langgraph-architecture

LangGraph 架构决策

何时使用 LangGraph

在以下情况下使用 LangGraph：

• 有状态对话 - 带记忆的多轮交互
• 人在环路中 - 审批关卡、纠正和干预
• 复杂控制流 - 循环、分支、条件路由
• 多 Agent 协调 - 多个大模型协同工作
• 持久化 - 从检查点恢复、通过时间旅行进行调试
• 流式处理 - 实时令牌流式传输、进度更新
• 可靠性 - 重试、错误恢复、持久性保障

在以下情况下考虑替代方案：

场景	替代方案	原因
单次大模型调用	直接 API 调用	不值得引入额外开销
线性管道	LangChain LCEL	抽象更简单
无状态工具使用	函数调用	无需持久化
简单 RAG	LangChain 检索器	内置模式
批处理	异步任务	执行模型不同

状态架构决策

TypedDict 与 Pydantic

TypedDict	Pydantic
轻量、速度更快	运行时验证
类似字典的访问方式	属性访问
无验证开销	类型强制转换
序列化更简单	复杂的嵌套模型

建议：大多数情况下使用 TypedDict。需要验证或复杂嵌套结构时使用 Pydantic。

Reducer 选择

用例	Reducer	示例
聊天消息	add_messages	处理 ID、RemoveMessage
简单追加	operator.add	Annotated[list, operator.add]
保留最新值	None (LastValue)	field: str
自定义合并	Lambda	Annotated[list, lambda a, b: ...]
覆盖列表	Overwrite	绕过 reducer

状态大小注意事项

# SMALL STATE (< 1MB) - Put in state
class State(TypedDict):
    messages: Annotated[list, add_messages]
    context: str

# LARGE DATA - Use Store
class State(TypedDict):
    messages: Annotated[list, add_messages]
    document_ref: str  # Reference to store

def node(state, *, store: BaseStore):
    doc = store.get(namespace, state["document_ref"])
    # Process without bloating checkpoints

图结构决策

单图与子图

单图适用于：

• 所有节点共享相同的状态架构
• 简单的线性流程或分支流程
• 节点数 < 10

子图适用于：

• 需要不同的状态架构
• 在多个图之间复用组件
• 团队职责分离
• 复杂的分层工作流

条件边与 Command

条件边	Command
基于状态进行路由	路由 + 状态更新
独立的路由函数	在节点中进行决策
可视化更清晰	更灵活
标准模式	动态目标

# Conditional Edge - when routing is the focus
def router(state) -> Literal["a", "b"]:
    return "a" if condition else "b"
builder.add_conditional_edges("node", router)

# Command - when combining routing with updates
def node(state) -> Command:
    return Command(goto="next", update={"step": state["step"] + 1})

静态路由与动态路由

静态边（add_edge）：

• 构建时已知的固定流程
• 图的可视化更清晰
• 更易于理解

动态路由（add_conditional_edges, Command, Send）：

• 基于状态在运行时做出决策
• 由 Agent 驱动导航
• 扇出模式

持久化策略

检查点器选择

检查点器	用例	特性
InMemorySaver	仅用于测试	重启后丢失
SqliteSaver	开发	单个文件、本地
PostgresSaver	生产环境	可扩展、支持并发
自定义	特殊需求	实现 BaseCheckpointSaver

检查点范围

# Full persistence (default)
graph = builder.compile(checkpointer=checkpointer)

# Subgraph options
subgraph = sub_builder.compile(
    checkpointer=None,   # Inherit from parent
    checkpointer=True,   # Independent checkpointing
    checkpointer=False,  # No checkpointing (runs atomically)
)

何时禁用检查点

• 应保持原子性的短期子图
• 状态架构不兼容的子图
• 无需恢复且对性能要求较高的路径

多 Agent 架构

监督者模式

适用于：

• 层级关系明确
• 集中决策
• Agent 专业领域各不相同

          ┌─────────────┐
          │  Supervisor │
          └──────┬──────┘
    ┌────────┬───┴───┬────────┐
    ▼        ▼       ▼        ▼
┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐
│Agent1│ │Agent2│ │Agent3│ │Agent4│
└──────┘ └──────┘ └──────┘ └──────┘

点对点模式

适用于：

• 协作型 Agent
• 没有明确层级
• 灵活通信

┌──────┐     ┌──────┐
│Agent1│◄───►│Agent2│
└──┬───┘     └───┬──┘
   │             │
   ▼             ▼
┌──────┐     ┌──────┐
│Agent3│◄───►│Agent4│
└──────┘     └──────┘

交接模式

适用于：

• 按顺序进行专业化处理
• 阶段转换明确
• 每个阶段的能力各不相同

┌────────┐    ┌────────┐    ┌────────┐
│Research│───►│Planning│───►│Execute │
└────────┘    └────────┘    └────────┘

流式处理策略

流模式选择

模式	适用场景	数据
updates	UI 更新	仅节点输出
values	状态检查	每个步骤的完整状态
messages	聊天体验	LLM 令牌
custom	进度／日志	通过 StreamWriter 传递的数据
debug	调试	任务 + 检查点

子图流式传输

# Stream from subgraphs
async for chunk in graph.astream(
    input,
    stream_mode="updates",
    subgraphs=True  # Include subgraph events
):
    namespace, data = chunk  # namespace indicates depth

人机协同设计

中断位置

策略	适用场景
interrupt_before	执行操作前审批
interrupt_after	完成后审核
节点中的 interrupt()	动态、结合上下文的暂停

恢复模式

# Simple resume (same thread)
graph.invoke(None, config)

# Resume with value
graph.invoke(Command(resume="approved"), config)

# Resume specific interrupt
graph.invoke(Command(resume={interrupt_id: value}), config)

# Modify state and resume
graph.update_state(config, {"field": "new_value"})
graph.invoke(None, config)

门槛（按顺序）

在将 LangGraph 设计定稿之前，按顺序完成以下步骤。每一步都有客观的通过条件（产物或明确标注“无”），而不是仅凭自我保证“我们考虑过了”。

1. 备选方案 — 通过条件： 对于该工作负载，满足以下任一条件：(a) 评估了 适合考虑备选方案的场景 中至少一个条目，并用一句话说明拒绝理由；或 (b) 用例明确符合 需要使用 LangGraph 的场景，且不符合任何“考虑备选方案”条目。
2. 状态契约 — 通过条件： 每个状态字段都指定了归约器（或默认值／LastValue），并与架构定义一同记录；大型负载使用引用或由 Store 支持，而不是内联的二进制块（参见 状态大小注意事项）。
3. 检查点保存器 — 通过条件： 根据 检查点保存器选择 为目标环境选择保存器类型（例如，生产环境不使用 InMemorySaver，除非明确仅用于测试）。
4. 循环与不稳定节点 — 通过条件： 对于任何可能形成循环的图，都设置了 recursion_limit（或等效项）；对于外部调用，已设置每节点的 RetryPolicy，或记录了“无重试”的选择（参见 重试配置）。

错误处理策略

重试配置

# Per-node retry
RetryPolicy(
    initial_interval=0.5,
    backoff_factor=2.0,
    max_interval=60.0,
    max_attempts=3,
    retry_on=lambda e: isinstance(e, (APIError, TimeoutError))
)

# Multiple policies (first match wins)
builder.add_node("node", fn, retry_policy=[
    RetryPolicy(retry_on=RateLimitError, max_attempts=5),
    RetryPolicy(retry_on=Exception, max_attempts=2),
])

回退模式

def node_with_fallback(state):
    try:
        return primary_operation(state)
    except PrimaryError:
        return fallback_operation(state)

# Or use conditional edges for complex fallback routing
def route_on_error(state) -> Literal["retry", "fallback", "__end__"]:
    if state.get("error") and state["attempts"] < 3:
        return "retry"
    elif state.get("error"):
        return "fallback"
    return END

扩展注意事项

水平扩展

• 使用 PostgresSaver 共享状态
• 考虑使用 LangGraph Platform 管理基础设施
• 使用 stores 在检查点之外存储大型数据

性能优化

1. 尽量减小状态大小 — 对大型数据使用引用
2. 并行节点 — 尽可能分散执行
3. 缓存开销大的操作 — 使用 CachePolicy
4. 全面异步 — 使用 ainvoke、astream

资源限制

# Set recursion limit
config = {"recursion_limit": 50}
graph.invoke(input, config)

# Track remaining steps in state
class State(TypedDict):
    remaining_steps: RemainingSteps

def check_budget(state):
    if state["remaining_steps"] < 5:
        return "wrap_up"
    return "continue"

决策检查清单

完成 门槛（按顺序） 后，在实现之前：

1. LangGraph 是合适的工具吗？（相较于更简单的替代方案）
2. 是否已定义包含合适归约器的状态架构？
3. 是否已选择持久化策略？（开发环境与生产环境的检查点保存器）
4. 是否已确定流式传输需求？
5. 是否已定义人机协同节点？
6. 错误处理和重试策略是什么？
7. 多 Agent 协调模式是什么？（如适用）
8. 是否已配置资源限制？