概览
在本教程中，你将学习如何使用 LangChain agents 构建一个可以回答 SQL 数据库相关问题的代理。 从高层来看，代理会：- 从数据库获取可用表和 schemas
- 判断哪些表与问题相关
- 获取相关表的 schemas
- 基于问题和 schemas 信息生成查询
- 使用 LLM 再次检查查询中的常见错误
- 执行查询并返回结果
- 修正数据库引擎暴露的错误，直到查询成功
- 基于结果组织响应
构建 SQL 数据库问答系统需要执行模型生成的 SQL 查询。这种做法存在固有风险。请确保数据库连接权限始终尽可能缩小到代理需求范围内。这会降低构建模型驱动系统的风险，但无法完全消除风险。
概念
本教程涵盖以下概念：- 用于读取 SQL 数据库的 Tools
- LangChain agents
- Human-in-the-loop 流程
设置
1
安装依赖
pip install langchain langgraph
2
设置 LangSmith
设置 LangSmith，以检查链或代理内部发生了什么。然后设置以下环境变量：
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."
构建 SQL 代理
1
选择 LLM
选择支持 tool-calling 的模型：下面示例中展示的输出使用 OpenAI。
- OpenAI
- Anthropic
- Azure
- Google Gemini
- AWS Bedrock
- HuggingFace
- OpenRouter
👉 阅读 OpenAI 聊天模型集成文档
pip install -U "langchain[openai]"
import os
from langchain.chat_models import init_chat_model
os.environ["OPENAI_API_KEY"] = "sk-..."
model = init_chat_model("gpt-5.5")
import os
from langchain_openai import ChatOpenAI
os.environ["OPENAI_API_KEY"] = "sk-..."
model = ChatOpenAI(model="gpt-5.5")
👉 阅读 Anthropic 聊天模型集成文档
pip install -U "langchain[anthropic]"
import os
from langchain.chat_models import init_chat_model
os.environ["ANTHROPIC_API_KEY"] = "sk-..."
model = init_chat_model("claude-sonnet-4-6")
import os
from langchain_anthropic import ChatAnthropic
os.environ["ANTHROPIC_API_KEY"] = "sk-..."
model = ChatAnthropic(model="claude-sonnet-4-6")
👉 阅读 Azure 聊天模型集成文档
pip install -U "langchain[openai]"
import os
from langchain.chat_models import init_chat_model
os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"
model = init_chat_model(
"azure_openai:gpt-5.5",
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)
import os
from langchain_openai import AzureChatOpenAI
os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"
model = AzureChatOpenAI(
model="gpt-5.5",
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]
)
👉 阅读 Google GenAI 聊天模型集成文档
pip install -U "langchain[google-genai]"
import os
from langchain.chat_models import init_chat_model
os.environ["GOOGLE_API_KEY"] = "..."
model = init_chat_model("google_genai:gemini-2.5-flash-lite")
import os
from langchain_google_genai import ChatGoogleGenerativeAI
os.environ["GOOGLE_API_KEY"] = "..."
model = ChatGoogleGenerativeAI(model="gemini-2.5-flash-lite")
👉 阅读 AWS Bedrock 聊天模型集成文档
pip install -U "langchain[aws]"
from langchain.chat_models import init_chat_model
# 请按照此页面的步骤配置您的凭证：
# https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html
model = init_chat_model(
"us.anthropic.claude-sonnet-4-6",
model_provider="bedrock_converse",
)
from langchain_aws import ChatBedrock
model = ChatBedrock(model="us.anthropic.claude-sonnet-4-6")
👉 阅读 HuggingFace 聊天模型集成文档
pip install -U "langchain[huggingface]"
import os
from langchain.chat_models import init_chat_model
os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."
model = init_chat_model(
"microsoft/Phi-3-mini-4k-instruct",
model_provider="huggingface",
temperature=0.7,
max_tokens=1024,
)
import os
from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint
os.environ["HUGGINGFACEHUB_API_TOKEN"] = "hf_..."
llm = HuggingFaceEndpoint(
repo_id="microsoft/Phi-3-mini-4k-instruct",
temperature=0.7,
max_length=1024,
)
model = ChatHuggingFace(llm=llm)
👉 阅读 OpenRouter 聊天模型集成文档
pip install -U "langchain-openrouter"
import os
from langchain.chat_models import init_chat_model
os.environ["OPENROUTER_API_KEY"] = "sk-..."
model = init_chat_model(
"auto",
model_provider="openrouter",
)
import os
from langchain_openrouter import ChatOpenRouter
os.environ["OPENROUTER_API_KEY"] = "sk-..."
model = ChatOpenRouter(model="auto")
2
配置数据库
本教程将创建一个 SQLite database。SQLite 是一种易于设置和使用的轻量数据库。这里会加载 将使用 Python 内置的
chinook 数据库，它是一个表示数字媒体商店的示例数据库。为了方便，数据库（Chinook.db）托管在公开 GCS bucket 上。import pathlib
import requests
url = "https://storage.googleapis.com/benchmarks-artifacts/chinook/Chinook.db"
local_path = pathlib.Path("Chinook.db")
if local_path.exists():
print(f"{local_path} already exists, skipping download.")
else:
response = requests.get(url, timeout=60)
if response.status_code == 200:
local_path.write_bytes(response.content)
print(f"File downloaded and saved as {local_path}")
else:
print(f"Failed to download the file. Status code: {response.status_code}")
sqlite3 模块与数据库交互：import sqlite3
con = sqlite3.connect("Chinook.db")
cursor = con.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = [row[0] for row in cursor.fetchall() if not row[0].startswith("sqlite_")]
print("Dialect: sqlite")
print(f"Available tables: {tables}")
cursor.execute("SELECT * FROM Artist LIMIT 5;")
print(f"Sample output: {cursor.fetchall()}")
con.close()
Dialect: sqlite
Available tables: ['Album', 'Artist', 'Customer', 'Employee', 'Genre', 'Invoice', 'InvoiceLine', 'MediaType', 'Playlist', 'PlaylistTrack', 'Track']
Sample output: [(1, 'AC/DC'), (2, 'Accept'), (3, 'Aerosmith'), (4, 'Alanis Morissette'), (5, 'Alice In Chains')]
3
添加用于数据库交互的工具
以下数据库工具只是用于演示的最小 wrapper。它们并不用于安全场景或生产环境。在执行模型生成的 SQL 之前，请使用范围尽可能小的数据库权限，并添加应用特定验证。
langchain.tools 中的 @tool 装饰器，将数据库 tools 实现为轻量 wrapper：import sqlite3
from langchain.tools import tool
# 以下是用于演示目的的最小化工具。
# 它们并非设计为安全或用于生产环境。
@tool
def sql_db_list_tables() -> str:
"""输入为空字符串，输出为数据库中以逗号分隔的表名列表。"""
con = sqlite3.connect("Chinook.db")
try:
cursor = con.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = [row[0] for row in cursor.fetchall() if not row[0].startswith("sqlite_")]
return ", ".join(tables)
finally:
con.close()
@tool
def sql_db_schema(table_names: str) -> str:
"""此工具的输入为以逗号分隔的表名列表，输出为这些表的模式信息及示例行。
务必先调用 sql_db_list_tables 以确认这些表确实存在！
示例输入: table1, table2, table3"""
con = sqlite3.connect("Chinook.db")
try:
cursor = con.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
valid_tables = {row[0] for row in cursor.fetchall() if not row[0].startswith("sqlite_")}
results = []
for table in table_names.split(","):
table = table.strip()
if table not in valid_tables:
results.append(f"错误: 数据库中未找到表名 {{{table!r}}}")
continue
cursor.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name=?;", (table,))
schema_row = cursor.fetchone()
if schema_row:
results.append(schema_row[0])
try:
quoted_table = '"' + table.replace('"', '""') + '"'
cursor.execute(f"SELECT * FROM {quoted_table} LIMIT 3;")
rows = cursor.fetchall()
if rows:
col_names = [description[0] for description in cursor.description]
results.append(
f"/*\n来自 {table} 表的3行数据:\n"
+ "\t".join(col_names)
+ "\n"
+ "\n".join("\t".join(str(x) for x in row) for row in rows)
+ "\n*/"
)
except Exception as e:
results.append(f"获取示例行时出错: {e}")
return "\n\n".join(results)
finally:
con.close()
@tool
def sql_db_query(query: str) -> str:
"""此工具的输入为一个详细且正确的 SQL 查询，输出为数据库查询结果。
如果查询不正确，将返回错误信息。
如果返回错误，请重写查询语句、检查查询并再次尝试。
如果遇到“Unknown column 'xxxx' in 'field list'”问题，请使用 sql_db_schema 查询正确的表字段。"""
con = sqlite3.connect("Chinook.db")
try:
cursor = con.cursor()
cursor.execute(query)
res = cursor.fetchall()
return str(res)
except Exception as e:
return f"错误: {e}"
finally:
con.close()
@tool
def sql_db_query_checker(query: str) -> str:
"""在执行查询前，使用此工具仔细检查您的查询是否正确。
在使用 sql_db_query 执行查询前，请务必使用此工具！"""
trigger_prompt = """{query}
请仔细检查上述 sqlite 查询的常见错误，包括:
- 使用 NOT IN 时包含 NULL 值
- 应该使用 UNION ALL 时使用了 UNION
- 使用 BETWEEN 进行排他范围查询
- 谓词中的数据类型不匹配
- 正确引用标识符
- 使用正确数量的函数参数
- 转换为正确的数据类型
- 使用正确的列进行联接
如果存在上述任何错误，请重写查询。如果没有错误，只需重现原始查询。
仅输出最终的 SQL 查询。
SQL 查询: """.format(query=query)
response = model.invoke(trigger_prompt)
return response.text.strip()
tools = [sql_db_list_tables, sql_db_schema, sql_db_query, sql_db_query_checker]
# 使用一个不同的循环变量，以免它遮蔽 `tool` 装饰器。
for t in tools:
print(f"{t.name}: {t.description}\n")
sql_db_query: Input to this tool is a detailed and correct SQL query, output is a result from the database.
If the query is not correct, an error message will be returned.
If an error is returned, rewrite the query, check the query, and try again.
If you encounter an issue with Unknown column 'xxxx' in 'field list', use sql_db_schema to query the correct table fields.
sql_db_schema: Input to this tool is a comma-separated list of tables, output is the schema and sample rows for those tables.
Be sure that the tables actually exist by calling sql_db_list_tables first!
Example Input: table1, table2, table3
sql_db_list_tables: Input is an empty string, output is a comma-separated list of tables in the database.
sql_db_query_checker: Use this tool to double check if your query is correct before executing it.
Always use this tool before executing a query with sql_db_query!
4
创建代理
使用 现在，用模型、工具和 prompt 创建代理：
create_agent 以最少代码构建 ReAct agent。代理会解释请求并生成 SQL 命令，然后由工具执行。如果命令有错误，错误消息会返回给模型。模型随后可以检查原始请求和新错误消息，并生成新命令。这个过程可以持续到 LLM 成功生成命令或达到终止次数。向模型提供反馈（这里是错误消息）的模式非常强大。使用描述性 system prompt 初始化代理，以自定义其行为：system_prompt = """
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 {top_k} 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 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.
To start you should ALWAYS look at the tables in the database to see what you
can query. Do NOT skip this step.
Then you should query the schema of the most relevant tables.
""".format(
dialect="sqlite",
top_k=5,
)
from langchain.agents import create_agent
agent = create_agent(
model,
tools,
system_prompt=system_prompt,
)
5
运行代理
用示例查询运行代理并观察其行为：代理正确编写了查询、检查了查询，并运行查询以支持最终响应。
question = "哪种流派的曲目平均时长最长？"
stream = agent.stream_events(
{"messages": [{"role": "user", "content": question}]},
version="v3",
)
for kind, item in stream.interleave("messages", "tool_calls"):
if kind == "messages":
for token in item.text:
print(token, end="", flush=True)
elif kind == "tool_calls":
print(f"\nTool call: {item.tool_name}({item.input})")
for delta in item.output_deltas:
print(delta, end="", flush=True)
print(f"\nTool result: {item.output}")
final_state = stream.output
================================ Human Message =================================
Which genre on average has the longest tracks?
================================== Ai Message ==================================
Tool Calls:
sql_db_list_tables (call_BQsWg8P65apHc8BTJ1NPDvnM)
Call ID: call_BQsWg8P65apHc8BTJ1NPDvnM
Args:
================================= Tool Message =================================
Name: sql_db_list_tables
Album, Artist, Customer, Employee, Genre, Invoice, InvoiceLine, MediaType, Playlist, PlaylistTrack, Track
================================== Ai Message ==================================
Tool Calls:
sql_db_schema (call_i89tjKECFSeERbuACYm4w0cU)
Call ID: call_i89tjKECFSeERbuACYm4w0cU
Args:
table_names: Track, Genre
================================= Tool Message =================================
Name: sql_db_schema
CREATE TABLE "Genre" (
"GenreId" INTEGER NOT NULL,
"Name" NVARCHAR(120),
PRIMARY KEY ("GenreId")
)
/*
3 rows from Genre table:
GenreId Name
1 Rock
2 Jazz
3 Metal
*/
CREATE TABLE "Track" (
"TrackId" INTEGER NOT NULL,
"Name" NVARCHAR(200) NOT NULL,
"AlbumId" INTEGER,
"MediaTypeId" INTEGER NOT NULL,
"GenreId" INTEGER,
"Composer" NVARCHAR(220),
"Milliseconds" INTEGER NOT NULL,
"Bytes" INTEGER,
"UnitPrice" NUMERIC(10, 2) NOT NULL,
PRIMARY KEY ("TrackId"),
FOREIGN KEY("MediaTypeId") REFERENCES "MediaType" ("MediaTypeId"),
FOREIGN KEY("GenreId") REFERENCES "Genre" ("GenreId"),
FOREIGN KEY("AlbumId") REFERENCES "Album" ("AlbumId")
)
/*
3 rows from Track table:
TrackId Name AlbumId MediaTypeId GenreId Composer Milliseconds Bytes UnitPrice
1 For Those About To Rock (We Salute You) 1 1 1 Angus Young, Malcolm Young, Brian Johnson 343719 11170334 0.99
2 Balls to the Wall 2 2 1 U. Dirkschneider, W. Hoffmann, H. Frank, P. Baltes, S. Kaufmann, G. Hoffmann 342562 5510424 0.99
3 Fast As a Shark 3 2 1 F. Baltes, S. Kaufman, U. Dirkscneider & W. Hoffman 230619 3990994 0.99
*/
================================== Ai Message ==================================
Tool Calls:
sql_db_query_checker (call_G64yYm6R6UauiVPCXJZMA49b)
Call ID: call_G64yYm6R6UauiVPCXJZMA49b
Args:
query: SELECT Genre.Name, AVG(Track.Milliseconds) AS AverageLength FROM Track INNER JOIN Genre ON Track.GenreId = Genre.GenreId GROUP BY Genre.Name ORDER BY AverageLength DESC LIMIT 5;
================================= Tool Message =================================
Name: sql_db_query_checker
SELECT Genre.Name, AVG(Track.Milliseconds) AS AverageLength FROM Track INNER JOIN Genre ON Track.GenreId = Genre.GenreId GROUP BY Genre.Name ORDER BY AverageLength DESC LIMIT 5;
================================== Ai Message ==================================
Tool Calls:
sql_db_query (call_AnO3SrhD0ODJBxh6dHMwvHwZ)
Call ID: call_AnO3SrhD0ODJBxh6dHMwvHwZ
Args:
query: SELECT Genre.Name, AVG(Track.Milliseconds) AS AverageLength FROM Track INNER JOIN Genre ON Track.GenreId = Genre.GenreId GROUP BY Genre.Name ORDER BY AverageLength DESC LIMIT 5;
================================= Tool Message =================================
Name: sql_db_query
[('Sci Fi & Fantasy', 2911783.0384615385), ('Science Fiction', 2625549.076923077), ('Drama', 2575283.78125), ('TV Shows', 2145041.0215053763), ('Comedy', 1585263.705882353)]
================================== Ai Message ==================================
On average, the genre with the longest tracks is "Sci Fi & Fantasy" with an average track length of approximately 2,911,783 milliseconds. This is followed by "Science Fiction," "Drama," "TV Shows," and "Comedy."
你可以在 LangSmith trace 中检查上述运行的各个方面，包括执行的步骤、调用的工具、LLM 看到的 prompts 等。
6
（可选）使用 Studio
Studio 提供“client side”循环和 memory，因此你可以将其作为聊天界面运行并查询数据库。你可以提出诸如 “Tell me the scheme of the database” 或 “Show me the invoices for the 5 top customers” 的问题。你会看到生成的 SQL 命令以及结果输出。下面是启动方式详情。
在 Studio 中运行代理
在 Studio 中运行代理
除了前面提到的包之外，还需要：在运行目录中，需要一个包含以下内容的 创建文件
pip install -U langgraph-cli[inmem]>=0.4.0
langgraph.json 文件：{
"dependencies": ["."],
"graphs": {
"agent": "./sql_agent.py:agent",
"graph": "./sql_agent_langgraph.py:graph"
},
"env": ".env"
}
sql_agent.py 并插入以下内容：# 适用于 studio 的 sql_agent.py
import pathlib
import sqlite3
import requests
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
# 初始化一个大语言模型
model = init_chat_model("gpt-5.5")
# 获取数据库文件，并保存在本地
url = "https://storage.googleapis.com/benchmarks-artifacts/chinook/Chinook.db"
local_path = pathlib.Path("Chinook.db")
if local_path.exists():
print(f"{local_path} 已存在，跳过下载。")
else:
response = requests.get(url, timeout=60)
if response.status_code == 200:
local_path.write_bytes(response.content)
print(f"文件已下载并保存为 {local_path}")
else:
print(f"文件下载失败。状态码: {response.status_code}")
# 以下为用于演示的最小化工具。
@tool
def sql_db_list_tables() -> str:
"""输入是一个空字符串，输出是数据库中以逗号分隔的表名列表。"""
con = sqlite3.connect("Chinook.db")
try:
cursor = con.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = [row[0] for row in cursor.fetchall() if not row[0].startswith("sqlite_")]
return ", ".join(tables)
finally:
con.close()
@tool
def sql_db_schema(table_names: str) -> str:
"""输入给此工具的是一个以逗号分隔的表名列表，输出是这些表的结构和示例行。
请务必先调用 sql_db_list_tables 以确认这些表确实存在！
输入示例：table1, table2, table3"""
con = sqlite3.connect("Chinook.db")
try:
cursor = con.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
valid_tables = {row[0] for row in cursor.fetchall() if not row[0].startswith("sqlite_")}
results = []
for table in table_names.split(","):
table = table.strip()
if table not in valid_tables:
results.append(f"错误：表名 {{{table!r}}} 在数据库中未找到")
continue
cursor.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name=?;", (table,))
schema_row = cursor.fetchone()
if schema_row:
results.append(schema_row[0])
try:
quoted_table = '"' + table.replace('"', '""') + '"'
cursor.execute(f"SELECT * FROM {quoted_table} LIMIT 3;")
rows = cursor.fetchall()
if rows:
col_names = [description[0] for description in cursor.description]
results.append(
f"/*\n来自 {table} 表的 3 行数据：\n"
+ "\t".join(col_names)
+ "\n"
+ "\n".join("\t".join(str(x) for x in row) for row in rows)
+ "\n*/"
)
except Exception as e:
results.append(f"获取示例行时出错：{e}")
return "\n\n".join(results)
finally:
con.close()
@tool
def sql_db_query(query: str) -> str:
"""输入给此工具的是一条详细且正确的 SQL 查询语句，输出是数据库返回的结果。
如果查询不正确，将返回错误信息。
如果返回错误，请重写查询语句，检查查询，然后重试。
如果遇到类似“字段列表中的未知列 'xxxx'”的问题，请使用 sql_db_schema 查询正确的表字段。"""
con = sqlite3.connect("Chinook.db")
try:
cursor = con.cursor()
cursor.execute(query)
res = cursor.fetchall()
return str(res)
except Exception as e:
return f"错误：{e}"
finally:
con.close()
@tool
def sql_db_query_checker(query: str) -> str:
"""使用此工具在执行查询前仔细检查你的查询是否正确。
在使用 sql_db_query 执行查询之前，务必先使用此工具！"""
trigger_prompt = """{query}
仔细检查上面的 sqlite 查询是否有常见错误，包括：
- 在包含 NULL 值的列上使用 NOT IN
- 本应使用 UNION ALL 时却使用了 UNION
- 对使用排他范围使用了 BETWEEN
- 谓词中的数据类型不匹配
- 标识符的引用是否正确
- 函数的参数数量是否正确
- 是否转换为了正确的数据类型
- 连接使用的列是否正确
如果存在上述任何错误，请重写查询。如果没有错误，则只需重现原始查询。
仅输出最终的 SQL 查询语句。
SQL 查询语句：""".format(query=query)
response = model.invoke(trigger_prompt)
return response.text.strip()
tools = [sql_db_list_tables, sql_db_schema, sql_db_query, sql_db_query_checker]
# 使用一个独立的循环变量，以免与 `tool` 装饰器的作用域冲突。
for t in tools:
print(f"{t.name}: {t.description}\n")
# 使用 create_agent
system_prompt = """
你是一个设计用于与 SQL 数据库交互的代理。
给定一个输入问题，创建一个语法正确的 {dialect} 查询语句来执行，
然后查看查询结果并返回答案。除非用户指定了他们希望获取的特定示例数量，
否则始终将你的查询结果限制在最多 {top_k} 条。
你可以通过一个相关的列对结果进行排序，以返回数据库中最有趣的示例。
永远不要查询特定表的所有列，只根据问题请求相关的列。
你必须在执行查询前仔细检查你的查询。如果执行查询时出错，请重写查询并重试。
不要对数据库执行任何 DML 语句（INSERT、UPDATE、DELETE、DROP 等）。
你应该首先查看数据库中有哪些表，以便了解可以查询什么。不要跳过此步骤。
然后，你应该查询最相关表的结构。
""".format(
dialect="sqlite",
top_k=5,
)
agent = create_agent(
model,
tools,
system_prompt=system_prompt,
)
7
实现 human-in-the-loop 审核
在执行代理的 SQL 查询之前，检查是否存在任何非预期动作或低效之处是一种谨慎做法。LangChain 代理支持内置 human-in-the-loop middleware，用于为代理工具调用添加监督。配置代理在调用 运行代理时，它现在会在执行 可以使用 Command 恢复执行，本例中接受该查询：详情请参阅 human-in-the-loop guide。
sql_db_query 工具时暂停并等待人工审核：from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent(
model,
tools,
system_prompt=system_prompt,
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={"sql_db_query": True},
description_prefix="Tool execution pending approval",
),
],
checkpointer=InMemorySaver(),
)
已向代理添加 checkpointer，以允许暂停和恢复执行。更多详情以及可用 middleware 配置，请参阅 human-in-the-loop guide。
sql_db_query 工具前暂停等待审核：question = "哪个流派的平均曲目时长最长？"
config = {"configurable": {"thread_id": "1"}}
stream = agent.stream_events(
{"messages": [{"role": "user", "content": question}]},
config,
version="v3",
)
for kind, item in stream.interleave("messages", "tool_calls"):
if kind == "messages":
for token in item.text:
print(token, end="", flush=True)
elif kind == "tool_calls":
print(f"\n工具调用：{item.tool_name}({item.input})")
if stream.interrupted:
print("已中断：")
interrupt = stream.interrupts[0]
for request in interrupt.value["action_requests"]:
print(request["description"])
...
INTERRUPTED:
Tool execution pending approval
Tool: sql_db_query
Args: {'query': 'SELECT g.Name AS Genre, AVG(t.Milliseconds) AS AvgTrackLength FROM Track t JOIN Genre g ON t.GenreId = g.GenreId GROUP BY g.Name ORDER BY AvgTrackLength DESC LIMIT 1;'}
from langgraph.types import Command
stream = agent.stream_events(
Command(resume={"decisions": [{"type": "approve"}]}),
config,
version="v3",
)
for kind, item in stream.interleave("messages", "tool_calls"):
if kind == "messages":
for token in item.text:
print(token, end="", flush=True)
elif kind == "tool_calls":
print(f"\n工具调用: {item.tool_name}({item.input})")
if stream.interrupted:
print("中断:")
interrupt = stream.interrupts[0]
for request in interrupt.value["action_requests"]:
print(request["description"])
================================== Ai Message ==================================
Tool Calls:
sql_db_query (call_7oz86Epg7lYRqi9rQHbZPS1U)
Call ID: call_7oz86Epg7lYRqi9rQHbZPS1U
Args:
query: SELECT Genre.Name, AVG(Track.Milliseconds) AS AvgDuration FROM Track JOIN Genre ON Track.GenreId = Genre.GenreId GROUP BY Genre.Name ORDER BY AvgDuration DESC LIMIT 5;
================================= Tool Message =================================
Name: sql_db_query
[('Sci Fi & Fantasy', 2911783.0384615385), ('Science Fiction', 2625549.076923077), ('Drama', 2575283.78125), ('TV Shows', 2145041.0215053763), ('Comedy', 1585263.705882353)]
================================== Ai Message ==================================
The genre with the longest average track length is "Sci Fi & Fantasy" with an average duration of about 2,911,783 milliseconds, followed by "Science Fiction" and "Drama."
下一步
如需更深入的自定义，请查看本教程，了解如何直接使用 LangGraph primitives 实现 SQL 代理。Connect these docs to Claude, VSCode, and more via MCP for real-time answers.