1 快速开始
1.1 框架介绍
01.框架定位
a.核心理念
a.功能说明
CrewAI是一个专为编排和自动化多智能体协作而设计的开源框架。其核心理念是通过结合两大核心抽象——Flows（流程）和Crews（团队），来构建能够处理复杂任务、具备生产级别稳定性的AI应用。Flows作为应用程序的结构化骨干，负责定义事件驱动的工作流、管理状态和控制执行逻辑；而Crews则作为工作单元，由多个自主的AI智能体组成，负责在Flow的指导下协同完成具体、复杂的任务。
b.代码示例
---
# 概念示例：Flows定义工作流程，Crews执行具体任务
from crewai import Flow, Crew
# 1. 定义一个工作流 (Flow)
financial_analysis_flow = Flow(
name="季度财报分析流程",
description="自动完成季度财报的数据提取、分析和报告生成。"
)
# 2. 在Flow中定义一个步骤，该步骤委派给一个Crew执行
@financial_analysis_flow.task
def analyze_earnings_call(company: str):
"""分析公司的财报电话会议记录。"""
# 创建并启动一个专门用于分析的Crew
analysis_crew = Crew(...)
result = analysis_crew.kickoff(inputs={"company": company})
return result
---
b.技术架构
a.功能说明
CrewAI的架构设计旨在实现自主性与控制之间的平衡。Flows作为顶层控制器，如同应用程序的“经理”，定义了任务的步骤、逻辑分支以及数据如何在系统中流转。Crews则是执行层，是完成具体工作的“团队”。当Flow遇到一个需要创造性协作和深度分析的复杂问题时，便会将该任务委派给一个专门的Crew。Crew内部的智能体通过角色扮演、自主协作和任务委派来共同解决问题，并将最终结果返回给Flow，由Flow决定下一步的走向。
b.代码示例
---
# 架构协同工作的概念示例
# Flow层面：管理状态和控制流程
class ApplicationFlow:
def handle_request(self, user_request):
# 1. Flow接收请求并管理状态
state = self.manage_state(user_request)
# 2. Flow将复杂任务委派给Crew
if state["needs_deep_analysis"]:
research_crew = self.create_research_crew()
analysis_result = research_crew.kickoff(inputs={"topic": state["topic"]})
state["analysis"] = analysis_result
# 3. Flow根据Crew的结果继续执行
final_report = self.generate_report(state)
return final_report
---
02.核心特性
a.自主运行
a.功能说明
CrewAI的智能体被设计为能够根据其角色、目标和可用工具自主做出决策。它们可以独立地规划行动、执行任务，并在需要时与其他智能体进行交互，而无需持续的人工干预。这种自主性是实现复杂任务自动化的关键。
b.代码示例
---
# Agent自主决策的概念示例
# 在一个研究任务中，研究员Agent可能会自主决定：
# 1. 使用搜索工具查找初始资料
# 2. 分析搜索结果，识别关键信息
# 3. 发现信息不足时，再次使用搜索工具进行更精确的查询
# 4. 整理所有信息，形成初步报告
---
b.自然交互
a.功能说明
智能体之间的协作通过类似人类团队成员的自然语言进行沟通。它们可以分享信息、委派子任务、请求帮助或提供反馈。这种基于对话的交互模式使得整个协作过程更加透明和易于理解。
b.代码示例
---
# 智能体交互对话示例
# Researcher Agent: "我已经收集了关于AI市场规模的初步数据，但需要更详细的财务分析。@Financial_Analyst，你能帮忙分析一下这些数据吗？"
# Financial_Analyst Agent: "收到。正在分析数据... 分析完成。数据显示，AI市场预计在未来五年内复合年增长率为25%。"
---
c.可扩展设计
a.功能说明
CrewAI的框架设计具有高度的可扩展性。开发者可以轻松地创建和集成自定义的工具（Tools），为智能体赋予新的能力。此外，也可以方便地定义新的智能体角色和任务类型，以适应各种不同的应用场景。
b.代码示例
---
from crewai_tools import BaseTool
# 创建一个自定义工具：读取本地文件
class MyFileReadTool(BaseTool):
name: str = "文件读取工具"
description: str = "读取指定路径的本地文件内容。"
def _run(self, file_path: str) -> str:
with open(file_path, 'r') as f:
return f.read()
# 在创建Agent时，将这个自定义工具加入其工具列表
custom_agent = Agent(
...,
tools=[MyFileReadTool()]
)
---
1.2 安装配置
01.环境准备
a.安装CrewAI
a.功能说明
安装CrewAI及其相关依赖是开始使用的第一步。官方推荐使用`pip`进行安装。为了使用CrewAI提供的命令行工具（CLI），例如快速创建项目脚手架，需要一并安装`crewai[tools]`。这个工具包包含了运行CLI所需的所有依赖。
b.代码示例
---
# 推荐使用虚拟环境以避免包冲突
python -m venv crewai-env
source crewai-env/bin/activate # Linux/macOS
# .\crewai-env\Scripts\activate # Windows
# 使用pip安装CrewAI和其工具包
pip install 'crewai[tools]'
# 验证安装
crewai --version
---
b.设置环境变量
a.功能说明
CrewAI的智能体通常需要与大型语言模型（LLM）进行交互，并且可能需要使用外部工具（如搜索引擎）。因此，必须在环境中配置相关的API密钥。这些密钥应保存在项目根目录下的`.env`文件中，CrewAI会自动加载这些变量。最基础的配置是LLM的API密钥，例如OpenAI。如果使用了需要认证的工具（如SerperDevTool），也需要一并设置。
b.代码示例
---
# 在项目根目录下创建 .env 文件
# .env 文件内容示例
# OpenAI API 密钥
OPENAI_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
# 如果使用Serper进行网络搜索，需要Serper API密钥
SERPER_API_KEY="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
# 如果使用其他LLM提供商，如Google Gemini
# GEMINI_API_KEY="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
---
02.项目结构
a.使用CLI创建项目
a.功能说明
CrewAI CLI提供了一个便捷的命令`crewai create`来快速生成一个结构化的项目模板。这个模板包含了预设的目录结构、配置文件（`agents.yaml`, `tasks.yaml`）和主程序入口，极大地简化了新项目的启动过程。开发者可以直接在这些模板文件的基础上进行修改，以适应自己的业务需求。
b.代码示例
---
# 在终端中运行以下命令创建一个名为 market-analysis-crew 的新项目
crewai create market-analysis-crew
# 命令执行后会生成如下目录结构：
# market-analysis-crew/
# ├── src/
# │ └── market_analysis_crew/
# │ ├── __init__.py
# │ ├── agents.py
# │ ├── crew.py
# │ ├── main.py
# │ └── tasks.py
# ├── config/
# │ ├── agents.yaml
# │ └── tasks.yaml
# └── .env
---
b.手动创建项目
a.功能说明
如果不使用CLI，也可以手动创建项目结构。核心是创建一个主Python文件（例如`main.py`），在其中定义Agent、Task和Crew。这种方式更加灵活，适合快速原型验证或集成到现有项目中。所有定义都在一个文件中，便于理解和调试。
b.代码示例
---
# 手动创建的项目结构可以非常简单
# my-simple-crew/
# ├── main.py
# └── .env
# main.py 文件内容
from crewai import Agent, Task, Crew, Process
import os
# os.environ['OPENAI_API_KEY'] = "YOUR_API_KEY"
# ... 在此处定义Agent, Task, Crew ...
# result = crew.kickoff()
# print(result)
---
1.3 快速开始
01.定义Agent
a.创建研究员Agent
a.功能说明
首先，我们定义一个“研究员”智能体。这个智能体的目标是围绕特定主题进行深入研究。我们为其分配“高级研究分析师”的角色，并提供一个背景故事来强化其专家身份。为了让它能够获取外部信息，我们为其配备了`SerperDevTool`，这是一个用于网络搜索的工具。
b.代码示例
---
from crewai import Agent
from crewai_tools import SerperDevTool
# 创建一个搜索工具实例
search_tool = SerperDevTool()
# 创建一个研究员智能体
researcher = Agent(
role=\'高级研究分析师\',
goal=\'发现关于 {topic} 的突破性见解\',
backstory=(
"作为一名传奇的研究分析师，你以能够从海量数据中挖掘出隐藏的叙事和趋势而闻名。"
"你擅长拆解复杂的数据，并以清晰、可操作的方式呈现出来。"
),
verbose=True,
allow_delegation=False, # 该Agent不委派任务给其他Agent
tools=[search_tool]
)
---
b.创建作家Agent
a.功能说明
接下来，我们定义一个“技术内容策略师”智能体。它的任务是基于研究员找到的资料，撰写一篇引人入胜的博客文章。我们为其设定的目标是创作有趣、易于理解且吸引人的内容。
b.代码示例
---
# 创建一个作家智能体
writer = Agent(
role=\'技术内容策略师\',
goal=\'撰写关于 {topic} 的引人入胜的博客文章\',
backstory=(
"作为一名著名的内容策略师，你以能够将复杂的技术概念转化为引人入胜的故事而闻名。"
"你拥有将复杂思想提炼为简单、引人入胜的格式的天赋。"
),
verbose=True,
allow_delegation=True # 该Agent可以委派任务
)
---
02.定义Task
a.创建研究任务
a.功能说明
现在，我们为研究员智能体创建一个具体的任务。任务描述清晰地说明了需要做什么——识别关于特定主题的最新AI趋势。期望输出则为任务的成功标准——一份包含5个要点的总结报告。我们将这个任务明确地分配给之前创建的`researcher`智能体。
b.代码示例
---
from crewai import Task
# 为researcher创建一个研究任务
task1 = Task(
description=(
"识别‘AI在2024年的应用’这一主题下的最新趋势。"
"重点关注主要的技术突破、行业应用和潜在的社会影响。"
),
expected_output="一份包含5个要点的项目符号列表总结，详细说明每个趋势及其重要性。",
agent=researcher
)
---
b.创建写作任务
a.功能说明
然后，我们为作家智能体创建一个写作任务。这个任务的目标是基于研究任务的结果，撰写一篇完整的博客文章。任务描述要求文章的格式为Markdown，并强调了内容的吸引力。
b.代码示例
---
# 为writer创建一个写作任务
task2 = Task(
description=(
"利用研究分析师提供的见解，撰写一篇关于‘AI在2024年的应用’的引人入胜的博客文章。"
"文章应突出最重要的趋势，并以引人入胜、信息丰富的方式呈现。"
),
expected_output="一篇格式完整的博客文章，采用Markdown格式，准备好发布。",
agent=writer
)
---
03.组建并运行Crew
a.实例化Crew
a.功能说明
最后一步是将我们创建的智能体和任务组建成一个`Crew`。我们将`researcher`和`writer`放入`agents`列表，将`task1`和`task2`放入`tasks`列表。我们选择`Process.sequential`作为执行流程，这意味着任务将按照它们在列表中的顺序依次执行。设置`verbose=2`可以让我们在运行时看到详细的执行日志。
b.代码示例
---
from crewai import Crew, Process
# 实例化你的Crew
crew = Crew(
agents=[researcher, writer],
tasks=[task1, task2],
process=Process.sequential, # 顺序执行
verbose=2, # 可以设置为1或2，查看不同详细程度的日志
)
---
b.启动任务
a.功能说明
通过调用`crew.kickoff()`方法，我们正式启动整个团队开始工作。我们可以在`inputs`字典中传入在任务和智能体定义中使用的动态占位符（如`{topic}`）。Crew会返回所有任务完成后的最终结果。
b.代码示例
---
# 启动Crew并获取结果
result = crew.kickoff(inputs={\'topic\': \'AI在2024年的应用\'})
print("######################")
print(result)
---
2 核心组件
2.1 核心组件：Agent详解
01.Agent基本构成
a.角色与目标 (Role & Goal)
a.功能说明
`role`和`goal`是定义一个Agent身份和使命的核心参数。`role`（角色）为其提供了一个行为框架和身份定位，例如“资深软件工程师”或“市场分析专家”。`goal`（目标）则清晰地指明了该Agent需要完成的具体任务。这两个参数共同作用，引导LLM在执行任务时产生更符合预期的行为和输出。
b.代码示例
---
# 定义一个具有明确角色和目标的Agent
data_analyst_agent = Agent(
role="数据分析师",
goal="分析用户行为数据并找出关键增长点",
# ... 其他参数
)
---
b.背景故事 (Backstory)
a.功能说明
`backstory`（背景故事）为Agent提供了更丰富的上下文和个性。一段精心编写的背景故事，可以使其行为更具深度和一致性，仿佛一个拥有真实经验和知识的专家。这有助于LLM更好地进行角色扮演，从而提高任务完成的质量。
b.代码示例
---
# 为数据分析师Agent添加背景故事
data_analyst_agent = Agent(
role="数据分析师",
goal="分析用户行为数据并找出关键增长点",
backstory=(
"你曾在多家顶尖科技公司担任数据分析主管，"
"擅长从复杂的数据集中洞察商业机会，并为产品决策提供数据支持。"
),
# ... 其他参数
)
---
02.Agent高级配置
a.任务委派 (Delegation)
a.功能说明
`allow_delegation`参数控制Agent是否能够将任务或子任务委派给其他Agent。当一个Agent认为某项任务超出了其能力范围，或者由另一个Agent处理更高效时，如果`allow_delegation`设置为`True`，它就可以主动发起委派。这使得Crew能够动态地、自主地进行任务分工和协作，极大地提高了灵活性和效率。
b.代码示例
---
# 创建一个可以委派任务的项目经理Agent
project_manager_agent = Agent(
role="项目经理",
goal="协调团队完成一个新功能的开发",
allow_delegation=True # 允许委派
)
# 创建一个不允许委派任务的初级开发Agent
junior_dev_agent = Agent(
role="初级开发者",
goal="根据分配的任务编写代码",
allow_delegation=False # 不允许委派
)
---
b.内存管理 (Memory)
a.功能说明
`memory`参数决定Agent是否在执行任务时拥有记忆能力。当设置为`True`时，Agent能够记住之前的对话历史、行动和结果，从而在后续的决策中利用这些信息。这对于需要长期上下文理解和连贯执行的复杂任务至关重要。CrewAI会利用记忆来提高后续步骤的质量和相关性。
b.代码示例
---
# 创建一个具有记忆能力的客服Agent
support_agent_with_memory = Agent(
role="高级客服代表",
goal="解决客户的复杂问题，并提供个性化支持",
memory=True, # 启用记忆功能
# ... 其他参数
)
---
c.详细日志 (Verbose)
a.功能说明
`verbose`参数用于控制Agent在执行过程中的日志输出详细程度。设置为`True`或具体的数字（如1或2）可以让你实时看到Agent的思考过程、工具使用情况、与其他智能体的交互等信息。这对于调试和理解Crew的工作流程非常有帮助。
b.代码示例
---
# 创建一个开启详细日志的Agent
debugging_agent = Agent(
role="调试助手",
goal="复现并定位一个软件缺陷",
verbose=True, # 开启详细日志
# ... 其他参数
)
---
d.语言模型配置 (LLM)
a.功能说明
`llm`参数允许为单个Agent指定特定的语言模型。默认情况下，Agent会使用在Crew层面定义的LLM。通过这个参数，可以为不同的Agent配置不同的模型，例如，为需要强大推理能力的研究员Agent配置GPT-4，而为执行简单文本处理任务的Agent配置更经济的模型。这有助于在性能和成本之间取得平衡。
b.代码示例
---
from langchain_openai import ChatOpenAI
# 为不同的Agent配置不同的LLM
gpt4_llm = ChatOpenAI(model_name="gpt-4-turbo")
gpt3_5_llm = ChatOpenAI(model_name="gpt-3.5-turbo")
# 研究员使用GPT-4
researcher = Agent(
role="研究员",
goal="进行深度研究",
llm=gpt4_llm,
# ...
)
# 总结员使用GPT-3.5
summarizer = Agent(
role="总结员",
goal="总结研究报告",
llm=gpt3_5_llm,
# ...
)
---
2.2 核心组件：Task详解
01.Task基本构成
a.描述与期望输出 (Description & Expected Output)
a.功能说明
`description`和`expected_output`是定义一个Task的核心。`description`清晰地阐述了该任务需要完成的具体工作内容，为Agent提供了明确的行动指令。`expected_output`则定义了任务成功的标准和交付物的格式，使得Agent的目标更加聚焦，也便于后续步骤或人工对结果进行评估。
b.代码示例
---
# 定义一个具有清晰描述和期望输出的任务
market_research_task = Task(
description=(
"对2024年第二季度全球智能手机市场进行全面分析。"
"需要涵盖主要厂商的市场份额、出货量变化以及关键增长区域。"
),
expected_output=(
"一份详细的市场分析报告，包含："
"1. 各厂商市场份额的饼状图数据。"
"2. 出货量同比和环比变化的表格。"
"3. 对亚洲、欧洲、北美三大市场的增长趋势分析。"
),
# ... 其他参数
)
---
b.分配Agent (Agent)
a.功能说明
`agent`参数用于将一个任务明确地分配给一个或多个特定的Agent来执行。如果只分配给一个Agent，则该Agent会独立完成此任务。如果分配给一个Agent列表，则这些Agent会协同完成该任务，但具体的协作方式取决于Crew的流程设置。这个参数是连接任务和执行者的桥梁。
b.代码示例
---
# 假设已创建 market_analyst 和 data_visualizer 两个Agent
# 将任务分配给单个Agent
analysis_task = Task(
description="分析销售数据",
expected_output="一份销售趋势报告",
agent=market_analyst
)
# 将任务分配给多个Agent协同完成（不常用，通常在Crew层面定义协作）
# visualization_task = Task(
# description="将分析报告可视化",
# expected_output="一组数据图表",
# agent=[market_analyst, data_visualizer]
# )
---
02.Task高级配置
a.任务上下文 (Context)
a.功能说明
`context`参数允许将其他任务的输出作为当前任务的输入。这使得任务之间可以形成依赖关系和数据流。例如，一个写作任务可以利用一个研究任务的输出作为其写作素材。通过在任务列表中按顺序组织，前一个任务的结果会自动传递给后一个任务，无需手动指定`context`。只有在需要使用非直接前驱任务的结果时，才需要手动配置。
b.代码示例
---
# research_task 和 writing_task 顺序执行
# writing_task 会自动接收 research_task 的输出作为上下文
writing_task = Task(
description="根据研究报告撰写一篇博客文章。",
expected_output="一篇完整的博客文章。",
agent=writer_agent
)
# 手动指定上下文的场景（较少见）
# translation_task = Task(
# description="将最终的博客文章翻译成西班牙语。",
# expected_output="西班牙语版本的博客文章。",
# context=[writing_task], # 明确指定使用writing_task的输出
# agent=translator_agent
# )
# crew = Crew(tasks=[research_task, writing_task, translation_task])
---
b.异步执行 (Async Execution)
a.功能说明
`async_execution`参数允许任务以异步方式执行。当设置为`True`时，Crew在启动该任务后不会等待其完成，而是会立即开始执行下一个可以执行的任务。这对于执行可以并行处理的、互不依赖的任务非常有用，可以显著缩短整个Crew的运行时间。
b.代码示例
---
# 创建两个可以并行执行的异步任务
# 任务1：从API获取天气数据
fetch_weather_task = Task(
description="获取纽约市今天的天气数据。",
expected_output="包含温度和天气的字符串。",
agent=api_agent,
async_execution=True
)
# 任务2：从API获取股票数据
fetch_stock_task = Task(
description="获取苹果公司今天的股价。",
expected_output="包含股价的字符串。",
agent=api_agent,
async_execution=True
)
# 汇总任务，它会等待前两个异步任务完成
summary_task = Task(
description="总结天气和股票信息。",
expected_output="一段总结性文字。",
agent=summary_agent,
context=[fetch_weather_task, fetch_stock_task]
)
---
c.输出文件 (Output File)
a.功能说明
`output_file`参数允许将任务的最终输出直接保存到一个文件中。这对于需要生成报告、代码、配置文件等产物的任务非常方便。只需提供一个文件路径，CrewAI会在任务成功完成后自动将结果写入该文件。
b.代码示例
---
# 创建一个将结果保存到文件的代码生成任务
code_generation_task = Task(
description="编写一个Python函数，用于计算斐波那契数列。",
expected_output="一个名为 fibonacci.py 的文件，包含完整的Python代码。",
agent=coder_agent,
output_file="fibonacci.py"
)
---
2.3 核心组件：Crew详解
01.Crew基本构成
a.智能体与任务列表 (Agents & Tasks)
a.功能说明
`agents`和`tasks`是组建一个Crew最核心的两个列表。`agents`列表包含了所有参与该Crew工作的智能体实例，定义了“谁来做”。`tasks`列表则包含了所有需要完成的任务实例，定义了“做什么”。Crew会根据这两个列表以及设定的流程来管理和执行整个工作流。
b.代码示例
---
# 假设已创建 researcher, writer 两个Agent
# 以及 research_task, write_task 两个Task
# 创建一个包含Agent和Task列表的Crew
blog_crew = Crew(
agents=[researcher, writer], # 定义团队成员
tasks=[research_task, write_task], # 定义待办任务
# ... 其他参数
)
---
02.流程控制 (Process)
a.顺序流程 (Sequential)
a.功能说明
`Process.sequential`是Crew最常用也是默认的执行流程。当设置为顺序流程时，Crew会严格按照`tasks`列表中定义的顺序，一个接一个地执行任务。前一个任务的输出会自动成为后一个任务的上下文，形成一条清晰的任务链。这种模式简单、可预测，适用于大多数线性工作流。
b.代码示例
---
from crewai import Crew, Process
# 创建一个使用顺序流程的Crew
sequential_crew = Crew(
agents=[...],
tasks=[task1, task2, task3],
process=Process.sequential # 设置为顺序执行
)
# 任务执行顺序: task1 -> task2 -> task3
---
b.层级流程 (Hierarchical)
a.功能说明
`Process.hierarchical`提供了一种更复杂的、由管理者驱动的协作模式。在这种模式下，你需要指定一个`manager_llm`。Crew会动态地决定任务的执行顺序，而不是严格遵循列表顺序。管理者（由`manager_llm`驱动）会评估所有任务和智能体的状态，决定下一步应该执行哪个任务，或者是否需要委派新任务。这种模式更加灵活和智能，适用于需要动态规划和决策的复杂场景。
b.代码示例
---
from langchain_openai import ChatOpenAI
# 创建一个使用层级流程的Crew
hierarchical_crew = Crew(
agents=[...],
tasks=[...],
process=Process.hierarchical, # 设置为层级执行
manager_llm=ChatOpenAI(model="gpt-4-turbo") # 必须指定一个管理者LLM
)
# 任务执行顺序由manager_llm动态决定
---
03.Crew高级配置
a.共享工具 (Share Crew)
a.功能说明
`share_crew`参数允许一个Crew将其自身作为一个工具，提供给另一个Crew使用。这是一种强大的组合模式，可以将一个专门负责某项复杂任务（如“深度市场分析”）的Crew，封装成一个单一的工具，供一个更高层次的“决策”Crew调用。这极大地促进了模块化和代码复用。
b.代码示例
---
# 1. 创建一个专门用于分析的子Crew
financial_analysis_crew = Crew(
agents=[analyst, researcher],
tasks=[analysis_task, research_task],
share_crew=True # 允许该Crew被共享
)
# 2. 创建一个使用子Crew作为工具的决策者Agent
ceo_agent = Agent(
role="CEO",
goal="根据市场分析做出战略决策",
tools=[financial_analysis_crew] # 将子Crew作为工具
)
# 3. 在更高层次的Crew中使用该Agent
strategy_crew = Crew(
agents=[ceo_agent],
tasks=[decision_task]
)
---
b.完整输出 (Full Output)
a.功能说明
默认情况下，`kickoff()`方法只返回最后一个任务的输出。当`full_output`设置为`True`时，`kickoff()`将返回一个包含所有任务执行细节的完整对象，包括每个任务的输入、输出、使用的工具、时间戳等。这对于需要详细审计、日志记录或对中间结果进行分析的场景非常有用。
b.代码示例
---
# 创建一个需要完整输出的Crew
audited_crew = Crew(
agents=[...],
tasks=[...],
full_output=True # 启用完整输出
)
# 启动Crew并获取详细结果
detailed_result = audited_crew.kickoff()
# detailed_result 将是一个包含所有执行信息的对象
# print(detailed_result.tasks_outputs)
---
3 相关信息
3.1 Flows工作流
01.Flows核心概念
a.事件驱动架构
a.功能说明
Flows是CrewAI中用于构建生产级AI应用的核心抽象，它采用事件驱动的架构。与线性的、以任务为中心的Crews不同，Flows将应用程序构建为一系列由事件触发的、相互连接的步骤。每个步骤都可以是一个简单的函数、一个复杂的逻辑块，或者是一个完整的Crew。这种架构使得应用能够响应外部事件（如用户输入、API回调、定时器），并以灵活、可扩展的方式处理它们。
b.代码示例
---
from crewai import Flow
# 创建一个Flow实例
customer_support_flow = Flow(
name="客户支持流程",
description="处理来自用户的支持请求。"
)
# 定义一个由事件触发的入口点
@customer_support_flow.on_event("user_request")
def handle_user_request(data):
# ... 处理用户请求的逻辑 ...
if data["is_complex"]:
# 触发另一个事件，将任务委派给Crew
customer_support_flow.trigger("escalate_to_crew", data)
else:
# 执行简单的回复逻辑
return {"reply": "这是一个简单的回复。"}
---
b.状态管理
a.功能说明
Flows内置了强大的状态管理能力。它可以在整个工作流的生命周期内维护一个状态对象，每个步骤都可以读取和更新这个状态。这使得在不同步骤之间传递数据和上下文变得非常简单和可靠。状态管理是构建能够处理多轮交互和复杂逻辑的应用的关键。
b.代码示例
---
# 在Flow的步骤中访问和更新状态
@customer_support_flow.task
def categorize_request(state):
# 读取状态
user_query = state["user_query"]
# 更新状态
category = classify_query(user_query) # 假设这是一个分类函数
state["category"] = category
return state
@customer_support_flow.task
def route_request(state):
# 根据状态中的分类来路由请求
if state["category"] == "billing":
# ... 处理账单问题 ...
else:
# ... 处理技术问题 ...
---
02.Flows与Crews的协同
a.委派复杂任务
a.功能说明
Flows和Crews的设计初衷是协同工作。Flows作为应用程序的“大脑”或“控制器”，负责处理高层逻辑、路由和状态管理。当Flow遇到一个需要深度分析、创造性协作或复杂工具使用的任务时，它就会将这个任务“委派”给一个专门的Crew。Crew作为“专家团队”，负责解决这个复杂问题，并将结果返回给Flow，由Flow决定下一步的行动。这种模式实现了控制与自主的完美结合。
b.代码示例
---
# 在Flow中定义一个委派给Crew的任务
@customer_support_flow.task
def complex_issue_analysis(state):
# 创建并配置一个专门用于分析复杂问题的Crew
analysis_crew = Crew(
agents=[expert_agent, researcher_agent],
tasks=[deep_dive_task]
)
# 启动Crew并传入所需数据
result = analysis_crew.kickoff(inputs={"query": state["user_query"]})
# 将Crew的结果更新到Flow的状态中
state["analysis_result"] = result
return state
---
b.构建生产级应用
a.功能说明
通过结合Flows和Crews，开发者可以构建出健壮、可维护且可扩展的生产级AI应用。Flows提供了清晰的结构和控制流，使得应用逻辑易于理解和修改。Crews则封装了复杂的多智能体协作，使得核心的“智能”部分可以被模块化和复用。这种分层架构使得应用既能处理结构化的业务流程，又能应对开放式的复杂问题。
b.代码示例
---
# 一个完整的生产级应用流程示例
# 1. Flow的入口点接收到一个API请求
@app.post("/support")
def start_support_flow(request: Request):
customer_support_flow.trigger("user_request", request.json())
# 2. Flow的第一个任务对请求进行初步处理和分类
# @customer_support_flow.task
# def categorize_request(state): ...
# 3. 如果是复杂问题，Flow将任务委派给一个Crew
# @customer_support_flow.task
# def complex_issue_analysis(state): ...
# 4. Flow的最后一个任务根据Crew的结果生成最终回复，并通过API返回
# @customer_support_flow.task
# def generate_final_response(state): ...
---
3.2 工具集成
01.使用内置工具
a.官方工具库 (crewai_tools)
a.功能说明
CrewAI提供了一个名为`crewai_tools`的官方工具库，其中包含了大量预置的、可以直接使用的工具。这些工具涵盖了网页搜索、文件读写、代码执行、网站内容抓取等多种常用功能。通过简单地导入和实例化，就可以为Agent配备强大的能力。
b.代码示例
---
from crewai import Agent
from crewai_tools import (
SerperDevTool,
FileReadTool,
ScrapeWebsiteTool,
DirectoryReadTool
)
# 创建工具实例
search_tool = SerperDevTool()
file_read_tool = FileReadTool()
scrape_tool = ScrapeWebsiteTool()
directory_read_tool = DirectoryReadTool()
# 在创建Agent时，将工具列表传入
research_agent = Agent(
role="研究员",
goal="进行全面的信息收集",
tools=[
search_tool,
file_read_tool,
scrape_tool,
directory_read_tool
]
)
---
b.工具使用示例
a.功能说明
Agent在执行任务时，会根据其目标和当前上下文，自主决定何时以及如何使用其配备的工具。例如，一个研究员Agent在接到研究任务后，可能会首先使用`SerperDevTool`进行网络搜索，然后使用`ScrapeWebsiteTool`抓取关键网页的内容，最后使用`FileReadTool`读取本地的参考资料，以完成其研究报告。
b.代码示例
---
# 任务描述
research_task = Task(
description=(
"研究LangChain的最新发展。首先，在网上搜索\'LangChain v0.2 release notes\'，"
"然后抓取官方发布说明页面的内容，并结合本地文件\'my_notes.md\'进行总结。"
),
expected_output="一份详细的LangChain v0.2更新总结报告。",
agent=research_agent
)
# 在执行过程中，Agent会依次调用 search_tool, scrape_tool, file_read_tool
---
02.创建自定义工具
a.继承BaseTool类
a.功能说明
当内置工具无法满足特定需求时，可以轻松创建自定义工具。只需创建一个继承自`crewai_tools.BaseTool`的类，并实现`name`、`description`和`_run`方法即可。`name`和`description`用于告诉Agent这个工具的用途，而`_run`方法则包含了工具的具体执行逻辑。
b.代码示例
---
from crewai_tools import BaseTool
# 创建一个用于计算两个数之和的自定义工具
class CalculatorTool(BaseTool):
name: str = "计算器工具"
description: str = "计算两个整数的和。"
def _run(self, a: int, b: int) -> str:
"""执行计算并返回结果。"""
return str(a + b)
# 创建Agent时使用这个自定义工具
math_agent = Agent(
role="数学家",
goal="解决数学问题",
tools=[CalculatorTool()]
)
---
b.使用函数工具
a.功能说明
对于更简单的场景，CrewAI提供了`FunctionTool`，可以将任何一个普通的Python函数快速封装成一个工具，而无需创建完整的类。只需将函数本身和其描述传递给`FunctionTool`即可。这对于快速集成现有函数或进行原型设计非常方便。
b.代码示例
---
from crewai_tools import FunctionTool
# 定义一个普通的Python函数
def get_weather(city: str) -> str:
"""获取指定城市的天气信息。"""
# 实际应用中会调用天气API
if city.lower() == "beijing":
return "北京今天晴天，气温25摄氏度。"
else:
return f"无法获取{city}的天气信息。"
# 将函数封装成工具
weather_tool = FunctionTool.from_function(
func=get_weather,
description="获取指定城市的实时天气信息。"
)
# 在Agent中使用该工具
travel_agent = Agent(
role="旅行规划师",
goal="为用户规划行程",
tools=[weather_tool]
)
---
3.3 高级特性
01.协作模式
a.任务委派 (Delegation)
a.功能说明
任务委派是CrewAI中实现动态协作的核心机制。当一个Agent在其`allow_delegation`参数被设置为`True`时，它可以在执行任务的过程中，判断是否需要将部分工作或整个任务委派给另一个更适合的Agent。这种决策是自主的，基于Agent对任务的理解和对团队中其他成员能力的认知。这使得Crew能够像一个真正的人类团队一样，灵活地分配工作，而不是死板地遵循预设流程。
b.代码示例
---
# 假设有一个项目经理Agent和一个开发Agent
project_manager = Agent(
role="项目经理",
goal="确保项目按时交付",
allow_delegation=True # 允许委派
)
developer = Agent(
role="软件工程师",
goal="编写高质量的代码",
allow_delegation=False # 通常执行者不委派
)
# 任务：规划并执行一个新功能的开发
planning_task = Task(
description="规划新功能的开发，并将其分解为具体的编码任务。",
expected_output="一份详细的开发计划和已委派的编码任务列表。",
agent=project_manager
)
# 在执行planning_task时，project_manager会自主地创建新的编码任务，并委派给developer
---
02.记忆功能 (Memory)
a.短期记忆
a.功能说明
CrewAI的Agent默认拥有短期记忆，即在同一个任务执行过程中，能够记住之前的思考和行动。这使得Agent可以进行连贯的、多步骤的推理和操作。例如，一个Agent可以先搜索信息，然后基于搜索结果进行分析，再根据分析结果做出决策。
b.代码示例
---
# Agent的短期记忆是内置的，无需特殊配置
# 在一个任务中，Agent的思考过程展示了其短期记忆：
# Thought: 我需要先搜索X公司的财报。
# Action: search(query="X公司最新财报")
# ... (得到结果)
# Thought: 财报显示利润增长了20%。现在我需要分析其原因。
# Action: search(query="X公司利润增长原因")
---
b.长期记忆 (RAG)
a.功能说明
为了让Crew拥有跨任务、跨会话的长期记忆，CrewAI集成了先进的检索增强生成（RAG）技术。通过在Crew层面配置`memory=True`，可以为整个团队启用长期记忆。Crew会将关键的对话历史、任务结果和学习到的知识存储在一个向量数据库中。在未来的任务中，Agent可以查询这个记忆库，以获取相关的上下文信息，从而做出更明智的决策，避免重复工作。
b.代码示例
---
# 在Crew层面启用长期记忆
crew_with_memory = Crew(
agents=[...],
tasks=[...],
memory=True, # 启用RAG长期记忆
verbose=2
)
# 第一次运行后，Crew会将关键信息存入记忆库
# result1 = crew_with_memory.kickoff(inputs={"customer_id": "123"})
# 第二次为同一客户服务时，Crew可以从记忆库中检索历史信息
# result2 = crew_with_memory.kickoff(inputs={"customer_id": "123"})
---
03.规划能力 (Planning)
a.任务分解
a.功能说明
CrewAI的Agent具备初步的任务规划和分解能力。当面对一个复杂的任务时，Agent可以将其分解为一系列更小、更易于管理的子任务。这种能力通常通过Agent的`backstory`和`goal`来引导，并结合其可用的工具来实现。例如，一个“旅行规划师”Agent可能会将“规划一次巴黎之旅”分解为“预订机票”、“预订酒店”、“规划每日行程”等子任务。
b.代码示例
---
# Agent的思考过程展示了其任务分解能力
# Thought: 我需要规划一次巴黎之旅。这个任务可以分解为几个步骤：
# 1. 查找并预订往返机票。
# 2. 根据预算和位置选择并预订酒店。
# 3. 规划每天的观光路线和活动。
# 我将首先开始第一步：查找机票。
# Action: search_flights(origin="NYC", destination="CDG", date="2024-10-10")
---
3.4 实战案例
01.市场分析报告
a.案例描述
a.功能说明
这是一个典型的研究与写作案例。目标是创建一个能够自动生成关于特定行业（如“AI行业”）市场分析报告的Crew。这个Crew由一个研究员Agent和一个作家Agent组成。研究员负责收集数据和趋势，作家负责将研究结果整理成一篇结构完整、语言流畅的报告。
b.代码示例
---
# 完整代码结构
# 1. 定义Agents: researcher, writer
# - researcher配备搜索工具
# 2. 定义Tasks: research_task, write_task
# - research_task分配给researcher
# - write_task分配给writer
# 3. 组建Crew:
# - agents=[researcher, writer]
# - tasks=[research_task, write_task]
# - process=Process.sequential
# 4. 启动Crew:
# - crew.kickoff(inputs={\'topic\': \'AI行业最新发展\'})
---
02.客户支持自动化
a.案例描述
a.功能说明
这个案例旨在构建一个多层次的客户支持系统。Crew由三个Agent组成：一个“分类员”Agent，一个“初级支持”Agent，和一个“高级工程师”Agent。当收到客户请求时，分类员首先判断问题的类型和紧急程度。简单问题由初级支持Agent使用知识库工具直接回答。复杂的技术问题则被委派给高级工程师Agent，该Agent可以使用更专业的诊断工具来解决问题。
b.代码示例
---
# 概念代码结构
# 1. Agents: classifier, junior_support, senior_engineer
# - classifier: 无特殊工具
# - junior_support: 配备知识库搜索工具
# - senior_engineer: 配备日志分析、数据库查询等高级工具
# 2. Tasks: classify_task, simple_support_task, complex_support_task
# 3. Crew:
# - 使用层级流程(Process.hierarchical)和管理者LLM来动态决定任务流转
# - 或者通过Agent的任务委派能力实现
---
03.自动化代码生成与审查
a.案例描述
a.功能说明
这个案例模拟了一个软件开发的迷你流程。Crew包含一个“产品经理”Agent，一个“程序员”Agent，和一个“代码审查员”Agent。产品经理负责将需求转化为具体的开发任务。程序员根据任务编写代码。代码完成后，审查员Agent会自动对代码进行分析，检查其是否符合编码规范、是否存在潜在的bug，并提出修改建议。整个流程实现了从需求到代码产出的自动化闭环。
b.代码示例
---
# 概念代码结构
# 1. Agents: product_manager, programmer, code_reviewer
# - programmer: 配备文件读写工具
# - code_reviewer: 配备静态代码分析工具
# 2. Tasks: define_spec_task, coding_task, review_task
# 3. Crew:
# - 使用顺序流程(Process.sequential)
# - coding_task的输出(代码文件路径)作为review_task的输入
---
3.5 生产部署
01.API服务封装
a.使用FastAPI
a.功能说明
将CrewAI应用部署为生产级服务的最常见方式是使用Web框架（如FastAPI）将其封装为API。你可以创建一个FastAPI应用，定义一个或多个API端点（例如`/run-crew`）。当这个端点被调用时，它会接收输入参数，启动Crew，并以HTTP响应的形式返回最终结果。这种方式使得Crew可以被任何客户端（网页、移动应用、其他后端服务）轻松集成。
b.代码示例
---
# main.py
from fastapi import FastAPI
from pydantic import BaseModel
from my_crew import MyCrew # 假设你的Crew在my_crew.py中定义
app = FastAPI()
class CrewInput(BaseModel):
topic: str
@app.post("/run-crew")
def run_crew_endpoint(crew_input: CrewInput):
inputs = {"topic": crew_input.topic}
result = MyCrew().crew().kickoff(inputs=inputs)
return {"result": result}
# 运行服务: uvicorn main:app --reload
---
02.Docker化部署
a.创建Dockerfile
a.功能说明
为了实现环境一致性和便捷部署，强烈建议使用Docker将你的CrewAI应用容器化。你需要创建一个`Dockerfile`，在其中定义应用的基础环境（如Python版本）、安装所有依赖项、复制项目文件，并指定启动命令（如`uvicorn`）。
b.代码示例
---
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 复制依赖文件并安装
COPY requirements.txt requirements.txt
RUN pip install --no-cache-dir -r requirements.txt
# 复制所有项目文件
COPY . .
# 暴露端口并设置启动命令
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
---
b.构建与运行容器
a.功能说明
有了`Dockerfile`之后，你可以使用`docker build`命令构建镜像，并使用`docker run`命令启动容器。记得通过`-p`参数将容器的端口映射到主机的端口，并通过`--env-file`参数传入包含API密钥的`.env`文件。
b.代码示例
---
# 1. 构建Docker镜像
docker build -t my-crew-app .
# 2. 运行Docker容器
docker run -d -p 8000:8000 --env-file .env --name crew-container my-crew-app
---
03.部署到云平台
a.选择云服务
a.功能说明
容器化的CrewAI应用可以轻松部署到任何支持Docker的云平台，例如Amazon ECS, Google Cloud Run, Microsoft Azure App Service, 或者 Heroku。这些平台提供了自动扩缩容、负载均衡、日志监控等生产级功能，可以确保你的服务高可用和可扩展。
b.代码示例
---
# 以Google Cloud Run为例的部署命令
# 1. 将Docker镜像推送到Google Artifact Registry
gcloud auth configure-docker
docker tag my-crew-app gcr.io/YOUR_PROJECT_ID/my-crew-app
docker push gcr.io/YOUR_PROJECT_ID/my-crew-app
# 2. 部署到Cloud Run
gcloud run deploy my-crew-service \
--image gcr.io/YOUR_PROJECT_ID/my-crew-app \
--platform managed \
--region us-central1 \
--allow-unauthenticated \
--set-env-vars="OPENAI_API_KEY=your_key"
---