本地大模型编程实战(25)用langgraph实现基于SQL数据构建的问答系统(4)

本文将演练使用基于 langgraph 链 ，对结构化数据库 SQlite 进行查询的方法。该系统建立以后，我们不需要掌握专业的 SQL 技能，可以用自然语言询问有关数据库中数据的问题并返回答案。
使 大语言模型(LLM) 查询结构化数据与非结构化文本数据有所不同。查询非结构化数据时，通常需要将待查询的文本嵌入到向量数据库中；而查询结构化数据的方法则是让 LLM 编写和执行 DSL（例如 SQL）进行查询。

在 用langchain实现基于SQL数据构建问答系统(1) 中，我们已经用 langchain 的 create_sql_query_chain 实现了类似功能，在本文结束后，我们将能看到它们之间的区别。

我们先看看完成的 langgraph 链的模样，情况一目了然，主要有两步：创建SQL查询语句->执行SQL查询语句：

使用 qwen2.5 、 deepseek 以及 llama3.1 做实验。

请注意:
构建 SQL 数据库的问答系统需要执行模型生成的 SQL 查询。这样做存在风险，请确保您的数据库连接权限始终尽可能小，这将减轻（但不能消除）构建模型驱动系统的风险。

准备

在正式开始撸代码之前，需要准备一下编程环境。

1. 计算机
   本文涉及的所有代码可以在没有显存的环境中执行。 我使用的机器配置为：

   • CPU: Intel i5-8400 2.80GHz
   • 内存: 16GB

2. Visual Studio Code 和 venv
   这是很受欢迎的开发工具，相关文章的代码可以在 Visual Studio Code 中开发和调试。 我们用 python 的 venv 创建虚拟环境, 详见：
   在Visual Studio Code中配置venv。

3. Ollama
   在 Ollama 平台上部署本地大模型非常方便，基于此平台，我们可以让 langchain 使用 llama3.1、qwen2.5、deepseek 等各种本地大模型。详见：
   在langchian中使用本地部署的llama3.1大模型 。

准备 SQLite 数据库

SQLite 是一个轻量级、嵌入式的关系型数据库管理系统，不需要独立的服务器进程，所有数据存储在单一文件中。它支持大部分 SQL 标准，适用于移动应用、本地存储和小型项目。

我们将使用 Chinook 数据库做练习，数据库文件放在本文最后提及的代码仓库中的 assert 文件夹，名为：Chinook.db 。
下图是该数据库的结构：

点击 sqlitestudio 可以下载 SQLite 的可视化管理工具，Sample Databases for SQLite 详细介绍了该数据库的情况。

• 创建数据库实例

from langchain_community.utilities import SQLDatabase
db = SQLDatabase.from_uri(f"sqlite:///{db_file_path}")

• 测试数据库

print(db.dialect)
print(db.get_usable_table_names())
print(db.run("SELECT * FROM Artist LIMIT 1;"))

sqlite
['Album', 'Artist', 'Customer', 'Employee', 'Genre', 'Invoice', 'InvoiceLine', 'MediaType', 'Playlist', 'PlaylistTrack', 'Track']
[(1, 'AC/DC')]

输出上述内容说明 SQLite 可以正常工作。

创建 langgraph 链

langgraph 链是可预测步骤/节点的组合。在 LangGraph 中，我们可以通过简单的节点序列来表示链。让我们创建一个节点序列，给定一个问题，它执行以下操作：

• 将问题转换为 SQL 查询；
• 执行查询；
• 使用查询结果回答原始问题。

因为此系统将对任何用户输入都会执行 SQL 查询，所以这种安排不支持某些场景。例如，我们问“你好”，它就无法运行。后面我们可以使用智能体解决这个问题。

状态

此应用程序的 LangGraph 状态控制哪些数据输入到应用程序、在步骤之间传输以及由应用程序输出。它通常是 TypedDict，但也可以是 Pydantic BaseModel。

from typing_extensions import TypedDictclass State(TypedDict):question: strquery: strresult: stranswer: str

根据问题生成SQL查询语句

为了可靠地获取 SQL 查询语句（缺少 markdown 格式和解释或说明），我们将利用 LangChain 的 with_structured_output 函数限定输出内容。

• 提示词

使用下面的提示词指导大模型输出SQL语句：

system = """You are an agent designed to interact with a SQL database.
Given an input question, create a syntactically correct {dialect} query to run, then look at the results of the query and return the answer.
Unless the user specifies a specific number of examples they wish to obtain, always limit your query to at most 5 results.
You can order the results by a relevant column to return the most interesting examples in the database.
Never query for all the columns from a specific table, only ask for the relevant columns given the question.
You have access to tools for interacting with the database.
Only use the given tools. Only use the information returned by the tools to construct your final answer.
You MUST double check your query before executing it. If you get an error while executing a query, rewrite the query and try again.DO NOT make any DML statements (INSERT, UPDATE, DELETE, DROP etc.) to the database.You have access to the following tables: {table_names}
""".format(table_names=db.get_usable_table_names(),dialect=db.dialect
)from langchain_core.prompts import ChatPromptTemplate
query_prompt_template = ChatPromptTemplate.from_messages([("system", system),("user", "Question:{input}")
])

提示包含我们需要填充的几个参数，例如 SQL 方言和表名称。下面定义生成SQL的函数：

def write_query(state: State):"""根据问题生成SQL查询语句"""prompt = query_prompt_template.invoke({"input": state["question"],})structured_llm = llm.with_structured_output(QueryOutput)result = structured_llm.invoke(prompt)print(f'Query is:\n{result["query"]}')return {"query": result["query"]}

• 用大模型测试

能否将问题生成正确的SQL语句是本系统的关键，我们用下面三个问题对三款大模型做一下简单的测试：

    questions = ["How many Employees are there?","Which country's customers spent the most?","Describe the PlaylistTrack table",     # 区分大小写，待改进。比如：用 PlaylistTrack 可以工作，但是用 playlisttrack 不准确]for question in questions:write_query({"question": question})

测试结果如下：

	How many Employees are there?	Which country’s customers spent the most?	Describe the PlaylistTrack table
qwen2.5	SELECT COUNT(*) AS EmployeeCount FROM Employee	SELECT T1.country FROM Customer AS T1 JOIN Invoice AS T2 ON T1.customerId = T2.customerId GROUP BY T1.country ORDER BY SUM(T2.total) DESC LIMIT 5	SELECT * FROM PlaylistTrack LIMIT 5
llama3.1	SELECT COUNT(*) FROM Employee	SELECT Customer.Country, SUM(Invoice.Total) FROM Invoice JOIN Customer ON Invoice.CustomerID = Customer.CustomerID GROUP BY Customer.Country ORDER BY SUM(Invoice.Total) DESC LIMIT 5	PRAGMA table_info(PlaylistTrack)

qwen2.5 和 llama3.1 针对前两个问题都生成了正确的SQL，llama3.1 针对第3个问题生成的SQL更好一点；而MFDoom/deepseek-r1-tool-calling:7b 和 deepseek-r1 执行时报错，无法生成SQL语句。

执行SQL查询

这是创建 SQL 链最危险的部分。请仔细考虑是否可以对数据运行自动查询，尽可能减少数据库连接权限。后面我们会考虑在查询执行之前为添加人工批准步骤。

def execute_query(state: State):"""执行SQL查询"""execute_query_tool = QuerySQLDatabaseTool(db=db)result = execute_query_tool.invoke(state["query"])print(f'Result is:\n{result}')return {"result": result}

生成答案

根据SQL执行的结果生成答案：

def generate_answer(state: State):"""使用检索到的信息作为上下文来回答问题。"""prompt = ("Given the following user question, corresponding SQL query, ""and SQL result, answer the user question.\n\n"f'Question: {state["question"]}\n'f'SQL Query: {state["query"]}\n'f'SQL Result: {state["result"]}')response = llm.invoke(prompt)print(f'answer is:\n{response.content}')return {"answer": response.content}

整合节点成为 langgraph 链

最后，我们将应用程序编译为单个 langgraph 对象。在本例中，我们只是将三个步骤连接成一个序列。

from langgraph.graph import START, StateGraphgraph_builder = StateGraph(State).add_sequence([write_query, execute_query, generate_answer]
)
graph_builder.add_edge(START, "write_query")
graph = graph_builder.compile()

测试大模型

现在我们依然用上面提到的三个问题，用大模型来测试一下前面创建的链。

下面是问答函数：

def ask(question):"""问答"""for step in graph.stream({"question": question}, stream_mode="updates"):print(step)

	How many Employees are there?	Which country’s customers spent the most?	Describe the PlaylistTrack table
qwen2.5	Based on the SQL result provided, there are 8 employees in total…	Based on the SQL query and the result provided, the country with the highest spending by its customers is USA…	The PlaylistTrack table contains two columns:…
llama3.1	There are 8 employees.	It looks like there’s an issue with the SQL query!..	…The PlaylistTrack table has two columns:…

在本轮测试中，qwen2.5 满分，llama3.1 推理出了第2个问题的SQL语句，但是不能执行。

总结

与 用langchain实现基于SQL数据构建问答系统(1) 中使用 create_sql_query_chain 相比，这次用 langgraph 创建的链看起来更复杂些，需要更多代码实现。
但是其优点也显而易见：结构清晰，容易理解，也适合定制。比如：deepseek 现在无法驱动 write_query 函数，但是我们实际上可以修改它，比如，用提示词指导大模型生成SQL，让它支持 deepseek 。
当然，目前此应用程序还有一些缺陷，比如：不支持人工审核是否执行SQL、只会查SQL数据库。我们后面的文章中会逐渐解决这些问题。

---

代码

本文涉及的所有代码以及相关资源都已经共享，参见：

• github
• gitee

为便于找到代码，程序文件名称最前面的编号与本系列文章的文档编号相同。

参考

• Build a Question/Answering system over SQL data

🪐感谢您观看，祝好运🪐