1. 为什么你的 Agent 需要一个审批阀Human-in-the-LoopHITL说白了就是Agent 在动手之前先停下来问人一句“我要这么干行不行”。它解决的是 Agent 太自主带来的失控问题——Agent 说“我已经帮你把邮件发给所有客户了”你根本没让它发Agent 说“已删除数据库里的临时表”那张表是生产环境的Agent 说“已转账 50000 元”你本意是转 5000。这些不是段子是真实会发生的场景。Agent Harness 是包裹在 LLM 外面的那层执行骨架负责调度工具、管理状态、控制循环。interrupt 机制则是 Harness 里的“暂停键”当 Agent 准备调用一个敏感工具时Harness 不直接执行而是把这次调用挂起序列化当前状态等人类给出批准、拒绝或修改参数的指令后再从断点恢复执行。这套机制适合谁适合正在从零搭 Agent 骨架、已经跑通“LLM→工具→LLM”基础循环、但发现 Agent 会乱调工具的开发者。如果你还在用最原始的 while 循环调工具没有状态持久化那 interrupt 会很难做——因为中断后状态丢了就恢复不了。所以前置条件是你的 Harness 得有一个可序列化的 State以及一个 Checkpointer 来保存断点。我试过直接在工具节点里塞 interrupt结果一次调三个工具就弹三次确认框用户体验极差。后来改成“审批队列”模式先把所有待审批项收集起来统一中断一次用户批量决策后再恢复。下面就从零把这套骨架搭出来。2. TaoToken 前置把模型接入层先跑通在写 interrupt 逻辑之前得先保证模型调用是通的。Agent Harness 里所有 LLM 调用都走同一个接入层这样后面加审批阀时不用改模型代码。我用 TaoToken 做统一入口它兼容 OpenAI 风格的接口换模型只改一个 model 字符串。先拿 API Key打开 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 登录后创建一个 Key复制出来。注意 Key 只在创建时显示一次丢了就重新建。然后配置环境变量别把 Key 硬编码进代码export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api如果你用 Python安装依赖pip install langgraph langchain langchain-openai初始化模型时指向 TaoToken 的 base_urlfrom langchain.chat_models import init_chat_model import os model init_chat_model( gpt-4o-mini, api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], temperature0, )这里有个坑init_chat_model默认走 OpenAI 官方地址必须显式传base_url否则会报 401。传了之后模型调用就走 TaoToken 的通道后面所有工具绑定、流式输出都基于这个 model 对象。想先验证模型通不通可以直接在模型对话页发一条消息测试 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。能正常回复就说明 Key 和地址没问题再往下写 Harness。3. 可复制配置审批队列 interrupt 骨架3.1 先理解原生 interrupt 的局限LangGraph 原生interrupt()长这样def review_node(state): approved interrupt({type: approval, action: state[action]}) if not approved: return {status: rejected} return {status: approved}问题在于一次只能中断一个点。如果 Agent 同时调了两个工具第一个工具触发 interrupt第二个工具的审批得等第一个恢复后才能开始。而且 interrupt 的 value 是一次性的resume 后就没了循环里多次中断很难管理。用户端体验是“弹一个框→确认→又弹一个框→又确认”三个工具就要点三次。我们想要的是审批队列Agent 准备“发邮件给张三、删除临时文件、备份数据库”队列里一次性列出三项用户批准“发邮件”、拒绝“删除文件”、修改“备份时间”然后 Agent 继续执行批准和修改后的项跳过被拒绝的。3.2 审批队列的数据结构核心思路是不在工具节点里 interrupt而是在工具节点前收集“待审批项”统一中断统一审批。from dataclasses import dataclass, field import uuid, json dataclass class ApprovalItem: id: str field(default_factorylambda: str(uuid.uuid4())[:8]) tool_name: str args: dict None summary: str status: str pending # pending | approved | rejected | modified modified_args: dict None staticmethod def from_tool_call(tc) - ApprovalItem: args tc.get(args, {}) summary f{tc[name]}({json.dumps(args, ensure_asciiFalse)}) return ApprovalItem(tool_nametc[name], argsargs, summarysummary) dataclass class ApprovalQueue: items: list[ApprovalItem] field(default_factorylist) def add(self, item: ApprovalItem): self.items.append(item) def approve(self, item_id: str): for item in self.items: if item.id item_id: item.status approved def reject(self, item_id: str): for item in self.items: if item.id item_id: item.status rejected def modify(self, item_id: str, new_args: dict): for item in self.items: if item.id item_id: item.status modified item.modified_args new_args def get_approved_tool_calls(self, original_tool_calls: list) - list: approved [] approved_ids {i.id for i in self.items if i.status in (approved, modified)} modified {i.id: i.modified_args for i in self.items if i.status modified} for tc in original_tool_calls: item_id self._find_item_id(tc) if item_id in approved_ids: tc_copy dict(tc) if item_id in modified: tc_copy[args] modified[item_id] approved.append(tc_copy) return approved def _find_item_id(self, tc) - str: for item in self.items: if item.tool_name tc[name] and item.args tc.get(args): return item.id return def has_pending(self) - bool: return any(i.status pending for i in self.items)3.3 State 里加审批字段Harness 的 State 需要新增几个字段来承载审批数据from typing import Annotated, TypedDict from langgraph.graph import add_messages class StandardState(TypedDict): messages: Annotated[list, add_messages] iteration_count: int pending_approvals: list[dict] # 待审批项列表 approved_tool_calls: list[dict] # 已批准的工具调用 needs_review: bool # 是否需要审批 subtask_tool_calls: list[dict] # Subgraph 提交的待审批调用3.4 图结构collect → approval → tools关键变化是 interrupt 不挂在工具节点而是挂在一个专门的审批节点上。审批节点只做审批不执行工具。审批通过的工具调用才传给 tool_node。from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import interrupt, Command def build_graph(harness): builder StateGraph(StandardState) builder.add_node(llm, harness._llm_node) builder.add_node(collect_approvals, harness._collect_approvals_node) builder.add_node(approval, harness._approval_node) builder.add_node(tools, harness._approved_tool_node) builder.add_edge(START, llm) builder.add_conditional_edges(llm, harness._hitl_route, { collect_approvals: collect_approvals, tools: tools, END: END, }) builder.add_edge(collect_approvals, approval) builder.add_edge(approval, tools) builder.add_edge(tools, llm) return builder.compile( checkpointerInMemorySaver(), interrupt_before[approval], # 在 approval 节点前中断 )interrupt_before[approval]是核心图执行到 approval 节点前会暂停把当前 State 存进 checkpointer等外部用Command(resume...)恢复。3.5 收集节点与审批节点收集节点把 LLM 输出的所有 tool_calls 转成审批项安全工具自动放行SAFE_TOOLS {get_time, calculate, echo} def _collect_approvals_node(self, state): last state[messages][-1] if not last.tool_calls: return {} queue ApprovalQueue() for tc in last.tool_calls: if tc[name] in self.safe_tools: continue # 安全工具自动放行 queue.add(ApprovalItem.from_tool_call(tc)) if not queue.has_pending(): return {approved_tool_calls: last.tool_calls} print(f待审批 {len(queue.items)} 项) return { pending_approvals: [i.to_dict() for i in queue.items], needs_review: True, }审批节点用interrupt()挂起等用户决策def _approval_node(self, state): pending state.get(pending_approvals, []) if not pending: return {} decisions interrupt({ type: approval_queue, items: pending, message: 请审批以下工具调用, }) queue ApprovalQueue() for p in pending: queue.items.append(ApprovalItem( idp[id], tool_namep[tool_name], argsp[args], summaryp.get(summary, ), )) if isinstance(decisions, list): for d in decisions: action d.get(action, reject) item_id d.get(id, ) if action approve: queue.approve(item_id) elif action reject: queue.reject(item_id) elif action modify: queue.modify(item_id, d.get(args, {})) last state[messages][-1] approved queue.get_approved_tool_calls(last.tool_calls) return {approved_tool_calls: approved, needs_review: False}工具节点只执行已批准的调用def _approved_tool_node(self, state): approved state.get(approved_tool_calls, []) if not approved: return {messages: []} results [] for tc in approved: fn self.tools_by_name.get(tc[name]) if fn: try: results.append(ToolMessage( contentstr(fn.invoke(tc[args])), tool_call_idtc.get(id, ), )) except Exception as e: results.append(ToolMessage( contentf工具错误: {e}, tool_call_idtc.get(id, ), )) return {messages: results, approved_tool_calls: []}4. 验证请求一次通过、一次拒绝4.1 定义两个敏感工具from langchain.tools import tool tool def send_email(to: str, content: str) - str: 给指定地址发送邮件 return f已发送邮件至 {to}: {content[:20]}... tool def transfer_money(to: str, amount: float) - str: 向指定账户转账 return f已向 {to} 转账 ¥{amount}4.2 场景一审批通过用户输入“给张三发邮件说你好再给李四转 500 元”。Agent 会调两个工具都进审批队列。agent MiniHarness( configHarnessConfig(enable_human_reviewTrue), tools[send_email, transfer_money], ) # 第一次 invoke会在 approval 节点前中断 result agent.graph.invoke( {messages: [(user, 给张三发邮件说你好再给李四转500元)]}, {configurable: {thread_id: t1}}, )此时图停在 approval 前pending_approvals里有两项。用get_state查看state agent.graph.get_state({configurable: {thread_id: t1}}) print(state.values[pending_approvals]) # [{id: a1b2, tool_name: send_email, args: {to: 张三, content: 你好}, status: pending}, # {id: c3d4, tool_name: transfer_money, args: {to: 李四, amount: 500}, status: pending}]用户批准两项用Command(resume...)恢复resume Command(resume[ {id: a1b2, action: approve}, {id: c3d4, action: approve}, ]) result agent.graph.invoke(resume, {configurable: {thread_id: t1}}) print(result[messages][-1].content) # 已发送邮件至 张三: 你好... 已向 李四 转账 ¥5004.3 场景二审批拒绝同样输入这次拒绝转账result agent.graph.invoke( {messages: [(user, 给张三发邮件说你好再给李四转500元)]}, {configurable: {thread_id: t2}}, ) resume Command(resume[ {id: a1b2, action: approve}, {id: c3d4, action: reject}, ]) result agent.graph.invoke(resume, {configurable: {thread_id: t2}}) print(result[messages][-1].content) # 已发送邮件至 张三: 你好... 转账操作已被拒绝未执行。被拒绝的工具调用不会进入approved_tool_calls工具节点直接跳过Agent 收到“未获审批”的反馈后继续对话。4.4 修改参数后执行用户还可以修改参数比如把转账金额从 500 改成 100resume Command(resume[ {id: a1b2, action: approve}, {id: c3d4, action: modify, args: {to: 李四, amount: 100}}, ])审批节点会把modified_args替换进原始 tool_call工具节点执行的是修改后的参数。5. 本篇常见错排查5.1 interrupt 后状态丢失最常见的问题是没配 checkpointer。interrupt_before依赖 checkpointer 保存断点如果compile()时没传checkpointer中断后状态就没了Command(resume...)会报找不到线程。# 错误没有 checkpointer builder.compile(interrupt_before[approval]) # 正确 builder.compile( checkpointerInMemorySaver(), interrupt_before[approval], )生产环境别用InMemorySaver进程重启就丢。换成SqliteSaver或PostgresSaver。5.2 一次调多个工具弹多次框如果你把interrupt()写在工具节点的 for 循环里每个工具都会中断一次。正确做法是收集节点一次性收集所有待审批项审批节点只interrupt()一次返回决策数组。5.3 安全工具也被拦SAFE_TOOLS白名单没配全或者auto_approve_tools没传。查询时间、做计算这类只读操作不该审批。在HarnessConfig里加HarnessConfig( enable_human_reviewTrue, auto_approve_tools[get_time, calculate], )5.4 Subgraph 里重复中断Planner 模式下子任务 Subgraph 内部如果也interrupt()用户会在同一线程看到多层中断。正确做法是 Subgraph 不自己中断只把待审批项写进subtask_pending_approvals主图的 approval 节点统一中断。Subgraph 的工具节点读取主图审批后的状态再执行。5.5 resume 格式不对Command(resume...)的格式必须和interrupt()返回的期望一致。我们约定返回决策数组每项含id和action。如果传了单个 dict 而不是 list审批节点里的isinstance(decisions, list)判断会走错分支。调试时先打印decisions看结构。5.6 工具调用 ID 对不上get_approved_tool_calls里用tool_name args匹配原始 tool_call如果两个工具同名同参数会匹配错。生产环境建议用 tool_call 自带的id字段做唯一标识而不是自己生成的短 ID。6. 把审批阀接进你的编码工作流审批队列跑通后你会发现它不只适用于“发邮件、转账”这类业务工具。在编码 Agent 场景里文件写入、命令执行、依赖安装同样需要审批阀——Agent 想rm -rf一个目录或者pip install一个来源不明的包都应该先停下来问一句。如果你在搭长期运行的编码 Agent需要更稳定的模型接入和额度管理可以看下 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。它适合需要持续调用模型、跑多轮 Agent 循环的场景比按次调用更省心。接入文档在这里里面有完整的 API 参数和错误码说明 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。遇到 401 或 429 先查文档里的鉴权和限流章节。最后留一个实用技巧审批队列的summary字段是给人看的别直接塞 JSON。把参数转成自然语言比如“向张三发送邮件内容包含‘你好’”用户扫一眼就能决策。我踩过的坑是 summary 写得太技术化审批时还得去翻 args体验很差。