Agent 越来越能干了——能读文件、发邮件、操作数据库、调用 API。但能力越大风险越大。试想这个场景你让 Agent「帮我给老板发封邮件汇报项目进度」Agent 二话不说就发了——但邮件内容把未完成的模块写成了「已完成」还 CC 了全组。问题不在于 Agent 不够聪明而在于它太自主了。在关键操作执行前应该有一个人来把把关。这就是 人机协作Human-in-the-Loop, HITL要解决的核心问题。今天这篇我们用 LangGraph 从原理到代码彻底讲透它。一、什么是人机协作人机协作不是让人替代 Agent 做事而是在 Agent 的执行链路中插入「检查点」让人类在关键时刻介入。主要有三种模式审批Agent 要执行高风险操作前先让人确认。比如发邮件、删数据、转账。修正Agent 的输出不完美时让人编辑后再继续。比如修改 SQL 查询条件。引导Agent 不知道怎么做时让人给方向。比如提供额外上下文或偏好。本质上是给 Agent 装了一个「刹车系统」——平时让它自己跑关键时刻人来踩刹车。二、LangGraph 为什么适合做 HITL市面上 Agent 框架不少但 LangGraph 在 HITL 方面有三个独特优势原生状态持久化LangGraph 的每个节点执行后都会通过 Checkpointer 自动保存状态。这意味着执行可以在任意节点暂停然后在任意时间点恢复——这是 HITL 的物理基础。内置中断机制提供两层中断 API编译时配置 interrupt_before 和运行时调用 interrupt() 函数灵活度很高。线程级隔离每个对话有独立的 thread_id多个用户的人机交互互不干扰暂停/恢复精确到线程级别。三、核心原理暂停 — 交互 — 恢复LangGraph 的 HITL 本质上是一个三步循环Agent 执行 → interrupt() 暂停 → 人类决策 → Command(resume) 恢复 → 继续执行 ↑ | └──── 状态通过 Checkpointer 持久化 ────────┘Step 1暂停InterruptAgent 执行到某个节点时调用 interrupt() 暂停执行。图的完整状态被保存到 Checkpointer当前线程进入「等待人类输入」状态。Step 2交互Human Input应用层从中断中获取 Agent 的请求信息比如「我想发这封邮件参数如下批准吗」展示给人类用户。人类做出决策。Step 3恢复Resume调用 Command(resume人类的决策) 恢复执行。interrupt() 函数的返回值就是人类输入的内容Agent 基于这个输入继续往下走。四、完整实战邮件审批 Agent下面是一个完整的「邮件发送审批」示例。Agent 想要发邮件时先暂停等待人类审批人类可以批准、拒绝或提供修改建议。第一部分定义状态和节点from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langgraph.types import interrupt, Command from langchain_core.messages import HumanMessage, AIMessage import operator # 1. 定义状态Agent 执行过程中需要携带的数据 class AgentState(TypedDict): messages: Annotated[list, operator.add] # 对话历史 tool_call: dict | None # 待执行的工具调用 human_feedback: str | None # 人类的审批结果 # 2. Agent 节点LLM 分析用户请求决定要做什么 def agent_node(state: AgentState) - dict: last_msg state[”messages”][-1].content # 实际场景中这里调用 LLM简化为规则判断 if ”发邮件” in last_msg: return { ”tool_call”: { ”name”: ”send_email”, ”args”: { ”to”: ”bosscompany.com”, ”subject”: ”项目进度汇报”, } } } return {”tool_call”: None} # 3. 人类审批节点暂停执行等待人类决策 def human_approval_node(state: AgentState) - dict: tool_call state[”tool_call”] # interrupt() 会暂停整个图的执行 # 传入的字典会返回给调用方用于展示给人类 decision interrupt({ ”type”: ”approval_request”, ”tool”: tool_call[”name”], ”args”: tool_call[”args”], }) # 当人类通过 Command(resume...) 恢复执行后 # decision 的值就是 resume 传入的内容 return {”human_feedback”: decision} # 4. 执行节点根据审批结果执行或取消 def execute_node(state: AgentState) - dict: feedback state[”human_feedback”] tool_call state[”tool_call”] if feedback ”approve”: # 实际执行工具调用发邮件等 return {”messages”: [AIMessage( contentf”已执行 {tool_call[name]}参数{tool_call[args]}” )]} elif feedback ”reject”: return {”messages”: [AIMessage(content”操作已取消”)]} else: # 人类提供了修改建议回到 agent 重新规划 return {”messages”: [AIMessage( contentf”收到修改建议{feedback}重新规划中...” )]}第二部分构建图# 5. 路由函数决定下一步去哪个节点 def route_after_agent(state: AgentState) - str: # 有工具调用 → 去人类审批 if state.get(”tool_call”): return ”human_approval” return ”end” def route_after_execute(state: AgentState) - str: feedback state.get(”human_feedback”, ””) # 非 approve/reject 说明是人类修改建议 → 回到 agent if feedback not in (”approve”, ”reject”): return ”agent” return ”end” # 6. 构建状态图 workflow StateGraph(AgentState) workflow.add_node(”agent”, agent_node) workflow.add_node(”human_approval”, human_approval_node) workflow.add_node(”execute”, execute_node) workflow.add_edge(START, ”agent”) workflow.add_conditional_edges(”agent”, route_after_agent, { ”human_approval”: ”human_approval”, ”end”: END, }) workflow.add_edge(”human_approval”, ”execute”) workflow.add_conditional_edges(”execute”, route_after_execute, { ”agent”: ”agent”, ”end”: END, }) # 7. 编译必须传入 checkpointer否则无法暂停/恢复 app workflow.compile(checkpointerMemorySaver())关键点checkpointer是 HITL 的基石。没有它interrupt()无法保存状态恢复也就无从谈起。生产环境建议用SqliteSaver或PostgresSaver替代内存版。第三部分运行 人机交互# 每个对话线程有唯一 ID暂停/恢复靠它定位 config {”configurable”: {”thread_id”: ”thread-1”}} # ---- 第一次调用Agent 执行到 human_approval 会暂停 ---- result app.invoke( {”messages”: [HumanMessage(content”帮我给老板发邮件汇报项目进度”)]}, config ) # 此时图在 human_approval 节点暂停 # result 中包含 interrupt() 传入的数据 # 查看当前状态 state app.get_state(config) print(state.next) # (human_approval,) 表示停在这里 # ---- 人类做出决策后恢复执行 ---- # 情况 1批准 result app.invoke(Command(resume”approve”), config) print(result[”messages”][-1].content) # 输出: 已执行 send_email参数{...} # 情况 2拒绝 # result app.invoke(Command(resume”reject”), config) # 输出: 操作已取消 # 情况 3提供修改建议 # result app.invoke( # Command(resume”语气委婉一些不要说延期了”), config # ) # 输出: 收到修改建议语气委婉一些... 重新规划中... # Agent 会回到 agent 节点重新决策整个流程的精妙之处在于Agent 不需要知道人类会怎么回应。它只需要在需要审批的地方调用 interrupt()剩下的暂停、保存、恢复都由 LangGraph 框架自动处理。五、两种中断方式对比LangGraph 提供了两种实现 HITL 的方式适用场景不同interrupt()函数interrupt_before配置触发方式在节点内部按需调用编译时指定在哪些节点前暂停自定义数据可以携带任意数据给人类通过 get_state 查看状态人类输入直接作为函数返回值需通过 update_state 修改灵活性高适合复杂交互低适合固定审批点代码侵入需修改节点代码无需修改节点代码如果审批点是固定的比如「所有发邮件操作都要审批」用 interrupt_before 更简洁# 方式二interrupt_before无需修改节点代码 app2 workflow.compile( checkpointerMemorySaver(), interrupt_before[”execute”] # execute 前暂停 ) # 第一次调用 → 暂停在 execute 之前 result app2.invoke( {”messages”: [HumanMessage(content”帮我发邮件”)]}, config ) # 查看状态确认 Agent 想做什么 state app2.get_state(config) print(state.values.get(”tool_call”)) # {name: send_email, args: {...}} # 人类确认后继续执行传入 None 不改状态直接继续 result app2.invoke(None, config)六、进阶条件中断实际业务中不是所有操作都需要人类审批。发个通知邮件可以自动执行但给老板发辞职信必须审批。这就需要条件中断——只在高风险操作时暂停。# 高风险工具白名单只有这些工具需要人类审批 HIGH_RISK_TOOLS { ”send_email”, # 发邮件 ”delete_file”, # 删文件 ”execute_sql”, # 执行 SQL ”transfer_money”, # 转账 } def smart_agent_node(state: AgentState) - dict: # LLM 决定要调用什么工具简化版 tool_call decide_tool_call(state[”messages”]) if tool_call and tool_call[”name”] in HIGH_RISK_TOOLS: # 高风险 → 走审批流程 return {”tool_call”: tool_call, ”needs_approval”: True} elif tool_call: # 低风险 → 直接执行 return {”tool_call”: tool_call, ”needs_approval”: False} return {”tool_call”: None} # 路由根据风险等级决定是否需要审批 def route_by_risk(state: AgentState) - str: if not state.get(”tool_call”): return ”end” if state.get(”needs_approval”): return ”human_approval” return ”execute” # 低风险直接执行这样 Agent 就能自主判断哪些操作需要人类把关在安全性和效率之间取得平衡。实际项目中风险等级可以做得更细比如按操作金额分档100 元以下自动、100-1000 元需组长审批、1000 元以上需总监审批。七、五条最佳实践Checkpointer 必须持久化到外部存储MemorySaver 只用于开发调试。生产环境用 PostgresSaver 或 RedisSaver否则进程重启所有暂停状态全丢。设置中断超时机制人类可能迟迟不响应。在应用层加超时逻辑——超过 30 分钟自动拒绝或转人工客服避免线程无限挂起。interrupt 信息要让人看得懂interrupt() 传入的字典是给人类看的。写清楚工具名、参数、潜在影响别让审批者猜。审批结果要可追溯把每次 interrupt/resume 的完整记录谁审批的、审批了什么、何时审批的写入审计日志出问题能回溯。最小化人工介入频率介入太频繁 没有自动化。通过条件中断只拦截高风险操作让低风险操作自动通过逐步建立信任后再放宽自动执行的范围。八、写在最后Agent 的自主性和安全性是一对永恒矛盾。给 Agent 完全自由它可能在某个角落闯祸给它太多约束又失去了自动化的意义。LangGraph 的 HITL 机制提供了一个优雅的解法让 Agent 自由探索但在关键路口设一个红绿灯。核心就三步——interrupt 暂停、human 决策、command resume 恢复。理解了这个循环你就能在任何 Agent 场景中插入人类把关点。剩下的就是工程细节什么操作需要审批、审批信息怎么展示、超时怎么处理——这些才是真正花时间的地方。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
