🚢 CrewAI 深度讲解

一句话定义

CrewAI = 用"角色扮演"的方式组建 AI 团队，让多个 Agent 像人类团队一样协作完成任务

如果说 LangGraph 是手动画流程图，那 CrewAI 就是组建一个虚拟公司：你定义每个员工的角色、目标、技能，然后给团队分配任务，它们自己协作完成。

---

CrewAI vs LangGraph 核心区别

先跟你刚学的 LangGraph 对比，帮你建立认知锚点：

维度	LangGraph	CrewAI
核心隐喻	流程图/状态机	人类团队/公司
你要定义的	Node + Edge + State	Agent 角色 + Task 任务
控制粒度	极细（每个节点、每条边）	较粗（定义角色和任务，框架自动编排）
学习成本	⭐⭐⭐ 较高	⭐⭐ 较低
灵活度	极高（想怎么编排都行）	中等（框架有自己的执行模式）
适合场景	复杂自定义工作流	多角色协作任务
思维方式	"这个流程怎么画"	"这个团队怎么分工"

类比：

• LangGraph = 你画 Visio 流程图，每一步自己定义
• CrewAI = 你当老板，招几个人，分配任务，他们自己干

---

CrewAI 的四大核心概念

┌────────────────────────────────────────────────┐

│ CrewAI 四要素 │

│ │

│ 1. Agent (智能体) — 团队里的"员工" │

│ 2. Task (任务) — 要完成的"工作" │

│ 3. Tool (工具) — 员工的"技能/装备" │

│ 4. Crew (团队) — 把员工和任务组装起来 │

│ │

└────────────────────────────────────────────────┘

---

概念一：Agent（智能体 = 团队成员）

每个 Agent 都有自己的身份设定，就像招聘一个员工要写 JD：

python

from crewai import Agent

# 研究员

researcher = Agent(

role="高级研究分析师", # 职位

goal="发现AI领域最具突破性的技术趋势", # 目标

backstory="""你是一位在科技研究领域工作了15年的资深分析师。

你擅长从海量信息中提取关键洞察，

你对AI和大模型领域有深刻的理解。""", # 背景故事

tools=[search_tool, web_scraper], # 可用工具

llm="gpt-4o", # 使用的大模型

verbose=True, # 打印推理过程

)

# 写手

writer = Agent(

role="资深科技博客作者",

goal="将复杂的技术概念转化为引人入胜的文章",

backstory="""你是一位获奖科技博主，擅长把晦涩的技术

用通俗易懂的方式讲清楚。你的文章风格幽默且深入。""",

tools=[], # 写手不需要搜索工具

llm="gpt-4o",

)

为什么要定义 backstory？

因为 LLM 是通过 prompt 来"变成"不同角色的。role + goal + backstory 最终会被拼接成 system prompt：

内部生成的 prompt（简化版）:

"你是一位 {role}。

你的目标是: {goal}。

背景信息: {backstory}。

你可以使用以下工具: {tools}。

请认真完成分配给你的任务。"

本质上就是更高级的 prompt engineering，用角色设定来引导 LLM 的行为。

---

概念二：Task（任务 = 工作分配）

Task 是你分配给 Agent 的具体工作：

python

from crewai import Task

# 研究任务

research_task = Task(

description="""调研2026年最重要的3个AI技术趋势。

对每个趋势，需要包含：

1. 趋势名称和简要描述

2. 主要推动公司/机构

3. 潜在的商业影响

4. 技术成熟度评估""",

expected_output="一份包含3个AI趋势的详细研究报告，格式为markdown",

agent=researcher, # 指定谁来做

)

# 写作任务

write_task = Task(

description="""基于研究报告，撰写一篇面向大众的科技博客文章。

要求：

1. 标题要吸引眼球

2. 语言通俗易懂，避免过多术语

3. 每个趋势用生动的案例说明

4. 文章长度约1500字""",

expected_output="一篇完整的中文科技博客文章",

agent=writer, # 指定谁来做

context=[research_task], # 🔑 依赖研究任务的结果

)

关键字段 context：它定义了任务之间的依赖关系。write_task 依赖 research_task 的输出，CrewAI 会自动把研究结果传给写手。

---

概念三：Tool（工具 = 技能装备）

跟 LangChain 的工具概念一样，但 CrewAI 有自己的封装方式：

python

from crewai.tools import tool

# 方式一：用装饰器定义

@tool("搜索互联网")

def search_tool(query: str) -> str:

"""在互联网上搜索最新信息。"""

# 实际实现...

return f"关于 '{query}' 的搜索结果..."

# 方式二：使用 CrewAI 内置工具

from crewai_tools import (

SerperDevTool, # Google 搜索

ScrapeWebsiteTool, # 网页抓取

FileReadTool, # 文件读取

DirectoryReadTool, # 目录读取

PDFSearchTool, # PDF 搜索

)

search_tool = SerperDevTool()

scrape_tool = ScrapeWebsiteTool()

# 方式三：直接用 LangChain 的工具（兼容）

from langchain_community.tools import DuckDuckGoSearchRun

langchain_tool = DuckDuckGoSearchRun()

---

概念四：Crew（团队 = 把一切组装起来）

python

from crewai import Crew, Process

crew = Crew(

agents=[researcher, writer], # 团队成员

tasks=[research_task, write_task], # 任务列表

process=Process.sequential, # 执行模式（顺序执行）

verbose=True, # 打印详细日志

)

# 🚀 启动团队！

result = crew.kickoff()

print(result)

---

完整的可运行示例

把上面所有概念串起来：

python

"""

CrewAI 完整示例：AI趋势研究 + 博客写作团队

"""

from crewai import Agent, Task, Crew, Process

from crewai.tools import tool

# ========== 1. 定义工具 ==========

@tool("搜索互联网")

def search_internet(query: str) -> str:

"""搜索互联网获取最新信息。"""

# 实际项目中接入搜索 API

return f"搜索 '{query}' 的结果: AI Agent框架在2026年快速发展..."

# ========== 2. 定义 Agent ==========

researcher = Agent(

role="AI技术研究员",

goal="找到最新最有价值的AI技术趋势",

backstory="你是AI领域的资深研究员，拥有15年经验，擅长技术分析",

tools=[search_internet],

llm="gpt-4o",

verbose=True,

)

writer = Agent(

role="科技博客作者",

goal="把复杂技术写成人人都能懂的好文章",

backstory="你是知名科技博主，文风幽默且深入浅出",

llm="gpt-4o",

verbose=True,

)

reviewer = Agent(

role="内容质量审核员",

goal="确保文章准确、有深度、没有事实错误",

backstory="你是出版社的资深编辑，对内容质量要求极高",

llm="gpt-4o",

verbose=True,

)

# ========== 3. 定义 Task ==========

research_task = Task(

description="调研并总结2026年最重要的3个AI技术趋势",

expected_output="结构化的研究报告（markdown格式）",

agent=researcher,

)

write_task = Task(

description="基于研究报告撰写一篇1500字的科技博客",

expected_output="一篇完整的博客文章",

agent=writer,

context=[research_task], # 依赖研究结果

)

review_task = Task(

description="审核文章质量，检查事实准确性，提出修改建议",

expected_output="审核意见和最终定稿",

agent=reviewer,

context=[research_task, write_task], # 依赖研究报告和文章

)

# ========== 4. 组建团队并执行 ==========

crew = Crew(

agents=[researcher, writer, reviewer],

tasks=[research_task, write_task, review_task],

process=Process.sequential,

verbose=True,

)

result = crew.kickoff()

print("=== 最终输出 ===")

print(result)

---

执行流程可视化

crew.kickoff() 被调用后发生了什么：

Step 1: 🔬 研究员开工

┌──────────────────────────────────────────────┐

│ Agent: AI技术研究员 │

│ Task: 调研AI趋势 │

│ │

│ [思考] 我需要搜索最新的AI动态 │

│ [行动] 调用 search_internet("2026 AI趋势") │

│ [观察] 搜索返回了结果... │

│ [思考] 我需要再深入搜索每个趋势 │

│ [行动] 调用 search_internet("AI Agent框架") │

│ [观察] ... │

│ [最终] 输出研究报告 │

└───────────────────────┬──────────────────────┘

│ 研究报告传递给下一个任务

▼

Step 2: ✍️ 写手开工

┌──────────────────────────────────────────────┐

│ Agent: 科技博客作者 │

│ Task: 写博客文章 │

│ Context: [研究员的报告] ← 自动注入 │

│ │

│ [思考] 基于研究报告，我要写一篇通俗的文章 │

│ [最终] 输出博客文章 │

└───────────────────────┬──────────────────────┘

│ 文章传递给下一个任务

▼

Step 3: 📋 审核员开工

┌──────────────────────────────────────────────┐

│ Agent: 内容质量审核员 │

│ Task: 审核文章 │

│ Context: [研究报告 + 博客文章] ← 自动注入 │

│ │

│ [思考] 我来核实文章中的事实... │

│ [最终] 输出审核意见和修改后的定稿 │

└──────────────────────────────────────────────┘

│

▼

返回最终结果

---

两种执行模式：Sequential vs Hierarchical

模式一：Sequential（顺序执行）

python

crew = Crew(

process=Process.sequential, # 按任务列表顺序，一个接一个

...

)

研究员 → 写手 → 审核员

| | |

串行执行，结果依次传递

适合：流水线型任务，前一步的输出是后一步的输入。

模式二：Hierarchical（层级执行）

python

crew = Crew(

process=Process.hierarchical, # 有一个"经理"自动分配任务

manager_llm="gpt-4o", # 经理用的模型

...

)

┌──────────┐

│ 🧑‍💼 经理 │ 自动协调，决定谁做什么、什么顺序

└────┬─────┘

┌──────┼──────┐

▼ ▼ ▼

研究员 写手 审核员

经理（Manager）是 CrewAI自动创建的，它会：

1. 分析所有任务
2. 决定执行顺序
3. 把任务分配给合适的 Agent
4. 审查结果质量
5. 必要时让 Agent 重做

适合：不确定最佳执行顺序，让 AI 经理自己判断。

---

Flow（工作流）— CrewAI 的高级编排

当你需要比 Crew 更精细的控制时，用 Flow：

python

from crewai.flow.flow import Flow, start, listen, router

class ContentPipeline(Flow):

"""内容生产流水线"""

# 初始状态

initial_state = {

"topic": "",

"research": "",

"article": "",

"quality_score": 0,

}

@start() # 入口节点

def get_topic(self):

self.state["topic"] = "AI Agent 2026趋势"

return self.state["topic"]

@listen(get_topic) # 监听 get_topic 完成后触发

def do_research(self, topic):

# 调用研究 Crew

result = research_crew.kickoff(inputs={"topic": topic})

self.state["research"] = result

return result

@listen(do_research)

def write_article(self, research):

# 调用写作 Crew

result = writing_crew.kickoff(inputs={"research": research})

self.state["article"] = result

return result

@router(write_article) # 🔑 路由节点：根据条件分支

def check_quality(self, article):

score = evaluate_quality(article)

self.state["quality_score"] = score

if score >= 0.9:

return "publish" # 质量够好，发布

else:

return "revise" # 质量不够，返工

@listen("publish")

def publish_article(self):

print(f"✅ 发布文章！质量分: {self.state['quality_score']}")

return self.state["article"]

@listen("revise")

def revise_article(self):

# 返工...可以再调用一个 Crew 来修改

revised = revision_crew.kickoff(inputs={...})

self.state["article"] = revised

# 修改后重新检查质量（形成循环）

return self.check_quality(revised)

# 运行 Flow

pipeline = ContentPipeline()

final = pipeline.kickoff()

Flow vs Crew 的关系：

• Crew = 一个团队执行一组相关任务（比如"研究团队"）
• Flow = 编排多个 Crew / 步骤的工作流（比如"整个内容生产流程"）

Flow 里面可以包含多个 Crew:

Flow: 内容生产流程

├── Step 1: research_crew.kickoff() ← 一个 Crew

├── Step 2: writing_crew.kickoff() ← 另一个 Crew

├── Step 3: 质量检查 ← 普通函数

└── Step 4: revision_crew.kickoff() ← 又一个 Crew

---

CrewAI 的记忆系统

CrewAI 内置了三种记忆，不需要你手动管理：

python

crew = Crew(

agents=[...],

tasks=[...],

memory=True, # 🔑 开启记忆

# 以下是自动管理的:

# short_term_memory → 当前任务执行中的上下文

# long_term_memory → 跨次执行的经验学习

# entity_memory → 记住关键实体（人名、公司名等）

)

┌─────────────────────────────────────────────┐

│ CrewAI 三层记忆 │

│ │

│ 短期记忆 (Short-term) │

│ ├─ 范围: 当前这一次 kickoff │

│ ├─ 内容: 任务之间传递的上下文 │

│ └─ 类比: 工作中的"当前讨论" │

│ │

│ 长期记忆 (Long-term) │

│ ├─ 范围: 跨多次 kickoff │

│ ├─ 内容: Agent 从过去执行中学到的经验 │

│ └─ 类比: 员工的"工作经验" │

│ │

│ 实体记忆 (Entity) │

│ ├─ 范围: 跨多次 kickoff │

│ ├─ 内容: 识别并记住关键实体的信息 │

│ └─ 类比: 员工的"客户关系簿" │

│ │

└─────────────────────────────────────────────┘

---

CrewAI vs LangGraph vs LangChain 三者关系总结

你要做什么？

"给一个 AI 几个工具让它回答问题"

└─ LangChain create_agent ✅ （最简单）

"自己精确控制每一步的执行流程"

└─ LangGraph ✅ （最灵活）

"组建 AI 团队，多角色协作完成复杂任务"

└─ CrewAI ✅ （最直觉）

	LangChain create_agent	LangGraph	CrewAI
核心隐喻	一个员工 + 工具箱	流程图	一个团队
Agent 数量	单 Agent	单/多 Agent	多 Agent
控制方式	中间件	图 (Node+Edge)	角色 + 任务
学习成本	⭐	⭐⭐⭐	⭐⭐
灵活度	中	极高	中
最佳场景	对话/问答	复杂工作流	多角色协作
底层关系	基于 LangGraph	独立框架	独立框架

---

安装和快速上手

bash

# 安装

pip install crewai crewai-tools

# 或者用 crewai CLI 快速创建项目

pip install crewai

crewai create crew my_project # 自动生成项目模板

生成的项目结构：

my_project/

├── src/my_project/

│ ├── config/

│ │ ├── agents.yaml # Agent 定义（用 YAML 配置）

│ │ └── tasks.yaml # Task 定义

│ ├── crew.py # Crew 组装逻辑

│ ├── main.py # 入口文件

│ └── tools/

│ └── custom_tool.py # 自定义工具

├── pyproject.toml

└── README.md

CrewAI 推荐用 YAML 定义 Agent 和 Task，Python 只做组装，分离配置和逻辑。