CrewAI 基础概念 #
安装与配置 #
安装 CrewAI #
bash
# 基础安装
pip install crewai
# 安装完整版（包含工具）
pip install 'crewai[tools]'
# 使用 poetry 安装
poetry add crewai
poetry add 'crewai[tools]'
验证安装 #
python
import crewai
print(crewai.__version__)
创建项目 #
bash
# 使用 CLI 创建项目
crewai create crew my_crew
# 项目结构
my_crew/
├── src/
│ └── my_crew/
│ ├── config/
│ │ ├── agents.yaml
│ │ └── tasks.yaml
│ ├── tools/
│ │ └── custom_tool.py
│ ├── crew.py
│ └── main.py
├── pyproject.toml
└── README.md
配置语言模型 #
CrewAI 支持多种语言模型后端，配置方式统一且简单。
OpenAI 配置 #
python
import os
from crewai import Agent
os.environ["OPENAI_API_KEY"] = "your-api-key"
os.environ["OPENAI_MODEL_NAME"] = "gpt-4o"
agent = Agent(
role="分析师",
goal="分析数据",
backstory="经验丰富的分析师"
)
Claude 配置 #
python
import os
from crewai import Agent, LLM
os.environ["ANTHROPIC_API_KEY"] = "your-api-key"
llm = LLM(model="anthropic/claude-3-opus-20240229")
agent = Agent(
role="分析师",
goal="分析数据",
backstory="经验丰富的分析师",
llm=llm
)
本地模型配置 #
python
from crewai import Agent, LLM
# Ollama
llm = LLM(
model="ollama/llama3",
base_url="http://localhost:11434"
)
# vLLM
llm = LLM(
model="vllm/meta-llama/Llama-3-70b",
base_url="http://localhost:8000"
)
agent = Agent(
role="分析师",
goal="分析数据",
backstory="经验丰富的分析师",
llm=llm
)
核心概念概览 #
text
┌─────────────────────────────────────────────────────────────┐
│ CrewAI 核心概念 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Crew（团队） │ │
│ │ 协调多个 Agent 完成复杂任务 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Task（任务） │ │
│ │ 定义具体工作内容和预期输出 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Agent（智能体） │ │
│ │ Role + Goal + Backstory + Tools │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Tools（工具） │ │
│ │ 扩展 Agent 能力的函数和技能 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
Agent（智能体） #
Agent 是 CrewAI 中最核心的概念，它是一个自主的执行单元。
Agent 的组成 #
python
from crewai import Agent
agent = Agent(
role="数据分析师", # 角色：定义职责
goal="分析数据并提供洞察", # 目标：指导决策
backstory="拥有 10 年数据分析经验", # 背景：塑造性格
tools=[analysis_tool], # 工具：扩展能力
verbose=True # 详细输出
)
Agent 属性详解 #
text
┌─────────────────────────────────────────────────────────────┐
│ Agent 属性 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 核心属性（必需）： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ role - 角色名称，定义 Agent 的职责 │ │
│ │ goal - 目标，指导 Agent 的决策和行动 │ │
│ │ backstory - 背景故事，塑造 Agent 的性格和经验 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 能力属性（可选）： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ tools - 可用工具列表 │ │
│ │ llm - 使用的语言模型 │ │
│ │ memory - 是否启用记忆 │ │
│ │ verbose - 是否输出详细日志 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 执行属性（可选）： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ max_iter - 最大迭代次数 │ │
│ │ max_rpm - 每分钟最大请求数 │ │
│ │ max_execution_time - 最大执行时间 │ │
│ │ allow_delegation - 是否允许委托任务 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
创建 Agent #
python
from crewai import Agent
from crewai_tools import SerperDevTool
search_tool = SerperDevTool()
researcher = Agent(
role="高级研究分析师",
goal="发现 AI 行业的前沿趋势和发展",
backstory="""你是一位经验丰富的研究分析师，
擅长发现最新的技术趋势。你以能够找到最相关的信息
并以清晰简洁的方式呈现而闻名。""",
tools=[search_tool],
verbose=True,
allow_delegation=False
)
writer = Agent(
role="技术内容作家",
goal="创作关于 AI 行业的引人入胜的内容",
backstory="""你是一位技术精湛的作家，
对技术充满热情。你擅长将复杂的技术概念
转化为易于理解的内容。""",
verbose=True,
allow_delegation=True
)
Task（任务） #
Task 定义了 Agent 需要完成的具体工作。
Task 的组成 #
python
from crewai import Task
task = Task(
description="研究 AI 行业的最新趋势", # 任务描述
expected_output="包含 3 个主要趋势的报告", # 预期输出
agent=researcher, # 负责的 Agent
output_file="report.md" # 输出文件
)
Task 属性详解 #
text
┌─────────────────────────────────────────────────────────────┐
│ Task 属性 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 核心属性： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ description - 任务描述，详细说明要做什么 │ │
│ │ expected_output - 预期输出，定义成功的标准 │ │
│ │ agent - 负责执行任务的 Agent │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 输出属性： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ output_file - 输出文件路径 │ │
│ │ output_json - JSON 格式输出 │ │
│ │ output_pydantic - Pydantic 模型输出 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 执行属性： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ tools - 任务特定工具 │ │
│ │ async_execution - 是否异步执行 │ │
│ │ context - 上下文任务 │ │
│ │ callback - 完成回调 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
创建 Task #
python
from crewai import Task
research_task = Task(
description="""
研究 AI 行业的最新趋势。
重点关注：
1. 大语言模型的发展
2. 多模态 AI 的进展
3. AI Agent 的应用
提供每个趋势的详细分析和示例。
""",
expected_output="""
一份包含 3 个主要趋势的详细报告，
每个趋势包括：
- 趋势描述
- 关键技术
- 主要玩家
- 未来展望
""",
agent=researcher,
output_file="research_report.md"
)
write_task = Task(
description="""
根据研究报告撰写一篇博客文章。
文章应该引人入胜、信息丰富、易于理解。
""",
expected_output="""
一篇 4 段落的 Markdown 格式博客文章，
包含标题、引言、正文和结论。
""",
agent=writer,
output_file="blog_post.md"
)
Crew（团队） #
Crew 协调多个 Agent 完成一系列任务。
Crew 的组成 #
python
from crewai import Crew, Process
crew = Crew(
agents=[researcher, writer], # Agent 列表
tasks=[research_task, write_task], # 任务列表
process=Process.sequential, # 执行流程
verbose=True # 详细输出
)
Crew 属性详解 #
text
┌─────────────────────────────────────────────────────────────┐
│ Crew 属性 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 核心属性： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ agents - Agent 列表 │ │
│ │ tasks - Task 列表 │ │
│ │ process - 执行流程（sequential/hierarchical） │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 执行属性： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ verbose - 详细输出级别 │ │
│ │ memory - 是否启用记忆 │ │
│ │ cache - 是否启用缓存 │ │
│ │ max_rpm - 每分钟最大请求数 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 管理属性（层级模式）： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ manager_llm - 管理者使用的 LLM │ │
│ │ manager_agent - 自定义管理者 Agent │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
执行流程 #
python
from crewai import Crew, Process
# 顺序执行
sequential_crew = Crew(
agents=[agent1, agent2, agent3],
tasks=[task1, task2, task3],
process=Process.sequential # 任务按顺序执行
)
# 层级管理
hierarchical_crew = Crew(
agents=[agent1, agent2, agent3],
tasks=[task1, task2, task3],
process=Process.hierarchical, # Manager 协调执行
manager_llm="gpt-4o"
)
第一个 CrewAI 程序 #
完整示例 #
python
import os
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
os.environ["SERPER_API_KEY"] = "your-serper-key"
os.environ["OPENAI_API_KEY"] = "your-openai-key"
search_tool = SerperDevTool()
researcher = Agent(
role="高级研究分析师",
goal="发现 AI 行业的前沿趋势",
backstory="""你是一位经验丰富的研究分析师，
擅长发现最新的技术趋势。""",
tools=[search_tool],
verbose=True
)
writer = Agent(
role="技术内容作家",
goal="创作引人入胜的技术内容",
backstory="""你是一位技术精湛的作家，
对技术充满热情。""",
verbose=True
)
research_task = Task(
description="研究 2025 年 AI 行业的最新趋势",
expected_output="包含 3 个主要趋势的详细报告",
agent=researcher
)
write_task = Task(
description="根据研究结果撰写一篇博客文章",
expected_output="一篇 4 段落的 Markdown 格式博客文章",
agent=writer
)
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, write_task],
process=Process.sequential,
verbose=True
)
result = crew.kickoff()
print(result)
使用 YAML 配置 #
yaml
# config/agents.yaml
researcher:
role: >
{topic} 高级研究分析师
goal: >
发现 {topic} 的前沿趋势和发展
backstory: >
你是一位经验丰富的研究分析师，
擅长发现最新的技术趋势。
writer:
role: >
{topic} 技术内容作家
goal: >
创作关于 {topic} 的引人入胜的内容
backstory: >
你是一位技术精湛的作家，对技术充满热情。
yaml
# config/tasks.yaml
research_task:
description: >
研究 {topic} 的最新趋势
expected_output: >
包含 3 个主要趋势的详细报告
agent: researcher
write_task:
description: >
根据研究结果撰写一篇博客文章
expected_output: >
一篇 4 段落的 Markdown 格式博客文章
agent: writer
python
# crew.py
from crewai import Agent, Crew, Task, Process
from crewai.project import CrewBase, agent, crew, task
@CrewBase
class MyCrew:
agents_config = 'config/agents.yaml'
tasks_config = 'config/tasks.yaml'
@agent
def researcher(self) -> Agent:
return Agent(
config=self.agents_config['researcher'],
verbose=True
)
@agent
def writer(self) -> Agent:
return Agent(
config=self.agents_config['writer'],
verbose=True
)
@task
def research_task(self) -> Task:
return Task(
config=self.tasks_config['research_task']
)
@task
def write_task(self) -> Task:
return Task(
config=self.tasks_config['write_task']
)
@crew
def crew(self) -> Crew:
return Crew(
agents=self.agents,
tasks=self.tasks,
process=Process.sequential,
verbose=True
)
# 运行
result = MyCrew().crew().kickoff(inputs={'topic': 'AI'})
执行与输出 #
启动 Crew #
python
# 同步执行
result = crew.kickoff()
# 带输入参数
result = crew.kickoff(inputs={'topic': 'AI'})
# 异步执行
result = await crew.kickoff_async()
# 批量执行
results = crew.kickoff_for_each([
{'topic': 'AI'},
{'topic': 'Blockchain'},
{'topic': 'IoT'}
])
访问输出 #
python
result = crew.kickoff()
# 原始输出
print(result.raw)
# JSON 输出
if result.json_dict:
print(result.json_dict)
# Pydantic 输出
if result.pydantic:
print(result.pydantic)
# 任务输出列表
for task_output in result.tasks_output:
print(f"Task: {task_output.description}")
print(f"Output: {task_output.raw}")
# Token 使用情况
print(result.token_usage)
调试与日志 #
启用详细日志 #
python
agent = Agent(
role="分析师",
goal="分析数据",
backstory="经验丰富的分析师",
verbose=True # 启用详细日志
)
crew = Crew(
agents=[agent],
tasks=[task],
verbose=True # 启用详细日志
)
保存日志 #
python
crew = Crew(
agents=[agent],
tasks=[task],
output_log_file=True # 保存为 logs.txt
)
# 或指定文件名
crew = Crew(
agents=[agent],
tasks=[task],
output_log_file="execution.json" # JSON 格式
)
最佳实践 #
1. Agent 设计 #
python
# 好的设计：明确的角色和目标
good_agent = Agent(
role="市场研究分析师",
goal="分析市场趋势，提供投资建议",
backstory="""你有 10 年的市场研究经验，
专注于科技行业。你擅长从海量数据中
发现关键趋势和投资机会。"""
)
# 不好的设计：模糊的角色
bad_agent = Agent(
role="助手",
goal="帮助用户",
backstory="我是一个 AI 助手"
)
2. Task 设计 #
python
# 好的设计：明确的描述和预期输出
good_task = Task(
description="""
分析 2025 年 AI 芯片市场：
1. 主要厂商和产品
2. 市场份额和趋势
3. 技术发展方向
""",
expected_output="""
一份结构化的报告，包含：
- 市场概览
- 竞争格局
- 技术趋势
- 投资建议
"""
)
# 不好的设计：模糊的任务
bad_task = Task(
description="分析 AI 市场",
expected_output="一份报告"
)
3. Crew 组织 #
python
# 好的设计：合理的团队结构
good_crew = Crew(
agents=[
researcher, # 研究
analyst, # 分析
writer, # 写作
editor # 审核
],
tasks=[
research_task,
analysis_task,
write_task,
edit_task
],
process=Process.sequential
)
下一步 #
现在你已经掌握了 CrewAI 的基础概念，接下来学习 Agent 详解，深入了解 Agent 的高级用法！
最后更新：2026-04-04