作者：PySuper | 来源：zhengxingtao.com
如果你曾经纠结过"怎么让多个 AI Agent 高效协作"，CrewAI 给了一个最直觉的答案——像管团队一样管 Agent。角色、目标、背景故事，三个要素定义一个 Agent；任务、流程、团队，三层抽象编排整个系统。这篇文章从最小 Crew 到 5 角色产品团队，手把手带你走一遍。
一、CrewAI 是什么
CrewAI 是一个角色制多 Agent 框架，由 João Moura 创建，MIT 开源，完全独立于 LangChain。
核心卖点：像定义团队一样定义 Agent 系统。
python
# 传统方式：用代码逻辑编排 Agent
if research_done:
writer.write(research_result)
if writer_done:
reviewer.review(writer_result)
# CrewAI 方式：用角色直觉编排 Agent
researcher = Agent(role="研究员", goal="...", backstory="...")
writer = Agent(role="写手", goal="...", backstory="...")
crew = Crew(agents=[researcher, writer], tasks=[...], process=Process.sequential)
result = crew.kickoff()
1.1 关键数据
表格
1.2 与其他框架的定位差异
plaintext
复杂度 ▲
│ LangGraph (状态机，精细控制)
│ ●
│
│ Google ADK (代码优先，GCP)
│ ●
│
│ CrewAI (角色制，直觉映射)
│ ●
│
│ OpenAI Agents SDK (轻量 Handoff)
│ ●
└──────────────────────────────────► 上手速度
二、四大核心概念
2.1 Agent（角色）
Agent 是 CrewAI 的核心抽象。每个 Agent 有三个灵魂要素：
python
from crewai import Agent, LLM
researcher = Agent(
role="Senior Battery Technology Analyst", # 角色：你是谁
goal="Uncover the latest breakthroughs in solid-state batteries", # 目标：你要做什么
backstory="""You are a veteran technology analyst with a Ph.D.
in materials science. You excel at finding obscure technical data
and identifying commercialization timelines.""", # 背景故事：你的经历
llm=LLM(model="gpt-4o", temperature=0.3),
tools=[search_tool], # 可用工具
verbose=True,
allow_delegation=False, # 是否允许委派
max_iter=25, # 最大推理迭代
memory=True # 启用记忆
)
为什么 backstory 重要？
plaintext
没有 backstory:
Agent 收到 "研究电池技术" → 输出泛泛的百科内容
有 backstory:
Agent 收到 "研究电池技术" → 输出具体的商业化时间线、
技术参数对比、市场数据，因为它"是材料科学博士"
2.2 Task（任务）
Task 是 Agent 要完成的具体工作：
python
from crewai import Task
research_task = Task(
description="""Investigate the top 3 companies leading solid-state
battery commercialization. Find specific data on energy density
improvements and projected mass production dates.
Topic: {topic}
Audience: {audience}""",
expected_output="A detailed bulleted list with data points and timelines",
agent=researcher, # 分配给哪个 Agent
context=[previous_task], # 接收哪些任务的输出作为上下文
output_file="research.md", # 输出保存到文件
async_execution=False # 是否异步执行
)
2.3 Crew（团队）
Crew 是 Agent + Task 的容器，定义执行环境：
python
from crewai import Crew, Process
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, write_task],
process=Process.sequential, # 执行模式
verbose=True,
memory=True, # 启用记忆系统
full_output=True # 返回完整输出
)
2.4 Process（流程）
plaintext
┌─────────────────────────────────────────────────────────────┐
│ Process.sequential (串行) │
│ │
│ Task 1 ──→ Task 2 ──→ Task 3 ──→ Done │
│ (Agent A) (Agent B) (Agent C) │
│ │
│ 特点：顺序执行，输出自动传递 │
│ 适合：流水线式工作流 │
├─────────────────────────────────────────────────────────────┤
│ Process.hierarchical (层级) │
│ │
│ ┌─────────────┐ │
│ │ Manager │ ← 自动创建的管理者 Agent │
│ └──────┬──────┘ │
│ ┌─────────┼─────────┐ │
│ ▼ ▼ ▼ │
│ Agent A Agent B Agent C │
│ │
│ 特点：Manager 动态分配任务，审核输出 │
│ 适合：复杂场景，任务顺序不确定 │
└─────────────────────────────────────────────────────────────┘
三、实战1：最小 Crew —— 研究员 + 写手
3.1 目标
输入一个主题，自动生成一篇技术调研报告。
3.2 完整代码
python
"""
最小 Crew 实战：研究员 + 写手
生成一篇技术调研报告
"""
import os
from crewai import Agent, Task, Crew, Process, LLM
from crewai_tools import SerperDevTool
# ============================================================
# 1. 配置
# ============================================================
# 设置 API Key（实际使用中请用环境变量）
# os.environ["OPENAI_API_KEY"] = "sk-..."
# os.environ["SERPER_API_KEY"] = "..."
llm = LLM(model="gpt-4o", temperature=0.3)
search_tool = SerperDevTool()
# ============================================================
# 2. 定义 Agent
# ============================================================
researcher = Agent(
role="Senior Technology Research Analyst",
goal="Conduct thorough, accurate research on the assigned topic with credible sources",
backstory="""You are a veteran research analyst with 15 years of experience
in technology trends analysis. You have a Ph.D. in Computer Science and
previously worked at Gartner as a principal analyst.
Your strengths:
- Finding obscure but reliable technical data
- Distinguishing signal from noise in AI hype
- Providing quantitative comparisons with source citations
You NEVER fabricate information. If you cannot verify a claim,
you explicitly flag it as 'UNVERIFIED'.""",
tools=[search_tool],
llm=llm,
verbose=True,
allow_delegation=False,
max_iter=25
)
writer = Agent(
role="Technical Content Strategist",
goal="Transform research findings into clear, engaging, well-structured reports",
backstory="""You are a skilled technical writer who spent 10 years at
O'Reilly Media. You translate complex research into compelling narratives
for technical audiences.
Your writing style:
- Short paragraphs with concrete examples
- Data-driven arguments with specific numbers
- No jargon without definition
- Every claim backed by the research""",
llm=llm,
verbose=True,
allow_delegation=False,
max_iter=20
)
# ============================================================
# 3. 定义 Task
# ============================================================
research_task = Task(
description="""Conduct comprehensive research on: {topic}
Your research should cover:
1. Current state of the art (2025-2026)
2. Key players and their approaches
3. Technical innovations and breakthroughs
4. Market data: size, growth rate, projections
5. Challenges and open problems
6. Future outlook (6-12 months)
Use the search tool extensively. Aim for at least 8-10 distinct findings
with specific data points.""",
expected_output="""A structured research brief containing:
- Executive summary (3-5 sentences)
- Key findings (8-10 items, each with source)
- Data points and comparisons
- Identified gaps in current solutions""",
agent=researcher
)
write_task = Task(
description="""Based on the research findings, write a comprehensive
report on: {topic}
Structure:
1. **Executive Summary** - Key takeaway in 100 words
2. **Current Landscape** - What's happening now
3. **Key Players & Approaches** - Who's doing what
4. **Technical Deep Dive** - How it works under the hood
5. **Market Analysis** - Numbers and trends
6. **Challenges** - What's still hard
7. **Future Outlook** - Where this is going
8. **Conclusion** - Actionable takeaways
Write in Markdown format. Target length: 1500-2000 words.""",
expected_output="A well-structured Markdown report (1500-2000 words)",
agent=writer,
context=[research_task] # 接收研究任务的输出
)
# ============================================================
# 4. 组装 Crew
# ============================================================
research_crew = Crew(
agents=[researcher, writer],
tasks=[research_task, write_task],
process=Process.sequential,
verbose=True,
full_output=True
)
# ============================================================
# 5. 运行
# ============================================================
if __name__ == "__main__":
result = research_crew.kickoff(
inputs={"topic": "MCP Protocol: How AI Agents Connect to Tools in 2026"}
)
# 打印最终结果
print("\n" + "=" * 60)
print("FINAL REPORT")
print("=" * 60)
print(result.raw)
# 保存到文件
with open("research_report.md", "w") as f:
f.write(result.raw)
print(f"\nReport saved to research_report.md")
3.3 运行流程
plaintext
kickoff()
│
├── Task 1: research_task
│ ├── Agent: researcher
│ ├── 工具调用: SerperDevTool (搜索 "MCP Protocol AI agents 2026")
│ ├── 推理: 分析搜索结果，提取关键发现
│ └── 输出: 结构化研究简报
│
├── Task 2: write_task
│ ├── Agent: writer
│ ├── 上下文: research_task 的输出
│ ├── 推理: 组织研究内容为报告结构
│ └── 输出: Markdown 格式的完整报告
│
└── Crew 完成
└── result.raw = 最终报告
四、实战2：复杂 Crew —— 5 角色产品开发团队
4.1 目标
模拟一个完整的产品开发流程：从需求到设计到开发到测试到部署。
4.2 团队结构
plaintext
┌─────────────────────────────────────────────────────────┐
│ Product Dev Crew │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 产品经理 │ │ 架构师 │ │ 开发者 │ │
│ │ PM Agent │ │Arch Agent│ │ Dev Agent│ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ ┌──────────┐ ┌──────────┐ │
│ │ 测试 │ │ 运维 │ │
│ │ QA Agent │ │ Ops Agent│ │
│ └──────────┘ └──────────┘ │
│ │
│ Process: Hierarchical (Manager 协调) │
└─────────────────────────────────────────────────────────┘
4.3 完整代码
python
"""
5 角色产品开发团队
产品经理 + 架构师 + 开发者 + 测试 + 运维
Hierarchical 模式，Manager 动态分配
"""
import os
from crewai import Agent, Task, Crew, Process, LLM
from crewai_tools import SerperDevTool
# ============================================================
# 配置
# ============================================================
llm = LLM(model="gpt-4o", temperature=0.2)
search_tool = SerperDevTool()
# ============================================================
# 定义 Agent（5 个角色）
# ============================================================
pm_agent = Agent(
role="Senior Product Manager",
goal="Define clear product requirements with measurable success criteria",
backstory="""You are a seasoned PM with 12 years at top tech companies.
You excel at translating vague ideas into precise specifications.
You think in terms of user stories, acceptance criteria, and KPIs.
You always start with 'why' before 'what'.""",
tools=[search_tool],
llm=llm,
verbose=True,
allow_delegation=True,
max_iter=20
)
architect_agent = Agent(
role="Principal System Architect",
goal="Design scalable, maintainable system architecture that meets requirements",
backstory="""You are a principal architect who designed systems handling
millions of users. You think in terms of trade-offs: consistency vs
availability, simplicity vs flexibility, cost vs performance.
You document decisions with ADRs (Architecture Decision Records).""",
llm=llm,
verbose=True,
allow_delegation=True,
max_iter=20
)
developer_agent = Agent(
role="Senior Full-Stack Developer",
goal="Implement clean, well-tested code that meets specifications",
backstory="""You are a senior developer with expertise in Python,
TypeScript, and cloud-native development. You follow SOLID principles,
write comprehensive tests, and document your code.
You prefer simplicity over cleverness.""",
llm=llm,
verbose=True,
allow_delegation=True,
max_iter=25
)
qa_agent = Agent(
role="Senior QA Engineer",
goal="Ensure the product meets quality standards through systematic testing",
backstory="""You are a QA engineer with a passion for breaking things
(in a good way). You think about edge cases, error paths, and
performance boundaries. You create test plans that cover:
- Happy path scenarios
- Error handling
- Boundary conditions
- Performance requirements""",
llm=llm,
verbose=True,
allow_delegation=True,
max_iter=15
)
ops_agent = Agent(
role="Senior DevOps Engineer",
goal="Design deployment pipeline and monitoring strategy for production readiness",
backstory="""You are a DevOps engineer who has managed production systems
at scale. You think about:
- Deployment strategies (blue-green, canary, rolling)
- Monitoring and alerting
- Rollback procedures
- Infrastructure as Code
You believe in 'you build it, you run it'.""",
llm=llm,
verbose=True,
allow_delegation=True,
max_iter=15
)
# ============================================================
# 定义 Task（5 个任务）
# ============================================================
requirements_task = Task(
description="""Analyze the product request and create a detailed PRD
(Product Requirements Document).
Product Request: {product_idea}
The PRD should include:
1. Problem Statement - What problem are we solving?
2. Target Users - Who will use this?
3. User Stories - At least 5 user stories with acceptance criteria
4. Success Metrics - Measurable KPIs
5. Scope - What's in and what's out (MVP vs Future)
6. Risks and Assumptions
Use the search tool to research market context if needed.""",
expected_output="A comprehensive PRD in Markdown format",
agent=pm_agent
)
architecture_task = Task(
description="""Based on the PRD, design the system architecture.
The architecture document should include:
1. System Overview - High-level component diagram (text description)
2. Technology Stack - Languages, frameworks, databases, cloud services
3. API Design - Key endpoints and data models
4. Data Flow - How data moves through the system
5. Scalability Strategy - How to handle growth
6. Security Considerations - Auth, data protection, compliance
7. ADRs - At least 2 Architecture Decision Records
Reference the PRD for requirements.""",
expected_output="A detailed architecture document with ADRs",
agent=architect_agent,
context=[requirements_task]
)
development_task = Task(
description="""Based on the architecture, provide implementation guidance.
For the MVP scope, provide:
1. Project Structure - Directory layout and module organization
2. Core Implementation - Key code structures (classes, functions, interfaces)
3. Database Schema - Tables, indexes, relationships
4. API Implementation - Request/response patterns
5. Test Strategy - Unit, integration, and E2E test approach
6. Code Examples - At least 2 critical code snippets
Focus on the MVP scope defined in the PRD.""",
expected_output="Implementation guide with code examples",
agent=developer_agent,
context=[requirements_task, architecture_task]
)
testing_task = Task(
description="""Based on the PRD and implementation guide, create a test plan.
The test plan should include:
1. Test Strategy - Overall approach and coverage targets
2. Unit Test Plan - Key test cases for core logic
3. Integration Test Plan - API and service integration tests
4. E2E Test Scenarios - Complete user flow tests
5. Performance Test Criteria - Response time, throughput targets
6. Security Test Checklist - OWASP top 10 coverage
7. Bug Report Template - Standardized format
Define pass/fail criteria for each test category.""",
expected_output="Comprehensive test plan with test cases",
agent=qa_agent,
context=[requirements_task, development_task]
)
deployment_task = Task(
description="""Based on the architecture and implementation, create a
deployment and operations plan.
The plan should include:
1. Deployment Architecture - Infrastructure setup
2. CI/CD Pipeline - Build, test, deploy stages
3. Deployment Strategy - Blue-green / canary / rolling
4. Monitoring & Alerting - Metrics, dashboards, alert rules
5. Incident Response - Runbook for common failures
6. Rollback Procedure - How to quickly revert
7. Cost Estimation - Monthly infrastructure cost
Consider the scalability requirements from the architecture.""",
expected_output="Production deployment and operations plan",
agent=ops_agent,
context=[architecture_task, development_task]
)
# ============================================================
# 组装 Crew
# ============================================================
product_crew = Crew(
agents=[pm_agent, architect_agent, developer_agent, qa_agent, ops_agent],
tasks=[
requirements_task,
architecture_task,
development_task,
testing_task,
deployment_task
],
process=Process.hierarchical, # 层级模式，Manager 动态分配
manager_llm=LLM(model="gpt-4o", temperature=0.1),
verbose=True,
memory=True,
full_output=True
)
# ============================================================
# 运行
# ============================================================
if __name__ == "__main__":
result = product_crew.kickoff(
inputs={
"product_idea": "AI-powered code review tool that automatically "
"analyzes pull requests, identifies bugs and security "
"vulnerabilities, and suggests improvements"
}
)
print("\n" + "=" * 60)
print("PRODUCT DEVELOPMENT COMPLETE")
print("=" * 60)
# 打印每个任务的输出
for task_output in result.tasks_output:
print(f"\n--- {task_output.description[:50]}... ---")
print(task_output.raw[:500] + "...")
五、Process 模式详解
5.1 Sequential（串行）
python
# 串行模式：任务按顺序执行
crew = Crew(
agents=[agent_a, agent_b, agent_c],
tasks=[task_1, task_2, task_3],
process=Process.sequential
)
# 执行流程：
# task_1 (agent_a) → task_2 (agent_b) → task_3 (agent_c)
# 每个 task 的输出自动作为下一个 task 的上下文
适用场景：
表格
优势：简单、可预测、易调试
劣势：不够灵活，无法处理动态依赖
5.2 Hierarchical（层级 / Supervisor）
python
# 层级模式：Manager Agent 动态分配
crew = Crew(
agents=[researcher, writer, reviewer],
tasks=[task_1, task_2, task_3],
process=Process.hierarchical,
manager_llm=LLM(model="gpt-4o") # 必须指定 Manager 的 LLM
)
# 执行流程：
# Manager 接收任务 → 决定分配给谁 → 审核输出 → 决定下一步
# Manager 可以：重新分配、要求返工、跳过任务
适用场景：
表格
优势：灵活、动态、有审核
劣势：Manager 额外消耗 Token、不确定性高
5.3 如何选择
plaintext
任务顺序确定吗？
├── 是 → Sequential ✅
│ ├── 任务 < 5 个 → Sequential
│ └── 任务 > 5 个 → 考虑 Hierarchical（管理复杂度）
│
└── 否 → Hierarchical ✅
├── 需要动态规划 → Hierarchical
├── 需要审核节点 → Hierarchical
└── 预算有限 → 考虑拆成多个 Sequential Crew
六、工具集成
6.1 @tool 装饰器：自定义工具
python
from crewai.tools import tool
@tool("database_query")
def query_database(sql: str) -> str:
"""Execute a SQL query on the analytics database.
Args:
sql: The SQL query to execute
Returns:
Query results as a formatted string
"""
import psycopg2
conn = psycopg2.connect(
host="localhost",
database="analytics",
user="reader",
password=" ***"
)
cursor = conn.cursor()
cursor.execute(sql)
results = cursor.fetchall()
conn.close()
return str(results)
@tool("send_notification")
def send_notification(message: str, channel: str = "slack") -> str:
"""Send a notification to a specified channel.
Args:
message: The notification message
channel: The channel to send to (slack, email, teams)
Returns:
Confirmation message
"""
# 实际实现中调用 Slack/Email/Teams API
return f"Notification sent to {channel}: {message[:50]}..."
# 使用自定义工具
analyst = Agent(
role="Data Analyst",
goal="Analyze data and send insights",
backstory="...",
tools=[query_database, send_notification]
)
6.2 MCP Server 对接
CrewAI 1.8+ 支持通过 MCP 协议对接外部工具：
python
from crewai.tools import MCPToolAdapter
# 连接远程 MCP Server
mcp_tool = MCPToolAdapter(
server_url="http://localhost:5000", # MCP Server 地址
tool_name="web_search" # 要使用的工具名
)
# 将 MCP 工具分配给 Agent
researcher = Agent(
role="Researcher",
goal="...",
backstory="...",
tools=[mcp_tool]
)
6.3 内置工具一览
python
from crewai_tools import (
SerperDevTool, # Google 搜索
ScrapeWebsiteTool, # 网页抓取
FileReadTool, # 文件读取
DirectoryReadTool, # 目录读取
CSVSearchTool, # CSV 搜索
JSONSearchTool, # JSON 搜索
MDXSearchTool, # MDX 搜索
CodeDocsSearchTool, # 代码文档搜索
GithubSearchTool, # GitHub 搜索
SeleniumScraper, # 浏览器自动化
DallETool, # 图片生成
PDFSearchTool, # PDF 搜索
)
七、记忆系统
7.1 三层记忆架构
plaintext
┌─────────────────────────────────────────────────────────┐
│ 记忆系统 │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 短期记忆 │ │ 长期记忆 │ │ 实体记忆 │ │
│ │ Short-term │ │ Long-term │ │ Entity │ │
│ │ │ │ │ │ │ │
│ │ 当前对话 │ │ 跨次运行 │ │ 关键实体 │ │
│ │ 自动管理 │ │ 向量存储 │ │ 人/组织/概念│ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
│ 作用域层级: │
│ /project/alpha │
│ ├── /agent/researcher │
│ │ ├── findings │
│ │ └── sources │
│ └── /agent/writer │
│ ├── style_preferences │
│ └── completed_reports │
└─────────────────────────────────────────────────────────┘
7.2 使用方式
python
# 启用所有记忆类型
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, write_task],
memory=True, # 一键启用三层记忆
verbose=True
)
# 记忆的作用：
# 1. 短期记忆：研究员的发现自动传递给写手
# 2. 长期记忆：第二次运行时，Agent 记得上次的研究方向
# 3. 实体记忆：Agent 记住关键公司、人物、技术的关联
7.3 记忆的注意事项
python
# 问题1：过时记忆干扰
# 解决：定期清理，或为新任务禁用记忆
crew = Crew(
agents=[researcher],
tasks=[new_topic_task],
memory=False # 新主题，不要受旧记忆影响
)
# 问题2：记忆膨胀
# 解决：限制记忆条目数
# CrewAI 的记忆系统使用 LLM 分析内容，
# 自动推断范围、类别和重要性。
# 回忆使用自适应深度评分：语义相似度 + 时效性 + 重要性
# 问题3：跨 Crew 记忆
# 解决：使用同一个 embedder 配置
# 目前跨 Crew 共享记忆需要手动配置向量存储
八、与 LangGraph 的对比
8.1 核心差异
表格
8.2 同一个任务的不同思维
任务 ：研究 AI 趋势 → 写报告 → 审核
CrewAI 思维 ：
plaintext
角色定义：
├── 研究员 (role: "Senior Analyst")
├── 写手 (role: "Technical Writer")
└── 审核员 (role: "QA Editor")
任务分配：
├── 研究员做研究
├── 写手基于研究写报告
└── 审核员检查质量
流程：Sequential，自动传递
LangGraph 思维 ：
plaintext
状态定义：
├── topic: str
├── research_notes: str
├── draft_report: str
├── review_status: str
└── iteration: int
节点定义：
├── research_node → 调用 LLM + 工具
├── write_node → 调用 LLM
└── review_node → 调用 LLM
边定义：
├── research_node → write_node
├── write_node → review_node
└── review_node → END (if approved) / write_node (if revision needed)
8.3 什么时候用哪个
plaintext
你的场景？
│
├── 任务流程是线性的？
│ └── 是 → CrewAI ✅（更直觉，更少代码）
│
├── 需要条件路由/循环？
│ └── 是 → LangGraph ✅（CrewAI 不支持自定义路由）
│
├── 需要 Human-in-the-Loop？
│ └── 精细控制 → LangGraph ✅
│ └── 简单审批 → CrewAI (Hierarchical + 委派)
│
├── 需要 Checkpointing/断点续跑？
│ └── 是 → LangGraph ✅（CrewAI 目前不支持）
│
├── 团队有非技术人员？
│ └── 是 → CrewAI ✅（角色描述更易理解）
│
└── 生产环境需要可观测性？
└── 是 → LangGraph + LangSmith ✅
九、生产部署
9.1 异步执行
python
import asyncio
from crewai import Crew, Process
async def run_crews_concurrently():
"""并发运行多个 Crew"""
# 创建多个 Crew
ai_crew = Crew(agents=[...], tasks=[...], process=Process.sequential)
biotech_crew = Crew(agents=[...], tasks=[...], process=Process.sequential)
fintech_crew = Crew(agents=[...], tasks=[...], process=Process.sequential)
# 并发执行
results = await asyncio.gather(
ai_crew.kickoff_async(inputs={"topic": "AI trends"}),
biotech_crew.kickoff_async(inputs={"topic": "Biotech trends"}),
fintech_crew.kickoff_async(inputs={"topic": "Fintech trends"}),
)
return results
if __name__ == "__main__":
results = asyncio.run(run_crews_concurrently())
for result in results:
print(result.raw[:200])
9.2 回调与事件监听
python
from crewai import Crew
from crewai.utilities.events import (
CrewKickoffStartedEvent,
CrewKickoffCompletedEvent,
AgentExecutionStartedEvent,
AgentExecutionCompletedEvent,
TaskStartedEvent,
TaskCompletedEvent,
ToolUsageStartedEvent,
ToolUsageCompletedEvent,
)
class MyEventListener:
"""自定义事件监听器"""
def on_crew_started(self, event: CrewKickoffStartedEvent):
print(f"Crew started: {event.crew_name}")
def on_agent_started(self, event: AgentExecutionStartedEvent):
print(f"Agent {event.agent_name} started task")
def on_tool_used(self, event: ToolUsageCompletedEvent):
print(f"Tool {event.tool_name} completed in {event.duration}s")
def on_task_completed(self, event: TaskCompletedEvent):
print(f"Task completed: {event.task_description[:50]}...")
def on_crew_completed(self, event: CrewKickoffCompletedEvent):
print(f"Crew completed in {event.duration}s")
# 注册监听器
crew = Crew(
agents=[...],
tasks=[...],
event_listeners=[MyEventListener()]
)
9.3 错误恢复
python
from crewai import Crew, Task, Agent
from crewai.utilities.logger import Logger
# 1. 设置最大迭代次数，防止无限循环
agent = Agent(
role="Researcher",
goal="...",
backstory="...",
max_iter=25, # 最大推理迭代次数
retry_on_fail=True # 失败时自动重试
)
# 2. 任务级别的超时
task = Task(
description="...",
expected_output="...",
agent=agent,
timeout=300 # 5 分钟超时
)
# 3. 使用 try-except 处理 Crew 级别异常
try:
result = crew.kickoff(inputs={"topic": "..."})
except Exception as e:
print(f"Crew execution failed: {e}")
# 可以尝试部分重试或降级执行
9.4 CrewAI AMP Suite（企业套件）
plaintext
CrewAI AMP = 企业级增强
├── 追踪与可观测性 ── 生产环境必备
├── 统一控制面板 ── 管理多个 Crew
├── 企业安全合规 ── SOC 2 进行中
├── 本地/云部署 ── 灵活选择
└── 24/7 企业支持
十、踩坑记录
坑1：Agent 之间不传递上下文
plaintext
问题: 写手 Agent 完全无视研究员的输出，从头写
原因: 没有设置 context 参数
修复: write_task = Task(..., context=[research_task])
教训: CrewAI 的 context 不会自动传递，必须显式声明
坑2：Hierarchical 模式的 Manager 循环
plaintext
问题: Manager 反复在 Agent A 和 B 之间委派
原因: Manager 的 LLM 没有明确的终止条件
修复: 在 Manager 的 instructions 中明确"每个任务最多执行 2 次"
教训: Hierarchical 模式下，给 Manager 的指令要包括终止条件
坑3：工具输出被截断
plaintext
问题: 工具返回大量数据，被截断到 4000 字符
原因: CrewAI 默认的 max_output_length 限制
修复:
1. 在工具端做数据摘要，返回精炼结果
2. 或调整 CrewAI 的输出长度配置
教训: 不要让 Agent 处理原始大数据，先在工具层做预处理
坑4：LLM 配置报 ValidationError
plaintext
问题: pydantic.ValidationError: 1 validation error for Agent: llm: field required
原因: 新版 CrewAI 要求显式配置 llm 参数
修复:
agent = Agent(
role="...",
llm=LLM(model="gpt-4o"), # 必须显式指定
...
)
教训: 升级 CrewAI 版本后，先检查 Agent 构造函数的变化
坑5：记忆系统导致 Token 膨胀
plaintext
问题: 启用 memory=True 后，每次调用的 Token 消耗翻了 3 倍
原因: 长期记忆的向量检索结果被注入了每次请求
修复:
1. 只在需要跨次运行的场景启用长期记忆
2. 为每个 Agent 单独配置 memory
教训: 记忆是有成本的，不是越多越好
坑6：CrewAI 的遥测数据收集
plaintext
问题: Trustpilot 上有用户投诉 CrewAI 未经同意收集使用数据
影响: 企业部署时可能违反数据合规要求
修复:
1. 设置环境变量 OLLAMA_TELEMETRY=0（如果用 Ollama）
2. 审查 CrewAI 的网络请求
3. share_crew=False（默认已关闭详细数据共享）
教训: 企业使用前务必做安全审计，特别是开源框架的数据收集行为
总结
CrewAI 的核心价值是**直觉映射 **：你不需要学会图论，只需要会"分工"。角色 + 目标 + 背景故事，三个要素定义 Agent；Sequential 或 Hierarchical，两种模式编排流程。
它不是最强大的框架——LangGraph 在控制力上更胜一筹，Google ADK 在 A2A 集成上领先，OpenAI Agents SDK 更轻量。但它是上手最快的框架，20 行代码就能跑通一个多 Agent 系统。
我的建议 ：
入门 ：从最小 Crew（研究员 + 写手）开始，理解核心概念
进阶 ：尝试 5 角色产品团队，体验 Hierarchical 模式
生产 ：评估是否需要迁移到 LangGraph（如果需要更精细的控制）
混合：CrewAI 做快速原型 + LangGraph 做核心工作流，两者可以共存
本文由 PySuper 撰写，首发于 zhengxingtao.com
参考来源：
据《CrewAI GitHub Repository》(GitHub, 2026-04)
据《CrewAI Review 2026: Multi-Agent Orchestration Made Simple》(VisionStack, 2026-04-11)
据《Fix: CrewAI Not Working — Agent Delegation, Task Context, and LLM Configuration Errors》(FixDevs, 2026-04-09)
据《Build Your First Multi-Agent AI System with CrewAI + Python》(Effloow, 2026-04-04)
评论区