Skip to main content

03 · LangGraph：把 Agent 当成一台可暂停的状态机

仓库langchain-ai/langgraph
Star39.9k（2026-08-19）
版本1.2.11
语言Python / TypeScript
许可证MIT
层级Runtime 运行时（LangChain 和 DeepAgents 都跑在它上面）
一句话不是「更强的 LangChain」，而是给 Agent 用的持久执行引擎

一、它解决的不是 LLM 问题，是分布式系统问题​

先看一个真实场景：

一个合同审查 Agent，跑到第 8 步要人工签字。审批人第二天早上才看邮件。

用普通框架，你有两个选择：进程挂在那儿等 12 小时，或者从头重跑。LangGraph 给的第三个选择是：把状态写进数据库，进程退出，明天开个新进程从第 8 步继续。

这就是它的定位 —— durable execution（持久执行）。同一层里还有 Temporal、Inngest 这些通用引擎，它们不懂 LLM 但持久执行做得同样好；LangGraph 的差异是它把 LLM 场景（消息归并、流式、工具中断）做成了一等公民。

四个核心能力：

能力含义
Durable execution崩溃 / 退出后从检查点恢复，不是从头
Streaming状态级、步骤级、token 级三种粒度
Human-in-the-loop任意位置中断，人改完状态再继续
Persistence线程内（checkpointer）+ 跨线程（store）双层

二、核心抽象：State + Node + Edge​

LangGraph 的心智模型只有三个词：

  • State：一个 TypedDict，整张图共享
  • Node：一个函数，(state) -> 状态更新片段
  • Edge：谁跑完之后跑谁，可以是条件的
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# ① State：整张图共享的一份数据，所有节点读它、改它
class State(TypedDict):
    # Annotated[类型, reducer]：add_messages 就是 reducer，
    # 它规定「节点返回新消息时是追加，而不是覆盖」——详见下一小节
    messages: Annotated[list, add_messages]

# ② Node：一个普通函数，入参是当前状态，返回「要改哪些字段」
def call_model(state: State):
    # 只返回 messages 这一个字段的增量，其余字段原样保留
    return {"messages": [llm.invoke(state["messages"])]}

# ③ Edge：把节点连起来，决定谁跑完之后跑谁
builder = StateGraph(State)
builder.add_node("model", call_model)     # 注册节点，名字随便起
builder.add_edge(START, "model")          # 入口 → model
builder.add_edge("model", END)            # model → 结束
graph = builder.compile()                 # 编译成可执行对象

Reducer：整个设计里最关键的一行​

Annotated[list, add_messages] 的意思是：节点返回 {"messages": [x]} 时，不是把 messages 覆盖成 [x]，而是追加。

没有 reducer 会怎样？

无 reducer（默认覆盖）有 add_messages
节点返回 {"messages": [新消息]}历史全没了追加到历史后面
两个节点并行返回后写的赢，前面的丢两边都保留

Reducer 是 LangGraph 能安全做并行分支的根本原因 —— 它把「多个节点同时改同一个字段」从竞态变成了确定性合并。这和 CRDT 的思路是一样的。

三、Checkpointer：状态落盘，随时可续​

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

checkpointer = InMemorySaver()   # 存「图状态快照」——会话内的短期记忆（生产要换 Postgres）
store = InMemoryStore()          # 存「应用数据」——跨会话的长期记忆

# 编译时把两者挂上，图才具备持久化能力
graph = builder.compile(checkpointer=checkpointer, store=store)

result = graph.invoke(
    {"messages": [{"role": "user", "content": "Hi, my name is Bob."}]},
    # thread_id 是这次会话的唯一标识，也是你的「持久化游标」：
    # 下次还传 thread-1 就接着上次的状态跑；换一个值就是全新会话
    {"configurable": {"thread_id": "thread-1"}},
)

Checkpoint 全景

图片来源：LangGraph Docs — Persistence

Checkpointer vs Store：两套持久化，别搞混​

CheckpointerStore
存什么图状态快照应用自定义的 KV 数据
作用域单个 thread跨 thread
记忆类型短期、会话内长期、跨会话
用于对话连续、HITL、时间旅行、容错用户偏好、事实、共享知识
怎么访问config 里传 thread_id节点里读写 item

thread_id 是你的持久化游标 —— 复用同一个 id 就接着上次跑，换一个就是全新会话。

生产环境的坑​

坑后果解法
用 InMemorySaver / MemorySaver 上线进程重启状态全丢换 PostgresSaver（生产）或 SqliteSaver（本地）
thread_id 太长Postgres 报列长度错误控制在 255 字符内，用 UUID 或 hash
checkpoint 无限增长延迟和存储成本飙升定期清理 / 设置保留策略

四、Interrupt：真正意义上的人工介入​

from langgraph.types import interrupt

def approval_node(state: State):
    # 执行到这一行：图立刻暂停，状态被 checkpointer 写进数据库，
    # 括号里的内容会抛给调用方（前端拿去渲染审批弹窗）。
    # 进程可以就地退出，人几小时后再来批也没关系。
    approved = interrupt("Do you approve this action?")
    # 恢复执行时，人给的答复会成为 interrupt() 的返回值落到 approved 上
    return {"approved": approved}

恢复：

from langgraph.types import Command

config = {"configurable": {"thread_id": "thread-1"}}

# 第一次运行：跑到 interrupt() 那行就停下
stream = graph.stream_events({"input": "data"}, config=config, version="v3")
final = stream.output

if stream.interrupted:              # True 表示这次运行是「停下等人」而不是跑完了
    print(stream.interrupts)        # 里面就是 interrupt() 传出来的内容
    # > (Interrupt(value='Do you approve this action?'),)

# ——— 这中间可以隔几小时、几天，进程也可以重启 ———

# 恢复：必须用同一个 thread_id，否则加载不到那个检查点。
# Command(resume=True) 里的 True 会成为 interrupt() 的返回值
resumed = graph.stream_events(Command(resume=True), config=config, version="v3")

断点

图片来源：LangGraph Docs — Interrupts

三条容易踩的规则：

  1. 恢复必须用同一个 thread_id，否则加载不到那个检查点。
  2. 恢复时节点是从头重跑的 —— interrupt() 之前的代码会再执行一遍。所以 interrupt() 前面不要放有副作用的操作（发邮件、扣款、写库）。
  3. Command(resume=...) 是唯一能当输入传给 invoke / stream 的 Command 形式；Command(update=...)、Command(goto=...) 是给节点函数返回用的。
为什么这比「回调式 HITL」强一个数量级

回调式要求你的进程一直活着等人。interrupt 式把状态写进 Postgres 后进程可以直接退出 —— 审批人三天后点「同意」，起一个新进程从断点继续。只有第二种能进真实的企业审批流。

五、多智能体：拓扑由你画​

多智能体架构

图片来源：LangGraph Docs — Multi-agent

LangGraph 不预设拓扑，两个最常用的：

Supervisor（主管制） —— 一个中心节点决定下一步派给谁：

Supervisor

Swarm（蜂群制） —— Agent 之间直接交接，没有中心：

Swarm

节点里返回 Command 可以同时更新状态并跳转，这是写路由最干净的方式：

from langgraph.types import Command
from typing import Literal

# 返回类型里的 Literal 列出所有可能去向，LangGraph 靠它推断出图的连线
def supervisor(state: State) -> Command[Literal["researcher", "writer", "__end__"]]:
    decision = route_with_llm(state)     # 让模型决定这一步该派给谁
    # Command 一次干两件事：goto 跳到哪个节点，update 顺手改状态。
    # 比「先 add_conditional_edges 再写路由函数」少绕一圈
    return Command(goto=decision, update={"next_step": decision})

子图（Subgraph） 是上下文隔离的手段 —— 子图有自己的 state schema，父图只看到约定的接口：

Subgraph

六、流式：三种粒度​

模式你收到什么用在哪
values每步之后的完整状态需要全量快照的场景
updates每步的增量更新大多数场景，省带宽
messagesLLM 的 token 流打字机效果
custom你在节点里手动 emit 的数据自定义进度条

values vs updates

图片来源：LangGraph Docs — Streaming

七、什么时候用 / 什么时候别用​

用它，如果​

  • 任务会跑很久（分钟到天级），中途可能崩、可能要等人
  • 要审计每一步 —— 检查点天然是完整的执行历史，可以时间旅行回放
  • 拓扑不是「循环到结束」 —— 有分支、并行、回环、多阶段
  • 要做混合控制 —— 大流程写死、局部交给模型，见 01 D2 维度

别用它，如果​

  • 你的 Agent 就是「调几个工具然后回答」 —— 用 LangChain create_agent 或 OpenAI Agents SDK，图的开销纯属自找麻烦
  • 团队不熟状态机 —— reducer、checkpoint、interrupt 语义有真实学习成本，没人带很容易写出难调的图
  • 只是想做 Demo —— 显式建图的代码量是声明式框架的好几倍
  • 公司已经在用 Temporal —— 那可能直接用 Temporal + 裸模型 SDK 更省一层依赖

八、常见误解​

误解实际
「LangGraph 是 LangChain 2.0」不是。LangChain 跑在 LangGraph 上，是上层框架
「用 LangGraph 就得用 LangChain」不用。LangGraph 可以直接调任何模型 SDK，节点里写什么都行
「图 = 工作流引擎，不够 agentic」图能表达「循环直到模型说完」，自主性由你在节点里放多少决定
「checkpointer 就是聊天记录」它存的是整个图状态，包括中间变量、待办、分支位置
参考资料