LangGraph状态图与工作流编排#

1. 为什么从 Chain 走向 Graph#

LCEL 链 prompt | llm | parser 是 DAG 且单向：不能回到上一步、条件分支只能塞进 RunnableBranch、更做不到”调工具→看结果→再决定”的循环。而 Agent 本质是状态机：ReAct 循环、失败重试、多 Agent 交接、等人审批，都要求”节点可重入 + 状态可持久化”。LangGraph 把这些从”藏在 AgentExecutor 黑盒里的 while 循环”变成显式、可画、可断点的图，调试和人工介入才有入口。

2. 核心抽象：State / Node / Edge / 编译#

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import InMemorySaver

class State(TypedDict):
    messages: Annotated[list, add_messages]      # 追加语义，而非覆盖
llm_with_tools = llm.bind_tools(tools)
def agent(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}   # 只返回增量

g = StateGraph(State)
g.add_node("agent", agent)
g.add_node("tools", ToolNode(tools))
g.add_edge(START, "agent")
g.add_conditional_edges("agent", tools_condition)   # 有 tool_calls → "tools"，否则 → END
g.add_edge("tools", "agent")                         # 工具结果回流，形成循环
app = g.compile(checkpointer=InMemorySaver())                     # 最小 ReAct 循环
app.invoke({"messages": [("user", "北京今天天气？")]}, {"configurable": {"thread_id": "u-1"}, "recursion_limit": 50})

3. 执行模型：超步、并行与递归上限#

LangGraph 借鉴 Google Pregel 的 BSP 模型：一次 super-step 里，所有被激活的节点并行执行，各自返回的更新在本步结束时由 reducer 统一合并写入 state，然后根据边计算下一批激活节点。几个推论：

• 并行分支 / fan-in：一个节点连出多条固定边就是 fan-out；它们在同一超步并行跑，汇聚节点要等所有上游分支完成才执行。并行节点写同一个无 reducer 字段会报 InvalidUpdateError——并行写入必须用 reducer（如 operator.add）。
• recursion_limit：限制一次 invoke 的超步数，默认 25，超过抛 GraphRecursionError，是防 ReAct 死循环的最后一道限制；在 invoke/stream 的 config 里按需调大（不是在 compile() 里），别当成 max_tokens 用。
• 预置 Agent：langgraph.prebuilt 的 create_react_agent(model, tools) 就是上面那张图的封装（agent 节点 + ToolNode + tools_condition），LangGraph 1.0 起已标记废弃，官方推荐改用 LangChain 1.x 的 create_agent（同样构建在 LangGraph 之上，并支持 middleware）。先用预置快速起步，需要自定义节点（如检索、反思、审批）再手写图。

4. 持久化、线程与 Human-in-the-loop#

• Checkpointer（通用原理——何时存、存什么、恢复三策略——见 [Agent Runtime与Checkpoint机制](/topics/ai-agent/Agent Runtime与Checkpoint机制)，这里只讲用法）：compile(checkpointer=...) 后，每个超步结束自动存一份 checkpoint。开发用 InMemorySaver（进程内，旧名 MemorySaver，重启即丢），单机持久化用 SqliteSaver，生产用 PostgresSaver（独立包 langgraph-checkpoint-postgres），也可自实现 BaseCheckpointSaver。
• thread_id：config={"configurable": {"thread_id": ...}} 标识一条会话线程，同线程多次 invoke 自动续上历史消息——多轮记忆不需要自己拼 history。线程内记忆靠 checkpointer，跨线程长期记忆靠 Store（InMemoryStore/Postgres），见 Agent记忆与上下文工程。
• Time travel：get_state_history(config) 列出每个超步的快照，取某个 checkpoint_id 放进 config 再 invoke，就从那一步分叉重跑；update_state() 可先人工改 state 再继续——调试和”回到出错前一步”都靠它。
• Human-in-the-loop：动态断点——节点内调 interrupt(payload) 抛出中断，图暂停并把 payload 交给调用方，审批后用 invoke(Command(resume=value), config) 续跑，interrupt() 返回 value；静态断点——compile(interrupt_before=["execute"])，invoke(None, config) 继续。两者都依赖 checkpointer。坑：恢复时该节点从头重跑，interrupt() 之前的副作用（发请求、写库）会重复执行，要幂等或放到中断之后。

5. 子图、多 Agent、流式、生产与选型#

• 子图（Subgraph）：编译好的图可直接 add_node 当节点用。父子 共享 state key 时直接嵌入；schema 不同时用函数包一层做输入/输出映射。Supervisor 模式 = 主管节点按 state 决定 Command(goto="researcher")，子 Agent 完成后 Command(goto=..., graph=Command.PARENT) 交还控制权（handoff）；多 Agent 拓扑与取舍见 多Agent协作架构。
• 流式（stream_mode）：values 每个超步后推送完整 state；updates 只推送该步各节点的增量（带节点名，适合进度条）；messages 逐 token 推送 LLM 输出（消息块 + 元数据，适合打字机效果）；可传列表同时订阅多种。
• 生产与可观测：LangSmith 一键记录每个节点的输入输出与 token，是 LangGraph 的默认追踪后端；部署侧原 LangGraph Platform 已于 2025-10 更名为 LangSmith Deployment（LangGraph Studio 改名 LangSmith Studio，LangGraph Server 改称 Agent Server），负责部署、队列和后台任务，支持云托管、混合部署、自托管和单独部署 Agent Server。
• 生态关系：LangChain 提供模型/工具/Prompt 等组件抽象，LangGraph 是其底层编排运行时，LangChain 1.x 的 agent 就跑在 LangGraph 上，LangGraph 也能脱离 LangChain 组件单独用；Java 栈的对应物是 [Spring AI Alibaba框架](/topics/ai-agent/Spring AI Alibaba框架) 的 Graph——同一套 StateGraph / Node / 条件边 / Checkpoint 思想的类比实现。
• 选型：单次问答、“检索→生成”线性链、无状态批处理——LCEL 或裸 SDK 更短更好调；需要循环（ReAct/重试）、分支（意图路由）、持久化（多轮会话、断点恢复）、人审批、多 Agent 交接中的任意两项以上，样板代码的投入才回本；横向框架对比见 [AI Agent技术趋势与框架选型](/topics/ai-agent/AI Agent技术趋势与框架选型)。

LangGraph 是 LangChain 团队做的底层编排运行时，解决的是 LCEL 链式调用表达不了循环、分支和回退的问题——Agent 本质是状态机，ReAct 循环、失败重试、等人审批都要求节点可重入、状态可持久化。它的核心抽象就四个：State 是一个 TypedDict，每个字段可以带 reducer，比如 messages 用 add_messages 就是追加语义，不带 reducer 就是覆盖；Node 是纯函数，读 state 返回部分更新；Edge 分固定边和条件边，条件边靠一个路由函数读 state 决定下一步；最后 compile 成可 invoke 的图，编译时注入 checkpointer。执行模型借鉴 Pregel 的超步，同一步激活的节点并行跑、步末统一用 reducer 合并，所以并行写同一字段必须有 reducer，另外有 recursion_limit 默认 25 步限制，防止死循环。第二块是持久化：挂上 InMemorySaver 或 PostgresSaver，每个超步自动存 checkpoint，按 thread_id 区分会话，同一线程多次调用自动续上历史，还能 get_state_history 做 time travel，从某一步分叉重跑。第三块是 Human-in-the-loop，节点里调 interrupt() 图就暂停，审批后用 Command(resume=…) 续跑，注意恢复时该节点从头重跑，中断前的副作用要幂等。第四是子图和多 Agent，编译好的图可以当节点嵌进去，Supervisor 用 Command(goto=…) 做交接。流式有 values、updates、messages 三种粒度。选型上，简单线性流程别上它，需要循环、分支、持久化、审批这几项里两项以上才值；Java 栈对应的是 Spring AI Alibaba 的 Graph。

追问方向：

• reducer 到底解决什么问题？不写会怎样？ → 决定多个节点对同一字段的更新怎么合并。不写就是覆盖语义，两个并行节点同一超步写同一字段会直接抛 InvalidUpdateError；add_messages 除了追加还会按消息 id 合并，因此可以用同 id 的消息做”修改/删除”
• 条件边和 Command 怎么选？ → 路由只依赖 state、不需要改 state 时用 add_conditional_edges，图结构静态可画；节点既要写 state 又要决定去向（典型是多 Agent handoff）用返回 Command(goto=..., update=...)，少一个节点、少一次状态同步
• LangGraph 的 checkpoint 粒度是什么？跟自研有何区别？ → 每个超步结束存一份，包含全部 state 通道值、下一步待执行节点和元数据，按 thread_id + checkpoint_id 寻址；自研通常只在工具调用后存业务状态。代价是写放大，生产要定期清理旧 checkpoint，或按线程 TTL 归档
• interrupt() 和 interrupt_before 有什么区别？ → interrupt_before 是编译期静态断点，只能停在节点边界；interrupt() 是运行期动态断点，能在节点内任意位置、带 payload 暂停，并把 Command(resume=...) 的值当返回值拿到，审批/补充信息场景用它
• 多轮对话历史越来越长怎么办？ → checkpointer 只负责存，不负责裁剪。常见做法是加一个 summarize 节点，在超过阈值时把旧消息摘要后用 RemoveMessage（配合 add_messages 的 id 合并）删掉原文；或在 agent 节点调用模型前 trim

易错点：

• ❌ “Node 要返回完整 state” → 只返回改动的 key，返回整份不仅浪费，遇到 reducer 字段还会把历史重复追加一遍
• ❌ “recursion_limit 是 LLM 最大调用次数” → 它是超步数上限（默认 25），一个 ReAct 轮次要消耗 agent + tools 两个超步，实际循环次数约为其一半
• ❌ “interrupt() 暂停后从中断点精确继续” → 是该节点从头重跑，中断前的副作用会重复，必须幂等或后置