纲要为什么要加入人工审核环节模型的不确定性高风险场景下的安全需求LangGraph 人机交互机制interrupt()在节点中暂停执行Command(resume...)恢复执行并传递人类反馈interrupt_before/interrupt_after在边处打断update_state直接修改图的状态实战一等待用户输入构建带反馈节点的简单图使用interrupt暂停Command恢复实战二审查工具调用在工具执行前插入审查节点支持批准、拒绝、修改参数实战三编辑图的状态使用interrupt_before在特定节点前暂停通过graph.update_state()修改状态后继续完整可运行代码总结与相关度说明为什么需要人工介入当前的大语言模型基于概率生成即使是最先进的模型也可能产生错误决策。在关键业务场景如金融交易、医疗建议、自动发布内容中一次错误的工具调用或回复可能造成严重影响。因此在生产级智能体系统中引入人机交互环节让人类能够审批、编辑甚至拒绝智能体的决策是保障系统可靠性的重要手段。LangGraph 为此提供了原语级的支持可以灵活地将人工审核嵌入到工作流的任意位置。核心机制interrupt与CommandLangGraph 的人机交互依赖于两个核心 APIinterrupt()在节点内部调用暂停图的执行并抛出一个需要人类处理的中断事件。Command(resume...)外部输入反馈后通过Command恢复执行同时可以将人类提供的数据注入到图中。此外还可以在编译图时通过interrupt_before或interrupt_after参数在指定节点执行前/后自动暂停无需在节点内部写interrupt()。实战一等待用户输入以下示例演示一个简单的工作流step_one→human_feedback→step_three。在human_feedback节点中调用interrupt()让人类提供反馈然后继续执行。importosfromtypingimportTypedDictfromdotenvimportload_dotenvfromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.typesimportinterrupt,Command load_dotenv()# 定义状态classFeedbackState(TypedDict):input:struser_feedback:str# 节点1不做具体处理只打印日志defstep_one(state:FeedbackState)-FeedbackState:print( 执行 Step 1)returnstate# 节点2等待人类反馈defhuman_feedback(state:FeedbackState)-FeedbackState:print( 等待人类反馈...)# interrupt 暂停执行提示信息会返回给调用方feedbackinterrupt(请提供你的反馈)# 当外部通过 Command(resume...) 恢复时feedback 会被赋值为传入的内容return{user_feedback:feedback}# 节点3使用反馈defstep_three(state:FeedbackState)-FeedbackState:print(f 执行 Step 3收到的反馈{state[user_feedback]})returnstate# 构建图builderStateGraph(FeedbackState)builder.add_node(step_one,step_one)builder.add_node(human_feedback,human_feedback)builder.add_node(step_three,step_three)builder.set_entry_point(step_one)builder.add_edge(step_one,human_feedback)builder.add_edge(human_feedback,step_three)builder.add_edge(step_three,END)# 激活短期记忆持久化用于中断恢复memoryMemorySaver()appbuilder.compile(checkpointermemory)# 首次调用会中断在 human_feedback 节点config{configurable:{thread_id:feedback-session-1}}input_data{input:你好}print( 首次调用会中断)foreventinapp.stream(input_data,config):fornode_name,valueinevent.items():ifisinstance(value,dict)anduser_feedbackinvalue:print(f节点{node_name}: 收到反馈 {value[user_feedback]})# 此时执行暂停我们需要获取中断事件并恢复# 实际上 stream 会抛出中断信息这里为简化演示直接使用 invoke 配合 Command 恢复# 方法使用 app.invoke 并传入 Command(resume...) 作为 inputprint(\n 恢复执行提供反馈)# 恢复时需要传入同一个 thread_id并用 Command 包裹 human 的反馈resume_inputCommand(resume我觉得很好继续吧)foreventinapp.stream(resume_input,config):fornode_name,valueinevent.items():ifisinstance(value,dict)anduser_feedbackinvalue:print(f节点{node_name}: 最终反馈 {value[user_feedback]})运行这段代码你会看到第一次调用在human_feedback处暂停然后通过Command(resume...)继续执行并最终在step_three看到反馈内容。实战二审查工具调用这是智能体最常用的场景在工具执行前暂停让人类审批工具的名称和参数。如果审批通过执行工具如果拒绝修改参数或终止流程。以下示例构建一个简单的天气查询智能体工具调用前需要人类审查。importosfromtypingimportTypedDict,Literalfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,AIMessage,ToolMessagefromlanggraph.graphimportStateGraph,END,MessagesStatefromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.prebuiltimportToolNode,tools_conditionfromlanggraph.typesimportinterrupt,Command load_dotenv()# 定义一个简单的天气工具模拟defget_weather(city:str)-str:查询指定城市的天气# 模拟返回固定天气returnf{city}天气晴朗25°Ctools[get_weather]# 绑定工具的LLMllmChatOpenAI(modelgpt-3.5-turbo,temperature0)llm_with_toolsllm.bind_tools(tools)# 定义状态classAgentState(MessagesState):pass# 模型调用节点defcall_model(state:AgentState)-AgentState:responsellm_with_tools.invoke(state[messages])return{messages:[response]}# 人类审查节点在工具执行前调用defhuman_review_node(state:AgentState)-AgentState:# 获取最后一条消息通常包含 tool_callslast_msgstate[messages][-1]ifnothasattr(last_msg,tool_calls)ornotlast_msg.tool_calls:returnstate# 无工具调用则跳过# 展示待审查的工具调用print( 需要审查的工具调用)forcallinlast_msg.tool_calls:print(f 工具:{call[name]}, 参数:{call[args]})# interrupt 等待人类决策返回字典格式的决策decisioninterrupt(请审批输入 continue 批准或提供修改后的参数 JSON)# decision 可以是字符串也可以是包含了修改后参数的对象# 这里为了简单只支持批准字符串 continueifdecisioncontinue:returnstateelse:# 其他情况可以拒绝或修改示例中我们简单拒绝print( 审查未通过终止执行)# 通过返回空消息终止流程实际可添加 ToolMessage 表示拒绝return{messages:[ToolMessage(content审查未通过,tool_call_idlast_msg.tool_calls[0][id])]}# 构建图builderStateGraph(AgentState)builder.add_node(call_model,call_model)builder.add_node(review,human_review_node)builder.add_node(tools,ToolNode(tools))builder.set_entry_point(call_model)# 条件边模型调用后如果有工具调用进入审查否则直接结束defshould_review(state:AgentState)-str:iftools_condition(state):returnreviewreturnEND builder.add_conditional_edges(call_model,should_review,{review:review,END:END})builder.add_edge(review,tools)builder.add_edge(tools,call_model)# 工具执行后回到模型继续思考memoryMemorySaver()appbuilder.compile(checkpointermemory,interrupt_before[review])# 在 review 节点前中断# 测试询问天气会触发工具调用和审查config{configurable:{thread_id:review-tools-1}}input_data{messages:[HumanMessage(content北京天气怎么样)]}print( 开始对话将中断在审查前)# 第一次调用会暂停在 review 节点前foreventinapp.stream(input_data,config):pass# 恢复批准工具调用print(\n 恢复批准工具调用 )resume_cmdCommand(resumecontinue)foreventinapp.stream(resume_cmd,config):fornode_name,valueinevent.items():ifmessagesinvalueandvalue[messages]:print(f节点{node_name}:{value[messages][-1].content})运行后你会看到工具调用被暂停你可以选择批准输入continue或修改。这里为了简化只支持批准。实际可以解析 JSON 来更新参数。实战三编辑图的状态有时我们不希望仅仅在节点内部暂停而是想在图的任意位置暂停并且可以修改状态例如修改之前某个节点产生的数据后再继续。使用interrupt_before在指定节点前暂停然后使用graph.update_state()直接修改状态最后再用Command或stream继续。fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.typesimportCommandclassEditState(TypedDict):text:strdefstep_one(state:EditState)-EditState:# 将输入大写return{text:state[text].upper()}defstep_two(state:EditState)-EditState:# 添加后缀return{text:state[text] [processed]}defstep_three(state:EditState)-EditState:# 最终输出return{text:最终结果: state[text]}builderStateGraph(EditState)builder.add_node(step_one,step_one)builder.add_node(step_two,step_two)builder.add_node(step_three,step_three)builder.set_entry_point(step_one)builder.add_edge(step_one,step_two)builder.add_edge(step_two,step_three)builder.add_edge(step_three,END)memoryMemorySaver()appbuilder.compile(checkpointermemory,interrupt_before[step_two])# 在 step_two 前暂停config{configurable:{thread_id:edit-state-1}}# 初始调用input1{text:hello world}print( 初始调用暂停在 step_two 前)eventslist(app.stream(input1,config))# 此时状态为 step_one 已执行text 为 HELLO WORLDprint(当前状态,app.get_state(config).values)# 我们想要修改这个状态比如把 text 改成其他内容print(\n 更新状态 )app.update_state(config,{text:CUSTOM TEXT})print(更新后状态,app.get_state(config).values)# 继续执行print(\n 继续执行 )foreventinapp.stream(None,config):fornode_name,valueinevent.items():print(f节点{node_name}:{value})运行后你会发现step_two接收到的text已经被改成了CUSTOM TEXT之后的结果也是基于修改后的值。这种能力允许人类在流程中任意点修正数据再继续执行。总结人机交互是 LangGraph 为生产级 AI 应用提供的关键能力。通过interrupt、Command、interrupt_before和update_state开发者可以灵活地在工作流中插入审批、修改状态、拒绝工具调用等操作从而在不完全信任模型决策时确保系统安全可控。本文的三个实战示例给出了最基础的用法实际项目中可以组合出更复杂的审核流程。本文完整覆盖了人机交互的概念、等待用户输入、审查工具调用、编辑图状态等所有演示场景并提供了可直接运行的代码示例。代码已可直接运行需安装 langgraph, langchain-openai 等依赖并配置 API Key。