CrewAI 框架
CrewAI 是一个基于角色扮演的多智能体协作框架,通过定义明确的角色、任务和协作流程,让多个 Agent 像团队一样协作完成任务。它强调"角色化设计"和"流程编排",非常适合内容创作、研究任务和自动化工作流。
一、核心原理
1.1 CrewAI 设计哲学
CrewAI 的核心思想是将 Agent 视为"团队成员",每个成员有明确的角色和职责:
┌─────────────────────────────────────────────────────────────┐
│ CrewAI 设计哲学 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 传统 Agent: │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 单一 Agent │ │
│ │ │ │
│ │ 输入 ──→ 处理 ──→ 输出 │ │
│ │ │ │
│ │ 问题:能力有限,难以处理复杂任务 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ CrewAI 模式: │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Crew (团队) │ │
│ │ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ Agent 1 │ │ Agent 2 │ │ Agent 3 │ │ │
│ │ │ (角色A) │ │ (角色B) │ │ (角色C) │ │ │
│ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │
│ │ │ │ │ │ │
│ │ └────────────┼────────────┘ │ │
│ │ ↓ │ │
│ │ ┌─────────────┐ │ │
│ │ │ Process │ ← 流程编排 │ │
│ │ │ (协作流程) │ │ │
│ │ └─────────────┘ │ │
│ │ │ │
│ │ 优势:角色明确,流程可控,易于调试 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.2 核心概念
| 概念 | 英文 | 描述 | 示例 |
|---|---|---|---|
| Agent | Agent | 具有特定角色和能力的智能体 | 研究员、撰稿人、编辑 |
| Task | Task | 需要完成的具体任务 | "研究AI发展趋势" |
| Crew | Crew | Agent 和 Task 的组织单位 | 一个内容创作团队 |
| Process | Process | 任务执行流程 | Sequential、Hierarchical |
| Tool | Tool | Agent 可使用的工具 | 搜索、数据库查询 |
1.3 四层架构
┌─────────────────────────────────────────────────────────────┐
│ CrewAI 四层架构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Layer 4: Crew Layer (团队层) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ Crew = Agents + Tasks + Process │ │
│ │ │ │
│ │ • 组织 Agent 和 Task │ │
│ │ • 定义协作流程 │ │
│ │ • 管理执行上下文 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ ↑ │
│ Layer 3: Process Layer (流程层) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ Sequential / Hierarchical │ │
│ │ │ │
│ │ • 定义任务执行顺序 │ │
│ │ • 协调 Agent 间协作 │ │
│ │ • 管理任务状态 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ ↑ │
│ Layer 2: Task Layer (任务层) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ Task = Description + Agent + Output │ │
│ │ │ │
│ │ • 定义任务目标 │ │
│ │ • 指定负责 Agent │ │
│ │ • 管理任务输出 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ ↑ │
│ Layer 1: Agent Layer (智能体层) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ Agent = Role + Goal + Tools + Backstory │ │
│ │ │ │
│ │ • 定义角色身份 │ │
│ │ • 设定目标方向 │ │
│ │ • 配置能力工具 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘二、核心组件详解
2.1 Agent 组件
Agent 是 CrewAI 的核心执行单元:
from crewai import Agent
# 创建 Agent
researcher = Agent(
# 基础属性
role="高级研究员", # 角色名称
goal="发现前沿技术趋势和洞察", # 目标
backstory=""" # 背景故事(塑造角色性格)
你是一位经验丰富的研究员,在科技行业工作了15年。
你擅长从海量信息中提取关键洞察,你的分析报告被业界广泛认可。
你注重事实和数据,总是给出可靠的结论。
""",
# 能力配置
verbose=True, # 详细日志
allow_delegation=False, # 是否允许委派任务
llm="gpt-4", # 使用的 LLM
# 工具配置
tools=[search_tool, scrape_tool], # 可用工具
# 行为约束
max_iter=15, # 最大迭代次数
max_rpm=100, # 每分钟最大请求数
memory=True, # 是否启用记忆
)Agent 属性详解:
| 属性 | 类型 | 说明 | 示例 |
|---|---|---|---|
role | str | 角色名称,定义身份 | "数据分析师" |
goal | str | 目标,定义方向 | "从数据中发现价值" |
backstory | str | 背景故事,塑造性格 | "10年经验的分析师..." |
tools | list | 可用工具列表 | [search_tool] |
llm | str | 使用的语言模型 | "gpt-4" |
verbose | bool | 是否输出详细日志 | True |
allow_delegation | bool | 是否允许委派 | False |
max_iter | int | 最大迭代次数 | 15 |
memory | bool | 是否启用记忆 | True |
2.2 Task 组件
Task 定义了需要完成的工作:
from crewai import Task
# 创建任务
research_task = Task(
description=""" # 任务描述
研究 {topic} 领域的最新发展趋势。
需要关注:
1. 主要技术突破
2. 市场动态和投资情况
3. 未来3年的发展预测
输出一份详细的研究报告。
""",
expected_output=""" # 期望输出格式
一份结构化的研究报告,包含:
- 摘要
- 主要发现(3-5条)
- 详细分析
- 参考资料
""",
agent=researcher, # 负责的 Agent
tools=[search_tool], # 任务专用工具
# 输出配置
output_file="report.md", # 输出文件
async_execution=False, # 是否异步执行
# 上下文
context=[other_task], # 依赖的其他任务
)2.3 Crew 组件
Crew 是 Agent 和 Task 的组织单位:
from crewai import Crew, Process
# 创建 Crew
crew = Crew(
agents=[researcher, writer, editor], # Agent 列表
tasks=[research_task, write_task, edit_task], # Task 列表
# 流程配置
process=Process.sequential, # 执行流程
# 运行配置
verbose=True, # 详细日志
memory=True, # 启用记忆
cache=True, # 启用缓存
# 输出配置
output_log_file="crew_log.txt", # 日志文件
share_crew_output=True, # 共享输出
)2.4 Process 组件
Process 定义了任务的执行方式:
┌─────────────────────────────────────────────────────────────┐
│ Process 执行模式 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. Sequential Process(顺序执行) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ Task 1 ──→ Task 2 ──→ Task 3 ──→ Output │ │
│ │ (Agent A) (Agent B) (Agent C) │ │
│ │ │ │
│ │ 特点: │ │
│ │ • 任务按顺序依次执行 │ │
│ │ • 前一任务输出作为后一任务输入 │ │
│ │ • 适合有依赖关系的流水线任务 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 2. Hierarchical Process(层次执行) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ ┌─────────────┐ │ │
│ │ │ Manager │ ← 自动创建的管理Agent │ │
│ │ │ Agent │ │ │
│ │ └──────┬──────┘ │ │
│ │ │ │ │
│ │ ┌───────────┼───────────┐ │ │
│ │ ↓ ↓ ↓ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Worker │ │ Worker │ │ Worker │ │ │
│ │ │ Agent A │ │ Agent B │ │ Agent C │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ │ │ │
│ │ 特点: │ │
│ │ • 自动创建 Manager Agent 进行协调 │ │
│ │ • Manager 负责任务分配和结果汇总 │ │
│ │ • 适合复杂任务自动分解 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘三、代码实现
3.1 基础示例:内容创作团队
"""
CrewAI 基础示例:内容创作团队
展示顺序协作模式的完整实现
"""
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileReadTool
# 初始化工具
search_tool = SerperDevTool() # 搜索工具
file_read_tool = FileReadTool() # 文件读取工具
# ==================== 定义 Agent ====================
# Agent 1: 研究员
researcher = Agent(
role="高级研究员",
goal="发现前沿技术趋势和洞察",
backstory="""
你是一位经验丰富的科技研究员,在硅谷工作了15年。
你擅长从海量信息中提取关键洞察,你的分析报告被业界广泛认可。
你注重事实和数据,总是给出可靠的结论。
""",
verbose=True,
allow_delegation=False,
tools=[search_tool],
llm="gpt-4"
)
# Agent 2: 撰稿人
writer = Agent(
role="技术撰稿人",
goal="创作引人入胜的技术文章",
backstory="""
你是一位技术作家,曾为多家知名科技媒体撰稿。
你擅长将复杂的技术概念转化为通俗易懂的文章。
你的文章风格清晰、有趣,深受读者喜爱。
""",
verbose=True,
allow_delegation=False,
tools=[file_read_tool],
llm="gpt-4"
)
# Agent 3: 编辑
editor = Agent(
role="资深编辑",
goal="确保文章质量和一致性",
backstory="""
你是一位资深编辑,有10年的编辑经验。
你注重文章的结构、逻辑和表达,追求完美。
你会仔细检查每一篇文章,确保其符合出版标准。
""",
verbose=True,
allow_delegation=True, # 允许委派给研究员或撰稿人
llm="gpt-4"
)
# ==================== 定义 Task ====================
# Task 1: 研究
research_task = Task(
description="""
研究 {topic} 领域的最新发展趋势。
需要关注:
1. 主要技术突破和创新
2. 市场动态和投资情况
3. 主要参与者和竞争格局
4. 未来3年的发展预测
提供详细的研究报告,包含数据和事实支撑。
""",
expected_output="""
一份结构化的研究报告,包含:
- 执行摘要(200字以内)
- 主要发现(5-7条)
- 详细分析(每个发现的数据支撑)
- 参考资料列表
""",
agent=researcher,
tools=[search_tool]
)
# Task 2: 写作
writing_task = Task(
description="""
根据研究报告,撰写一篇关于 {topic} 的深度文章。
要求:
1. 标题吸引人
2. 开篇引人入胜
3. 内容结构清晰
4. 技术深度适中
5. 结尾有洞察
文章长度:2000-3000字
""",
expected_output="""
一篇完整的技术文章,包含:
- 标题
- 导语
- 正文(分章节)
- 总结
""",
agent=writer,
context=[research_task] # 依赖研究任务
)
# Task 3: 编辑
editing_task = Task(
description="""
审校和优化文章,确保质量符合出版标准。
检查要点:
1. 逻辑连贯性
2. 表达准确性
3. 错别字和语法错误
4. 格式规范
如发现问题,可以委派给撰稿人修改。
""",
expected_output="""
最终版文章,包含:
- 修改说明
- 最终文章内容
- 质量评分(1-10)
""",
agent=editor,
context=[writing_task] # 依赖写作任务
)
# ==================== 创建 Crew ====================
content_crew = Crew(
agents=[researcher, writer, editor],
tasks=[research_task, writing_task, editing_task],
process=Process.sequential, # 顺序执行
verbose=True,
memory=True # 启用记忆,共享上下文
)
# ==================== 执行 ====================
def run_content_pipeline(topic: str):
"""运行内容创作流水线"""
result = content_crew.kickoff(inputs={"topic": topic})
print("\n" + "="*50)
print("最终文章:")
print("="*50)
print(result)
return result
if __name__ == "__main__":
run_content_pipeline("AI Agent 技术发展趋势")3.2 进阶示例:层次协作模式
"""
CrewAI 进阶示例:层次协作模式
展示 Manager Agent 自动协调任务
"""
from crewai import Agent, Task, Crew, Process
# 定义工作 Agent
agents = [
Agent(
role="数据分析师",
goal="分析数据并提供洞察",
backstory="专业的数据分析师,擅长Python和数据可视化",
tools=[python_repl_tool],
),
Agent(
role="报告撰写员",
goal="撰写专业的分析报告",
backstory="资深分析师,擅长撰写清晰的报告",
),
Agent(
role="质量控制员",
goal="确保输出质量符合标准",
backstory="有严格标准的质量专家",
),
]
# 定义任务(不指定 Agent,由 Manager 分配)
tasks = [
Task(
description="分析销售数据,找出增长最快的产品类别",
expected_output="包含数据洞察的分析报告",
),
Task(
description="根据分析结果撰写商业报告",
expected_output="结构化的商业报告文档",
),
Task(
description="审核报告质量,确保无错误",
expected_output="质量审核报告和最终版本",
),
]
# 创建层次协作 Crew
hierarchical_crew = Crew(
agents=agents,
tasks=tasks,
process=Process.hierarchical, # 层次执行
manager_llm="gpt-4", # Manager 使用的 LLM
verbose=True,
)
# 执行
result = hierarchical_crew.kickoff()3.3 工具集成
"""
CrewAI 工具集成示例
展示如何创建自定义工具
"""
from crewai_tools import tool
from typing import Optional
# 使用装饰器定义工具
@tool("Database Query Tool")
def query_database(query: str, database: Optional[str] = None) -> str:
"""
执行数据库查询
Args:
query: SQL 查询语句
database: 数据库名称(可选)
Returns:
查询结果
"""
# 实际实现中这里会连接真实数据库
mock_result = {
"sales": [
{"product": "A", "revenue": 10000},
{"product": "B", "revenue": 15000},
]
}
return str(mock_result)
@tool("Code Execution Tool")
def execute_code(code: str, language: str = "python") -> str:
"""
执行代码片段
Args:
code: 要执行的代码
language: 编程语言
Returns:
执行结果
"""
try:
# 实际实现中这里会使用安全的代码执行环境
exec_globals = {}
exec(code, exec_globals)
return "代码执行成功"
except Exception as e:
return f"执行错误: {str(e)}"
# 在 Agent 中使用工具
analyst = Agent(
role="数据分析师",
goal="分析数据并提供洞察",
backstory="专业的数据分析师",
tools=[query_database, execute_code], # 使用自定义工具
)四、适用场景
4.1 最佳适用场景
| 场景 | 描述 | 推荐配置 |
|---|---|---|
| 内容创作 | 研究、写作、编辑流水线 | Sequential + 3+ Agent |
| 市场研究 | 数据收集、分析、报告生成 | Hierarchical |
| 自动化工作流 | 多步骤业务流程 | Sequential |
| 代码审查 | 代码分析、改进建议、验证 | Sequential |
4.2 场景实现示例
┌─────────────────────────────────────────────────────────────┐
│ 典型应用场景 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 场景1: 新闻内容生产 │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 选题 │ ─→ │ 采访 │ ─→ │ 撰写 │ │
│ │ Agent │ │ Agent │ │ Agent │ │
│ └─────────┘ └─────────┘ └────┬────┘ │
│ │ │
│ ↓ │
│ ┌─────────────┐ │
│ │ 编辑发布 │ │
│ │ Agent │ │
│ └─────────────┘ │
│ │
│ 场景2: 数据分析报告 │
│ ┌──────────────────────────────────────────────────┐ │
│ │ Manager Agent │ │
│ │ (自动任务分配) │ │
│ └─────────────────────┬────────────────────────────┘ │
│ ┌──────────┼──────────┐ │
│ ↓ ↓ ↓ │
│ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │数据收集│ │数据分析│ │报告撰写│ │
│ │Agent │ │Agent │ │Agent │ │
│ └────────┘ └────────┘ └────────┘ │
│ │
│ 场景3: 代码审查流水线 │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 代码 │ ─→ │ 安全 │ ─→ │ 性能 │ │
│ │ 审查 │ │ 检查 │ │ 分析 │ │
│ │ Agent │ │ Agent │ │ Agent │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │ │ │ │
│ └──────────────┼──────────────┘ │
│ ↓ │
│ ┌─────────────┐ │
│ │ 综合报告 │ │
│ │ Agent │ │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘4.3 不适用场景
| 场景 | 原因 | 推荐替代 |
|---|---|---|
| 实时交互 | 响应延迟较高 | 单 Agent |
| 高度灵活对话 | 流程相对固定 | AutoGen |
| 大规模部署 | 资源消耗较大 | 自定义架构 |
五、优缺点分析
5.1 优点
| 优点 | 描述 |
|---|---|
| 角色清晰 | Agent 角色定义明确,职责清晰 |
| 流程可控 | Process 机制确保任务执行有序 |
| 易于理解 | 概念直观,学习曲线平缓 |
| 工具丰富 | 内置多种工具,支持自定义 |
| 记忆系统 | 支持短期和长期记忆 |
5.2 缺点
| 缺点 | 描述 | 缓解方案 |
|---|---|---|
| 灵活性有限 | 流程定义相对固定 | 使用 Hierarchical 模式 |
| 资源消耗 | 多 Agent 并行内存占用大 | 优化 Agent 数量 |
| 通信开销 | Agent 间通信有延迟 | 减少不必要的协作 |
六、面试问答
Q1: CrewAI 的核心概念是什么?
回答: CrewAI 的核心概念包括:
- Agent:具有特定角色和能力的智能体
- Task:需要完成的具体任务
- Crew:Agent 和 Task 的组织单位
- Process:任务执行流程(Sequential 或 Hierarchical)
Q2: Sequential 和 Hierarchical Process 有什么区别?
回答:
| 对比项 | Sequential | Hierarchical |
|---|---|---|
| 执行方式 | 按顺序执行 | Manager 自动分配 |
| 任务依赖 | 明确指定 | 自动识别 |
| 适用场景 | 流水线任务 | 复杂任务分解 |
| 灵活性 | 较低 | 较高 |
Q3: 如何选择使用 CrewAI 还是 AutoGen?
回答:
- 选择 CrewAI:角色明确、流程固定、需要严格控制
- 选择 AutoGen:需要灵活对话、代码执行、研究探索
Q4: CrewAI 如何处理 Agent 间的协作?
回答:
- 上下文传递:通过
context参数传递任务依赖 - 记忆系统:启用
memory=True共享上下文 - 任务委派:启用
allow_delegation=True允许委派 - 输出传递:前一任务输出自动传递给后一任务
七、小结
| 概念 | 一句话总结 | 面试关键词 |
|---|---|---|
| CrewAI | 基于角色扮演的多智能体协作框架 | Agent、Task、Crew、Process |
| Sequential | 任务按顺序依次执行 | 流水线、上下文传递 |
| Hierarchical | Manager Agent 自动协调 | 自动分配、灵活调度 |
| 适用场景 | 内容创作、工作流自动化 | 角色明确、流程固定 |
一句话总结:CrewAI 通过角色化的 Agent 设计和流程化的任务编排,实现了结构清晰的多智能体协作,非常适合内容创作和自动化工作流场景。
最后更新:2026年3月19日