LangGraph 系统学习文档

适合人群：已经了解 LangChain 或基础 LLM 应用，想进一步掌握复杂 Agent、状态机、可恢复执行、人类审批、多 Agent 编排和生产级 AI 工作流的学习者。
学习目标：从图、状态、节点、边这些基础概念入门，逐步掌握 LangGraph 的 StateGraph、Reducer、Command、Checkpoint、Persistence、Interrupt、Human-in-the-loop、Memory、Subgraph 和生产实践。
核对时间：2026-06-25。本文以 LangGraph 官方文档当前信息为基础整理。官方将 LangGraph 定位为低层级 orchestration framework 和 runtime，用于构建、管理和部署 long-running、stateful agents。
官方资料：
LangGraph Overview：https://docs.langchain.com/oss/python/langgraph/overview
LangGraph Graph API：https://docs.langchain.com/oss/python/langgraph/graph-api
LangGraph Persistence：https://docs.langchain.com/oss/python/langgraph/persistence
LangGraph Human-in-the-loop：https://docs.langchain.com/oss/python/langgraph/human-in-the-loop
LangGraph Memory：https://docs.langchain.com/oss/python/langgraph/memory

第一篇：基础认知

1. LangGraph 是什么

LangGraph 是一个用于构建有状态 Agent 和复杂 AI 工作流的框架。它把 Agent 流程表示成图：

text

State
  -> Node A
  -> Node B
  -> Conditional Edge
  -> Node C 或 END

它关注：

状态管理
多步骤执行
持久化
故障恢复
人类审批
长时间运行任务
多 Agent 协作

如果 LangChain 更像“快速组装 LLM 应用的组件库和 Agent harness”，LangGraph 更像“Agent 工作流运行时”。

2. 为什么需要 LangGraph

简单 LLM 应用：

text

用户输入 -> 模型输出

普通 RAG：

text

用户输入 -> 检索 -> 模型回答

复杂 Agent：

text

用户目标
  -> 分析任务
  -> 调用工具
  -> 等待外部结果
  -> 人工审批
  -> 继续执行
  -> 失败恢复
  -> 生成最终报告

复杂流程的问题：

需要记住当前执行到哪一步。
需要在失败后恢复。
需要暂停等待用户或人工审批。
需要多分支条件逻辑。
需要记录每一步的状态变化。

LangGraph 正是为这些问题设计的。

3. LangChain 和 LangGraph 的区别

维度	LangChain	LangGraph
核心定位	快速构建 LLM/Agent 应用	构建有状态、可恢复、长期运行 Agent
抽象方式	模型、工具、Prompt、Agent	State、Node、Edge、Graph
复杂流程	可做，但复杂后不够清晰	更适合
持久化	create_agent 支持 checkpointer	一等能力
Human-in-the-loop	可实现	官方重点能力
适合阶段	入门和中等复杂度	复杂生产 Agent

学习建议：

text

先学 LangChain 的模型、工具、RAG
再学 LangGraph 的状态图和持久化
最后用 LangGraph 承载复杂 Agent

4. 图模型思维

图由节点和边组成。

text

节点：执行某个动作
边：决定下一步去哪
状态：节点之间共享和更新的数据

例如客服 Agent：

text

START
  -> classify_intent
  -> 如果是退款：refund_flow
  -> 如果是物流：shipping_flow
  -> summarize_answer
  -> END

LangGraph 的关键是把“隐含在代码 if/else 里的流程”显式建模成图。这样更容易调试、恢复和扩展。

4.1 底层机制：Pregel 与超级步骤（super-step）

LangGraph 的执行引擎借鉴 Google 的 Pregel 论文，用**消息传递（message passing）**驱动图，按离散的"超级步骤"推进：

text

每个 super-step（图的一次"心跳"）:
1. 收到新消息（state 更新）的节点被激活；同一 super-step 内的节点并行执行
2. 活跃节点运行函数，返回更新，沿出边传给下游节点
3. super-step 结束：框架按各字段的 reducer 合并更新；没有收到消息的节点投票 halt

当所有节点都 inactive 且没有消息在途时，整个执行终止。这个模型直接决定了三件事：

事实	后果
checkpoint 写在 super-step 边界	恢复和时间旅行只能从某个 super-step 开始，不能从任意代码行恢复
同一 super-step 的节点并行执行	并行分支写同一个 state 字段时，必须靠 reducer 定义合并规则（见第 10 章）
节点本质是消息处理函数	节点应近似纯函数：重放和 fault-tolerance 会重新执行节点，副作用必须幂等（见第 19 章）

💡 顺序图 START -> A -> B -> END 有 3 个 super-step（输入、A、B）；分支汇聚 A -> B, A -> C, B -> D, C -> D 中 B 和 C 在同一个 super-step 并行。理解 super-step 边界是理解 checkpoint、并发 reducer 和恢复语义的前提。

5. 第一个 StateGraph

简化示例：

python

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    topic: str
    answer: str

def generate_answer(state: State):
    return {
        "answer": f"你想学习的是：{state['topic']}"
    }

builder = StateGraph(State)
builder.add_node("generate_answer", generate_answer)
builder.add_edge(START, "generate_answer")
builder.add_edge("generate_answer", END)

graph = builder.compile()

result = graph.invoke({
    "topic": "LangGraph"
})

print(result["answer"])

这个例子里：

State 定义图状态。
generate_answer 是节点。
add_edge 定义执行顺序。
compile 编译图。
invoke 执行图。

第二篇：核心概念

6. State

State 是图的共享状态。所有节点都读取 State，并返回对 State 的更新。

示例：

python

class AgentState(TypedDict):
    messages: list
    user_id: str
    intent: str
    result: str

设计 State 时要注意：

不要把所有东西都塞进去。
保留流程真正需要的状态。
大对象可以放数据库，只在 State 中保存引用 ID。
敏感信息要谨慎存储。

State 是 LangGraph 的核心。流程是否清晰，取决于 State 是否设计得好。

7. Node

Node 是图中的执行单元。它接收 State，返回 State 更新。

python

def classify_intent(state: AgentState):
    user_message = state["messages"][-1]
    intent = "refund" if "退款" in user_message else "general"
    return {"intent": intent}

Node 可以做：

调用模型
调用工具
查询数据库
检索知识库
校验输出
等待审批
汇总结果

Node 应保持职责单一。一个节点里塞太多逻辑，会让图失去可读性。

8. Edge

Edge 定义节点之间的执行顺序。

普通边：

python

builder.add_edge("classify_intent", "answer")

含义：

text

classify_intent 执行后，进入 answer 节点。

Edge 让流程显式化。比起在一个巨大的函数里写 if/else，图结构更容易观察和调试。

9. START 和 END

START 表示图入口，END 表示图结束。

python

builder.add_edge(START, "classify_intent")
builder.add_edge("answer", END)

复杂图可能有多个路径通向 END。

text

START -> A -> B -> END
START -> A -> C -> END

10. Reducer

Reducer 用于定义状态字段如何合并。

例如消息列表通常不是覆盖，而是追加：

python

from typing import Annotated
from operator import add

class State(TypedDict):
    messages: Annotated[list, add]

如果没有 reducer，节点返回的字段通常会覆盖原字段。对于列表、日志、消息历史等字段，Reducer 非常重要。

常见用途：

追加 messages
合并中间结果
累积日志
汇总多个分支输出

10.1 reducer 的左右参数

每个 key 有独立的 reducer。它是一个二元函数：

python

new_value = reducer(left=current_state[key], right=node_update[key])

左参数永远来自已累积的状态，右参数永远来自节点最新的更新。未显式声明 reducer 的字段使用默认 reducer：丢弃 left、只保留 right（覆盖语义）。自定义 reducer 合并 left + right（累积语义）。并行节点在同一 super-step 写同一个字段时，reducer 就是冲突仲裁规则——没有 reducer 的字段被并行写会互相覆盖，只剩一份。

10.2 add_messages：消息通道为什么不能只用 operator.add

operator.add 只能追加，不能修改。而 human-in-the-loop、时间旅行编辑等场景需要“更新已有消息”。预置的 add_messages（langgraph.graph.message）按消息 id 处理：新消息追加、同 id 消息原位覆盖，并把 dict 形式的输入反序列化成 LangChain Message 对象：

python

from typing import Annotated
from typing_extensions import TypedDict
from langchain.messages import AnyMessage
from langgraph.graph.message import add_messages

class GraphState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

因为更新总是被反序列化成 Message，读消息用点号：state["messages"][-1].content。配合 RemoveMessage（携带目标消息 id）可以删除指定消息。这个模式太常用，官方直接提供预置状态 MessagesState（只有带 add_messages 的 messages 一个键），实际项目通常继承它再加字段：

python

from langgraph.graph import MessagesState

class State(MessagesState):
    documents: list[str]

⚠️ 想绕过 reducer 直接整体覆盖某个累积字段（例如整段替换 messages），用 langgraph.types.Overwrite 包住值再返回，而不是硬塞裸列表。

11. Command

Command 可以让节点同时更新状态并决定下一步。

适合场景：

节点内部判断路由。
根据工具结果决定下一步。
多 Agent handoff。

概念示意：

python

from typing import Literal
from langgraph.types import Command

def route(state) -> Command[Literal["refund_node", "general_node"]]:
    if state["intent"] == "refund":
        return Command(goto="refund_node")
    return Command(goto="general_node")

Command 让某些动态跳转更自然，但不要滥用。简单条件路由用 conditional edge 更清晰。

11.1 补充：Send 动态扇出（map-reduce）

Command 解决“跳到哪个节点、带什么更新”；Send（langgraph.types，since v0.2）解决另一个问题：在运行时动态决定某个节点跑几次、每次喂什么输入。在 conditional edge 的返回函数里构造一个 Send 列表，就能对同一节点并行扇出 N 个任务，每个任务收到的是你自定义的 arg，而不是全局 state：

python

from typing import Annotated
import operator
from typing_extensions import TypedDict
from langgraph.types import Send
from langgraph.graph import END, START, StateGraph

class OverallState(TypedDict):
    subjects: list[str]
    jokes: Annotated[list[str], operator.add]

def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

builder = StateGraph(OverallState)
builder.add_node("generate_joke", lambda state: {"jokes": [f"Joke about {state['subject']}"]})
builder.add_conditional_edges(START, continue_to_jokes)
builder.add_edge("generate_joke", END)
graph = builder.compile()

graph.invoke({"subjects": ["cats", "dogs"]})
# {'subjects': ['cats', 'dogs'], 'jokes': ['Joke about cats', 'Joke about dogs']}

底层语义：每个 Send 相当于往目标节点投递一条独立消息，N 个 Send 在同一 super-step 并行执行；结果要汇聚回主 state，目标节点写回的通道必须有 reducer（如上面的 operator.add），否则并行写互相覆盖。这是多 Agent“并行探索再汇总”和 map-reduce 的标准实现。Send 还支持可选 timeout 参数，为单个推送任务设置超时策略。

Command 与 Send 的选择：

需求	用
固定节点、按状态选一条路	conditional edge
节点内同时更新 state + 指定去向（含多 Agent handoff）	`Command(goto=..., update=...)`（节点返回值）
运行时才知道个数的并行扇出，每次喂不同输入	conditional edge 返回 `Send` 列表

12. Conditional Edge

Conditional Edge 根据状态选择下一步。

示意：

python

def route_by_intent(state):
    return state["intent"]

builder.add_conditional_edges(
    "classify_intent",
    route_by_intent,
    {
        "refund": "refund_flow",
        "shipping": "shipping_flow",
        "general": "general_answer",
    }
)

适合：

意图分类
成功/失败分支
是否需要工具
是否需要人工审批
是否继续循环

第三篇：持久化与记忆

13. Checkpoint

Checkpoint 是图执行状态的保存点。

它让系统可以：

保存每一步状态。
崩溃后恢复。
暂停后继续。
回看历史。
实现短期记忆。

如果没有 checkpoint，Agent 一旦中断，就很难知道之前执行到了哪里。

13.1 底层机制：checkpoint 到底存了什么

checkpointer 把图状态快照按 super-step 边界组织进 thread。对 START -> A -> B -> END 的一次调用会产生 4 个 checkpoint：空初始状态（next=START）、输入写入后（next=A）、A 输出后（next=B）、B 输出后（next 为空）。除了 super-step 快照，LangGraph 还在节点（task）级别持久化写入（checkpoint_writes 表）：同一 super-step 里如果某个节点失败，其他已成功节点的写入已经落盘，恢复时不必重跑——这就是 pending writes 容错机制。

配套 API：

API	作用
`graph.get_state(config)`	取 thread 最新状态，返回 `StateSnapshot`
`graph.get_state_history(config)`	按时间倒序列出历史 checkpoint（time travel / replay 的入口）
`graph.update_state(config, values, as_node=...)`	人工改状态：创建新 checkpoint，不修改旧的；更新同样走 reducer；`as_node` 决定“这条更新算哪个节点发的”，影响下一步路由

⚠️ MemorySaver / InMemorySaver 只存内存，进程重启即丢；生产用 PostgresSaver（langgraph-checkpoint-postgres，需单独安装，setup() 建表）或 SqliteSaver。checkpoint 会随长对话无限增长，要配定期清理/保留策略。

13.2 序列化与加密

状态要写进数据库就必须序列化。默认实现是 JsonPlusSerializer（基于 msgpack + JSON，来自 langgraph-checkpoint 包），能处理 LangChain/LangGraph 原语、datetime、enum、Pydantic 模型等。不被 msgpack 支持的对象（如 pandas DataFrame）可开启 JsonPlusSerializer 的 pickle_fallback 参数回退到 pickle。敏感数据场景可以给任何 BaseCheckpointSaver 实现的 serde 参数传入 EncryptedSerializer（可用 EncryptedSerializer.from_pycryptodome_aes 创建，AES 密钥读环境变量 LANGGRAPH_AES_KEY 或直接传 key 参数）实现整体加密。

💡 设计启示：State 里只放“值得被序列化并长期保存”的东西。连接对象、文件句柄、大型二进制这类不可靠/不宜序列化的内容，放外部存储，State 里只存引用 ID（与第 6 章 State 设计原则一致）。

14. Thread

Thread 表示一次会话或一次任务执行的逻辑上下文。

例如：

text

用户 A 的客服会话 thread_id = user-a-session-1
用户 B 的客服会话 thread_id = user-b-session-1

每个 thread 有自己的状态历史。前端聊天系统中，可以把会话 ID 映射为 LangGraph thread ID。

14.1 使用规则与陷阱

只要编译时挂了 checkpointer，每次 invoke/stream 必须在 config 里给 thread_id：{"configurable": {"thread_id": "..."}}。没有它，checkpointer 既存不了状态也没法恢复 interrupt。
同一个 thread_id = 续上这条会话（读旧状态、追加新 checkpoint）；换一个 thread_id = 全新空状态。thread_id 就是持久化游标。
⚠️ 用 PostgresSaver 时 thread_id 有列长限制（超过 255 字符会报数据库错误），建议 UUID 或 hash。
子图有自己独立的 checkpoint namespace（checkpoint_ns，形如 "node_name:uuid"，嵌套用 | 连接）：父图不一定立即可见子图对共享键的更新。需要跨图边界共享数据时，优先用 Store（第 16 章），或按官方文档显式配置子图写父 checkpoint。

15. Short-term Memory

短期记忆通常就是当前 thread 内的上下文。

它适合保存：

当前对话消息
当前任务状态
最近工具结果
当前审批状态

短期记忆不适合保存长期用户画像和跨会话知识。长期信息应进入数据库、向量库或专门 memory store。

16. Long-term Memory

长期记忆跨 thread 存在。

例如：

用户偏好
历史项目背景
常用配置
长期知识

设计长期记忆要谨慎：

什么信息值得长期保存？
用户是否授权？
如何修改和删除？
如何避免错误记忆污染后续回答？
如何处理隐私？

长期记忆不是越多越好。错误记忆比没有记忆更危险。

16.1 官方机制：Checkpointer vs Store

LangGraph 用两套互补的持久化系统分别承载短期/长期记忆：

	Checkpointer	Store
存什么	图状态快照	应用定义的 key-value 数据
作用域	单个 thread	跨 thread
记忆类型	短期、thread 内	长期、跨会话
用途	对话连续性、HITL、time travel、容错	用户偏好、事实、共享知识
访问方式	config 里传 `thread_id`	节点或应用代码里读写 item

Store 用 namespace/key 组织数据（namespace 是元组，如 ("users",)），生产用持久实现（PostgresStore、RedisStore、MongoDBStore），开发用 InMemoryStore。节点内通过 runtime.store 访问，工具里通过 ToolRuntime.store 访问。典型做法是给 Agent 配“读记忆/写记忆”两个工具，让模型自己决定何时存取，再由人在环或策略层控制写入质量。

python

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

graph = builder.compile(checkpointer=InMemorySaver(), store=InMemoryStore())

两者经常同时用：checkpointer 管当前会话，store 管跨会话知识。

17. Persistence 的价值

Persistence 让 Agent 从一次“内存中的函数调用”变成一个可恢复的系统。

价值：

长任务可持续运行。
服务重启不丢状态。
支持人工审批。
支持调试历史状态。
支持时间旅行和回放。

典型应用：

自动生成报告
多步骤数据分析
AI 客服工单
代码审查助手
复杂企业审批流

18. Durable Execution

Durable Execution 是持久、可恢复执行。

通俗理解：

text

Agent 每走一步都保存状态。
如果进程崩溃、服务重启或等待人工输入，下次可以从上一步继续。

这和普通脚本不同。普通脚本失败后通常从头再来；durable execution 允许从中间恢复。

对于生产 Agent，这是非常关键的能力。

18.1 durability 三档与节点级重试

执行时可指定 durability 模式，在性能和数据安全之间取舍：

模式	落盘时机	权衡
`"exit"`	仅在图执行退出时（成功/报错/interrupt）	长图性能最好；进程中途崩溃则本次运行全丢
`"async"`（默认档位思路）	下一步执行的同时异步写	性能好；崩溃瞬间有小概率丢最新 checkpoint
`"sync"`	每个 super-step 开始前同步写完	最耐久；每步都付写库开销

除 checkpoint 恢复外，LangGraph 支持节点级容错配置（官方 Fault tolerance 文档）：add_node() 可直接挂 retry_policy=RetryPolicy(...) 和 timeout=...（langgraph>=1.2）。RetryPolicy 默认值：max_attempts=3、initial_interval=0.5 秒、backoff_factor=2.0、max_interval=128.0、jitter=True；默认重试 ValueError/OSError/requests/httpx 等瞬时异常（可用 retry_on 自定义）。超时支持 run_timeout（墙钟上限）与 idle_timeout（无进展判定，配合 runtime.heartbeat() 心跳），超时抛 NodeTimeoutError 并可被 retry policy 捕获重试。节点内可用 runtime.execution_info.node_attempt 感知当前是第几次尝试，做“首次走主接口、重试降级到备用接口”的逻辑。两者分工：retry/timeout 处理“再试一次大概率就好”的错误，durable execution 处理“进程没了也要接着跑”的场景。

第四篇：Human-in-the-loop 与 Agent

19. Interrupt

Interrupt 用于中断图执行，等待外部输入后再恢复。

适合：

人工审批
用户补充信息
高风险操作确认
中间结果编辑

示例场景：

text

Agent 准备发送退款申请
  -> interrupt 暂停
  -> 前端展示审批卡片
  -> 人工点击同意
  -> 图继续执行退款工具

19.1 中断-恢复的底层语义

interrupt()（langgraph.types）不是线程挂起，而是干净的“保存-退出-重放”：

text

1. 节点函数执行到 interrupt(payload) —— 整个图停在这条指令上
2. checkpointer 落盘当前图状态；执行以“暂停”形式结束，不占任何进程或线程
3. payload 返回给调用方：invoke() 结果里的 __interrupt__，或 stream_events(version="v3") 的 stream.interrupts（stream.interrupted 为 True）
4. 外部（几秒或几天后）用同一个 thread_id 重新调用图，输入为 Command(resume=human_value)
5. ⚠️ 恢复时该节点从头重新执行——interrupt() 之前的代码会被再跑一遍，直到再次走到 interrupt()，此时它不再暂停，而是直接返回 resume 值作为自己的返回值

因此三条硬规则（官方明确列为反模式）：

⚠️ interrupt 之前的代码必须幂等。在 interrupt 前建 DB 记录、往历史列表追加、发外部请求，每次恢复都会重复执行一次，造成脏数据。副作用要么放到 interrupt 之后，要么做存在性检查。
恢复必须复用同一 thread_id（payload 必须 JSON 可序列化）。
只有 Command(resume=...) 是设计给 invoke/stream 当输入用的；Command(update/goto/graph) 是节点返回值用的。多轮对话继续执行传普通 dict 输入，不要传 Command(update=...)。

python

from langgraph.types import interrupt, Command

def approval_node(state):
    # payload 给前端渲染审批卡片
    decision = interrupt({"question": "Approve this action?", "details": state["action_details"]})
    return Command(goto="proceed" if decision else "cancel")

config = {"configurable": {"thread_id": "approval-123"}}
graph.invoke({"action_details": "Transfer $500"}, config)   # 暂停
graph.invoke(Command(resume=True), config)                   # 批准并继续

并行分支同时 interrupt 时（fan-out 到多个节点各自 interrupt()），恢复可以把 {interrupt_id: resume_value} 的映射一次性传给 Command(resume=...)，每个回答对应该 id 的中断。

19.2 动态 interrupt vs 静态断点

	`interrupt()` 函数	`interrupt_before` / `interrupt_after`
时机	代码任意位置、可按条件触发	固定在某节点前/后
传值	payload 出去、resume 值回来	只暂停，无交换（恢复传 `None`）
定位	生产 HITL	调试单步、LangSmith Studio 里看状态

两者都要求编译时挂 checkpointer。官方态度明确：human-in-the-loop 用 interrupt()，静态断点留给调试。

20. 人工审批

人工审批流程：

text

模型提出操作
  -> 生成审批请求
  -> 图暂停
  -> 前端展示操作详情
  -> 用户批准/拒绝/修改
  -> 图恢复

前端要展示：

Agent 当前步骤
准备执行的动作
参数详情
风险提示
批准、拒绝、修改按钮

这正是前端开发者的优势：把复杂 Agent 状态转成用户能理解和控制的交互。

21. Tool Calling Agent

工具调用 Agent 常见循环：

text

LLM 判断是否需要工具
  -> 调用工具
  -> 工具返回结果
  -> LLM 继续判断
  -> 最终回答

在 LangGraph 中，这个循环可以显式建模：

text

agent_node
  -> 如果有 tool_call：tools_node
  -> tools_node -> agent_node
  -> 如果没有 tool_call：END

显式图结构让你能控制最大循环次数、工具权限、错误分支和中断点。

21.1 官方实现：ToolNode 与 ToolCallState

手写这个循环时，工具执行节点通常直接用 langgraph.prebuilt.ToolNode：它读取最后一条 AIMessage 的 tool_calls，执行各工具（支持并行调用），把结果打包成 ToolMessage 列表返回。声明了 runtime: ToolRuntime 参数的工具，由 ToolNode 在调用时自动注入（与 LangChain 文档的 ToolRuntime 同一套机制），模型看不到这个参数。另一条路是官方正在推的 Middleware for LangGraph：ToolCallState 等中间件状态让工具调用生命周期（请求、执行、结果、错误）可被拦截和扩展，而不用把逻辑埋进节点函数。

控制循环上限的正式手段是 recursion_limit（config 级，默认 25）：超过抛 GraphRecursionError。⚠️ 一个常见坑——一次工具调用往返会消耗 2 个 super-step（agent 节点 + tools 节点），所以默认 25 步大约只够 12 轮工具调用。更优雅的做法是用 managed value RemainingSteps 在图内主动感知剩余额度（见第 24 章）。

22. ReAct 思想

ReAct 是 Reasoning + Acting。

大致流程：

text

思考要做什么
  -> 执行动作
  -> 观察结果
  -> 继续思考

现代工具调用 Agent 不一定显式输出思考过程，但工程上仍然有类似循环：

text

模型决策 -> 工具执行 -> 观察工具结果 -> 再决策

生产系统不应把模型的内部推理直接暴露给用户，用户需要的是清晰结果、必要过程状态和可控操作。

23. 多 Agent 与 Subgraph

Subgraph 是图里的子图。

适合：

把复杂流程拆成模块。
多 Agent 协作。
复用某个流程。
隔离状态。

示例：

text

主图：任务分发
  -> research_subgraph
  -> coding_subgraph
  -> review_subgraph
  -> final_report

多 Agent 不要为了“角色好听”而拆。只有当上下文隔离、流程复用或并行探索真的有价值时，再使用多 Agent。

23.1 两种组合方式的机制差别

官方把子图用法分成两类，选择依据是父子图是否共享状态键：

方式	适用	写法
在节点函数里 `subgraph.invoke(...)`	父子状态 schema 不同（无共享键），需要显式转换	节点内把父 state 映射为子图输入，拿到输出再映射回父 state 返回
把编译后的子图直接 `add_node(name, subgraph)`	父子共享状态键，子图直接读写父图通道	无需包装函数；子图作为普通节点参与父图调度

要点：

子图默认拥有独立 checkpoint namespace（"node_name:uuid"，见 14.1），所以“每个 Agent 保留私有消息历史”要用节点内 invoke 的方式；共享键的读写才会同步反映到父图。
子图内 interrupt() 时，恢复会从父图中调用子图那个节点的开头重新执行（节点里 interrupt 之前的代码同样重跑，见 19.1 幂等规则）。
create_agent 返回的编译图可以直接当子图节点用（配 name= 参数），middleware 钩子照常运行——这是官方推荐的“多 Agent = 图里嵌 Agent 节点”路线。

24. 错误恢复和重试

Agent 常见错误：

模型输出格式不对。
工具参数不合法。
工具超时。
检索为空。
外部 API 失败。
审批被拒绝。

处理方式：

节点内重试。
错误写入 State。
路由到修复节点。
降级回答。
等待人工处理。
从 checkpoint 恢复。

不要让错误直接炸掉整个流程。生产 Agent 要把失败也当成状态。

24.1 recursion limit 的主动处理：RemainingSteps

react 循环失控的兜底是 GraphRecursionError，但被动 catch 意味着图被强制终止、拿不到优雅结果。官方推荐主动方案：在 State 里声明 managed value RemainingSteps（langgraph.managed），运行时自动注入剩余额度，配合 conditional edge 提前收尾：

python

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.managed import RemainingSteps

class State(TypedDict):
    messages: Annotated[list, lambda x, y: x + y]
    remaining_steps: RemainingSteps  # managed value，无需节点写它

def route_decision(state: State) -> Literal["agent", "fallback"]:
    if state["remaining_steps"] <= 2:
        return "fallback"
    return "agent"

方案	检测时机	结果
主动：`RemainingSteps` + 路由到 fallback 节点	超限前	图正常走到收尾节点，可返回部分结果、写审计
被动：外层 try/except `GraphRecursionError`	超限后	执行被终止，中间产出可能丢失

也可以直接在节点里读 config["metadata"]["langgraph_step"] 观察当前步数（同族元数据还有 langgraph_node、langgraph_triggers 等），用于日志和追踪。

第五篇：工程化与生产

25. LangGraph 项目结构

推荐结构：

text

ai-service/
├── app/
│   ├── graphs/
│   │   ├── customer_support.py
│   │   └── research_agent.py
│   ├── nodes/
│   ├── tools/
│   ├── state/
│   ├── memory/
│   ├── prompts/
│   └── api/
├── tests/
└── pyproject.toml

职责：

graphs：图定义和编译。
nodes：节点函数。
tools：工具实现。
state：TypedDict/Pydantic 状态定义。
memory：短期/长期记忆。
api：FastAPI/NestJS 网关对接。

26. 前端如何展示 LangGraph 状态

前端不应该只展示最终答案，还可以展示：

当前步骤
是否正在调用工具
工具名称
是否等待审批
可恢复任务列表
中间结果预览
错误和重试入口

交互示例：

text

用户提交：“帮我分析这些文档并生成报告”
前端展示：
1. 正在解析文档
2. 正在检索资料
3. 正在生成大纲
4. 等待你确认大纲
5. 正在生成最终报告

这类体验比单纯 loading 更可靠，也更适合复杂 Agent。

27. 与数据库和队列集成

生产架构：

text

前端
  -> API Gateway
  -> LangGraph Service
  -> Checkpointer Database
  -> Business Database
  -> Queue / Worker
  -> Vector Store

建议：

会话和任务记录进数据库。
长任务用队列异步执行。
Checkpoint 使用可靠存储。
工具调用写审计日志。
前端通过轮询、SSE 或 WebSocket 获取状态。

28. 调试、观测和评估

LangGraph 调试要看：

每个节点输入输出。
State 变化。
走了哪条边。
工具调用参数。
interrupt 发生位置。
恢复后的状态。

评估：

路由是否正确。
工具调用是否必要。
最终回答是否准确。
是否遵守审批规则。
是否能从失败中恢复。

复杂 Agent 没有观测和评估，几乎无法稳定维护。

29. 常见误区

误区一：所有 LangChain 应用都要改成 LangGraph。

纠正：简单链路没必要。复杂状态、长任务、审批和恢复才更适合 LangGraph。

误区二：图越复杂越高级。

纠正：图是为了让流程清晰。节点和边过多且命名混乱，反而更难维护。

误区三：State 里什么都放。

纠正：State 应保存流程必要数据。大文件、大结果、敏感数据应放外部存储。

误区四：有 checkpoint 就不用考虑错误。

纠正：checkpoint 只负责保存状态，错误处理仍要设计。

30. 学习路线和项目实战

第一阶段：图基础

State
Node
Edge
START/END
Conditional Edge

第二阶段：状态更新

Reducer
Message state
Command
循环控制

第三阶段：持久化

Checkpoint
Thread
短期记忆
失败恢复

第四阶段：Human-in-the-loop

Interrupt
审批 UI
恢复执行
操作审计

第五阶段：复杂 Agent

Tool Calling Agent
多 Agent
Subgraph
长任务
评估

项目建议：

text

AI 工单处理系统
- 前端：Nuxt/Vue + Tailwind
- API：NestJS
- AI 服务：Python + LangGraph
- 数据库：PostgreSQL
- 功能：工单分类、知识库检索、建议回复、退款审批、执行状态追踪

快速记忆清单

LangGraph 是有状态 Agent 编排框架和运行时。
核心抽象是 State、Node、Edge、Graph。
Checkpoint 让 Agent 能保存状态和恢复执行。
Interrupt 是 human-in-the-loop 的关键。
简单 Agent 用 LangChain 就够，复杂长期任务再用 LangGraph。
前端 AI 产品要展示 Agent 状态，而不是只展示最终回复。

LangGraph 系统学习文档 ​

目录 ​

第一篇：基础认知 ​

第二篇：核心概念 ​

第三篇：持久化与记忆 ​

第四篇：Human-in-the-loop 与 Agent ​

第五篇：工程化与生产 ​

第一篇：基础认知 ​

1. LangGraph 是什么 ​

2. 为什么需要 LangGraph ​

3. LangChain 和 LangGraph 的区别 ​

4. 图模型思维 ​

4.1 底层机制：Pregel 与超级步骤（super-step） ​

5. 第一个 StateGraph ​

第二篇：核心概念 ​

6. State ​

7. Node ​

8. Edge ​

9. START 和 END ​

10. Reducer ​

10.1 reducer 的左右参数 ​

10.2 add_messages：消息通道为什么不能只用 operator.add ​

11. Command ​

11.1 补充：Send 动态扇出（map-reduce） ​

12. Conditional Edge ​

第三篇：持久化与记忆 ​

13. Checkpoint ​

13.1 底层机制：checkpoint 到底存了什么 ​

13.2 序列化与加密 ​

14. Thread ​

14.1 使用规则与陷阱 ​

15. Short-term Memory ​

16. Long-term Memory ​

16.1 官方机制：Checkpointer vs Store ​

17. Persistence 的价值 ​

18. Durable Execution ​

18.1 durability 三档与节点级重试 ​

第四篇：Human-in-the-loop 与 Agent ​

19. Interrupt ​

19.1 中断-恢复的底层语义 ​

19.2 动态 interrupt vs 静态断点 ​

20. 人工审批 ​

21. Tool Calling Agent ​

21.1 官方实现：ToolNode 与 ToolCallState ​

22. ReAct 思想 ​

23. 多 Agent 与 Subgraph ​

23.1 两种组合方式的机制差别 ​

24. 错误恢复和重试 ​

24.1 recursion limit 的主动处理：RemainingSteps ​

第五篇：工程化与生产 ​

25. LangGraph 项目结构 ​

26. 前端如何展示 LangGraph 状态 ​

27. 与数据库和队列集成 ​

28. 调试、观测和评估 ​

29. 常见误区 ​

30. 学习路线和项目实战 ​

快速记忆清单 ​

LangGraph 系统学习文档

目录

第一篇：基础认知

第二篇：核心概念

第三篇：持久化与记忆

第四篇：Human-in-the-loop 与 Agent

第五篇：工程化与生产

第一篇：基础认知

1. LangGraph 是什么

2. 为什么需要 LangGraph

3. LangChain 和 LangGraph 的区别

4. 图模型思维

4.1 底层机制：Pregel 与超级步骤（super-step）

5. 第一个 StateGraph

第二篇：核心概念

6. State

7. Node

8. Edge

9. START 和 END

10. Reducer

10.1 reducer 的左右参数

10.2 add_messages：消息通道为什么不能只用 operator.add

11. Command

11.1 补充：Send 动态扇出（map-reduce）

12. Conditional Edge

第三篇：持久化与记忆

13. Checkpoint

13.1 底层机制：checkpoint 到底存了什么

13.2 序列化与加密

14. Thread

14.1 使用规则与陷阱

15. Short-term Memory

16. Long-term Memory

16.1 官方机制：Checkpointer vs Store

17. Persistence 的价值

18. Durable Execution

18.1 durability 三档与节点级重试

第四篇：Human-in-the-loop 与 Agent

19. Interrupt

19.1 中断-恢复的底层语义

19.2 动态 interrupt vs 静态断点

20. 人工审批

21. Tool Calling Agent

21.1 官方实现：ToolNode 与 ToolCallState

22. ReAct 思想

23. 多 Agent 与 Subgraph

23.1 两种组合方式的机制差别

24. 错误恢复和重试

24.1 recursion limit 的主动处理：RemainingSteps

第五篇：工程化与生产

25. LangGraph 项目结构

26. 前端如何展示 LangGraph 状态

27. 与数据库和队列集成

28. 调试、观测和评估

29. 常见误区

30. 学习路线和项目实战

快速记忆清单