系列：高级Agent框架（2/6）
难度：⭐⭐⭐ 中级
预计阅读时间：15分钟

---

问题场景

使用AutoGen时，你是否遇到这些问题？

• Agent角色定义不够清晰
• 任务分配需要手动管理
• 缺乏标准化的工作流程
• 团队协作模式不够灵活

痛点：如何像管理真实团队一样管理Agent？

解决方案：使用CrewAI框架，让Agent像团队成员一样协作。

---

CrewAI简介

什么是CrewAI？

CrewAI是一个面向任务的Multi-Agent协作框架，核心理念是将Agent组织成"团队"（Crew），每个Agent有明确的角色和职责。

核心特点：

• 👥 团队化组织：Agent、Task、Crew三层结构
• 🎯 任务驱动：明确的任务定义和交付物
• 🔄 流程控制：顺序执行、分层处理两种模式
• 💾 记忆机制：短期记忆和长期记忆
• 🚀 简单易用：API设计直观，学习曲线平缓

GitHub：https://github.com/joaomdmoura/crewAI
Star数：15,000+（快速增长）
最新版本：0.28.x（2026年5月）

CrewAI vs AutoGen

维度	CrewAI	AutoGen
设计理念	团队化管理	对话式协作
学习曲线	⭐⭐ 简单	⭐⭐⭐ 中等
任务管理	✅ 内置Task抽象	❌ 需自定义
流程控制	✅ Sequential/Hierarchical	⚠️ 需手动编排
代码执行	⚠️ 基础支持	✅ 强大
灵活性	⭐⭐⭐ 中等	⭐⭐⭐⭐⭐ 高
适用场景	结构化任务	开放式探索

选择建议：

• ✅ 选CrewAI：任务明确、流程固定、需要标准化
• ✅ 选AutoGen：开放对话、代码生成、高度定制

---

核心概念

1. Agent（智能体）

Agent是CrewAI的基本执行单元，代表团队中的一个成员。

from crewai import Agent

# 创建Agent
researcher = Agent(
    role='高级研究员',
    goal='深入研究指定主题，提供全面准确的分析报告',
    backstory="""你是一位经验丰富的研究员，擅长：
    - 信息搜集和整理
    - 数据分析和洞察提取
    - 撰写专业的研究报告
    
    你的工作风格严谨、细致，注重事实依据。""",
    verbose=True,  # 输出详细日志
    allow_delegation=False,  # 是否允许委派任务
    max_iter=3,  # 最大迭代次数
    max_rpm=10  # 每分钟最大请求数（限流）
)

关键参数：

• role：角色名称（如"研究员"、"作家"）
• goal：目标描述（Agent的使命）
• backstory：背景故事（塑造Agent个性）
• verbose：是否输出详细日志
• allow_delegation：是否可以将任务委派给其他Agent
• tools：可用的工具列表

2. Task（任务）

Task定义了Agent需要完成的具体工作。

from crewai import Task

# 创建任务
research_task = Task(
    description="""
    研究人工智能在医疗领域的应用现状：
    1. 收集最新的研究论文和行业报告
    2. 分析主要应用场景和技术趋势
    3. 识别关键挑战和未来发展方向
    
    确保信息来源可靠，数据准确。
    """,
    expected_output="""一份详细的研究简报，包含：
    - 主要应用场景（至少5个）
    - 技术发展趋势
    - 面临的挑战
    - 未来展望
    
    格式：Markdown，1500-2000字""",
    agent=researcher,  # 指定执行Agent
    tools=[],  # 可选工具
    async_execution=False,  # 是否异步执行
    context=[]  # 依赖的前置任务
)

关键参数：

• description：任务描述（要做什么）
• expected_output：期望输出（交付物标准）
• agent：执行任务的Agent
• tools：任务可用的工具
• async_execution：是否异步执行
• context：依赖的其他任务输出

3. Crew（团队）

Crew将多个Agent和Task组织成一个协作团队。

from crewai import Crew, Process

# 创建团队
crew = Crew(
    agents=[researcher, writer],  # 团队成员
    tasks=[research_task, writing_task],  # 任务列表
    process=Process.sequential,  # 执行流程：顺序执行
    verbose=True,  # 详细日志
    memory=True,  # 启用记忆
    cache=True,  # 启用缓存
    max_rpm=100,  # 团队级别的限流
    language='zh'  # 输出语言
)

# 执行团队任务
result = crew.kickoff()
print(result)

关键参数：

• agents：团队成员列表
• tasks：任务列表
• process：执行流程
  • Process.sequential：顺序执行
  • Process.hierarchical：分层执行（需要manager agent）
• memory：是否启用记忆
• cache：是否启用缓存
• language：输出语言（'zh'中文，'en'英文）

---

快速入门示例

示例1：简单的内容创作团队

场景：创建一个博客文章创作团队

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

# 配置LLM
llm = ChatOpenAI(
    model="qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="your-api-key",
    temperature=0.7
)

# 1. 创建Agent
researcher = Agent(
    role='内容研究员',
    goal='深入研究Python编程最佳实践',
    backstory='你是一位资深Python开发者，拥有10年经验，擅长总结最佳实践。',
    llm=llm,
    verbose=True
)

writer = Agent(
    role='技术作家',
    goal='将研究成果转化为通俗易懂的博客文章',
    backstory='你是一位优秀的技术作家，擅长用简洁的语言解释复杂概念。',
    llm=llm,
    verbose=True
)

editor = Agent(
    role='编辑',
    goal='审查和优化文章，确保质量和可读性',
    backstory='你是一位经验丰富的编辑，注重文章结构和语言表达。',
    llm=llm,
    verbose=True
)

# 2. 创建Task
research_task = Task(
    description='研究Python性能优化的5个关键技巧',
    expected_output='一份包含5个技巧的研究笔记，每个技巧包含原理、示例代码、效果对比',
    agent=researcher
)

writing_task = Task(
    description='基于研究笔记撰写博客文章',
    expected_output='一篇1500字的技术博客，包含引言、正文、结论，格式Markdown',
    agent=writer,
    context=[research_task]  # 依赖研究任务
)

editing_task = Task(
    description='审查和优化博客文章',
    expected_output='优化后的最终版本，改进语言表达和结构',
    agent=editor,
    context=[writing_task]  # 依赖写作任务
)

# 3. 创建Crew
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,
    verbose=True
)

# 4. 执行
result = crew.kickoff()
print("\n=== 最终文章 ===\n")
print(result)

执行流程：

1. Researcher → 研究Python性能优化技巧
   ↓ 输出研究笔记
2. Writer → 基于笔记撰写博客
   ↓ 输出初稿
3. Editor → 审查和优化
   ↓ 输出最终版本

示例2：市场调研团队

场景：自动生成市场调研报告

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool

# 配置工具（需要API Key）
search_tool = SerperDevTool()
web_search_tool = WebsiteSearchTool()

# 1. 创建Agent
market_researcher = Agent(
    role='市场研究员',
    goal='收集和分析市场数据',
    backstory='你擅长使用各种工具收集市场信息，数据分析能力强。',
    tools=[search_tool, web_search_tool],
    verbose=True
)

competitor_analyst = Agent(
    role='竞争分析师',
    goal='分析竞争对手的策略和优势劣势',
    backstory='你善于从公开信息中提取竞争对手的关键策略。',
    tools=[search_tool],
    verbose=True
)

strategy_consultant = Agent(
    role='战略顾问',
    goal='基于市场和竞争分析制定战略建议',
    backstory='你是一位资深战略顾问，擅长制定可执行的商业策略。',
    verbose=True
)

# 2. 创建Task
market_analysis = Task(
    description="""
    调研2026年AI助手市场：
    1. 市场规模和增长趋势
    2. 主要玩家和市场份额
    3. 用户需求痛点
    """,
    expected_output='市场调研报告，包含数据和图表说明',
    agent=market_researcher
)

competitor_analysis = Task(
    description="""
    分析主要竞争对手：
    1. Cursor、GitHub Copilot、Codeium的产品特点
    2. 定价策略
    3. 优势和劣势
    """,
    expected_output='竞争分析报告，SWOT分析',
    agent=competitor_analyst,
    context=[market_analysis]
)

strategy_recommendation = Task(
    description="""
    制定市场进入策略：
    1. 目标客户群体
    2. 差异化定位
    3. 营销策略
    4. 风险评估
    """,
    expected_output='战略建议书，包含具体行动计划',
    agent=strategy_consultant,
    context=[market_analysis, competitor_analysis]
)

# 3. 创建Crew
crew = Crew(
    agents=[market_researcher, competitor_analyst, strategy_consultant],
    tasks=[market_analysis, competitor_analysis, strategy_recommendation],
    process=Process.sequential,
    verbose=True,
    memory=True  # 启用记忆，Agent可以记住之前的分析
)

# 4. 执行
result = crew.kickoff()
print(result)

示例3：异步执行提升效率

场景：并行执行独立任务

from crewai import Agent, Task, Crew, Process

# 创建Agent
data_analyst_1 = Agent(role='数据分析师A', goal='分析销售数据', llm=llm)
data_analyst_2 = Agent(role='数据分析师B', goal='分析用户数据', llm=llm)
data_analyst_3 = Agent(role='数据分析师C', goal='分析产品数据', llm=llm)
report_writer = Agent(role='报告撰写者', goal='整合分析结果', llm=llm)

# 创建并行任务
sales_analysis = Task(
    description='分析2024年Q1销售数据',
    expected_output='销售分析报告',
    agent=data_analyst_1,
    async_execution=True  # 异步执行
)

user_analysis = Task(
    description='分析用户行为和留存率',
    expected_output='用户分析报告',
    agent=data_analyst_2,
    async_execution=True  # 异步执行
)

product_analysis = Task(
    description='分析产品性能和用户反馈',
    expected_output='产品分析报告',
    agent=data_analyst_3,
    async_execution=True  # 异步执行
)

# 汇总任务（等待所有分析完成）
final_report = Task(
    description='整合三份分析报告，撰写综合报告',
    expected_output='综合分析报告',
    agent=report_writer,
    context=[sales_analysis, user_analysis, product_analysis]  # 依赖所有分析
)

# 创建Crew
crew = Crew(
    agents=[data_analyst_1, data_analyst_2, data_analyst_3, report_writer],
    tasks=[sales_analysis, user_analysis, product_analysis, final_report],
    process=Process.sequential,
    verbose=True
)

# 执行（前3个任务并行，最后1个串行）
result = crew.kickoff()

执行时序：

时间轴：
T0: [Sales Analysis] [User Analysis] [Product Analysis] ← 并行执行
T1:      ↓                ↓                ↓
T2:                [Final Report]          ← 等待完成后执行

性能提升：

• 串行执行：3个分析 × 2分钟 + 报告 2分钟 = 8分钟
• 并行执行：max(2分钟) + 报告 2分钟 = 4分钟
• 提升50%

---

高级特性

1. 分层处理（Hierarchical Process）

使用Manager Agent自动协调团队。

from crewai import Agent, Task, Crew, Process

# 创建专业Agent
coder = Agent(
    role='Python开发工程师',
    goal='编写高质量的Python代码',
    backstory='你是一位资深Python开发者，注重代码质量和最佳实践。',
    llm=llm
)

tester = Agent(
    role='测试工程师',
    goal='编写全面的测试用例',
    backstory='你擅长发现代码中的bug，编写高质量的单元测试。',
    llm=llm
)

reviewer = Agent(
    role='代码审查专家',
    goal='审查代码质量和安全性',
    backstory='你是一位安全专家，擅长发现潜在的安全漏洞。',
    llm=llm
)

# 创建Manager Agent（自动协调）
manager = Agent(
    role='项目经理',
    goal='协调团队完成软件开发任务',
    backstory='你是一位经验丰富的项目经理，擅长任务分解和资源调配。',
    llm=llm,
    allow_delegation=True  # 允许委派任务
)

# 创建任务
development_task = Task(
    description='开发一个REST API服务，包含用户注册、登录、查询功能',
    expected_output='完整的Flask应用代码，包含路由、模型、验证',
    agent=coder
)

testing_task = Task(
    description='为API服务编写单元测试',
    expected_output='pytest测试用例，覆盖率>80%',
    agent=tester
)

review_task = Task(
    description='审查代码安全性和规范性',
    expected_output='代码审查报告，包含改进建议',
    agent=reviewer
)

# 创建Crew（分层模式）
crew = Crew(
    agents=[coder, tester, reviewer, manager],
    tasks=[development_task, testing_task, review_task],
    process=Process.hierarchical,  # 分层模式
    manager_agent=manager,  # 指定Manager
    verbose=True
)

# 执行
result = crew.kickoff()

分层模式工作流程：

Manager接收任务
    ↓
分解任务 → 分配给Coder
    ↓
Coder完成 → 报告Manager
    ↓
Manager分配给Tester
    ↓
Tester完成 → 报告Manager
    ↓
Manager分配给Reviewer
    ↓
最终汇总

vs 顺序模式：

• Sequential：任务按预定义顺序执行
• Hierarchical：Manager动态决定下一步

2. 记忆机制

Agent可以记住历史交互。

from crewai import Agent, Task, Crew, Process
from crewai.memory import ShortTermMemory, LongTermMemory

# 创建带记忆的Agent
researcher = Agent(
    role='研究员',
    goal='持续跟踪AI技术发展',
    backstory='你是一位长期关注AI领域的研究员。',
    llm=llm,
    memory=True  # 启用记忆
)

# 第1次调研
task1 = Task(
    description='调研2024年AI大模型发展',
    expected_output='调研报告',
    agent=researcher
)

crew1 = Crew(
    agents=[researcher],
    tasks=[task1],
    memory=True,  # 启用团队记忆
    verbose=True
)

result1 = crew1.kickoff()

# 第2次调研（Agent记得之前的内容）
task2 = Task(
    description='基于之前的调研，分析2025年的新趋势',
    expected_output='趋势分析报告',
    agent=researcher
)

crew2 = Crew(
    agents=[researcher],
    tasks=[task2],
    memory=True,
    verbose=True
)

result2 = crew2.kickoff()
# Agent会引用第1次调研的结果

记忆类型：

• Short-term Memory：当前会话的记忆
• Long-term Memory：跨会话的持久化记忆
• Entity Memory：实体信息记忆（人名、公司名等）

3. 缓存机制

避免重复的LLM调用。

from crewai import Crew

# 启用缓存
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    cache=True  # 启用缓存
)

# 相同输入会返回缓存结果
# 显著提升速度，降低成本

缓存策略：

• 基于任务描述和输入生成缓存键
• 默认TTL：24小时
• 可自定义缓存后端（Redis、文件系统）

4. 回调机制

监控执行过程。

from crewai import Crew

# 定义回调函数
def on_step_start(step):
    print(f"开始执行: {step.task.description}")

def on_step_end(step, output):
    print(f"完成执行: {output[:100]}...")

def on_crew_start(crew):
    print(f"团队启动: {len(crew.agents)}个Agent, {len(crew.tasks)}个任务")

def on_crew_end(crew, result):
    print(f"团队完成，结果长度: {len(str(result))}")

# 创建Crew并注册回调
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    step_callback=on_step_start,  # 每步开始
    task_callback=on_step_end,  # 每步结束
)

# 执行
crew.kickoff()

---

实战案例：市场调研报告自动生成

项目背景

构建一个自动化市场调研系统，能够：

1. 收集行业数据
2. 分析竞争格局
3. 识别市场机会
4. 生成完整报告

系统架构

┌─────────────┐
│   Manager   │ ← 协调整个流程
└──────┬──────┘
       │
       ├──────────────┬──────────────┐
       ▼              ▼              ▼
┌──────────┐  ┌──────────┐  ┌──────────┐
│ Researcher│  │ Analyst  │  │ Strategist│
│ 数据收集  │  │ 竞争分析  │  │ 战略规划  │
└────┬─────┘  └────┬─────┘  └────┬─────┘
     │              │              │
     └──────────────┼──────────────┘
                    ▼
            ┌──────────────┐
            │   Writer     │ ← 撰写报告
            └──────┬───────┘
                   ▼
            ┌──────────────┐
            │  Final Report│
            └──────────────┘

完整代码

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
from langchain_openai import ChatOpenAI
import os

# 配置LLM
llm = ChatOpenAI(
    model="qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    temperature=0.7
)

# 配置工具
search_tool = SerperDevTool(api_key=os.getenv("SERPER_API_KEY"))
scrape_tool = ScrapeWebsiteTool()

# ==================== 创建Agent ====================

# 1. 市场研究员
market_researcher = Agent(
    role='高级市场研究员',
    goal='深入调研目标市场，收集全面准确的数据',
    backstory="""你是一位拥有15年经验的市场研究专家，擅长：
    - 使用多种渠道收集市场数据
    - 识别关键市场趋势
    - 量化市场规模和增长率
    
    你的研究报告以数据详实、洞察深刻著称。""",
    tools=[search_tool, scrape_tool],
    llm=llm,
    verbose=True,
    max_iter=5
)

# 2. 竞争分析师
competitor_analyst = Agent(
    role='竞争情报分析师',
    goal='全面分析竞争对手的战略、优势和劣势',
    backstory="""你是一位竞争情报专家，擅长：
    - 竞品功能对比
    - 定价策略分析
    - SWOT分析
    - 市场份额估算
    
    你能从公开信息中提取有价值的竞争情报。""",
    tools=[search_tool],
    llm=llm,
    verbose=True
)

# 3. 战略顾问
strategy_consultant = Agent(
    role='首席战略顾问',
    goal='基于市场和竞争分析制定可执行的战略建议',
    backstory="""你是一位顶级战略顾问，曾服务于多家 Fortune 500 企业。
    你擅长：
    - 市场进入策略
    - 差异化定位
    - 商业模式设计
    - 风险评估和管理
    
    你的建议总是 actionable 且可量化。""",
    llm=llm,
    verbose=True
)

# 4. 报告撰写者
report_writer = Agent(
    role='资深商业作家',
    goal='将复杂的分析转化为清晰专业的商业报告',
    backstory="""你是一位获奖的商业作家，擅长：
    - 结构化报告撰写
    - 数据可视化说明
    - Executive Summary提炼
    - 专业商务语言
    
    你的报告被多家咨询公司采用为标准模板。""",
    llm=llm,
    verbose=True
)

# ==================== 创建Task ====================

# 任务1：市场调研
market_research_task = Task(
    description="""
    调研2026年AI编程助手市场：
    
    1. 市场规模：全球和中国市场的规模、增长率
    2. 用户画像：主要用户群体、使用场景、付费意愿
    3. 技术趋势：关键技术发展方向
    4. 政策环境：相关政策和监管要求
    
    数据来源：行业报告、新闻、研究机构发布
    """,
    expected_output="""
    一份详细的市场调研报告，包含：
    - 市场规模数据（2024-2026）
    - 用户画像分析
    - 技术趋势总结
    - 政策环境说明
    
    格式：Markdown，2000-2500字，包含数据表格
    """,
    agent=market_researcher
)

# 任务2：竞争分析
competitor_analysis_task = Task(
    description="""
    分析主要竞争对手：
    
    1. GitHub Copilot：功能、定价、优劣势
    2. Cursor：产品特点、市场定位
    3. Codeium：免费策略、用户反馈
    4. 通义灵码：本土化优势
    
    进行SWOT分析和功能对比
    """,
    expected_output="""
    竞争分析报告，包含：
    - 竞品功能对比表
    - 定价策略对比
    - SWOT分析（每个竞品）
    - 市场份额估算
    
    格式：Markdown，1500-2000字
    """,
    agent=competitor_analyst,
    context=[market_research_task]
)

# 任务3：战略建议
strategy_task = Task(
    description="""
    制定市场进入战略：
    
    1. 目标客户：优先切入的客户群体
    2. 差异化定位：与竞品的差异点
    3. 定价策略：价格区间和套餐设计
    4. 营销渠道：获客策略
    5. 风险与应对：主要风险和缓解措施
    """,
    expected_output="""
    战略建议书，包含：
    - 目标客户定义
    - 价值主张
    - 定价方案
    - 营销计划（3个月）
    - 风险矩阵
    
    格式：Markdown，2000字
    """,
    agent=strategy_consultant,
    context=[market_research_task, competitor_analysis_task]
)

# 任务4：报告整合
report_writing_task = Task(
    description="""
    整合以上所有分析，撰写完整的市场调研报告：
    
    结构：
    1. Executive Summary（执行摘要）
    2. 市场概况
    3. 竞争格局
    4. 战略建议
    5. 附录（数据表格）
    
    要求：
    - 逻辑清晰，层次分明
    - 数据支撑观点
    - 可执行的建议
    - 专业的商务语言
    """,
    expected_output="""
    完整的市场调研报告，3000-4000字，
    包含所有章节，格式专业，可直接用于商业决策
    """,
    agent=report_writer,
    context=[market_research_task, competitor_analysis_task, strategy_task]
)

# ==================== 创建Crew ====================

crew = Crew(
    agents=[
        market_researcher,
        competitor_analyst,
        strategy_consultant,
        report_writer
    ],
    tasks=[
        market_research_task,
        competitor_analysis_task,
        strategy_task,
        report_writing_task
    ],
    process=Process.sequential,
    verbose=True,
    memory=True,  # 启用记忆
    cache=True,   # 启用缓存
    max_rpm=50    # 限流
)

# ==================== 执行 ====================

print("🚀 启动市场调研团队...\n")
result = crew.kickoff()

print("\n" + "="*60)
print("📊 市场调研报告生成完成！")
print("="*60 + "\n")
print(result)

# 保存报告
with open('market_research_report.md', 'w', encoding='utf-8') as f:
    f.write(str(result))

print("\n✅ 报告已保存到 market_research_report.md")

运行结果

生成的报告结构：

# 2026年AI编程助手市场调研报告

## Executive Summary
- 市场规模：2026年全球市场预计达到$XX亿
- 增长率：年复合增长率XX%
- 关键发现：...

## 1. 市场概况
### 1.1 市场规模
[数据表格]

### 1.2 用户画像
- 主要用户：软件开发者（70%）
- 使用场景：代码补全、代码审查、Bug修复
...

## 2. 竞争格局
### 2.1 竞品对比
[功能对比表]

### 2.2 SWOT分析
**GitHub Copilot**
- Strengths: ...
- Weaknesses: ...
...

## 3. 战略建议
### 3.1 目标客户
- 优先级1：中小型科技公司
- 优先级2：独立开发者
...

### 3.2 差异化定位
...

## 4. 风险评估
[风险矩阵]

## 附录
- 数据来源
- 研究方法

执行时间：

• 市场调研：~3分钟
• 竞争分析：~2分钟
• 战略建议：~2分钟
• 报告整合：~2分钟
• 总计：~9分钟（串行）

成本估算：

• LLM调用：约20次
• Token消耗：~50,000 tokens
• 成本：约¥5-10（取决于模型）

---

与AutoGen对比

功能对比

功能	CrewAI	AutoGen
Agent定义	Role/Goal/Backstory	System Message
任务管理	✅ Task抽象	❌ 需自定义
流程控制	✅ Sequential/Hierarchical	⚠️ GroupChat
记忆机制	✅ 内置	⚠️ 需手动管理
缓存	✅ 内置	❌ 需自定义
代码执行	⚠️ 基础	✅ 强大
工具集成	✅ crewai_tools	✅ 灵活
学习曲线	⭐⭐ 简单	⭐⭐⭐ 中等
灵活性	⭐⭐⭐ 中等	⭐⭐⭐⭐⭐ 高

选择建议

使用CrewAI的场景：

• ✅ 任务明确、流程固定
• ✅ 需要标准化的团队协作
• ✅ 快速原型开发
• ✅ 业务导向的应用

使用AutoGen的场景：

• ✅ 开放式对话和探索
• ✅ 代码生成和执行
• ✅ 高度自定义的Agent行为
• ✅ 研究和实验性质

混合使用：

# 可以用CrewAI管理团队，内部Agent使用AutoGen
from crewai import Agent
from autogen import ConversableAgent

# CrewAI的Agent底层可以使用AutoGen的LLM配置
autogen_llm_config = {...}
crewai_agent = Agent(
    role='...',
    llm_config=autogen_llm_config  # 兼容
)

---

最佳实践

1. 明确Agent角色

# ❌ 模糊的角色定义
agent = Agent(
    role='助手',
    goal='帮助用户'
)

# ✅ 具体的角色定义
agent = Agent(
    role='Python性能优化专家',
    goal='识别代码性能瓶颈并提供优化建议',
    backstory='你是一位有10年经验的Python性能优化专家...'
)

2. 详细的Expected Output

# ❌ 模糊的输出要求
task = Task(
    expected_output='一份报告'
)

# ✅ 具体的输出要求
task = Task(
    expected_output="""
    一份详细的技术报告，包含：
    - 问题分析（300字）
    - 解决方案（500字）
    - 代码示例（完整可运行）
    - 性能对比数据
    
    格式：Markdown，总字数1000-1500字
    """
)

3. 合理使用Context

# ❌ 任务之间无关联
task1 = Task(..., agent=agent1)
task2 = Task(..., agent=agent2)  # 不知道task1的结果

# ✅ 建立任务依赖
task1 = Task(..., agent=agent1)
task2 = Task(
    ...,
    agent=agent2,
    context=[task1]  # 可以访问task1的输出
)

4. 启用记忆和缓存

# ✅ 对于重复性任务，启用缓存
crew = Crew(
    ...,
    memory=True,  # 记住历史交互
    cache=True    # 缓存LLM响应
)

# 显著降低成本，提升速度

5. 限流保护

# ✅ 设置合理的限流
crew = Crew(
    ...,
    max_rpm=50  # 每分钟最多50次请求
)

# 避免API配额耗尽

---

常见问题

Q1: CrewAI适合什么规模的项目？

回答：

• 小型：3-5个Agent，简单任务链
• 中型：5-10个Agent，复杂工作流
• 大型：10+ Agent，建议使用Hierarchical模式

Q2: 如何处理长任务？

回答：

# 增加迭代次数
agent = Agent(
    ...,
    max_iter=10  # 默认3，可增加
)

# 或者拆分任务
task1 = Task(description='第1部分', ...)
task2 = Task(description='第2部分', context=[task1])

Q3: 如何调试Crew？

回答：

# 启用详细日志
crew = Crew(..., verbose=True)

# 查看每一步的输出
for task in crew.tasks:
    print(f"Task: {task.description}")
    print(f"Output: {task.output}")

---

总结

CrewAI核心价值

🎯 团队化管理

• Agent、Task、Crew三层结构
• 清晰的角色和职责分工

🚀 快速上手

• API设计直观
• 学习曲线平缓

💡 标准化流程

• Sequential/Hierarchical两种模式
• 适合业务场景

下一步学习

1. 实践：用CrewAI重构现有的单Agent应用
2. 深入：尝试Hierarchical模式和记忆机制
3. 对比：下一篇将对比LangGraph

相关资源：

• CrewAI官方文档：https://docs.crewai.com/
• GitHub仓库：https://github.com/joaomdmoura/crewAI

---

💡 觉得有用？点赞+收藏+关注，不错过下一篇文章！

下一篇预告：《LangGraph深度教程：状态机驱动的Agent工作流》