LangGraph 与可恢复 Agent
Agent 不是「一条链」,是带环的图:可能重试、可能分支、可能停下来等人审批、可能跑到一半崩了要接着跑。 LangGraph 就是把这个「图」显式建模出来的框架。核心价值不是编排,是状态可持久化、可恢复、可回放。
一、为什么不直接用 Chain
| Chain(LCEL) | LangGraph | |
|---|---|---|
| 结构 | 有向无环(DAG) | 允许环(循环、重试、反思) |
| 状态 | 一路传递的中间值 | 显式 State,可读写、可合并 |
| 中断 | 不支持 | 可在任意节点暂停等人 |
| 恢复 | 崩了就重来 | checkpoint 断点续跑 |
| 可观测 | 调用链 | 每步状态快照(可回放) |
判断标准:需要循环、需要人审批、需要断点续跑 → LangGraph;一次性线性处理 → Chain 就够。
二、四个核心概念
State(共享状态) + Node(做事) + Edge(去哪) + Compile(跑)State(共享状态) + Node(做事) + Edge(去哪) + Compile(跑)State:用 TypedDict 定义
python
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages] # reducer:新消息追加而不是覆盖
order_id: str
attempts: intfrom typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages] # reducer:新消息追加而不是覆盖
order_id: str
attempts: int关键在 reducer:add_messages 表示「节点返回的消息要追加」,不加 reducer 就是覆盖。 这是新手最常踩的坑 —— 发现对话上下文丢了,八成是这里。
Node:一个函数,返回状态的局部更新
python
def check_stock(state: State):
stock = inventory.get(state["order_id"])
return {"attempts": state["attempts"] + 1, "stock": stock} # 只返回要改的字段def check_stock(state: State):
stock = inventory.get(state["order_id"])
return {"attempts": state["attempts"] + 1, "stock": stock} # 只返回要改的字段节点返回的是 partial update,LangGraph 用 reducer 合并进 State。
Edge:固定边与条件边
python
from langgraph.graph import StateGraph, START, END
builder = StateGraph(State)
builder.add_node("check_stock", check_stock)
builder.add_node("create_order", create_order)
builder.add_node("ask_human", ask_human)
builder.add_edge(START, "check_stock")
# 条件边:根据状态决定去哪
def route(state):
if state["stock"] > 0:
return "create_order"
return "ask_human"
builder.add_conditional_edges("check_stock", route, {
"create_order": "create_order",
"ask_human": "ask_human",
})
builder.add_edge("create_order", END)
app = builder.compile()
app.invoke({"messages": [], "order_id": "A1", "attempts": 0})from langgraph.graph import StateGraph, START, END
builder = StateGraph(State)
builder.add_node("check_stock", check_stock)
builder.add_node("create_order", create_order)
builder.add_node("ask_human", ask_human)
builder.add_edge(START, "check_stock")
# 条件边:根据状态决定去哪
def route(state):
if state["stock"] > 0:
return "create_order"
return "ask_human"
builder.add_conditional_edges("check_stock", route, {
"create_order": "create_order",
"ask_human": "ask_human",
})
builder.add_edge("create_order", END)
app = builder.compile()
app.invoke({"messages": [], "order_id": "A1", "attempts": 0})三、checkpointer:持久化与断点续跑
python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver # 生产用 Postgres/Sqlite
app = builder.compile(checkpointer=SqliteSaver.from_conn_string("lg.db"))
config = {"configurable": {"thread_id": "user-42"}} # thread_id = 一次会话/任务
app.invoke(input_state, config)from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver # 生产用 Postgres/Sqlite
app = builder.compile(checkpointer=SqliteSaver.from_conn_string("lg.db"))
config = {"configurable": {"thread_id": "user-42"}} # thread_id = 一次会话/任务
app.invoke(input_state, config)有了 checkpointer 才能做的事:
| 能力 | 怎么用 |
|---|---|
| 断点续跑 | 进程崩了,同 thread_id 再 invoke(None, config) 从最后一个 checkpoint 继续 |
| 时间旅行 | app.get_state_history(config) 列出历史快照,选一个 update_state 回退重跑 |
| 多轮会话 | 同 thread_id 自动带着上一轮状态 |
| 人机协同 | 中断后等外部输入,再恢复 |
注意:thread_id 是你自己的业务概念(会话 id、任务 id),不是数据库线程 —— 命名容易误导。
四、human-in-the-loop:中断与恢复
高危动作(转账、删数据、发邮件)必须让人确认,LangGraph 的做法是在图里真的停下来。
python
from langgraph.types import interrupt, Command
def approve_transfer(state: State):
# 执行到这里会暂停,把 payload 抛给调用方
decision = interrupt({"action": "transfer", "amount": state["amount"]})
if decision != "yes":
return {"status": "rejected"}
return {"status": "approved"}from langgraph.types import interrupt, Command
def approve_transfer(state: State):
# 执行到这里会暂停,把 payload 抛给调用方
decision = interrupt({"action": "transfer", "amount": state["amount"]})
if decision != "yes":
return {"status": "rejected"}
return {"status": "approved"}python
# 第一次调用:会停在 interrupt
result = app.invoke(state, config)
# result 里带着待确认信息
# 外部确认后:用 Command 带着答案恢复
app.invoke(Command(resume="yes"), config)# 第一次调用:会停在 interrupt
result = app.invoke(state, config)
# result 里带着待确认信息
# 外部确认后:用 Command 带着答案恢复
app.invoke(Command(resume="yes"), config)两个要点:
- interrupt 之前的代码会重跑(恢复时节点从开头执行)→ 所以确认前的副作用要幂等或放到确认后;
- 老版本用
compile(interrupt_before=["node"])静态中断,新版本推荐用interrupt()动态中断。
五、并行与 Map-Reduce
python
from langgraph.types import Send
def fan_out(state: State):
# 一个节点拆出 N 个并行分支
return [Send("process_doc", {"doc": d}) for d in state["docs"]]
builder.add_conditional_edges("split", fan_out, ["process_doc"])
builder.add_edge("process_doc", "merge") # 自动等待所有分支完成from langgraph.types import Send
def fan_out(state: State):
# 一个节点拆出 N 个并行分支
return [Send("process_doc", {"doc": d}) for d in state["docs"]]
builder.add_conditional_edges("split", fan_out, ["process_doc"])
builder.add_edge("process_doc", "merge") # 自动等待所有分支完成fan-in 的语义:多条边指向同一节点时,默认等全部完成再执行一次(不是执行 N 次)。如果需要聚合列表,用 reducer(Annotated[list, operator.add])。
六、流式输出
stream_mode | 输出 |
|---|---|
values | 每步的完整状态 |
updates | 每步的增量(最常用) |
messages | LLM 的 token 流(前端流式渲染用这个) |
custom | 节点里 get_stream_writer() 自己写的进度 |
python
for chunk in app.stream(input_state, config, stream_mode="messages"):
token, meta = chunk
yield token.content # 直接推给前端 SSEfor chunk in app.stream(input_state, config, stream_mode="messages"):
token, meta = chunk
yield token.content # 直接推给前端 SSE七、生产的三个注意点
1. 节点必须可重放
恢复、时间旅行、重试都会重新执行节点。所以节点里的副作用(写库、发消息、调外部 API)必须:
- 幂等(用业务唯一键去重),或
- 移到「确认之后」,或
- 用 outbox 模式(先落库再异步发)。
2. checkpoint 表会膨胀
每步都存快照,长会话会积累大量行。要做:定期清理过期 thread、只保留关键节点的 checkpoint、用 Postgres 而不是文件。
3. 环要有出口
builder.add_edge("reflect", "generate") 这种环,必须配最大轮次:
python
def should_continue(state):
if state["attempts"] >= 3:
return END
return "generate"def should_continue(state):
if state["attempts"] >= 3:
return END
return "generate"否则模型一旦不收敛就是死循环 + 烧钱。
八、和别的编排方式怎么选
| 方案 | 适合 |
|---|---|
| if/else + 函数调用 | 步骤少、逻辑简单(别过度设计) |
| LangChain Chain | 线性、无状态、无中断 |
| LangGraph | 有环、要人审批、要断点续跑、要回放 |
| Temporal / Airflow | 长周期定时任务、跨系统编排(不涉及 LLM 决策) |
九、速查
| 问题 | 答案 |
|---|---|
| LangGraph 的核心价值 | 状态可持久化 + 可恢复 + 可回放 |
| State 怎么定义 | TypedDict + Annotated[..., reducer] |
| 节点返回什么 | 状态的局部更新,不是完整状态 |
| 为什么对话上下文丢了 | 忘了给 messages 加 add_messages reducer |
| checkpointer 干什么 | 存每步快照,支持续跑与时间旅行 |
| thread_id 是什么 | 业务上的会话/任务 id |
| 人体循环怎么做 | interrupt() 暂停 + Command(resume=...) 恢复 |
| 恢复时节点会重跑吗 | 会,所以副作用必须幂等 |
| 环怎么防死循环 | 条件边里加最大轮次判断 |
| 前端流式用哪个模式 | stream_mode="messages" |
Cao Note