CrewAI 是一个专为多 Agent 协作场景设计的 AI 框架，它通过"角色扮演"和"任务编排"的抽象，让多个 LLM Agent 能够像一支团队一样协同工作，共同完成复杂任务。CrewAI 的设计哲学强调结构化协作和可复用的 Agent 角色，使得从简单的双 Agent 对话到大规模多 Agent 系统的构建都变得简洁而可控。
CrewAI 的核心抽象层围绕三个基本构件展开：Agent（角色）、Task（任务）、Crew（团队）。理解这三个概念的关系是掌握框架的关键。
Agent 是 CrewAI 中的基本执行单元，代表一个拥有特定角色、技能和行为模式的 AI 实体。每个 Agent 的配置包含以下核心属性：
Agent 的配置本质上是在为 LLM 构建一个系统提示词（System Prompt），通过角色设定来引导模型的行为范式。
from crewai import Agent
analyst = Agent(
role="高级数据分析师",
goal="从原始数据中提取关键洞察并生成可视化报告",
backstory="你是一位拥有 10 年经验的数据分析师，擅长发现数据中的隐藏模式",
llm="gpt-4",
tools=[search_tool, database_tool],
verbose=True,
allow_delegation=True,
memory=True
)
Task 定义了 Agent 需要完成的具体工作单元。任务可以包含输入、输出规格、上下文依赖和执行约束：
from crewai import Task
analysis_task = Task(
description="""
分析用户行为数据集，重点考察：
1. 日活跃用户(DAU)趋势
2. 功能使用频率分布
3. 用户流失的关键指标
4. 按用户群组的留存率对比
数据源：data/user_events_2024.db
输出格式：Markdown 报告 + CSV 汇总表
""",
expected_output="一份包含趋势图表描述、关键指标表格和策略建议的分析报告",
agent=analyst,
async_execution=False
)
Crew 是任务编排的最高层次，它将 Agent 和 Task 组织为一个完整的协作工作流。Crew 的配置决定了 Agent 之间的交互方式和任务执行流程：
from crewai import Crew, Process
crew = Crew(
agents=[analyst, writer, reviewer],
tasks=[collect_data, analyze_data, write_report, review_report],
process=Process.sequential,
verbose=True,
memory=True
)
result = crew.kickoff()
CrewAI 提供了三种核心的协作流程模式，分别适用于不同复杂度的任务场景。
顺序流程是最基本的执行模式，Task 按照预定义的顺序依次执行，每个 Task 的输出可以传递给下一个 Task 作为输入。这种模式适用于有明确步骤依赖关系的任务管线。
典型场景：文档生成管线——先收集资料，再撰写初稿，然后审阅修改，最后格式化输出。
执行特性：
# 顺序流程：研究→撰写→审阅
research_task = Task(description="收集 AI Agent 领域的最新论文", agent=researcher)
write_task = Task(description="基于研究结果撰写综述文章", agent=writer, context=[research_task])
review_task = Task(description="审阅文章，检查事实准确性", agent=reviewer, context=[write_task])
crew = Crew(
agents=[researcher, writer, reviewer],
tasks=[research_task, write_task, review_task],
process=Process.sequential
)
层级流程引入了一个"管理 Agent"（Manager Agent）来协调工作。管理 Agent 负责任务拆分、分配给合适的 Team Agent、监督执行进度和处理异常。这种模式更接近真实团队的管理结构。
执行特性：
# 层级流程：Manager 协调多个 Specialist
crew = Crew(
agents=[data_scientist, ml_engineer, devops_engineer],
tasks=[deploy_model, monitor_performance, optimize_pipeline],
process=Process.hierarchical,
manager_llm="gpt-4-turbo",
verbose=True
)
共识流程允许 Agent 之间通过协商来达成最终输出。在这种模式下，Agent 之间可以进行多轮对话，交换意见、修正错误，直到达成共识。
执行特性：
典型用例：
CrewAI 的工具系统是 Agent 与环境交互的接口。工具可以是简单的函数、API 封装器，也可以是自定义的复杂操作。
CrewAI 提供了一组开箱即用的工具：
| 工具名 | 功能 | 典型场景 |
|---|---|---|
| SerperDevTool | Google 搜索集成 | 实时信息检索 |
| ScrapeWebsiteTool | 网页内容抓取 | 数据采集 |
| FileReadTool | 本地文件读取 | 数据处理 |
| FileWriteTool | 文件写入 | 结果持久化 |
| PDFSearchTool | PDF 内容搜索 | 文档分析 |
| DOCXSearchTool | Word 文档搜索 | 办公处理 |
| CSVSearchTool | CSV 搜索 | 表格数据处理 |
| JSONSearchTool | JSON 搜索 | 结构化数据处理 |
| MySQLTool | MySQL 数据库查询 | 数据库操作 |
| CodeInterpreterTool | 代码执行沙箱 | 数据分析 |
from crewai.tools import BaseTool
class MarketResearchTool(BaseTool):
name: str = "市场调研工具"
description: str = "查询指定行业的最新市场规模和增长数据"
def _run(self, industry: str) -> str:
# 实现市场数据查询逻辑
data = query_market_database(industry)
return format_market_report(data)
# 在 Agent 中使用
analyst = Agent(
role="市场分析师",
tools=[MarketResearchTool(), SerperDevTool()]
)
CrewAI 的记忆系统为 Agent 提供了跨任务的上下文保持能力，分为三个层次：
每个 Agent 在单次 Crew 执行过程中维护当前会话的上下文，用于保持对话连贯性和跟踪任务进度。短期记忆在 Crew 执行结束后自动清除。
Agent 可以将重要的学习结果持久化存储，供未来的 Crew 执行引用。长期记忆使用嵌入向量（Embedding）进行语义索引和检索。
记录与特定用户相关的偏好、历史交互和上下文，允许 Agent 在多次交互中保持一致的用户体验。
memory_agent = Agent(
role="客户支持专家",
goal="根据用户的偏好和历史记录提供个性化支持",
memory=True,
# 默认使用 OpenAI Embeddings，可配置
embedder={
"provider": "openai",
"config": {"model": "text-embedding-3-small"}
}
)
Agent 可以将自己不擅长或无法独立完成的任务委派给团队中的其他 Agent。委派机制通过 allow_delegation=True 启用。
team_lead = Agent(
role="项目经理",
goal="协调团队完成项目交付",
allow_delegation=True, # 允许委派
verbose=True
)
developer = Agent(
role="全栈开发工程师",
goal="编写高质量的代码",
tools=[code_search_tool, code_review_tool]
)
# 在层级流程中，Manager 会自动进行委派
委派的工作机制：
Task 可以明确引用其他 Task 的输出，这在创建依赖链时非常有用：
task1 = Task(description="生成产品需求文档", agent=pm)
task2 = Task(
description="基于 PRD 设计系统架构",
agent=architect,
context=[task1] # 显式引用 task1 的输出
)
task3 = Task(
description="实现核心模块",
agent=developer,
context=[task1, task2] # 多任务依赖
)
系统支持在任务和步骤级别注册回调，用于监控、日志记录和外部集成：
def task_completed_callback(task_output):
"""任务完成时的通知回调"""
log_to_database(task_output)
send_notification(f"Task {task_output.task_id} completed")
def step_started_callback(step_info):
"""每个步骤开始前触发"""
print(f"Starting step: {step_info['description']}")
analysis = Task(
description="执行数据分析",
agent=data_scientist,
callback=task_completed_callback
)
crew = Crew(
agents=[data_scientist],
tasks=[analysis],
step_callback=step_started_callback
)
下面是一个完整的实战案例，展示如何使用 CrewAI 构建一个自动化市场研究报告生成系统：
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
# 工具
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
# Agent 定义
researcher = Agent(
role="行业研究员",
goal="收集并整理指定行业的最新信息",
backstory="你是一名资深的行业研究员，擅长从海量信息中提取有价值的内容",
tools=[search_tool, scrape_tool],
verbose=True
)
analyst = Agent(
role="数据分析师",
goal="对收集到的数据进行定量和定性分析",
backstory="你擅长将原始数据转化为可执行的商业洞察",
tools=[search_tool],
allow_delegation=True,
verbose=True
)
writer = Agent(
role="报告撰写人",
goal="将分析结论转化为结构化的专业报告",
backstory="你是一名专业的商业报告撰写者，文笔简练、逻辑清晰",
verbose=True
)
# Task 定义
research_task = Task(
description="调研 AI Agent 行业市场规模，包括：
"
"1. 2024-2025 年全球 AI Agent 市场规模
"
"2. 主要玩家（OpenAI、Anthropic、Google、Meta）的战略布局
"
"3. 行业关键趋势与增长驱动因素
"
"4. 地域分布：北美、中国、欧洲的市场差异",
expected_output="一份全面的行业调研汇总，包含关键数据和结论摘要",
agent=researcher
)
analysis_task = Task(
description="基于研究数据进行分析：
"
"1. 市场规模预测与增长率计算
"
"2. 竞争格局分析（波特五力模型）
"
"3. SWOT 分析
"
"4. 关键风险和机会",
expected_output="包含数据分析、图表描述和战略洞察的分析报告",
agent=analyst,
context=[research_task]
)
report_task = Task(
description="将调研和分析结果整理为正式报告，包含：
"
"1. 执行摘要
"
"2. 行业概览
"
"3. 竞争分析
"
"4. 趋势展望
"
"5. 战略建议",
expected_output="一份 3000 字以上的专业市场研究报告（Markdown 格式）",
agent=writer,
context=[research_task, analysis_task]
)
# Crew 执行
market_research_crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, report_task],
process=Process.sequential,
verbose=True,
memory=True
)
result = market_research_crew.kickoff()
print(result)
| 维度 | CrewAI | AutoGen |
|---|---|---|
| 抽象层次 | 角色/任务/团队 | Agent/对话/会话 |
| 协作模式 | 顺序/层级/共识 | 对话式/群组对话 |
| 可扩展性 | 中等（适合中小规模团队） | 高（支持大规模 Agent） |
| 学习曲线 | 低（API 简洁易懂） | 中（概念较多） |
| 调试 | 内建 verbose 和回调 | 对话日志 |
| 代码生成 | 适合 | 更适合 |
AutoGen 更擅长处理开放式的对话协作场景，适合复杂的、需要 Agent 间深入交互的任务。CrewAI 则更适合结构化的工作流，任务编排更加明确可控。
| 维度 | CrewAI | LangGraph |
|---|---|---|
| 设计哲学 | 高层抽象，快速开发 | 底层控制，灵活编排 |
| 状态管理 | 内建的 Agent/任务状态 | 显式 StateGraph |
| 流程控制 | 预定义 Process | 自定义图结构 |
| 集成度 | 内建工具和记忆 | 丰富但复杂 |
LangGraph 提供了更细粒度的流程控制，适合需要自定义状态转换和复杂分支逻辑的场景。CrewAI 以"少代码"为目标，适合快速搭建标准的多 Agent 协作系统。
| 维度 | CrewAI | Semantic Kernel |
|---|---|---|
| 语言 | Python | Python、C#、Java |
| 架构 | Agent 为中心 | 插件/技能/记忆 |
| 目标 | 多 Agent 协作 | 企业级 AI 集成 |
| 成熟度 | 快速发展中 | 更成熟 |
Semantic Kernel 更适合企业级场景，需要集成现有系统基础设施。CrewAI 则专注于 Agent 协作本身。
from crewai import Task
robust_task = Task(
description="执行关键数据分析",
agent=analyst,
max_retries=3, # 失败重试次数
expected_output="结果数组",
# 在 callback 中处理错误
callback=lambda output: handle_error(output) if output.is_error else None
)
# 使用异步任务加速
task_a = Task(description="搜索文献", agent=researcher, async_execution=True)
task_b = Task(description="搜索专利", agent=researcher, async_execution=True)
task_c = Task(description="合并分析", agent=analyst, context=[task_a, task_b])
CrewAI 在 2024 年下半年以来经历了快速发展：
CrewAI 通过简洁的 Agent/Task/Crew 三层抽象，降低了多 Agent 系统的构建门槛。它的设计特别适合以下场景：
相比 AutoGen 和 LangGraph，CrewAI 的优势在于 API 简洁、直观，开发效率高。对于需要快速搭建结构化多 Agent 协作系统的场景，CrewAI 是一个出色的选择。
此页面为 AI 知识体系 的一部分，内容持续更新中。