13 | 多 Agent 架构——为什么一个 Agent 不够用？

128K 窗口（单 Agent 独享）
├── System Prompt:       500 tokens
├── 历史循环 (20 轮):    80,000 tokens  ← 几乎占满
├── RAG 检索结果:        30,000 tokens
├── 用户指令:            300 tokens
└── 留给模型输出:        ~17,000 tokens  ← 不够写一份完整报告

[Manager Agent]
  ├── 拆解任务: "调研量子计算行业"
  ├── 分配子任务:
  │     ├── Worker A: "搜索量子计算市场规模"
  │     ├── Worker B: "分析主要玩家和竞争格局"
  │     └── Worker C: "总结技术趋势和挑战"
  ├── 收集结果 → 综合生成最终报告

Agent A (正方): "Transformer 架构在未来 3 年内不会被取代"
Agent B (反方): "Mamba 等线性架构已经在长序列上展现出优势..."
Agent A: "但 Transformer 的生态优势（工具链、预训练模型）..."
Agent B: "生态可以迁移，架构的数学效率才是核心..."

[Judge Agent] 综合双方论点，给出最终结论

[Agent 1: 搜集] → 原始资料
     ↓
[Agent 2: 清洗] → 结构化数据
     ↓
[Agent 3: 分析] → 洞察和结论
     ↓
[Agent 4: 撰写] → 最终报告

# ❌ 差的 Agent 定义
agent_research = Agent(
    role="研究员",
    goal="搜索信息、分析数据、生成报告、制作图表、发送邮件"
)

# ✅ 好的 Agent 定义
search_agent = Agent(role="搜索员", goal="搜索并返回相关资料")
analyst_agent = Agent(role="分析师", goal="分析数据并提取洞察")
writer_agent = Agent(role="写手", goal="撰写结构化报告")

# Agent 2 的定义（清晰的上下游关系）
analyst_agent = Agent(
    role="数据分析师",
    goal="接收搜索员返回的原始资料，提取关键数据点并生成结构化分析",
    backstory="你是一位资深数据分析师，擅长从杂乱信息中提取核心洞察",
    inputs_from=["search_agent"],   # 从搜索员接收
    outputs_to=["writer_agent"],    # 输出给写手
)

# 主搜索 Agent + 备用搜索 Agent
primary_search = Agent(role="主搜索员", tools=[search_web, search_academic])
fallback_search = Agent(role="备用搜索员", tools=[search_news, search_social])

# 如果主搜索 Agent 连续失败 2 次，切换到备用

# 搜索员只有搜索工具
search_agent = Agent(
    role="搜索员",
    tools=[search_web, fetch_url]  # 没有 write_file、calculator
)

# 分析师只有分析工具
analyst_agent = Agent(
    role="分析师",
    tools=[calculator, run_python, create_chart]  # 没有 search_web
)

# 写手只有写作工具
writer_agent = Agent(
    role="写手",
    tools=[write_file, format_markdown]  # 没有搜索和分析工具
)

# 共享状态
blackboard = {
    "raw_data": None,       # 搜索员写入
    "analysis": None,       # 分析师写入
    "report": None,         # 写手写入
    "status": "searching"   # Manager 更新
}

# 搜索员完成后
blackboard["raw_data"] = search_results
blackboard["status"] = "analyzing"

# 分析师检测到状态变化，开始工作
if blackboard["status"] == "analyzing":
    analysis = analyze(blackboard["raw_data"])
    blackboard["analysis"] = analysis
    blackboard["status"] = "writing"

# 搜索员完成后，发消息给分析师
message = {
    "from": "search_agent",
    "to": "analyst_agent",
    "type": "data_ready",
    "payload": search_results
}
message_bus.send(message)

# 分析师监听消息
@analyst_agent.on_message("data_ready")
def handle_data(msg):
    analysis = analyze(msg.payload)
    ...