Back to Details

langgraph

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

LangGraph Workflow Patterns

LangGraph工作流模式

Comprehensive patterns for building production LangGraph workflows. Each category has individual rule files in 
rules/
loaded on-demand.
用于构建生产级LangGraph工作流的全面模式集合。每个分类都有独立的规则文件存储在
rules/
目录中，可按需加载。

Quick Reference

快速参考

CategoryRulesImpactWhen to Use
State Management4CRITICALDesigning workflow state schemas, accumulators, reducers
Routing & Branching3HIGHDynamic routing, retry loops, semantic routing
Parallel Execution3HIGHFan-out/fan-in, map-reduce, concurrent agents
Supervisor Patterns3HIGHCentral coordinators, round-robin, priority dispatch
Tool Calling4CRITICALBinding tools, ToolNode, dynamic selection, approvals
Checkpointing3HIGHPersistence, recovery, cross-thread Store memory
Human-in-Loop3MEDIUMApproval gates, feedback loops, interrupt/resume
Streaming3MEDIUMReal-time updates, token streaming, custom events
Subgraphs3MEDIUMModular composition, nested graphs, state mapping
Functional API3MEDIUM@entrypoint/@task decorators, migration from StateGraph
Total: 32 rules across 10 categories
分类规则数量影响级别适用场景
状态管理4关键设计工作流状态 schema、累加器、归约器
路由与分支3高动态路由、重试循环、语义路由
并行执行3高扇出/扇入、Map-Reduce、并发Agent
监督者模式3高中央协调器、轮询调度、优先级分发
工具调用4关键工具绑定、ToolNode、动态选择、审批机制
检查点3高持久化、恢复、跨线程Store内存
人机协同3中审批网关、反馈循环、中断/恢复
流式传输3中实时更新、Token流式传输、自定义事件
子图3中模块化组合、嵌套图、状态映射
函数式API3中@entrypoint/@task装饰器、从StateGraph迁移
总计：10个分类共32条规则

State Management

状态管理

State schemas determine how data flows between nodes. Wrong schemas cause silent data loss.
RuleFileKey Pattern
TypedDict State
rules/state-typeddict.md
TypedDict
+ 
Annotated[list, add]
for accumulators
Pydantic Validation
rules/state-pydantic.md
BaseModel
at boundaries, TypedDict internally
MessagesState
rules/state-messages.md
MessagesState
or 
add_messages
reducer
Custom Reducers
rules/state-reducers.md
Annotated[T, reducer_fn]
for merge/overwrite
状态 schema 决定了数据在节点间的流动方式。错误的 schema 会导致无提示的数据丢失。
规则文件核心模式
TypedDict 状态
rules/state-typeddict.md
TypedDict
+ 
Annotated[list, add]
实现累加器
Pydantic 验证
rules/state-pydantic.md
边界使用
BaseModel
，内部使用TypedDict
MessagesState
rules/state-messages.md
MessagesState
或 
add_messages
归约器
自定义归约器
rules/state-reducers.md
Annotated[T, reducer_fn]
实现合并/覆盖

Routing & Branching

路由与分支

Control flow between nodes. Always include END fallback to prevent hangs.
RuleFileKey Pattern
Conditional Edges
rules/routing-conditional.md
add_conditional_edges
with explicit mapping
Retry Loops
rules/routing-retry-loops.md
Loop-back edges with max retry counter
Semantic Routing
rules/routing-semantic.md
Embedding similarity or 
Command
API routing
控制节点间的数据流。务必包含END分支回退，避免工作流挂起。
规则文件核心模式
条件边
rules/routing-conditional.md
使用
add_conditional_edges
并配置显式映射
重试循环
rules/routing-retry-loops.md
带最大重试计数器的循环边
语义路由
rules/routing-semantic.md
基于嵌入相似度或
Command
API的路由

Parallel Execution

并行执行

Run independent nodes concurrently. Use 
Annotated[list, add]
to accumulate results.
RuleFileKey Pattern
Fan-Out/Fan-In
rules/parallel-fanout-fanin.md
Send
API for dynamic parallel branches
Map-Reduce
rules/parallel-map-reduce.md
asyncio.gather
+ result aggregation
Error Isolation
rules/parallel-error-isolation.md
return_exceptions=True
+ per-branch timeout
并发运行独立节点。使用
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 Patterns

监督者模式

Central coordinator routes to specialized workers. Workers return to supervisor.
RuleFileKey Pattern
Basic Supervisor
rules/supervisor-basic.md
Command
API for state update + routing
Priority Routing
rules/supervisor-priority.md
Priority dict ordering agent execution
Round-Robin
rules/supervisor-round-robin.md
Completion tracking with 
agents_completed
中央协调器将任务路由至专用工作者节点，工作者执行后返回至协调器。
规则文件核心模式
基础监督者
rules/supervisor-basic.md
使用
Command
API实现状态更新+路由
优先级路由
rules/supervisor-priority.md
基于优先级字典排序Agent执行顺序
轮询调度
rules/supervisor-round-robin.md
用
agents_completed
跟踪执行完成情况

Tool Calling

工具调用

Integrate function calling into LangGraph agents. Keep tools under 10 per agent.
RuleFileKey Pattern
Tool Binding
rules/tools-bind.md
model.bind_tools(tools)
+ 
tool_choice
ToolNode Execution
rules/tools-toolnode.md
ToolNode(tools)
prebuilt parallel executor
Dynamic Selection
rules/tools-dynamic.md
Embedding-based tool relevance filtering
Tool Interrupts
rules/tools-interrupts.md
interrupt()
for approval gates on tools
将函数调用集成到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()
为工具添加审批网关

Checkpointing

检查点

Persist workflow state for recovery and debugging.
RuleFileKey Pattern
Checkpointer Setup
rules/checkpoints-setup.md
MemorySaver
dev / 
PostgresSaver
prod
State Recovery
rules/checkpoints-recovery.md
thread_id
resume + 
get_state_history
Cross-Thread Store
rules/checkpoints-store.md
Store
for long-term memory across threads
持久化工作流状态以支持恢复与调试。
规则文件核心模式
检查点设置
rules/checkpoints-setup.md
开发环境用
MemorySaver
/ 生产环境用
PostgresSaver
状态恢复
rules/checkpoints-recovery.md
thread_id
恢复 + 
get_state_history
跨线程存储
rules/checkpoints-store.md
使用
Store
实现跨线程的长期内存

Human-in-Loop

人机协同

Pause workflows for human intervention. Requires checkpointer for state persistence.
RuleFileKey Pattern
Interrupt/Resume
rules/human-in-loop-interrupt.md
interrupt()
function + 
Command(resume=)
Approval Gate
rules/human-in-loop-approval.md
interrupt_before
+ state update + resume
Feedback Loop
rules/human-in-loop-feedback.md
Iterative interrupt until approved
暂停工作流以等待人工干预。需要检查点来实现状态持久化。
规则文件核心模式
中断/恢复
rules/human-in-loop-interrupt.md
interrupt()
函数 + 
Command(resume=)
审批网关
rules/human-in-loop-approval.md
interrupt_before
+ 状态更新 + 恢复
反馈循环
rules/human-in-loop-feedback.md
迭代中断直至获得批准

Streaming

流式传输

Real-time updates and progress tracking for workflows.
RuleFileKey Pattern
Stream Modes
rules/streaming-modes.md
5 modes: values, updates, messages, custom, debug
Token Streaming
rules/streaming-tokens.md
messages
mode with node/tag filtering
Custom Events
rules/streaming-custom-events.md
get_stream_writer()
for progress events
为工作流提供实时更新与进度跟踪。
规则文件核心模式
流式模式
rules/streaming-modes.md
5种模式：values、updates、messages、custom、debug
Token 流式传输
rules/streaming-tokens.md
带节点/标签过滤的
messages
模式
自定义事件
rules/streaming-custom-events.md
使用
get_stream_writer()
实现进度事件

Subgraphs

子图

Compose modular, reusable workflow components with nested graphs.
RuleFileKey Pattern
Invoke from Node
rules/subgraphs-invoke.md
Different schemas, explicit state mapping
Add as Node
rules/subgraphs-add-as-node.md
Shared state, 
add_node(name, compiled_graph)
State Mapping
rules/subgraphs-state-mapping.md
Boundary transforms between parent/child
通过嵌套图组合模块化、可复用的工作流组件。
规则文件核心模式
从节点调用
rules/subgraphs-invoke.md
独立schema，显式状态映射
作为节点添加
rules/subgraphs-add-as-node.md
共享状态，
add_node(name, compiled_graph)
状态映射
rules/subgraphs-state-mapping.md
父图/子图间的边界转换

Functional API

函数式API

Build workflows using 
@entrypoint
and 
@task
decorators instead of explicit graph construction.
RuleFileKey Pattern
@entrypoint
rules/functional-entrypoint.md
Workflow entry point with optional checkpointer
@task
rules/functional-task.md
Returns futures, 
.result()
to block
Migration
rules/functional-migration.md
StateGraph
to Functional API conversion
使用
@entrypoint
和
@task
装饰器构建工作流，替代显式的图构造方式。
规则文件核心模式
@entrypoint
rules/functional-entrypoint.md
带可选检查点的工作流入口
@task
rules/functional-task.md
返回futures，使用
.result()
阻塞获取结果
迁移指南
rules/functional-migration.md
从
StateGraph
转换为函数式API

Quick Start Example

快速开始示例

python
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()
python
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 Key Patterns

2026核心模式

  • Command API: Use 
    Command(update=..., goto=...)
    when updating state AND routing together
  • context_schema: Pass runtime config (temperature, provider) without polluting state
  • CachePolicy: Cache expensive node results with TTL via 
    InMemoryCache
  • RemainingSteps: Proactively handle recursion limits
  • Store: Cross-thread memory separate from Checkpointer (thread-scoped)
  • interrupt(): Dynamic interrupts inside node logic (replaces 
    interrupt_before
    for conditional cases)
  • add_edge(START, node): Not 
    set_entry_point()
    (deprecated)
  • Command API: 同时更新状态和路由时使用
    Command(update=..., goto=...)
  • context_schema: 传递运行时配置（温度、提供商），避免污染状态
  • CachePolicy: 通过
    InMemoryCache
    为耗时节点结果添加TTL缓存
  • RemainingSteps: 主动处理递归限制
  • Store: 与检查点分离的跨线程内存（线程作用域）
  • interrupt(): 在节点逻辑内实现动态中断（替代条件场景下的
    interrupt_before
    ）
  • add_edge(START, node): 替代已弃用的
    set_entry_point()

Key Decisions

关键决策指南

DecisionRecommendation
State typeTypedDict internally, Pydantic at boundaries
Entry point
add_edge(START, node)
not 
set_entry_point()
Routing + state updateCommand API
Routing onlyConditional edges
Accumulators
Annotated[list[T], add]
always
Dev checkpointerMemorySaver
Prod checkpointerPostgresSaver
Short-term memoryCheckpointer (thread-scoped)
Long-term memoryStore (cross-thread, namespaced)
Max parallel branches5-10 concurrent
Tools per agent5-10 max (dynamic selection for more)
Approval gates
interrupt()
for high-risk operations
Stream modes
["updates", "custom"]
for most UIs
Subgraph patternInvoke for isolation, Add-as-Node for shared state
Functional vs GraphFunctional for simple flows, Graph for complex topology
决策项推荐方案
状态类型内部使用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"]
子图模式隔离场景用调用模式，共享状态场景用添加为节点模式
函数式vs图API简单流用函数式，复杂拓扑用图API

Common Mistakes

常见错误

  1. Forgetting 
    add
    reducer (overwrites instead of accumulates)
  2. Mutating state in place (breaks checkpointing)
  3. No END fallback in routing (workflow hangs)
  4. Infinite retry loops (no max counter)
  5. Side effects in router functions
  6. Too many tools per agent (context overflow)
  7. Raising exceptions in tools (crashes agent loop)
  8. No checkpointer in production (lose progress on crash)
  9. Wrapping 
    interrupt()
    in try/except (breaks the mechanism)
  10. Not transforming state at subgraph boundaries
  11. Forgetting 
    .result()
    on Functional API tasks
  12. Using 
    set_entry_point()
    (deprecated, use 
    add_edge(START, ...)
    )
  1. 遗漏
    add
    归约器（覆盖而非累加结果）
  2. 原地修改状态（破坏检查点机制）
  3. 路由未设置END回退（工作流挂起）
  4. 无限重试循环（未设置最大计数器）
  5. 路由函数中包含副作用
  6. 单Agent绑定过多工具（上下文溢出）
  7. 工具中抛出异常（导致Agent循环崩溃）
  8. 生产环境未配置检查点（崩溃时丢失进度）
  9. 用try/except包裹
    interrupt()
    （破坏中断机制）
  10. 子图边界未做状态转换
  11. 函数式API任务遗漏
    .result()
  12. 使用已弃用的
    set_entry_point()
    （应使用
    add_edge(START, ...)
    ）

Evaluations

测试验证

See 
test-cases.json
for consolidated test cases across all categories.
所有分类的综合测试用例请参考
test-cases.json
。

Related Skills

相关技能

  • agent-orchestration
    - Higher-level multi-agent coordination, ReAct loop patterns, and framework comparisons
  • temporal-io
    - Durable execution alternative
  • llm-integration
    - General LLM function calling
  • type-safety-validation
    - Pydantic model patterns
  • agent-orchestration
    - 更高阶的多Agent协调、ReAct循环模式及框架对比
  • temporal-io
    - 持久化执行替代方案
  • llm-integration
    - 通用LLM函数调用
  • type-safety-validation
    - Pydantic模型模式