Agent 架构文档 · Markdown
LangGraph Agent 架构全景讲解
以 Agent 的生命脉络 为主线,讲清楚 LangGraph 这套"低层图编排框架"是怎么把 agent 的 状态、节点、边、记忆、流式输出 串成一个可恢复、可中断、可重放的 长生命周期工作流。 跟"包了一层 LLM 的 chain"不同,LangGraph 把 状态显式化、控制流显式化、持久化 显式化 ,让你既能用 create react agent
以 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; PyPIlanggraph-prebuilt为1.1.0,langgraph-checkpoint为4.1.1; npm@langchain/langgraph为1.3.6,@langchain/langgraph-cli为1.2.5。 本文主体继续按图编排架构讲解,版本号以本条与官方包为准。
#目录
- 写在前面:LangGraph 是什么 / 不是什么 0.5. 一张图看懂:LangGraph = Model + Harness(积木而非成品)
- Agent 是什么 —— StateGraph + Node 函数 + state schema
- Agent 怎么活起来 ——
invoke/astream/ Cloud 三条触发链路 - Agent 怎么思考 —— ReAct prebuilt + 自定义图 + Supervisor / Swarm
- Agent 怎么行动 —— ToolNode + Handoff + Interrupt
- Agent 怎么记忆 —— state + Checkpointer + Store
- Agent 怎么开口 —— 五种流模式
- Agent 的外脑 —— LangChain ChatModel 抽象
- 关键设计权衡(架构师视角)
- 附录:组件地图 / 关键概念 / 源码索引
#零、写在前面: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:
- 仓库:https://github.com/langchain-ai/langgraph
- 文档站:https://langchain-ai.github.io/langgraph/ →(已迁)https://docs.langchain.com/oss/python/langgraph/
- API Reference:https://reference.langchain.com/python/langgraph
- Deep Agents:https://github.com/langchain-ai/deepagents
- Supervisor / Swarm:https://github.com/langchain-ai/langgraph-supervisor-py / https://github.com/langchain-ai/langgraph-swarm-py
#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 | TypedDict 或 pydantic.BaseModel |
描述图运行期间会出现的所有字段(agent 的"血液") |
| Node | def(state) -> dict 或 async 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()
三件事值得展开:
- State 字段必须有 reducer。
Annotated[list, add_messages]这一行是说:messages通道用add_messages这个 reducer 来合并新值(追加而非覆盖、按 id 去重、自动给新消息打 uuid)。不写 reducer 就默认覆盖,每次 node 返回的字段 会替换前值。 - Node 只返回"差量"。返回
{"messages": [new_msg]}不是替换整个 messages, 而是交给 reducer 合并。这一点跟传统 chain 的 "input → output" 范式区别巨大。 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)
发生了什么:
invoke把入参跟 checkpointer 里thread_id="user-42"的最近 state 合并(reducer 起作用)- Pregel 引擎进入超步循环:每一步找出"该跑的 node 集合"(输入通道有更新且未消费)
- 并发执行该集合,收集每个 node 的返回,按 reducer 合并到 state
- 把新 state 序列化进 checkpointer(一个 checkpoint = 一个 superstep 末尾的快照)
- 直到没有 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
)
它内部干了什么(看源码):
- 定义
AgentState:messages: Annotated[Sequence[BaseMessage], add_messages]remaining_steps: RemainingSteps
- 把
prompt转成Runnable(str / SystemMessage / callable / Runnable 都接受) - 校验
model.bind_tools()是否已经绑了所有 tools,没绑就帮你 bind - 加两个节点:
agent(call model)+tools(ToolNode(tools)) - 加条件边:agent → 有 tool_call 走 tools / 否则 END / 步数耗尽抛错
- tools 节点回到 agent,循环
- 返回
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=...)
它干的事:
- 把每个子 agent 当作 graph 的一个 node(子 agent 本身也是一张图,subgraph 嵌套)
- supervisor 节点是一个 LLM,给它一份"delegate_to_research / delegate_to_math / finish" 的 handoff tool 列表
- supervisor 根据当前 messages 选一个 handoff tool → 该子 agent 节点接管
- 子 agent 跑完回到 supervisor,supervisor 再判断要不要继续派活
- 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=...)
它干的事:
- state 里多一个
active_agent: str字段 - 每个 handoff tool 返回
Command(goto="Bob", graph=Command.PARENT, update={"active_agent": "Bob"}) - ToolNode 检测到 Command 直接跳节点(不再回到调用 agent)
- 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])
它的工作很机械:
- 从 state 拿最后一条 AIMessage
- 遍历
tool_calls字段 - 每个 tool_call 找对应 tool,并发执行(v2 用 Send)
- 把每个返回包成 ToolMessage,挂回 messages(reducer 是
add_messages) - 出现异常按
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 用 InjectedState
和 InjectedToolCallId 注解告诉 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)
发生了什么:
interrupt()抛GraphInterrupt异常- Pregel 引擎捕获,把异常负载(
{"draft": ..., "type": ...})写进当前 checkpoint 的interrupts列表 - invoke / stream 提前返回,用户拿到
__interrupt__chunk - 外部决策回来后,调用方
invoke(Command(resume=...)) - 引擎从最近 checkpoint 恢复,把
resume值塞回interrupt()调用点 - 节点继续执行,跟没中断过一样
关键认知: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) # 第二轮自动接上
两条规则:
- 没有
thread_id的 invoke 没有记忆(每次新 state) - 同一个
thread_id的多次 invoke 会自动接 messages(reducer 合并)
老消息怎么裁?用 pre_model_hook(create_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-redis(redis-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})
InjectedStore 跟 InjectedState 一个套路——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_endon_tool_start/on_tool_endon_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 对模型的唯一要求:
- 实现
BaseChatModel(invoke/ainvoke/astream) - 实现
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_agent 的 model 参数也能传 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 为啥要
把循环显式画成图?
回答:把控制流移到图上有三个直接收益:
- 可观测——每一跳都是 superstep,Studio / LangSmith 直接出节点级 timeline
- 可中断——任意节点前后插
interrupt,业务不需要自己埋 hook - 可重放——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。