19. LangGraph 状态机编排

很多人第一次接触 LangChain， 会觉得 Runnable 已经够用了。 Prompt、Model、Parser、 再加几个工具调用， 一个链条就能跑起来。

问题出在系统一旦变复杂， 链式抽象很快就会失真。 你真正拥有的， 已经不是一条线， 而是一个会分支、会回退、会暂停、会恢复、 还可能把任务交给别的 Agent 的流程图。

这时如果仍坚持用线性的 Runnable 组合来表达， 代码表面上还能写， 但控制流会开始散落在各种 if、 回调、 外层 while 循环和临时变量里。 调试时你看到的是一堆函数调用， 而不是业务上真实存在的执行路径。

LangGraph 解决的不是“再发明一种链”。 它解决的是： 当 Agent 进入多步、 长时、 有状态的阶段之后， 如何把流程本身提升为一等公民。

它的基本判断非常直接： Agent 不是若干 Runnable 的自然堆叠， 而是一个围绕共享状态运转的状态机。 节点负责执行， 边负责流转， 状态负责让每一步都能看见同一份上下文， 而 checkpoint 负责让这份状态可以被保存、 恢复、 审计。

为什么链式 Runnable 不够用

Runnable 的长处在于顺手。 把一个输入依次交给几个步骤处理， 这类场景它非常自然。 例如“先改写问题，再检索，再总结”， 本来就是线性的。

但 Agent 工作流有三个特征， 会让这种线性抽象逐渐吃力。

第一， 它经常需要条件分支。 模型判断“需要先查资料”还是“可以直接回答”， 这不是链条上的下一个固定步骤， 而是路由。

第二， 它需要共享状态。 你不只是在传递一个字符串， 而是在维护 messages、 工具结果、 中间产物、 审批结论、 重试计数、 当前负责人等多种字段。

第三， 它需要暂停与恢复。 只要你引入 Human-in-the-Loop、 长任务、 异步执行或失败恢复， 就必须承认一次执行不会总是从头跑到尾。

Runnable 当然也能勉强表达这些能力。 你可以在某个节点里手写 if， 可以在外层维护一个字典， 也可以自己把中间状态落进数据库。 问题是， 这些能力不再被框架显式建模， 而是退化成“你自己想办法补”。

一旦系统进入生产， 真正稀缺的不是“还能不能实现”， 而是“别人能不能看懂这套流程、 能不能恢复、 能不能审计、 能不能在第二个月还敢改”。

LangGraph 的价值， 就在于把这些工程问题直接收编进了抽象层。

StateGraph 的四个核心抽象

如果只记 LangGraph 的一件事， 应该记住 StateGraph。 它不是在一堆 Runnable 外再包一层皮， 而是让你先定义“共享状态长什么样”， 再把节点和边挂上去。

Node：执行单元

Node 就是一个执行函数。 它接收当前 state， 做一段明确的工作， 返回对 state 的局部更新。

这个更新不是“重建整份状态”， 而是只返回本步新增或修改的字段。 编排层再把这些更新合并回共享 state。

Edge：无条件转移

Edge 表示从一个节点稳定流向下一个节点。 它适合表达没有分歧的控制流， 例如“检索完成后一定进入总结”。

Conditional Edge：路由器

Conditional Edge 是真正把图和链分开的地方。 它允许你写一个路由函数， 根据当前 state 决定去哪个节点， 甚至直接结束。

这比把 if 写在节点内部更清晰。 节点只负责干活， 路由器负责决定下一步。 职责分工更像真实的工作流系统。

State：共享数据结构

State 是整个图的中心。 你通常会用 TypedDict 或 dataclass 定义它的字段。 关键不只是“有哪些字段”， 还包括“同名字段如何合并”。

LangGraph 支持为某些字段声明 reducer。 例如消息列表、 日志列表、 产物列表这类字段， 更适合追加而不是覆盖。

这一步很重要。 很多第一次用 LangGraph 的人， 只是把 state 当成一个大字典。 真正有价值的地方其实在于： 你把“状态如何演化”也变成了显式设计。

一个最小可跑的 LangGraph

先看一个足够小、 但已经具备实战味道的例子。 它有三个节点： analyze、 search_docs、 answer。 其中 analyze 后面接一个条件路由， 决定是否需要检索； 整个图使用 SQLite checkpoint 持久化， 因此可以按 thread_id 恢复执行。

from __future__ import annotations

from operator import add
from typing import Annotated, Literal, TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph


# 共享状态定义：
# messages 和 docs 都用 add reducer，表示返回值会追加到列表里。
class AgentState(TypedDict):
    question: str
    route: str
    messages: Annotated[list[BaseMessage], add]
    docs: Annotated[list[str], add]
    answer: str


def analyze(state: AgentState) -> dict:
    # 这里为了示例可运行，用关键词代替真正的 LLM 分类。
    # 实战里通常会在这个节点里调用模型判断“是否需要检索”。
    question = state["question"]
    needs_search = "LangGraph" in question or "原理" in question
    route = "search" if needs_search else "direct"

    return {
        "route": route,
        "messages": [
            AIMessage(content=f"analyze 节点判断路由为: {route}")
        ],
    }


def search_docs(state: AgentState) -> dict:
    # 模拟检索系统返回的文档片段。
    # 真实项目里这里可以换成向量库、SQL 或外部 API。
    return {
        "docs": [
            "LangGraph 用图和共享状态表达多步工作流。",
            "Checkpoint 可以让执行在中断后继续。",
        ],
        "messages": [
            AIMessage(content="search_docs 节点补充了两段参考资料。")
        ],
    }


def answer(state: AgentState) -> dict:
    # 最终回答节点统一收敛，避免“直接回答”和“检索后回答”两套出口逻辑。
    docs_text = "\n".join(state["docs"]) if state["docs"] else "无额外资料"
    final_answer = (
        f"问题: {state['question']}\n"
        f"检索资料:\n{docs_text}\n"
        "结论: 这是一个可恢复、可路由的最小 LangGraph 示例。"
    )
    return {
        "answer": final_answer,
        "messages": [AIMessage(content="answer 节点生成了最终答案。")],
    }


def route_after_analyze(state: AgentState) -> Literal["search_docs", "answer"]:
    # 条件边的路由函数只负责决定去哪里，不做额外副作用。
    return "search_docs" if state["route"] == "search" else "answer"


builder = StateGraph(AgentState)
builder.add_node("analyze", analyze)
builder.add_node("search_docs", search_docs)
builder.add_node("answer", answer)

builder.add_edge(START, "analyze")
builder.add_conditional_edges("analyze", route_after_analyze)
builder.add_edge("search_docs", "answer")
builder.add_edge("answer", END)


# SQLite checkpointer 会把每一步状态存下来。
# thread_id 相同的调用，会沿着同一条执行线程继续。
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "demo-thread-1"}}
    result = graph.invoke(
        {
            "question": "LangGraph 的原理是什么？",
            "route": "",
            "messages": [HumanMessage(content="请解释一下 LangGraph。")],
            "docs": [],
            "answer": "",
        },
        config=config,
    )

    print(result["answer"])

这个例子里最值得注意的， 不是节点函数本身， 而是控制流终于被写在了图上。 你一眼能看出入口在哪里， 分支在哪里， 收敛点在哪里。

如果以后要加一个 human_review 节点， 或者在 search_docs 后面插一个 rerank， 修改的是图结构， 而不是在某个巨大函数里继续堆 if。

多 Agent 编排：supervisor、handoff、subgraph

LangGraph 真正发力的场景， 通常不是单 Agent 三节点， 而是多 Agent 协作。 这里要区分三种常见组织方式。

第一种是 supervisor。 一个总控节点根据任务状态决定当前该由谁工作。 它自己不一定产出最终内容， 但负责维持队列、 切换角色、 判断是否收束。

第二种是 handoff。 这不是“多个 Agent 同时说话”， 而是当前负责人把上下文交接给下一位。 交接后的核心问题， 不是语言上说“请你继续”， 而是状态里有没有一个明确字段能表明谁是 active agent， 以及交接时保留哪些共享产物。

第三种是 subgraph。 当某个子流程内部已经足够复杂， 例如“研究团队”本身又要规划、 检索、 去重、 总结， 最好的做法不是把所有节点塞进总图， 而是先把它编译成一个子图， 再把子图当成父图的一个节点。

下面看一个精简写法。 它展示 supervisor、 handoff 和 subgraph 是如何协同的。

from typing import Annotated, Literal, TypedDict
from operator import add

from langgraph.graph import END, START, StateGraph


class TeamState(TypedDict):
    task: str
    active_agent: str
    notes: Annotated[list[str], add]
    final_report: str


def supervisor(state: TeamState) -> dict:
    # supervisor 根据已有笔记决定下一位负责人。
    # 实战里这里通常由模型输出路由标签。
    if not state["notes"]:
        return {"active_agent": "researcher"}
    return {"active_agent": "writer"}


def handoff_router(state: TeamState) -> Literal["research_team", "writer", "__end__"]:
    if state["active_agent"] == "researcher":
        return "research_team"
    if state["active_agent"] == "writer":
        return "writer"
    return "__end__"


def research_plan(state: TeamState) -> dict:
    return {"notes": [f"研究计划: 先搜集与 {state['task']} 相关的事实。"]}


def research_collect(state: TeamState) -> dict:
    return {"notes": ["补充事实: LangGraph 擅长长时状态编排。"]}


research_builder = StateGraph(TeamState)
research_builder.add_node("research_plan", research_plan)
research_builder.add_node("research_collect", research_collect)
research_builder.add_edge(START, "research_plan")
research_builder.add_edge("research_plan", "research_collect")
research_builder.add_edge("research_collect", END)
research_subgraph = research_builder.compile()


def writer(state: TeamState) -> dict:
    joined_notes = "\n".join(state["notes"])
    return {
        "final_report": f"任务: {state['task']}\n研究记录:\n{joined_notes}",
        "active_agent": "done",
    }


team_builder = StateGraph(TeamState)
team_builder.add_node("supervisor", supervisor)
team_builder.add_node("research_team", research_subgraph)
team_builder.add_node("writer", writer)

team_builder.add_edge(START, "supervisor")
team_builder.add_conditional_edges("supervisor", handoff_router)
team_builder.add_edge("research_team", "supervisor")
team_builder.add_edge("writer", END)

team_graph = team_builder.compile()

这个模式的好处在于层次分明。 总图只关心谁接棒， 子图只关心本团队内部如何完成任务。 如果未来研究团队再拆成检索、 验证、 摘要三步， 父图一行都不用动。

这里也能看出 LangGraph 和很多“多 Agent 框架”的根本差异。 它并不试图先定义一套人格化的员工世界， 而是先把协作关系还原成图和状态。 角色可以有， 但角色只是状态机上的一个组织方式， 不是抽象的出发点。

与 Runnable 的取舍

LangGraph 不是 Runnable 的替代品， 更像是 Runnable 的上层编排器。 很多节点内部照样可以调用 Prompt、 Model、 Parser 组成的小链。

真正的分界线在于： 如果你的流程是“单次请求、 线性步骤、 失败就整体重跑”， Runnable 仍然更轻。 你不需要为了一个三步摘要流程， 硬把系统建成图。

但只要出现下面任意一个信号， LangGraph 往往更合适。

• 需要条件路由或循环。
• 需要暂停、 恢复、 回放。
• 需要长时共享状态而不是一次性参数传递。
• 需要把多 Agent 协作写成可视化、 可审计的执行图。

这不是“LangGraph 更先进”， 而是两者优化目标不同。 Runnable 优先快速把能力串起来， LangGraph 优先把执行过程工程化。

与 CrewAI、Claude Agent SDK 的编排哲学差异

CrewAI 的默认叙事是团队协作。 你先定义 researcher、 writer、 manager 这类角色， 再让任务在角色之间流动。 它更像把组织结构做成一等公民。

LangGraph 的默认叙事不是“谁是谁”， 而是“状态怎样流动”。 角色可以映射成节点， 也可以映射成路由结果， 但本质仍是图。 因此它在需要精确控制执行路径时更稳， 代价是写法没有 CrewAI 那么直观。

Claude Agent SDK 则走另一端。 它倾向于保留一个非常薄的 Agent 循环， 把大部分控制交还给开发者。 这在单 Agent、 TypeScript、 直接围绕 Anthropic 能力开发时很舒服， 但复杂编排要你自己补结构。

如果把三者放在一条轴上看， 大致可以这样理解： Claude Agent SDK 最薄， CrewAI 最强调角色语义， LangGraph 最强调显式控制流和持久化状态。

生产里的关键点

把图跑起来只是开始。 真正进入生产， 有三类能力会决定 LangGraph 是不是只是“图画得漂亮”。

interrupt：Human-in-the-Loop

前面第 15 篇讲过， 高风险动作必须让人能介入。 在 LangGraph 里， 这件事之所以自然， 是因为图本来就是一步一步推进的， 每一步之后都可以 checkpoint。

因此审批点最适合设计成明确节点。 例如 draft_email 之后进入 human_review， 只有审批通过才流向 send_email。 流程图上看得见， 状态里留得下， 失败后也能从审批节点继续。

time travel：回到历史状态再分叉

很多团队第一次觉得 LangGraph 值回票价， 不是因为“能画图”， 而是因为它让你真正可以回到过去的某个状态重新走。

这对调试极其有用。 你不必每次都从头重现一个十五步任务。 只要第十二步路由错了， 就从第十一步的 checkpoint 分叉， 改路由逻辑再跑。

checkpoint 持久化：存储怎么选

存储选型不要神化。 原则其实很朴素。

开发机和单机原型， SQLite 足够。 它部署成本最低， 方便本地重放， 也容易直接查看状态。

需要多人共享、 多实例恢复、 更强事务语义时， Postgres 更稳。 它不是专门为 Agent 生的， 但对“保存结构化状态、 按 thread_id 查询、 做审计”这件事已经非常成熟。

如果你追求极低延迟、 状态体积小、 更偏临时会话， Redis 也可以考虑。 但要意识到它在审计、 历史追溯和复杂查询上通常不如关系型数据库顺手。

真正的经验是： 不要只看“存得下”， 要看“出事故时能不能查、 能不能恢复、 能不能解释”。

本篇要点

• LangGraph 解决的核心问题不是把链写成图，而是把多步 Agent 的控制流、共享状态与恢复点显式建模。
• StateGraph 的四个关键抽象是 Node、Edge、Conditional Edge 与 State，其中 reducer 决定状态如何安全合并。
• 最小可跑示例里，三个节点加一个条件路由就已经能体现 LangGraph 的价值；再接上 checkpoint，流程才真正具备生产味道。
• 多 Agent 编排时，supervisor 负责总控，handoff 负责交接，subgraph 负责把复杂子流程封装成更高层节点。
• 与 Runnable 相比，LangGraph 更重但更可控；与 CrewAI、Claude Agent SDK 相比，它最强调状态机与持久化编排。
• 生产里真正关键的是 interrupt、time travel 和 checkpoint 存储选型，而不是图画得多复杂。

参考资料

• LangGraph Docs
• LangChain Runnable Interface
• LangGraph Persistence
• LangSmith