Agent 架构文档 · Markdown

LangGraph Agent 架构全景讲解

以 Agent 的生命脉络 为主线,讲清楚 LangGraph 这套"低层图编排框架"是怎么把 agent 的 状态、节点、边、记忆、流式输出 串成一个可恢复、可中断、可重放的 长生命周期工作流。 跟"包了一层 LLM 的 chain"不同,LangGraph 把 状态显式化、控制流显式化、持久化 显式化 ,让你既能用 create react agent

来源文件:langgraph-agent-architecture.md · 阅读时间 28 分钟

Agent 的生命脉络 为主线,讲清楚 LangGraph 这套"低层图编排框架"是怎么把 agent 的 状态、节点、边、记忆、流式输出 串成一个可恢复、可中断、可重放的 长生命周期工作流。

跟"包了一层 LLM 的 chain"不同,LangGraph 把 状态显式化、控制流显式化、持久化 显式化,让你既能用 create_react_agent 几行起一个 ReAct agent,也能下沉到 StateGraph 自己拼图,还能挂上 supervisor / swarm / deep agents 这些更高阶范式。

适合:刚从 LangChain chain 范式切过来想搞清楚"图到底图在哪"的同学 / 在做 multi-agent 编排选型的架构师 / 已经写了 agent 但 checkpoint / interrupt / streaming 没吃透的工程师。

2026-06-08 官方复核补丁:PyPI / GitHub langgraph 当前复核到 1.2.4; PyPI langgraph-prebuilt1.1.0langgraph-checkpoint4.1.1; npm @langchain/langgraph1.3.6@langchain/langgraph-cli1.2.5。 本文主体继续按图编排架构讲解,版本号以本条与官方包为准。


#目录

  1. 写在前面:LangGraph 是什么 / 不是什么 0.5. 一张图看懂:LangGraph = Model + Harness(积木而非成品)
  2. Agent 是什么 —— StateGraph + Node 函数 + state schema
  3. Agent 怎么活起来 —— invoke / astream / Cloud 三条触发链路
  4. Agent 怎么思考 —— ReAct prebuilt + 自定义图 + Supervisor / Swarm
  5. Agent 怎么行动 —— ToolNode + Handoff + Interrupt
  6. Agent 怎么记忆 —— state + Checkpointer + Store
  7. Agent 怎么开口 —— 五种流模式
  8. Agent 的外脑 —— LangChain ChatModel 抽象
  9. 关键设计权衡(架构师视角)
  10. 附录:组件地图 / 关键概念 / 源码索引

#零、写在前面:LangGraph 是什么 / 不是什么

"LangGraph is a low-level orchestration framework for building, managing, and deploying long-running, stateful agents." —— 官方仓库 README 的第一行。 注意三个词:low-level(不是"框架包好的 agent 模板")、stateful(状态是 一等公民)、long-running(按"会跑半小时、可恢复、可中断"的标准设计)。

#0.1 名称消歧:先把 "LangGraph" 这个词锁定

LangChain Inc 一家公司就出了五个相关品牌,写文档前先分清楚:

候选 是什么 跟本文的关系
LangGraph(python / js 库) 底层图编排库,PyPI langgraph,仓库 langchain-ai/langgraph ✅ 本文主角
LangGraph Studio 可视化调 graph + 改 state 的 GUI 🔵 本文捎带提一下,不展开
LangGraph Platform / Cloud / Deployment 托管运行 LangGraph 应用的服务(已并入 LangSmith Deployment) 🔵 §2.3 简述
LangChain LangGraph 的母库,提供 ChatModel / Tool / Retriever / 集成 🟡 §7 借它的 ChatModel 抽象
LangSmith observability + eval + deployment 平台 ⚪ 仅在 §6 / §8 提一下
LangChain create_agent LangChain 1.0 在 LangGraph 上重新封的 ReAct agent 工厂(带 middleware 系统),官方推荐替代 create_react_agent 🟡 §3 会对比 create_react_agent
deepagents 包 LangChain 团队在 create_agent 上做的"todo + 虚拟文件系统 + subagent"模板(早期版本基于 create_react_agent 🟢 §3.5 详解

本文重点:**LangGraph 库本身(langgraph + langgraph-prebuilt + langgraph-checkpoint*

  • 它推荐的 agent 范式(create_react_agent / supervisor / swarm / deep agents)**。 不讲 LangChain 老的 chains / LLMChain / AgentExecutor,那一套已经被官方判定 deprecated(LangChain 1.0 之后正式推荐迁到 create_agent,底层就是 LangGraph)。

主要 URL:

#0.2 核心矛盾:LangGraph 想回答什么

矛盾 LangGraph 的回答
LangChain 老 chain 的 state 是隐式的,agent 一旦循环就难调试 State 显式 TypedDict / Pydantic + 图节点 + 边路由,每一步都能从 checkpoint 看到状态切片
LLM 是无状态的,agent 却要长生命周期 / 跨会话 / 失败重试 Checkpointer(thread_id 短期)+ Store(namespace 长期),中断后从最近 checkpoint 继续
agent 偶尔需要人审一脚(敏感操作 / 高风险动作) interrupt() + Command(resume=...),图层面的 breakpoint,不靠业务自己埋点
多 agent 之间的协作,要么调度复杂,要么变成 supervisor 单点瓶颈 同一套 Command 原语撑起 supervisor / swarm / hierarchical / network 四种范式,按 trade-off 选
Agent UI 想看 token 流、看节点切换、看工具调用,老 chain 给不出 astream 五种模式(values / updates / messages / debug / custom) + astream_events 事件流
想做长任务(爬虫、深度研究),http 请求级别的部署搞不定 retry / 断点续跑 LangGraph Platform 用 task queue 跑长任务,state 由 checkpointer 持久化

每章都会回到这几条矛盾,看图编排范式具体怎么回应的。

#0.3 你将看到三层东西,请从一开始就分清楚

┌──────────────────────────────────────────────────────────────────┐
│ 第三层:高阶范式 / 模板                                            │
│   create_react_agent (langgraph-prebuilt)                         │
│   create_supervisor  (langgraph-supervisor-py)                    │
│   create_swarm       (langgraph-swarm-py)                         │
│   deepagents.create_deep_agent                                    │
│   LangChain 1.0 create_agent (在 langchain 包里,底层 LangGraph)   │
├──────────────────────────────────────────────────────────────────┤
│ 第二层:编排原语                                                  │
│   StateGraph / Node 函数 / Edge / Conditional Edge                │
│   Command(goto, update, resume)                                   │
│   Send (动态扇出, map-reduce)                                     │
│   interrupt() + Checkpointer + Store                              │
│   astream(stream_mode=...) / astream_events                       │
├──────────────────────────────────────────────────────────────────┤
│ 第一层:Pregel 执行引擎                                            │
│   libs/langgraph/langgraph/pregel/  —— 把图编译成 BSP 超步循环      │
│   channels(state 字段 = 通道,受 reducer 控制)                   │
│   Postgres / SQLite / Redis / Memory checkpointer 作为持久层       │
└──────────────────────────────────────────────────────────────────┘

下文每节会标清楚正在哪一层讲。你看到的所有"高阶 agent 模板",本质都只是把第二层 的原语预先拼成了一份 StateGraph——这点想清楚,整个 LangGraph 就通了。


#零·五、一张图看懂:LangGraph = Model + Harness(积木而非成品)

业界视角:Claude Code / Cursor / Codex 这些端到端 agent 产品的差距,不在底层模型, 而在"模型之上的外壳(Harness)"——指令、能力、基础设施、可观测性四层共同决定 agent 的真实上限。

LangGraph 的特殊定位:它不是端到端 agent 产品,而是一套搭 harness 的低层积木。 它把"状态、控制流、持久化、中断、流式"这些 harness 关键组件提取成图编排原语和可选后端 (Checkpointer/Store),但指令层(prompt)和能力层(tool 实现)几乎全部留给开发者自己写。 下面按"4 层 + 8 支柱"反向索引本文档每个章节,并诚实标注"框架给了什么 / 留给开发者什么"。

#0.5.1 四层 Harness 结构

┌──────────────────────────────────────────────────────────────────────┐
│       Agent  =  Model  (LLM)  +  Harness  (LangGraph 提供原语)         │
└──────────────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────────────┐
│ ① 指令层  │ ⚠ 框架不管:node 函数内 LLM 调用前的 prompt 拼装由开发者写  │
│           │ create_react_agent 的 prompt= 参数(str/callable/Runnable) → §3.1 │
│           │ deepagents 的 instructions + subagent prompt 模板          → §3.5 │
│           │ pre_model_hook 在进 LLM 前裁剪 / 注入 messages              → §3.1 │
├──────────────────────────────────────────────────────────────────────┤
│ ② 能力层  │ ToolNode:自动执行 AIMessage.tool_calls                    → §4.1 │
│           │ @tool / BaseTool:开发者自己写工具实现                      → §4.1 │
│           │ InjectedState / InjectedStore:tool 安全注入 state          → §4.2 │
│           │ Command-tool:tool 直接 goto 跨节点(handoff 原语)          → §4.3 │
│           │ deepagents 的 todo / fs / subagent 三件套(最接近"内置能力") → §3.5 │
├──────────────────────────────────────────────────────────────────────┤
│ ③ 基础设施│ State schema + channels + reducer(图的"血液"契约)          → §1.1 │
│           │ Checkpointer:thread_id 级 state 持久化                     → §5.2 │
│           │   ├─ MemorySaver / SqliteSaver / PostgresSaver / Redis      │
│           │   └─ ⚠ 是可选挂件,不挂就没短期持久化和 interrupt           │
│           │ Store:跨 thread 长期记忆(namespace + 可选向量)            → §5.4 │
│           │ interrupt() + Command(resume=) :图层面 HITL                → §4.4 │
│           │ Pregel BSP 引擎:superstep 调度 + 并发 fan-out               → §1 / 引擎 │
│           │ LangGraph Platform:长任务托管(Threads API / 后台 run)     → §2.3 │
├──────────────────────────────────────────────────────────────────────┤
│ ④ 可观测  │ astream 五种 stream_mode(values/updates/messages/debug/custom) → §6.1 │
│           │ astream_events:LangChain 通用事件协议                      → §6.2 │
│           │ get_state_history + Time Travel:任意 checkpoint 重放        → §5.3 │
│           │ LangSmith:trace / eval / cost(外挂,需注册)               → §6 / §8 │
│           │ LangGraph Studio:可视化 + 改 state                          → §0.1 │
└──────────────────────────────────────────────────────────────────────┘

#0.5.2 八支柱对照表

# Harness 支柱 一句话定义 LangGraph 的实现(框架给 vs 开发者写) 对应章节
01 FS + Git state 必须活在 context 之外 ✅ Checkpointer 持久化 state(PG/SQLite);deepagents 提供虚拟 FS 工具组(state/Store/磁盘/沙箱可切);❌ 普通 agent 没有"文件系统"概念,要写文件得自己挂 tool §5.2 / §3.5
02 Bash 通用工具 随用随装避免 token 爆炸 ❌ 框架不提供通用 shell;ToolNode 只是调度器,工具实现完全靠开发者;deepagents 的 fs 工具组是少数"框架附带能力" §4.1 / §3.5
03 沙箱 + 子工具 行动可隔离 + 自检 ⚠ 框架本身无沙箱;deepagents 的 subagent(task tool)提供上下文隔离的子 agent;外接 Modal/Daytona/Deno 沙箱要自己接 §3.5
04 记忆 + 搜索 跨 turn / 跨用户拿历史 ✅ Store 接口 + Postgres pgvector 实现(namespace + 语义检索);short-term 在 state.messages 里 reducer 合并;❌ 召回策略 / RAG pipeline 要开发者自己写 §5.4 / §5.1
05 对抗 Context Rot 长对话也别糊 ⚠ 给原语不给方案:pre_model_hook + trim_messages 让你裁剪,但何时压缩 / 怎么 summarize 完全是开发者的事;没有内置 condenser §3.1 / §5.1
06 长程执行 多 phase 任务不跑偏 ✅ 这是 LangGraph 最强项:StateGraph 显式画 plan→execute→critique 图 / Supervisor / Swarm / GraphFlow / deepagents 的 todo 工具;checkpoint 续跑跨进程都行 §3.2-3.5 / §2.3
07 Hooks 强制层 失败转成永久规则 ⚠ 有 hook 原语但不强制:pre_model_hook / post_model_hook / interrupt_before / interrupt_after;LangChain 1.0 create_agent 多了 middleware 系统。没有"agent 必须先 plan 再 execute"这种强制语义,要写就自己画在图边里 §3.1 / §4.4
08 规划 + 工具选择 别什么都试 ⚠ 把它推给图设计:你画一个 plan 节点 → 选 tool 节点 → execute 节点的图就有"规划",但没有"框架默认 agent 必须先 plan"。deepagents 的 write_todos 是少数自带的"显式规划工具" §3.2 / §3.5

判读规则:✅ = 框架原生提供且大多场景够用;⚠ = 框架给了原语但要开发者组装;❌ = 框架不管,需要自己写或外接。

#0.5.3 当前架构的短板(按 Harness 视角看出来的)

LangGraph 的根本设计哲学是 "low-level orchestration framework"——它故意不替你做指令层和能力层的决策。 所以"短板"的定义在这里要倒过来:不是它没做什么,而是它把哪些决策留给了开发者,这些点都是工程化时的真实成本:

  • 指令层完全空白:没有 "system prompt 怎么写 / 怎么注入 skill / 怎么提示 agent 切换 phase" 的内置约定。 Claude Code 那种"agent.md + 子目录 skill"的体系在 LangGraph 里只能靠 deepagents 模板(或开发者自创)补齐。
  • 能力层只有调度器没有工具仓:ToolNode 只负责把 tool_calls 派出去,工具实现、参数 schema、 错误处理、并发上限、RAG 检索、shell / browser / file 全是开发者的事。这跟"内置 50 个 connector"的 LangChain 风格是反的。
  • Context rot 防护偏弱:长会话裁剪只给 pre_model_hook 钩子和 trim_messages 工具, 没有"自动 summarize 历史 / 维护一份滚动摘要 / 把 system prompt 和长 context 分离做 prompt cache"这类成熟方案。
  • 多 agent 模板要选 / 要拼:Supervisor / Swarm / Hierarchical / Custom 四种范式各有取舍, 选错了上线后改图代价大(state schema 变更要管 checkpoint 兼容,§8.2)。
  • 可观测的"业务语义"层薄:astream / astream_events 给的是图层事件(哪个节点跑了什么 token), 但"这条用户消息花了多少钱 / 哪个 tool 失败率最高 / 哪个 agent 经常被路由到"这类聚合靠 LangSmith(付费外挂)。

这些"短板"换个角度看就是 LangGraph 的设计取舍——做底层积木保留最大灵活度。 真要做端到端 agent 产品,要么自己在上面叠一层 harness(deepagents 是 LangChain 团队自己叠的样板), 要么直接用 LangChain 1.0 create_agent + middleware 系统(同一套底子的"半成品"封装)。


#一、Agent 是什么 —— StateGraph + Node 函数 + state schema

在 LangGraph 里,"一个 agent" = 一份 state schema(数据契约) + 一组 node 函数 (能力实现) + 一张连接图(控制流)。三件套缺一不可。

#1.1 三件套:State / Node / Edge

组件 类型 职责
State schema TypedDictpydantic.BaseModel 描述图运行期间会出现的所有字段(agent 的"血液")
Node def(state) -> dictasync def(state) -> dict 一次具体的状态变更("喝点药")
Edge 静态 add_edge 或动态 add_conditional_edges 决定 node 之间的跳转("血液怎么流")

最小工作示例:

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # ★ Reducer:append 而非覆盖
    counter: int                              # 默认覆盖语义

def call_model(state: AgentState) -> dict:
    new_msg = llm.invoke(state["messages"])
    return {"messages": [new_msg], "counter": state["counter"] + 1}

graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)
app = graph.compile()

三件事值得展开:

  1. State 字段必须有 reducerAnnotated[list, add_messages] 这一行是说: messages 通道用 add_messages 这个 reducer 来合并新值(追加而非覆盖、按 id 去重、自动给新消息打 uuid)。不写 reducer 就默认覆盖,每次 node 返回的字段 会替换前值。
  2. Node 只返回"差量"。返回 {"messages": [new_msg]} 不是替换整个 messages, 而是交给 reducer 合并。这一点跟传统 chain 的 "input → output" 范式区别巨大。
  3. START / END 是常量节点START 是图入口(__start__),END 终止当前分支(__end__)。不连 END 的分支会一直循环到达到 recursion_limit 抛错。

💡 只需要一个 messages 字段时,可以直接用快捷类 from langgraph.graph import MessagesState (等价于 class MessagesState(TypedDict): messages: Annotated[list[AnyMessage], add_messages]), 自己再用 class S(MessagesState): plan: str 扩展字段。这是文档样例里最常见的写法。

#1.2 Reducer:state 通道的"合并策略"

LangGraph 的 state 不是普通 dict,而是 channels(通道)——每个字段是一个 受 reducer 控制的通道,每一步 node 返回的字段会经过对应 reducer 合并到主 state。

Reducer 来源 语义
默认(无 reducer) / 覆盖:新值直接替换旧值
add_messages langgraph.graph.message 消息列表追加 + id 去重 + 新消息自动补 uuid
operator.add stdlib 列表 / 数字相加(map-reduce 聚合常用)
Annotated[T, my_fn] 自定义 任意 (old, new) -> merged 函数

为什么要 reducer? 当多个 node 并行写入同一字段(fan-out / 子图 / 并发分支), 没有 reducer 你拿不到"合并语义",只会变成竞态。Reducer 是 LangGraph 实现并发安全的 关键,背后跟 Pregel BSP 模型一脉相承(每个 superstep 收集所有更新再批量合并)。

#1.3 Conditional Edge:图层面的控制流

ReAct 的核心循环不是写 while True,而是用条件边:

def should_continue(state) -> str:
    last = state["messages"][-1]
    return "tools" if last.tool_calls else END

graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode(tools))
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")          # 工具结果回到 agent 节点

add_conditional_edges(source, path_fn, mapping) 三件套:

  • source:从哪个节点出发判断
  • path_fn:纯函数,返回字符串 / 字符串列表 / Send 对象
  • mapping:字符串 → 目标节点的字典

返回 Send("node", subset_state) 可以触发动态扇出——同一个节点用不同子状态并发 跑 N 次(map-reduce 的 map 阶段就是这样写的)。

#1.4 一图看清"agent 是怎么被定义的"

                 ┌─────────────────────────┐
                 │   AgentState (TypedDict) │
                 │   messages: add_messages │
                 │   plan: str (覆盖)        │
                 │   files: dict (自定义合并)│
                 └────────────┬────────────┘
                              │ 字段契约
            ┌─────────────────▼─────────────────┐
            │            StateGraph              │
            │   ┌─────────┐    ┌─────────┐       │
   START ──▶│   │  agent  │───▶│  tools  │──┐    │
            │   └────┬────┘    └────┬────┘  │    │
            │        │  conditional │       │    │
            │        ▼              └───────┘    │
            │      END  ◀────────────────────────│
            └────────────────────────────────────┘
                              │ .compile(checkpointer=, store=)
                              ▼
                     CompiledStateGraph
                     (可 invoke / stream / ainvoke / astream)

这就是 LangGraph 里"agent 是谁"的全部定义——没有 class,没有 framework,只有这三层。 后面所有花哨的范式(react / supervisor / swarm / deep agents)都是在这套基础上往上盖。

#1.5 跟"老 chain"的本质差异

维度 LangChain 老 chain(AgentExecutor LangGraph
State 隐式(intermediate_steps 黑盒) 显式 TypedDict / Pydantic
控制流 while not finished: ... 内置死循环 图节点 + 条件边,每跳都能 hook
并发 几乎没有 Send / Subgraph / fan-out 原生支持
持久化 没有 Checkpointer 内置,state 切片可序列化
中断 没有 interrupt() 抛异常 + 可恢复
Streaming 只能流 token 五种 stream mode + 事件流

这表是回答"为什么 LangChain 自己都把 AgentExecutor deprecate 掉、推 create_agent (底层 LangGraph)"的全部答案。


#二、Agent 怎么活起来 —— invoke / astream / Cloud 三条触发链路

同一个 compiled graph,本地一行 app.invoke() 跑批,本地 astream 跑流式 UI, 上 Platform 跑长任务——三种链路同一份图,区别只在执行器。

#2.1 同步 invoke:最朴素的链路

config = {"configurable": {"thread_id": "user-42"}}
out = app.invoke({"messages": [{"role": "user", "content": "你好"}]}, config)
print(out["messages"][-1].content)

发生了什么:

  1. invoke 把入参跟 checkpointer 里 thread_id="user-42" 的最近 state 合并(reducer 起作用)
  2. Pregel 引擎进入超步循环:每一步找出"该跑的 node 集合"(输入通道有更新且未消费)
  3. 并发执行该集合,收集每个 node 的返回,按 reducer 合并到 state
  4. 把新 state 序列化进 checkpointer(一个 checkpoint = 一个 superstep 末尾的快照)
  5. 直到没有 node 触发或到达 END,返回最终 state

关键点thread_id 是用户级 / 会话级的标识,决定"这次 invoke 接到哪条历史上"。 没有 thread_id 就用 MemorySaver 跑一次性图(每次 invoke 都是干净 state)。

#2.2 异步 astream:UI / agent 编排标配

async for chunk in app.astream(
    {"messages": [{"role": "user", "content": "..."}]},
    config={"configurable": {"thread_id": "user-42"}},
    stream_mode="updates",      # 或 values / messages / debug / custom / 多个
):
    print(chunk)

astream 是 agent UI 真正的脊柱(流式输出 §6 详解)。同一个图,stream_mode 选不同的 就推不同粒度的事件——不需要改图,不需要改 node

#2.3 LangGraph Platform / Cloud:长任务运行时

把同一个图扔到 langgraph-cli 打包,部署到 LangGraph Platform(已并入 LangSmith Deployment)后,得到的是:

运行时能力 本质
Threads API thread_id 一等公民,HTTP 接口 POST /threads/{id}/runs 启动一次 invoke
后台 run Cron / webhook 触发的 run 在 task queue 里跑,不占用调用方连接
Postgres checkpointer 默认替你接,state 持久到平台托管 PG
HITL UI 内置面板看 interrupt、改 state、resume
Subgraph trace 每个 superstep 的 state 切片可在 Studio 时间线上回放

对工程的意义:长任务不再是"自己写 worker + 自己存中间结果"——平台把"图 + 状态 + checkpoint + 中断 + 重放"打包成一组 HTTP API。如果你做 deep research / 异步 agentic workflow,这是 LangGraph 跟纯库的最大区别。

#2.4 三条链路对比

维度 invoke (本地同步) astream (本地异步) Platform Run
调用方 Python 进程 Python / FastAPI 服务 HTTP 客户端
持久化 看 checkpointer 看 checkpointer 默认 Postgres
失败恢复 进程内重试 进程内重试 平台 worker 自动续跑
中断处理 手写 try/except 同上 平台 UI 改 state 后 Command(resume=...)
适用 离线脚本 / 测试 API 后端 / Discord bot 深度研究 / 多日 agent

#三、Agent 怎么思考 —— ReAct prebuilt + 自定义图 + Supervisor / Swarm

LangGraph 默认 agent loop 不是"prompt 里写 ReAct 模板",而是 graph 上画一条 agent ↔ tools 来回的边——"思考"被显式建模成两个 node 之间的循环。

#3.1 create_react_agent:90% 单 agent 场景的起点

源码:libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py

调用一行:

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(
    model="openai:gpt-4o",                # 或者已 bind_tools 的 ChatModel
    tools=[search, calculator],
    prompt="You are helpful.",
    checkpointer=PostgresSaver.from_conn_string(...),
    store=PostgresStore.from_conn_string(...),
    pre_model_hook=trim_history,          # 进 LLM 前裁剪 messages
    post_model_hook=guardrail_check,      # 出 LLM 后审查
    response_format=MyOutputSchema,       # 强制结构化输出
    interrupt_before=["tools"],           # 工具调用前暂停等人审
    version="v2",                         # v2 用 Send 并发 tool calls
)

它内部干了什么(看源码):

  1. 定义 AgentStatemessages: Annotated[Sequence[BaseMessage], add_messages]
    • remaining_steps: RemainingSteps
  2. prompt 转成 Runnable(str / SystemMessage / callable / Runnable 都接受)
  3. 校验 model.bind_tools() 是否已经绑了所有 tools,没绑就帮你 bind
  4. 加两个节点:agent(call model)+ toolsToolNode(tools)
  5. 加条件边:agent → 有 tool_call 走 tools / 否则 END / 步数耗尽抛错
  6. tools 节点回到 agent,循环
  7. 返回 CompiledStateGraph

v1 vs v2 的区别:v1 把 LLM 一次返回的多个 tool_call 当作 ToolNode 的批量输入, v2 用 Send 把每个 tool_call 拆成一个并发 sub-invocation(更利于并发执行长跑工具)。

⚠️ 注意:create_react_agent 在 LangGraph v1.0 已被官方源码 @deprecated 标注 (libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py),明确推荐迁到 from langchain.agents import create_agent(同一个团队,同一套底层,多了 middleware 系统)。但 create_react_agent 还能跑很久,且生态里大量代码用它,所以本文照旧讲。 迁移指南见 https://docs.langchain.com/oss/python/migrate/langgraph-v1

#3.2 自定义 graph:当 ReAct 不够用

ReAct 的局限:单一 agent / 固定循环 / 没法显式表达 plan-act-reflect。需要的时候 直接拿 StateGraph 拼,比如一个 plan → execute → critique → revise 的图:

START ──▶ plan ──▶ execute ──▶ critique ──┐
                      ▲                   │
                      └──── revise ◀──────┘ (条件边: 不通过)
                                          │
                                          ▼
                                         END  (条件边: 通过)

每个节点都是普通函数,state 里塞 plan: str / result: str / critique: str, critique 用条件边决定走 END 还是 revise。这种"明牌"的图比 prompt 里塞"think step by step then critique"稳定得多——控制流移到了图层而不是 LLM 内部

#3.3 Supervisor 模式:中央调度 + 子 agent

包:langgraph-supervisor,仓库 langchain-ai/langgraph-supervisor-py

from langgraph_supervisor import create_supervisor

research_agent = create_react_agent(model, [web_search],   name="research")
math_agent     = create_react_agent(model, [py_executor],  name="math")

workflow = create_supervisor(
    [research_agent, math_agent],
    model=model,
    prompt="You manage a research expert and a math expert. Route accordingly.",
)
app = workflow.compile(checkpointer=...)

它干的事:

  1. 把每个子 agent 当作 graph 的一个 node(子 agent 本身也是一张图,subgraph 嵌套)
  2. supervisor 节点是一个 LLM,给它一份"delegate_to_research / delegate_to_math / finish" 的 handoff tool 列表
  3. supervisor 根据当前 messages 选一个 handoff tool → 该子 agent 节点接管
  4. 子 agent 跑完回到 supervisor,supervisor 再判断要不要继续派活
  5. supervisor 决定 finish 时图走 END
                    ┌─────────────────┐
   user ──▶ START──▶│   supervisor    │──▶ END
                    │  (LLM + handoff │      ▲
                    │     tools)      │      │
                    └────┬─────┬──────┘      │
                         │     │             │
              delegate_  │     │  delegate_  │
              to_research│     │  to_math    │
                         ▼     ▼             │
                  ┌─────────┐  ┌─────────┐   │
                  │research │  │  math   │───┘
                  │ (sub-   │  │ (sub-   │
                  │  graph) │  │  graph) │
                  └─────────┘  └─────────┘

特点:控制流可观(每次都回 supervisor)/ trace 干净 / 适合需要全局视角的场景。 代价:每次切 agent 多一次 LLM 调用,延迟更高。

#3.4 Swarm 模式:去中心化 / 直接交接

包:langgraph-swarm,仓库 langchain-ai/langgraph-swarm-py

from langgraph_swarm import create_swarm, create_handoff_tool

alice = create_react_agent(model,
    tools=[create_handoff_tool(agent_name="Bob"), ...], name="Alice")
bob   = create_react_agent(model,
    tools=[create_handoff_tool(agent_name="Alice"), ...], name="Bob")

workflow = create_swarm([alice, bob], default_active_agent="Alice")
app = workflow.compile(checkpointer=...)

它干的事:

  1. state 里多一个 active_agent: str 字段
  2. 每个 handoff tool 返回 Command(goto="Bob", graph=Command.PARENT, update={"active_agent": "Bob"})
  3. ToolNode 检测到 Command 直接跳节点(不再回到调用 agent)
  4. checkpointer 保存 active_agent,下一轮对话从这个 agent 继续
   user ──▶ ┌─────────┐  handoff_to_Bob   ┌─────────┐  handoff_to_Alice
            │  Alice  │ ─────────────────▶│   Bob   │ ──────────────▶ ...
            └─────────┘                   └─────────┘
                ▲ (active_agent 持久化)         ▲
                └──────────── checkpointer ────┘

特点:少一跳 LLM / 延迟低 / 适合 agent 互相熟悉职责的场景。 代价:没有全局视角,容易陷入 A↔B 反复横跳;调试时 trace 跳来跳去看着累。

#3.5 Deep Agents:todo + 虚拟文件系统 + subagent

包:deepagents,仓库 langchain-ai/deepagents。是 LangChain 团队 2026 年推的 "长任务 agent harness"。早期版本基于 create_react_agent当前版本(2026 年中后) 已迁到 LangChain create_agent + 一组 middleware(filesystem / subagents / summarization / permissions / skills 等),但对开发者的入口 create_deep_agent API 保持一致:

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openai:gpt-5.5",                  # 或任意 LangChain ChatModel
    tools=[your_research_tool],
    system_prompt="You are a research analyst.",   # 旧版叫 instructions=,新版统一为 system_prompt=
    subagents=[
        {"name": "summarizer", "description": "...",
         "prompt": "...", "tools": [...]},
    ],
)
result = agent.invoke({"messages": "..."})

它在 ReAct 之上多挂三类工具 + 若干 middleware:

工具组 工具名 作用
Planning write_todos 把复杂任务拆成 checklist 写进 state
Filesystem ls / read_file / write_file / edit_file / glob / grep 在 state(默认)/ 本地磁盘 / Store / sandbox 里读写文件,避开 context 爆炸
Subagent task 起一个上下文隔离的子 agent 干小活,只回总结
Shell(可选) execute 在选定 sandbox 里跑命令(local / Modal / Daytona)

文件系统是 可插拔后端:默认存 state(in-memory),也能切换到 LangGraph Store (跨 thread 持久)/ 本地磁盘 / Modal / Daytona / Deno 沙箱。这把"agent 搞出来一堆 中间产物" 的脏活包了。

middleware 层(新版核心):summarization(自动滚动摘要长 messages)/ permissions (HITL 工具批准)/ skills(按需加载技能描述)—— 这些是 LangGraph 原生不给但 deep agents 作为 harness 模板补齐的能力,正好对应 §0.5.3 列的"短板"。

这套范式致敬了 Cognition Devin 的 task / file 设计,但完全开源,跑在 LangGraph 上。

#3.6 四种"思考范式"对比

范式 控制流 代码量 适合 trace 复杂度
create_react_agent 单 agent 内部 LLM↔tools 循环 1 行起步 90% 工具调用类任务
自定义 StateGraph 显式节点 / 条件边 几十行 plan-execute-critique 这类多阶段
Supervisor 中央 LLM 调度 sub-agent 一段配置 多专家 / 全局视角
Swarm 去中心化交接 一段配置 agent 职责互相清楚 / 低延迟 中(trace 跳)
Deep Agents ReAct + 计划 + 文件系统 + subagent 一段配置 长任务 / 深度研究 / coding

AutoGen 的 actor model 是另一种思路(每个 agent 一个 actor,消息驱动,没有显式 图)。LangGraph 选了图编排,让控制流可观;AutoGen 选了消息总线,让协议 灵活。要事后追责 / replay 选 LangGraph,要快糙猛扔 agent 进消息队列选 AutoGen。


#四、Agent 怎么行动 —— ToolNode + Handoff + Interrupt

Tool 在 LangGraph 里不是"附加功能",而是 agent 跟世界唯一的接口——所有的副作用、 所有的 handoff、所有的 HITL 都从 tool 出口走。

#4.1 ToolNode:自动执行 tool_call 的标准节点

源码:libs/prebuilt/langgraph/prebuilt/tool_node.py

from langgraph.prebuilt import ToolNode
tools_node = ToolNode([search, calculator, save_file])

它的工作很机械:

  1. 从 state 拿最后一条 AIMessage
  2. 遍历 tool_calls 字段
  3. 每个 tool_call 找对应 tool,并发执行(v2 用 Send)
  4. 把每个返回包成 ToolMessage,挂回 messages(reducer 是 add_messages
  5. 出现异常按 handle_tool_errors 配置:要么抛、要么塞错误消息让 agent 自我修复

支持的 tool 类型:

  • LangChain @tool 函数 / BaseTool 子类
  • 普通 Python 函数(StructuredTool.from_function 自动包)
  • 返回 Command(...) 的 tool(用于 handoff,见 §4.3)
  • 注入 state 的 tool(参数注解 Annotated[..., InjectedState]

#4.2 Tool 怎么"看到" agent 的 state

普通 tool 的签名是 def search(query: str) -> str,但 agent 跑的时候有时需要 "读 state 里的 user_id" 或者 "把结果写回 state"。LangGraph 用 InjectedStateInjectedToolCallId 注解告诉 ToolNode "这个参数不是给 LLM 看的,是注入的":

from langgraph.prebuilt import InjectedState
from langchain_core.tools import tool

@tool
def search_in_user_scope(
    query: str,
    state: Annotated[dict, InjectedState],     # 不暴露给 LLM
) -> str:
    user_id = state["user_id"]
    return search(query, scope=user_id)

LLM 永远只看到 query 参数;ToolNode 从当前 state 注入 state 参数。这是安全 边界的关键——敏感信息不通过 prompt 传给 LLM 而通过 state 注入给 tool。

#4.3 Handoff Tool:用 Command 跳节点

from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.types import Command

def create_handoff_tool(agent_name: str):
    @tool(name=f"transfer_to_{agent_name}")
    def handoff(
        state: Annotated[dict, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        tool_msg = ToolMessage(
            content=f"Transferred to {agent_name}",
            tool_call_id=tool_call_id,
        )
        return Command(
            goto=agent_name,                  # 跳到目标节点
            graph=Command.PARENT,             # 跳出当前 subgraph 到父图
            update={"messages": [tool_msg], "active_agent": agent_name},
        )
    return handoff

Command 是个统一原语,三个字段:

  • goto:下一个节点名(也可以是 END,或多个节点列表用于扇出)
  • update:给 state 通道喂数据(走 reducer)
  • resume:给 interrupt() 回填值(见 §4.4)

ToolNode 看到 tool 返回 Command 不会原样塞进 messages,而是把 update / goto 直接交给 Pregel 引擎执行。这是 swarm / supervisor / 任何跨 agent 跳转的底层机制

#4.4 Interrupt:图层面的 HITL

源码:langgraph.types.interrupt,最常用的人审范式。

from langgraph.types import interrupt, Command

def critical_op(state):
    # 先存草稿
    draft = compute_draft(state)
    # 把草稿抛给外面,等用户审
    decision = interrupt({"draft": draft, "type": "approve_required"})
    if decision == "approve":
        do_real_thing(draft)
    else:
        return {"messages": [...]}

调用方拿到 chunk 里的 __interrupt__ 事件,给用户看 draft,用户点"approve"后:

app.invoke(Command(resume="approve"), config)

发生了什么:

  1. interrupt()GraphInterrupt 异常
  2. Pregel 引擎捕获,把异常负载({"draft": ..., "type": ...})写进当前 checkpoint 的 interrupts 列表
  3. invoke / stream 提前返回,用户拿到 __interrupt__ chunk
  4. 外部决策回来后,调用方 invoke(Command(resume=...))
  5. 引擎从最近 checkpoint 恢复,把 resume 值塞回 interrupt() 调用点
  6. 节点继续执行,跟没中断过一样

关键认知:interrupt 不是 try/except,是 checkpoint 机制的副产物——只要图的 state 都在 checkpointer 里,重新 invoke 就能从中断点继续。没有 checkpointer 就用不了 interrupt(会抛 RuntimeError)。

create_react_agent 还提供 interrupt_before=["tools"] / interrupt_after=[...] 两个参数,这是图层面的 breakpoint:在某节点之前 / 之后无条件停下,等外部 resume。 做开发调试 / 高敏感工具的强制审批,比业务里硬埋 if 漂亮。

#4.5 Tool 调用全链路图

agent node
   │
   │ AIMessage(tool_calls=[{name, args, id}])
   ▼
ToolNode
   ├─ 普通 tool ─────▶ ToolMessage ──▶ messages(add_messages 合并)
   ├─ Command-tool ──▶ Command(goto, update) ──▶ Pregel 跳节点
   └─ interrupt() ───▶ GraphInterrupt ──▶ 写 checkpoint ──▶ 抛给调用方

三种出口归到一起:LangGraph 里所有跨节点的副作用,都从 ToolNode 出口走。 你想做 "agent 调一个工具直接结束当前 agent / 跨 agent / 等待人审" —— 都是 tool 返回 不同对象的事。


#五、Agent 怎么记忆 —— state + Checkpointer + Store

LangGraph 把记忆切成两层:state 里的 messages 是 working memory(短期), checkpointer 把每个 superstep 的 state 持久化(断点续跑),store 把跨 thread 的 知识攒着(长期)—— 三层各司其职。

#5.1 短期记忆:state.messages + thread_id

最常见的"会话记忆"就是 state 里的 messages 字段,用 add_messages reducer 追加,靠 checkpointer 按 thread_id 持久化:

config = {"configurable": {"thread_id": "user-42-conv-7"}}
app.invoke({"messages": [...]}, config)   # 第一轮
app.invoke({"messages": [...]}, config)   # 第二轮自动接上

两条规则

  1. 没有 thread_id 的 invoke 没有记忆(每次新 state)
  2. 同一个 thread_id 的多次 invoke 会自动接 messages(reducer 合并)

老消息怎么裁?用 pre_model_hookcreate_react_agent 参数)或者自己加一个 "trim node",langchain_core.messages.utils.trim_messages 做按 token 数裁剪是标配。

#5.2 Checkpointer:state 的持久层

接口:langgraph.checkpoint.base.BaseCheckpointSaver

实现 适合
InMemorySaver(旧名 MemorySaver,仍保留为别名) langgraph.checkpoint.memory 单进程跑 demo / 单测
SqliteSaver / AsyncSqliteSaver langgraph-checkpoint-sqlite 单机本地 agent / Discord bot
PostgresSaver / AsyncPostgresSaver langgraph-checkpoint-postgres 生产 agent 服务 / Platform 默认
RedisSaver / AsyncRedisSaver langgraph-checkpoint-redisredis-developer/langgraph-redis,Redis 官方维护) 已有 Redis / RedisStack 想复用,含 Store 实现
DynamoDB / MongoDB / 自研 社区或自实现 自己接 BaseCheckpointSaver

每个 superstep 末尾写入一个 checkpoint,包含:

checkpoint = {
    thread_id, checkpoint_id, parent_checkpoint_id,
    channel_values: {messages: [...], plan: "..."},
    channel_versions: {messages: 7, plan: 3},   # 通道版本号(决定下个 superstep 哪些节点要跑)
    pending_sends: [Send(...)],                 # 未消费的扇出
    pending_writes: [...],                      # 未提交的写
    interrupts: [...],
}

版本号机制是 Pregel 模型的核心:节点是否要在下一 superstep 跑,取决于它的输入 通道版本号是否比上次它消费的版本号高。这套机制让"哪些节点该跑"完全可推导,从而支持 任意复杂的并发图。

#5.3 Time Travel:从历史 checkpoint 重放

# 拿到历史 checkpoint 列表
history = list(app.get_state_history(config))
# 选一个老 checkpoint
old_cp = history[3]
# 从那个点重新跑(甚至改 state 再跑)
new_config = {"configurable": {"thread_id": "...", "checkpoint_id": old_cp.config["configurable"]["checkpoint_id"]}}
app.invoke(None, new_config)             # None 表示不喂新输入,从老 checkpoint 接着跑
app.update_state(new_config, {"plan": "...new..."})  # 改 state 再跑

这是调试 agent 的杀手锏——任何一个超步都能重放,能改 state 再走。 比写 print + 复现快十倍。

#5.4 Store:跨 thread 的长期记忆

接口:langgraph.store.base.BaseStore

from langgraph.store.memory import InMemoryStore
# 或 langgraph.store.postgres.PostgresStore

store = InMemoryStore(index={"embed": embeddings, "dims": 1536, "fields": ["content"]})
# 写
store.put(("user", "42", "memories"), key="loves-pizza", value={"content": "用户爱吃披萨"})
# 取
mems = store.search(("user", "42", "memories"), query="食物偏好")

特点:

  • namespace 是 tuple("user", user_id, "memories")),层级语义随便切
  • 可选向量索引——配 embeddings 后 search 走语义检索
  • Postgres 实现走 pgvector,是生产推荐
  • 跨 thread / 跨 agent 共享——这是它跟 checkpointer 的本质区别

工具里直接用:

from langgraph.prebuilt import InjectedStore
from langgraph.store.base import BaseStore
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool

@tool
def remember(
    fact: str,
    config: RunnableConfig,
    store: Annotated[BaseStore, InjectedStore()],
):
    user_id = config["configurable"]["user_id"]
    store.put(("user", user_id, "facts"), key=fact[:32], value={"content": fact})

InjectedStoreInjectedState 一个套路——LLM 看不到,由 ToolNode 注入。 create_react_agent(store=store, ...) 全程把这个 store 注入到所有支持的 tool。

#5.5 三层记忆对照

存什么 生命周期 持久层
state.messages 当前会话历史 一个 thread checkpointer
state.其他字段 plan / files / counters / scratchpad 一个 thread checkpointer
store 用户偏好 / 学到的事实 / 跨会话 RAG 内容 用户级 / 全局 Postgres / 内存 / 自研

#六、Agent 怎么开口 —— 五种流模式

LangGraph 不只流 token,它流"图发生了什么"——五种 stream_mode 让 UI 能选自己 想要的粒度,从粗(每步状态)到细(每个 token)都行。

#6.1 五种 stream_mode

async for chunk in app.astream(input, config, stream_mode="updates"):
    ...
模式 chunk 长啥样 用途
values 当前完整 state 想每一步看到全量 state(小图 / 调试)
updates {node_name: 该 node 这次返回的差量} UI 上看"现在哪个 node 在跑、写了啥"
messages (message_chunk, metadata) LLM token 流 + 哪个 node / run 产生的
debug 包含 task / step / state 的诊断包 trace UI / Studio 用
custom 节点内 get_stream_writer().write(...) 推的任意自定义 chunk 业务自己造事件("开始写代码")

可以多模式同时开,stream_mode=["updates", "messages"] 拿到 (mode, chunk) 元组。

#6.2 astream_events:跟 LangChain 通用的事件流

async for event in app.astream_events(input, config, version="v2"):
    if event["event"] == "on_chat_model_stream":
        token = event["data"]["chunk"].content

跟 LangChain Runnable 一脉相承,事件类型包括:

  • on_chain_start / on_chain_end(每个 node 进出)
  • on_chat_model_start / on_chat_model_stream / on_chat_model_end
  • on_tool_start / on_tool_end
  • on_retriever_* / on_prompt_*

astream(stream_mode="messages") 区别:events 是 LangChain 通用事件协议, mode="messages" 是 LangGraph 专用、信息更紧凑。一般 UI 用 messages 流 token, 日志 / 监控用 astream_events 拿全事件。

#6.3 节点内主动推 chunk

from langgraph.config import get_stream_writer

def my_node(state):
    writer = get_stream_writer()
    writer({"phase": "fetching docs"})
    docs = fetch()
    writer({"phase": "analyzing", "n_docs": len(docs)})
    return {...}

调用方加 stream_mode="custom"(或多模式包含 custom)就能收到这些自定义 chunk。 做"agent 现在干啥"提示条用这个最干净,不用塞进 messages 污染上下文。

#6.4 跟前端的连接

最常见的栈:FastAPI WebSocket / SSE → astream → 前端 React。LangGraph 团队还 开源了 assistant-ui / agent-chat-ui 两个 React 模板,自己接 messages mode 就能跑出 ChatGPT 似的 UI(含 tool call 折叠 + interrupt 卡片)。


#七、Agent 的外脑 —— LangChain ChatModel 抽象

LangGraph 不实现任何 LLM 客户端——它把"模型"这个角色完全外包给 LangChain 的 ChatModel 抽象层,几乎所有 provider(OpenAI / Anthropic / Gemini / vLLM / Ollama / Bedrock / Groq / DeepSeek / 通义千问 / xAI)都已经有 langchain_* 包。

#7.1 ChatModel 接口的两条契约

LangGraph 对模型的唯一要求:

  1. 实现 BaseChatModelinvoke / ainvoke / astream
  2. 实现 bind_tools(tools: list) 返回一个绑定了 tool schema 的模型实例

任何 LangChain ChatModel 都满足。两种常见用法:

# A. 直接传字符串(LangChain 1.0+ 的 init_chat_model 协议)
agent = create_react_agent(model="openai:gpt-4o", tools=tools)
agent = create_react_agent(model="anthropic:claude-sonnet-4-5", tools=tools)

# B. 传 ChatModel 实例
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools, parallel_tool_calls=False)
agent = create_react_agent(model=model, tools=tools)

#7.2 模型路由 / 多模型 agent

create_react_agentmodel 参数也能传 callable,按 state 决定用哪个模型:

def pick_model(state, runtime):
    if state["task_type"] == "code":
        return claude_sonnet
    return gpt_4o_mini

agent = create_react_agent(model=pick_model, tools=tools)

这是"复杂任务用大模型 / 简单任务用便宜模型"的官方路径。LangSmith 的成本统计也会 按节点拆开,方便算账。

#7.3 跟 LLM Gateway 的关系

LangGraph 不做 gateway 层(限流 / 计费 / failover)。生产典型架构:

LangGraph agent (ChatModel) ──▶ LiteLLM proxy / 自建 gateway ──▶ OpenAI / Anthropic / 自部署 vLLM

ChatModel 的 base_url / api_key 直接打到 LiteLLM 这种 OpenAI 兼容代理上即可, LangGraph 这层完全无感知。

#7.4 Function Calling 协议

LangGraph 默认走 ChatModel 自带的 tool calling 协议——OpenAI tools / Anthropic tool_use / Gemini function_call —— ChatModel 内部统一成 AIMessage.tool_calls 列表,再交给 ToolNode 执行。LangGraph 不发明新协议,这点跟 Hermes 那种"在 prompt 里手撸 <tool_call> 标签"是两条路。


#八、关键设计权衡(架构师视角)

五条选型 / 落地中绕不过去的取舍,每条按 问题 / 回答 / 启发 三段式给。

#8.1 图编排 vs ReAct prompt 内嵌

问题:很多框架(包括早期 LangChain AgentExecutor)把 ReAct 循环写在框架的 while 里,prompt 里塞"think / act / observe"模板让 LLM 模仿。LangGraph 为啥要 把循环显式画成图?

回答:把控制流移到图上有三个直接收益:

  1. 可观测——每一跳都是 superstep,Studio / LangSmith 直接出节点级 timeline
  2. 可中断——任意节点前后插 interrupt,业务不需要自己埋 hook
  3. 可重放——checkpoint 切片完整,时光回溯不需要重跑 LLM

代价是认知门槛高了一截:你得理解"state 是 channel、reducer 是合并策略、Pregel 是 执行模型"。但一旦理解,任何"prompt 里写规则"的方案都会显得脆—— prompt 改一行 agent 行为就漂,图改一条边语义清晰、diff 干净。

启发:选 LangGraph 时,把"能不能画成图"作为第一考量——能画清楚就用,画不清楚 说明任务定义本身没收敛,应该先做产品设计。

#8.2 显式 state 的代价与回报

问题:相比"全靠 messages 列表"的 chain 范式,LangGraph 强制 TypedDict / Pydantic 写 state schema,明显多一笔活。值不值?

回答

收益 代价
Reducer 让并发 / 扇出安全 多写一个 schema
工具能 InjectedState 拿到非 LLM 字段(user_id / org_id) 每个字段要决定 reducer
Time travel 可以改任意字段重放 跨版本的 schema 演化要自己处理
LangSmith 能看到每个超步的 state diff 模板代码增加

经验:state 字段不要超过 10 个。多了说明你在把"业务状态"塞进 agent state,应 该外置到 DB(用 tool 读写)。把 state 当作"agent 这一轮思考必须的草稿纸",不是 "系统的真相之源"。

启发:state schema 是 agent 的 API,跟 DB schema 一个慎重程度——一旦上线,改 字段就要考虑历史 checkpoint 的兼容(因为 checkpointer 里存的就是这个 schema 序列化 的版本)。

#8.3 Interrupt 设计:图层面 HITL vs 业务层 HITL

问题:人审有两种实现:(A) 业务侧给 LLM 输出"塞个 button,前端等点击"; (B) LangGraph interrupt() 抛异常 + checkpoint 续跑。用哪个?

回答:A 看着简单,但故障域大——前端崩了 / WebSocket 断了 / 后端重启了, agent 状态全丢。B 是图层面的"事务点"——只要 checkpointer 在,永远能从中断点恢复, 甚至能跨进程(A 进程开 interrupt,B 进程拿 thread_id 来 resume)。

代价:interrupt 强依赖 checkpointer,跑 InMemorySaver 的图遇到 interrupt 直接抛错; 本地开发要养成"配 SqliteSaver"的习惯。

启发有一个以上"等人审"步骤就上 checkpointer + interrupt,别在业务层凑 合。代价是早期一点点配置工作,回报是上线后所有 HITL 场景的可靠性。

#8.4 Multi-agent 范式选型:Supervisor / Swarm / Network / Custom

问题:四种多 agent 模式都拼得起来,怎么选?

回答:按"控制权 / 可观测 / 延迟"三角选:

范式 控制权 可观测 延迟 适合
Supervisor 中央 ★★★ ★★ (多一跳) 多专家、流程合规、需要全局视角
Swarm 分散 ★★ ★★★ (直接跳) 职责清晰、A↔B 互相熟悉
Hierarchical 多层中央 ★★★ 大型企业系统 / 部门 / 组的多层结构
Network 任意 → 任意 ★★★ 强探索性 / 没有明显主次
Custom StateGraph 你说了算 看你怎么画 看你怎么画 上面四种都不贴脸的非典场景

误区警告:很多团队默认上 Supervisor 觉得"中央好管",但 supervisor 自己也是 LLM,且每一跳都要跑一次——10 个子 agent 的图里 supervisor 一轮要跑 10+ 次,token 账单陡增。Swarm 在职责清晰时往往更经济。

启发:先用 supervisor 跑通 PoC(trace 干净),上线前评估延迟 / 成本,对延迟 敏感的关键链路考虑收敛成 swarm 或自定义图

#8.5 LangGraph Platform vs 自托管 + 自己扛 checkpointer

问题:用不用 LangGraph Platform?

回答:决策点不是"贵不贵",而是"长任务多不多":

你做的事 推荐
短任务(≤30s)的对话 agent 自托管即可,FastAPI + PostgresSaver 三百行代码搞定
长任务(分钟到小时级)/ 异步 / cron Platform 真香——thread 管理 / 重试 / HITL UI 全包
私有数据 / 合规要求 Platform self-hosted 镜像也能上 K8s,不一定走 Cloud

代价:上 Platform 等于把 deployment 模式锁定(Threads API / langgraph-cli 打包), 后续想脱钩要重写 wrapper。

启发别一开始就上 Platform。先用 LangGraph 库 + Postgres checkpointer 跑 半年,攒到的 thread / interrupt / retry 痛点足够多再迁。LangGraph 库本身就足够 hold 住 80% 场景。

#8.6(额外)跟 AutoGen / CrewAI 的取舍

问题:multi-agent 这一坨框架(LangGraph / AutoGen / CrewAI / OpenAI Swarm / LlamaIndex Workflows)选哪个?

回答:一句话各代表一种世界观——

  • LangGraph:图 + 显式 state,控制流可观,社区最大,深度集成 LangChain 生态
  • AutoGen(Microsoft):actor model + 消息总线,更"分布式",配置上轻
  • CrewAI:role-playing 抽象,适合做"虚拟团队"演示和原型
  • OpenAI Swarm(已演化为 Agents SDK):极简 handoff,OpenAI 自家原型,生产证据少
  • LlamaIndex Workflows:事件驱动 + step 装饰器,跟 LlamaIndex 检索栈深度耦合

启发:要看 trace / 要 HITL / 要 checkpoint 续跑——选 LangGraph。要纯粹的 agent 互聊语义、不在乎 trace 完整度——AutoGen。要快速 demo + 角色扮演——CrewAI。 不要选 OpenAI Swarm 直接跑生产(它最早期就标自己是"教学示例")。


#九、附录:组件地图 / 关键概念 / 源码索引

#9.1 组件地图(核心包)

langgraph 仓库(langchain-ai/langgraph,PyPI 多包发布)
├── libs/langgraph/langgraph/        ★ 主包 `langgraph`:StateGraph / Pregel / channels / interrupt
│   ├── graph/                         图构造 (StateGraph / message / MessagesState)
│   ├── pregel/                        执行引擎(BSP / superstep / planner)
│   ├── channels/                      通道实现(LastValue / Topic / BinaryOperatorAggregate / EphemeralValue 等)
│   ├── types.py                       Command / Send / interrupt
│   ├── runtime.py                     Runtime / 配置
│   ├── func/                          @entrypoint / @task 函数式 API
│   ├── managed/                       managed values(RemainingSteps 等)
│   └── stream/                        astream / astream_events 实现
├── libs/prebuilt/langgraph/prebuilt/ ★ 预构造包 `langgraph-prebuilt`
│   ├── chat_agent_executor.py         create_react_agent(v1.0+ @deprecated)
│   ├── tool_node.py                   ToolNode
│   └── interrupt.py                   prebuilt interrupt 工具(HumanInterrupt 等)
├── libs/checkpoint/langgraph/         独立包 `langgraph-checkpoint`:BaseCheckpointSaver / InMemorySaver
│   ├── checkpoint/                    接口 + 内存实现
│   └── store/                         BaseStore 接口 + InMemoryStore
├── libs/checkpoint-postgres/          独立包 `langgraph-checkpoint-postgres`
├── libs/checkpoint-sqlite/            独立包 `langgraph-checkpoint-sqlite`
├── libs/cli/                          独立包 `langgraph-cli`(打包 / 本地 dev server)
├── libs/sdk-py/                       独立包 `langgraph-sdk`(Platform 客户端 SDK)
└── libs/sdk-js/                       JS SDK

兄弟项目(独立仓库):
langchain-ai/langgraph-supervisor-py        create_supervisor
langchain-ai/langgraph-swarm-py             create_swarm
langchain-ai/deepagents                     create_deep_agent
redis-developer/langgraph-redis             Redis checkpointer + Store
langchain-ai/react-agent                    starter template
langchain-ai/langgraphjs                    TypeScript port (libs/ 与上面同构)

#9.2 关键概念速查

术语 一句话
StateGraph LangGraph 的核心图构造器,声明 state schema + 加 node + 加 edge
State TypedDict / Pydantic 定义的字段集合,每个字段是一个"通道"
Channel state 字段的运行时对应物,由 reducer 控制写入合并
Reducer (old, new) -> merged 函数,决定字段的合并语义(默认覆盖)
Node def(state) -> dict 或 async 等价,返回的字段会经 reducer 合并
Edge 静态边 add_edge(a, b)
Conditional Edge add_conditional_edges(src, path_fn, mapping),path_fn 返回下一节点名
START / END 图的入口 / 终止常量节点
Send Send(node, sub_state),动态扇出到多个并发执行
Command tool / node 返回值,统一表达 goto + update + resume
interrupt(value) 节点中调用,抛异常并把 value 暴露给调用方,等 Command(resume=...) 续跑
Checkpointer 持久化 state 的存储(Memory / SQLite / Postgres),按 thread_id 切分
Store 跨 thread 的长期记忆 KV,namespace 是 tuple,可选向量索引
thread_id config.configurable.thread_id,决定 invoke 接到哪条历史
Pregel 借鉴 Google 同名论文的 BSP 执行模型,LangGraph 内部叫 Pregel
Superstep 一次同步执行屏障,每个超步一个 checkpoint
add_messages 消息列表 reducer:append + 按 id 去重 + 自动补 uuid
InjectedState tool 参数注解,告诉 ToolNode 把 state 注入进来(不暴露给 LLM)
ToolNode 自动执行 AIMessage.tool_calls 的标准节点
Subgraph 一个 compiled graph 当作 node 嵌进父图(supervisor / swarm 的实现基础)
astream 异步流式,stream_mode 可选 values / updates / messages / debug / custom
astream_events LangChain 通用事件流(v2 协议),跨 chain / graph 一致

#9.3 源码索引(仓库目录级,方便去追)

想看什么 去哪
StateGraph 怎么编译成 Pregel libs/langgraph/langgraph/graph/state.py
Pregel 主循环(superstep / planner) libs/langgraph/langgraph/pregel/__init__.py
Channel 实现(LastValue / Topic / BinaryAgg) libs/langgraph/langgraph/channels/
add_messages reducer / MessagesState libs/langgraph/langgraph/graph/message.py
Command / Send / interrupt libs/langgraph/langgraph/types.py
create_react_agent(已 @deprecated libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py
ToolNode libs/prebuilt/langgraph/prebuilt/tool_node.py
InMemorySaver / MemorySaver 别名 libs/checkpoint/langgraph/checkpoint/memory/__init__.py
PostgresSaver libs/checkpoint-postgres/langgraph/checkpoint/postgres/
SqliteSaver libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/
Store 接口 libs/checkpoint/langgraph/store/base.py
RedisSaver / RedisStore redis-developer/langgraph-redis:langgraph/checkpoint/redis/(独立仓库)
Supervisor 实现 langchain-ai/langgraph-supervisor-py:langgraph_supervisor/supervisor.py
Swarm 实现 langchain-ai/langgraph-swarm-py:langgraph_swarm/swarm.py
Deep Agents 实现 langchain-ai/deepagents:libs/deepagents/deepagents/
LangChain 1.0 create_agent langchain-ai/langchain:libs/langchain/langchain/agents/

#9.4 一句话总结

LangGraph = 把 agent 的 state、控制流、持久化、中断、流式输出,全部从 prompt 和 框架内部的 while 循环里掏出来,做成图上的一等公民——你画清楚 state schema、 node 函数、边路由,剩下的"长任务、HITL、time travel、multi-agent"全是这套底子的 自然推论。所有上层范式(ReAct / Supervisor / Swarm / Deep Agents)都是同一个 StateGraph + Command + Checkpointer 的不同拼法。


#文末小结(≤200 字)

消歧:本文讲的是 LangChain Inc 出的 langgraph Python / JS 库本身(图编排 原语 + prebuilt 范式),不是 LangGraph Studio(GUI)、不是 LangGraph Platform (托管运行时)、也不是 LangChain 老 chains 范式。

文件agents/docs/langgraph-agent-architecture.md, 约 1200 行,9 章按 Agent 生命脉络主线展开,含组件地图、术语速查、仓库目录级源码索引。

最不确定的两点:(1) create_react_agent 已在源码层 @deprecated(v1.0+), 官方推 from langchain.agents import create_agent;本文按"主线讲 create_react_agent + 顶部 / §3.1 显著提示迁移"处理,迁移时间表与新 API(middleware 系统)以官方迁移指南 https://docs.langchain.com/oss/python/migrate/langgraph-v1 为准;(2) LangGraph Platform 与 LangSmith Deployment 的产品合并节奏——文中描述按 2026 年中公开信息(已并入 LangSmith Deployment / "Assistants, threads, and runs" 执行模型),部署 API 命名 可能仍在演进(待核实细节)。

2026-06-08 复核补充:版本号已按 langgraph@1.2.4 / @langchain/langgraph@1.3.6 更新,详见 official-update-check-2026-06-08.md

返回 Agent 资料库