从脚本调度到状态机编排：LangGraph 异常中断与检查点恢复实战

摘要

在构建多智能体系统时，传统的脚本调度方式难以应对复杂的状态管理和错误恢复需求。本文通过一个 Researcher→Coder→Reviewer 三智能体协作的实战案例，深入剖析 LangGraph 框架的核心能力。我们将重点澄清一个常见误区：业务异常中断与框架级 interrupt 机制的区别，并展示如何利用 MemorySaver 实现状态持久化与断点恢复。同时，文章会纠正条件路由的反模式用法，给出更符合 LangGraph 设计哲学的线性流程实现。通过本文，你将掌握构建生产级多智能体工作流的关键技术，包括状态图编排、检查点快照原理、自定义 Reducer 以及生产环境持久化方案的选择。

问题背景：脚本调用的天花板

想象一个典型的智能体协作场景：Researcher 负责调研技术方案，Coder 根据调研结果编写代码，Reviewer 最终审查并输出报告。在原型阶段，我们可能会这样实现：

1
2
3

research_result = research_agent(task)
code_result = coder_agent(research_result)
final_answer = reviewer_agent(code_result)

这种串行调用看似简单，但一旦遇到以下问题就会变得难以维护：

状态丢失：某个智能体执行失败后，整个流程需要从头开始
错误处理混乱：异常散布在各处，难以统一管理
可观测性差：无法精确追踪每个智能体的输入输出
扩展性受限：增加分支逻辑或重试机制需要大量样板代码

这正是状态机编排框架的用武之地。LangGraph 通过有向图模型，将智能体间的协作抽象为状态节点和边的组合，从根本上解决了上述痛点。

技术方案：LangGraph 的编排哲学

LangGraph 的核心思想很简单：将多智能体工作流建模为一个状态图。每个节点代表一个处理步骤（如智能体调用），边定义了状态流转的路径。共享的 State 对象在节点间传递，确保每个智能体都能访问到上下文。

相比传统调度，LangGraph 提供了三个关键能力：

状态化编排：所有节点共享同一个状态对象，天然支持上下文传递
检查点机制：每次节点执行后自动保存状态快照，支持精确恢复
灵活的控制流：支持条件路由、循环、并行等复杂拓扑

但这里需要澄清一个常见误区：业务异常与框架中断是两回事。业务异常（如 API 调用失败）应该通过 try/except 捕获并写入状态，而框架级的 interrupt 机制（通过 interrupt_before/interrupt_after 或 Command 实现）用于在特定节点前/后暂停执行，等待外部输入。本文的 Demo 使用异常传播来触发中断，这是一种简化方案，适合演示检查点恢复的能力。生产环境中应优先使用显式的 interrupt 机制。

核心实现解析

1. 定义共享状态

状态是工作流的灵魂。我们使用 Pydantic 的 BaseModel 定义明确的字段和类型：

from pydantic import BaseModel, Field
from typing import List, Dict, Any

class AgentState(BaseModel):
    messages: List[Dict[str, str]] = Field(default_factory=list)
    task: str = ""
    research_result: str = ""
    code_result: str = ""
    final_answer: str = ""
    errors: List[str] = Field(default_factory=list)

每个字段都定义了默认值，LangGraph 的 Reducer 机制会确保状态合并的正确性。对于列表字段，默认使用 add 操作（追加合并），标量字段则使用 replace 操作（覆盖合并）。

2. 实现智能体节点

每个智能体节点都是一个接收 AgentState 并返回更新的函数：

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def research_agent(state: AgentState) -> Dict[str, Any]:
    """调研任务并返回发现"""
    try:
        prompt = f"Research the following topic and provide key findings: {state.task}"
        response = llm.invoke(prompt)
        return {"research_result": response.content}
    except Exception as e:
        return {"errors": [f"Research failed: {str(e)}"]}

def coder_agent(state: AgentState) -> Dict[str, Any]:
    """根据调研结果编写代码"""
    try:
        prompt = f"Based on this research: {state.research_result}\nWrite production-quality code."
        response = llm.invoke(prompt)
        return {"code_result": response.content}
    except Exception as e:
        return {"errors": [f"Coding failed: {str(e)}"]}

def reviewer_agent(state: AgentState) -> Dict[str, Any]:
    """审查代码并输出最终答案"""
    try:
        prompt = f"Review this code: {state.code_result}\nProvide final answer with explanation."
        response = llm.invoke(prompt)
        return {"final_answer": response.content}
    except Exception as e:
        return {"errors": [f"Review failed: {str(e)}"]}

注意每个节点都包含 try/except，将异常信息写入 errors 字段。这是业务异常处理的正确方式——不影响图执行，而是将错误信息传递给后续节点或路由逻辑。

3. 构建图并配置检查点

这是文章需要重点修正的部分。对于线性工作流，最简洁的方式是直接使用 add_edge：

from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver

# 初始化图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("researcher", research_agent)
workflow.add_node("coder", coder_agent)
workflow.add_node("reviewer", reviewer_agent)

# 设置入口点
workflow.set_entry_point("researcher")

# 线性连接：researcher → coder → reviewer → END
workflow.add_edge("researcher", "coder")
workflow.add_edge("coder", "reviewer")
workflow.add_edge("reviewer", END)

# 配置 MemorySaver 检查点
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

这里的核心改进有三点：

去除了条件路由：线性流程用 add_edge 直接连接，避免不必要的路由逻辑
正确配置检查点：MemorySaver 会在每个节点执行后自动保存状态快照
清晰的拓扑结构：一眼就能看出工作流的执行顺序

如果确实需要演示条件路由，应该设计一个明确的分支场景。例如，当 errors 字段非空时重试，否则进入下一节点：

def should_retry(state: AgentState) -> str:
    """根据错误信息决定是否重试"""
    if state.errors:
        return "retry"
    return "continue"

# 条件边示例（非当前 Demo 使用）
workflow.add_conditional_edges(
    "coder",
    should_retry,
    {
        "retry": "coder",     # 重试编码
        "continue": "reviewer"  # 进入审查
    }
)

4. 检查点机制原理

LangGraph 的检查点机制基于快照（Snapshot） 实现。每次节点执行完成后，框架会：

序列化当前状态对象
记录执行上下文（当前节点、父节点、配置等）
将快照存入配置的持久化后端

MemorySaver 将快照保存在内存中，适合开发和测试。生产环境应使用 PostgresSaver 或 RedisSaver：

# 生产环境持久化方案示例
from langgraph.checkpoint import PostgresSaver

# PostgreSQL 持久化
pg_saver = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/langgraph"
)
app = workflow.compile(checkpointer=pg_saver)

# 或 Redis 持久化
from langgraph.checkpoint import RedisSaver

redis_saver = RedisSaver.from_conn_string("redis://localhost:6379/0")
app = workflow.compile(checkpointer=redis_saver)

5. 执行与断点恢复

执行工作流并演示断点恢复：

import asyncio

async def run_workflow():
    # 初始状态
    initial_state = AgentState(
        task="Write a Python function to calculate Fibonacci numbers using recursion, and explain how it works."
    )
    
    config = {"configurable": {"thread_id": "fibonacci-demo"}}
    
    # 第一次执行：可能会因异常中断
    try:
        async for event in app.astream_events(initial_state, config, version="v2"):
            if event["event"] == "on_chain_end":
                print(f"Node completed: {event['name']}")
    except Exception as e:
        print(f"Workflow interrupted: {e}")
    
    # 检查当前状态
    state = app.get_state(config)
    print(f"Errors: {state.values.get('errors', [])}")
    
    # 恢复执行（LangGraph v0.2+ 标准范式）
    # 注意：这里使用 None 作为输入，表示从断点处继续
    # 实际生产环境中，可以通过 Command(resume=...) 传递恢复数据
    async for event in app.astream_events(None, config, version="v2"):
        if event["event"] == "on_chain_end":
            node_name = event["name"]
            if node_name == "reviewer":
                final_state = app.get_state(config)
                print(f"Final Answer: {final_state.values['final_answer']}")

关于 astream_events(None, config) 的说明：

适用边界：当工作流因异常中断后，None 表示从断点处继续执行，不传递新输入
潜在限制：如果中断发生在节点内部（而非节点之间），恢复时可能需要重新执行该节点
标准恢复范式：LangGraph v0.2+ 推荐使用 Command(resume=...) 显式指定恢复数据，适用于需要外部输入的场景

运行效果

执行上述代码后，你会看到类似以下的输出：

Node completed: researcher
Node completed: coder
Workflow interrupted: Error in reviewer node
Errors: ['Review failed: API timeout']
--- 恢复执行 ---
Node completed: reviewer
Final Answer: Here's a recursive Fibonacci function...

整个过程展示了：

Researcher 和 Coder 节点成功执行
Reviewer 节点因 API 超时抛出异常
检查点机制保存了 Researcher 和 Coder 的结果
恢复执行时，Reviewer 从断点处重新运行
最终输出完整的代码和解释

总结与展望

本文通过一个实战案例，展示了 LangGraph 在多智能体编排中的核心能力。我们澄清了业务异常与框架中断的区别，纠正了条件路由的反模式，并深入探讨了检查点机制的原理和生产环境持久化方案。

关键 takeaways：

**线性工作流优先使用 add_edge**：避免不必要的条件路由，保持代码简洁
异常处理与中断分离：业务异常写入状态，框架中断使用显式 interrupt 机制
检查点是核心能力：MemorySaver 适合开发，生产环境选择 PostgresSaver/RedisSaver
恢复执行需谨慎：理解 astream_events(None, config) 的适用边界，生产环境使用 Command(resume=...)

未来方向：

探索 LangGraph 的 Command 机制实现精确的中断与恢复
结合 Human-in-the-loop 模式，在关键节点引入人工审核
使用 PostgresSaver 实现跨进程的状态持久化，支持分布式部署

状态机编排不是银弹，但对于复杂多智能体工作流，它提供的状态管理、错误恢复和可观测性能力，是传统脚本调度无法比拟的。希望本文能帮助你正确理解和使用 LangGraph，构建更健壮的多智能体系统。