SQL Agent 是 LangChain 生态里最常被问到的落地场景之一而 Human-in-the-loop 又是把 SQL Agent 从“玩具”变成“能上生产”的关键一步。这套组合解决的是同一个痛点大模型生成 SQL 的能力已经够用但谁也不敢让一个概率模型在没有监督的情况下直接对线上库执行 DELETE、UPDATE 或者 DROP。第二十集我就围绕 SQL Agent Human-in-the-loop 做一次完整拆解从设计思路到两条实现路径再到一套可以直接抄作业的 Demo。这一集不依赖前面任何一集即使你第一次接触这两个概念也能照着落地。先说结论真正适合生产的 SQL Agent不是让模型自己“想干嘛干嘛”而是让它“能查、能写、但动手前必须有人点头”。下面我把为什么需要人工节点、两条主流实现路线怎么选、核心代码怎么写、以及我实际踩过的坑一次讲清楚。1. 为什么 SQL Agent 需要“人在回路”1.1 全自动 SQL Agent 的翻车现场早几年大家追求“全自动数据问答”最理想的画面是业务同学在对话框里打一句“帮我查一下上个月华东区的销售额”Agent 自己写 SQL、自己执行、自己返回结果。这个画面在受限的只读库、固定报表库上确实能跑通但一旦把 Agent 的数据库权限从 SELECT 放开到 INSERT、UPDATE、DELETE画风就完全变了。我自己见过最典型的翻车案例是模型被诱导写出一条没有 WHERE 条件的 UPDATE直接把整张表的某个状态字段刷成了同一个值。还有一次Agent 为了让查询“看起来更完整”连续做了三次大的表关联扫描把业务库的慢查询日志直接打满。这些问题的根源不是 SQL 语法能力不够而是大模型缺少两个东西第一它不知道这条 SQL 会影响到多少行第二它不知道这个数据库背后的业务价值。DELETE 一张配置表和 DELETE 一张订单表在模型眼里都是“执行一条语句”在业务眼里却是天壤之别。所以生产环境的 SQL Agent默认权限应该永远是“最小够用”同时把所有写操作、危险查操作都纳入人工确认通道。这就是 Human-in-the-loop 存在的意义让机器负责效率和生成质量让人负责风险判断和最终拍板。它不是一种可选的“加强功能”而是从全自动退回到半自动的必要妥协。1.2 哪些场景必须引入人工审批不是所有 SQL 都需要人工介入否则 Agent 就没有存在价值了。我习惯把数据库操作分成三类分别对应不同的管控策略。第一类是“安全读操作”包括纯 SELECT、只读视图查询、EXPLAIN 等。这类操作让 Agent 全自动执行最多加一个返回行数限制防止有人写出全表扫描。第二类是“敏感读操作”典型代表是查询包含手机号、身份证号、薪资字段的明细数据。这类操作虽然不修改数据但存在数据泄露风险需要审批后才能执行或脱敏后返回。第三类是“写操作”包括 INSERT、UPDATE、DELETE以及 DDL 里的 DROP、ALTER、TRUNCATE。这类操作必须走后置审批也就是说 Agent 先生成 SQL把 SQL 和影响范围预判一起提交给人工同意之后才真正执行。硬性要求我建议这样定任何 DDL 和任何不带 WHERE 条件的 UPDATE/DELETE 直接禁止自动执行带 WHERE 的写操作也必须人工确认影响行数。这个规则不需要模型理解直接在代码层用 SQL 解析和关键词匹配做兜底比靠模型自觉可靠得多。2. 方案设计与技术选型2.1 两条实现路线的对比实现 SQL Agent Human-in-the-loop目前主流有两套方案。第一套是“工具层包装”在 LangChain 的工具定义里做文章把每个写操作封装成一个自定义 ToolTool 内部先发起人工审批审批通过后再真正调用数据库执行器。第二套是“流程层编排”用 LangGraph 把 Agent 的“思考、生成 SQL、人工审批、执行、生成回答”拆成显式的状态节点在审批节点用 graph 的中断机制把流程挂起等人回传结果后续跑。两套方案我都分别在生产项目里试过。工具层包装实现简单不需要引入额外的状态图框架适合审批逻辑比较单一、一个问题只有一个写操作、团队对 LangGraph 不熟的场景。但它有硬伤模型可能绕过被包装的工具直接去调用底层原始的 SQL 执行工具因为 Tool 列表里只要存在一个“裸”执行器模型就有概率选择更顺手的那个。流程层编排则把每个决策点变成图中可见的节点执行顺序由图结构强制约束不依赖模型自觉安全性高一个量级。我的选型建议很明确如果是给内部业务用的工具工具层包装够用如果 Agent 要同时服务多条业务线、对接线上核心库直接上 LangGraph 做流程编排不要犹豫。对比维度工具层包装LangGraph 流程编排实现门槛低只需要写自定义 Tool中高需要理解状态图和中断机制审批控制力依赖模型是否调用包装后的工具图结构强制模型无法绕过多步写操作审批困难容易出现多次审批容易可以集中审批或逐节点审批支持人工反馈上下文较弱强可以携带反馈继续生成上线风险中等低2.2 环境准备与版本约束我下面所有代码基于 Python 3.10 以上环境依赖版本用的是 langchain 0.3.x、langchain-community 0.3.x、langchain-openai 0.3.x、langgraph 0.2.x、sqlalchemy 2.x。为什么要特别强调版本因为 LangChain 的 API 变化非常快早先的SQLDatabaseToolkit里不少工具签名在 0.3 里已经调整过如果你本地还是 0.1.x 的老项目照抄代码大概率跑不通。安装依赖用一条命令搞定pip install langchain langchain-community langchain-openai langgraph sqlalchemy模型接口我下面统一用 ChatOpenAI 类名举例实际使用中你完全可以用 ChatAnthropic、ChatOllama、或者任何 OpenAI 协议兼容的推理服务只需要把base_url和api_key换成你自己的。有一点要特别提醒SQL Agent 的模型不要开高温度temperature0是必须的。生成 SQL 不是创意写作模型需要的是确定性和可复现性温度一高同一个问题两次生成的 SQL 可能完全不同人工审批的成本会被无限放大。2.3 演示库与风险规则设计为了不依赖外部数据库我直接用 SQLite 做演示。SQLite 的好处是零部署适合做快速原型坏处是它和线上 MySQL/PostgreSQL 在函数、锁机制、权限粒度上都有差异所以 Demo 归 Demo生产环境一定要用目标数据库的真实方言重新测一遍。演示库我设计成一张订单表import sqlite3 conn sqlite3.connect(order_demo.db) cur conn.cursor() cur.execute( CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY, customer_name TEXT, product_name TEXT, amount REAL, status TEXT, created_at TEXT ) ) cur.execute(DELETE FROM orders) cur.executemany( INSERT INTO orders (customer_name, product_name, amount, status, created_at) VALUES (?, ?, ?, ?, ?, ?), [ (客户A, 机械键盘, 299.0, paid, 2025-01-10), (客户B, 显示器, 1299.0, unpaid, 2025-01-12), (客户C, 鼠标, 59.5, refunded, 2025-01-15), (客户A, 耳机, 399.0, paid, 2025-02-01), (客户D, 机械键盘, 299.0, unpaid, 2025-02-03), ], ) conn.commit() conn.close()风险规则我定义了三条都是代码级兜底不依赖模型判断第一SQL 中以INSERT、UPDATE、DELETE、ALTER、DROP、TRUNCATE开头的语句一律算写操作必须进审批第二UPDATE和DELETE语句如果没解析出 WHERE 条件直接拦截拒绝不给审批机会第三SELECT 查询显式加上 LIMIT 100 的默认限制防止一次拉全表。3. 核心细节两种人工审批通道的实现3.1 通道一在工具层做审批包装工具层包装的核心思路是给模型暴露一个“写操作执行器”但这个执行器内置了审批流程。模型要想执行写操作只能走这个工具工具第一步先打印 SQL 和审批原因等人输入 y 或 n再决定是否放行。from typing import Type from langchain_core.tools import BaseTool from pydantic import BaseModel, Field class WriteApproveInput(BaseModel): sql: str Field(description需要人工审批的 SQL 语句) reason: str Field(description触发人工审批的原因) class WriteApproveTool(BaseTool): name: str write_approval description: str ( 当 SQL 是 INSERT/UPDATE/DELETE/ALTER/DROP/TRUNCATE 等写操作时 必须调用本工具获批后才能执行。 ) args_schema: Type[BaseModel] WriteApproveInput def _run(self, sql: str, reason: str) - str: print(f[审批请求] 原因: {reason}) print(f[审批请求] SQL: {sql}) decision input(批准请输入 y拒绝请输入 n: ) if decision.strip().lower() y: return APPROVED return REJECTED这段代码在本地跑起来效果很直观模型一旦想执行写操作终端就会停下来等输入。但这里有一个必须提醒的坑你不能只给模型这一个工具还必须提供一个真正的只读查询工具。否则模型发现自己写不了 SQL会直接告诉你“权限不足”而不是先去申请审批。正确的做法是给模型两个工具一个只读查询工具一个上面这个审批工具两个工具的描述里写清楚各自的使用边界。另一个容易被忽略的细节是args_schema。如果你不定义 Pydantic 入参模型LangChain 默认会从_run函数的签名里推断参数这在小工具上没问题但一旦参数多了模型生成的 JSON 经常和你的函数签名对不上。显式定义args_schema等于告诉模型和解释器“我只要两个字段别乱发挥”实测能显著减少参数解析失败的比例。3.2 通道二用 LangGraph 状态机实现暂停与恢复工具层包装虽然简单但我在真实项目中还是碰到了一个绕不开的问题审批通过之后模型往往忘记了刚才审批的是哪条 SQL或者因为中间对话过长上下文里出现了多条 SQL模型不知道该执行哪一条。原因很好理解工具调用只是对话中的一个环节模型拿到“APPROVED”返回值时并不保证它还记得最初生成的那条 SQL 全文。LangGraph 方案把这个问题从根上解决了。它的核心思路是把“生成 SQL”和“执行 SQL”拆成两个独立节点中间插入一个人工审批节点节点之间通过显式的状态字段传递数据。这样审批节点拿到的 SQL 来自上游状态执行节点读取的也是同一份状态根本不依赖模型记忆。from typing import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import interrupt, Command from langgraph.checkpoint.memory import InMemorySaver class AgentState(TypedDict): question: str sql_plan: str write_sql: str approved: bool answer: str def decide_node(state: AgentState): # 真实项目中这里是大模型根据问题生成SQL演示用规则替代 if delete in state[sql_plan].lower() or update in state[sql_plan].lower(): return {write_sql: state[sql_plan]} return {write_sql: } def human_approval(state: AgentState): result interrupt({ sql: state[write_sql], question: state[question], }) return {approved: result.get(approve, False)} def run_execute(state: AgentState): if state[approved]: return {answer: SQL 已批准并执行} return {answer: 用户拒绝执行数据库无任何变更} builder StateGraph(AgentState) builder.add_node(decide, decide_node) builder.add_node(approval, human_approval) builder.add_node(execute, run_execute) builder.add_edge(START, decide) builder.add_conditional_edges( decide, lambda s: approval if s[write_sql] else execute, {approval: approval, execute: execute}, ) builder.add_edge(approval, execute) builder.add_edge(execute, END) graph builder.compile(checkpointerInMemorySaver())这里最关键的函数就是interrupt。它做的事情和 Python 的input()类似但更强大它会把整个图的计算状态完整地持久化到 checkpointer 里然后抛出一个中断。这时候图的状态相当于被冻结了你可以把这段状态序列化发到前端、IM、邮件让真实的人类在任何地方审批而不是傻等终端输入。3.3 审批结果如何安全回写到对话流人工审批完结果怎么传回给被冻结的图LangGraph 用的是Command(resume...)。你不需要重新调用整条图只要拿着之前的线程 ID把审批结果塞回去即可。config {configurable: {thread_id: order-demo-001}} # 第一次调用流程会停在 human_approval 的 interrupt 位置 result graph.invoke( { question: 把客户A的订单状态改成paid, sql_plan: UPDATE orders SET statuspaid WHERE customer_name客户A, }, config, ) # 人工确认后回传审批结果 result graph.invoke(Command(resume{approve: True}), config)这段代码里我特意用了thread_id来区分不同的会话。这是并发场景的生命线如果没有 thread_id两个用户同时发起审批请求状态就会串线。有了它每个人工审批请求都对应一个独立的状态通道恢复时各回各家。回传结果时还有一个容易被忽视的坑Command(resume...)只能传一次。如果你在审批界面点了两次“通过”第二次 resuming 会报错或者覆盖状态。所以在开发审批前端时按钮要加状态锁提交成功后按钮必须变成“已处理”。4. 完整实操把 SQL Agent 改造成可审批的半自动系统4.1 初始化数据库与模型前面已经建好 order_demo.db下一步是初始化数据库连接和模型。SQLDatabase 是从 LangChain 社区库里拿的标准封装它会自动读取表结构、字段名、主键信息这些元数据后续会拼进 Agent 的系统提示词里是模型写 SQL 的基础。from langchain_community.utilities import SQLDatabase from langchain_openai import ChatOpenAI db SQLDatabase.from_uri( sqlite:///order_demo.db, sample_rows_in_table_info3, ) llm ChatOpenAI( modelgpt-4o-mini, temperature0, max_tokens1024, )sample_rows_in_table_info3这个参数值得多说一句。它会让 Agent 在提示词里看到每张表的前三行真实样例数据。有了样例模型对字段格式的把握会准很多比如它会知道status字段用的是 paid/unpaid/refunded 而不是 0/1知道created_at的格式是 YYYY-MM-DD。但这个东西不是越多越好样例行数设太大会导致提示词超长、Token 费用飙升三行是一个比较均衡的取值。4.2 完整可运行的审批 Agent 代码下面这段代码是把前面所有片段拼起来的完整版。它包含了只读工具、写审批工具、状态图定义以及主流程调用。为了控制篇幅我略去了decide_node内部真正调用大模型生成 SQL 的细节而是用规则代替你接入自己的模型时把那段换成llm.invoke()即可。from langchain_core.tools import tool from langgraph.graph import StateGraph, START, END from langgraph.types import interrupt, Command from langgraph.checkpoint.memory import InMemorySaver from typing import TypedDict tool def query_database(sql: str) - str: 只读查询只能执行 SELECT 语句。 if not sql.strip().lower().startswith(select): return 只允许 SELECT 查询 return str(db.run(sql)) class AgentState(TypedDict): question: str sql_plan: str write_sql: str approved: bool answer: str def generate_sql_node(state: AgentState): # 生产环境用 llm.invoke 结合 db.get_table_info() 生成 sql_plan # 演示环境从 question 里暴力提取关键词生成 UPDATE question state[question] if UPDATE in question.upper(): sql question.split(SQL:)[-1].strip() return {sql_plan: sql} return {sql_plan: } def decide_node(state: AgentState): sql state[sql_plan] lowered sql.lower().strip() if lowered.startswith((insert, update, delete, alter, drop, truncate)): return {write_sql: sql} return {write_sql: } def human_approval(state: AgentState): confirm interrupt({ sql: state[write_sql], question: state[question], }) approved confirm.get(approve, False) return {approved: approved} def execute_node(state: AgentState): if state[write_sql]: if state[approved]: db.run(state[write_sql]) return {answer: f已批准并执行: {state[write_sql]}} return {answer: 已拒绝执行数据库无变更} result query_database.invoke(state[sql_plan]) return {answer: result} builder StateGraph(AgentState) builder.add_node(generate_sql, generate_sql_node) builder.add_node(decide, decide_node) builder.add_node(approval, human_approval) builder.add_node(execute, execute_node) builder.add_edge(START, generate_sql) builder.add_edge(generate_sql, decide) builder.add_conditional_edges( decide, lambda s: approval if s[write_sql] else execute, {approval: approval, execute: execute}, ) builder.add_edge(approval, execute) builder.add_edge(execute, END) graph builder.compile(checkpointerInMemorySaver()) config {configurable: {thread_id: demo-001}} first graph.invoke( {question: 把客户A的订单状态改成paid SQL: UPDATE orders SET statuspaid WHERE customer_name客户A}, config, ) # 到这里流程会停住等待审批 next_output graph.invoke(Command(resume{approve: True}), config) print(next_output[answer])这段代码你可以直接跑感受一下审批前后流程的变化。第一次invoke会进入 interrupt 挂起第二次invoke带着Command(resume{approve: True})恢复会把approved置为 True然后执行节点才真正碰数据库。4.3 关键参数的选择逻辑跑通 Demo 之后有三处参数值得你根据业务自己调。第一是模型温度前面已经说了必须 0不重复。第二是max_tokens我给它设了 1024不要太小。因为模型生成的可能不是一条 SQL 而是一段带解释的自然语言加 SQLToken 太小会把 SQL 截断产生半条语句比不生成还危险。第三是图执行时的迭代上限用 LangGraph 时虽然没有 LangChain AgentExecutor 的max_iterations参数但你仍然可以在节点内部对大模型生成 SQL 的调用次数做计数超过三次就强制走人工通道防止模型反复生成失败死循环。还有一个经验值SQL Agent 的回答里要强制模型带上“影响行数预估”。怎么拿到这个值最土也最稳的办法是在审批前先在只读事务里执行一次等价的 COUNT 查询。比如要执行UPDATE orders SET statuspaid WHERE customer_name客户A就先跑SELECT COUNT(*) FROM orders WHERE customer_name客户A把统计结果和 SQL 一起发给审批人。人工审批者看到“将影响 1 行”签字的底气完全不一样。这也是我在真实系统里强烈建议加的一个环节。5. 常见问题与排查技巧实录5.1 Agent 绕过审批直接执行写操作这是工具层方案最容易踩的坑症状是审批流程写得挺完整但实际生产日志里发现写操作是直接从某个底层 SQL 工具执行的根本没经过审批节点。原因我刚才提过模型在 Tool 列表里看到了不止一个能执行 SQL 的工具它没有义务一定选中你包装好的那个。解决办法有三个。第一是做工具隔离只读执行器和写执行器不要同时暴露给模型写执行器必须经过审批工具第二是行为审计在数据库侧开启全量 SQL 审计日志写操作如果来自非审批通道直接告警第三是干脆放弃工具层方案改用 LangGraph 的图结构强约束。我个人测试下来的结论是方案三最干净因为你不再依赖“模型愿意选哪个工具”而是从架构上就把“不经过审批就无法执行写操作”变成了事实。5.2 人工审批后上下文丢失另一个高频问题审批通过了但是图节点拿不到之前生成的 SQL或者拿到的是空值。排查重点先看状态字段命名。LangGraph 的状态字段有覆盖和叠加两种语义默认是覆盖语义即上一个节点返回什么下一个节点读到的就是什么。如果你的decide_node返回的字段名是write_sql但execute_node读的是exec_sql那一定是空。命名保持一致是最笨也最有效的防呆手段。还有一类的“上下文丢失”其实是 checkpointer 没配置。没有配置 checkpointer 的图虽然也能跑但interrupt无法持久化第二次Command(resume...)时找不到之前的图状态直接报错。解决方法是graph.compile(checkpointerInMemorySaver())这一步不能省。生产环境建议把InMemorySaver换成基于数据库的持久化实现否则服务一重启所有冻结状态全部丢失。5.3 并发会话审批状态串线我用thread_id区分会话一开始觉得万无一失直到有一次测试发现两个还没审批的流程互相拿到了对方的审批结果。查到最后原因是前端在恢复流程时把thread_id当成了全局变量A 用户的操作覆盖了 B 用户的配置。这个问题的根子不在 LangGraph而在应用层。你要保证一个原则thread_id必须随会话上下文传递而不是从某个全局变量读取。另外恢复流程的Command(resume...)一定要和原线程绑定前端展示审批单列表时每条审批单都要携带自己的thread_id和interrupt值不要只看一眼 SQL 就提交。5.4 问题速查表问题现象常见原因解决动作审批流程没触发Tool 列表里有裸执行工具隐藏裸执行工具只暴露审批入口审批后 SQL 为空状态字段命名不一致统一 write_sql 字段名全局搜索检查第二次恢复报错checkpointer 未配置compile 时传入 InMemorySaver 或持久化版本A 用户审批影响 B 用户thread_id 用了全局变量把 thread_id 绑定到每个会话上下文审批界面重复提交resume 不能重复调用前端加按钮状态锁成功后禁用模型不写 SQL 却说没权限模型只看到了审批工具没看到查询工具同时暴露只读查询工具和写审批工具查询结果一次拉超大SELECT 没有 LIMIT代码层强制追加 LIMIT 100UPDATE 无 WHERE 被审批人工审批也能过代码硬拦截无 WHERE 直接拒绝不进审批最后再分享一点个人体会SQL Agent Human-in-the-loop 的技术难点其实不在模型而在“审批体验”。如果审批界面只有一串 SQL没有任何上下文解释和影响行数预估人工根本审不下去最后要么全部点同意要么全部点拒绝流程形同虚设。我踩过几次坑之后现在做审批卡片时一定会带上三样东西原始问题、生成的 SQL、影响行数预估。审批人一眼就能判断“这是不是用户想问的、这条 SQL 会不会出事故”。把这三样东西做好比换更强的大模型更能提升 SQL Agent 的可用性。