企业官网建设流程全解析

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 的工具定义里做文章：把每个写操作封装成一个自定义 Tool，Tool 内部先发起人工审批，审批通过后再真正调用数据库执行器。第二套是“流程层编排”，用 LangGraph 把 Agent 的“思考、生成 SQL、人工审批、执行、生成回答”拆成显式的状态节点，在审批节点用 graph 的中断机制把流程挂起，等人回传结果后续跑。

两套方案我都分别在生产项目里试过。工具层包装实现简单，不需要引入额外的状态图框架，适合审批逻辑比较单一、一个问题只有一个写操作、团队对 LangGraph 不熟的场景。但它有硬伤：模型可能绕过被包装的工具，直接去调用底层原始的 SQL 执行工具，因为 Tool 列表里只要存在一个“裸”执行器，模型就有概率选择更顺手的那个。流程层编排则把每个决策点变成图中可见的节点，执行顺序由图结构强制约束，不依赖模型自觉，安全性高一个量级。

我的选型建议很明确：如果是给内部业务用的工具，工具层包装够用；如果 Agent 要同时服务多条业务线、对接线上核心库，直接上 LangGraph 做流程编排，不要犹豫。

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 的模型不要开高温度，temperature=0是必须的。生成 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(checkpointer=InMemorySaver())

这里最关键的函数就是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 status='paid' 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_info=3, ) llm = ChatOpenAI( model="gpt-4o-mini", temperature=0, max_tokens=1024, )

sample_rows_in_table_info=3这个参数值得多说一句。它会让 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(checkpointer=InMemorySaver()) config = {"configurable": {"thread_id": "demo-001"}} first = graph.invoke( {"question": "把客户A的订单状态改成paid SQL: UPDATE orders SET status='paid' 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 而是一段带解释的自然语言加 SQL，Token 太小会把 SQL 截断，产生半条语句，比不生成还危险。第三是图执行时的迭代上限，用 LangGraph 时虽然没有 LangChain AgentExecutor 的max_iterations参数，但你仍然可以在节点内部对大模型生成 SQL 的调用次数做计数，超过三次就强制走人工通道，防止模型反复生成失败死循环。

还有一个经验值：SQL Agent 的回答里要强制模型带上“影响行数预估”。怎么拿到这个值？最土也最稳的办法是，在审批前先在只读事务里执行一次等价的 COUNT 查询。比如要执行UPDATE orders SET status='paid' 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(checkpointer=InMemorySaver())，这一步不能省。生产环境建议把InMemorySaver换成基于数据库的持久化实现，否则服务一重启，所有冻结状态全部丢失。

5.3 并发会话审批状态串线

我用thread_id区分会话一开始觉得万无一失，直到有一次测试发现两个还没审批的流程互相拿到了对方的审批结果。查到最后原因是：前端在恢复流程时，把thread_id当成了全局变量，A 用户的操作覆盖了 B 用户的配置。

这个问题的根子不在 LangGraph，而在应用层。你要保证一个原则：thread_id必须随会话上下文传递，而不是从某个全局变量读取。另外，恢复流程的Command(resume=...)一定要和原线程绑定，前端展示审批单列表时，每条审批单都要携带自己的thread_id和interrupt值，不要只看一眼 SQL 就提交。

5.4 问题速查表

最后再分享一点个人体会：SQL Agent + Human-in-the-loop 的技术难点其实不在模型，而在“审批体验”。如果审批界面只有一串 SQL，没有任何上下文解释和影响行数预估，人工根本审不下去，最后要么全部点同意，要么全部点拒绝，流程形同虚设。我踩过几次坑之后，现在做审批卡片时一定会带上三样东西：原始问题、生成的 SQL、影响行数预估。审批人一眼就能判断“这是不是用户想问的、这条 SQL 会不会出事故”。把这三样东西做好，比换更强的大模型更能提升 SQL Agent 的可用性。