Contents

用CrewAI构建多智能体协作系统:让多个Agent协同完成复杂任务

为什么需要多智能体?

一个Agent能回答问题、调用工具,但面对复杂任务时力不从心:

  • 写一篇深度技术文章:需要调研、写大纲、写正文、审校
  • 做竞品分析:需要搜集数据、分析对比、生成报告
  • 开发一个小项目:需要设计架构、写代码、测试

这些任务拆分给多个专业Agent协作,效果远好于让一个Agent包揽一切。这就是多智能体系统(Multi-Agent System)的核心思想。

CrewAI:像组建团队一样搭建Agent

CrewAI的隐喻非常直觉——Agent就是员工,Task就是任务,Crew就是团队

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
┌─────────────────────────────────────────┐
│                 Crew                     │
│                                         │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐│
│  │ 研究员   │  │ 写手    │  │ 编辑    ││
│  │ Agent   │  │ Agent   │  │ Agent   ││
│  └────┬────┘  └────┬────┘  └────┬────┘│
│       │            │            │       │
│  ┌────▼────┐  ┌────▼────┐  ┌───▼─────┐│
│  │调研任务  │  │写作任务  │  │审校任务  ││
│  └─────────┘  └─────────┘  └─────────┘│
└─────────────────────────────────────────┘

安装与配置

1
pip install crewai crewai-tools
1
2
3
import os
os.environ["OPENAI_API_KEY"] = "your-api-key"
os.environ["OPENAI_API_BASE"] = "http://localhost:18090/v1"  # 自定义代理

实战:搭建一个技术文章写作团队

第一步:定义Agent(角色)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool  # 搜索工具

# Agent 1:技术调研员
researcher = Agent(
    role="技术调研员",
    goal="深入调研指定技术主题,收集最新的资料、案例和数据",
    backstory="""你是一位资深技术调研员,擅长从互联网搜集高质量技术资料。
    你会关注最新的技术趋势、GitHub项目、论文和博客文章。
    你擅长区分一手资料和二手资料,确保信息的准确性。""",
    verbose=True,
    allow_delegation=False,
    tools=[SerperDevTool()],  # 搜索工具
    llm="gpt-5.4"  # 可指定不同模型
)

# Agent 2:技术写手
writer = Agent(
    role="技术写手",
    goal="基于调研资料,撰写清晰、有深度、有代码示例的技术文章",
    backstory="""你是一位经验丰富的技术博主,擅长将复杂的技术概念用通俗易懂的语言解释清楚。
    你的文章特点:
    1. 开头直入主题,不废话
    2. 大量使用代码示例和图表
    3. 包含实际踩坑经验和最佳实践
    4. 结尾有总结和延伸阅读""",
    verbose=True,
    allow_delegation=False,
    llm="gpt-5.4"
)

# Agent 3:文章编辑
editor = Agent(
    role="文章编辑",
    goal="审核文章质量,检查技术准确性、逻辑连贯性和可读性",
    backstory="""你是一位严格的技术编辑,专注于:
    1. 技术准确性——代码能否运行,概念是否正确
    2. 逻辑连贯性——论证是否合理,过渡是否自然
    3. 可读性——段落长度、标题层级、重点突出
    4. SEO优化——标题吸引力、关键词密度、meta description""",
    verbose=True,
    allow_delegation=False,
    llm="gpt-5.4"
)

第二步:定义Task(任务)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
# 任务1:技术调研
research_task = Task(
    description="""深入调研"AI Agent记忆系统"这个主题:
    1. 什么是Agent记忆系统?有哪些类型(短期/长期/工作记忆)?
    2. 主流实现方案有哪些?(向量数据库、知识图谱、KV存储等)
    3. 有哪些开源项目实现了Agent记忆?(如Mem0、Zep等)
    4. 实际应用中的最佳实践和常见陷阱
    
    请输出结构化的调研报告,包含关键信息和参考链接。""",
    expected_output="一份结构化的调研报告,包含技术概述、方案对比、开源项目推荐和最佳实践",
    agent=researcher
)

# 任务2:技术写作
writing_task = Task(
    description="""基于调研报告,撰写一篇1500-3000字的技术博客文章:
    
    要求:
    - 标题吸引人,包含关键词
    - 开头用一个具体场景引入
    - 包含完整的Python代码示例
    - 对比不同方案的优劣
    - 包含踩坑记录
    - 用Markdown格式输出
    
    风格:实战导向,面向有一定Python基础的开发者""",
    expected_output="一篇完整的Markdown格式技术博客文章",
    agent=writer,
    context=[research_task]  # 依赖调研任务的结果
)

# 任务3:文章审校
review_task = Task(
    description="""审校技术文章,输出修改意见:
    1. 技术准确性:代码能否运行?概念是否正确?
    2. 逻辑连贯性:论证是否合理?
    3. 可读性:段落过长?重点不突出?
    4. 如果文章质量合格,直接输出"审核通过"并给出最终版本
    
    请直接给出修改后的最终版本。""",
    expected_output="修改意见列表,或修改后的最终文章",
    agent=editor,
    context=[writing_task, research_task]  # 依赖写作和调研结果
)

第三步:组建Crew并执行

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
# 组建团队
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,  # 按顺序执行
    verbose=True
)

# 开始工作
result = crew.kickoff()

print("=" * 50)
print("最终输出:")
print(result)

执行流程

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
[2026-06-15 10:00:00] Crew 启动
  ├─ 📋 研究任务分配给 → 技术调研员
  │   ├─ 🔍 搜索: "AI Agent memory system 2026"
  │   ├─ 🔍 搜索: "Mem0 Zep agent memory comparison"
  │   └─ ✅ 输出调研报告
  ├─ 📋 写作任务分配给 → 技术写手
  │   ├─ 📖 阅读调研报告
  │   ├─ ✍️ 撰写文章初稿
  │   └─ ✅ 输出Markdown文章
  └─ 📋 审校任务分配给 → 文章编辑
      ├─ 🔍 检查技术准确性
      ├─ ✏️ 修改并优化
      └─ ✅ 输出最终版本

进阶技巧

1. 并行执行(层级模式)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from crewai import Process

# 层级模式:Manager Agent自动分配任务
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, review_task],
    process=Process.hierarchical,  # 层级管理
    manager_llm="gpt-5.4",          # Manager用的模型
    verbose=True
)

2. 自定义工具

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class CodeRunnerInput(BaseModel):
    code: str = Field(description="要执行的Python代码")

class CodeRunnerTool(BaseTool):
    name: str = "code_runner"
    description: str = "执行Python代码并返回结果"
    args_schema: type[BaseModel] = CodeRunnerInput

    def _run(self, code: str) -> str:
        """安全执行代码"""
        import subprocess
        result = subprocess.run(
            ["python", "-c", code],
            capture_output=True, text=True, timeout=30
        )
        if result.returncode == 0:
            return result.stdout
        return f"执行错误: {result.stderr}"

# 注册到Agent
writer_with_tools = Agent(
    role="技术写手",
    goal="撰写可运行的代码示例",
    backstory="...",
    tools=[CodeRunnerTool()]  # 可以验证代码
)

3. 记忆共享

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from crewai import Crew

# Crew级别共享记忆
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, review_task],
    memory=True,  # 启用共享记忆
    verbose=True
)

# Agent之间可以共享发现
# Researcher发现的资料会自动传递给Writer

4. 输出为文件

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
# 保存结果到文件
import datetime

result = crew.kickoff()

output_path = f"/opt/data/output/article_{datetime.date.today()}.md"
with open(output_path, "w") as f:
    f.write(str(result))

print(f"文章已保存到: {output_path}")

成本与性能

配置 预估Token消耗 执行时间 质量
全用gpt-5.4 ~15000 tokens 3-5分钟 ⭐⭐⭐⭐⭐
混合(gpt-5.4+mini) ~8000 tokens 2-4分钟 ⭐⭐⭐⭐
全用gpt-5.4-mini ~5000 tokens 1-3分钟 ⭐⭐⭐

建议:调研和审校用强模型(gpt-5.4),写作用中等模型(gpt-5.4-mini),平衡质量和成本。

踩坑记录

1. Agent互相委派死循环

allow_delegation=True时,Agent可能把任务踢回给对方。小团队建议设为False,用Process.sequential控制执行顺序。

2. 上下文太长

3个Agent的输出累积后,context可能超过模型上下文窗口。解决:每个Task的expected_output尽量精简,或用更强的长上下文模型。

3. 调试困难

多Agent系统的调试比单Agent难得多。建议开启verbose=True,并在每个Agent的backstory中明确输出格式要求。

总结

多智能体系统的核心价值是专业化分工。就像一个创业团队不会让CEO一个人干所有事,AI系统也不应该让一个Agent包揽一切。

CrewAI的"员工-任务-团队"隐喻降低了理解门槛,但本质是三个核心概念:

  • Agent:角色定义(我是谁、擅长什么)
  • Task:任务描述(要做什么、输出什么格式)
  • Process:协作方式(顺序、层级、并行)

掌握了这三点,你就能搭建出真正能干活的AI团队。