LangGraph 让你用状态机精确控制 Agent 流转

2025 年的 Agent 世界已进入“生产落地深水区”。几个月前，你或许还能用一串精心编写的提示词和几个 IF-ELSE 让单个 Agent “看起来”很智能；但当需求演进到让三个 Agent 协同完成一份合同审查——一个负责抽取条款、一个负责合规检查、一个负责生成修订建议——你会发现，那些藏在 prompt 里的隐性约定正在崩塌。某个 Agent 返回了非预期格式的 JSON，下游就彻底卡死；你找不到任何日志来复现那个“偶尔发生的”路径分叉。

这就是 LangGraph 要解决的核心问题。它不是又一个“更好用的 LLM 封装器”，而是一个将 Agent 控制流编译为确定性状态机的运行时框架。本章将带你从零构建一个可调试、可回放、可精确控制流转的多 Agent 工作流——就像编写普通的工程代码那样，而不是像在黑暗中掷骰子。

你需要什么

• 运行环境：Python 3.10+，建议使用虚拟环境
• 依赖包：langgraph、langchain-core、langchain-anthropic（或 langchain-openai）
• API 密钥：Anthropic API Key（本教程使用 Claude Skills 示例）或 OpenAI API Key
• 预计时间：45 分钟（含编码与调试）

最终成果

你将得到一个能处理文档审阅请求的三节点工作流：

1. 分类节点判断文档类型（合同 / 技术规范 / 通用文本）
2. 技能节点调用对应的 Claude Skill 执行专业处理
3. 审核节点在人工视图中展示结果，等待确认或修改

最终成果不是一段 demo 视频，而是一个你可以在项目里继承、扩展的可工程化基础架构。

为什么做这个？多 Agent 系统真正的成本不是 API 调用，而是逻辑失控后的排查时间。用 LangGraph 把流转逻辑显式化为图，你就获得了一张“逻辑地图”——无论系统多复杂，你都能精准定位每一步的状态变化。

步骤 1：定义共享状态

在 LangGraph 中，所有节点共享一个“状态字典”。这相当于贯穿整个工作流的数据总线。

from typing import TypedDict, Union, Annotated
from langgraph.graph import StateGraph, END
import operator

# 定义状态的形状
class AgentState(TypedDict):
    # 使用 reducer：operator.add 表示追加到列表，而非覆盖
    messages: Annotated[list, operator.add]  # 消息历史
    document_type: str                       # 分类结果
    draft_result: str                        # 技能处理结果
    final_output: str                        # 人工确认后的最终输出
    next_step: str                           # 用于条件路由的控制字段

预期结果：一个强类型的状态字典。其中 messages 使用了 operator.add 作为 reducer，这意味着每次节点返回 {"messages": new_msgs} 时，这些消息会被追加到已有列表，而不是替换。对于并行节点（比如同时调用多个技能），这种设计能避免状态覆盖的并发问题。

踩坑提醒：reducer 缺失的后果

如果某个字段（比如 draft_result）未定义 reducer，多次写入会直接覆盖旧值。这听起来无害，但当你后续加上重试机制、让节点可能被重复执行时，你会发现中间结果神秘消失。规则很简单：

要追加的字段用 operator.add，要覆盖的字段用默认行为（不指定 reducer），要累加的数值用自定义 reducer。

步骤 2：构建图的基本块——节点、边、条件边

现在开始用 StateGraph 将三个步骤注册为图的节点，并定义它们之间的流转。

2.1 创建 Graph 实例并注册节点

# 初始化状态图
graph = StateGraph(AgentState)

# 定义三个节点的逻辑（目前还是占位函数）
def classify_document(state: AgentState) -> AgentState:
    """节点1：文档分类"""
    print("→ [分类节点] 正在分析文档类型...")
    # 稍后会接入 LLM 调用
    return {"document_type": "contract", "next_step": "route_to_skill"}

def execute_skill(state: AgentState) -> AgentState:
    """节点2：执行技能处理"""
    print(f"→ [技能节点] 正在处理 {state['document_type']} ...")
    return {"draft_result": "条款合规检查完成，发现2处风险点。", "next_step": "review"}

def human_review(state: AgentState) -> AgentState:
    """节点3：人工审核（稍后会改为交互式）"""
    print(f"→ [审核节点] 待审核结果：{state['draft_result']}")
    return {"final_output": state['draft_result'], "next_step": "end"}

# 注册节点：第一个参数是节点名（用于路由），第二个是函数
graph.add_node("classify", classify_document)
graph.add_node("skill", execute_skill)
graph.add_node("review", human_review)

2.2 定义普通边与条件边

# ========== 普通边 ==========
# 设置入口点：图启动时第一个执行的节点
graph.set_entry_point("classify")

# ========== 条件边 ==========
def router(state: AgentState) -> str:
    """根据 next_step 字段决定下一个节点"""
    if state["next_step"] == "route_to_skill":
        return "skill"
    elif state["next_step"] == "review":
        return "review"
    elif state["next_step"] == "end":
        return END  # LangGraph 内置常量，表示终止
    else:
        raise ValueError(f"未知路由步骤: {state['next_step']}")

# 从 "classify" 出发的边由 router 函数决定
graph.add_conditional_edges("classify", router, {
    "skill": "skill",
    "review": "review",
    END: END
})

# 从 "skill" 到 "review" 是固定路径
graph.add_edge("skill", "review")

# 从 "review" 结束后，根据条件判断
graph.add_conditional_edges("review", router, {
    END: END
})

关键解释：条件边的工作方式是——每次执行完一个节点，调用 router 函数，根据返回值映射到目标节点。映射字典 {"skill": "skill", ...} 不是冗余，它声明了所有可能的路径，LangGraph 在 compile() 时做拓扑校验，检测出悬挂节点或不可达分支。

步骤 3：编译并运行第一个版本

# 编译图（验证拓扑，生成运行时）
app = graph.compile()

# 运行：传入初始状态
initial_state = {
    "messages": [],
    "document_type": "",
    "draft_result": "",
    "final_output": "",
    "next_step": "route_to_skill"  # 初始路由指令
}

result = app.invoke(initial_state)
print(f"\n最终输出: {result['final_output']}")

预期输出：

→ [分类节点] 正在分析文档类型...
→ [技能节点] 正在处理 contract ...
→ [审核节点] 待审核结果：条款合规检查完成，发现2处风险点。

最终输出: 条款合规检查完成，发现2处风险点。

此时，你拥有的是一个确定性的、可追溯的控制流。任何一个 next_step 的变更都能精准指向下一个节点，这比“根据 prompt 描述让 LLM 自己决定下一步”要可靠得多。

视觉断点：这是你的工作流图

┌──────────┐   route_to_skill   ┌──────────┐      ┌──────────┐
│ classify ├──────────────────►│   skill   ├─────►│  review  │
└──────────┘                    └──────────┘      └─────┬────┘
     │                                                  │
     │ 其他条件                                         │ end
     ▼                                                  ▼
   (END)                                              (END)

这张图就是你的 Agent 的“逻辑电路图”。当系统出现非预期行为时，你能直接在图上标注出哪条边触发了问题，而不是在几百行日志中猜。

步骤 4：与 Claude Skills 的集成模式——将已有技能封装为节点

前面章节中，你应当已经积累了一些 Claude Skills（代码解释、合同审查、文档总结等）。现在，把这些技能“装进”LangGraph 节点。

from langchain_anthropic import ChatAnthropic
import anthropic

def build_skill_node(skill_name: str, skill_prompt: str):
    """工厂函数：为指定技能创建一个节点函数"""
    client = anthropic.Anthropic()  # 确保环境变量 ANTHROPIC_API_KEY 已设置

    def skill_node(state: AgentState) -> AgentState:
        # 从消息历史中获取最新用户输入
        user_input = state["messages"][-1]["content"] if state["messages"] else ""

        # 构建完整的技能提示
        full_prompt = f"{skill_prompt}\n\n上下文: {user_input}"

        # 调用 Claude（这里使用原生 API，你也可以用 LangChain 封装）
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": full_prompt}]
        )

        return {
            "draft_result": response.content[0].text,
            "next_step": "review",
            # 将这次对话记录追加到消息历史
            "messages": [{"role": "assistant", "content": response.content[0].text}]
        }

    return skill_node

# 技能定义（通常从外部加载，这里简写）
skills_registry = {
    "contract_review": {
        "prompt": "你是一位资深合同律师，请审查以下文档中的风险条款。",
        "doc_types": ["contract", "agreement"]
    },
    "tech_spec_check": {
        "prompt": "你是一位技术架构师，请检查技术规范的一致性和可行性。",
        "doc_types": ["technical_spec", "design_doc"]
    }
}

# 动态注册技能节点
for skill_id, skill_config in skills_registry.items():
    # 创建节点函数
    node_func = build_skill_node(skill_id, skill_config["prompt"])
    # 注册到图
    graph.add_node(skill_id, node_func)

重要细节：这里是混合编排的关键点。LangGraph 节点不关心内部是调用 LLM、运行 SQL 查询还是执行本地脚本——只要它返回符合 AgentState 的字典。这意味着你可以把 Claude Skills 封装为“黑盒”节点，在图中像搭积木一样组装它们，同时保留 Skill 内部的 prompt 工程优势。

动态路由增强：根据文档类型分发到不同技能

现在更新 router 和分类逻辑，让工作流真正做到不同的输入走不同的处理路径。

def classify_document(state: AgentState) -> AgentState:
    """增强版分类节点"""
    # ... LLM 分类逻辑 ...
    doc_type = "contract"  # 示例结果

    # 映射文档类型到技能
    for skill_id, config in skills_registry.items():
        if doc_type in config["doc_types"]:
            return {"document_type": doc_type, "next_step": skill_id}

    # 无匹配技能时直接送审
    return {"document_type": doc_type, "next_step": "review"}

def router(state: AgentState) -> str:
    next_step = state["next_step"]
    # 检查是否是注册过的技能节点
    if next_step in skills_registry:
        return next_step
    elif next_step == "review":
        return "review"
    elif next_step == "end":
        return END
    else:
        # 生产代码中应记录告警并走降级路径
        return "review"

步骤 5：检查点与人工干预——让关键决策回到人类手中

在所有自动化工作流中，最终输出的质量控制必须经过人类判断。LangGraph 的检查点机制让你可以在任意节点暂停执行，等待人类输入后再恢复。这不是简单的“弹个窗口”，而是状态保存、中断、恢复的完整生命周期。

5.1 启用检查点并定义中断点

from langgraph.checkpoint.memory import MemorySaver

# 使用内存保存检查点（生产环境应替换为 PostgresSaver）
checkpointer = MemorySaver()

# 重新编译图，并传入检查点
app = graph.compile(
    checkpointer=checkpointer,  # 关键参数
    # 在 review 节点之前暂停
    interrupt_before=["review"]
)

5.2 交互式人工审核

# 运行到 "review" 节点前会自动暂停
thread_config = {"configurable": {"thread_id": "doc-review-001"}}

# 启动执行
result = app.invoke(initial_state, thread_config)
# 此时状态是中断后的快照

print(f"中断前技能输出：{app.get_state(thread_config).values['draft_result']}")
user_decision = input("是否批准此结果？(yes/no/edit): ")

if user_decision.lower() == "yes":
    # 恢复执行，更新状态中的审批标记
    app.update_state(thread_config, {"approval": "approved", "next_step": "end"})
    final_result = app.invoke(None, thread_config)  # None 表示从当前状态继续
elif user_decision.lower() == "edit":
    # 人类修改结果
    edited = input("请输入修改后的文本: ")
    app.update_state(thread_config, {"draft_result": edited, "next_step": "end"})
    final_result = app.invoke(None, thread_config)
else:
    print("任务已取消")

预期行为：

1. 序列执行 classify → skill，然后在执行 review 节点之前暂停
2. 控制台打印技能输出，等待用户输入
3. 用户批准后，继续执行 review 节点，或直接跳转到结束

踩坑经验：这些细节会让你少花两天 debug

1. invoke(None, config) 是恢复，不是重新开始
当工作流被中断后，调用 invoke(None, thread_config) 表示“从我上次停下的地方继续”。如果错误地传入了新的状态字典，你会覆盖整个快照，导致前序执行结果全部丢失。

2. 条件边里的映射字典必须完整
缺少任何可能返回值的映射，compile() 会抛 InvalidUpdateError。防御性编程建议：始终加上一个兜底映射 "fallback": "review" 或其他安全节点。

3. 状态中的 messages 字段必须用 reducer
不配置 operator.add 时，每次节点返回 messages 都会覆盖历史。你会在调试时发现“明明我记得有 3 条消息，怎么只剩 1 条了”——这就是原因。

4. 检查点在生产环境必须持久化
MemorySaver 只能用于开发和测试。生产环境请替换为 SqliteSaver 或 PostgresSaver，否则服务器重启后所有中断状态都会丢失。

回顾：你做了什么，花了多久

总计 45 分钟，你完成了：

1. 理解 LangGraph 的核心理念：把 Agent 控制流显式表达为有向图
2. 用 StateGraph 构建了一个分类→技能处理→人工审核的三步工作流
3. 学会了 reducer、条件边、编译验证三个工程化关键机制
4. 将 Claude Skills 封装为可复用的 LangGraph 节点
5. 接入了人工干预检查点，实现了暂停-修改-恢复的完整循环

你获得的不只是一个 demo，而是一个可扩展的确定性 Agent 编排框架。当你的同事还在为“Agent 为什么跳过了某个步骤”而困惑时，你可以直接打开状态历史，指给他看第 37 行的 next_step 值。

行动清单

1. 创建你的第一个 StateGraph：用本章的代码骨架，把自己的一个简易 Agent 拆成 2-3 个节点，体验显式控制的便利
2. 给列表类字段加上 reducer：检查现有代码，所有 messages 类字段都改为 Annotated[list, operator.add]
3. 引入一个人工审核节点：在关键输出节点前插入检查点，配置 interrupt_before，运行一次完整的中断-恢复流程
4. 设计 5 个以内的 next_step 值：把它们当作工作流的“指令集”，避免使用自然语言描述路由（如 "请让skill处理" 这种字符串）
5. 画一张你的 Agent 拓扑图：每次加新节点前，先在纸上或 Mermaid 中画出所有边，再编码——这是成本最低的 bug 预防手段

下一章：AutoGen 通过对话驱动实现更自然的 Agent 合作

LangGraph 给了你精确到每一张边的控制力——正因如此，它适合那些“逻辑需要被证明正确”的任务。但在探索性场景中（比如让两个 Agent 自由讨论方案、互相挑战假设），这种确定性的图结构反而成了束缚。这时，你会希望 Agent 能像人类团队那样，在对话中自然涌现协调，而不是被预先编排。

下一章，我们将切换到 AutoGen 的世界，看它如何以“对话即协议”的哲学，让你用几行代码构建出能够自主谈判、相互纠错的 Agent 团队。你会在两者的对比中，清晰地看到：何时该精确控制，何时该放任涌现。