LangGraph 1.x 工作流模式指南：状态管理、路由与并行执行最佳实践

langgraph by yonatangross/orchestkit

80 周安装量

134 GitHub Stars

GitHub

安装命令

npx skills add https://github.com/yonatangross/orchestkit --skill langgraph

AI/机器学习自动化系统架构

🇨🇳中文介绍

LangGraph 工作流模式

构建生产级 LangGraph 工作流的综合模式。LangGraph 1.x 是 LTS（长期支持版本）——首个稳定主版本，为 Uber、LinkedIn 和 Klarna 的智能体提供支持。每个类别在 rules/ 目录下都有按需加载的独立规则文件。

快速参考

类别	规则数	影响程度	使用场景
状态管理	4	关键	设计工作流状态模式、累加器、归约器
路由与分支	4	高	动态路由、重试循环、语义路由、跨图路由
并行执行	3	高	扇出/扇入、映射归约、并发智能体
监督者模式	3	高	中央协调器、轮询调度、优先级调度
工具调用	4	关键

广告位招租

在这里展示您的产品或服务

触达数万 AI 开发者，精准高效

联系我们

相关 Skills

FlyClaw：零登录航班聚合查询工具，Python实现多源航班信息与价格搜索

4,000,000 周安装

find-skills 技能搜索工具 - Vercel Labs 开源智能体技能包管理器

896,800 周安装

Azure RBAC 权限管理工具：查找最小角色、创建自定义角色与自动化分配

135,700 周安装

React 组合模式指南：Vercel 组件架构最佳实践，提升代码可维护性

120,000 周安装

规则	文件	关键模式
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]` 用于合并/覆盖

规则	文件	关键模式
条件边	`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)` 用于父/兄弟图路由

规则	文件	关键模式
扇出/扇入	`rules/parallel-fanout-fanin.md`	使用 `Send` API 实现动态并行分支
映射归约	`rules/parallel-map-reduce.md`	`asyncio.gather` + 结果聚合
错误隔离	`rules/parallel-error-isolation.md`	`return_exceptions=True` + 每个分支的超时设置

规则	文件	关键模式
基础监督者	`rules/supervisor-basic.md`	使用 `Command` API 进行状态更新和路由
优先级路由	`rules/supervisor-priority.md`	使用优先级字典排序智能体执行
轮询调度	`rules/supervisor-round-robin.md`	使用 `agents_completed` 跟踪完成情况

规则	文件	关键模式
工具绑定	`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()` 实现工具审批关卡

规则	文件	关键模式
检查点设置	`rules/checkpoints-setup.md`	开发环境用 `MemorySaver` / 生产环境用 `PostgresSaver`
状态恢复	`rules/checkpoints-recovery.md`	`thread_id` 恢复 + `get_state_history`
跨线程存储	`rules/checkpoints-store.md`	使用 `Store` 实现跨线程的长期记忆

规则	文件	关键模式
中断/恢复	`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.1 引入了 version="v2" —— 一种可选的流式格式，为 stream()、astream()、invoke() 和 ainvoke() 提供完整的类型安全。

规则	文件	关键模式
流模式	`rules/streaming-modes.md`	5 种模式：值、更新、消息、自定义、调试
令牌流式传输	`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`	不同的模式，显式的状态映射
添加为节点	`rules/subgraphs-add-as-node.md`	共享状态，`add_node(name, compiled_graph)`
状态映射	`rules/subgraphs-state-mapping.md`	父/子图之间的边界转换

规则	文件	关键模式
@entrypoint	`rules/functional-entrypoint.md`	工作流入口点，可选检查点机制
@task	`rules/functional-task.md`	返回 futures，使用 `.result()` 阻塞等待
迁移	`rules/functional-migration.md`	从 `StateGraph` 迁移到函数式 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()

流式处理 v2 (LG 1.1) : 使用 version="v2" 进行类型安全的流式处理 —— stream() 和 astream() 返回完全类型化的结果。默认仍为 "v1" 以保持向后兼容。
Command API : 当需要同时更新状态和路由时，使用 Command(update=..., goto=...)
context_schema : 传递运行时配置（如 temperature、provider），避免污染状态
CachePolicy : 通过 InMemoryCache 缓存开销大的节点结果，并设置 TTL
RemainingSteps : 主动处理递归限制
Store : 跨线程内存，独立于检查点（线程作用域）
interrupt() : 节点逻辑内部的动态中断（替代条件性场景中的 interrupt_before）
add_edge(START, node) : 而非 set_entry_point()（已弃用）
LTS 发布 : LangGraph 1.x 是 LTS 版本 —— 在 v2.0 发布前将保持活跃状态

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

忘记 add 归约器（导致覆盖而非累积）
原地修改状态（破坏检查点机制）
路由中没有 END 回退（工作流挂起）
无限重试循环（没有最大计数器）
在路由函数中产生副作用
每个智能体的工具过多（上下文溢出）
在工具中抛出异常（导致智能体循环崩溃）
生产环境没有检查点（崩溃时丢失进度）
将 interrupt() 包裹在 try/except 中（破坏中断机制）
未在子图边界转换状态
忘记在函数式 API 任务上调用 .result()
使用 set_entry_point()（已弃用，请使用 add_edge(START, ...)）

ork:agent-orchestration - 更高级别的多智能体协调、ReAct 循环模式及框架比较
temporal-io - 持久化执行的替代方案
ork:llm-integration - 通用 LLM 函数调用
type-safety-validation - Pydantic 模型模式

🇺🇸English

LangGraph Workflow Patterns

Comprehensive patterns for building production LangGraph workflows. LangGraph 1.x is LTS (Long Term Support) — the first stable major release, powering agents at Uber, LinkedIn, and Klarna. Each category has individual rule files in rules/ loaded on-demand.

Quick Reference

Category	Rules	Impact	When to Use
State Management	4	CRITICAL	Designing workflow state schemas, accumulators, reducers
Routing & Branching	4	HIGH	Dynamic routing, retry loops, semantic routing, cross-graph
Parallel Execution	3	HIGH	Fan-out/fan-in, map-reduce, concurrent agents
Supervisor Patterns	3	HIGH	Central coordinators, round-robin, priority dispatch
Tool Calling	4	CRITICAL	Binding tools, ToolNode, dynamic selection, approvals
Checkpointing	3	HIGH	Persistence, recovery, cross-thread Store memory
Human-in-Loop	3	MEDIUM	Approval gates, feedback loops, interrupt/resume
Streaming	3	MEDIUM	Real-time updates, token streaming, custom events
Subgraphs	3	MEDIUM	Modular composition, nested graphs, state mapping
Functional API	3	MEDIUM	@entrypoint/@task decorators, migration from StateGraph
Platform	3	HIGH	Deployment, RemoteGraph, double-texting strategies

Total: 37 rules across 11 categories

State Management

State schemas determine how data flows between nodes. Wrong schemas cause silent data loss.

Rule	File	Key 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 reducer

Routing & Branching

Control flow between nodes. Always include END fallback to prevent hangs.

Rule	File	Key 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
Cross-Graph Navigation	`rules/routing-cross-graph.md`

Parallel Execution

Run independent nodes concurrently. Use Annotated[list, add] to accumulate results.

Rule	File	Key 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

Supervisor Patterns

Central coordinator routes to specialized workers. Workers return to supervisor.

Rule	File	Key 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`

Tool Calling

Integrate function calling into LangGraph agents. Keep tools under 10 per agent.

Rule	File	Key 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

Checkpointing

Persist workflow state for recovery and debugging.

Rule	File	Key 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`

Human-in-Loop

Pause workflows for human intervention. Requires checkpointer for state persistence.

Rule	File	Key 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

Streaming

Real-time updates and progress tracking for workflows. LangGraph 1.1 introducesversion="v2" — an opt-in streaming format with full type safety on stream(), astream(), invoke(), and ainvoke().

Rule	File	Key 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
Streaming v2	`rules/streaming-v2-format.md`

Subgraphs

Compose modular, reusable workflow components with nested graphs.

Rule	File	Key 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

Functional API

Build workflows using @entrypoint and @task decorators instead of explicit graph construction.

Rule	File	Key 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

Platform

Deploy graphs as managed APIs with persistence, streaming, and multi-tenancy.

Rule	File	Key Pattern
Deployment	`rules/platform-deployment.md`	`langgraph.json` + CLI + Assistants API
RemoteGraph	`rules/platform-remote-graph.md`	`RemoteGraph` for calling deployed graphs
Double Texting	`rules/platform-double-texting.md`	4 strategies: reject, rollback, enqueue, interrupt

Quick Start Example

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

Streaming v2 (LG 1.1) : Use version="v2" for type-safe streaming — fully typed stream() and astream() returns. Default remains "v1" for backwards compat.
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)

Key Decisions

Decision	Recommendation
State type	TypedDict internally, Pydantic at boundaries
Entry point	`add_edge(START, node)` not `set_entry_point()`
Routing + state update	Command API
Routing only	Conditional edges
Accumulators	`Annotated[list[T], add]` always
Dev checkpointer	MemorySaver
Prod checkpointer	PostgresSaver
Short-term memory	Checkpointer (thread-scoped)
Long-term memory	Store (cross-thread, namespaced)
Max parallel branches

Common Mistakes

Forgetting add reducer (overwrites instead of accumulates)
Mutating state in place (breaks checkpointing)
No END fallback in routing (workflow hangs)
Infinite retry loops (no max counter)
Side effects in router functions
Too many tools per agent (context overflow)
Raising exceptions in tools (crashes agent loop)
No checkpointer in production (lose progress on crash)
Wrapping interrupt() in try/except (breaks the mechanism)
Not transforming state at subgraph boundaries
Forgetting .result() on Functional API tasks
Using set_entry_point() (deprecated, use add_edge(START, ...))

Evaluations

See test-cases.json for consolidated test cases across all categories.

Related Skills

ork:agent-orchestration - Higher-level multi-agent coordination, ReAct loop patterns, and framework comparisons
temporal-io - Durable execution alternative
ork:llm-integration - General LLM function calling
type-safety-validation - Pydantic model patterns

Weekly Installs

Repository

yonatangross/orchestkit

GitHub Stars

134

First Seen

Feb 14, 2026

Security Audits

Gen Agent Trust HubPass SocketPass SnykPass

Installed on

codex76

opencode75

gemini-cli74

github-copilot74

cursor69

kimi-cli67

LangGraph 1.x 工作流模式指南：状态管理、路由与并行执行最佳实践

🇨🇳中文介绍

LangGraph 工作流模式

快速参考

相关 Skills

状态管理

路由与分支

并行执行

监督者模式

工具调用

检查点

人在回路

流式处理

子图

函数式 API

平台

快速开始示例

2026 关键模式

关键决策

常见错误

评估

相关技能