第四篇：《LangGraph 实战：用状态机构建可控的 Agent 工作流》-平芜编程栈

在上一篇文章中，我们通过 create_agent 几行代码就构建了一个可运行的 Agent。但 create_agent 是一个“黑盒”——它内部构建了一个 StateGraph，但你无法控制流程的细节。当你需要条件分支、循环控制、中断恢复、人工审批时，就必须直接使用 LangGraph 的 StateGraph。简单 Agent 的核心是 Model → Action → Observation → Model 的自由循环，这种方式灵活但不可预测。进入生产环境后，系统关注的问题变为：Agent 执行到哪里了？哪些步骤不能跳过？一个节点最多重复几次？服务重启后怎样继续？什么时候必须让人确认？本文从 StateGraph 的核心概念讲起，深入讲解条件边、中断恢复和 Human-in-the-Loop，并通过一个完整的客服 Agent 实战帮你掌握 LangGraph 的编排能力。

一、为什么需要 LangGraph？
LangChain 的 LCEL 适合表达线性数据流，但真实 Agent 的流程往往不是线性的：

核心原则：状态机负责确定性的流程控制，Agent 负责节点内部的理解、推理与工具调用，实现 “确定性的骨架 + 概率性的智能” 。

二、StateGraph 的核心概念
LangGraph 的核心抽象只有三个：

State（状态）：应用当前快照，一个共享的数据结构

Node（节点）：一个函数，读取并更新 State

Edge（边）：定义节点之间的连接和执行顺序

最简单的 StateGraph：

fromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDictclassState(TypedDict):input:stroutput:strdefprocess_node(state:State)->State:return{"output":f"处理结果:{state['input']}"}builder=StateGraph(State)builder.add_node("process",process_node)builder.add_edge(START,"process")builder.add_edge("process",END)graph=builder.compile()result=graph.invoke({"input":"测试"})print(result)# {"input": "测试", "output": "处理结果: 测试"}

三、条件边：让流程“会拐弯”
条件边是 LangGraph 实现循环和分支的核心机制。它通过一个路由函数根据当前状态动态决定下一个节点。

fromlanggraph.graphimportStateGraph,START,ENDdefshould_continue(state:State)->str:"""路由函数：根据状态决定下一步"""ifstate.get("needs_tool"):return"call_tool"elifstate.get("is_done"):return"end"else:return"continue"builder.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 checkpointer=InMemorySaver()graph=builder.compile(checkpointer=checkpointer)# 每个线程有唯一的 thread_idconfig={"configurable":{"thread_id":"user-123"}}# 第一次执行result=graph.invoke({"input":"开始任务"},config=config)# 后续执行可以恢复之前的状态result2=graph.invoke({"input":"继续"},config=config)

Checkpointer 的价值：会话可以暂停和恢复。这对于长时间运行的任务和人工审批流程至关重要。

五、中断与恢复：Human-in-the-Loop
Human-in-the-Loop（HITL）模式让 Agent 在关键节点暂停执行，等待人工确认后再继续。

5.1 基本用法

fromlanggraph.typesimportinterrupt,Commanddefapproval_node(state:State)->State:"""需要人工审批的节点"""# interrupt() 会暂停执行，返回人工输入的值human_decision=interrupt({"question":"是否批准这个操作？","operation":state.get("pending_operation"),})ifhuman_decision=="approved":return{"status":"approved"}else:return{"status":"rejected"}

执行流程：

图以初始状态启动

执行到 interrupt() 时暂停，返回一个中断对象

调用方检查中断对象，获取需要人工决策的 payload

调用方通过 Command(resume=…) 恢复执行，传入人工输入

interrupt() 返回人工输入的值，节点继续执行

5.2 恢复执行

# 第一次执行：会停在 interruptresult=graph.invoke(initial_state,config=config)print(result["__interrupt__"])# 查看中断信息# 恢复执行：传入人工决策final=graph.invoke(Command(resume="approved"),# 人工批准config=config)

关键注意：恢复执行时，必须使用发生中断时使用的同一个 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:"""对用户问题进行分类"""query=state["user_query"].lower()if"退款"inqueryor"退钱"inquery:category="refund"elif"订单"inqueryor"物流"inquery:category="order"else:category="other"return{"category":category}defroute_by_category(state:CustomerServiceState)->str:"""根据分类路由到不同处理节点"""category=state["category"]ifcategory=="refund":return"refund_handler"elifcategory=="order":return"order_handler"else:return"human_escalation"deforder_handler(state:CustomerServiceState)->CustomerServiceState:"""处理订单查询"""result=check_order_status.invoke({"order_id":"ORD-001"})return{"response":result,"resolved":True}defrefund_handler(state:CustomerServiceState)->CustomerServiceState:"""处理退款请求，需要人工审批"""ifstate.get("human_decision")=="approved":result=process_refund.invoke({"order_id":"ORD-001"})return{"response":result,"resolved":True}else:# 触发中断，等待人工审批decision=interrupt({"question":"是否批准退款？","order_id":"ORD-001",})ifdecision=="approved":result=process_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}# ===== 构建图 =====builder=StateGraph(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)checkpointer=InMemorySaver()graph=builder.compile(checkpointer=checkpointer)# ===== 执行 =====config={"configurable":{"thread_id":"customer-001"}}# 测试订单查询（自动处理）result=graph.invoke({"user_query":"我的订单到哪了？"},config=config)print(result["response"])# 订单 ORD-001 已发货，预计明天到达。# 测试退款（需要人工审批）config2={"configurable":{"thread_id":"customer-002"}}result=graph.invoke({"user_query":"我要退款"},config=config2)print(result["__interrupt__"])# 触发中断# 人工批准后恢复final=graph.invoke(Command(resume="approved"),config=config2)print(final["response"])# 订单 ORD-001 的退款申请已提交。