为什么需要多智能体?
一个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团队。