在上一篇文章中我们通过 create_agent 几行代码就构建了一个可运行的 Agent。但 create_agent 是一个“黑盒”——它内部构建了一个 StateGraph但你无法控制流程的细节。当你需要条件分支、循环控制、中断恢复、人工审批时就必须直接使用 LangGraph 的 StateGraph。简单 Agent 的核心是 Model → Action → Observation → Model 的自由循环这种方式灵活但不可预测。进入生产环境后系统关注的问题变为Agent 执行到哪里了哪些步骤不能跳过一个节点最多重复几次服务重启后怎样继续什么时候必须让人确认 本文从 StateGraph 的核心概念讲起深入讲解条件边、中断恢复和 Human-in-the-Loop并通过一个完整的客服 Agent 实战帮你掌握 LangGraph 的编排能力。一、为什么需要 LangGraphLangChain 的 LCEL 适合表达线性数据流但真实 Agent 的流程往往不是线性的核心原则状态机负责确定性的流程控制Agent 负责节点内部的理解、推理与工具调用实现 “确定性的骨架 概率性的智能” 。二、StateGraph 的核心概念LangGraph 的核心抽象只有三个State状态 应用当前快照一个共享的数据结构Node节点 一个函数读取并更新 StateEdge边 定义节点之间的连接和执行顺序最简单的 StateGraphfromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDictclassState(TypedDict):input:stroutput:strdefprocess_node(state:State)-State:return{output:f处理结果:{state[input]}}builderStateGraph(State)builder.add_node(process,process_node)builder.add_edge(START,process)builder.add_edge(process,END)graphbuilder.compile()resultgraph.invoke({input:测试})print(result)# {input: 测试, output: 处理结果: 测试}三、条件边让流程“会拐弯”条件边是 LangGraph 实现循环和分支的核心机制。 它通过一个路由函数根据当前状态动态决定下一个节点。fromlanggraph.graphimportStateGraph,START,ENDdefshould_continue(state:State)-str:路由函数根据状态决定下一步ifstate.get(needs_tool):returncall_toolelifstate.get(is_done):returnendelse:returncontinuebuilder.add_conditional_edges(agent,# 源节点should_continue,# 路由函数{call_tool:tools,# 返回 call_tool → 去 tools 节点continue:agent,# 返回 continue → 回到 agent 节点end:END,# 返回 end → 结束})条件边的典型应用场景Agent 调用工具后判断是否需要继续调用循环质量检查后判断是通过还是重试分支人工审批后判断是批准还是拒绝分流四、Checkpoint让 Agent“记得住”LangGraph 的持久化层通过 Checkpointer 实现。每个线程thread的状态会被自动保存即使服务重启也能恢复。fromlanggraph.checkpoint.memoryimportInMemorySaver checkpointerInMemorySaver()graphbuilder.compile(checkpointercheckpointer)# 每个线程有唯一的 thread_idconfig{configurable:{thread_id:user-123}}# 第一次执行resultgraph.invoke({input:开始任务},configconfig)# 后续执行可以恢复之前的状态result2graph.invoke({input:继续},configconfig)Checkpointer 的价值会话可以暂停和恢复。这对于长时间运行的任务和人工审批流程至关重要。五、中断与恢复Human-in-the-LoopHuman-in-the-LoopHITL 模式让 Agent 在关键节点暂停执行等待人工确认后再继续。5.1 基本用法fromlanggraph.typesimportinterrupt,Commanddefapproval_node(state:State)-State:需要人工审批的节点# interrupt() 会暂停执行返回人工输入的值human_decisioninterrupt({question:是否批准这个操作,operation:state.get(pending_operation),})ifhuman_decisionapproved:return{status:approved}else:return{status:rejected}执行流程图以初始状态启动执行到 interrupt() 时暂停返回一个中断对象调用方检查中断对象获取需要人工决策的 payload调用方通过 Command(resume…) 恢复执行传入人工输入interrupt() 返回人工输入的值节点继续执行5.2 恢复执行# 第一次执行会停在 interruptresultgraph.invoke(initial_state,configconfig)print(result[__interrupt__])# 查看中断信息# 恢复执行传入人工决策finalgraph.invoke(Command(resumeapproved),# 人工批准configconfig)关键注意恢复执行时必须使用发生中断时使用的同一个 thread ID。5.3 HITL 的典型场景六、实战用 LangGraph 构建一个客服 Agent以下是一个完整的客服 Agent包含条件分支、工具调用和人工升级。fromtypingimportTypedDict,Literalfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.typesimportinterrupt,Commandfromlangchain_openaiimportChatOpenAIfromlangchain_core.toolsimporttool# State 定义 classCustomerServiceState(TypedDict):user_query:strcategory:strresponse:strneeds_human:boolhuman_decision:strresolved:bool# 工具定义 tooldefcheck_order_status(order_id:str)-str:查询订单状态。returnf订单{order_id}已发货预计明天到达。tooldefprocess_refund(order_id:str)-str:处理退款请求。returnf订单{order_id}的退款申请已提交。# 节点定义 defclassify_node(state:CustomerServiceState)-CustomerServiceState:对用户问题进行分类querystate[user_query].lower()if退款inqueryor退钱inquery:categoryrefundelif订单inqueryor物流inquery:categoryorderelse:categoryotherreturn{category:category}defroute_by_category(state:CustomerServiceState)-str:根据分类路由到不同处理节点categorystate[category]ifcategoryrefund:returnrefund_handlerelifcategoryorder:returnorder_handlerelse:returnhuman_escalationdeforder_handler(state:CustomerServiceState)-CustomerServiceState:处理订单查询resultcheck_order_status.invoke({order_id:ORD-001})return{response:result,resolved:True}defrefund_handler(state:CustomerServiceState)-CustomerServiceState:处理退款请求需要人工审批ifstate.get(human_decision)approved:resultprocess_refund.invoke({order_id:ORD-001})return{response:result,resolved:True}else:# 触发中断等待人工审批decisioninterrupt({question:是否批准退款,order_id:ORD-001,})ifdecisionapproved:resultprocess_refund.invoke({order_id:ORD-001})return{response:result,human_decision:approved,resolved:True}else:return{response:退款被拒绝,human_decision:rejected,resolved:True}defhuman_escalation(state:CustomerServiceState)-CustomerServiceState:升级到人工客服return{response:已转接人工客服请稍候。,needs_human:True,resolved:False}# 构建图 builderStateGraph(CustomerServiceState)builder.add_node(classify,classify_node)builder.add_node(order_handler,order_handler)builder.add_node(refund_handler,refund_handler)builder.add_node(human_escalation,human_escalation)builder.add_edge(START,classify)builder.add_conditional_edges(classify,route_by_category,{order_handler:order_handler,refund_handler:refund_handler,human_escalation:human_escalation,})builder.add_edge(order_handler,END)builder.add_edge(refund_handler,END)builder.add_edge(human_escalation,END)checkpointerInMemorySaver()graphbuilder.compile(checkpointercheckpointer)# 执行 config{configurable:{thread_id:customer-001}}# 测试订单查询自动处理resultgraph.invoke({user_query:我的订单到哪了},configconfig)print(result[response])# 订单 ORD-001 已发货预计明天到达。# 测试退款需要人工审批config2{configurable:{thread_id:customer-002}}resultgraph.invoke({user_query:我要退款},configconfig2)print(result[__interrupt__])# 触发中断# 人工批准后恢复finalgraph.invoke(Command(resumeapproved),configconfig2)print(final[response])# 订单 ORD-001 的退款申请已提交。七、LangGraph 与 create_agent 的关系官方推荐从 create_agent 开始快速原型当需要更精细的控制时下降到 StateGraph。两者可以混合使用——外层用 StateGraph 控制业务边界节点内部用 create_agent 处理不确定问题。八、小结StateGraph 三要素State共享状态、Node执行单元、Edge流转控制。条件边通过路由函数根据状态动态决定下一个节点实现分支和循环。Checkpointer持久化会话状态支持中断恢复。Human-in-the-Loop通过 interrupt() 暂停通过 Command(resume…) 恢复实现人工审批。核心原则状态机负责确定性流程控制Agent 负责节点内部的智能决策。create_agent vs StateGraph前者快速原型后者精细控制按需选择。