【Agents篇】14：CrewAI——基于角色的团队协作
·
🎯 核心主题：深入解析CrewAI框架的核心架构与设计理念，掌握多智能体协作系统的构建方法
📑 目录
- 引言：为什么需要CrewAI
- CrewAI核心架构概览
- 核心组件深度解析
- 3.1 Agent（智能体）
- 3.2 Task（任务）
- 3.3 Crew（团队）
- 3.4 Process（流程）
- 角色定义与设计艺术
- 工作流编排实战
- Flows：事件驱动的高级编排
- 状态管理与持久化
- 实战案例：构建AI研究团队
- 高级特性与最佳实践
- CrewAI与其他框架对比
- 总结与展望
- 参考文献
1. 引言：为什么需要CrewAI 🤔
1.1 多智能体系统的崛起
在人工智能领域，单一模型的能力边界正在被不断突破，但现实世界的复杂任务往往需要多种专业技能的协同配合。想象一下：
- 一个内容创作项目需要研究员收集资料、作家撰写内容、编辑校对润色
- 一个软件开发任务需要架构师设计系统、开发者编写代码、测试员验证功能
- 一个商业分析项目需要数据分析师处理数据、市场专家解读趋势、战略顾问制定方案
这些场景都有一个共同点：需要不同角色的专家协同工作。
💡 思考：如何让多个AI Agent像人类团队一样协作？
🤔 解答：CrewAI正是为解决这个问题而生的框架。它引入了"角色扮演"的核心理念，让每个Agent都有明确的角色、目标和背景故事，从而实现更自然、更高效的团队协作。
1.2 CrewAI的诞生背景
CrewAI是一个精简、快速的Python框架，完全从零开始构建——完全独立于LangChain或其他Agent框架。它赋予开发者高层次的简洁性和精确的底层控制，非常适合创建针对任何场景的自主AI Agent。
┌─────────────────────────────────────────────────────────────────┐
│ CrewAI 设计哲学 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 🎭 角色扮演 (Role-Playing) │
│ ↓ │
│ 🎯 目标驱动 (Goal-Oriented) │
│ ↓ │
│ 🤝 协作智能 (Collaborative Intelligence) │
│ ↓ │
│ ⚡ 高性能执行 (High Performance) │
│ │
└─────────────────────────────────────────────────────────────────┘
1.3 核心优势
| 特性 | 描述 |
|---|---|
| 独立框架 | 完全独立，不依赖LangChain，执行更快、资源占用更少 |
| 灵活精确 | 通过Crews（自主性）或Flows（精确控制）轻松编排Agent |
| 无缝集成 | Crews和Flows可以完美结合，创建复杂的真实世界自动化 |
| 深度定制 | 从高层工作流到底层内部提示和Agent行为，全面可定制 |
| 可靠性能 | 从简单任务到复杂企业级自动化，表现一致 |
2. CrewAI核心架构概览 🏗️
2.1 整体架构图
┌───────────────────────────────────────────────────────────────────────────┐
│ CrewAI 架构总览 │
├───────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Flows Layer (事件驱动层) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ @start │───▶│ @listen │───▶│ @router │───▶│ @listen │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │
│ │ │ │ │ │ │ │
│ │ ▼ ▼ ▼ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────────┐ │ │
│ │ │ State Management (状态管理) │ │ │
│ │ └─────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Crew Layer (团队协作层) │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────────────────────────┐ │ │
│ │ │ Crew │────────▶│ Process │ │ │
│ │ │ (团队) │ │ ┌───────────┬───────────┐ │ │ │
│ │ └─────────────┘ │ │Sequential │Hierarchical│ │ │ │
│ │ │ │ │ (顺序) │ (层级) │ │ │ │
│ │ │ │ └───────────┴───────────┘ │ │ │
│ │ ▼ └─────────────────────────────────┘ │ │
│ │ ┌─────────────────────────────────────────┐ │ │
│ │ │ Agents (智能体群) │ │ │
│ │ │ ┌───────┐ ┌───────┐ ┌───────┐ │ │ │
│ │ │ │Agent 1│ │Agent 2│ │Agent 3│ ... │ │ │
│ │ │ │ Role │ │ Role │ │ Role │ │ │ │
│ │ │ │ Goal │ │ Goal │ │ Goal │ │ │ │
│ │ │ │Backsty│ │Backsty│ │Backsty│ │ │ │
│ │ │ └───┬───┘ └───┬───┘ └───┬───┘ │ │ │
│ │ └──────┼──────────┼─────────┼────────────┘ │ │
│ │ │ │ │ │ │
│ │ ▼ ▼ ▼ │ │
│ │ ┌─────────────────────────────────────────┐ │ │
│ │ │ Tasks (任务队列) │ │ │
│ │ │ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │
│ │ │ │Task 1│─▶│Task 2│─▶│Task 3│ ... │ │ │
│ │ │ └──────┘ └──────┘ └──────┘ │ │ │
│ │ └─────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Foundation Layer (基础设施层) │ │
│ │ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ LLM │ │ Tools │ │ Memory │ │Knowledge │ │ │
│ │ │ 连接器 │ │ 工具库 │ │ 记忆系统 │ │ 知识库 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└───────────────────────────────────────────────────────────────────────────┘
2.2 核心概念关系图
┌─────────────────┐
│ Crew │
│ (团队) │
└────────┬────────┘
│
┌──────────────┼──────────────┐
│ │ │
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Agent │ │ Task │ │ Process │
│ (智能体) │ │ (任务) │ │ (流程) │
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │
┌────────┼────────┐ │ ┌───────┴───────┐
│ │ │ │ │ │
▼ ▼ ▼ │ ▼ ▼
┌──────┐ ┌──────┐ ┌──────┐ │ ┌────────┐ ┌────────┐
│ Role │ │ Goal │ │Back- │ │ │Sequential│ │Hierarchical│
│ │ │ │ │story │ │ │ 顺序执行 │ │ 层级执行 │
└──────┘ └──────┘ └──────┘ │ └────────┘ └────────┘
│
┌─────────────┼─────────────┐
│ │ │
▼ ▼ ▼
┌────────┐ ┌────────┐ ┌────────┐
│Description│ │Expected│ │Context │
│ 描述 │ │ Output │ │ 上下文 │
└────────┘ └────────┘ └────────┘
2.3 两种编排范式
CrewAI提供两种强大而互补的方法来构建AI应用：
2.3.1 Crews（团队自主协作）
┌────────────────────────────────────────────────────────────┐
│ Crews 特点 │
├────────────────────────────────────────────────────────────┤
│ │
│ ✅ 自然的自主决策 │
│ ✅ 动态任务委派和协作 │
│ ✅ 具有明确目标和专业知识的专门角色 │
│ ✅ 灵活的问题解决方法 │
│ │
│ 适用场景：需要Agent自主协作、动态决策的复杂任务 │
│ │
└────────────────────────────────────────────────────────────┘
2.3.2 Flows（精确事件驱动）
┌────────────────────────────────────────────────────────────┐
│ Flows 特点 │
├────────────────────────────────────────────────────────────┤
│ │
│ ✅ 对执行路径的精细控制 │
│ ✅ 任务间安全、一致的状态管理 │
│ ✅ AI Agent与生产Python代码的无缝集成 │
│ ✅ 复杂业务逻辑的条件分支 │
│ │
│ 适用场景：需要精确控制执行流程的生产级应用 │
│ │
└────────────────────────────────────────────────────────────┘
💡 思考：什么时候用Crews，什么时候用Flows？
🤔 解答：
- 当任务需要创造性协作和动态决策时，使用Crews
- 当需要精确的执行控制和复杂的业务逻辑时，使用Flows
- 最佳实践是结合使用：Flows控制整体流程，在关键节点调用Crews执行复杂的协作任务
3. 核心组件深度解析 🔍
3.1 Agent（智能体）
Agent是CrewAI框架中的核心自主单元，它可以：
- 执行特定任务
- 基于角色和目标做出决策
- 使用工具完成目标
- 与其他Agent通信协作
- 维护交互记忆
- 在允许时委派任务
3.1.1 Agent核心属性
from crewai import Agent
# 完整的Agent定义示例
agent = Agent(
# 🎭 核心身份三要素（必填）
role="高级数据科学家", # 角色定义
goal="分析和解读复杂数据集，提供可操作的洞察", # 目标
backstory="""
你拥有超过10年的数据科学和机器学习经验，
擅长在复杂数据集中发现模式。你曾在多家
财富500强企业担任首席数据科学家。
""", # 背景故事
# 🤖 LLM配置
llm="gpt-4", # 默认：OPENAI_MODEL_NAME 或 "gpt-4"
function_calling_llm=None, # 用于工具调用的专用LLM
# ⚙️ 执行控制
verbose=True, # 启用详细日志
allow_delegation=False, # 是否允许委派任务
max_iter=20, # 最大迭代次数
max_rpm=None, # API调用速率限制
max_execution_time=None, # 最大执行时间（秒）
max_retry_limit=2, # 错误重试次数
# 🛠️ 工具配置
tools=[], # 可用工具列表
allow_code_execution=False, # 是否允许执行代码
code_execution_mode="safe", # 代码执行模式："safe"（Docker）或"unsafe"
# 🧠 记忆与上下文
memory=True, # 启用记忆功能
respect_context_window=True, # 自动处理上下文窗口
knowledge_sources=None, # 知识源配置
# 🎨 高级特性
reasoning=False, # 启用推理规划
max_reasoning_attempts=None, # 推理尝试次数
multimodal=False, # 多模态支持
inject_date=False, # 自动注入日期
# 📝 自定义模板
system_template=None, # 自定义系统提示
prompt_template=None, # 自定义提示模板
response_template=None, # 自定义响应模板
# 📊 回调与监控
step_callback=None, # 步骤回调函数
cache=True, # 启用工具缓存
)
3.1.2 Agent属性详解表
| 属性 | 类型 | 默认值 | 描述 |
|---|---|---|---|
role |
str | 必填 | 定义Agent在团队中的功能和专业领域 |
goal |
str | 必填 | 指导Agent决策的个人目标 |
backstory |
str | 必填 | 为Agent提供背景和个性，丰富交互内容 |
llm |
str/LLM | gpt-4 | 驱动Agent的语言模型 |
tools |
List[BaseTool] | [] | Agent可使用的能力或功能 |
verbose |
bool | False | 启用详细执行日志 |
allow_delegation |
bool | False | 允许Agent将任务委派给其他Agent |
max_iter |
int | 20 | Agent必须给出最佳答案前的最大迭代次数 |
memory |
bool | False | 启用交互记忆 |
reasoning |
bool | False | 启用执行前的反思和规划 |
multimodal |
bool | False | 支持多模态能力 |
3.1.3 不同类型Agent的设计模式
研究型Agent
research_agent = Agent(
role="研究分析师",
goal="查找并总结特定主题的信息",
backstory="你是一位经验丰富的研究员，注重细节",
tools=[SerperDevTool()],
verbose=True
)
代码开发Agent
dev_agent = Agent(
role="高级Python开发者",
goal="编写和调试Python代码",
backstory="拥有10年经验的Python专家开发者",
allow_code_execution=True,
code_execution_mode="safe", # 使用Docker保证安全
max_execution_time=300, # 5分钟超时
max_retry_limit=3 # 复杂代码任务允许更多重试
)
推理规划Agent
strategic_agent = Agent(
role="战略规划师",
goal="分析复杂问题并创建详细的执行计划",
backstory="擅长系统性分解复杂挑战的专家战略规划师",
reasoning=True, # 启用推理和规划
max_reasoning_attempts=3, # 限制推理尝试次数
max_iter=30, # 复杂规划允许更多迭代
verbose=True
)
3.2 Task（任务）
Task是Agent执行的具体工作单元，提供执行所需的所有必要细节。
3.2.1 Task核心属性
from crewai import Task
task = Task(
# 📋 核心定义（必填）
description="""
对人工智能最新发展进行深入研究。
确保找到任何有趣且相关的信息，
当前年份是2025年。
""",
expected_output="""
一份包含10个要点的列表，
记录关于AI最相关的信息
""",
# 👤 执行者
agent=research_agent, # 负责执行任务的Agent
# 🛠️ 工具配置
tools=[search_tool], # 任务特定的工具
# 🔗 依赖与上下文
context=[previous_task], # 依赖的其他任务输出
# ⚙️ 执行控制
async_execution=False, # 是否异步执行
human_input=False, # 是否需要人工审核
# 📤 输出配置
output_file="report.md", # 输出文件路径
output_json=None, # JSON输出的Pydantic模型
output_pydantic=None, # Pydantic输出模型
markdown=True, # 启用Markdown格式化
# 🛡️ 验证与回调
guardrail=validate_function, # 输出验证函数
guardrails=[], # 多重验证器
guardrail_max_retries=3, # 验证失败最大重试次数
callback=callback_function, # 完成后的回调函数
)
3.2.2 Task属性详解表
| 属性 | 类型 | 描述 |
|---|---|---|
description |
str | 任务内容的清晰简明陈述 |
expected_output |
str | 任务完成的详细描述 |
agent |
Agent | 负责执行任务的Agent |
tools |
List[BaseTool] | Agent可用于此任务的工具 |
context |
List[Task] | 将用作此任务上下文的其他任务 |
async_execution |
bool | 是否异步执行任务 |
human_input |
bool | 是否需要人工审核最终答案 |
output_file |
str | 存储任务输出的文件路径 |
output_pydantic |
Type[BaseModel] | 用于任务输出的Pydantic模型 |
output_json |
Type[BaseModel] | 用于结构化JSON输出的Pydantic模型 |
guardrail |
Callable | 验证任务输出的函数 |
callback |
Callable | 任务完成后执行的函数 |
3.2.3 任务依赖与上下文
# 任务1：研究任务
research_task = Task(
description="研究AI领域的最新发展",
expected_output="近期AI发展列表",
agent=researcher
)
# 任务2：分析任务（依赖于研究任务）
analysis_task = Task(
description="分析研究发现并识别关键趋势",
expected_output="AI趋势分析报告",
agent=analyst,
context=[research_task] # 此任务将等待research_task完成
)
┌─────────────────────────────────────────────────────────────┐
│ 任务执行流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ │
│ │ Research Task│ │
│ │ 研究任务 │ │
│ └──────┬───────┘ │
│ │ │
│ │ output (输出作为上下文) │
│ ▼ │
│ ┌──────────────┐ │
│ │Analysis Task │◀─── context=[research_task] │
│ │ 分析任务 │ │
│ └──────┬───────┘ │
│ │ │
│ │ output │
│ ▼ │
│ ┌──────────────┐ │
│ │ Report Task │◀─── context=[analysis_task] │
│ │ 报告任务 │ │
│ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
3.2.4 Task Guardrails（任务护栏）
Guardrails提供了在将任务输出传递给下一个任务之前验证和转换的方法：
函数式Guardrail
from typing import Tuple, Any
from crewai import TaskOutput
def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
"""验证博客内容是否符合要求"""
try:
word_count = len(result.raw.split())
if word_count > 200:
return (False, "博客内容超过200字")
return (True, result.raw.strip())
except Exception as e:
return (False, "验证过程中发生意外错误")
blog_task = Task(
description="写一篇关于AI的博客文章",
expected_output="200字以内的博客文章",
agent=blog_agent,
guardrail=validate_blog_content
)
LLM式Guardrail（字符串描述）
research_task = Task(
description="研究量子计算的最新发展",
expected_output="一份全面的研究报告",
agent=researcher_agent,
guardrail="""
研究报告必须：
- 至少1000字
- 包含至少5个可信来源
- 涵盖技术和实际应用
- 采用专业、学术的语气
- 避免推测或未经验证的声明
"""
)
3.2.5 结构化输出
from pydantic import BaseModel
from crewai import Agent, Crew, Process, Task
class Blog(BaseModel):
title: str
content: str
blog_agent = Agent(
role="博客内容生成Agent",
goal="生成博客标题和内容",
backstory="你是一位专业的内容创作者，擅长撰写引人入胜的博客文章。",
llm="gpt-4o",
)
task = Task(
description="创建一个关于给定主题的博客标题和内容，确保内容在200字以内。",
expected_output="引人注目的博客标题和精心撰写的内容。",
agent=blog_agent,
output_pydantic=Blog, # 使用Pydantic模型确保结构化输出
)
crew = Crew(
agents=[blog_agent],
tasks=[task],
process=Process.sequential,
)
result = crew.kickoff()
# 访问结构化输出
print("标题:", result.pydantic.title)
print("内容:", result.pydantic.content)
3.3 Crew（团队）
Crew代表一组协作工作的Agent团队，定义任务执行策略、Agent协作方式和整体工作流程。
3.3.1 Crew核心属性
from crewai import Crew, Process
crew = Crew(
# 📋 核心配置（必填）
agents=[agent1, agent2], # 团队成员列表
tasks=[task1, task2], # 任务列表
# ⚙️ 流程控制
process=Process.sequential, # 执行流程：sequential或hierarchical
# 🤖 管理配置（层级流程需要）
manager_llm="gpt-4o", # 管理Agent使用的LLM
manager_agent=None, # 自定义管理Agent
# 🧠 记忆与缓存
memory=True, # 启用记忆（短期、长期、实体记忆）
cache=True, # 启用工具执行结果缓存
embedder={"provider": "openai"}, # 嵌入器配置
# 📊 监控与日志
verbose=True, # 详细日志
output_log_file="logs.txt", # 日志文件
step_callback=None, # 步骤回调
task_callback=None, # 任务回调
# 🔧 高级配置
max_rpm=None, # 最大请求速率
planning=False, # 启用规划能力
planning_llm=None, # 规划使用的LLM
knowledge_sources=None, # 团队级知识源
# 📡 输出配置
stream=False, # 启用流式输出
)
3.3.2 Crew属性详解表
| 属性 | 参数 | 描述 |
|---|---|---|
tasks |
tasks | 分配给团队的任务列表 |
agents |
agents | 团队成员Agent列表 |
process |
process | 执行流程（sequential或hierarchical），默认sequential |
verbose |
verbose | 执行期间的日志详细程度，默认False |
manager_llm |
manager_llm | 层级流程中管理Agent使用的语言模型 |
memory |
memory | 用于存储执行记忆（短期、长期、实体记忆） |
cache |
cache | 是否缓存工具执行结果，默认True |
planning |
planning | 启用规划能力，在每次迭代前发送数据给AgentPlanner |
stream |
stream | 启用流式输出，实时接收更新 |
3.3.3 使用YAML配置（推荐方式）
agents.yaml
# src/my_project/config/agents.yaml
researcher:
role: >
{topic} 高级数据研究员
goal: >
发掘 {topic} 领域的前沿发展
backstory: >
你是一位经验丰富的研究员，擅长发现 {topic} 领域的最新发展。
以能够找到最相关的信息并以清晰简洁的方式呈现而闻名。
reporting_analyst:
role: >
{topic} 报告分析师
goal: >
基于 {topic} 数据分析和研究发现创建详细报告
backstory: >
你是一位注重细节的分析师。以能够将复杂数据转化为清晰
简洁的报告而闻名，使他人易于理解和采取行动。
tasks.yaml
# src/my_project/config/tasks.yaml
research_task:
description: >
对 {topic} 进行深入研究
确保找到任何有趣且相关的信息，当前年份是2025年。
expected_output: >
一份包含10个要点的列表，记录关于 {topic} 最相关的信息
agent: researcher
reporting_task:
description: >
审查你获得的上下文，并将每个主题扩展为报告的完整章节。
确保报告详尽，包含所有相关信息。
expected_output: >
一份完整的报告，包含主要主题，每个主题都有完整的信息章节。
格式为markdown，不带 '```'
agent: reporting_analyst
markdown: true
output_file: report.md
crew.py
# src/my_project/crew.py
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import SerperDevTool
@CrewBase
class LatestAiDevelopmentCrew():
"""最新AI发展研究团队"""
agents_config = 'config/agents.yaml'
tasks_config = 'config/tasks.yaml'
@agent
def researcher(self) -> Agent:
return Agent(
config=self.agents_config['researcher'],
verbose=True,
tools=[SerperDevTool()]
)
@agent
def reporting_analyst(self) -> Agent:
return Agent(
config=self.agents_config['reporting_analyst'],
verbose=True
)
@task
def research_task(self) -> Task:
return Task(
config=self.tasks_config['research_task'],
)
@task
def reporting_task(self) -> Task:
return Task(
config=self.tasks_config['reporting_task'],
output_file='report.md'
)
@crew
def crew(self) -> Crew:
"""创建最新AI发展团队"""
return Crew(
agents=self.agents,
tasks=self.tasks,
process=Process.sequential,
verbose=True,
)
3.3.4 Crew输出结构
# 执行Crew
crew_output = crew.kickoff()
# 访问输出
print(f"原始输出: {crew_output.raw}")
# JSON格式输出
if crew_output.json_dict:
print(f"JSON输出: {json.dumps(crew_output.json_dict, indent=2)}")
# Pydantic模型输出
if crew_output.pydantic:
print(f"Pydantic输出: {crew_output.pydantic}")
# 各任务输出
print(f"任务输出: {crew_output.tasks_output}")
# Token使用统计
print(f"Token使用: {crew_output.token_usage}")
3.4 Process（流程）
Process定义了团队中任务的执行方式，决定了Agent之间的协作模式。
3.4.1 流程类型
┌─────────────────────────────────────────────────────────────────────┐
│ Process 流程类型 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Sequential (顺序流程) │ │
│ │ │ │
│ │ Task 1 ───▶ Task 2 ───▶ Task 3 ───▶ Task 4 │ │
│ │ │ │ │ │ │ │
│ │ ▼ ▼ ▼ ▼ │ │
│ │ Agent A Agent B Agent C Agent D │ │
│ │ │ │
│ │ 特点：任务按顺序执行，前一任务的输出作为下一任务的上下文 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Hierarchical (层级流程) │ │
│ │ │ │
│ │ ┌──────────────┐ │ │
│ │ │ Manager │ │ │
│ │ │ (管理者) │ │ │
│ │ └──────┬───────┘ │ │
│ │ │ │ │
│ │ ┌───────────────┼───────────────┐ │ │
│ │ │ │ │ │ │
│ │ ▼ ▼ ▼ │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ │ │
│ │ │Agent A │ │Agent B │ │Agent C │ │ │
│ │ │ Task 1 │ │ Task 2 │ │ Task 3 │ │ │
│ │ └────────┘ └────────┘ └────────┘ │ │
│ │ │ │
│ │ 特点：管理者Agent协调团队，根据能力分配任务，验证结果 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Consensual (共识流程) [计划中] │ │
│ │ │ │
│ │ 特点：Agent之间协作决策，民主化的任务管理方式 │ │
│ │ 状态：计划中，尚未实现 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
3.4.2 顺序流程（Sequential）
from crewai import Crew, Process
# 创建顺序执行的团队
crew = Crew(
agents=my_agents,
tasks=my_tasks,
process=Process.sequential # 默认值
)
特点：
- 任务按照定义的顺序依次执行
- 前一任务的输出自动成为下一任务的上下文
- 适用于线性工作流
3.4.3 层级流程（Hierarchical）
from crewai import Crew, Process
# 创建层级执行的团队
crew = Crew(
agents=my_agents,
tasks=my_tasks,
process=Process.hierarchical,
manager_llm="gpt-4o" # 必须指定manager_llm或manager_agent
# 或者使用自定义管理Agent
# manager_agent=my_manager_agent
)
特点：
- 自动创建或使用指定的管理Agent
- 管理者负责任务规划、委派和验证
- 任务不预先分配，由管理者根据Agent能力动态分配
- 适用于复杂的多步骤任务
💡 思考：层级流程中的管理者Agent是如何工作的？
🤔 解答：管理者Agent会：
- 分析所有待完成的任务
- 评估每个Agent的角色、目标和能力
- 将任务委派给最合适的Agent
- 审查任务输出，评估完成质量
- 在必要时重新分配或调整任务
4. 角色定义与设计艺术 🎭
4.1 角色设计的三要素
┌─────────────────────────────────────────────────────────────────┐
│ 角色设计三要素 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Role (角色) │ │
│ │ │ │
│ │ 定义Agent的职能和专业领域 │ │
│ │ 例如："高级数据科学家"、"内容策略师" │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Goal (目标) │ │
│ │ │ │
│ │ 指导Agent决策的个人目标 │ │
│ │ 例如："分析复杂数据集并提供可操作的洞察" │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Backstory (背景故事) │ │
│ │ │ │
│ │ 为Agent提供上下文和个性 │ │
│ │ 例如："你拥有10年数据科学经验，曾在多家 │ │
│ │ 财富500强企业担任首席数据科学家..." │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
4.2 角色设计最佳实践
4.2.1 明确且具体的角色定义
# ❌ 不推荐：模糊的角色定义
bad_agent = Agent(
role="助手",
goal="帮助完成任务",
backstory="一个有用的助手"
)
# ✅ 推荐：具体明确的角色定义
good_agent = Agent(
role="金融科技领域高级市场研究分析师",
goal="识别金融科技市场的新兴趋势和投资机会，为战略决策提供数据驱动的洞察",
backstory="""
你是一位在金融服务行业拥有15年经验的市场研究专家。
你曾在高盛和摩根士丹利的研究部门工作，专注于金融科技
和数字银行领域。你以对市场趋势的敏锐洞察和对复杂数据
的深入分析而闻名。你发表过多篇关于支付技术、区块链
和数字货币的行业报告。
"""
)
4.2.2 角色协作设计
# 研究团队的角色设计示例
researcher = Agent(
role="首席研究员",
goal="收集和验证来自多个来源的高质量研究数据",
backstory="""
你是一位资深的学术研究员，拥有博士学位和20年的研究经验。
你以严谨的研究方法和对信息准确性的执着追求而闻名。
你的研究被顶级期刊引用超过1000次。
""",
tools=[SerperDevTool(), WikipediaTools()],
allow_delegation=True
)
analyst = Agent(
role="数据分析专家",
goal="对研究数据进行深入分析，发现隐藏的模式和洞察",
backstory="""
你是一位数据科学专家，擅长统计分析和机器学习。
你曾帮助多家《财富》500强企业从数据中提取商业价值。
你特别擅长将复杂的分析结果转化为易于理解的洞察。
""",
tools=[],
allow_delegation=False
)
writer = Agent(
role="技术写作专家",
goal="将研究发现转化为清晰、专业的书面报告",
backstory="""
你是一位获奖的技术作家，拥有工程学位和新闻学背景。
你擅长将复杂的技术概念解释给非专业读者。你的文章
以清晰、准确和引人入胜而著称。
""",
tools=[],
allow_delegation=False
)
4.3 常见角色模板
# 📊 数据分析师
data_analyst = Agent(
role="高级数据分析师",
goal="从数据中提取有价值的洞察，支持业务决策",
backstory="精通SQL、Python和各种数据可视化工具的分析专家"
)
# 💻 软件开发者
developer = Agent(
role="全栈软件开发工程师",
goal="设计和实现高质量、可维护的软件解决方案",
backstory="拥有多年企业级应用开发经验的资深工程师"
)
# 📝 内容创作者
content_creator = Agent(
role="内容策略师",
goal="创作引人入胜的内容，提升品牌影响力",
backstory="曾为多个知名品牌打造成功内容营销策略的专家"
)
# 🔍 质量检查员
qa_specialist = Agent(
role="质量保证专家",
goal="确保输出符合最高质量标准",
backstory="对细节有极致追求的质量控制专家"
)
# 🎯 项目经理
project_manager = Agent(
role="项目管理专家",
goal="协调资源、管理进度，确保项目成功交付",
backstory="PMP认证的项目管理专家，曾成功交付数百个项目"
)
5. 工作流编排实战 ⚡
5.1 顺序工作流
from crewai import Agent, Task, Crew, Process
# 步骤1：定义Agent
researcher = Agent(
role="AI研究员",
goal="研究AI领域的最新进展",
backstory="经验丰富的AI研究专家"
)
writer = Agent(
role="技术作家",
goal="撰写易于理解的技术文章",
backstory="擅长将复杂概念简单化的作家"
)
editor = Agent(
role="编辑",
goal="确保内容质量和准确性",
backstory="严谨的内容编辑专家"
)
# 步骤2：定义任务
research_task = Task(
description="研究大语言模型的最新发展趋势",
expected_output="包含5个关键发现的研究摘要",
agent=researcher
)
writing_task = Task(
description="基于研究摘要撰写一篇技术博客",
expected_output="1500字的技术博客文章",
agent=writer,
context=[research_task] # 依赖研究任务的输出
)
editing_task = Task(
description="编辑和优化博客文章",
expected_output="经过润色的最终文章",
agent=editor,
context=[writing_task], # 依赖写作任务的输出
output_file="final_article.md"
)
# 步骤3：创建团队
content_crew = Crew(
agents=[researcher, writer, editor],
tasks=[research_task, writing_task, editing_task],
process=Process.sequential,
verbose=True
)
# 步骤4：启动执行
result = content_crew.kickoff()
print(result)
5.2 层级工作流
from crewai import Agent, Task, Crew, Process
# 定义专家Agent
market_analyst = Agent(
role="市场分析师",
goal="分析市场数据和趋势",
backstory="拥有10年金融市场分析经验"
)
financial_analyst = Agent(
role="财务分析师",
goal="评估财务数据和风险",
backstory="CFA持证人，专注于投资分析"
)
strategy_consultant = Agent(
role="战略顾问",
goal="制定投资策略建议",
backstory="顶级咨询公司合伙人背景"
)
# 定义任务（不预先分配Agent）
market_analysis = Task(
description="分析科技行业2024年的市场表现和趋势",
expected_output="详细的市场分析报告"
)
financial_assessment = Task(
description="评估目标公司的财务健康状况",
expected_output="财务评估报告，包含关键指标分析"
)
strategy_recommendation = Task(
description="基于市场和财务分析，制定投资建议",
expected_output="投资策略建议书"
)
# 创建层级团队
investment_crew = Crew(
agents=[market_analyst, financial_analyst, strategy_consultant],
tasks=[market_analysis, financial_assessment, strategy_recommendation],
process=Process.hierarchical,
manager_llm="gpt-4o", # 管理Agent使用的模型
verbose=True
)
result = investment_crew.kickoff()
5.3 异步任务执行
# 异步执行多个独立任务
research_ai_task = Task(
description="研究AI领域的最新发展",
expected_output="AI发展报告",
async_execution=True, # 异步执行
agent=research_agent
)
research_blockchain_task = Task(
description="研究区块链技术的最新应用",
expected_output="区块链应用报告",
async_execution=True, # 异步执行
agent=research_agent
)
# 汇总任务等待异步任务完成
summary_task = Task(
description="整合AI和区块链研究，撰写综合分析报告",
expected_output="技术趋势综合分析报告",
agent=writer_agent,
context=[research_ai_task, research_blockchain_task] # 等待两个异步任务
)
┌─────────────────────────────────────────────────────────────────┐
│ 异步任务执行流程 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌────────────────┐ ┌────────────────┐ │
│ │ AI Research │ │ Blockchain │ │
│ │ (async=True) │ │ Research │ │
│ │ │ │ (async=True) │ │
│ └───────┬────────┘ └───────┬────────┘ │
│ │ │ │
│ │ 并行执行 │ │
│ │◀───────────────────────│ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────────────────────────────┐ │
│ │ Summary Task │ │
│ │ context=[ai_task, blockchain_task] │ │
│ │ 等待所有异步任务完成 │ │
│ └─────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
5.4 带回调的工作流
from crewai import TaskOutput
def task_completed_callback(output: TaskOutput):
"""任务完成后的回调函数"""
print(f"""
✅ 任务完成！
📋 任务描述: {output.description}
📄 输出结果: {output.raw}
🤖 执行Agent: {output.agent}
""")
# 可以在这里发送通知、记录日志等
def step_callback(step_output):
"""每个步骤完成后的回调"""
print(f"步骤输出: {step_output}")
# 创建带回调的任务
research_task = Task(
description="研究AI最新进展",
expected_output="研究报告",
agent=researcher,
callback=task_completed_callback # 任务完成回调
)
# 创建带回调的团队
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, write_task],
process=Process.sequential,
step_callback=step_callback, # 步骤回调
task_callback=task_completed_callback # 任务回调
)
6. Flows：事件驱动的高级编排 🌊
6.1 Flows概述
CrewAI Flows是一个强大的功能，专为简化AI工作流的创建和管理而设计。Flows允许开发者高效地组合和协调编码任务和Crews。
┌─────────────────────────────────────────────────────────────────────┐
│ Flows 核心特点 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 🔗 简化的工作流创建 │
│ 轻松链接多个Crews和任务，创建复杂的AI工作流 │
│ │
│ 📊 状态管理 │
│ Flows使工作流中不同任务之间的状态管理和共享变得超级简单 │
│ │
│ ⚡ 事件驱动架构 │
│ 基于事件驱动模型构建，实现动态和响应式的工作流 │
│ │
│ 🔀 灵活的控制流 │
│ 在工作流中实现条件逻辑、循环和分支 │
│ │
└─────────────────────────────────────────────────────────────────────┘
6.2 核心装饰器
from crewai.flow.flow import Flow, start, listen, router
class MyFlow(Flow):
@start() # 标记流程入口点
def begin_flow(self):
"""流程开始"""
return "开始数据"
@listen(begin_flow) # 监听begin_flow的输出
def process_data(self, data):
"""处理数据"""
return f"处理后: {data}"
@router(process_data) # 基于输出进行路由
def route_decision(self):
"""决定下一步"""
if self.state.success:
return "success_path"
return "failure_path"
@listen("success_path") # 监听路由结果
def handle_success(self):
"""处理成功情况"""
pass
@listen("failure_path")
def handle_failure(self):
"""处理失败情况"""
pass
6.3 完整Flow示例
from crewai.flow.flow import Flow, listen, start
from litellm import completion
class ResearchFlow(Flow):
model = "gpt-4o-mini"
@start()
def generate_topic(self):
"""生成研究主题"""
print("开始Flow...")
print(f"Flow状态ID: {self.state['id']}")
response = completion(
model=self.model,
messages=[{
"role": "user",
"content": "生成一个有趣的AI研究主题",
}],
)
topic = response["choices"][0]["message"]["content"]
self.state["topic"] = topic
print(f"研究主题: {topic}")
return topic
@listen(generate_topic)
def research_topic(self, topic):
"""研究选定的主题"""
response = completion(
model=self.model,
messages=[{
"role": "user",
"content": f"提供关于'{topic}'的5个关键研究发现",
}],
)
findings = response["choices"][0]["message"]["content"]
self.state["findings"] = findings
return findings
@listen(research_topic)
def generate_summary(self, findings):
"""生成研究摘要"""
response = completion(
model=self.model,
messages=[{
"role": "user",
"content": f"基于以下发现撰写简短摘要:\n{findings}",
}],
)
summary = response["choices"][0]["message"]["content"]
self.state["summary"] = summary
return summary
# 运行Flow
flow = ResearchFlow()
flow.plot("research_flow") # 生成可视化图表
result = flow.kickoff()
print(f"最终结果: {result}")
6.4 条件控制：or_ 和 and_
from crewai.flow.flow import Flow, start, listen, or_, and_
class ConditionalFlow(Flow):
@start()
def start_method(self):
return "开始"
@start()
def another_start(self):
return "另一个开始"
# or_: 任一方法完成即触发
@listen(or_(start_method, another_start))
def log_any(self, result):
"""任意一个start方法完成就执行"""
print(f"收到结果: {result}")
# and_: 所有方法完成才触发
@listen(and_(start_method, another_start))
def log_all(self, results):
"""两个start方法都完成后执行"""
print(f"所有结果: {results}")
6.5 路由器模式
from crewai.flow.flow import Flow, start, listen, router
from random import random
class RouterFlow(Flow):
@start()
def analyze_data(self):
"""分析数据并设置置信度"""
confidence = random()
self.state["confidence"] = confidence
return {"confidence": confidence}
@router(analyze_data)
def route_by_confidence(self):
"""根据置信度路由"""
if self.state["confidence"] > 0.8:
return "high_confidence"
elif self.state["confidence"] > 0.5:
return "medium_confidence"
return "low_confidence"
@listen("high_confidence")
def handle_high_confidence(self):
print("置信度高，直接执行策略")
return "执行策略A"
@listen("medium_confidence")
def handle_medium_confidence(self):
print("置信度中等，需要进一步分析")
return "执行策略B"
@listen("low_confidence")
def handle_low_confidence(self):
print("置信度低，收集更多数据")
return "执行策略C"
┌─────────────────────────────────────────────────────────────────┐
│ Router 路由流程 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ │
│ │ analyze_data │ │
│ │ @start() │ │
│ └────────┬────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ route_by_ │ │
│ │ confidence │ │
│ │ @router() │ │
│ └────────┬────────┘ │
│ │ │
│ ┌─────────────┼─────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ "high │ │ "medium │ │ "low │ │
│ │ confidence"│ │ confidence"│ │ confidence"│ │
│ └─────┬──────┘ └─────┬──────┘ └─────┬──────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ handle_ │ │ handle_ │ │ handle_ │ │
│ │ high │ │ medium │ │ low │ │
│ │ @listen() │ │ @listen() │ │ @listen() │ │
│ └────────────┘ └────────────┘ └────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
6.6 在Flow中集成Crew
from crewai.flow.flow import Flow, listen, start
from crewai import Crew, Agent, Task, Process
from pydantic import BaseModel
class AnalysisState(BaseModel):
topic: str = ""
research_result: str = ""
analysis_result: str = ""
class ResearchAnalysisFlow(Flow[AnalysisState]):
@start()
def initialize(self):
"""初始化研究主题"""
print(f"开始研究: {self.state.topic}")
return {"topic": self.state.topic}
@listen(initialize)
def research_with_crew(self, data):
"""使用Crew进行研究"""
# 创建研究Agent
researcher = Agent(
role="高级研究员",
goal=f"深入研究{self.state.topic}",
backstory="经验丰富的领域研究专家"
)
# 创建研究任务
research_task = Task(
description=f"全面研究{self.state.topic}的最新发展",
expected_output="详细的研究报告",
agent=researcher
)
# 创建并执行Crew
research_crew = Crew(
agents=[researcher],
tasks=[research_task],
process=Process.sequential,
verbose=True
)
result = research_crew.kickoff()
self.state.research_result = result.raw
return result.raw
@listen(research_with_crew)
def analyze_with_crew(self, research_data):
"""使用另一个Crew进行分析"""
analyst = Agent(
role="数据分析专家",
goal="分析研究数据并提取洞察",
backstory="专注于数据分析和趋势预测"
)
analysis_task = Task(
description=f"分析以下研究数据并提供关键洞察:\n{research_data}",
expected_output="分析报告和行动建议",
agent=analyst
)
analysis_crew = Crew(
agents=[analyst],
tasks=[analysis_task],
process=Process.sequential
)
result = analysis_crew.kickoff()
self.state.analysis_result = result.raw
return result.raw
@listen(analyze_with_crew)
def generate_report(self, analysis):
"""生成最终报告"""
print("=== 最终报告 ===")
print(f"主题: {self.state.topic}")
print(f"研究结果: {self.state.research_result[:200]}...")
print(f"分析结果: {analysis}")
return analysis
# 运行Flow
async def run_flow():
flow = ResearchAnalysisFlow()
flow.plot("research_analysis_flow")
result = await flow.kickoff_async(inputs={"topic": "人工智能在医疗领域的应用"})
return result
# asyncio.run(run_flow())
7. 状态管理与持久化 💾
7.1 非结构化状态管理
from crewai.flow.flow import Flow, listen, start
class UnstructuredFlow(Flow):
@start()
def first_method(self):
# 状态自动包含'id'字段
print(f"状态ID: {self.state['id']}")
# 动态添加任意属性
self.state['counter'] = 0
self.state['message'] = "Hello"
self.state['data'] = {"key": "value"}
@listen(first_method)
def second_method(self):
# 更新状态
self.state['counter'] += 1
self.state['message'] += " - 已更新"
@listen(second_method)
def third_method(self):
print(f"最终状态: {self.state}")
特点：
- 灵活性高，可动态添加属性
- 适合简单或高度动态的工作流
- 快速原型开发友好
7.2 结构化状态管理
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel
class WorkflowState(BaseModel):
# 注意：'id'字段会自动添加到所有状态
counter: int = 0
message: str = ""
completed_steps: list = []
class StructuredFlow(Flow[WorkflowState]):
@start()
def first_method(self):
# 访问自动生成的ID
print(f"状态ID: {self.state.id}")
# 类型安全的状态更新
self.state.message = "开始结构化流程"
self.state.completed_steps.append("first_method")
@listen(first_method)
def second_method(self):
self.state.counter += 1
self.state.message += " - 第二步"
self.state.completed_steps.append("second_method")
@listen(second_method)
def third_method(self):
self.state.counter += 1
self.state.completed_steps.append("third_method")
print(f"最终状态: {self.state}")
特点：
- 类型安全，IDE支持良好
- 验证确保数据一致性
- 适合需要严格状态结构的应用
7.3 状态持久化
from crewai.flow.flow import Flow, start, listen
from crewai.flow.persistence import persist
# 类级别持久化 - 所有方法状态自动持久化
@persist # 默认使用SQLiteFlowPersistence
class PersistentFlow(Flow):
@start()
def initialize(self):
self.state["counter"] = 1
print(f"初始化。状态ID: {self.state['id']}")
@listen(initialize)
def increment(self):
self.state["counter"] += 1
print(f"计数器: {self.state['counter']}")
# 方法级别持久化 - 更精细的控制
class SelectivePersistFlow(Flow):
@persist # 仅此方法持久化
@start()
def critical_step(self):
if "runs" not in self.state:
self.state["runs"] = 0
self.state["runs"] += 1
print(f"关键步骤执行次数: {self.state['runs']}")
┌─────────────────────────────────────────────────────────────────┐
│ 状态持久化工作原理 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ │
│ │ Flow 执行 │ │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌─────────────────────────────────┐ │
│ │ 状态更新 │───▶│ SQLite数据库 │ │
│ └──────────────┘ │ - 唯一状态ID │ │
│ │ │ - 序列化的状态数据 │ │
│ ▼ │ - 时间戳 │ │
│ ┌──────────────┐ └─────────────────────────────────┘ │
│ │ 系统重启 │ │
│ └──────┬───────┘ │ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────┐ ┌─────────────────────────────────┐ │
│ │ 状态恢复 │◀───│ 从数据库加载状态 │ │
│ └──────────────┘ └─────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ 继续执行 │ │
│ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
8. 实战案例：构建AI研究团队 🎯
8.1 项目结构
ai_research_team/
├── src/
│ └── ai_research_team/
│ ├── __init__.py
│ ├── main.py
│ ├── crew.py
│ ├── config/
│ │ ├── agents.yaml
│ │ └── tasks.yaml
│ └── tools/
│ └── custom_tool.py
├── pyproject.toml
├── README.md
└── .env
8.2 配置文件
agents.yaml
# src/ai_research_team/config/agents.yaml
lead_researcher:
role: >
首席AI研究员
goal: >
领导{topic}领域的前沿研究，发现突破性的技术洞察
backstory: >
你是人工智能领域的顶尖研究员，拥有MIT博士学位和15年研究经验。
你曾在DeepMind和OpenAI工作，发表过多篇顶级会议论文。
你以敏锐的洞察力和严谨的研究方法著称，善于发现技术发展的
关键趋势和潜在突破点。
data_analyst:
role: >
高级数据分析师
goal: >
分析{topic}相关数据，提供量化洞察和趋势预测
backstory: >
你是一位数据科学专家，精通统计分析、机器学习和数据可视化。
你曾帮助多家科技巨头从海量数据中提取关键洞察。你的分析报告
以数据驱动、逻辑严密而闻名。
technical_writer:
role: >
技术内容专家
goal: >
将复杂的{topic}研究转化为清晰、专业的书面报告
backstory: >
你是一位获奖的技术作家，拥有计算机科学和新闻学双重背景。
你擅长将复杂的技术概念转化为易于理解的内容，你的文章
被《MIT Technology Review》和《Wired》等媒体转载。
quality_reviewer:
role: >
质量审核专家
goal: >
确保关于{topic}的所有输出内容准确、完整、专业
backstory: >
你是一位严谨的技术审核专家，拥有10年的学术审稿经验。
你以对细节的极致追求和对质量的严格要求而闻名，
没有任何错误能逃过你的审查。
tasks.yaml
# src/ai_research_team/config/tasks.yaml
comprehensive_research:
description: >
对{topic}进行全面深入的研究。
要求：
1. 收集最新的学术论文、行业报告和技术博客
2. 分析当前的技术发展状态和关键突破
3. 识别主要玩家和他们的技术路线
4. 预测未来12-24个月的发展趋势
当前年份是{year}，确保所有信息都是最新的。
expected_output: >
一份结构化的研究报告，包含：
- 技术概述和发展历程
- 当前最先进技术分析
- 关键玩家和竞争格局
- 10个关键发现
- 未来趋势预测
agent: lead_researcher
data_analysis:
description: >
基于研究报告，进行深入的数据分析。
要求：
1. 量化分析{topic}的市场规模和增长趋势
2. 分析技术采用率和应用场景分布
3. 评估投资和资金流向
4. 进行竞争力量化对比
expected_output: >
一份数据分析报告，包含：
- 市场规模和增长预测（带数据支持）
- 技术采用率分析
- 投资趋势分析
- 竞争格局量化评估
- 至少3个数据可视化描述
agent: data_analyst
report_writing:
description: >
整合研究和分析结果，撰写专业的技术报告。
要求：
1. 结构清晰，逻辑严密
2. 语言专业但易于理解
3. 包含执行摘要和详细章节
4. 提供可操作的建议
expected_output: >
一份3000字左右的专业技术报告，包含：
- 执行摘要（300字）
- 技术背景介绍
- 当前发展状态
- 数据分析洞察
- 未来趋势预测
- 战略建议
- 参考文献
agent: technical_writer
markdown: true
output_file: research_report.md
quality_review:
description: >
对研究报告进行全面的质量审核。
审核要点：
1. 事实准确性检验
2. 逻辑一致性检查
3. 内容完整性评估
4. 语言质量和专业性审核
5. 格式规范性检查
expected_output: >
质量审核报告，包含：
- 整体质量评分（1-10分）
- 发现的问题列表
- 改进建议
- 最终审核意见
如果报告质量达标（8分以上），输出"APPROVED"；
否则提供详细的改进建议。
agent: quality_reviewer
8.3 团队实现
# src/ai_research_team/crew.py
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task, before_kickoff, after_kickoff
from crewai_tools import SerperDevTool, WebsiteSearchTool
from datetime import datetime
@CrewBase
class AIResearchTeam():
"""AI研究团队"""
agents_config = 'config/agents.yaml'
tasks_config = 'config/tasks.yaml'
@before_kickoff
def prepare_inputs(self, inputs):
"""准备输入数据"""
inputs['year'] = datetime.now().year
print(f"🚀 启动AI研究团队 - 主题: {inputs.get('topic', '未指定')}")
return inputs
@after_kickoff
def process_output(self, output):
"""处理输出"""
print("✅ 研究任务完成!")
return output
@agent
def lead_researcher(self) -> Agent:
"""首席研究员"""
return Agent(
config=self.agents_config['lead_researcher'],
verbose=True,
tools=[SerperDevTool(), WebsiteSearchTool()],
allow_delegation=True,
memory=True
)
@agent
def data_analyst(self) -> Agent:
"""数据分析师"""
return Agent(
config=self.agents_config['data_analyst'],
verbose=True,
tools=[SerperDevTool()],
memory=True
)
@agent
def technical_writer(self) -> Agent:
"""技术作家"""
return Agent(
config=self.agents_config['technical_writer'],
verbose=True,
memory=True
)
@agent
def quality_reviewer(self) -> Agent:
"""质量审核员"""
return Agent(
config=self.agents_config['quality_reviewer'],
verbose=True,
memory=True
)
@task
def comprehensive_research(self) -> Task:
"""综合研究任务"""
return Task(
config=self.tasks_config['comprehensive_research']
)
@task
def data_analysis(self) -> Task:
"""数据分析任务"""
return Task(
config=self.tasks_config['data_analysis'],
context=[self.comprehensive_research()]
)
@task
def report_writing(self) -> Task:
"""报告撰写任务"""
return Task(
config=self.tasks_config['report_writing'],
context=[self.comprehensive_research(), self.data_analysis()]
)
@task
def quality_review(self) -> Task:
"""质量审核任务"""
return Task(
config=self.tasks_config['quality_review'],
context=[self.report_writing()]
)
@crew
def crew(self) -> Crew:
"""创建AI研究团队"""
return Crew(
agents=self.agents,
tasks=self.tasks,
process=Process.sequential,
verbose=True,
memory=True,
cache=True,
output_log_file="research_logs.json"
)
8.4 主程序入口
# src/ai_research_team/main.py
import sys
from ai_research_team.crew import AIResearchTeam
def run():
"""运行AI研究团队"""
inputs = {
'topic': 'Large Language Models and AI Agents'
}
try:
result = AIResearchTeam().crew().kickoff(inputs=inputs)
print("\n" + "="*60)
print("📊 研究报告生成完成")
print("="*60)
print(result)
return result
except Exception as e:
print(f"❌ 执行出错: {e}")
raise
def plot():
"""生成团队工作流可视化图"""
AIResearchTeam().crew().plot("research_team_workflow")
print("📈 工作流图已生成: research_team_workflow.html")
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "plot":
plot()
else:
run()
8.5 运行项目
# 安装依赖
crewai install
# 激活虚拟环境
source .venv/bin/activate
# 运行研究团队
crewai run
# 或者直接运行
python src/ai_research_team/main.py
# 生成工作流可视化
python src/ai_research_team/main.py plot
9. 高级特性与最佳实践 🌟
9.1 记忆系统
CrewAI支持三种类型的记忆：
┌─────────────────────────────────────────────────────────────────┐
│ CrewAI 记忆系统 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 短期记忆 (Short-term) │ │
│ │ │ │
│ │ - 存储当前对话/任务的上下文 │ │
│ │ - 帮助Agent保持对话连贯性 │ │
│ │ - 任务结束后清除 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 长期记忆 (Long-term) │ │
│ │ │ │
│ │ - 存储跨会话的知识和经验 │ │
│ │ - 帮助Agent从过去的任务中学习 │ │
│ │ - 持久化存储 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 实体记忆 (Entity) │ │
│ │ │ │
│ │ - 存储关于特定实体的信息 │ │
│ │ - 如人物、地点、概念等 │ │
│ │ - 跨任务保持实体信息一致性 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
# 启用记忆的Crew
crew = Crew(
agents=[agent1, agent2],
tasks=[task1, task2],
memory=True, # 启用记忆系统
embedder={
"provider": "openai",
"config": {
"model": "text-embedding-3-small"
}
}
)
9.2 知识源集成
from crewai import Agent, Crew, Task
from crewai.knowledge.source import TextKnowledgeSource, PDFKnowledgeSource
# 创建知识源
text_knowledge = TextKnowledgeSource(
content="这是一些专业领域知识...",
metadata={"domain": "AI"}
)
pdf_knowledge = PDFKnowledgeSource(
file_path="research_paper.pdf"
)
# Agent使用知识源
expert_agent = Agent(
role="领域专家",
goal="回答专业问题",
backstory="拥有深厚专业知识的专家",
knowledge_sources=[text_knowledge, pdf_knowledge]
)
# 或在Crew级别共享
crew = Crew(
agents=[expert_agent],
tasks=[...],
knowledge_sources=[text_knowledge, pdf_knowledge] # 所有Agent可访问
)
9.3 人机协作（Human-in-the-Loop）
from crewai import Task
# 需要人工审核的任务
review_task = Task(
description="撰写重要的客户提案",
expected_output="专业的商业提案文档",
agent=proposal_writer,
human_input=True # 启用人工审核
)
# 使用Guardrail进行验证
def human_approval(result):
print(f"请审核以下内容:\n{result.raw}")
approval = input("是否批准？(yes/no): ")
if approval.lower() == "yes":
return (True, result.raw)
return (False, "内容需要修改，请重新生成")
approval_task = Task(
description="生成营销文案",
expected_output="吸引人的营销文案",
agent=copywriter,
guardrail=human_approval
)
9.4 流式输出
# 启用流式输出
crew = Crew(
agents=[researcher],
tasks=[task],
stream=True # 启用流式输出
)
# 迭代接收输出
streaming = crew.kickoff(inputs={"topic": "AI"})
for chunk in streaming:
print(chunk.content, end="", flush=True)
# 获取最终结果
result = streaming.result
9.5 错误处理与重试
from crewai import Agent, Task
# 配置重试机制
robust_agent = Agent(
role="可靠执行者",
goal="完成任务并处理错误",
backstory="经验丰富的执行专家",
max_retry_limit=3, # 最大重试3次
max_iter=25, # 最大迭代25次
max_execution_time=300, # 5分钟超时
respect_context_window=True # 自动处理上下文溢出
)
# 任务级别的Guardrail重试
task = Task(
description="执行关键任务",
expected_output="符合要求的输出",
agent=robust_agent,
guardrail=validation_function,
guardrail_max_retries=5 # Guardrail失败最多重试5次
)
9.6 性能优化建议
# 1. 使用缓存减少重复计算
crew = Crew(
agents=[...],
tasks=[...],
cache=True # 启用工具结果缓存
)
# 2. 合理设置速率限制
agent = Agent(
role="API调用者",
goal="调用外部API",
backstory="...",
max_rpm=10 # 限制每分钟最多10次请求
)
# 3. 异步执行独立任务
task1 = Task(
description="独立任务1",
async_execution=True,
agent=agent1
)
task2 = Task(
description="独立任务2",
async_execution=True,
agent=agent2
)
# 4. 使用function_calling_llm优化成本
agent = Agent(
role="研究员",
goal="...",
backstory="...",
llm="gpt-4", # 主要推理用高端模型
function_calling_llm="gpt-4o-mini" # 工具调用用低成本模型
)
# 5. 上下文窗口管理
agent = Agent(
role="分析师",
goal="...",
backstory="...",
respect_context_window=True # 自动压缩超长上下文
)
10. CrewAI与其他框架对比 ⚔️
10.1 对比总览
┌─────────────────────────────────────────────────────────────────────────┐
│ 多Agent框架对比分析 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌───────────────────────────────────────────────────────────────────┐ │
│ │ CrewAI │ │
│ │ │ │
│ │ ✅ 独立框架，不依赖LangChain │ │
│ │ ✅ 角色扮演设计，自然的团队协作 │ │
│ │ ✅ Crews + Flows 双重编排模式 │ │
│ │ ✅ 高性能，资源占用低 │ │
│ │ ✅ 企业级就绪 │ │
│ │ │ │
│ │ 适用：需要团队协作的复杂任务 │ │
│ │ │ │
│ └───────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────────┐ │
│ │ LangGraph │ │
│ │ │ │
│ │ ✅ 灵活的图结构定义 │ │
│ │ ✅ 与LangChain生态深度集成 │ │
│ │ ⚠️ 需要较多样板代码 │ │
│ │ ⚠️ 状态管理模式复杂 │ │
│ │ │ │
│ │ 适用：需要复杂图结构的工作流 │ │
│ │ │ │
│ └───────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────────┐ │
│ │ AutoGen │ │
│ │ │ │
│ │ ✅ 对话式Agent协作 │ │
│ │ ✅ 微软背景，企业支持 │ │
│ │ ⚠️ 缺乏内置流程概念 │ │
│ │ ⚠️ 大规模任务编排复杂 │ │
│ │ │ │
│ │ 适用：对话为主的Agent协作 │ │
│ │ │ │
│ └───────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────────┐ │
│ │ ChatDev │ │
│ │ │ │
│ │ ✅ 首创流程概念 │ │
│ │ ✅ 软件开发场景优化 │ │
│ │ ⚠️ 实现较为刚性 │ │
│ │ ⚠️ 定制化困难 │ │
│ │ │ │
│ │ 适用：代码生成和软件开发任务 │ │
│ │ │ │
│ └───────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
10.2 性能对比
根据官方benchmarks，CrewAI在某些任务上的性能优势：
10.3 选择建议
| 场景 | 推荐框架 | 原因 |
|---|---|---|
| 团队协作任务 | CrewAI | 角色扮演设计，自然协作 |
| 复杂图工作流 | LangGraph | 灵活的图结构定义 |
| 对话式协作 | AutoGen | 对话驱动设计 |
| 软件开发 | ChatDev/CrewAI | 专门优化 |
| 企业生产环境 | CrewAI | 企业级就绪，高性能 |
11. 总结与展望 🔮
11.1 核心要点回顾
┌─────────────────────────────────────────────────────────────────┐
│ CrewAI 核心要点 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 🎭 角色设计 │
│ - Role + Goal + Backstory 三要素定义Agent身份 │
│ - 明确具体的角色描述带来更好的执行效果 │
│ │
│ 📋 任务定义 │
│ - 清晰的description和expected_output │
│ - 使用context建立任务依赖关系 │
│ - Guardrails确保输出质量 │
│ │
│ 👥 团队协作 │
│ - Sequential流程适合线性工作流 │
│ - Hierarchical流程适合复杂决策任务 │
│ - 合理分配Agent能力和职责 │
│ │
│ 🌊 高级编排 │
│ - Flows提供事件驱动的精确控制 │
│ - Crews + Flows组合实现最大灵活性 │
│ - 状态管理和持久化确保可靠性 │
│ │
└─────────────────────────────────────────────────────────────────┘
11.2 最佳实践总结
-
角色设计
- 每个Agent应有明确、具体的角色定义
- 背景故事应该丰富且相关
- 目标应该清晰且可衡量
-
任务规划
- 将复杂任务分解为可管理的子任务
- 建立清晰的任务依赖关系
- 使用Guardrails确保输出质量
-
流程选择
- 简单线性流程用Sequential
- 需要动态决策时用Hierarchical
- 复杂业务逻辑用Flows
-
性能优化
- 启用缓存减少重复计算
- 合理设置速率限制
- 使用异步执行并行任务
11.3 未来展望
CrewAI正在快速发展，未来值得期待的方向：
- Consensual Process：计划中的共识流程，实现Agent之间的民主决策
- 更强大的记忆系统：增强跨会话学习能力
- 企业级功能增强：更多的可观测性、安全性和合规性功能
- 多模态支持深化：更好的图像、音频处理能力
💡 思考：多Agent系统的未来是什么？
🤔 解答：随着LLM能力的不断提升和Agent框架的成熟，我们正在迈向一个AI团队可以自主完成复杂任务的时代。CrewAI代表了这一趋势的最佳实践——通过角色扮演和团队协作，让AI不仅仅是工具，而是真正的协作伙伴。
12. 参考文献 📚
官方资源
-
CrewAI官方文档
- https://docs.crewai.com/
- 最权威的CrewAI学习资源
-
CrewAI GitHub仓库
- https://github.com/crewAIInc/crewAI
- 源代码和最新更新
-
CrewAI示例仓库
- https://github.com/crewAIInc/crewAI-examples
- 丰富的实战案例
学习资源
-
DeepLearning.AI课程
-
CrewAI社区
- https://community.crewai.com
- 社区讨论和问题解答
相关工具
-
CrewAI Tools
- https://github.com/joaomdmoura/crewai-tools
- 官方工具集
-
LiteLLM
- https://github.com/BerriAI/litellm
- 统一的LLM调用接口
📝 作者说明：本文基于CrewAI官方文档和GitHub仓库编写，力求准确全面地介绍CrewAI框架。由于框架持续更新，建议读者参考官方文档获取最新信息。
如果这篇文章对你有帮助，欢迎点赞👍、收藏⭐、评论💬！有任何问题也欢迎在评论区讨论。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
9
0- 0
已为社区贡献2条内容
所有评论(0)