Crewai
CrewAI 从零到进阶学习教程
1. CrewAI 是什么
CrewAI 是一个用于构建多 Agent 系统的 Python 框架。这里的 Agent 可以理解为一个“带角色、目标、背景和工具能力的 LLM 工作单元”。多个 Agent 可以组成 Crew,也就是一个协作团队,由 CrewAI 按照指定的 Process 执行一组 Task。
CrewAI 主要解决的问题是:当一个任务不是一次简单 LLM 调用能稳定完成时,如何把它拆成多个角色、多个步骤、多个工具调用,并让这些步骤形成可理解、可调试、可复用的协作流程。
典型问题包括:
- 自动调研:搜索资料、提炼观点、生成报告。
- 内容生产:选题、写作、润色、审核。
- 数据分析:读取数据、计算指标、解释结论。
- 业务自动化:把人工流程拆成多个 Agent 协作执行。
- 原型验证:快速搭建多 Agent 协作 demo。
2. 环境准备
2.1 Python 版本
CrewAI 官方安装文档要求 Python 版本满足:
python --version
推荐使用:
Python >= 3.10 and < 3.14
如果你是 Windows + PyCharm 用户,建议:
- 创建独立虚拟环境。
- 在项目根目录维护
.env。 - 把 API Key 放在环境变量或
.env中,不要硬编码到源码。
2.2 安装依赖
最小安装:
pip install crewai
如果需要使用搜索、文件读取、网站检索等工具:
pip install "crewai[tools]"
也可以使用 uv:
uv add crewai
uv add "crewai[tools]"
2.3 配置模型 API Key
CrewAI 最终仍然需要调用大模型。以 OpenAI 兼容模型为例:
$env:OPENAI_API_KEY="你的 API Key"
如果你使用搜索工具 SerperDevTool,还需要:
$env:SERPER_API_KEY="你的 Serper API Key"
在 Python 代码中,也可以通过 python-dotenv 加载 .env:
pip install python-dotenv
.env 示例:
OPENAI_API_KEY=你的 API Key
SERPER_API_KEY=你的 Serper API Key
3. 核心概念总览
CrewAI 的最小运行模型是:
Agent + Task + Crew + Process
一个典型执行流程如下:
3.1 一句话理解五个核心概念
| 概念 | 英文原文 | 一句话解释 |
|---|---|---|
| Agent | Agent | 带角色、目标、背景和工具能力的执行者 |
| Task | Task | 交给 Agent 完成的具体任务 |
| Tool | Tool | Agent 可调用的函数、搜索、文件、API 等能力 |
| Crew | Crew | 多个 Agent 和 Task 组成的协作团队 |
| Process | Process | Crew 执行 Task 的流程策略 |
4. Agent 详解
Agent 是 CrewAI 的核心执行单元。一个 Agent 通常包含:
| 参数 | 说明 |
|---|---|
role |
Agent 的角色,例如“市场研究员” |
goal |
Agent 的目标,例如“找出行业趋势” |
backstory |
背景设定,帮助模型稳定扮演角色 |
tools |
可选,Agent 可以调用的工具列表 |
verbose |
是否输出详细日志 |
allow_delegation |
是否允许委派任务,常用于层级式流程 |
示例:
from crewai import Agent
researcher = Agent(
role="AI 行业研究员",
goal="收集并总结 AI Agent 框架的关键趋势",
backstory="你是一名严谨的技术研究员,擅长从复杂资料中提炼结构化结论。",
verbose=True,
)
4.1 写好 Agent 的关键
好的 Agent 定义要回答三个问题:
- 它是谁?
- 它负责什么?
- 它应该用什么标准完成任务?
不好的写法:
Agent(
role="助手",
goal="帮我完成任务",
backstory="你很聪明",
)
更好的写法:
Agent(
role="技术文档编辑",
goal="把技术资料整理成适合初学者阅读的 Markdown 教程",
backstory="你长期编写开发者文档,擅长用清晰结构解释复杂概念。",
)
5. Task 详解
Task 是交给 Agent 的具体任务。常见参数包括:
| 参数 | 说明 |
|---|---|
description |
任务描述 |
expected_output |
预期输出格式和质量标准 |
agent |
执行任务的 Agent |
context |
可选,指定依赖哪些前置任务输出 |
output_file |
可选,把任务结果写入文件 |
示例:
from crewai import Task
research_task = Task(
description="调研 CrewAI 的核心概念,并总结 Agent、Task、Tool、Crew、Process 的区别。",
expected_output="一段结构清晰的中文说明,包含表格。",
agent=researcher,
)
5.1 description 和 expected_output 的区别
| 字段 | 重点 |
|---|---|
description |
让 Agent 知道要做什么 |
expected_output |
让 Agent 知道结果应该长什么样 |
建议写法:
Task(
description="分析 Python Agent 框架 CrewAI 的适用场景。",
expected_output="输出 5 个适用场景,每个场景包含:场景名称、为什么适合 CrewAI、潜在风险。",
agent=researcher,
)
6. Tool 详解
Tool 是 Agent 能调用的外部能力。没有 Tool 的 Agent 只能基于模型已有知识和上下文进行推理;有 Tool 的 Agent 可以搜索网页、读取文件、查询数据库或执行受控函数。
CrewAI 支持两类常见工具:
- 官方或社区提供的工具,例如
SerperDevTool、FileReadTool、DirectoryReadTool、WebsiteSearchTool。 - 自定义工具,例如继承
BaseTool或使用@tool装饰器。
6.1 工具安装
pip install "crewai[tools]"
6.2 自定义工具基本结构
from crewai.tools import tool
@tool("文本长度计算器")
def count_text_length(text: str) -> str:
"""计算输入文本的字符数量。"""
return f"输入文本共有 {len(text)} 个字符。"
注意
Tool 的 docstring 很重要。Agent 会根据工具名称、参数和说明判断什么时候调用它。
6.3 关于代码执行工具的变化
早期 CrewAI 资料中常见 CodeInterpreterTool 或 allow_code_execution=True。但在较新的官方工具文档中,CodeInterpreterTool 已被标注为从 crewai-tools 移除,allow_code_execution 和 code_execution_mode 也被标注为废弃方向,推荐使用 E2B、Modal 等专门沙箱服务进行安全隔离。
因此,初学阶段不建议直接教 Agent 执行任意 Python 代码。更稳妥的方式是:
- 把需要执行的逻辑封装成受控 Tool。
- Tool 输入使用 Pydantic 或函数签名约束。
- Tool 内部只执行白名单逻辑。
- 生产环境如需任意代码执行,接入独立沙箱服务。
7. Crew 详解
Crew 表示一组 Agent 和 Task 组成的协作团队。
常见参数:
| 参数 | 说明 |
|---|---|
agents |
Agent 列表 |
tasks |
Task 列表 |
process |
执行流程,默认顺序执行 |
verbose |
是否输出详细日志 |
manager_llm |
层级式流程中经理 Agent 使用的模型 |
最小结构:
from crewai import Crew, Process
crew = Crew(
agents=[researcher],
tasks=[research_task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff()
print(result)
8. Process 详解
Process 是 Crew 执行任务的流程策略。
CrewAI 当前常用流程包括:
| Process | 说明 | 适用场景 |
|---|---|---|
Process.sequential |
按任务列表顺序执行 | 线性流程、报告生成、内容生产 |
Process.hierarchical |
由经理模型或经理 Agent 分派和审核任务 | 复杂任务、多角色协作、需要管理者判断 |
8.1 顺序流程
顺序流程最容易理解:
调研任务 -> 分析任务 -> 写作任务 -> 审核任务
每个 Task 按列表顺序执行,后一个 Task 可以参考前一个 Task 的输出。
8.2 层级流程
层级流程更像团队管理:
经理 Agent 判断任务 -> 分派给合适 Agent -> 汇总和验证输出
使用 Process.hierarchical 时,需要配置 manager_llm 或 manager_agent。
9. 基础示例:第一个 Agent 和 Task
这个示例创建一个“学习教练 Agent”,让它生成 CrewAI 学习计划。
9.1 安装依赖
pip install crewai python-dotenv
9.2 准备 .env
OPENAI_API_KEY=你的 API Key
9.3 完整代码:hello_crewai.py
from dotenv import load_dotenv
from crewai import Agent, Crew, Process, Task
def main() -> None:
load_dotenv()
learning_coach = Agent(
role="Python Agent 学习教练",
goal="帮助初学者用清晰路线学习 CrewAI",
backstory=(
"你是一名耐心的技术教练,熟悉 Python、LLM 应用和多 Agent 框架,"
"擅长把复杂技术拆成可执行的学习步骤。"
),
verbose=True,
)
plan_task = Task(
description=(
"请为一个有 Java 后端经验、刚开始学习 Python Agent 框架的开发者,"
"设计一份 3 天 CrewAI 入门学习计划。"
),
expected_output=(
"输出 Markdown 格式的学习计划,包含每天的学习目标、练习任务和验收标准。"
),
agent=learning_coach,
)
crew = Crew(
agents=[learning_coach],
tasks=[plan_task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff()
print("\n===== 最终结果 =====")
print(result)
if __name__ == "__main__":
main()
9.4 运行方式
python hello_crewai.py
9.5 预期输出说明
你会看到 CrewAI 的详细执行日志,最后输出类似:
===== 最终结果 =====
# 3 天 CrewAI 入门学习计划
## 第 1 天:理解核心概念
...
具体措辞由模型生成,不会每次完全一致。
10. 进阶示例一:多 Agent 顺序协作
这个示例模拟一个小型内容团队:
- 研究员负责收集观点。
- 分析师负责提炼结构。
- 作者负责写最终文章。
10.1 完整代码:multi_agent_report.py
from dotenv import load_dotenv
from crewai import Agent, Crew, Process, Task
def main() -> None:
load_dotenv()
researcher = Agent(
role="技术研究员",
goal="找出 CrewAI 在多 Agent 协作中的核心价值",
backstory="你擅长阅读技术资料,并把信息整理成准确、紧凑的要点。",
verbose=True,
)
analyst = Agent(
role="技术分析师",
goal="把研究材料转化为结构化分析框架",
backstory="你擅长比较技术方案,能从适用场景、优点和风险三个维度分析问题。",
verbose=True,
)
writer = Agent(
role="技术文章作者",
goal="写出适合初学者阅读的中文技术文章",
backstory="你长期撰写开发者教程,语言清晰,例子具体,不堆砌概念。",
verbose=True,
)
research_task = Task(
description=(
"总结 CrewAI 的核心概念,包括 Agent、Task、Tool、Crew、Process。"
"重点说明它们分别解决什么问题。"
),
expected_output="一份结构化研究笔记,包含 5 个核心概念的解释。",
agent=researcher,
)
analysis_task = Task(
description=(
"基于研究员的输出,分析 CrewAI 适合和不适合的场景。"
"请用表格输出。"
),
expected_output="一个 Markdown 表格,包含场景、是否适合、原因、风险。",
agent=analyst,
context=[research_task],
)
writing_task = Task(
description=(
"基于前两个任务的结果,写一篇 800 字以内的中文入门文章,"
"标题为《为什么要学习 CrewAI》。"
),
expected_output="一篇结构清晰的 Markdown 中文文章。",
agent=writer,
context=[research_task, analysis_task],
)
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff()
print("\n===== 最终文章 =====")
print(result)
if __name__ == "__main__":
main()
11. 进阶示例二:搜索工具参与自动调研
这个示例使用 SerperDevTool 进行网页搜索,让 Agent 调研某个主题并生成摘要。
11.1 准备依赖和环境变量
pip install "crewai[tools]" python-dotenv
.env:
OPENAI_API_KEY=你的 API Key
SERPER_API_KEY=你的 Serper API Key
11.2 完整代码:research_with_search.py
from dotenv import load_dotenv
from crewai import Agent, Crew, Process, Task
from crewai_tools import SerperDevTool
def main() -> None:
load_dotenv()
search_tool = SerperDevTool()
researcher = Agent(
role="AI 新闻研究员",
goal="搜索并总结指定主题的最新公开信息",
backstory=(
"你是一名谨慎的研究员,擅长使用搜索工具查找资料,"
"并把来源、事实和推断分开。"
),
tools=[search_tool],
verbose=True,
)
writer = Agent(
role="研究报告编辑",
goal="把研究员找到的信息整理成清晰的中文报告",
backstory="你擅长把零散信息改写成结构化报告,并明确标注不确定性。",
verbose=True,
)
research_task = Task(
description=(
"请搜索 CrewAI、LangGraph、AutoGen 在多 Agent 框架方向的公开资料,"
"总结它们各自的定位。请优先关注官方文档或项目主页。"
),
expected_output=(
"输出中文研究笔记,包含每个框架的定位、适合场景和至少 3 条关键信息。"
),
agent=researcher,
)
report_task = Task(
description=(
"基于研究结果,写一份中文对比报告。要求包含:"
"1. 三个框架的定位;2. 初学者学习建议;3. 风险和注意事项。"
),
expected_output="一份 Markdown 格式中文报告,包含标题、表格和结论。",
agent=writer,
context=[research_task],
)
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, report_task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff()
print("\n===== 调研报告 =====")
print(result)
if __name__ == "__main__":
main()
12. 进阶示例三:受控代码执行工具
很多 Agent 教程会直接让模型执行任意 Python 代码,但这在安全上风险很高。更推荐的方式是:把你允许执行的逻辑封装成受控 Tool。
下面示例中,我们创建一个“数字统计工具”。Agent 不能执行任意系统命令,只能把一组数字传给 Tool,由 Tool 计算平均值、最大值、最小值。
12.1 安装依赖
pip install crewai python-dotenv
12.2 完整代码:controlled_code_tool.py
from statistics import mean
from dotenv import load_dotenv
from crewai import Agent, Crew, Process, Task
from crewai.tools import tool
@tool("数字统计工具")
def calculate_number_stats(numbers: str) -> str:
"""
计算一组数字的统计信息。
输入格式示例:10, 20, 30, 40
"""
values = [float(item.strip()) for item in numbers.split(",") if item.strip()]
if not values:
return "没有可计算的数字。"
return (
f"数量: {len(values)}\n"
f"平均值: {mean(values):.2f}\n"
f"最小值: {min(values):.2f}\n"
f"最大值: {max(values):.2f}"
)
def main() -> None:
load_dotenv()
data_analyst = Agent(
role="数据分析师",
goal="使用受控工具完成基础数据统计,并解释结果",
backstory=(
"你是一名谨慎的数据分析师。你不会编造计算结果,"
"需要计算时必须调用提供的数字统计工具。"
),
tools=[calculate_number_stats],
verbose=True,
)
analysis_task = Task(
description=(
"请分析这组数据:18, 22, 35, 40, 41, 60。"
"你必须先使用数字统计工具完成计算,然后用中文解释这组数据的特点。"
),
expected_output=(
"输出包含两部分:1. 工具计算结果;2. 用 3 条要点解释数据特点。"
),
agent=data_analyst,
)
crew = Crew(
agents=[data_analyst],
tasks=[analysis_task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff()
print("\n===== 数据分析结果 =====")
print(result)
if __name__ == "__main__":
main()
13. 进阶示例四:层级式协作
层级式协作使用 Process.hierarchical,由经理模型或经理 Agent 负责分派任务、协调 Agent、验证结果。
适合场景:
- 任务拆解不固定。
- 需要一个管理者判断谁来执行。
- 多个 Agent 能力差异明显。
- 任务结果需要统一审核。
13.1 完整代码:hierarchical_crew.py
from dotenv import load_dotenv
from crewai import Agent, Crew, Process, Task
def main() -> None:
load_dotenv()
market_researcher = Agent(
role="市场研究员",
goal="分析目标用户和市场机会",
backstory="你擅长调研用户需求、市场趋势和竞品差异。",
verbose=True,
)
product_designer = Agent(
role="产品设计师",
goal="把市场信息转化为产品功能方案",
backstory="你擅长设计 MVP 功能,并能控制产品复杂度。",
verbose=True,
)
technical_writer = Agent(
role="技术文档作者",
goal="把方案整理成开发团队可读的 Markdown 文档",
backstory="你擅长输出清晰的技术方案文档,结构完整、语言准确。",
verbose=True,
)
project_task = Task(
description=(
"我们计划做一个面向个人开发者的 AI Agent 学习助手。"
"请完成一次小型产品立项分析,内容包括:"
"目标用户、核心痛点、MVP 功能、技术风险、下一步建议。"
),
expected_output=(
"一份 Markdown 格式产品立项简报,包含标题、表格和明确建议。"
),
)
crew = Crew(
agents=[market_researcher, product_designer, technical_writer],
tasks=[project_task],
process=Process.hierarchical,
manager_llm="gpt-4o-mini",
verbose=True,
)
result = crew.kickoff()
print("\n===== 产品立项简报 =====")
print(result)
if __name__ == "__main__":
main()
13.2 运行方式
python hierarchical_crew.py
14. 实际应用场景
14.1 自动调研
流程:
搜索 Agent -> 资料整理 Agent -> 分析 Agent -> 报告 Agent
适合任务:
- 行业趋势报告。
- 竞品调研。
- 技术选型对比。
- 开源项目调研。
优势:
- 可以把“查资料”和“写报告”拆开。
- 可以要求研究 Agent 标注来源,写作 Agent 只负责结构化表达。
风险:
- 搜索结果可能过时或噪声很大。
- Agent 可能混淆事实和推断。
- 需要人工审核关键结论。
14.2 内容生成
流程:
选题 Agent -> 大纲 Agent -> 写作 Agent -> 审核 Agent
适合任务:
- 博客文章。
- 产品文案。
- 视频脚本。
- 公众号草稿。
注意事项:
- 给
expected_output明确格式。 - 审核 Agent 要检查事实、语气和结构。
- 不要让一个 Agent 同时负责选题、搜索、写作、审核。
14.3 数据分析
流程:
数据读取 Tool -> 分析 Agent -> 解释 Agent -> 报告 Agent
适合任务:
- CSV 指标分析。
- 运营数据摘要。
- 错误日志归因。
- 实验结果解释。
建议:
- 计算逻辑封装成 Tool。
- Agent 负责解释,不负责凭空算数。
- 重要指标由确定性代码计算。
14.4 开发辅助
流程:
需求分析 Agent -> 方案 Agent -> 代码审查 Agent -> 文档 Agent
适合任务:
- 生成接口设计草案。
- 整理重构方案。
- 写测试计划。
- 生成 README。
注意事项:
- 不要让 Agent 未经约束地修改大量代码。
- 代码执行必须隔离。
- 生产代码要经过测试、review 和人工确认。
15. 常见问题与调试技巧
15.1 为什么 Agent 输出不稳定
常见原因:
role太模糊。goal没有验收标准。expected_output没有格式约束。- Task 太大,一个 Agent 同时承担多个职责。
改进方式:
expected_output=(
"输出 Markdown 表格,列名固定为:概念、作用、例子、注意事项。"
"每行不超过 80 个中文字符。"
)
15.2 为什么工具没有被调用
常见原因:
- 工具名称不清楚。
- 工具 docstring 没说明输入格式。
- Task 没明确要求必须调用工具。
- Agent 认为自己可以直接回答。
改进方式:
description=(
"你必须先调用数字统计工具计算结果,再解释数据。"
"如果没有工具结果,不允许直接给出结论。"
)
15.3 为什么搜索工具报错
检查:
- 是否安装了
crewai[tools]。 - 是否设置了
SERPER_API_KEY。 - 当前网络是否能访问搜索服务。
- API Key 是否额度不足或权限错误。
15.4 为什么层级流程报错
使用 Process.hierarchical 时,通常需要设置:
manager_llm="gpt-4o-mini"
或者提供自定义 manager_agent。
如果没有 manager,CrewAI 无法知道谁来分派任务。
15.5 如何查看执行过程
把 Agent 和 Crew 的 verbose 打开:
agent = Agent(..., verbose=True)
crew = Crew(..., verbose=True)
这样可以看到任务执行、工具调用和中间推理过程。
15.6 如何减少幻觉
建议:
- 需要事实时,给 Agent 搜索工具或知识库工具。
- 在 Task 里要求区分“事实”和“推断”。
- 对关键输出增加审核 Agent。
- 对数值计算使用确定性 Tool。
- 输出中要求列出来源或依据。
示例:
expected_output=(
"请分成三部分输出:事实、推断、仍需人工确认的信息。"
)
15.7 如何控制任务复杂度
一个 Task 只做一类事情:
不推荐:
搜索资料、分析竞品、生成报告、写推广文案、制定开发计划。
推荐拆分:
Task 1: 搜索资料
Task 2: 提炼观点
Task 3: 生成报告
Task 4: 审核报告
15.8 如何在代码中传入变量
Task 描述中可以写占位符:
task = Task(
description="请调研主题:{topic}",
expected_output="输出中文摘要。",
agent=researcher,
)
启动时传入:
result = crew.kickoff(inputs={"topic": "CrewAI 多 Agent 框架"})
完整示例:
from dotenv import load_dotenv
from crewai import Agent, Crew, Process, Task
def main() -> None:
load_dotenv()
researcher = Agent(
role="主题研究员",
goal="围绕用户输入主题生成结构化摘要",
backstory="你擅长快速理解主题,并生成适合初学者阅读的说明。",
verbose=True,
)
task = Task(
description="请解释这个主题:{topic}",
expected_output="输出 Markdown 格式说明,包含定义、用途和一个例子。",
agent=researcher,
)
crew = Crew(
agents=[researcher],
tasks=[task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff(inputs={"topic": "CrewAI 的 kickoff(inputs=...) 参数"})
print(result)
if __name__ == "__main__":
main()
预期输出:
# CrewAI 的 kickoff(inputs=...) 参数
## 定义
...
15.9 如何把输出写入文件
可以在 Task 中设置 output_file:
from dotenv import load_dotenv
from crewai import Agent, Crew, Process, Task
def main() -> None:
load_dotenv()
writer = Agent(
role="Markdown 文档作者",
goal="生成结构清晰的 Markdown 文档",
backstory="你擅长编写开发者学习笔记。",
verbose=True,
)
task = Task(
description="写一份 CrewAI 入门说明,控制在 500 字以内。",
expected_output="Markdown 格式文档。",
agent=writer,
output_file="output/crewai_intro.md",
)
crew = Crew(
agents=[writer],
tasks=[task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff()
print(result)
if __name__ == "__main__":
main()
运行后会生成:
output/crewai_intro.md
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐

所有评论(0)