二十集了，评论区催得最狠的就是SQL这一块。前几集把工具调用、输出解析、记忆机制都过了，很多人真正上手LangChain之后发现，文档和API网关都好说，一接数据库就露怯。单独把自然语言转成SQL并不算难，难的是把这条链路交给Agent之后，你根本猜不到它下一步要干什么。这集就把SQL Agent和Human-in-the-loop彻底聊透：一边讲LangChain怎么把自然语言查询变成一条可执行的数据库工具链，另一边讲怎么在关键节点把表决权留在人手里。

1. SQL Agent设计思路与核心链路

1.1 SQL Agent到底在解决什么问题

从早期的Text2SQL开始，大家的做法基本是：写一段prompt让大模型生成SQL，把SQL拿到数据库里执行，再把结果丢回给模型生成人话。这套流程在小样本上看起来很顺，一放到真实业务里就垮。垮的原因不是SQL生成得不对，而是真实数据库的表结构往往藏着一堆隐含信息：某张表是冗余的、某个字段是废弃状态、某些表必须靠特定条件过滤才能查，否则性能就崩了。你不可能把所有信息都塞进一段prompt里。

SQL Agent的解法是把数据库本身变成Agent可以触达的工具环境。Agent拿到用户问题之后，不再是一步到位生成SQL，而是一个反思式的多轮行动链：先查看当前数据库有哪些表，再看目标表的字段结构，然后构造SQL、执行、观察结果，发现结果不对再修正。这个链路本质上是把“程序员查数据库”的工作方式教给了大模型。

LangChain里实现这个能力不需要从零拼装，直接用SQLDatabaseToolkit就能组装出完整工具集，再用create_sql_agent把大模型和工具集绑定成一个AgentExecutor。我在本地一个模拟销售数据库上跑过完整流程，用一个很开放的自然语言问题“帮我看看今年各季度销售额趋势”，Agent的执行日志是下图这种形态的：

> Entering new SQL Agent Executor chain... Thought: 用户想看销售额趋势，先确认哪些表包含订单和销售数据。 Action: sql_db_list_tables Observation: orders, customers, products, order_items Thought: 订单金额在orders表，订单时间也在orders表，需要查看orders表结构。 Action: sql_db_schema Observation: 返回orders表的列名和类型... Thought: 现在构造按季度聚合的SQL。 Action: sql_db_query Observation: 返回季度、销售额聚合行... Thought: 数据已经拿到，组织成中文回答。

这个循环比“一句prompt生成SQL”健壮得多，因为每一步都有可观测的工具返回值，模型可以根据观测结果修正下一步动作。

1.2 为什么Human-in-the-loop不是可选而是必有

这里要面对一个现实问题：SQL Agent在无约束环境下能做什么，完全取决于模型的判断。模型觉得用户说“把30天未活跃用户标记为流失”，就真的很可能生成一条UPDATE语句直接改掉线上用户表。如果是DELETE或TRUNCATE，那后果更是灾难性的。

我在公司内部搭建数据分析台的时候，最开始按常规思路直接生成一个完全自动的SQL Agent，上线第一天就遇到状况。某位使用者问Agent“把那些测试订单清理掉”，Agent生成了DELETE语句，如果不是发现目标表是生产订单表并且手动终止，这条语句就直接提交了。原因很简单：模型并不真正理解“测试订单”在业务上意味着什么，它只是根据关键词推测。

引入Human-in-the-loop的价值主要有三层：

• 第一层是权限。数据库层面可以做只读账号，但只有“不再被盲操作”的语义保障，Agent本身并不知道哪些表敏感，人可以在执行前最后把关。
• 第二层是审计。每次人工审批都会留下Who-What-When记录，方便追责和复盘。
• 第三层是提高Agent的“胆量”。没有人类兜底的Agent会被各种安全规则限制得缩手缩脚，有了审批闸门，反而可以把能力放开给更多业务人员用。

很多人误以为Human-in-the-loop是降低自动化程度，实际上它是在给Agent戴上安全带之后，允许它开快车。

1.3 SQL Agent与普通问答链/文本转SQL脚本的区别

拿普通Retrieval QA链和SQL Agent对比，最容易混淆的能力边界在于“查什么”和“怎么查”。普通的文档问答链适合检索非结构化文档，它的工具是向量库，返回的是相似段落。SQL Agent面对的是结构化数据，工具是SQL执行器，返回的是表格。两者看似都在回答自然语言问题，实则数据形态完全不同，不能互相替代。

再看单纯文本转SQL脚本方案，这个方案也有它的存在价值：执行计划固定、适合跑批任务、结果不依赖模型二轮判断。但它的致命短板是没有反馈回路。脚本一旦生成，哪怕SQL语法完全正确，执行后产生了业务上荒谬的结果，脚本也不会自我修正。SQL Agent则多了一个“执行后观察”的环节，结果不对劲它可以换一个写法，甚至换一张表。

所以这一集的选型结论很明确：需要交互探索、临时性查询、结果要对业务负责的场景，直接上SQL Agent；固定报表、批处理、确定性高的SQL，别用Agent，浪费资源还引入不确定性。

2. 从零搭建SQL Agent：工具包、参数与首跑

2.1 环境准备与数据库连接

先说依赖，我本地用的是Python 3.10，安装这几项：

pip install langchain langchain-community langchain-openai sqlalchemy

数据库连接直接用SQLAlchemy的URI。平常做Demo用SQLite最省事，不用单独启动服务，但在生产环境里建议换成PostgreSQL或同类企业级数据库，因为SQLite并发能力弱，Agent一旦并发起来会把数据库锁死。

from sqlalchemy import create_engine from langchain_community.utilities import SQLDatabase engine = create_engine("sqlite:///demo.db") db = SQLDatabase(engine)

这里有一个关键习惯：连接数据库之后，先把只读会话配置好。比如在PostgreSQL里创建一个专用账号，只授SELECT权限。SQL Agent再聪明，账号层面没有写权限，它就发不出INSERT、UPDATE、DELETE。这属于纵深防御的最外层，后面谈Human-in-the-loop时还会再加一层。

2.2 SQLDatabaseToolkit中四件工具的职责

构建SQL Agent核心用的是SQLDatabaseToolkit，里面默认提供四个工具，每一个都对应Agent在数据库操作中的一个基础动作：

• sql_db_list_tables：列出数据库里的所有表。Agent处理一个新问题之前，通常先调用它了解全局。
• sql_db_schema：查看指定表的建表语句、字段列表、主外键信息。Agent据此判断能用哪些字段做条件、哪些字段做关联。
• sql_db_query：执行SQL语句并返回结果。这是最重要的工具，也是风险最大的工具。
• sql_db_query_checker：对即将执行的SQL做一次合法性检查。它会用模型再读一遍SQL，看有没有语法错误或者明显的不安全模式。

把这四个工具排列组合一下，Agent的典型工作路径就是：list_tables定位目标表，schema确认字段细节，query执行查询，必要时用query_checker反复校正SQL。

from langchain_community.agent_toolkits import SQLDatabaseToolkit toolkit = SQLDatabaseToolkit(db=db, llm=llm) tools = toolkit.get_tools() for tool in tools: print(tool.name, "-", tool.description)

输出里能看到每个工具的name和description，这些描述会被Agent当成工具使用说明书。所以一个小技巧是：如果你不希望Agent轻易碰某类表，可以在include_tables参数里直接限定白名单，从源头减小工具的探索范围。

db = SQLDatabase(engine, include_tables=["orders", "customers", "products"])

这样即使Agent想象力再丰富，它也无法读取被排除的表。

2.3 create_sql_agent的参数选择思路

LangChain提供的create_sql_agent封装得比较高层，核心参数并不多，但每一个都影响行为走向：

• llm：指定大模型。SQL Agent对模型的工具调用能力要求比较高，建议选择函数调用能力强的模型。
• toolkit：上面创建的SQLDatabaseToolkit实例。
• agent_type：常见的是ZERO_SHOT_REACT_DESCRIPTION，让模型通过ReAct循环做思考-行动-观察；如果模型支持函数调用，也可以使用OPENAI_FUNCTIONS。
• verbose：设为True时会把Agent每一步的Thought/Action/Observation打印出来，调试期强烈建议打开。
• max_iterations：限制Agent最多执行多少轮工具调用，防止它陷入死循环。
• max_query_length：限制SQL语句最大长度，防止生成超长恶意SQL。
• max_query_rows：限制查询返回的最大行数，默认通常是几十行，防止一次查询拉爆内存或上下文。

我习惯在调试阶段把max_iterations设成5左右，够用又不至于让失控的Agent反复试探。等压测完再调高。

2.4 首个完整示例：让Agent自动查报表

下面这个例子可以直接跑，我用的是SQLite内存文件：

from langchain_openai import ChatOpenAI from langchain.agents import create_sql_agent from langchain.agents.agent_types import AgentType llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, verbose=False) agent_executor = create_sql_agent( llm=llm, toolkit=toolkit, agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True, max_iterations=5, max_query_rows=30, )

然后提问：

result = agent_executor.invoke("近30天每个客户的下单金额排名前5是谁？")

我在本地看到的效果是：Agent先列出表，再获取客户表和订单表的schema，发现关联键是customer_id，随后构造出带GROUP BY和ORDER BY的SQL，执行后给出客户名单。整个过程没有人工干预，正确性依赖表结构和字段命名是否清晰。如果字段叫c_id、o_id这种缩写，模型就比较容易猜错，建议在字段注释里把业务含义写清楚，或者把schema信息预先放到系统提示词中。

3. Human-in-the-loop的三种落地方式

3.1 方案一：Tool包装拦截（轻量级、上手快）

在AgentExecutor的执行循环里，模型每次决定调用工具，都是调用Tool对象的run方法。那么最简单的Human-in-the-loop就是在工具的run上包一层“审批守卫”，在Agent真正执行SQL之前拦住它，向人类展示SQL内容并等待确认。

def make_approval_wrapper(tool): original_run = tool.run def wrapped_run(sql): # 这里仅示例，生产环境需要更严格的判断 sql_str = sql.strip().lower() if not sql_str.startswith("select"): answer = input(f"[审批] Agent准备执行非只读操作：\n{sql}\n是否批准？(y/n): ") if answer.strip().lower() != "y": return "用户拒绝了本次操作。请停止执行，并向用户解释为什么不能执行。" return original_run(sql) tool.run = wrapped_run return tool

这段代码的核心思想是：任何非SELECT语句都要经过终端人工输入确认。Agent拿到的返回值如果是一个拒绝通知字符串，它就能理解用户的意图并停止行动。这个方法非常适合本地原型、IPython Notebook以及内部人员使用的命令行工具。

它的缺点也很明显：input()是一个阻塞式等待，Agent的进程会一直挂着；它不能在Web服务、异步框架里使用；如果Agent在多个线程里并发运行，审批消息可能混乱。

3.2 方案二：LangGraph状态机实现暂停恢复（生产级）

对于需要部署成服务的场景，我还是推荐用LangGraph把Agent流程改造成状态图。核心思路是把整个执行过程拆成多个节点，在“执行SQL”这个节点之前插入“人工审批”节点，利用持久化Checkpoint保存状态，审批通过后再从断点恢复。

from langgraph.graph import StateGraph, END from typing import TypedDict, List class AgentState(TypedDict): messages: List[str] pending_sql: str approved: bool def model_node(state: AgentState): # 调用大模型决定下一步动作，产出工具调用 return state def approval_node(state: AgentState): sql = state["pending_sql"] if not is_readonly(sql): # 发通知给值班人员，等待审批结果写回 state["approved"] = wait_for_human_approval(sql) if not state["approved"]: state["messages"].append("用户拒绝了操作请求。回复用户无法继续。") return state def execute_sql_node(state: AgentState): if state["approved"]: result = run_sql(state["pending_sql"]) state["messages"].append(f"查询结果：{result}") return state

这套方案的价值在于，Agent不是在进程里傻等，而是通过Checkpoint把状态序列化存储。Web后端收到人工审批结果之后，可以恢复这个状态继续执行。真正做到了“暂停-等待-恢复”的语义。代价是代码复杂度明显上升，需要理解LangGraph的状态管理和持久化机制。

3.3 方案三：异步审批队列（团队协作场景）

更重一点的方案是把人工审批做成一个异步消息队列。当Agent产生非安全SQL时，不等待结果，而是把审批请求发送到队列中，同时挂起这一个任务的状态。值班人员通过内部审批平台查看SQL、表名、影响行数，点击通过或拒绝，结果再通过回调接口回传。

这种方案适合Agent被多个业务方共用、需要审计留痕、甚至需要多人会签的场景。每一次操作都有完整记录，且审批人可以不是开发者本人，而是懂业务的数据管理员。我目前在公司落地的是这个形态，审批记录直接对接内部审计日志。相比前两种方案，它最接近生产环境的要求，但工程量也最大，一般的个人项目不建议一步到位。

3.4 三种方案选型对比

我的建议是：第一版先做方案一，把提示词和拦截逻辑调顺；上了真实业务需要并发，再迁移到方案二；如果公司对审批流程有审计要求，直接上方案三。

4. 完整实操：带审批的SQL Agent从开发到运行

4.1 准备示例数据与库表结构

为了演示，我建了一个迷你销售库，三张表：

CREATE TABLE customers ( id INTEGER PRIMARY KEY, name TEXT, city TEXT, created_at DATE ); CREATE TABLE products ( id INTEGER PRIMARY KEY, name TEXT, price REAL ); CREATE TABLE orders ( id INTEGER PRIMARY KEY, customer_id INTEGER, product_id INTEGER, amount REAL, status TEXT, created_at DATE );

往里插入一些示例数据。注意我给orders表加了status字段，用来模拟“已取消”“已支付”“已发货”这类状态，这样Agent在面对“删除已取消订单”之类的请求时，才会有真实的操作空间。

4.2 审批工具与Agent封装代码

完整代码如下，我加了注释：

import sqlite3 from sqlalchemy import create_engine from langchain_community.utilities import SQLDatabase from langchain_openai import ChatOpenAI from langchain.agents import create_sql_agent from langchain.agents.agent_types import AgentType from langchain_community.agent_toolkits import SQLDatabaseToolkit # 第一步：建库连接 engine = create_engine("sqlite:///demo.db") db = SQLDatabase(engine) # 第二步：实例化大模型 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 第三步：创建Toolkit toolkit = SQLDatabaseToolkit(db=db, llm=llm) # 第四步：给敏感工具加审批 tools = toolkit.get_tools() def is_readonly_sql(sql: str) -> bool: s = sql.strip().lstrip("(").lower() return s.startswith("select") or s.startswith("with") def apply_human_approval(tool): original_run = tool.run def guarded_run(sql: str): if tool.name == "sql_db_query" and not is_readonly_sql(sql): decision = input(f"[人工审批] SQL: {sql}\n放行? (y/n): ") if decision.strip().lower() != "y": return "申请人拒绝了该SQL，必须停止。回复用户：当前操作未经批准，未执行。" return original_run(sql) tool.run = guarded_run return tool tools = [apply_human_approval(t) for t in tools] # 第五步：创建Agent agent = create_sql_agent( llm=llm, toolkit=toolkit, tools=tools, agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True, max_iterations=6, max_query_rows=20, )

这里有个容易踩的坑：你单独对toolkit实例加了包装，但create_sql_agent如果接收独立的tools参数，要确保传入的是包装后的列表。否则Agent拿到的还是原始工具，审批逻辑等于没有生效。我在调试阶段反复确认过这一点。

4.3 运行效果一：只读查询自动放行

输入：

result = agent.invoke("每个城市的客户数量分别是多少？")

我在终端看到的运行日志大概是这样的：

Thought: 先看看有哪些表可用。 Action: sql_db_list_tables Observation: customers, products, orders Thought: 客户数量需要在customers表里按city分组统计。 Action: sql_db_schema Observation: customers(id, name, city, created_at) Action: sql_db_query Observation: [('北京', 35), ('上海', 42), ('广州', 18)]

因为是SELECT查询，审批包装器直接放行，Agent把结果整理成一句人话：“北京35位、上海42位、广州18位”。

4.4 运行效果二：更新操作被拦截

输入：

result = agent.invoke("把今年之前创建但状态为已取消的订单统一标记为已关闭")

Agent会去orders表查询状态相关字段，然后生成类似这样一句SQL：

UPDATE orders SET status = '已关闭' WHERE created_at < '2025-01-01' AND status = '已取消'

此时审批包装器捕获到非SELECT开头，终端停下来等我输入y/n。我刻意输入n，Agent收到的工具返回值是拒绝通知，它的下一步Thought会是“用户拒绝了，我不能执行这个更新”，最终回复用户“操作未经批准，未执行”。

这一步就是Human-in-the-loop的直观价值。你以为它在自动操作，其实每次危险动作背后都有一个人按着闸门。

4.5 运行效果三：多表关联查询的细节处理

再试一个稍微绕的问题：“销售额最高的前三名客户是谁？”这需要customers和orders做JOIN，还要注意status为已取消的订单是否应该排除。Agent如果只看schema，很可能把取消订单也算进去。

我的建议是在系统提示词里补充业务规则。比如：

from langchain_core.prompts import PromptTemplate prompt = PromptTemplate.from_template( "你是一个数据分析Agent。业务规则：统计销售额时只计算status='paid'的订单。\n{input}" )

把这条规则注入之后，Agent生成的SQL会带上WHERE status = 'paid'。这类业务知识是模型无法靠推理自己获得的，必须由人来提前注入。这也是Human-in-the-loop的另一种形态：人在系统设计时介入，而不是等到执行时再介入。

5. 常见问题与排查技巧实录

5.1 Agent反复调用工具或死循环

症状：日志里Agent不停调用同一个查询工具，或者反复执行同一个SQL，明显陷入循环。

排查思路：循环通常是模型没有得到足够信息去判断“已经完成”。比如表很多但Agent没有充分浏览，或字段描述不清晰导致SQL反复不过校验。先看verbose日志中最后一次Observation返回什么，再调整。

解决建议：

• 把max_iterations调小，比如3到5，强制它收敛。
• 在prompt里要求Agent“每次调用工具后必须记录结论”。
• 数据库表非常多时，用include_tables过滤掉无关表。

5.2 返回结果太大把上下文塞爆

症状：Agent执行一个不带LIMIT的查询，返回几千行数据，上下文窗口被撑爆，报错或者后续推理明显变差。

解决步骤：

• 设置max_query_rows，让它默认加LIMIT。
• 在prompt里提醒模型：所有查询都必须LIMIT 100。
• 查询结果传给下一轮前，可以先做摘要化处理，比如只保留前20行。

LangChain这个限制只在工具层控制，不是绝对安全。如果Agent直接调用底层的database API绕过工具，那拦不住。所以在设计上要约束Agent只能使用Toolkit暴露的方法，别把原生Session给它。

5.3 审批结果没有回传给模型

症状：人工在终端输入n之后，Agent仍然继续执行原始SQL，或者表现得很困惑。

大多数时候是包装函数返回的数据格式有问题。LangChain的Tool.run返回值如果是空字符串、None，或者不带明确的语义，模型就无法理解发生了什么。审批被拒绝时，一定要返回一段完整的、指令明确的字符串，让它能读明白当前状态。

5.4 数据库连接数打满

症状：Agent运行一段时间后，数据库连接超时或被拒绝。

Agent每次调用SQL工具都可能新建连接，如果没有复用连接池，并发一高连接数就爆了。SQLAlchemy默认有连接池，但你要确保在工具内部使用的是同一个engine实例，而不是每次新建。另外，Agent本身是串行执行工具的，不需要大量并发连接。真要做并发，建议把查询放到队列里逐个消费。

5.5 权限设计建议速查表

这五层是配合关系，不是替代关系。单靠任何一层都不够，组合起来才是一个能上生产环境的SQL Agent。

我个人现在的习惯是：任何接数据库的Agent，第一版就用最小权限账号加工具层白名单上线，等跑通之后再把审批层加上。别一上来就上一整套重方案，否则你会发现，连Agent正常跑通一次查询都要花半天时间调试权限。把最轻的防线先立起来，后面逐步加厚，这才是SQL Agent接入业务的最舒服路径。