multi-agent-orchestration

多 Agent 编排

设计并编排复杂的多 Agent 系统，让专业 Agent 协作解决复杂问题，结合不同的专业知识和视角。

快速入门

通过示例和实用工具开始多 Agent 实现：

• 示例：查看 examples/ 目录中的完整实现：

  • orchestration_patterns.py - 顺序、并行、层级和共识编排
  • framework_implementations.py - CrewAI、AutoGen、LangGraph 和 Swarm 的模板

• 实用工具：查看 scripts/ 目录中的辅助模块：

  • agent_communication.py - 消息代理、共享内存和通信协议
  • workflow_management.py - 工作流执行、优化和监控
  • benchmarking.py - 团队性能和 Agent 效能指标

概述

多 Agent 系统将复杂问题拆解为专业的子任务，分配给具备相关专业知识的 Agent，然后协调其工作以实现统一目标。

多 Agent 系统的优势

• 复杂工作流：需要多个专业角色的任务
• 领域专业知识：金融、法律、人力资源和工程领域需要不同的知识
• 并行处理：多个 Agent 同时处理不同方面
• 协作推理：Agent 通过讨论、完善和改进方案
• 韧性：一个 Agent 出故障不会导致整个系统瘫痪
• 可扩展性：可轻松添加新的专业 Agent

架构概览

User Request
    ↓
Orchestrator
    ├→ Agent 1 (Specialist) → Task 1
    ├→ Agent 2 (Specialist) → Task 2
    ├→ Agent 3 (Specialist) → Task 3
    ↓
Result Aggregator
    ↓
Final Response

核心概念

Agent 定义

Agent 由以下要素定义：

• 角色：它承担什么职责？（例如，“财务分析师”）
• 目标：它应该完成什么？（例如，“分析财务风险”）
• 专业知识：它具备哪些知识／工具？
• 工具：它可以使用哪些能力？
• 上下文：它需要哪些信息才能有效工作？

编排模式

1. 顺序编排

• Agent 依次工作
• 每个 Agent 使用前一个 Agent 的输出
• 适用场景：步骤必须按顺序进行（研究 → 分析 → 写作）

2. 并行编排

• 多个 Agent 同时工作
• 最后汇总结果
• 适用场景：彼此独立的任务（分析竞争对手、市场和用户）

3. 层级编排

• 高级 Agent 将任务委派给初级 Agent
• 由管理者协调工作流程
• 适用场景：需要监督的大型项目

4. 基于共识的编排

• 多个 Agent 分析问题
• 讨论并完善想法
• 投票或达成共识
• 适用场景：需要多种视角的复杂决策

5. 工具中介编排

• Agent 使用共享工具／数据库
• 尽量减少直接通信
• 适用场景：大型系统、间接协调

多 Agent 团队示例

金融团队

Coordinator Agent
    ├→ Market Analyst Agent
    │   ├ Tools: Market data API, financial news
    │   └ Task: Analyze market conditions
    ├→ Financial Analyst Agent
    │   ├ Tools: Financial statements, ratio calculations
    │   └ Task: Analyze company financials
    ├→ Risk Manager Agent
    │   ├ Tools: Risk models, scenario analysis
    │   └ Task: Assess investment risks
    └→ Report Writer Agent
        ├ Tools: Document generation
        └ Task: Synthesize findings into report

法律团队

Case Manager Agent (Coordinator)
    ├→ Contract Analyzer Agent
    │   └ Task: Review contract terms
    ├→ Precedent Research Agent
    │   └ Task: Find relevant case law
    ├→ Risk Assessor Agent
    │   └ Task: Identify legal risks
    └→ Document Drafter Agent
        └ Task: Prepare legal documents

客户支持团队

Support Coordinator
    ├→ Issue Classifier Agent
    │   └ Task: Categorize customer issue
    ├→ Knowledge Base Agent
    │   └ Task: Find relevant documentation
    ├→ Escalation Agent
    │   └ Task: Determine if human escalation needed
    └→ Solution Synthesizer Agent
        └ Task: Prepare comprehensive response

实现框架

1. CrewAI

最适合：角色明确且具有层级结构的团队

from crewai import Agent, Task, Crew

# Define agents
analyst = Agent(
    role="Financial Analyst",
    goal="Analyze financial data and provide insights",
    backstory="Expert in financial markets with 10+ years experience"
)

researcher = Agent(
    role="Market Researcher",
    goal="Research market trends and competition",
    backstory="Data-driven researcher specializing in market analysis"
)

# Define tasks
analysis_task = Task(
    description="Analyze Q3 financial results for {company}",
    agent=analyst,
    tools=[financial_tool, data_tool]
)

research_task = Task(
    description="Research competitive landscape in {market}",
    agent=researcher,
    tools=[web_search_tool, industry_data_tool]
)

# Create crew and execute
crew = Crew(
    agents=[analyst, researcher],
    tasks=[analysis_task, research_task],
    process=Process.sequential
)

result = crew.kickoff(inputs={"company": "TechCorp", "market": "AI"})

2. AutoGen（Microsoft）

最适合：复杂的多轮对话和协商

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

# Define agents
analyst = AssistantAgent(
    name="analyst",
    system_message="You are a financial analyst..."
)

researcher = AssistantAgent(
    name="researcher",
    system_message="You are a market researcher..."
)

# Create group chat
groupchat = GroupChat(
    agents=[analyst, researcher],
    messages=[],
    max_round=10,
    speaker_selection_method="auto"
)

# Manage group conversation
manager = GroupChatManager(groupchat=groupchat)

# User proxy to initiate conversation
user = UserProxyAgent(name="user")

# Have conversation
user.initiate_chat(
    manager,
    message="Analyze if Company X should invest in Y market"
)

3. LangGraph

最适合：需要状态管理的复杂工作流

from langgraph.graph import Graph, StateGraph
from langgraph.prebuilt import create_agent_executor

# Define state
class AgentState:
    research_findings: str
    analysis: str
    recommendations: str

# Create graph
graph = StateGraph(AgentState)

# Add nodes for each agent
graph.add_node("researcher", research_agent)
graph.add_node("analyst", analyst_agent)
graph.add_node("writer", writer_agent)

# Define edges (workflow)
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")

# Set entry/exit points
graph.set_entry_point("researcher")
graph.set_finish_point("writer")

# Compile and run
workflow = graph.compile()
result = workflow.invoke({"topic": "AI trends"})

4. OpenAI Swarm

最适合：简单的 Agent 交接和对话式工作流

from swarm import Agent, Swarm

# Define agents
triage_agent = Agent(
    name="Triage Agent",
    instructions="Determine which specialist to route the customer to"
)

billing_agent = Agent(
    name="Billing Specialist",
    instructions="Handle billing and payment questions"
)

technical_agent = Agent(
    name="Technical Support",
    instructions="Handle technical issues"
)

# Define handoff functions
def route_to_billing(reason: str):
    return billing_agent

def route_to_technical(reason: str):
    return technical_agent

# Add tools to triage agent
triage_agent.functions = [route_to_billing, route_to_technical]

# Execute swarm
client = Swarm()
response = client.run(
    agent=triage_agent,
    messages=[{"role": "user", "content": "I have a billing question"}]
)

编排模式

模式 1：顺序任务链

Agent 按顺序执行任务，每个 Agent 都基于之前的结果继续工作：

# Task 1: Research
research_output = research_agent.work("Analyze AI market trends")

# Task 2: Analysis (uses research output)
analysis = analyst_agent.work(f"Analyze these findings: {research_output}")

# Task 3: Report (uses analysis)
report = writer_agent.work(f"Write report on: {analysis}")

适用场景：步骤之间存在依赖关系，且每一步都基于之前的结果

模式 2：并行执行

多个 Agent 同时工作，最后汇总结果：

import asyncio

async def parallel_teams():
    # All agents work in parallel
    market_task = market_agent.work_async("Analyze market")
    technical_task = tech_agent.work_async("Analyze technology")
    user_task = user_agent.work_async("Analyze user needs")

    # Wait for all to complete
    market_results, tech_results, user_results = await asyncio.gather(
        market_task, technical_task, user_task
    )

    # Synthesize results
    return synthesize(market_results, tech_results, user_results)

适用场景：需要快速获得独立分析结果，或希望结果更多元

模式 3：层级结构

Manager Agent 协调各专家：

manager_agent.orchestrate({
    "market_analysis": {
        "agents": [competitor_analyst, trend_analyst],
        "task": "Comprehensive market analysis"
    },
    "technical_evaluation": {
        "agents": [architecture_agent, security_agent],
        "task": "Technical feasibility assessment"
    },
    "synthesis": {
        "agents": [strategy_agent],
        "task": "Create strategic recommendations"
    }
})

适用场景：层级关系明确、团队各异、协调复杂

模式 4：辩论与共识

多个 Agent 讨论并达成共识：

agents = [bull_agent, bear_agent, neutral_agent]
question = "Should we invest in this startup?"

# Debate round 1
arguments = {agent: agent.argue(question) for agent in agents}

# Debate round 2 (respond to others)
counter_arguments = {
    agent: agent.respond(arguments) for agent in agents
}

# Reach consensus
consensus = mediator_agent.synthesize_consensus(counter_arguments)

适用场景：复杂决策、需要多种观点、风险评估

Agent 通信模式

1. 直接通信

Agent 彼此直接传递消息：

agent_a.send_message(agent_b, {
    "type": "request",
    "action": "analyze_document",
    "document": doc_content,
    "context": {"deadline": "urgent"}
})

2. 通过工具进行通信

Agent 使用共享工具或数据库：

# Agent A writes to shared memory
shared_memory.write("findings", {"market_size": "$5B", "growth": "20%"})

# Agent B reads from shared memory
findings = shared_memory.read("findings")

3. 基于 Manager 的通信

中央协调者管理 Agent 之间的通信：

manager.broadcast("update_all_agents", {
    "new_deadline": "tomorrow",
    "priority": "critical"
})

最佳实践

Agent 设计

• ✓ 角色和目标清晰、具体
• ✓ 为角色配备合适的工具
• ✓ 背景或专业知识相关
• ✓ 与其他 Agent 有明确区分
• ✓ 工作范围合理

工作流设计

• ✓ 任务依赖关系明确
• ✓ 确定交接点
• ✓ 处理 Agent 之间的错误
• ✓ 制定回退策略
• ✓ 监控性能

通信

• ✓ 消息格式规范
• ✓ 上下文共享清晰
• ✓ 错误传播策略明确
• ✓ 处理超时
• ✓ 审计日志记录

编排

• ✓ 明确定义流程（顺序、并行等）
• ✓ 设定明确的成功标准
• ✓ 监控 Agent 性能
• ✓ 实现反馈循环
• ✓ 留出人工干预点

常见挑战与解决方案

挑战：Agent 冲突

解决方案：

• 明确划分角色
• 明确决策规则
• 建立共识机制
• 设置冲突解决 Agent
• 明确权责层级

挑战：执行缓慢

解决方案：

• 尽可能采用并行执行
• 缓存开销较大的操作结果
• 预处理数据
• 优化 Agent 逻辑
• 实现超时处理

挑战：结果质量差

解决方案：

• 改进 Agent 提示词或指令
• 使用更相关的工具
• 整合反馈
• 设置质量验证 Agent
• 制定结果汇总策略

挑战：工作流复杂

解决方案：

• 拆分为更小的团队
• 采用层级结构
• 明确定义任务
• 做好状态管理
• 编写工作流文档

评估指标

团队表现：

• 任务完成率
• 结果质量
• 执行时间
• 成本（token/API 调用）
• 错误率

Agent 效能：

• 任务成功率
• 响应质量
• 工具使用效率
• 沟通清晰度
• 协作评分

高级技巧

1. 自组织团队

Agent 自主决定角色和工作流：

# Agents negotiate roles based on task
agents = [agent1, agent2, agent3]
task = "complex financial analysis"

# Agents determine best structure
negotiated_structure = self_organize(agents, task)
# Returns optimal workflow for this task

2. 自适应工作流

工作流根据进展情况进行调整：

# Monitor progress
if progress < expected_rate:
    # Increase resources
    workflow.add_agent(specialist_agent)
elif quality < threshold:
    # Increase validation
    workflow.insert_review_step()

3. 跨 Agent 学习

Agent 从彼此的工作中学习：

# After team execution
execution_trace = crew.get_execution_trace()

# Extract learnings
learnings = extract_patterns(execution_trace)

# Update agent knowledge
for agent, learning in learnings.items():
    agent.update_knowledge(learning)

资源

框架

• CrewAI：https://crewai.com/
• AutoGen：https://microsoft.github.io/autogen/
• LangGraph：https://langchain-ai.github.io/langgraph/
• Swarm：https://github.com/openai/swarm

论文

• “生成式 Agent”（Park 等人）
• “自组织多 Agent 系统”（研究论文）

实施清单

• 定义每个 Agent 的角色、目标和专业知识
• 确定每个 Agent 可用的工具或能力
• 规划工作流（顺序、并行、层级）
• 定义通信模式
• 实现任务定义
• 为每项任务设定成功标准
• 添加错误处理和回退方案
• 实现监控和日志记录
• 测试团队协作
• 评估质量和性能
• 根据结果进行优化
• 记录工作流和决策

入门指南

1. 从小规模开始：先从 2-3 个 Agent 开始
2. 明确工作流：记录 Agent 的交互方式
3. 充分测试：分别验证 Agent 的行为，并验证其协作情况
4. 密切监控：跟踪性能和结果
5. 持续迭代：根据结果不断改进
6. 逐步扩展：按需增加 Agent 和复杂度