langgraph

LangGraph 工作流模式

用于构建生产级 LangGraph 工作流的全面模式。LangGraph 1.x 是 LTS（长期支持版）——首个稳定的主要版本，为 Uber、LinkedIn 和 Klarna 的 Agent 提供支持。每个类别都有位于 rules/ 中的独立规则文件，按需加载。

LangGraph 1.2（于 2026-05-12 发布）——故障容错版本。 除特别说明外，以下内容均适用于 StateGraph.add_node(...)：

• 节点级超时 — timeout= 接受 float | timedelta | TimeoutPolicy。 TimeoutPolicy(run_timeout=, idle_timeout=, refresh_on="auto"|"heartbeat") 将硬性 墙钟时间上限与可由进度刷新计时的空闲上限分开。超时后，LangGraph 会引发 NodeTimeoutError（包含 kind="idle"|"run" 和 elapsed），丢弃该次尝试的写入，并 交由重试策略处理。此机制依赖协作式取消，因此不会中断阻塞 GIL 的节点。参见 rules/resilience-node-timeouts.md。
• 节点错误处理器 — error_handler= 注册一个恢复节点，在重试次数耗尽后运行。它通过 声明类型为 NodeError 的参数接收失败上下文（字段为 node、error），并返回 Command 来更新状态并重新路由。参见 rules/resilience-error-handlers.md。
• RunControl（langgraph.runtime）— 协作式优雅关闭。可从任意线程调用 request_drain(reason)； 节点轮询 runtime.drain_requested，并在检查点边界停止，留下可恢复的线程，而不是留下 只应用了一半的超步。参见 rules/resilience-graceful-drain.md。
• DeltaChannel（langgraph.channels.delta，beta）— 检查点只存储增量写入，并通过批量 reducer 重放，每隔 snapshot_frequency 次更新创建一个快照。可避免检查点成本随线程长度增长。 其 reducer 接受一个批次，且必须满足批处理不变性。参见 rules/state-delta-channel.md。
• runtime.heartbeat() — 显式进度信号；在 refresh_on="heartbeat" 时，只有它会刷新空闲超时。

更早发布的功能，属于 1.1，而非 1.2（这些功能目前仍受支持；此前版本的此技能仅错误标注了 它们所属的版本）：延迟节点（defer=True）、节点级缓存（CachePolicy + graph.compile(cache=...)），以及 create_agent 上的模型中间件（before_model / after_model）。

快速参考

类别	规则	影响	使用场景
状态管理	5	严重	设计工作流状态架构、累加器、reducer、增量通道
弹性	3	严重	节点超时、错误处理器、优雅排空（1.2+）
路由与分支	4	高	动态路由、重试循环、语义路由、跨图路由
并行执行	3	高	扇出/扇入、Map-Reduce、并发 Agent
主管模式	3	高	中央协调器、轮询、优先级调度
工具调用	4	严重	绑定工具、ToolNode、动态选择、审批
检查点	3	高	持久化、恢复、跨线程 Store 记忆
人机协作	3	中	审批关卡、反馈循环、中断/恢复
流式传输	3	中	实时更新、令牌流、自定义事件
子图	3	中	模块化组合、嵌套图、状态映射
函数式 API	3	中	@entrypoint/@task 装饰器、从 StateGraph 迁移
平台	3	高	部署、RemoteGraph、双重文本输入策略

总计：12 个类别中的 41 条规则

状态管理

状态模式决定数据如何在节点之间流动。错误的模式会导致数据悄无声息地丢失。

规则	文件	关键模式
TypedDict 状态	rules/state-typeddict.md	累加器使用 TypedDict + Annotated[list, add]
Pydantic 验证	rules/state-pydantic.md	在边界处使用 BaseModel，内部使用 TypedDict
消息状态	rules/state-messages.md	MessagesState 或 add_messages reducer
自定义 Reducer	rules/state-reducers.md	使用 Annotated[T, reducer_fn] 进行合并/覆盖
Delta 通道（1.2，beta）	rules/state-delta-channel.md	对大型累加器使用 DeltaChannel(reducer, snapshot_frequency=)

弹性

为与外部世界交互的节点提供容错能力。1.2 中新增——在此之前，唯一可用的手段是 retry_policy，但它无法帮助一个从不失败的节点，因为它始终不会返回。

规则	文件	关键模式
节点超时	rules/resilience-node-timeouts.md	add_node(..., timeout=TimeoutPolicy(run_timeout=, idle_timeout=))
错误处理器	rules/resilience-error-handlers.md	add_node(..., error_handler=) + 参数类型为 NodeError → Command
优雅排空	rules/resilience-graceful-drain.md	RunControl().request_drain() + runtime.drain_requested

from langgraph.types import RetryPolicy, TimeoutPolicy
from langgraph.errors import NodeError

builder.add_node(
    "call_vendor",
    call_vendor,
    timeout=TimeoutPolicy(run_timeout=300, idle_timeout=30),
    retry_policy=RetryPolicy(max_attempts=3),
    error_handler=lambda state, error: Command(
        update={"failure": f"{error.node}: {error.error}"}, goto="degraded_path"
    ),
)

路由与分支

节点之间的控制流。始终包含 END 回退，以防止流程挂起。

规则	文件	关键模式
条件边	rules/routing-conditional.md	使用显式映射的 add_conditional_edges
重试循环	rules/routing-retry-loops.md	带最大重试计数器的回边
语义路由	rules/routing-semantic.md	基于嵌入的相似度或 Command API 路由
跨图导航	rules/routing-cross-graph.md	使用 Command(graph=Command.PARENT) 进行父级/同级路由

并行执行

并发运行相互独立的节点。使用 Annotated[list, add] 累积结果。

规则	文件	关键模式
扇出/扇入	rules/parallel-fanout-fanin.md	使用 Send API 实现动态并行分支
Map-Reduce	rules/parallel-map-reduce.md	asyncio.gather + 结果聚合
错误隔离	rules/parallel-error-isolation.md	return_exceptions=True + 各分支超时

Supervisor 模式

由中央协调器将任务路由给专门的工作节点。工作节点完成后返回 Supervisor。

规则	文件	关键模式
基本 Supervisor	rules/supervisor-basic.md	使用 Command API 更新状态并进行路由
优先级路由	rules/supervisor-priority.md	使用优先级字典排序 Agent 执行顺序
轮询	rules/supervisor-round-robin.md	使用 agents_completed 跟踪完成情况

工具调用

将函数调用集成到 LangGraph Agent 中。每个 Agent 使用的工具不要超过 10 个。

规则	文件	关键模式
工具绑定	rules/tools-bind.md	model.bind_tools(tools) + tool_choice
ToolNode 执行	rules/tools-toolnode.md	ToolNode(tools) 预构建并行执行器
动态选择	rules/tools-dynamic.md	基于嵌入的工具相关性筛选
工具中断	rules/tools-interrupts.md	使用 interrupt() 为工具设置审批关卡

检查点

持久化工作流状态，以便恢复和调试。

规则	文件	关键模式
Checkpointer 设置	rules/checkpoints-setup.md	开发环境使用 MemorySaver / 生产环境使用 PostgresSaver
状态恢复	rules/checkpoints-recovery.md	使用 thread_id 恢复 + get_state_history
跨线程存储	rules/checkpoints-store.md	使用 Store 在多个线程间保存长期记忆

节点级缓存（1.2+）

独立于检查点机制。缓存单个节点的输出，使使用相同输入重新运行时可完全跳过执行。

from langgraph.graph import StateGraph
from langgraph.types import CachePolicy
from langgraph.cache.sqlite import SqliteCache

graph = StateGraph(State)
graph.add_node(
    "expensive_fetch",
    fetch_fn,
    cache_policy=CachePolicy(ttl=3600, key_func=lambda s: s["query"]),
)
# RedisCache(url=...) for distributed workers
compiled = graph.compile(cache=SqliteCache("cache.db"))

适用于幂等且开销较大的节点（嵌入、外部 API）。除非 key_func 能捕获这些变化，否则不要用于输出依赖于挂钟时间或可变外部状态的节点。

延迟节点与模型中间件（1.2+）

# defer=True — node execution is deferred until the run is about to end,
# i.e. after every other upstream node has completed
graph.add_node("aggregate", aggregate_fn, defer=True)

# Model middleware — no subclassing required.
# create_react_agent is @deprecated since v1.0; use create_agent from langchain.agents.
# The legacy pre_model_hook/post_model_hook are now before_model/after_model middleware.
from langchain.agents import create_agent

agent = create_agent(
    model=model,
    tools=tools,
    middleware=[compress_history, redact_pii],  # before_model / after_model hooks
    system_prompt="...",                          # prompt= renamed to system_prompt
)

人机协作

暂停工作流以进行人工干预。需要检查点机制来持久化状态。

规则	文件	关键模式
中断/恢复	rules/human-in-loop-interrupt.md	interrupt() 函数 + Command(resume=)
审批关卡	rules/human-in-loop-approval.md	interrupt_before + 状态更新 + 恢复
反馈循环	rules/human-in-loop-feedback.md	持续迭代中断，直到获批

流式处理

为工作流提供实时更新和进度跟踪。LangGraph 1.2 支持 version="v2"（在 1.1 中引入），这是一种可选的流式格式，可为 stream()、astream()、invoke() 和 ainvoke() 提供完整的类型安全。

规则	文件	关键模式
流模式	rules/streaming-modes.md	5 种模式：values、updates、messages、custom、debug
Token 流式传输	rules/streaming-tokens.md	使用 messages 模式，并按节点/标签进行筛选
自定义事件	rules/streaming-custom-events.md	使用 get_stream_writer() 发送进度事件
流式格式 v2	rules/streaming-v2-format.md	使用 version="v2" 实现类型化流式处理（LG 1.1+）

子图

通过嵌套图组合模块化、可复用的工作流组件。

规则	文件	关键模式
从节点调用	rules/subgraphs-invoke.md	不同的 schema，显式状态映射
添加为节点	rules/subgraphs-add-as-node.md	共享状态，add_node(name, compiled_graph)
状态映射	rules/subgraphs-state-mapping.md	父级/子级之间的边界转换

函数式 API

使用 @entrypoint 和 @task 装饰器构建工作流，而非显式构建图。

规则	文件	关键模式
@entrypoint	rules/functional-entrypoint.md	工作流入口点，可选检查点机制
@task	rules/functional-task.md	返回 future，使用 .result() 阻塞
迁移	rules/functional-migration.md	将 StateGraph 转换为函数式 API

平台

将图部署为托管 API，支持持久化、流式处理和多租户。

规则	文件	关键模式
部署	rules/platform-deployment.md	langgraph.json + CLI + Assistants API
RemoteGraph	rules/platform-remote-graph.md	使用 RemoteGraph 调用已部署的图
重复消息	rules/platform-double-texting.md	4 种策略：拒绝、回滚、排队、中断

快速入门示例

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import TypedDict, Annotated, Literal
from operator import add

class State(TypedDict):
    input: str
    results: Annotated[list[str], add]

def supervisor(state) -> Command[Literal["worker", END]]:
    if not state.get("results"):
        return Command(update={"input": state["input"]}, goto="worker")
    return Command(goto=END)

def worker(state) -> dict:
    return {"results": [f"Processed: {state['input']}"]}

graph = StateGraph(State)
graph.add_node("supervisor", supervisor)
graph.add_node("worker", worker)
graph.add_edge(START, "supervisor")
graph.add_edge("worker", "supervisor")
app = graph.compile()

2026 年关键模式

• 流式处理 v2（LG 1.1）：使用 version="v2" 实现类型安全的流式处理——stream() 和 astream() 的返回值均有完整类型定义。默认仍为 "v1"，以保持向后兼容。
• Command API：需要同时更新状态和进行路由时，使用 Command(update=..., goto=...)
• context_schema：传递运行时配置（temperature、provider），避免污染状态
• CachePolicy：使用 SqliteCache（生产环境）或 langgraph.cache.memory 中的 InMemoryCache（开发环境），通过 TTL 缓存开销较大的节点结果
• RemainingSteps：主动处理递归限制
• Store：与检查点机制分离的跨线程记忆（限定于线程）
• interrupt()：在节点逻辑中使用动态中断（在条件场景下替代 interrupt_before）
• add_edge(START, node)：不要使用已弃用的 set_entry_point()
• LTS 版本：LangGraph 1.x 是 LTS 版本——在 v2.0 发布前将保持 ACTIVE

关键决策

决策	建议
状态类型	内部使用 TypedDict，边界处使用 Pydantic
入口点	使用 add_edge(START, node)，而不是 set_entry_point()
路由 + 状态更新	Command API
仅路由	条件边
累加器	始终使用 Annotated[list[T], add]
开发环境检查点保存器	MemorySaver
生产环境检查点保存器	PostgresSaver
短期记忆	检查点保存器（按线程隔离）
长期记忆	Store（跨线程，使用命名空间）
最大并行分支数	并发 5-10 个
每个 Agent 的工具数	最多 5-10 个（更多时动态选择）
审批关卡	对高风险操作使用 interrupt()
流模式	大多数 UI 使用 ["updates", "custom"]
子图模式	使用 Invoke 实现隔离，使用 Add-as-Node 共享状态
Functional 与 Graph	简单流程使用 Functional，复杂拓扑使用 Graph

常见错误

1. 忘记添加 add reducer（会覆盖而不是累加）
2. 原地修改状态（会破坏检查点保存）
3. 路由中没有 END 后备路径（工作流会挂起）
4. 无限重试循环（没有最大计数器）
5. 在路由器函数中产生副作用
6. 每个 Agent 配置过多工具（上下文溢出）
7. 在工具中抛出异常（会导致 Agent 循环崩溃）
8. 生产环境未配置检查点保存器（崩溃时会丢失进度）
9. 将 interrupt() 包装在 try/except 中（会破坏其机制）
10. 未在子图边界转换状态
11. 忘记在 Functional API 任务上调用 .result()
12. 使用 set_entry_point()（已弃用，应使用 add_edge(START, ...)）

评估

请参阅 test-cases.json，查看所有类别汇总的测试用例。

相关技能

• ork:agent-orchestration - 更高层次的多 Agent 协调、ReAct 循环模式和框架对比
• temporal-io - 持久化执行的替代方案
• ork:llm-integration - 通用 LLM 函数调用
• type-safety-validation - Pydantic 模型模式