LangGraph[2] ---- 控制流与持久化:让图“活”起来

前言

在上一篇博客里，我们认识了 LangGraph 的三要素——状态、节点和边，并且用它们搭出了一个会循环的简单图。但那个例子离真正的“智能体”还差得远：它只会机械地重复同样的话，没有根据对话内容做出判断的能力，而且一旦跑起来就再也停不下，更无法在中间插手。

今天这篇，我们就为图注入真正的决策能力和记忆。你将学到：

• 如何用条件边实现智能路由，让 LLM 决定下一步做什么
• 如何安全地构建思考-行动循环，模拟一个 Agent 的核心逻辑
• 如何通过检查点让图拥有持久化“记忆”，支持暂停、恢复和时间旅行
• 如何嵌入人机交互，让关键操作必须经过人工审批才能继续

读完这篇，你就能搭出真正“活”的流程：它会根据情况选择不同的分支，会在适当的时候停下来等你点头，也会记住自己做过的事。

1. 条件边：让图学会“看情况”

条件边的本质，是在节点执行完之后，根据状态中的某些信息动态选择下一个节点。这给了我们实现“意图识别”、“超时处理”、“工具调用循环”等高级模式的基础。

1.1 路由函数的写法

add_conditional_edges需要三个参数：

• 源节点名称
• 一个路由函数，接收当前状态，返回一个字符串（代表分支名称）
• 一个映射字典，把分支名称映射到实际的目标节点（或END）

来看一个实际的例子：假设我们的聊天机器人要能区分“普通闲聊”和“需要计算”两类问题。我们先用一个 LLM 节点来分析用户的意图，然后根据意图走不同的分支。

fromtypingimportTypedDict,Annotated,Literalfromlanggraph.graphimportStateGraph,END,add_messagesfromlangchain_core.messagesimportHumanMessage,AIMessage# 定义状态classState(TypedDict):messages:Annotated[list,add_messages]intent:str# 新增一个字段，存放意图分类结果# 意图分类节点defclassify_intent(state:State)->dict:# 模拟分类逻辑：如果用户消息包含“计算”，则归类为 mathlast_msg=state["messages"][-1].contentif"计算"inlast_msg:return{"intent":"math"}else:return{"intent":"chat"}# 闲聊处理节点defchat_node(state:State)->dict:return{"messages":[AIMessage(content="这是个闲聊问题，我简单回复一下。")]}# 数学处理节点defmath_node(state:State)->dict:return{"messages":[AIMessage(content="这是需要计算的问题，我来认真算一下。")]}# 路由函数：根据 intent 字段决定下一步defroute_by_intent(state:State)->Literal["chat_node","math_node"]:returnstate["intent"]# 直接返回 "chat" 或 "math"# 构建图graph=StateGraph(State)graph.add_node("classify",classify_intent)graph.add_node("chat_node",chat_node)graph.add_node("math_node",math_node)graph.set_entry_point("classify")graph.add_conditional_edges("classify",route_by_intent,{"chat":"chat_node","math":"math_node"})# 两个处理节点执行完直接结束graph.add_edge("chat_node",END)graph.add_edge("math_node",END)app=graph.compile()

测试一下：

initial_state={"messages":[HumanMessage(content="帮我计算 2+3")],"intent":""}result=app.invoke(initial_state)print(result["messages"][-1].content)# 输出：这是需要计算的问题，我来认真算一下。

路由函数route_by_intent根据状态里的intent字段，决定走哪个分支。注意它的返回值只能是映射字典中存在的键，否则会报错。这种模式非常清晰——节点负责产生状态，边负责读取状态并决定流向。

2. 安全的循环：Agent 的思考-行动循环

真正的 Agent 往往需要多次调用 LLM，每次调用都可能请求新的工具或产生新的思考结果。这个过程天然就是循环的。但循环也带来了风险：如果终止条件没写好，图就会一直跑下去，直到资源耗尽。

LangGraph 为我们提供了内建的最大步数限制（recursion_limit），但更优雅的做法是在循环体内自己控制何时跳出，就像在 while 循环里写break。

2.1 一个极简的 ReAct 循环

我们用伪代码模拟一个“智能体”做决策：每次运行“思考”节点，如果它觉得可以结束了，就返回一条不包含工具调用的消息；否则继续调用工具，然后再次思考。

classAgentState(TypedDict):messages:Annotated[list,add_messages]defthink(state:AgentState)->dict:# 模拟 LLM 的思考：如果消息数量已经较多，就决定停止msg_count=len(state["messages"])ifmsg_count<6:# 返回一个要求调用工具的消息（模拟）return{"messages":[AIMessage(content="我需要更多信息，调用 search 工具。")]}else:return{"messages":[AIMessage(content="我已收集足够信息，最终答案：……")]}defshould_continue(state:AgentState)->Literal["action","end"]:last_message=state["messages"][-1]# 如果最后一条消息是最终答案，则结束if"最终答案"inlast_message.content:return"end"else:return"action"# 假设的工具节点deftake_action(state:AgentState)->dict:# 模拟执行工具并返回结果return{"messages":[AIMessage(content="搜索结果：……")]}graph=StateGraph(AgentState)graph.add_node("think",think)graph.add_node("action",take_action)graph.set_entry_point("think")graph.add_conditional_edges("think",should_continue,{"action":"action","end":END})graph.add_edge("action","think")# 行动后再次思考app=graph.compile()

执行这个图，它会按照think → action → think → action → ...的路径循环，直到think节点产出一条包含“最终答案”的消息，才走向END。这就实现了一个安全的、由自身状态决定的终止条件。

在实际开发中，你可以把think替换成真正的 LLM 调用，让模型自己决定是调用工具还是直接回答，终止条件则变为“消息列表最后一条不是工具调用请求”。这个模式我们会在第三篇博客中完整实现。

3. 持久化与检查点：为图装上“记忆”

到目前为止，我们的图每次执行都是无状态的：跑完一次invoke后，内部状态就消失了。但在真实场景中，我们可能需要：

• 暂停执行，几天后从同一点继续
• 查看某次对话的完整历史轨迹
• 回溯到之前的某个步骤，重新选择一条分支

这些能力都依赖于检查点（Checkpointer）。

3.1 内置检查点

LangGraph 内置了内存级检查点MemorySaver，它把所有状态快照保存在内存字典里。使用时，只需在compile时传入它，并为每次执行提供一个唯一的thread_id。

fromlanggraph.checkpoint.memoryimportMemorySaver checkpointer=MemorySaver()app=graph.compile(checkpointer=checkpointer)config={"configurable":{"thread_id":"user-001"}}initial_state={"messages":[HumanMessage(content="开始会话")]}# 第一次执行，每个步骤自动保存状态result=app.invoke(initial_state,config=config)

MemorySaver会在每一个步骤（节点执行之后）自动保存一份状态快照，键是thread_id和自动生成的checkpoint_id。你可以随时查看某个会话的当前状态：

current_state=app.get_state(config)print(current_state.values)# 当前状态字典print(current_state.next)# 下一个将要执行的节点（空则表示已结束）

3.2 暂停与恢复

因为每一步都有快照，所以你可以随时“重启”一个已经结束的图——只需用相同的config再次调用invoke或stream，图会从最后保存的状态开始继续执行。

如果图由于某种原因抛出了异常，状态也会保存在最后一个成功的步骤处，你可以修正输入后，用同一个config继续执行（通常需要配合人工介入，见下一节）。

3.3 时间旅行：回溯历史

检查点还允许你重新播放历史。通过get_state_history获取某个线程的所有快照，然后取出某个旧快照的config（包含checkpoint_id），就可以从那个点重新执行，选择另一条分支。

history=list(app.get_state_history(config))old_state=history[2]# 第三个快照old_config=old_state.config# 用旧配置重新执行，从那个时间点开始app.invoke(None,config=old_config)

这在调试和 A/B 测试中非常有用：你可以回到“LLM 生成回复”之前，给模型换一个提示词，看看回复会怎么变。

生产环境中，你应当把MemorySaver换成持久化存储，如SqliteSaver或AsyncPostgresSaver，这样即使服务重启，会话也不会丢失。

4. 人机交互：关键时刻让人来把关

在很多业务场景里，Agent 不能擅自行动：退款申请、对外发送邮件、调用高风险 API……这些动作必须经过人工审核。LangGraph 提供了Human-in-the-loop（HITL）机制，允许图在指定节点前/后暂停，等待人类干预。

4.1 中断节点的执行

在compile时，可以传入interrupt_before或interrupt_after参数，指定哪些节点要被“打断”。

app=graph.compile(checkpointer=checkpointer,interrupt_before=["take_action"]# 在执行 take_action 之前暂停)

此时，当图运行到take_action节点前时，会自动抛出GraphInterrupt异常（实际上是暂停并返回当前状态，你可以通过get_state获取）。此时图处于“等待人工确认”的状态，你可以查看当前状态，决定是批准、修改还是拒绝。

4.2 暂停后如何处理

实践中，我们通常使用stream或astream来逐步运行图，因为它可以产生多个中间状态，不会因为中断而抛出异常导致流程终结。当遇到中断时，stream会停止产生新的事件，你可以在外部代码中捕获当前状态，展示给审批人。

更直接的方法是用invoke，它会抛出一个GraphInterrupt异常，你捕获异常后可以获取状态。

不过更推荐的是使用update_state和后续的继续执行流程。以下是典型的处理步骤：

1. 用app.stream(initial_state, config)逐步运行，直到某个节点被中断。
2. 中断发生后，不再有新的流事件。此时调用app.get_state(config)查看当前状态和下一步。
3. 如果需要修改状态（比如人工编辑了 Agent 生成的草稿），调用app.update_state(config, values={"messages": [HumanMessage(content="审批人修改的回复")]})。
4. 然后可以调用app.stream(None, config)继续执行，图会从中断的节点之前/之后继续。

4.3 完整示例：带人工审批的回复发送

我们来构建一个真实场景：用户提问 → Agent 生成回复草稿 → 人工审批 → 发送回复。

fromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,Annotatedfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportHumanMessage,AIMessageclassChatState(TypedDict):messages:Annotated[list,add_messages]draft:strapproved:bool# 节点：LLM 生成回复defgenerate_draft(state:ChatState)->dict:# 实际中调用 LLM，这里模拟user_msg=state["messages"][-1].content draft=f"这是对 '{user_msg}' 的自动回复草稿。"return{"draft":draft,"approved":False}# 节点：发送回复（仅当 approved）defsend_reply(state:ChatState)->dict:return{"messages":[AIMessage(content=state["draft"])]}# 路由：根据 approved 字段决定是否发送defreview_decision(state:ChatState):ifstate["approved"]:return"send"else:return"end"# 未批准，直接结束（不发）graph=StateGraph(ChatState)graph.add_node("generate_draft",generate_draft)graph.add_node("send_reply",send_reply)graph.set_entry_point("generate_draft")graph.add_conditional_edges("generate_draft",review_decision,{"send":"send_reply","end":END})graph.add_edge("send_reply",END)# 在 generate_draft 之后中断，让人工审核checkpointer=MemorySaver()app=graph.compile(checkpointer=checkpointer,interrupt_after=["generate_draft"]# 生成草稿后暂停)config={"configurable":{"thread_id":"chat-1"}}initial_state={"messages":[HumanMessage(content="请帮我退款")],"draft":"","approved":False}# 用 stream 运行，遇到中断时会停止events=list(app.stream(initial_state,config))print("流程暂停，当前事件：",events)state=app.get_state(config)print("草稿内容：",state.values["draft"])

这时，审批人看到草稿觉得不满意，想修改回复内容并批准。我们通过update_state把修改后的草稿和approved=True写回状态，然后继续执行。

# 人工审批：修改草稿并批准app.update_state(config,values={"draft":"亲，您的退款已处理，请查收。","approved":True})# 继续执行，从中断点之后开始（即条件边判断阶段）foreventinapp.stream(None,config):print(event)final_state=app.get_state(config)print("最终发送的消息：",final_state.values["messages"][-1].content)

输出将是人工修改后的那条消息。整个过程中，Agent 绝不可能在审批通过前擅自发送回复。

中断位置的选择很灵活：interrupt_before在节点执行前暂停，让你有机会阻止它运行；interrupt_after在节点执行后暂停，让你可以审阅它的输出并决定是否继续。你甚至可以在同一张图中组合使用多个中断点。

5. 完整动手实验：智能客服审批流

现在，我们把上面的知识点串联起来，构建一个稍复杂的智能客服：

• 用户提问，模型先进行意图分类（普通问题 / 投诉）
• 如果是投诉，必须生成处理草稿并送人工审批，审批通过才能发送；如果人工驳回，则重新生成。
• 如果是普通问题，直接回复。
• 整个对话过程被持久化保存，支持暂停恢复。

这里给出核心代码框架：

# 状态定义classCustomerState(TypedDict):messages:Annotated[list,add_messages]intent:strdraft:strapproved:bool# 节点：意图分类defclassify(state:CustomerState)->dict:last=state["messages"][-1].content intent="complaint"if"投诉"inlastelse"general"return{"intent":intent}# 节点：生成回复（普通）defgeneral_reply(state:CustomerState)->dict:return{"messages":[AIMessage(content="这是普通问题的自动回复。")]}# 节点：生成投诉草稿defdraft_complaint_reply(state:CustomerState)->dict:return{"draft":"【投诉草稿】我们将尽快处理您的问题。","approved":False}# 节点：发送草稿defsend_reply(state:CustomerState)->dict:return{"messages":[AIMessage(content=state["draft"])]}# 路由函数：根据意图分流defroute_intent(state:CustomerState):ifstate["intent"]=="complaint":return"draft_complaint_reply"else:return"general_reply"# 审批判断：检查 approved 字段defcheck_approval(state:CustomerState):ifstate["approved"]:return"send_reply"else:return"end"# 构建图builder=StateGraph(CustomerState)builder.add_node("classify",classify)builder.add_node("general_reply",general_reply)builder.add_node("draft_complaint_reply",draft_complaint_reply)builder.add_node("send_reply",send_reply)builder.set_entry_point("classify")builder.add_conditional_edges("classify",route_intent,{"general_reply":"general_reply","draft_complaint_reply":"draft_complaint_reply"})builder.add_edge("general_reply",END)builder.add_conditional_edges("draft_complaint_reply",check_approval,{"send_reply":"send_reply","end":END})builder.add_edge("send_reply",END)# 编译并开启持久化和中断checkpointer=MemorySaver()app=builder.compile(checkpointer=checkpointer,interrupt_after=["draft_complaint_reply"]# 草稿生成后暂停，等待审批)

使用方式与上一节的审批示例完全一致，只是这里增加了意图分流。这样，我们就拥有了一套完整的、可干预、有记忆的智能客服引擎。

6. 小结与下篇预告

今天我们把 LangGraph 的控制流和持久化能力挖掘了一遍：

• 条件边让图能够根据状态做出动态决策，实现意图路由、循环跳出等模式。
• 循环是 Agent 思考-行动逻辑的骨架，配合明确的终止条件可以安全运行。
• 检查点为图提供了事务性的状态保存，支撑暂停、恢复、时间旅行等能力。
• 人机交互通过中断机制和update_state，让人安全地介入关键决策，防止模型乱来。

有了这些，你的图已经是一个“活”的系统了。但它还缺少最后一块拼图：如何跟真正的工具（搜索、计算、API）集成，并且以流式方式把执行过程实时推送给前端？

下一篇博客，我们将接上工具调用和流式输出，构建一个完整的、生产可用的 LangGraph Agent，并探讨部署与可观测性。