langgraph-agent-patterns

LangGraph Agent 模式

为 LangGraph 应用实现并配置多 Agent 协作模式。

模式选择

根据协作需求选择合适的模式：

模式	最适用场景	使用时机
Supervisor	复杂工作流、动态路由	Agent 需要协作，路由取决于上下文
Router	简单分类、独立任务	一次性路由，决策确定
Orchestrator-Worker	并行执行、高吞吐量	子任务相互独立，且需要汇总结果
Handoffs	顺序工作流、保留上下文	顺序清晰，每个 Agent 都在前一个 Agent 的基础上继续处理

快速决策：

• 需要动态路由？ → Supervisor
• 任务可以并行运行？ → Orchestrator-Worker
• 简单分类？ → Router
• 线性顺序？ → Handoffs

详细对比：请参阅 references/pattern-comparison.md

模式实现指南

Supervisor-Subagent 模式

概述： 中央协调器根据上下文将任务委派给专门的子 Agent。

快速入门：

# Generate supervisor graph boilerplate
uv run scripts/generate_supervisor_graph.py my-team \
  --subagents "researcher,writer,reviewer"

# TypeScript
uv run scripts/generate_supervisor_graph.py my-team \
  --subagents "researcher,writer,reviewer" \
  --typescript

关键组件：

1. 带路由的状态：用于路由决策的 next 字段
2. Supervisor 节点：根据上下文做出路由决策
3. 子 Agent 节点：具备不同能力的专用 Agent
4. 条件边：从 Supervisor 路由到子 Agent

示例流程：

User Request → Supervisor → Researcher → Supervisor → Writer → Supervisor → FINISH

完整实现： 请参阅 references/supervisor-subagent.md

Router 模式

概述： 根据初始请求，一次性路由到专门的 Agent。

关键组件：

1. 带路由的状态：用于单次路由决策的字段
2. Router 节点：对请求进行分类（基于关键词、LLM 或语义）
3. 专用 Agent：针对每个类别的独立 Agent
4. 条件路由：路由到 Agent，然后结束

示例流程：

User Request → Router → Sales Agent → END
                   ├→ Support Agent → END
                   └→ Billing Agent → END

路由策略：

• 基于关键词：快速、简单的字符串匹配
• 基于 LLM：语义理解，灵活性高
• 基于嵌入：相似度匹配
• 基于模型：经过微调的分类器

完整实现： 请参阅 references/router-pattern.md

Orchestrator-Worker 模式

概述： 将任务拆分为并行子任务，并汇总结果。

关键组件：

1. 带子任务的状态：用于任务拆分和结果累积
2. Orchestrator 节点：将任务拆分为相互独立的子任务
3. Worker 节点：并行处理子任务
4. Aggregator 节点：综合结果
5. Send 扇出：从条件边返回 Send(...) 对象，并使用列表 reducer（例如 Annotated[list[dict], operator.add]）累积 Worker 输出

示例流程：

Task → Orchestrator → Worker 1 ┐
                  → Worker 2  ├→ Aggregator → Result
                  → Worker 3 ┘

最佳实践：

• 确保子任务相互独立
• 妥善处理 Worker 故障
• 限制并发 Worker 数量以管理资源
• 使用 LLM 综合结果

完整实现： 请参阅 references/orchestrator-worker.md

Handoffs 模式

概述： 按顺序移交 Agent，同时保留上下文。

关键组件：

1. 带上下文的状态：在移交过程中共享上下文
2. Agent 节点：每个 Agent 都会将任务移交给下一个 Agent
3. 移交逻辑：显式或有条件地移交
4. 上下文管理：保留并传递信息

示例流程：

Request → Researcher → Writer → Editor → FINISH
         (with context preservation)

移交策略：

• 显式：Agent 指定下一个 Agent
• 条件式：根据完成标准决定
• 循环式：Agent 可以将任务移交回去进行修订

完整实现： 请参阅 references/handoffs.md

示例

可运行的迷你项目（Python + JavaScript）：

• assets/examples/supervisor-example/
• assets/examples/router-example/
• assets/examples/orchestrator-example/
• assets/examples/handoff-example/

多 Agent 模式的状态设计

每种模式都需要特定的状态架构设计：

监督者模式：

class SupervisorState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next: Literal["agent1", "agent2", "FINISH"]
    current_agent: str

路由器模式：

class RouterState(TypedDict):
    messages: list[BaseMessage]
    route: Literal["category1", "category2"]

编排者-工作者模式：

import operator
class OrchestratorState(TypedDict):
    task: str
    subtasks: list[dict]
    results: Annotated[list[dict], operator.add]

交接模式：

class HandoffState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next_agent: str
    context: dict

详细的状态模式： 请参阅 references/state-management-patterns.md

验证和可视化

验证图结构

# Validate agent graph for issues
uv run scripts/validate_agent_graph.py path/to/graph.py:graph

# Checks for:
# - Unreachable nodes
# - Cycles without termination
# - Dead ends
# - Invalid routing

可视化图

# Generate Mermaid diagram
uv run scripts/visualize_graph.py path/to/graph.py:graph --output diagram.md

# View in browser or IDE with Mermaid support

常见模式和反模式

最佳实践

1. 明确 Agent 职责

• 定义互不重叠的能力
• 记录每个 Agent 的用途
• 避免 Agent 重复

2. 防止循环

• 在状态中跟踪迭代次数
• 设置最大迭代次数
• 实现循环检测

3. 上下文管理

• 上下文过大时进行摘要
• 只传递必要的信息
• 尽可能使用结构化上下文

4. 错误处理

• 验证路由决策
• 妥善处理无效路由
• 默认使用安全回退方案

应避免的反模式

1. 过度监督

# ❌ Bad: Supervisor for simple linear flow
User → Supervisor → Agent1 → Supervisor → Agent2 → Supervisor

# ✅ Good: Use handoffs instead
User → Agent1 → Agent2 → FINISH

2. 复杂的路由器逻辑

# ❌ Bad: Complex routing rules in router
if complex_condition_A and (condition_B or condition_C):
    route = determine_complex_route()

# ✅ Good: Use supervisor with LLM
route = llm.invoke("Analyze and route: {query}")

3. 未受管理的状态增长

# ❌ Bad: Accumulating all messages forever
messages: list[BaseMessage]  # Grows unbounded

# ✅ Good: Summarize or limit
if len(messages) > 20:
    messages = summarize_context(messages)

多 Agent 系统调试

1. 跟踪 Agent 流程

使用 LangSmith 可视化 Agent 交互：

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "<your-api-key>"
os.environ["LANGSMITH_PROJECT"] = "multi-agent-debug"

result = graph.invoke(input_state)

2. 记录路由决策

在路由节点中添加日志：

def supervisor_node(state: SupervisorState) -> dict:
    decision = make_routing_decision(state)

    print(f"Supervisor routing to: {decision}")
    print(f"Current state: {len(state['messages'])} messages")
    print(f"Iteration: {state.get('iteration', 0)}")

    return {"next": decision}

3. 验证图结构

# Detect common issues
uv run scripts/validate_agent_graph.py my_agent/graph.py:graph

# Check for:
# - Unreachable nodes
# - Infinite loops
# - Dead ends

4. 可视化流程

# Generate diagram
uv run scripts/visualize_graph.py my_agent/graph.py:graph -o flow.md

性能优化

延迟优化

监督者模式：

• 使用更快的模型进行路由（gpt-4o-mini）
• 缓存路由决策
• 实现提前终止

路由器模式：

• 简单情况使用关键词匹配
• 缓存相似查询的路由结果
• 尽可能避免调用 LLM

编排者-工作者模式：

• 真正的并行化已达到最优
• 限制工作者数量以避免触发速率限制
• 将结果流式传输给聚合器

交接模式：

• 尽量缩小上下文
• 跳过不必要的交接
• 适当使用成本更低的模型

成本优化

Token 用量：

• 定期摘要上下文
• 使用结构化输出以提高可靠性
• 简单任务使用成本更低的模型

LLM 调用：

• 缓存路由决策
• 尽可能使用确定性逻辑
• 批量处理相似请求

模式选择：

• 路由器 < 交接 < 编排者 < 监督者（成本）

多 Agent 模式测试

单元测试路由逻辑

def test_supervisor_routing():
    """Test supervisor routes correctly."""
    state = {
        "messages": [HumanMessage(content="Need research")],
        "next": "",
        "current_agent": ""
    }

    result = supervisor_node(state)
    assert result["next"] == "researcher"

集成测试

def test_full_workflow():
    """Test complete multi-agent workflow."""
    graph = create_supervisor_graph()

    result = graph.invoke({
        "messages": [HumanMessage(content="Write article about AI")]
    })

    # Verify agents were called in correct order
    assert "researcher" in result["agent_history"]
    assert "writer" in result["agent_history"]

测试图结构

# Validate before deployment
python3 scripts/validate_agent_graph.py graph.py:graph

模式之间的迁移

从路由器迁移到监督者

当路由逻辑变得复杂时：

# Before: Complex router
def route(query):
    if complex_rules(query):
        return category

# After: Supervisor with LLM
def supervisor(state):
    return llm_routing_decision(state)

从交接迁移到监督者

当需要动态路由时：

# Before: Fixed sequence
Agent1 → Agent2 → Agent3

# After: Dynamic routing
Supervisor ⇄ Agent1/Agent2/Agent3

从串行迁移到并行

当任务变得彼此独立时：

# Before: Sequential
Agent1 → Agent2 → Agent3

# After: Parallel
Orchestrator → [Agent1, Agent2, Agent3] → Aggregator

常见用例

客户支持系统

模式： 路由器 + 监督者

Router → Sales Supervisor → Sales Agents
     ↓
     Support Supervisor → Support Agents

研究与写作流水线

模式： 监督者或交接

Supervisor ⇄ Researcher
         ⇄ Writer
         ⇄ Editor

数据分析流水线

模式： 编排者-工作者

Orchestrator → Data Collectors → Aggregator

文档处理

模式： 编排者-工作者 + 监督者

Router → PDF Orchestrator → Workers → Aggregator
     ↓
     DOCX Orchestrator → Workers → Aggregator

脚本参考

generate_supervisor_graph.py

生成监督者-子 Agent 样板代码：

uv run scripts/generate_supervisor_graph.py <name> [options]

Options:
  --subagents AGENTS    Comma-separated list (default: researcher,writer,reviewer)
  --output DIR          Output directory (default: current directory)
  --typescript          Generate TypeScript instead of Python

validate_agent_graph.py

验证图结构：

uv run scripts/validate_agent_graph.py <module_path>

Format: path/to/module.py:graph_name

Checks:
  - Unreachable nodes
  - Cycles
  - Dead ends
  - Invalid routing

visualize_graph.py

生成 Mermaid 图：

uv run scripts/visualize_graph.py <module_path> [options]

Options:
  --output FILE         Output file (default: stdout)
  --diagram-only        Skip documentation, output diagram only

故障排除

“Agent 未能正确协调”

检查：

1. 状态架构是否支持你的模式（参见 state-management-patterns.md）
2. 路由逻辑是否经过正确验证
3. 上下文是否在各 Agent 之间得到保留

“检测到无限循环”

解决方案：

1. 在状态中添加迭代计数器
2. 设置最大迭代次数
3. 添加循环检测逻辑
4. 使用 validate_agent_graph.py 进行验证

“路由决策效果不佳”

解决方案：

1. 改进监督者提示词，清晰描述各 Agent
2. 使用结构化输出以提高可靠性
3. 在路由提示词中添加示例
4. 使用更好的模型进行路由决策

“高延迟”

解决方案：

1. 对于简单情况，考虑使用路由器模式
2. 使用速度更快的模型进行路由
3. 尽可能并行执行
4. 缓存路由决策

“Token 使用量高”

解决方案：

1. 定期总结上下文
2. 对简单任务使用成本更低的模型
3. 实现上下文窗口管理
4. 选择更高效的模式

其他资源

• 模式详情：

  • Supervisor：references/supervisor-subagent.md
  • Router：references/router-pattern.md
  • Orchestrator-Worker：references/orchestrator-worker.md
  • Handoffs：references/handoffs.md

• 状态管理： references/state-management-patterns.md

• 模式比较： references/pattern-comparison.md

• 可运行示例： assets/examples/

• LangGraph 文档：

  • 多 Agent 模式：https://docs.langchain.com/oss/python/langchain/multi-agent/subagents-personal-assistant; https://docs.langchain.com/oss/python/langchain/multi-agent/handoffs-customer-support; https://docs.langchain.com/oss/python/langchain/multi-agent/router-knowledge-base
  • 条件边：https://docs.langchain.com/oss/python/langgraph/graph-api
  • Map-Reduce：https://docs.langchain.com/oss/python/langgraph/graph-api#map-reduce-and-the-send-api