Cao Note

外观

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: int
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: 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)

两个要点：

  1. interrupt 之前的代码会重跑（恢复时节点从开头执行）→ 所以确认前的副作用要幂等或放到确认后；
  2. 老版本用 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每步的增量（最常用）
messagesLLM 的 token 流（前端流式渲染用这个）
custom节点里 get_stream_writer() 自己写的进度
python
for chunk in app.stream(input_state, config, stream_mode="messages"):
    token, meta = chunk
    yield token.content      # 直接推给前端 SSE
for 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"

Copyright © 2024-present xiaocao