1. 引言在构建 AI 智能体Agent时我们通常希望它能够自主完成尽可能多的任务。然而在真实业务场景中完全放手让模型自主决策往往伴随着风险它可能执行了不可逆的操作如删除数据、发送邮件、扣款或者需要用户提供额外的信息才能继续。Human-in-the-Loop人机协同简称 HITL正是为了解决这一问题而生的设计模式。它让智能体在关键节点暂停执行将控制权交还给人类等待人工确认、审批或补充信息后再继续运行。LangGraph 作为专为构建有状态、可编排的智能体应用而设计的框架原生提供了对 Human-in-the-Loop 的完善支持。本文将深入讲解 LangGraph 中实现 HITL 的三种核心机制中断interrupt、动态断点dynamic breakpoints与静态断点static breakpoints并通过可运行的代码示例帮助你快速上手。2. 为什么需要 Human-in-the-Loop在引入具体实现之前我们先明确 HITL 的价值。它主要解决以下三类问题安全性在执行删除、转账、发布等高风险操作前加入人工审批环节避免不可逆的损失。信息补全当模型发现输入信息不足时暂停并向用户提问获取必要信息后再继续。质量把关在生成关键内容如合同条款、对外公告后由人工审核确认确保输出符合预期。一个典型的 HITL 流程可以用下图表示否是智能体开始执行是否需要人工介入?继续自动执行暂停并等待人工输入人工审批 / 补充信息恢复执行任务结束3. 核心概念StateGraph 与 checkpointer在深入 HITL 之前我们需要先了解 LangGraph 的两个基础概念StateGraph和checkpointer。3.1 StateGraphLangGraph 的核心是图Graph。我们用StateGraph定义节点Node和边Edge节点是执行逻辑的函数边定义了节点之间的流转关系。每个节点接收当前状态State并返回状态的更新。3.2 Checkpointer检查点Checkpointer 是 LangGraph 实现 HITL 的基石。它负责在图的每一步执行后保存状态快照。有了检查点图才能在中断后从保存的状态恢复执行。LangGraph 提供了多种 checkpointer 实现最常用的是基于 SQLite 的SqliteSaver和基于内存的MemorySaver。生产环境推荐使用持久化的SqliteSaver或PostgresSaver。fromlanggraph.checkpoint.sqliteimportSqliteSaver# 使用内存中的 SQLite 数据库checkpointerSqliteSaver.from_conn_string(:memory:)4. 中断Interrupt最直接的 HITL 实现interrupt是 LangGraph 中实现 HITL 最直接、最推荐的方式。它像一个暂停按钮当图执行到interrupt函数时会暂停执行并抛出一个特殊的中断信号将控制权交还给调用方。4.1 基本用法fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportinterrupt,Commandfromlanggraph.checkpoint.sqliteimportSqliteSaverclassState(TypedDict):user_input:strapproved:booldefhuman_approval_node(state:State):在关键操作前暂停等待人工审批。# interrupt 会暂停图执行并将问题抛给外部调用方decisioninterrupt({question:是否批准执行该操作,context:state[user_input]})return{approved:decisionyes}defexecute_node(state:State):人工批准后执行的操作。ifstate[approved]:return{result:操作已执行}return{result:操作已取消}# 构建图builderStateGraph(State)builder.add_node(approval,human_approval_node)builder.add_node(execute,execute_node)builder.add_edge(START,approval)builder.add_edge(approval,execute)builder.add_edge(execute,END)# 绑定 checkpointer这是中断恢复的前提graphbuilder.compile(checkpointercheckpointer)4.2 运行与恢复# 第一次调用图会执行到 interrupt 处暂停config{configurable:{thread_id:thread-1}}resultgraph.invoke({user_input:删除用户数据},config)# 此时图已暂停result 中不包含最终结果# 人工审批后通过 Command 恢复执行resultgraph.invoke(Command(resumeyes),config)print(result)# 输出: {user_input: 删除用户数据, approved: True, result: 操作已执行}关键点在于恢复执行时必须传入相同的thread_id这样 LangGraph 才能从正确的检查点恢复状态。5. 动态断点Dynamic Breakpoints动态断点允许我们在图执行过程中根据当前状态动态地决定是否暂停。它通过interrupt配合条件判断实现比静态断点更灵活。5.1 实现方式defreview_node(state:State):# 根据状态动态决定是否需要人工介入ifstate[risk_level]high:# 高风险操作暂停等待审批decisioninterrupt(高风险操作请确认是否继续)return{approved:decisionconfirm}# 低风险操作直接放行return{approved:True}5.2 应用场景动态断点非常适合风险分级场景。例如转账金额超过阈值时暂停审批涉及删除操作时暂停确认检测到敏感词时暂停人工审核。6. 静态断点Static Breakpoints静态断点在编译图时通过interrupt_before或interrupt_after参数指定在特定节点执行前或执行后自动暂停。6.1 基本用法# 在 execute 节点执行前暂停graphbuilder.compile(checkpointercheckpointer,interrupt_before[execute])# 在 execute 节点执行后暂停graphbuilder.compile(checkpointercheckpointer,interrupt_after[execute])6.2 与动态断点的对比特性动态断点静态断点暂停时机运行时根据状态决定编译时固定指定灵活性高可动态判断低固定位置适用场景风险分级、条件审批固定流程节点审核实现方式interrupt()函数interrupt_before/interrupt_after7. 实战带人工审批的客服工单系统下面我们综合运用所学知识构建一个完整的客服工单处理系统。该系统在工单升级为高优先级时会暂停等待人工确认。fromtypingimportTypedDict,Literalfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportinterrupt,Commandfromlanggraph.checkpoint.sqliteimportSqliteSaverclassTicketState(TypedDict):ticket_id:strcontent:strpriority:Literal[low,high]approved:boolstatus:strdefclassify_node(state:TicketState):模拟工单分类判断优先级。# 实际项目中可调用 LLM 进行分类if投诉instate[content]or紧急instate[content]:return{priority:high}return{priority:low}defhuman_review_node(state:TicketState):高优先级工单需要人工确认。decisioninterrupt({ticket_id:state[ticket_id],content:state[content],message:该工单被标记为高优先级是否升级处理})return{approved:decisionapprove}defprocess_node(state:TicketState):处理工单。ifstate[priority]highandnotstate[approved]:return{status:rejected}return{status:processed}# 构建图builderStateGraph(TicketState)builder.add_node(classify,classify_node)builder.add_node(review,human_review_node)builder.add_node(process,process_node)builder.add_edge(START,classify)builder.add_edge(classify,review)builder.add_edge(review,process)builder.add_edge(process,END)# 使用动态断点仅高优先级工单触发人工审批defshould_review(state:TicketState)-bool:returnstate[priority]highbuilder.add_conditional_edges(classify,lambdastate:reviewifshould_review(state)elseprocess,{review:review,process:process})checkpointerSqliteSaver.from_conn_string(:memory:)graphbuilder.compile(checkpointercheckpointer)# 模拟运行config{configurable:{thread_id:ticket-001}}# 第一步提交工单resultgraph.invoke({ticket_id:T001,content:紧急投诉服务不可用},config)# 第二步人工审批通过resultgraph.invoke(Command(resumeapprove),config)print(result)# 输出: {ticket_id: T001, content: 紧急投诉服务不可用, priority: high, approved: True, status: processed}8. 常见问题与最佳实践8.1 忘记传 thread_id中断恢复时如果未传入相同的thread_idLangGraph 将无法找到对应的检查点导致恢复失败。务必在config中保持一致。8.2 未配置 checkpointerinterrupt依赖 checkpointer 保存状态。如果编译图时未绑定 checkpointer调用interrupt会直接报错。8.3 生产环境使用持久化存储开发环境可以使用MemorySaver或内存版SqliteSaver但生产环境务必使用持久化的PostgresSaver或文件版SqliteSaver防止进程重启后状态丢失。8.4 设计清晰的审批超时机制在实际业务中人工审批可能长时间无响应。建议结合外部任务队列如 Celery设计超时与自动降级策略避免工单长期卡死。9. 总结Human-in-the-Loop 是构建可靠、安全 AI 智能体的关键设计模式。LangGraph 通过 checkpointer 与 interrupt 机制为我们提供了简洁而强大的 HITL 支持中断interrupt在代码中显式暂停等待人工输入后恢复动态断点运行时根据状态灵活决定是否暂停静态断点编译时固定暂停位置适合流程化审核。掌握这三种机制你就能在 LangGraph 中轻松构建「该放手时放手该等人时等人」的智能体应用。建议你从本文的工单系统示例出发结合自身业务场景动手实践逐步完善审批、超时与异常处理逻辑。