第01章 为什么需要多Agent：突破单Agent的天花板
第01章 为什么需要多Agent：突破单Agent的天花板
“没有任何一位伟大的将领独自打赢了战争。是协作，让力量倍增。” —— 军事历史研究
你已经有了一个能调用工具、有记忆、会规划的 Agent。它能搜索网页、写代码、分析数据……
那为什么还需要多个 Agent？
这一章，我们从根本上回答这个问题：单 Agent 有哪些无法逾越的天花板？多 Agent 又是如何打破这些限制的？
1.1 单 Agent 的三堵墙
# ====================================================
# 单 Agent 的局限性：三个真实场景
# ====================================================
"""
让我们用一个真实的任务来揭示单 Agent 的天花板。
任务：为一家 B2B SaaS 公司制作季度竞品分析报告
- 调研 10 个竞品
- 每个竞品分析：功能、定价、用户口碑、技术栈
- 生成对比矩阵
- 撰写战略建议
预计需要：40,000 token 的上下文
"""
import asyncio
import time
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def single_agent_competitor_analysis(competitors: list[str]) -> str:
"""
单 Agent 尝试处理复杂的竞品分析任务
这会暴露三个核心问题：
1. 上下文窗口溢出
2. 注意力分散导致质量下降
3. 串行执行耗时长
"""
all_research = []
print("单 Agent 开始工作...")
start = time.time()
for i, competitor in enumerate(competitors):
print(f" 正在分析第 {i+1}/{len(competitors)} 个竞品：{competitor}")
# 模拟研究每个竞品（真实场景会调用搜索工具）
response = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": f"""
分析竞品：{competitor}
请提供：
1. 核心功能列表（5个）
2. 定价模式
3. 主要用户群体
4. 技术优势
5. 已知弱点
已收集的其他竞品数据（上下文）：
{chr(10).join(all_research[-3:]) if all_research else "（首个竞品，暂无对比数据）"}
"""
}],
max_tokens=800,
)
research = response.choices[0].message.content
all_research.append(f"=== {competitor} ===\n{research}")
# 问题1：随着循环进行，上下文越来越长，性能越来越差
current_context_size = sum(len(r) for r in all_research)
print(f" 当前累积上下文：{current_context_size} 字符")
if current_context_size > 50000:
print(" ⚠️ 上下文接近上限，后续分析质量将下降！")
elapsed = time.time() - start
print(f"\n单 Agent 完成，耗时：{elapsed:.1f}秒")
# 问题2：串行执行，10个竞品需要10倍时间
print(f"串行执行的问题：{len(competitors)} 个竞品需要 {elapsed:.1f}秒")
print(f"如果是 100 个竞品，需要：约 {elapsed * 10:.0f}秒")
# 问题3：单个 Agent 试图同时扮演"研究员"、"分析师"、"写手"，注意力分散
print("\n单 Agent 的角色冲突：")
print(" - 研究每个竞品（需要广度）")
print(" - 深度分析各维度（需要专注）")
print(" - 保持跨竞品一致性（需要全局视野）")
print(" - 最终撰写报告（需要写作技巧）")
print(" → 一个 Agent 同时做这些，像让一个人同时烹饪、服务、收银、采购")
return "\n\n".join(all_research[:3]) + "\n...(因上下文限制，剩余分析质量已降低)"
# 演示：3个竞品就能看到问题
competitors = ["Notion", "Obsidian", "Roam Research"]
asyncio.run(single_agent_competitor_analysis(competitors))
1.2 三堵墙的本质：认知、注意力、时间
# ====================================================
# 深入理解三堵墙
# ====================================================
"""
墙1：上下文窗口（认知容量）
- GPT-4o：128K token ≈ 约 10 万汉字
- 听起来很多？一份完整的竞品调研报告可能需要 200K+
- 更重要的是：上下文越长，"注意力"越分散，质量越差
- 研究发现：当上下文超过 60K token，模型开始"遗忘"早期信息
墙2：注意力焦点（专业深度）
- 人类顾问团队有专家分工：技术专家 + 市场专家 + 财务专家
- 单个 LLM 必须同时扮演所有角色
- 结果：每个角色都是"通才"水平，没有专家深度
- 专注度原理：同一时间聚焦的范围越窄，输出质量越高
墙3：串行时间（执行效率）
- 单 Agent 只能串行执行步骤
- 10 个独立子任务 × 每个 30 秒 = 总计 5 分钟
- 多 Agent 并行：10 个任务同时执行 = 30 秒
- 效率提升：10 倍
"""
import asyncio
from dataclasses import dataclass, field
from typing import List, Dict, Any
@dataclass
class SingleAgentLimitations:
"""量化单 Agent 的局限性"""
# 上下文窗口局限
max_context_tokens: int = 128_000
quality_degradation_threshold: int = 60_000 # 超过此值质量开始下降
typical_complex_task_tokens: int = 200_000 # 复杂任务需要的 token 数
# 注意力局限
roles_required: List[str] = field(default_factory=lambda: [
"研究员", "分析师", "策略师", "写手", "校对者"
])
# 时间局限
avg_subtask_seconds: float = 30.0
def analyze(self, num_subtasks: int = 10) -> Dict[str, Any]:
"""分析完成 num_subtasks 个子任务的单 Agent vs 多 Agent"""
# 上下文是否够用？
tokens_needed = num_subtasks * 10_000 # 每子任务约1万 token
context_ok = tokens_needed <= self.max_context_tokens
context_degraded = tokens_needed > self.quality_degradation_threshold
# 单 Agent 时间
single_time = self.avg_subtask_seconds * num_subtasks
# 多 Agent 并行时间（理想情况）
multi_time = self.avg_subtask_seconds # 所有任务并行
return {
"子任务数量": num_subtasks,
"上下文需求（tokens）": tokens_needed,
"上下文是否够用": "✅" if context_ok else "❌ 超出上限",
"质量是否受影响": "⚠️ 会下降" if context_degraded else "✅ 正常",
"单Agent耗时": f"{single_time:.0f}秒",
"多Agent耗时（并行）": f"{multi_time:.0f}秒",
"效率提升": f"{single_time/multi_time:.0f}倍",
"角色冲突": f"需要同时扮演 {len(self.roles_required)} 个角色",
}
def print_comparison(self):
"""打印不同规模任务的对比"""
print("=== 单 Agent vs 多 Agent 能力对比 ===\n")
for n_tasks in [3, 10, 20, 50]:
result = self.analyze(n_tasks)
print(f"任务规模：{n_tasks} 个子任务")
for k, v in result.items():
if k != "子任务数量":
print(f" {k}: {v}")
print()
limits = SingleAgentLimitations()
limits.print_comparison()
1.3 多 Agent 如何打破这三堵墙
# ====================================================
# 多 Agent 的解决方案
# ====================================================
import asyncio
from openai import AsyncOpenAI
from typing import List
client = AsyncOpenAI()
class MultiAgentCompetitorAnalysis:
"""
多 Agent 版本的竞品分析
架构：
- 研究员 Agents（并行）：每人只研究1个竞品，上下文纯净
- 分析师 Agent：汇总研究结果，做对比分析
- 策略师 Agent：基于分析生成战略建议
- 协调器：统筹全局
"""
def __init__(self, model: str = "gpt-4o"):
self.model = model
async def researcher_agent(self, competitor: str) -> dict:
"""
研究员 Agent：专注研究单个竞品
好处：
1. 上下文只包含这一个竞品的信息（纯净、无干扰）
2. 角色单一：只需要做"研究员"
3. 可以和其他研究员同时工作
"""
response = await client.chat.completions.create(
model="gpt-4o-mini", # 研究用小模型，省成本
messages=[
{
"role": "system",
"content": "你是一名专业的竞品研究员。你的唯一任务是深入研究指定的产品，提供结构化的竞品情报。"
},
{
"role": "user",
"content": f"""
深入研究产品：{competitor}
提供以下信息：
1. 核心功能（列出5个最重要的）
2. 定价模式（免费/付费层级）
3. 目标用户（谁在用它）
4. 技术特点（差异化能力）
5. 用户痛点（主要投诉/不满）
要求：具体、简洁、有依据
"""
}
],
max_tokens=600,
)
return {
"competitor": competitor,
"research": response.choices[0].message.content,
"agent": "researcher"
}
async def analyst_agent(self, research_results: List[dict]) -> str:
"""
分析师 Agent：基于多个研究结果做对比分析
好处：
1. 只负责"分析"这一个职责
2. 接收已经整理好的研究结果（不需要做研究）
3. 专注于找出差异、模式、机会点
"""
research_summary = "\n\n".join([
f"### {r['competitor']}\n{r['research']}"
for r in research_results
])
response = await client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "你是一名竞争分析师。你擅长从多个竞品研究中找出规律、差距和机会。"
},
{
"role": "user",
"content": f"""
基于以下竞品研究，生成结构化的对比分析：
{research_summary}
输出：
1. 功能对比矩阵（表格形式）
2. 定价策略对比
3. 共同痛点（多个竞品都有的问题）
4. 市场空白（没有人做好的地方）
"""
}
],
max_tokens=1000,
)
return response.choices[0].message.content
async def strategist_agent(self, analysis: str, our_product: str) -> str:
"""
策略师 Agent：基于分析生成战略建议
好处：
1. 只负责"战略建议"这一职责
2. 输入是已经处理好的分析结果
3. 专注于"我们应该怎么做"
"""
response = await client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "你是一名产品战略顾问。你的任务是基于竞品分析，提出具体可执行的战略建议。"
},
{
"role": "user",
"content": f"""
我们的产品：{our_product}
竞品分析报告：
{analysis}
请提供：
1. 我们的核心差异化机会（3条）
2. 需要立刻跟进的竞品动作（2条）
3. 不值得竞争的方向（需要放弃的地方）
4. 6个月优先级建议
"""
}
],
max_tokens=800,
)
return response.choices[0].message.content
async def run(self, competitors: List[str], our_product: str) -> dict:
"""
协调器：统筹多 Agent 工作流
"""
print(f"多 Agent 团队启动，分析 {len(competitors)} 个竞品...\n")
start = __import__("time").time()
# 阶段1：所有研究员 Agent 并行工作
print("阶段1：研究员 Agents 并行研究（同时进行）")
research_results = await asyncio.gather(*[
self.researcher_agent(competitor)
for competitor in competitors
])
phase1_time = __import__("time").time() - start
print(f" 所有竞品研究完成，耗时：{phase1_time:.1f}秒（并行！）")
# 阶段2：分析师 Agent 处理汇总结果
print("\n阶段2：分析师 Agent 进行对比分析")
analysis = await self.analyst_agent(list(research_results))
phase2_time = __import__("time").time() - start - phase1_time
print(f" 对比分析完成，耗时：{phase2_time:.1f}秒")
# 阶段3：策略师 Agent 生成战略建议
print("\n阶段3：策略师 Agent 生成战略建议")
strategy = await self.strategist_agent(analysis, our_product)
total_time = __import__("time").time() - start
print(f"\n✅ 多 Agent 团队完成，总耗时：{total_time:.1f}秒")
# 对比单 Agent 估算时间
single_agent_estimate = len(competitors) * 30
print(f"相比单 Agent（约 {single_agent_estimate}秒），效率提升：{single_agent_estimate/total_time:.1f}倍")
return {
"research": [r["research"] for r in research_results],
"analysis": analysis,
"strategy": strategy,
"total_seconds": total_time,
}
# 运行多 Agent 团队
team = MultiAgentCompetitorAnalysis()
result = asyncio.run(team.run(
competitors=["Notion", "Obsidian", "Roam Research"],
our_product="我们的知识管理SaaS"
))
print("\n=== 战略建议（摘要）===")
print(result["strategy"][:500] + "...")
本章小结
-
单 Agent 有三堵无法逾越的墙：上下文窗口（认知容量）、注意力焦点（专业深度）和串行时间（执行效率）。任何一堵墙都会成为复杂任务的瓶颈。
-
上下文窗口不只是"装不下"的问题：即使还没满，当上下文超过 60K token，模型的注意力就会开始分散，早期信息逐渐被"遗忘"，质量下降往往比开发者意识到的更早。
-
角色分离是多 Agent 的核心价值：研究员专注研究，分析师专注分析，策略师专注战略——每个 Agent 只做一件事，像人类专家团队一样运作，输出质量显著提升。
-
并行执行带来的效率不是线性的：10 个任务并行 vs 串行，理论上是 10 倍效率，实际中因调度开销可能是 5-8 倍，但这已经是巨大的优势。
-
多 Agent 不是"更复杂"，而是"更合理"：当任务本身有多个独立维度时，用多 Agent 处理是符合问题本质的架构，而不是过度设计。
# 核心行动：判断你的任务是否需要多 Agent
def should_use_multi_agent(task_description: str) -> dict:
"""简单的多 Agent 适用性检查"""
triggers = {
"上下文溢出风险": len(task_description) > 5000 or "大量" in task_description,
"需要专业深度": any(kw in task_description for kw in ["专业", "深入", "详细分析"]),
"可并行子任务": any(kw in task_description for kw in ["多个", "所有", "每个", "批量"]),
"需要多个角色": any(kw in task_description for kw in ["研究", "分析", "策略", "报告"]),
}
score = sum(triggers.values())
return {
"触发条件": triggers,
"建议": "使用多 Agent" if score >= 2 else "单 Agent 足够",
"信心": f"{score}/4 个指标触发",
}
# 测试
print(should_use_multi_agent("帮我调研10个竞品并生成对比报告"))
本章提示词模板
【模板1：任务适用性评估提示词】
请评估以下任务是否适合使用多 Agent 协作：
任务描述：{task_description}
评估维度：
1. 上下文复杂度：这个任务需要处理多少信息？单个上下文窗口装得下吗？
2. 角色多样性：完成这个任务需要哪些不同的专业角色？
3. 并行机会：任务中有哪些步骤可以同时执行而不互相依赖？
4. 质量要求：每个子任务需要专家级输出还是通才水平即可？
请给出：
- 是否建议使用多 Agent（是/否/视情况）
- 如果是，建议的 Agent 数量和角色划分
- 如果否，原因是什么
【模板2：单 Agent 瓶颈诊断提示词】
我的 Agent 在处理以下任务时遇到了问题：
任务：{task_description}
症状：{symptoms}（如：回答越来越短、前后不一致、遗漏早期要求）
请诊断：
1. 这是否是上下文窗口溢出导致的问题？
2. 是否存在角色冲突（Agent同时承担太多职责）？
3. 是否有可以并行执行但实际上串行执行的步骤？
4. 推荐的解决方案：调整提示词、拆分为多 Agent、还是重构任务？