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

本文将将扩展上一篇文章完成的 langgraph 链，继续使用基于 langgraph 链 ，对结构化数据库 SQlite 进行查询的方法。该系统建立以后，我们不需要掌握专业的 SQL 技能，可以用自然语言询问有关数据库中数据的问题并返回答案。主要完善一下两点内容：

• 自动记录消息历史
• 增加人工审核环节，防止 LLM(大语言模型) 运行危险的SQL语句

我们先看看完成的 langgraph 链的模样，主要有两步：创建SQL查询语句->执行SQL查询语句，在执行SQL查询前中断进行人工审核，上一篇文章的 链 没有人工审核：

使用 qwen2.5 、 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大模型 。

定义 langgraph 步骤/节点

用langgraph实现基于SQL数据构建的问答系统(4) 中对这部分工作有详细的阐述，这里仅贴出主要代码，以使得本文内容比较连贯：

"""
1. 创建SQLite对象
"""from langchain_community.utilities import SQLDatabasedb = SQLDatabase.from_uri(f"sqlite:///{db_file_path}")"""
2. 状态
"""from typing_extensions import TypedDictclass State(TypedDict):question: strquery: strresult: stranswer: strfrom langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3.1",temperature=0, verbose=True)def set_llm(llm_model_name):"""设置大模型，用于测试不同大模型"""global llm llm = ChatOllama(model=llm_model_name,temperature=0, verbose=True)"""
3. 定义langgraph节点
"""from typing_extensions import Annotatedclass QueryOutput(TypedDict):"""生成的SQL查询语句"""query: Annotated[str, ..., "Syntactically valid SQL query."]# 提示词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}")
])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"]}from langchain_community.tools.sql_database.tool import QuerySQLDatabaseTooldef 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}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 可以在敏感步骤（例如执行 SQL 查询）之前中断应用程序以供人工审核。这是由 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()from langgraph.checkpoint.memory import MemorySavermemory = MemorySaver()
graph_with_human = graph_builder.compile(checkpointer=memory, interrupt_before=["execute_query"])

上述代码在 execute_query 执行前中断了流程，以使得人可以进行人工审核。
MemorySaver 将链的执行过程存储在内存中，它实际上也记录了聊天历史，使得链具有记忆功能，拥有聊天的上下文信息，可以与用户进行多轮连续对话。

在问答中增加人工审核

下面我们定义问答方法：

def ask_with_human(question,thread_id):"""问答：增加了人工审核"""config = {"configurable": {"thread_id": thread_id}}for step in graph_with_human.stream({"question": question},config,stream_mode="updates",):print(step)try:user_approval = input("您确定要执行查询么？(yes/no): ")except Exception:user_approval = "no"if user_approval.lower() == "yes":# 如果获得批准，再继续执行for step in graph_with_human.stream(None, config, stream_mode="updates"):print(step)else:print("操作已被取消。")

上面的代码中增加了人工审核逻辑。

定义测试方法

为方便对多款大模型进行对比测试，我们定义一个简单的测试方法,其中定义了两个问题：

def test_model(llm_model_name):"""测试大模型"""print(f'============{llm_model_name}==========')set_llm(llm_model_name)thread_id = "liu23"questions = ["How many Employees are there?","Which country's customers spent the most?",]for question in questions:ask_with_human( question,thread_id)

见证效果

qwen2.5 和 llama3.1 处理这些逻辑都没有问题，我们用 qwen2.5 执行第1个问题，了解一下执行过程：

• 同意/yes

{'write_query': {'query': 'SELECT COUNT(*) AS EmployeeCount FROM Employee'}}
{'__interrupt__': ()}
您确定要执行查询么？(yes/no): yes
{'execute_query': {'result': '[(8,)]'}}
{'generate_answer': {'answer': 'Based on the SQL result provided, there are 8 employees in total. The result `(8,)` indicates that the count of employees is 8.'}}

• 不同意/no

{'write_query': {'query': 'SELECT COUNT(*) AS EmployeeCount FROM Employee'}}
{'__interrupt__': ()}
您确定要执行查询么？(yes/no): no
操作已被取消。

总结

langgraph 可以在不修改步骤/节点逻辑的情况下增加人工审批环节，nice。

---

代码

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

• github
• gitee

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

参考

• Build a Question/Answering system over SQL data

🪐感谢您观看，祝好运🪐