多Agent协作：构建AI团队的架构与实践

一个Agent是一个超级个体，多个Agent就是一个超级团队。当AI学会分工协作，能完成的事情远超单个Agent的能力边界。

为什么需要多Agent

单Agent的局限

问题	具体表现
上下文膨胀	工具越多、prompt越长，推理质量越下降
角色混淆	一个Agent同时扮演研究员、分析师、写作者，品质打折
注意力分散	面对复杂任务，不知从何下手
缺乏制衡	没有”第二双眼睛”审视自己的工作

多Agent的优势

单Agent：一个人包办所有 → 快但粗糙
多Agent：专业分工 + 互相审查 → 更高质量

类比：为什么即使有全栈工程师，大项目还是需要多个人？因为分工带来深度，协作带来质量。

多Agent架构模式

模式一：顺序流水线

Agent 1（研究员）→ Agent 2（分析师）→ Agent 3（写作者）
     ↓                  ↓                 ↓
   收集资料           提炼洞察          撰写报告

适用场景：任务步骤天然有先后顺序

模式二：并行分发

                   ┌→ Agent 1（安全审查）
主Agent（分发器）  ─┼→ Agent 2（性能审查）
                   └→ Agent 3（代码规范审查）
                          ↓
                  主Agent（汇总）

适用场景：同一对象需要多维度审查

模式三：辩论模式

Agent A（正方）──→ 辩论 ──→ 裁判Agent
Agent B（反方）──→        ──→ 综合判断

适用场景：需要充分权衡的决策

模式四：层级结构

               CEO Agent
              /    |    \
         CTO    CPO    CMO Agent
        /   \     |      /   \
    工程师1 工程师2 设计师  市场1 市场2

适用场景：大型复杂项目的端到端执行

实战：用 CrewAI 构建多Agent系统

环境准备

pip install crewai crewai-tools

构建内容创作团队

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
import os

# 配置LLM
os.environ["OPENAI_API_KEY"] = "..."
os.environ["SERPER_API_KEY"] = "..."  # 搜索API

# ============ 定义 Agent ============

# 研究员Agent
researcher = Agent(
    role="资深研究员",
    goal="深入研究指定课题，收集全面、准确、有深度的信息",
    backstory="""你是一位拥有15年经验的研究员，曾在顶级咨询公司工作。
    你擅长从海量信息中提取关键事实，善于发现信息之间的关联，
    对信息源的可信度有敏锐的判断力。""",
    tools=[SerperDevTool(), ScrapeWebsiteTool()],
    verbose=True,
    allow_delegation=False,
    llm="gpt-4"
)

# 分析师Agent
analyst = Agent(
    role="战略分析师",
    goal="从研究资料中提炼洞察，识别模式、机会和风险",
    backstory="""你是一位前麦肯锡战略顾问，擅长数据驱动的分析。
    你能从复杂的材料中看到别人看不到的pattern，
    并用清晰的框架呈现分析结论。""",
    tools=[],
    verbose=True,
    allow_delegation=False,
    llm="gpt-4"
)

# 写作者Agent
writer = Agent(
    role="资深内容创作者",
    goal="将研究和分析转化为引人入胜的内容",
    backstory="""你是一位获奖的科技写作者，曾在知名媒体担任主编。
    你的文章既有深度又通俗易懂，擅长用故事和数据打动读者。
    你注重文章的结构、节奏和可读性。""",
    tools=[],
    verbose=True,
    allow_delegation=False,
    llm="gpt-4"
)

# 审校Agent
reviewer = Agent(
    role="总编辑",
    goal="确保内容质量达到出版标准，核查事实并优化表达",
    backstory="""你是一位严厉但公正的总编辑，有20年从业经验。
    你不会放过任何事实错误、逻辑漏洞或表达问题。
    你知道好文章是改出来的。""",
    tools=[SerperDevTool()],  # 用于事实核查
    verbose=True,
    allow_delegation=False,
    llm="gpt-4"
)

定义任务

# 为每个Agent定义任务

research_task = Task(
    description="""
    研究课题：{topic}

    请完成以下研究：
    1. 搜索该课题的最新发展（2025-2026年）
    2. 找到3-5个权威数据来源
    3. 收集不同观点的专家意见
    4. 识别关键趋势和事件

    输出：结构化的研究报告（含信息来源）
    """,
    expected_output="一份包含数据、引述和来源的研究报告",
    agent=researcher
)

analysis_task = Task(
    description="""
    基于研究报告，进行深度分析：

    1. 识别2-3个核心趋势及其驱动力
    2. 评估各方观点的可信度和立场
    3. 发现被忽视的角度
    4. 提出2个反直觉的观点

    输出：分析报告（使用框架呈现）
    """,
    expected_output="一份结构化的分析报告，含框架图和洞察",
    agent=analyst
)

writing_task = Task(
    description="""
    基于研究和分析，撰写一篇面向行业从业者的深度文章：

    要求：
    - 字数：2000-3000字
    - 开头hook吸引注意力
    - 主体包含具体数据、案例和专家观点
    - 使用小标题和过渡句保持阅读节奏
    - 结尾有明确的takeaway

    目标读者：行业从业者，有一定基础但不一定是专家
    """,
    expected_output="一篇完整的深度文章",
    agent=writer
)

review_task = Task(
    description="""
    审查终稿，检查以下维度：

    1. 事实准确性（重点核查数据和引述）
    2. 逻辑一致性（前后是否有矛盾）
    3. 可读性（结构、节奏、语言）
    4. 价值密度（是否有实质性内容）
    5. 合规性（是否有不当表述）

    给出：通过/修改后通过/需要重写，以及具体修改建议
    """,
    expected_output="一份审查报告，含修改建议和最终评级",
    agent=reviewer
)

组装和运行

# 创建Crew - 跨Agent协作团队
crew = Crew(
    agents=[researcher, analyst, writer, reviewer],
    tasks=[research_task, analysis_task, writing_task, review_task],
    process=Process.sequential,  # 顺序执行
    verbose=True
)

# 运行
result = crew.kickoff(inputs={
    "topic": "2026年AI Agent在企业服务中的应用趋势"
})

print(result)

高级模式：有条件的Agent路由

基于任务复杂度的Agent选择

class AgentRouter:
    """根据任务复杂度路由到不同的Agent配置"""

    def route(self, task: str) -> Crew:
        complexity = self._assess_complexity(task)

        if complexity == "simple":
            # 简单任务：研究员 → 写作者（跳过分
析和审校）
            return Crew(
                agents=[researcher, writer],
                tasks=[simple_research, simple_write],
                process=Process.sequential
            )
        elif complexity == "medium":
            # 中等任务：标准四Agent流水线
            return standard_crew
        else:
            # 复杂任务：增加辩论环节
            return debate_crew

    def _assess_complexity(self, task: str) -> str:
        prompt = f"""
        评估以下任务的复杂度（simple/medium/complex）：

        维度：
        - 需要研究的深度
        - 涉及的利益方数量
        - 不确定性水平
        - 对专业知识的要求

        任务：{task}

        只输出一个词：simple, medium, 或 complex。
        """
        return llm.invoke(prompt).content.strip().lower()

Agent间通信模式

结构化交接

class AgentHandoff:
    """Agent之间的结构化交接"""

    def __init__(self, from_agent: str, to_agent: str):
        self.from_agent = from_agent
        self.to_agent = to_agent

    def format_handoff(self, result, context):
        return f"""
        ========================================
        HANDOFF: {self.from_agent} → {self.to_agent}
        ========================================

        ## 交付成果
        {result}

        ## 特别注意事项
        - 置信度较低的部分：[标注]
        - 需要验证的假设：[标注]
        - 建议下一个Agent重点关注：[标注]

        ## 上下文摘要
        {context}
        """

# 在任务定义中使用
research_task.human_input = False
research_task.callback = lambda output: logger.info(
    AgentHandoff("研究员", "分析师").format_handoff(output, context)
)

调试与监控

Agent执行追踪

import logging
from datetime import datetime

class AgentMonitor:
    """监控多Agent系统的运行"""

    def __init__(self):
        self.execution_trace = []
        self.token_usage = {}
        self.task_timing = {}

    def on_task_start(self, agent_name, task_name):
        trace_id = f"{agent_name}/{task_name}/{datetime.now().isoformat()}"
        self.execution_trace.append({
            "trace_id": trace_id,
            "agent": agent_name,
            "task": task_name,
            "status": "started",
            "timestamp": datetime.now()
        })
        self.task_timing[trace_id] = datetime.now()
        return trace_id

    def on_task_complete(self, trace_id, tokens_used):
        duration = (datetime.now() - self.task_timing[trace_id]).total_seconds()
        self.execution_trace.append({
            "trace_id": trace_id,
            "status": "completed",
            "duration_seconds": duration,
            "tokens_used": tokens_used
        })

    def generate_report(self):
        """生成执行报告"""
        total_tokens = sum(t.get("tokens_used", 0) for t in self.execution_trace if "tokens_used" in t)
        return f"""
        多Agent执行报告
        ────────────────
        总任务数: {len([t for t in self.execution_trace if t.get("status") == "completed"])}
        总Token使用: {total_tokens}
        估算成本: ${total_tokens * 0.00001:.2f}
        """

多Agent系统的最佳实践

设计原则

1. 单一职责
   每个Agent只做一件事，把它做到最好

2. 最小化耦合
   Agent之间通过结构化输出通信，而非自然语言闲聊

3. 渐进复杂
   先用2个Agent跑通，再扩展到5个

4. 成本意识
   每增加一个Agent = 增加推理成本
   衡量"多Agent带来的质量提升"是否值得额外成本

5. 失败隔离
   一个Agent失败不应导致整个流水线崩溃
   关键任务要有fallback方案

何时用单Agent vs 多Agent

单Agent更好：
├── 任务简单，一次对话就能完成
├── 延迟敏感（单Agent更快）
└── 成本敏感

多Agent更好：
├── 任务需要多种不同能力
├── 质量要求高，需要审查机制
├── 任务可以自然分解
└── 需要从多个视角看待同一问题

---

多Agent系统就像是组建一支AI团队。你不是在管理一个超级员工，而是在设计一个组织——需要明确分工、定义协作流程、建立反馈机制。最成功的多Agent系统，往往是最接近人类高效团队工作方式的那些。