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 用户,建议:

  1. 创建独立虚拟环境。
  2. 在项目根目录维护 .env
  3. 把 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

一个典型执行流程如下:

定义 Agent: 角色、目标、背景、工具

定义 Task: 描述、预期输出、执行者

创建 Crew: agents + tasks + process

调用 crew.kickoff()

LLM 推理和工具调用

返回 CrewOutput

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 定义要回答三个问题:

  1. 它是谁?
  2. 它负责什么?
  3. 它应该用什么标准完成任务?

不好的写法:

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 descriptionexpected_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 支持两类常见工具:

  1. 官方或社区提供的工具,例如 SerperDevToolFileReadToolDirectoryReadToolWebsiteSearchTool
  2. 自定义工具,例如继承 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 资料中常见 CodeInterpreterToolallow_code_execution=True。但在较新的官方工具文档中,CodeInterpreterTool 已被标注为从 crewai-tools 移除,allow_code_executioncode_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_llmmanager_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 为什么搜索工具报错

检查:

  1. 是否安装了 crewai[tools]
  2. 是否设置了 SERPER_API_KEY
  3. 当前网络是否能访问搜索服务。
  4. 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
Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐