本文將將擴(kuò)展上一篇文章完成的 langgraph 鏈，繼續(xù)使用基于 langgraph 鏈 ，對(duì)結(jié)構(gòu)化數(shù)據(jù)庫(kù) SQlite 進(jìn)行查詢的方法。該系統(tǒng)建立以后，我們不需要掌握專業(yè)的 SQL 技能，可以用自然語(yǔ)言詢問(wèn)有關(guān)數(shù)據(jù)庫(kù)中數(shù)據(jù)的問(wèn)題并返回答案。主要完善一下兩點(diǎn)內(nèi)容：

• 自動(dòng)記錄消息歷史
• 增加人工審核環(huán)節(jié)，防止 LLM(大語(yǔ)言模型) 運(yùn)行危險(xiǎn)的SQL語(yǔ)句

我們先看看完成的 langgraph 鏈的模樣，主要有兩步：創(chuàng)建SQL查詢語(yǔ)句->執(zhí)行SQL查詢語(yǔ)句，在執(zhí)行SQL查詢前中斷進(jìn)行人工審核，上一篇文章的 鏈 沒(méi)有人工審核：

使用 qwen2.5 、 llama3.1 做實(shí)驗(yàn)。

請(qǐng)注意:
構(gòu)建 SQL 數(shù)據(jù)庫(kù)的問(wèn)答系統(tǒng)需要執(zhí)行模型生成的 SQL 查詢。這樣做存在風(fēng)險(xiǎn)，請(qǐng)確保您的數(shù)據(jù)庫(kù)連接權(quán)限始終盡可能小，這將減輕（但不能消除）構(gòu)建模型驅(qū)動(dòng)系統(tǒng)的風(fēng)險(xiǎn)。

準(zhǔn)備開(kāi)發(fā)環(huán)境

在正式開(kāi)始擼代碼之前，需要準(zhǔn)備一下編程環(huán)境。

1. 計(jì)算機(jī)
   本文涉及的所有代碼可以在沒(méi)有顯存的環(huán)境中執(zhí)行。 我使用的機(jī)器配置為：

   • CPU: Intel i5-8400 2.80GHz
   • 內(nèi)存: 16GB

2. Visual Studio Code 和 venv
   這是很受歡迎的開(kāi)發(fā)工具，相關(guān)文章的代碼可以在 Visual Studio Code 中開(kāi)發(fā)和調(diào)試。 我們用 python 的 venv 創(chuàng)建虛擬環(huán)境, 詳見(jiàn)：
   在Visual Studio Code中配置venv。

3. Ollama
   在 Ollama 平臺(tái)上部署本地大模型非常方便，基于此平臺(tái)，我們可以讓 langchain 使用 llama3.1、qwen2.5、deepseek 等各種本地大模型。詳見(jiàn)：
   在langchian中使用本地部署的llama3.1大模型 。

定義 langgraph 步驟/節(jié)點(diǎn)

用langgraph實(shí)現(xiàn)基于SQL數(shù)據(jù)構(gòu)建的問(wèn)答系統(tǒng)(4) 中對(duì)這部分工作有詳細(xì)的闡述，這里僅貼出主要代碼，以使得本文內(nèi)容比較連貫：

"""
1. 創(chuàng)建SQLite對(duì)象
"""from langchain_community.utilities import SQLDatabasedb = SQLDatabase.from_uri(f"sqlite:///{db_file_path}")"""
2. 狀態(tài)
"""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):"""設(shè)置大模型，用于測(cè)試不同大模型"""global llm llm = ChatOllama(model=llm_model_name,temperature=0, verbose=True)"""
3. 定義langgraph節(jié)點(diǎn)
"""from typing_extensions import Annotatedclass QueryOutput(TypedDict):"""生成的SQL查詢語(yǔ)句"""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):"""根據(jù)問(wèn)題生成SQL查詢語(yǔ)句"""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):"""執(zhí)行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):"""使用檢索到的信息作為上下文來(lái)回答問(wèn)題。"""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}

增加人工審核節(jié)點(diǎn)

LangGraph 可以在敏感步驟（例如執(zhí)行 SQL 查詢）之前中斷應(yīng)用程序以供人工審核。這是由 LangGraph 的持久層啟用的，它將運(yùn)行進(jìn)度保存到存儲(chǔ)中。

使用內(nèi)存存儲(chǔ)

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 執(zhí)行前中斷了流程，以使得人可以進(jìn)行人工審核。
MemorySaver 將鏈的執(zhí)行過(guò)程存儲(chǔ)在內(nèi)存中，它實(shí)際上也記錄了聊天歷史，使得鏈具有記憶功能，擁有聊天的上下文信息，可以與用戶進(jìn)行多輪連續(xù)對(duì)話。

在問(wèn)答中增加人工審核

下面我們定義問(wèn)答方法：

def ask_with_human(question,thread_id):"""問(wèn)答：增加了人工審核"""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("您確定要執(zhí)行查詢么？(yes/no): ")except Exception:user_approval = "no"if user_approval.lower() == "yes":# 如果獲得批準(zhǔn)，再繼續(xù)執(zhí)行for step in graph_with_human.stream(None, config, stream_mode="updates"):print(step)else:print("操作已被取消。")

上面的代碼中增加了人工審核邏輯。

定義測(cè)試方法

為方便對(duì)多款大模型進(jìn)行對(duì)比測(cè)試，我們定義一個(gè)簡(jiǎn)單的測(cè)試方法,其中定義了兩個(gè)問(wèn)題：

def test_model(llm_model_name):"""測(cè)試大模型"""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)

見(jiàn)證效果

qwen2.5 和 llama3.1 處理這些邏輯都沒(méi)有問(wèn)題，我們用 qwen2.5 執(zhí)行第1個(gè)問(wèn)題，了解一下執(zhí)行過(guò)程：

• 同意/yes

{'write_query': {'query': 'SELECT COUNT(*) AS EmployeeCount FROM Employee'}}
{'__interrupt__': ()}
您確定要執(zhí)行查詢么？(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__': ()}
您確定要執(zhí)行查詢么？(yes/no): no
操作已被取消。

總結(jié)

langgraph 可以在不修改步驟/節(jié)點(diǎn)邏輯的情況下增加人工審批環(huán)節(jié)，nice。

---

代碼

本文涉及的所有代碼以及相關(guān)資源都已經(jīng)共享，參見(jiàn)：

• github
• gitee

為便于找到代碼，程序文件名稱最前面的編號(hào)與本系列文章的文檔編號(hào)相同。

參考

• Build a Question/Answering system over SQL data

🪐感謝您觀看，祝好運(yùn)🪐