本文將將擴展上一篇文章完成的 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

🪐感謝您觀看，祝好運🪐