Multi-Agent 协作:CrewAI 实战

一个 Agent 是员工,多个 Agent 是团队。本文用 CrewAI 搭建一个「产品经理 + 开发 + 测试」三人开发小组,看他们怎么协作完成一个真实需求。


一、为什么需要 Multi-Agent

单 Agent 的瓶颈:

用户:帮我做一个待办事项 App,需要前端后端和数据库

单 Agent 的困境:
- 前端要 React,后端要 FastAPI,数据库要 PostgreSQL
- 一个 Agent 得同时掌握三种技术栈
- 上下文很快被占满
- 生成的代码风格不统一

Multi-Agent 的解法:

              用户需求
                 │
         ┌───────▼───────┐
         │   PM Agent    │ 分析需求,拆解任务
         └───────┬───────┘
                 │
     ┌───────────┼───────────┐
     ▼           ▼           ▼
前端 Agent   后端 Agent   测试 Agent
(React)      (FastAPI)    (Pytest)
     │           │           │
     └───────────┼───────────┘
                 ▼
           整合 + 交付

二、CrewAI 核心概念

概念 含义 示例
Agent 一个 AI 角色 PM、前端开发、QA
Task 分配给 Agent 的具体任务 “设计 API 接口”
Crew Agent 团队 “开发小组”
Process 协作模式 sequential(顺序)/ hierarchical(层级)

三、搭建三人开发团队

# pip install crewai langchain-openai

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

# ── 共用 LLM ──
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key="your-key",
    base_url="https://api.deepseek.com/v1"
)

# ── 1. 定义三个 Agent ──

pm_agent = Agent(
    role="产品经理",
    goal="分析用户需求,输出清晰的技术需求文档",
    backstory="你有 5 年 SaaS 产品经验,擅长把模糊需求转化为可执行的技术方案",
    llm=llm,
    verbose=True
)

frontend_agent = Agent(
    role="前端开发工程师",
    goal="根据技术需求,用 React + TypeScript 实现前端界面",
    backstory="你是一个 React 专家,擅长组件化开发和状态管理",
    llm=llm,
    verbose=True
)

backend_agent = Agent(
    role="后端开发工程师",
    goal="根据技术需求,用 FastAPI + SQLAlchemy 实现后端 API",
    backstory="你是一个 Python 后端专家,擅长 RESTful API 设计和数据库建模",
    llm=llm,
    verbose=True
)

qa_agent = Agent(
    role="测试工程师",
    goal="验证前后端代码的功能完整性和质量",
    backstory="你有 3 年自动化测试经验,擅长单元测试、集成测试和边界条件测试",
    llm=llm,
    verbose=True
)

Agent 设计的核心要素

Role(角色):越具体越好。「前端开发」不如「React 前端工程师,擅长 TypeScript 和 Tailwind CSS」。

Goal(目标):单一、明确。「写代码」不如「根据 PRD 用 React + TypeScript 实现前端界面,包含 Loading/空数据/错误三种状态」。

Backstory(背景):给 Agent 一段「故事」,让它的回答更有风格。不是闲聊,而是影响它如何做决策。


四、定义任务

# ── 2. 定义任务链 ──

task_prd = Task(
    description="""
    用户需求:做一个个人记账 Web 应用。
    功能要求:
    1. 记录收入/支出(金额、分类、备注、日期)
    2. 查看月度收支汇总
    3. 分类统计饼图
    
    请输出一份技术需求文档,包含:
    - 数据模型设计
    - API 接口定义
    - 前端页面结构
    - 技术栈选型
    """,
    expected_output="一份 Markdown 格式的技术需求文档",
    agent=pm_agent
)

task_backend = Task(
    description="""
    根据 PM 的技术需求文档,实现后端 API。
    
    要求:
    - FastAPI + SQLAlchemy + SQLite
    - 完整的 CRUD 接口
    - 月度汇总接口
    - 添加请求参数校验和错误处理
    """,
    expected_output="完整的 Python 后端代码,包含所有 API 接口",
    agent=backend_agent
)

task_frontend = Task(
    description="""
    根据 PM 的技术需求文档和后端 API,实现前端界面。
    
    要求:
    - React + TypeScript
    - 记账表单(金额、分类下拉、备注、日期选择)
    - 交易列表(分页、筛选)
    - 月度汇总卡片
    - 分类饼图(用 recharts)
    """,
    expected_output="完整的前端代码,包含所有页面和组件",
    agent=frontend_agent
)

task_qa = Task(
    description="""
    审查前后端代码,检查:
    1. 前端是否处理了 Loading/空数据/错误三种状态
    2. 后端 API 是否有输入校验
    3. 前后端数据格式是否一致
    
    输出测试报告和改进建议。
    """,
    expected_output="测试报告,列出发现的问题和建议",
    agent=qa_agent
)

五、组建 Crew 并运行

# ── 3. 组建团队 ──

dev_team = Crew(
    agents=[pm_agent, frontend_agent, backend_agent, qa_agent],
    tasks=[task_prd, task_backend, task_frontend, task_qa],
    process=Process.sequential,  # 顺序执行:PM → 后端 → 前端 → 测试
    verbose=True
)

# ── 4. 启动! ──
result = dev_team.kickoff()
print(result)

执行过程

🚀 Crew 启动
  ├── PM Agent 正在写 PRD...
  │   ✅ PRD 完成(技术栈:React + FastAPI + SQLite)
  ├── 后端 Agent 正在写代码...
  │   ✅ 5 个 API 接口完成
  ├── 前端 Agent 正在写代码...
  │   ✅ 4 个页面组件完成
  └── 测试 Agent 正在审查...
      ✅ 发现 3 个问题,建议 2 处改进

📦 交付物:
  - 技术需求文档 × 1
  - 后端代码 × 5 文件
  - 前端代码 × 6 文件
  - 测试报告 × 1

六、两种协作模式

Sequential(顺序模式)

Task 1 → Task 2 → Task 3 → Task 4
 (PM)    (后端)    (前端)    (QA)

每个任务的输出自动作为下一个任务的上下文。适合有明确前后依赖的任务。

Hierarchical(层级模式)

           ┌─────────┐
           │ Manager  │  ← 分配任务、审核结果
           │  Agent   │
           └────┬─────┘
    ┌───────────┼───────────┐
    ▼           ▼           ▼
Agent 1     Agent 2     Agent 3

Manager Agent 负责拆任务、分派、汇总。适合复杂、需要动态决策的场景。

# 层级模式示例
manager = Agent(
    role="技术总监",
    goal="协调开发团队,确保项目按时高质量交付",
    backstory="你是技术总监,10 年经验,擅长资源调配和项目管控",
    llm=llm,
    allow_delegation=True  # 允许委派任务
)

hierarchical_crew = Crew(
    agents=[manager, frontend_agent, backend_agent, qa_agent],
    tasks=[task_prd, task_backend, task_frontend, task_qa],
    process=Process.hierarchical,
    manager_agent=manager
)

七、Multi-Agent 实战要点

7.1 Agent 角色要互补而非重叠

❌ 两个 Agent 都会写 React → 代码冲突
✅ 前端 Agent 写 UI,UI 设计 Agent 出设计稿

7.2 任务描述要能被下游消费

# ❌ 太模糊
Task(description="写后端代码", ...)

# ✅ 下游 Agent 能直接用
Task(
    description="根据 PRD 实现 API。输出格式:Python 代码,每个文件用 === FILE: xxx.py === 分隔",
    expected_output="格式化的 Python 代码",
    ...
)

7.3 设置重试和超时

Task(
    ...,
    max_retries=2,      # 失败重试 2 次
    max_execution_time=300  # 最多 5 分钟
)

八、CrewAI vs AutoGen vs LangGraph

维度 CrewAI AutoGen LangGraph
学习曲线 ⭐⭐ 简单 ⭐⭐⭐ 中等 ⭐⭐⭐⭐ 较陡
角色定义 Role + Goal + Backstory Agent + System Message Node 函数
协作模式 顺序 / 层级 对话式 状态图
适合场景 固定流程 动态对话 精细控制
中文支持 ✅ 好 🟡 一般 ✅ 好

建议:先用 CrewAI 入门,理解 Multi-Agent 的逻辑后再学 LangGraph 做精细控制。


九、总结

  1. Multi-Agent 解决单 Agent 的复杂度瓶颈——术业有专攻
  2. CrewAI 用角色+目标+背景定义 Agent——像招人一样招 AI
  3. 顺序模式适合固定流程,层级模式适合动态决策
  4. Agent 角色要互补、任务描述要明确、输出格式要规范

十、生产实战:Multi-Agent 的隐藏成本

10.1 成本:Agent 越多不等于越好

单 Agent 完成任务:
  1 个 LLM 调用 → 5000 Token → 完成

CrewAI 4 个 Agent 完成同一个任务:
  PM Agent:     2000 Token
  后端 Agent:   3000 Token
  前端 Agent:   3000 Token
  测试 Agent:   2000 Token
  ───────────────────────
  总计:        10000 Token  (2×)
  时间:       4×(串行模式下)

经验:评估一个任务是否真的需要 Multi-Agent。如果任务能在一个 Agent 的 Context Window 内完成,单 Agent 更便宜更快。Multi-Agent 用于「单 Agent 真的搞不定」的场景。

10.2 Agent 间的沟通成本

Agent 的输出给下一个 Agent ——如果格式不一致,后面全乱:

# PM Agent 输出的数据模型 → 前后端 Agent 必须严格一致
{
  "api": {
    "GET /transactions": {
      "params": {"month": "string"},
      "response": {"transactions": "array", "total": "number"}
    }
  }
}

# 如果 PM 输出 {"total": "int"} 但后端写成 {"total": "float"}
# 前端解析会炸 → 而 Agent 不会主动告诉你

经验:Multi-Agent 项目中,PM Agent 的输出用严格的 JSON Schema 约束。后续 Agent 在开始干活前,先验证上游输入是否合法。

10.3 错误级联:一个小错放大 4 倍

PM Agent 理解错了一个需求 → 后端按错的写 → 前端按错的对接 → 测试按错的验证 → 4 个 Agent 都在做无用功。

# 每步执行后做验证
class ValidatedTask(Task):
    def execute(self, context):
        result = super().execute(context)
        # 如果是 PM 的输出,先给用户看一眼
        if self.agent.role == "产品经理":
            if not confirm_with_human(result):
                return {"error": "用户撤销了 PRD"}
        return result

经验:在 PM Agent 之后加一个「人类确认」环节,收益远大于成本。PM 错 = 全线错。

10.4 什么时候不该用 Multi-Agent

场景 建议 理由
单步骤简单任务 单 Agent Multi-Agent 纯属浪费
任务可在一个 Context Window 完成 单 Agent 串行 Multi-Agent 更慢更贵
Agent 角色高度重叠 单 Agent 两个 Agent 干同样的事 = 内耗
需要极致低延迟(< 1s) 单 Agent Multi-Agent 串行延迟叠加
复杂多领域任务 Multi-Agent 单 Agent Context 不够用

🎉 系列完结。8 篇文章从 Token 原理写到 Multi-Agent 协作,每篇都带了生产环境的真实经验和坑。接下来的关键是把这些串起来做一个你自己的 Agent 产品——代码是学出来的,经验是踩坑踩出来的。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐