langgraph-agents

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

<objective> Build production-grade multi-agent systems with LangGraph using supervisor, swarm, handoff, router, or master patterns. Enables cost-optimized orchestration with multi-provider routing (Claude, DeepSeek, Gemini - NO OpenAI), guardrails, durable execution, observability, and scalable agent coordination. </objective>

<quick_start> State schema (foundation):

python

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class AgentState(TypedDict, total=False):
    messages: Annotated[list, add_messages]  # Auto-merge
    next_agent: str  # For handoffs

Pattern selection:

Pattern	When	Agents
Supervisor	Clear hierarchy	3-10
Swarm	Peer collaboration	5-15
Handoff	Sequential pipeline	2-5
Router	Classify and dispatch	2-10
Master	Learning systems	10-30+

API choice: Graph API (explicit nodes/edges) vs Functional API (

@entrypoint

@task

decorators)

Key packages:

pip install langchain langgraph langgraph-supervisor langgraph-swarm langchain-mcp-adapters

</quick_start>

<success_criteria> Multi-agent system is successful when:

State uses
```
Annotated[..., add_messages]
```
for proper message merging
Termination conditions prevent infinite loops
Routing uses conditional edges (not hardcoded paths) OR Functional API tasks
Cost optimization: simple tasks → cheaper models (DeepSeek)
Complex reasoning → quality models (Claude)
NO OpenAI used anywhere
Checkpointers enabled for context preservation
Human-in-the-loop: interrupt() for approval workflows
Guardrails: PII detection, budget limits, call limits
MCP tools standardized via MultiServerMCPClient when appropriate
Observability: LangSmith tracing enabled in production </success_criteria>

<core_content> Production-tested patterns for building scalable, cost-optimized multi-agent systems with LangGraph and LangChain.

<objective> 使用LangGraph，通过监督者、集群、移交、路由或主模式构建生产级多智能体系统。支持多供应商路由（Claude、DeepSeek、Gemini - 不支持OpenAI）、防护机制、持久化执行、可观测性，以及可扩展的智能体协调，实现成本优化的编排。 </objective>

<quick_start> 状态架构（基础）：

python

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class AgentState(TypedDict, total=False):
    messages: Annotated[list, add_messages]  # Auto-merge
    next_agent: str  # For handoffs

模式选择：

模式	适用场景	智能体数量
监督者模式	存在明确层级结构	3-10
集群模式	对等协作场景	5-15
移交模式	顺序流水线场景	2-5
路由模式	分类与调度场景	2-10
主模式	学习型系统	10-30+

API选择： Graph API（显式节点/边） vs Functional API（

@entrypoint

@task

装饰器）

核心依赖包：

pip install langchain langgraph langgraph-supervisor langgraph-swarm langchain-mcp-adapters

</quick_start>

<success_criteria> 多智能体系统满足以下条件即为成功：

状态使用
```
Annotated[..., add_messages]
```
实现消息自动合并
包含终止条件以避免无限循环
路由使用条件边（而非硬编码路径）或Functional API任务
成本优化：简单任务→低成本模型（DeepSeek）
复杂推理→高质量模型（Claude）
全程未使用OpenAI
启用检查点以保留上下文
支持人机协作：通过interrupt()实现审批工作流
防护机制：PII检测、预算限制、调用次数限制
适当时通过MultiServerMCPClient标准化MCP工具
可观测性：生产环境中启用LangSmith追踪 </success_criteria>

<core_content> 基于LangGraph与LangChain构建可扩展、成本优化的生产级多智能体系统的经测试模式。

When to Use This Skill

适用场景

Symptoms:

"State not updating correctly between agents"
"Agents not coordinating properly"
"LLM costs spiraling out of control"
"Need to choose between supervisor vs swarm vs handoff patterns"
"Unclear how to structure agent state schemas"
"Agents losing context or repeating work"
"Need guardrails for PII, budget, or safety"
"How to test agent graphs"
"Need durable execution with crash recovery"
"Setting up LangSmith tracing / observability"
"Deploying LangGraph to production"

Use Cases:

Multi-agent systems with 3+ specialized agents
Complex workflows requiring orchestration
Cost-sensitive production deployments
Self-learning or adaptive agent systems
Enterprise applications with multiple LLM providers

常见痛点：

"智能体间状态更新异常"
"智能体协作不畅"
"大语言模型成本失控"
"不确定选择监督者/集群/移交哪种模式"
"智能体状态架构设计不清晰"
"智能体丢失上下文或重复工作"
"需要针对PII、预算或安全的防护机制"
"如何测试智能体图谱"
"需要带崩溃恢复的持久化执行"
"设置LangSmith追踪/可观测性"
"将LangGraph部署到生产环境"

典型用例：

包含3个以上专业智能体的多智能体系统
需要编排的复杂工作流
对成本敏感的生产部署
自学习或自适应智能体系统
集成多LLM供应商的企业应用

Quick Reference: Orchestration Pattern Selection

编排模式选择速查

Pattern	Use When	Complexity	Reference
Supervisor	Clear hierarchy, centralized routing	Low-Medium	`reference/orchestration-patterns.md`
Swarm	Peer collaboration, dynamic handoffs	Medium	`reference/orchestration-patterns.md`
Handoff	Sequential pipelines, escalation	Low	`reference/orchestration-patterns.md`
Router	Classify-and-dispatch, fan-out	Low	`reference/orchestration-patterns.md`
Skills	Progressive disclosure, on-demand	Low	`reference/orchestration-patterns.md`
Master	Learning systems, complex workflows	High	`reference/orchestration-patterns.md`

模式	适用场景	复杂度	参考文档
监督者模式	存在明确层级、集中式路由	低-中	`reference/orchestration-patterns.md`
集群模式	对等协作、动态移交	中	`reference/orchestration-patterns.md`
移交模式	顺序流水线、升级处理	低	`reference/orchestration-patterns.md`
路由模式	分类调度、扇出处理	低	`reference/orchestration-patterns.md`
技能模式	渐进式披露、按需调用	低	`reference/orchestration-patterns.md`
主模式	学习型系统、复杂工作流	高	`reference/orchestration-patterns.md`

Core Patterns

核心模式

1. State Schema (Foundation)

1. 状态架构（基础）

python

from typing import TypedDict, Annotated, Dict, Any
from langchain_core.messages import BaseMessage
from langgraph.graph import add_messages

class AgentState(TypedDict, total=False):
    messages: Annotated[list[BaseMessage], add_messages]  # Auto-merge
    agent_type: str
    metadata: Dict[str, Any]
    next_agent: str  # For handoffs

Deep dive:

reference/state-schemas.md

(reducers, annotations, multi-level state)

python

from typing import TypedDict, Annotated, Dict, Any
from langchain_core.messages import BaseMessage
from langgraph.graph import add_messages

class AgentState(TypedDict, total=False):
    messages: Annotated[list[BaseMessage], add_messages]  # Auto-merge
    agent_type: str
    metadata: Dict[str, Any]
    next_agent: str  # For handoffs

深度解析：

reference/state-schemas.md

（归约器、注解、多层级状态）

2. Multi-Provider Configuration (via lang-core)

2. 多供应商配置（基于lang-core）

python

undefined

python

undefined

Use lang-core for unified provider access (NO OPENAI)

使用lang-core实现统一供应商访问（不支持OPENAI）

from lang_core.providers import get_llm_for_task, LLMPriority

llm_cheap = get_llm_for_task(priority=LLMPriority.COST) # DeepSeek llm_smart = get_llm_for_task(priority=LLMPriority.QUALITY) # Claude llm_fast = get_llm_for_task(priority=LLMPriority.SPEED) # Cerebras llm_local = get_llm_for_task(priority=LLMPriority.LOCAL) # Ollama

**Deep dive:** `reference/base-agent-architecture.md`, `reference/cost-optimization.md`

from lang_core.providers import get_llm_for_task, LLMPriority

**深度解析：** `reference/base-agent-architecture.md`, `reference/cost-optimization.md`

3. Supervisor Pattern

3. 监督者模式

python

from langgraph_supervisor import create_supervisor  # pip install langgraph-supervisor
from langgraph.prebuilt import create_react_agent

research_agent = create_react_agent(model, tools=research_tools, prompt="Research specialist")
writer_agent = create_react_agent(model, tools=writer_tools, prompt="Content writer")

supervisor = create_supervisor(agents=[research_agent, writer_agent], model=model)
result = supervisor.invoke({"messages": [("user", "Write article about LangGraph")]})

python

from langgraph_supervisor import create_supervisor  # pip install langgraph-supervisor
from langgraph.prebuilt import create_react_agent

research_agent = create_react_agent(model, tools=research_tools, prompt="Research specialist")
writer_agent = create_react_agent(model, tools=writer_tools, prompt="Content writer")

supervisor = create_supervisor(agents=[research_agent, writer_agent], model=model)
result = supervisor.invoke({"messages": [("user", "Write article about LangGraph")]})

4. Swarm Pattern

4. 集群模式

python

from langgraph_swarm import create_swarm, create_handoff_tool  # pip install langgraph-swarm

handoff_to_bob = create_handoff_tool(agent_name="Bob", description="Transfer for Python tasks")
alice = create_react_agent(model, tools=[query_db, handoff_to_bob], prompt="SQL expert")
bob = create_react_agent(model, tools=[execute_code], prompt="Python expert")

swarm = create_swarm(agents=[alice, bob], default_active_agent="Alice")

python

from langgraph_swarm import create_swarm, create_handoff_tool  # pip install langgraph-swarm

handoff_to_bob = create_handoff_tool(agent_name="Bob", description="Transfer for Python tasks")
alice = create_react_agent(model, tools=[query_db, handoff_to_bob], prompt="SQL expert")
bob = create_react_agent(model, tools=[execute_code], prompt="Python expert")

swarm = create_swarm(agents=[alice, bob], default_active_agent="Alice")

5. Functional API (Alternative to Graph)

5. 函数式API（图谱API的替代方案）

python

from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

@task
def research(query: str) -> str:
    return f"Results for: {query}"

@entrypoint(checkpointer=InMemorySaver())
def workflow(query: str) -> dict:
    result = research(query).result()
    return {"output": result}

Deep dive:

reference/functional-api.md

(durable execution, time travel, testing)

python

from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

@task
def research(query: str) -> str:
    return f"Results for: {query}"

@entrypoint(checkpointer=InMemorySaver())
def workflow(query: str) -> dict:
    result = research(query).result()
    return {"output": result}

深度解析：

reference/functional-api.md

（持久化执行、时间回溯、测试）

6. MCP Tool Integration

6. MCP工具集成

python

from langchain_mcp_adapters.client import MultiServerMCPClient

async with MultiServerMCPClient(
    {"tools": {"transport": "stdio", "command": "python", "args": ["./mcp_server.py"]}}
) as client:
    tools = await client.get_tools()
    agent = create_react_agent(model, tools=tools)

Deep dive:

reference/mcp-integration.md

python

from langchain_mcp_adapters.client import MultiServerMCPClient

async with MultiServerMCPClient(
    {"tools": {"transport": "stdio", "command": "python", "args": ["./mcp_server.py"]}}
) as client:
    tools = await client.get_tools()
    agent = create_react_agent(model, tools=tools)

深度解析：

reference/mcp-integration.md

7. Deep Agents Framework (Production)

7. Deep Agents框架（生产级）

python

from deep_agents import create_deep_agent
from deep_agents.backends import CompositeBackend, StateBackend, StoreBackend

backend = CompositeBackend({
    "/workspace/": StateBackend(),      # Ephemeral
    "/memories/": StoreBackend()        # Persistent
})
agent = create_deep_agent(
    model=ChatAnthropic(model="claude-opus-4-6"),
    backend=backend,
    interrupt_on=["deploy", "delete"],
    skills_dirs=["./skills/"]
)

Deep dive:

reference/deep-agents.md

(subagents, skills, long-term memory)

python

from deep_agents import create_deep_agent
from deep_agents.backends import CompositeBackend, StateBackend, StoreBackend

backend = CompositeBackend({
    "/workspace/": StateBackend(),      # 临时存储
    "/memories/": StoreBackend()        # 持久化存储
})
agent = create_deep_agent(
    model=ChatAnthropic(model="claude-opus-4-6"),
    backend=backend,
    interrupt_on=["deploy", "delete"],
    skills_dirs=["./skills/"]
)

深度解析：

reference/deep-agents.md

（子智能体、技能、长期记忆）

8. Guardrails

8. 防护机制

python

undefined

python

undefined

Recursion limit prevents runaway agents (default: 25 steps)

递归限制防止智能体失控（默认：25步）

config = {"recursion_limit": 25, "configurable": {"thread_id": "user-123"}} result = graph.invoke(input_data, config=config)

Add guardrail nodes for PII, safety checks, HITL — see reference

添加防护节点处理PII、安全检查、人机协作 — 详见参考文档

**Deep dive:** `reference/guardrails.md` (input/output validation, tripwires, graph-node guardrails)

**深度解析：** `reference/guardrails.md`（输入/输出验证、触发机制、图谱节点防护）

Reference Files (14 Deep Dives)

参考文档（14篇深度解析）

Architecture:

reference/state-schemas.md
- TypedDict, Annotated reducers, multi-level state
reference/base-agent-architecture.md
- Multi-provider setup, agent templates
reference/tools-organization.md
- Modular tool design, InjectedState/InjectedStore

Orchestration:

reference/orchestration-patterns.md
- Supervisor, swarm, handoff, router, skills, master, HITL
reference/context-engineering.md
- Three context types, memory compaction, Anthropic best practices
reference/cost-optimization.md
- Provider routing, caching, token budgets, fallback chains

APIs:

reference/functional-api.md
- @entrypoint/@task, durable execution, time travel, testing
reference/mcp-integration.md
- MultiServerMCPClient, async context manager, tool composition
reference/deep-agents.md
- Harness, backends, subagents, skills, long-term memory
reference/streaming-patterns.md
- 5 streaming modes, v2 format, custom streaming

Production:

reference/guardrails.md
- PII detection, prompt injection, budget tripwires, output filtering
reference/testing-patterns.md
- Unit/integration testing, mocking, snapshot tests, CI/CD
reference/observability.md
- LangSmith tracing, custom metrics, evaluation, monitoring
reference/deployment-patterns.md
- App structure, local server, LangGraph Platform, Docker

架构类：

reference/state-schemas.md
- TypedDict、注解归约器、多层级状态
reference/base-agent-architecture.md
- 多供应商配置、智能体模板
reference/tools-organization.md
- 模块化工具设计、InjectedState/InjectedStore

编排类：

reference/orchestration-patterns.md
- 监督者、集群、移交、路由、技能、主模式、人机协作
reference/context-engineering.md
- 三种上下文类型、内存压缩、Anthropic最佳实践
reference/cost-optimization.md
- 供应商路由、缓存、令牌预算、降级链

API类：

reference/functional-api.md
- @entrypoint/@task、持久化执行、时间回溯、测试
reference/mcp-integration.md
- MultiServerMCPClient、异步上下文管理器、工具组合
reference/deep-agents.md
- Harness、后端、子智能体、技能、长期记忆
reference/streaming-patterns.md
- 5种流式模式、v2格式、自定义流式处理

生产类：

reference/guardrails.md
- PII检测、提示注入、预算触发、输出过滤
reference/testing-patterns.md
- 单元/集成测试、模拟、快照测试、CI/CD
reference/observability.md
- LangSmith追踪、自定义指标、评估、监控
reference/deployment-patterns.md
- 应用结构、本地服务器、LangGraph平台、Docker

Common Pitfalls

常见陷阱与解决方案

Issue	Solution
State not updating	Add `Annotated[..., add_messages]` reducer
Infinite loops	Add termination condition or set `recursion_limit` in config
High costs	Route simple tasks to cheaper models; use fallback chains
Context loss	Use checkpointers or memory systems
Wrong imports	`create_supervisor` from `langgraph_supervisor` , not `langgraph.prebuilt`
Wrong imports	`create_swarm` from `langgraph_swarm` , not `langgraph.prebuilt`
MCP API mismatch	Use `await client.get_tools()` , not `get_langchain_tools()`
PII leakage	Add PII redaction guard node (see `reference/guardrails.md` )
No observability	Set `LANGSMITH_TRACING=true` for zero-config tracing
Fragile agents	Add guardrails: call limits, budget tripwires, structured output

问题	解决方案
状态不更新	添加 `Annotated[..., add_messages]` 归约器
无限循环	添加终止条件或在配置中设置 `recursion_limit`
成本过高	将简单任务路由到低成本模型；使用降级链
上下文丢失	使用检查点或内存系统
导入错误	`create_supervisor` 来自 `langgraph_supervisor` ，而非 `langgraph.prebuilt`
导入错误	`create_swarm` 来自 `langgraph_swarm` ，而非 `langgraph.prebuilt`
MCP API不匹配	使用 `await client.get_tools()` ，而非 `get_langchain_tools()`
PII泄露	添加PII脱敏防护节点（详见 `reference/guardrails.md` ）
无观测能力	设置 `LANGSMITH_TRACING=true` 启用零配置追踪
智能体不稳定	添加防护机制：调用次数限制、预算触发、结构化输出

lang-core Integration

lang-core集成

For production deployments, use lang-core for:

Middleware: Cost tracking, budget enforcement, retry, caching, PII safety
LangSmith: Unified tracing with
```
@traced_agent
```
decorators
Providers: Auto-selection via
```
get_llm_for_task(priority=...)
```
Celery: Background agent execution with progress tracking
Redis: Distributed locks, rate limiting, event pub/sub

python

from lang_core import traced_agent, get_llm_for_task, LLMPriority
from lang_core.middleware import budget_enforcement_middleware

@traced_agent("QualificationAgent", tags=["sales"])
async def run_qualification(data):
    llm = get_llm_for_task(priority=LLMPriority.SPEED)
    # ... agent logic

生产部署时，使用lang-core实现：

中间件：成本追踪、预算控制、重试、缓存、PII安全
LangSmith：通过
```
@traced_agent
```
装饰器实现统一追踪
供应商：通过
```
get_llm_for_task(priority=...)
```
自动选择
Celery：带进度追踪的智能体后台执行
Redis：分布式锁、速率限制、事件发布/订阅

python

from lang_core import traced_agent, get_llm_for_task, LLMPriority
from lang_core.middleware import budget_enforcement_middleware

@traced_agent("QualificationAgent", tags=["sales"])
async def run_qualification(data):
    llm = get_llm_for_task(priority=LLMPriority.SPEED)
    # ... 智能体逻辑

Emit Outcome Sidecar

输出结果侧写

As the final step, write to

~/.claude/skill-analytics/last-outcome-langgraph-agents.json

json

{"ts":"[UTC ISO8601]","skill":"langgraph-agents","version":"2.0.0","variant":"default",
 "status":"[success|partial|error]","runtime_ms":[estimated ms from start],
 "metrics":{"agents_created":[n],"nodes_configured":[n],"graphs_built":[n]},
 "error":null,"session_id":"[YYYY-MM-DD]"}

Use status "partial" if some stages failed but results were produced. Use "error" only if no output was generated.

最后一步，写入

~/.claude/skill-analytics/last-outcome-langgraph-agents.json

：

json

{"ts":"[UTC ISO8601]","skill":"langgraph-agents","version":"2.0.0","variant":"default",
 "status":"[success|partial|error]","runtime_ms":[estimated ms from start],
 "metrics":{"agents_created":[n],"nodes_configured":[n],"graphs_built":[n]},
 "error":null,"session_id":"[YYYY-MM-DD]"}

若部分阶段失败但仍生成结果，状态设为"partial"；仅当未生成任何输出时设为"error"。