1. 当 Multi-Agent 跑偏时我们到底缺什么LangGraph 的 Multi-Agent 系统能做什么简单说它让多个各有所长的 Agent 协同完成一个复杂任务一个负责搜索、一个负责写代码、一个负责审核彼此通过共享状态传递消息。适合谁适合那些已经把单 Agent 玩明白、开始处理真实业务流订单、工单、数据写入的开发者。但问题也随之而来——Agent 越自主越容易在关键节点做出你无法接受的决策。我试过让一个搜索 Agent 自动调用外部接口结果它把参数里的城市名拼错了接口照样返回 200下游流程一路跑到底最后生成了一份完全错误的报告。这类事故的根因不是模型不够强而是缺少一个“暂停按钮”在 Agent 真正执行高风险动作之前把控制权交还给人类。LangGraph 给出的答案就是 interrupt 机制也就是 Human-in-the-Loop。它的核心思路是Graph 每执行一步就用 Checkpointer 保存一次状态当执行流遇到interrupt()时立即暂停并持久化人类可以在几秒甚至几天后回来处理再用Command(resume...)从断点恢复。这套机制支撑起四大经典干预模式批准/拒绝、编辑图状态、审查工具调用、验证人工输入。这篇内容聚焦落地配置会给出可复制的config.toml与settings.json骨架、TaoToken 统一 Key/API 通道接入 AI 工具的配置示例并附上中断触发与恢复的验证动作。你不需要重写现有 Agent只要在关键节点包一层就能用。2. 前置准备TaoToken 统一通道与 LangGraph 环境在写中断逻辑之前先把模型调用通道理顺。Multi-Agent 系统里往往要同时调用多个模型规划用强模型、执行用快模型如果每个模型都单独配 Key配置会迅速失控。TaoToken 提供统一的 API 通道一个 Key 就能覆盖多种模型省去在多个平台之间来回切换的麻烦。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。你需要先在控制台创建一个 API Key然后把它写进环境变量避免硬编码进代码。2.1 环境变量与依赖安装先装依赖LangGraph 的中断能力依赖 checkpointer所以langgraph和langchain-openai都要装pip install -U langgraph langchain langchain-openai接着配置环境变量。把 Key 和 Base URL 分开管理方便在不同环境切换export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api2.2 config.toml 骨架如果你用 TOML 管理项目配置可以这样组织。把模型通道和 Agent 参数分开中断相关的开关单独放一段[llm] provider openai base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY model gpt-4o temperature 0.0 [graph] checkpointer memory # 生产环境换成 postgres thread_prefix agent [interrupt] enabled true allow_accept true allow_edit true allow_respond true2.3 settings.json 骨架如果你的工具链读 JSON等价配置如下。注意interrupt_config里的三个开关直接决定前端能显示哪些操作按钮{ llm: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, model: gpt-4o, temperature: 0 }, graph: { checkpointer: memory, thread_prefix: agent }, interrupt: { enabled: true, interrupt_config: { allow_accept: true, allow_edit: true, allow_respond: true } } }注意allow_edit打开后人类可以修改工具参数再放行这是防止参数拼错最有效的一道闸门。生产环境建议至少保留allow_accept和allow_edit。3. 可复制配置四大模式的 interrupt 落地四大模式共用同一套底层机制区别只在中断点放在哪里、恢复时传什么。下面逐个给出可直接运行的骨架。3.1 模式一批准/拒绝Approve or Reject这是最常用的模式适合 API 调用前、敏感操作确认。核心是让中断节点返回Command(goto...)根据人类决策走不同分支from typing import Literal, TypedDict import uuid from langgraph.constants import END from langgraph.graph import StateGraph from langgraph.types import interrupt, Command from langgraph.checkpoint.memory import InMemorySaver class State(TypedDict): llm_output: str decision: str def generate_llm_output(state: State) - State: return {llm_output: 这是AI生成的一段需要审批的文本。} def human_approval(state: State) - Command[Literal[approved_path, rejected_path]]: decision interrupt({ question: 请审批以下内容回复 approve 或 reject, llm_output: state[llm_output] }) if decision approve: return Command(gotoapproved_path, update{decision: approved}) return Command(gotorejected_path, update{decision: rejected}) def approved_node(state: State) - State: return state def rejected_node(state: State) - State: return state builder StateGraph(State) builder.add_node(generate_llm_output, generate_llm_output) builder.add_node(human_approval, human_approval) builder.add_node(approved_path, approved_node) builder.add_node(rejected_path, rejected_node) builder.set_entry_point(generate_llm_output) builder.add_edge(generate_llm_output, human_approval) builder.add_edge(approved_path, END) builder.add_edge(rejected_path, END) graph builder.compile(checkpointerInMemorySaver())3.2 模式二编辑图状态Review and Edit State当模型输出有错别字或信息缺失时让人类直接改状态再往下走。中断的返回值就是人类编辑后的内容class EditState(TypedDict): summary: str def generate_summary(state: EditState) - EditState: return {summary: The cat sat on the mat and looked at the stars.} def human_review_edit(state: EditState) - EditState: result interrupt({ task: 请审查并按需编辑摘要, generated_summary: state[summary] }) return {summary: result[edited_summary]} def downstream_use(state: EditState) - EditState: print(f使用编辑后的摘要: {state[summary]}) return state3.3 模式三审查工具调用Review Tool Calls这是安全性的终极防线。做法是写一个通用包装器把任意工具包一层在真正调用前插入中断。这样不用改原工具代码复用性最高from typing import Callable from langchain_core.tools import BaseTool, tool as create_tool from langchain_core.runnables import RunnableConfig from langgraph.types import interrupt from langgraph.prebuilt.interrupt import HumanInterruptConfig, HumanInterrupt def add_human_in_the_loop( tool: Callable | BaseTool, *, interrupt_config: HumanInterruptConfig None, ) - BaseTool: if not isinstance(tool, BaseTool): tool create_tool(tool) if interrupt_config is None: interrupt_config { allow_accept: True, allow_edit: True, allow_respond: True, } create_tool(tool.name, descriptiontool.description, args_schematool.args_schema) def call_tool_with_interrupt(config: RunnableConfig, **tool_input): request: HumanInterrupt { action_request: {action: tool.name, args: tool_input}, config: interrupt_config, description: 请审查工具调用 } response interrupt([request])[0] if response[type] accept: tool_response tool.invoke(tool_input, config) elif response[type] edit: tool_input response[args][args] tool_response tool.invoke(tool_input, config) elif response[type] response: tool_response response[args] else: raise ValueError(f不支持的响应类型: {response[type]}) return tool_response return call_tool_with_interrupt如果工具是异步的比如走 MCP 协议的网络搜索把def改成async def并把tool.invoke换成await tool.ainvoke其余逻辑不变。3.4 模式四验证人工输入Validate Human Input当需要用户提供特定格式输入时在节点内部用循环反复请求直到输入合法class AgeState(TypedDict): age: int def get_valid_age(state: AgeState) - AgeState: prompt 请输入年龄非负整数 while True: user_input interrupt(prompt) try: age int(user_input) if age 0: raise ValueError(年龄不能为负) break except (ValueError, TypeError): prompt f{user_input} 不合法请重新输入非负整数 return {age: age}4. 验证请求中断触发与恢复的完整动作配置写完必须验证否则你不知道中断到底有没有生效。下面用模式一跑一遍完整闭环。4.1 首次调用触发中断config {configurable: {thread_id: fthread-{uuid.uuid4()}}} result graph.invoke({}, configconfig) print(result[__interrupt__])预期输出里会出现__interrupt__键值是Interrupt(value{question: ..., llm_output: ...}, id...)。看到这个就说明中断成功Graph 已暂停并保存状态。4.2 用 Command 恢复执行final_result graph.invoke(Command(resumeapprove), configconfig) print(final_result)resume里的值会成为interrupt()的返回值赋给decision变量。传approve走批准分支传reject走拒绝分支。最终状态里应包含decision: approved。4.3 工具调用中断的验证用包装器包一个搜索工具跑一次 Agentfrom langgraph.prebuilt import create_react_agent from langgraph.checkpoint.memory import InMemorySaver def search_hotel(hotel_name: str): 根据酒店名称查询酒店信息 return f为您查询{hotel_name}. agent create_react_agent( modelmodel, tools[add_human_in_the_loop(search_hotel)], checkpointerInMemorySaver(), namesearch_assistant, prompt你是一个酒店查询工具 ) config {configurable: {thread_id: 1}} for chunk in agent.stream({messages: [{role: user, content: 帮我查下北京瑰丽酒店的信息}]}, config): print(chunk)当模型识别出hotel_name北京瑰丽酒店、参数完备时会触发中断。此时用Command(resume[{type: accept}])恢复工具才会真正执行。注意如果参数没填全比如只说“我要查酒店”模型会先追问参数不会触发中断。中断只在工具即将被调用的那一刻发生。5. 本篇常见错排查5.1 报错没有 checkpointer 导致 interrupt 失效最常见的坑是编译 Graph 时忘了传 checkpointer。中断的本质是状态保存与恢复没有持久化层interrupt()调用会直接报错或静默失效。检查你的compile()是否带了checkpointerInMemorySaver()生产环境换成 Postgres 或 SQLite 持久化。5.2 恢复后副作用被重复执行这是理解 interrupt 最关键的一点恢复执行不是从interrupt()那一行继续而是从包含它的节点开头重新执行整个节点。重跑期间再次遇到interrupt()时不会暂停直接返回resume的值。这意味着任何位于interrupt()之前的副作用操作API 调用、数据库写入都会被重复执行。解决办法是把副作用放到interrupt()之后或者拆到独立的后续节点里。比如不要在中断节点开头就写数据库而是等人类批准后再写。5.3 thread_id 不一致导致恢复失败恢复时必须使用与首次调用完全相同的config尤其是thread_id。如果两次调用的 thread_id 不同Graph 找不到对应的检查点会当成一次全新执行中断信息丢失。建议把 thread_id 存到会话上下文里统一管理。5.4 并行中断的批量恢复如果 Graph 并行执行了多个含中断的节点result[__interrupt__]会是一个列表。此时不能只传一个值需要构造映射resume_map {i.id: value for i, value in zip(result[__interrupt__], values)} graph.invoke(Command(resumeresume_map), configconfig)5.5 子图中的中断重跑范围当中断发生在子图里时恢复会从父图调用该子图的节点开头、以及子图内含中断的节点开头同时重跑。如果你的子图里有副作用同样要按 5.2 的原则处理。6. 把中断接进你的真实工作流到这里四大模式的骨架和验证动作都齐了。回到实际项目建议按这个顺序推进先用模式三的通用包装器把所有高风险工具包一遍这是投入产出比最高的一步再在关键决策节点加模式一的批准分支模式二和模式四按业务需要补。模型通道方面如果你还在为多个模型分别配 Key 而头疼可以到控制台创建一个统一 Key接入文档里有各语言的调用示例。需要长期跑编码类 Agent、或者想让 Agent 在后台持续执行任务的可以看看 Coding Plan它更适合这种长周期场景。想先验证模型输出质量的直接进模型对话试几轮确认中断逻辑符合预期再上生产。最后留一个实用技巧调试阶段把interrupt_before和interrupt_after打开相当于在节点前后打断点能清楚看到每一步的状态变化。但这两个静态中断不推荐用于生产环境的人机交互生产还是用interrupt()动态中断更灵活。