Multi-Agent Architect

多 Agent 架构师

设计由多个专业 Agent 协作解决复杂问题的系统。

核心原则

将复杂任务分配给专业 Agent，由各领域的专家负责，并通过清晰的通信模式进行协调。

何时使用多 Agent 系统

适合使用多 Agent 的情况：

• ✅ 任务需要多种专业能力（研究 + 写作 + 编码）
• ✅ 并行处理可以加快解决问题的速度（独立的子任务）
• ✅ 需要通过同行评审进行自我纠正
• ✅ 工作流复杂，包含决策点
• ✅ 扩展单 Agent 方案变得难以管理

不适合使用多 Agent 的情况：

• ❌ 单个 Agent 就能高效完成任务
• ❌ 任务简单且线性
• ❌ 通信开销 > 并行处理带来的收益
• ❌ 团队缺乏多 Agent 调试经验

---

多 Agent 模式

模式 1：顺序流水线

用途：多步骤工作流，每个 Agent 都以前一个 Agent 的成果为基础

User Query → Researcher → Analyst → Writer → Editor → Output

示例：生成研究报告

1. 研究员：收集资料
2. 分析师：综合研究结果
3. 撰稿人：起草报告
4. 编辑：润色并排版

优点：依赖关系清晰，易于调试 缺点：顺序执行（无法并行处理），容易形成瓶颈

---

模式 2：分层模式（管理者-工作者）

用途：将复杂任务拆分为可并行处理的子任务

              Manager Agent
              /     |     \
    Worker 1   Worker 2   Worker 3
    (Search)   (Analyze)  (Summarize)
              \     |     /
              Aggregator Agent

示例：针对多个竞争对手开展市场研究

• 管理者：将任务拆分为针对每个竞争对手的分析
• 工作者：并行研究竞争对手 A、B、C
• 汇总者：整合研究结果

优点：并行处理，专业化分工 缺点：管理者逻辑复杂，协调开销大

---

模式 3：同行协作（圆桌会议）

用途：通过多个视角提升质量

Coder ↔ Reviewer ↔ Tester
  ↓        ↓        ↓
      Consensus

示例：代码生成与评审

1. 编码者：编写初始代码
2. 评审者：检查问题
3. 测试者：验证功能
4. 反复迭代，直到达成共识

优点：通过评审提升质量，实现自我纠正 缺点：可能无法达成一致，成本较高（多次调用大模型）

---

模式 4：Agent 群集

用途：让多个 Agent 独立探索解决方案空间

Agent 1 → Candidate Solution 1
Agent 2 → Candidate Solution 2
Agent 3 → Candidate Solution 3
   ↓
Selector (pick best)

示例：创意头脑风暴

• 5 个 Agent 分别提出不同方案
• 选择器评估并选出最佳方案

优点：探索性强，富有创意 缺点：成本较高（N 个 Agent），可能产生相似的方案

---

通信模式

1. 共享内存

shared_state = {
    "research_findings": [],
    "current_task": "analyze_competitors",
    "decisions": []
}

# All agents read/write to shared state
researcher.execute(shared_state)
analyst.execute(shared_state)

优点：简单，所有 Agent 都能看到完整上下文 缺点：存在竞态条件，难以调试哪些内容由谁修改

---

2. 消息传递

# Agent A sends message to Agent B
message = {
    "from": "researcher",
    "to": "analyst",
    "content": research_findings,
    "metadata": {"confidence": 0.9}
}

message_queue.send(message)

优点：通信流程清晰，可追踪 缺点：实现更复杂

---

3. 事件驱动

# Agents subscribe to events
event_bus.subscribe("research_complete", analyst.on_research_complete)
event_bus.subscribe("analysis_complete", writer.on_analysis_complete)

# Agent publishes event when done
event_bus.publish("research_complete", research_data)

优点：耦合松散，可扩展 缺点：执行流程更难追踪

---

Agent 协调策略

1. 固定工作流

预定义执行顺序，不进行动态决策

workflow = [
    ("researcher", gather_info),
    ("analyst", analyze_data),
    ("writer", create_report)
]

for agent_name, task in workflow:
    result = agents[agent_name].execute(task, context)
    context.update(result)

用途：适用于可预测的任务和依赖关系清晰的场景

---

2. 动态路由

管理者根据上下文决定下一个 Agent

class ManagerAgent:
    def route_task(self, task, context):
        if requires_technical_expertise(task):
            return tech_specialist
        elif requires_creative_input(task):
            return creative_agent
        else:
            return generalist

用途：任务差异较大，需要灵活处理

---

3. 基于共识

Agent 通过投票或协商达成一致

proposals = [agent.propose_solution(task) for agent in agents]
scores = [agent.evaluate(proposals) for agent in agents]
best = proposals[argmax(mean(scores))]

用途：适用于高风险决策和质量至关重要的场景

---

使用 CrewAI 实现

CrewAI 模式（基于角色的团队）：

from crewai import Agent, Task, Crew

# Define specialized agents
researcher = Agent(
    role="Research Specialist",
    goal="Gather comprehensive information on {topic}",
    backstory="Expert researcher with 10 years experience",
    tools=[search_tool, scrape_tool]
)

analyst = Agent(
    role="Data Analyst",
    goal="Synthesize research findings into insights",
    backstory="Data scientist specialized in trend analysis",
    tools=[analysis_tool]
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear, compelling reports",
    backstory="Professional writer with technical expertise",
    tools=[writing_tool]
)

# Define tasks
research_task = Task(
    description="Research {topic} thoroughly",
    agent=researcher,
    expected_output="Comprehensive research findings with sources"
)

analysis_task = Task(
    description="Analyze research findings for key insights",
    agent=analyst,
    context=[research_task],  # Depends on research_task
    expected_output="List of key insights and trends"
)

writing_task = Task(
    description="Write executive summary based on analysis",
    agent=writer,
    context=[research_task, analysis_task],
    expected_output="500-word executive summary"
)

# Create crew and execute
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    verbose=True
)

result = crew.kickoff(inputs={"topic": "AI market trends"})

---

使用 LangGraph 实现

LangGraph 模式（状态机）：

from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    input: str
    research: str
    analysis: str
    output: str

def research_node(state):
    research = researcher_agent.run(state["input"])
    return {"research": research}

def analysis_node(state):
    analysis = analyst_agent.run(state["research"])
    return {"analysis": analysis}

def writing_node(state):
    output = writer_agent.run(state["analysis"])
    return {"output": output}

# Build graph
workflow = StateGraph(AgentState)

workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("writing", writing_node)

workflow.set_entry_point("research")
workflow.add_edge("research", "analysis")
workflow.add_edge("analysis", "writing")
workflow.add_edge("writing", END)

app = workflow.compile()

# Execute
result = app.invoke({"input": "Analyze AI market trends"})

---

最佳实践

1. 明确 Agent 角色

每个 Agent 都应具备特定的专业能力和职责

2. 减少通信

Agent 越多，协调开销越大。应从简单方案开始。

3. 幂等操作

重启 Agent 不应产生副作用

4. 故障处理

设计时应考虑 Agent 故障（重试、回退、跳过）

5. 执行过程可观测

记录 Agent 的决策，追踪执行流程

6. 成本管理

追踪每个 Agent 的 Token 用量，优化成本高昂的调用

---

常见的多 Agent 错误

❌ Agent 太多 → 从 2-3 个开始，仅在需要时增加 ❌ 职责不明确 → 明确界定角色 ❌ 未处理故障 → 一个 Agent 故障就会导致整个系统中断 ❌ 同步瓶颈 → 并行运行互不依赖的 Agent ❌ 忽视成本 → N 个 Agent = N× 次 LLM 调用 ❌ 过度设计 → 单个 Agent 通常就足够

---

决策框架：单 Agent 还是多 Agent

Task Complexity?
│
├─ Simple, linear → Single Agent
│
├─ Complex, requires specialization?
│  │
│  ├─ Sequential steps → Pipeline Pattern
│  ├─ Parallel subtasks → Hierarchical Pattern
│  ├─ Need review → Peer Collaboration
│  └─ Explore solutions → Swarm Pattern
│
└─ Uncertain → Start with Single Agent, refactor to Multi if needed

---

监控与调试

# Track agent execution
class TrackedAgent(Agent):
    def execute(self, task, context):
        start = time.time()
        logger.info(f"{self.name} starting: {task}")

        result = super().execute(task, context)

        duration = time.time() - start
        logger.info(f"{self.name} completed in {duration}s")

        metrics.record({
            "agent": self.name,
            "task": task,
            "duration": duration,
            "tokens": result.token_count,
            "cost": result.cost
        })

        return result

关键指标：

• Agent 执行时间
• 每个 Agent 的 Token 使用量
• 成功率／失败率
• 交接延迟
• 整体工作流时长

---

相关资源

相关技能：

• rag-implementer - 用于基于知识的 Agent
• knowledge-graph-builder - 用于 Agent 知识库
• api-designer - 用于 Agent 通信 API

相关模式：

• META/DECISION-FRAMEWORK.md - 框架选择（CrewAI 与 LangGraph）
• STANDARDS/architecture-patterns/multi-agent-pattern.md - Agent 架构（创建后提供）

相关操作手册：

• PLAYBOOKS/deploy-multi-agent-system.md - 部署指南（创建后提供）
• PLAYBOOKS/debug-agent-workflows.md - 调试流程（创建后提供）