langgraph-agents

使用 LangGraph，通过主管、群体协作、交接、路由器或主控模式，构建可用于生产环境的多 Agent 系统。支持成本优化型编排、多提供商路由（Claude、DeepSeek、Gemini——不使用 OpenAI）、护栏、持久执行、可观测性和可扩展的 Agent 协调。

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

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

模式选择：

模式	适用场景	Agent 数量
主管	层级明确	3-10
群体协作	同级协作	5-15
交接	顺序管线	2-5
路由器	分类并分派	2-10
主控	学习系统	10-30+

**API 选择：**Graph API（显式定义节点/边）或 Functional API（@entrypoint/@task 装饰器）

关键软件包：pip install langchain langgraph langgraph-supervisor langgraph-swarm langchain-mcp-adapters </quick_start>

<success_criteria> 满足以下条件时，多 Agent 系统即为成功：

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

<core_content> 使用 LangGraph 和 LangChain 构建可扩展、成本优化的多 Agent 系统的生产验证模式。

何时使用此技能

症状：

• “Agent 之间的状态未能正确更新”
• “Agent 协调不当”
• “LLM 成本失控”
• “需要在主管、群体协作和交接模式之间做出选择”
• “不清楚如何设计 Agent 状态架构”
• “Agent 丢失上下文或重复工作”
• “需要针对 PII、预算或安全性的护栏”
• “如何测试 Agent 图”
• “需要具备崩溃恢复能力的持久执行”
• “设置 LangSmith 跟踪 / 可观测性”
• “将 LangGraph 部署到生产环境”

使用场景：

• 包含 3 个或更多专用 Agent 的多 Agent 系统
• 需要编排的复杂工作流
• 对成本敏感的生产部署
• 自学习或自适应 Agent 系统
• 使用多个 LLM 提供商的企业应用

快速参考：编排模式选择

模式	适用场景	复杂度	参考资料
主管	层级明确、集中式路由	低—中	reference/orchestration-patterns.md
群体协作	同级协作、动态交接	中	reference/orchestration-patterns.md
交接	顺序管线、升级处理	低	reference/orchestration-patterns.md
路由器	分类并分派、扇出	低	reference/orchestration-patterns.md
技能	渐进式披露、按需使用	低	reference/orchestration-patterns.md
主控	学习系统、复杂工作流	高	reference/orchestration-patterns.md

核心模式

1. 状态架构（基础）

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（reducer、注解、多级状态）

2. 多提供商配置（通过 lang-core）

# Use lang-core for unified provider access (NO 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

深入了解：reference/base-agent-architecture.md、reference/cost-optimization.md

3. 主管模式

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. 群体协作模式

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（Graph 的替代方案）

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 工具集成

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 框架（生产环境）

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/"]
)

深入了解：reference/deep-agents.md（子 Agent、技能、长期记忆）

</core_content>

8. 防护措施

# Recursion limit prevents runaway agents (default: 25 steps)
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

深入了解： reference/guardrails.md（输入／输出验证、触发保护机制、图节点防护措施）

参考文件（14 个深入主题）

架构：

• reference/state-schemas.md - TypedDict、Annotated reducers、多层状态
• reference/base-agent-architecture.md - 多提供商配置、Agent 模板
• reference/tools-organization.md - 模块化工具设计、InjectedState/InjectedStore

编排：

• reference/orchestration-patterns.md - Supervisor、swarm、移交、路由器、技能、主控、HITL
• 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、后端、子 Agent、技能、长期记忆
• 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 Platform、Docker

常见问题

问题	解决方案
状态未更新	添加 Annotated[..., add_messages] reducer
无限循环	添加终止条件，或在配置中设置 recursion_limit
成本高昂	将简单任务路由到更便宜的模型；使用回退链
上下文丢失	使用检查点保存器或记忆系统
导入错误	从 langgraph_supervisor 导入 create_supervisor，而非 langgraph.prebuilt
导入错误	从 langgraph_swarm 导入 create_swarm，而非 langgraph.prebuilt
MCP API 不匹配	使用 await client.get_tools()，而非 get_langchain_tools()
PII 泄露	添加 PII 脱敏防护节点（参见 reference/guardrails.md）
缺少可观测性	设置 LANGSMITH_TRACING=true，以实现零配置跟踪
Agent 不稳定	添加防护措施：调用次数限制、预算触发保护机制、结构化输出

lang-core 集成

对于生产部署，请使用 lang-core：

• Middleware：成本跟踪、预算执行、重试、缓存、PII 安全
• LangSmith：通过 @traced_agent 装饰器实现统一跟踪
• Providers：通过 get_llm_for_task(priority=...) 自动选择
• Celery：后台 Agent 执行与进度跟踪
• Redis：分布式锁、速率限制、事件发布／订阅

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

输出结果侧车文件

最后一步，将内容写入 ~/.claude/skill-analytics/last-outcome-langgraph-agents.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"。