CrewAI 多智能体协作开发完全教程

从零开始掌握 CrewAI 多智能体协作框架，构建专业的 AI Agent 团队

---

目录

1. CrewAI 简介与核心概念
2. 环境搭建与安装
3. Agent 智能体详解
4. Task 任务系统
5. Tool 工具系统
6. Crew 团队协作
7. 流程控制与记忆系统
8. 实战项目一：自动化研究报告生成
9. 实战项目二：智能内容创作团队
10. 实战项目三：代码审查与优化系统
11. 高级技巧与最佳实践
12. 常见问题与排错指南

---

1. CrewAI 简介与核心概念

1.1 什么是 CrewAI

CrewAI 是一个开源的多智能体协作框架，它让多个 AI Agent 像一个高效团队一样协同工作。与单一 AI Agent 不同，CrewAI 通过角色分工、任务委派和协作机制，让多个专业化 Agent 共同完成复杂任务。

想象一下，你有一个由 AI 专家组成的团队：一个负责研究，一个负责写作，一个负责审查——它们各司其职，互相配合，最终产出高质量的工作成果。这就是 CrewAI 的核心理念。

1.2 为什么选择 CrewAI

在 AI 应用开发领域，我们面临几个关键挑战：

单一 Agent 的局限性

当任务变得复杂时，单一 Agent 往往力不从心。比如，让它同时做市场调研、数据分析和报告撰写，效果通常不如让专业化的 Agent 分工协作。

CrewAI 的优势

• 角色专业化：每个 Agent 有明确的角色、目标和专长
• 任务编排：支持顺序、并行和层级等多种任务执行模式
• 工具集成：内置丰富的工具支持，也支持自定义工具
• 记忆系统：Agent 之间可以共享上下文和记忆
• 简洁 API：设计优雅，上手快，代码量少

1.3 CrewAI 与其他框架对比

特性	CrewAI	LangChain Agent	AutoGen	MetaGPT
多智能体协作	原生支持	需手动编排	原生支持	原生支持
角色定义	直观简洁	较复杂	中等	直观
任务编排	灵活	灵活	中等	结构化
学习曲线	低	中	中	中
社区活跃度	高	很高	高	中
适用场景	通用协作	链式调用	对话式	软件开发

1.4 核心概念一览

CrewAI 的核心概念包括四个部分：

Agent（智能体）：具有特定角色和能力的 AI 实体，类似于团队中的成员。每个 Agent 都有自己的目标、背景故事和工具集。

Task（任务）：需要完成的具体工作单元，包含描述、预期输出和负责的 Agent。

Tool（工具）：Agent 可以使用的外部能力，如搜索、读写文件、调用 API 等。

Crew（团队）：由多个 Agent 组成的协作单元，定义了协作策略和任务执行流程。

---

2. 环境搭建与安装

2.1 系统要求

在开始之前，确保你的环境满足以下要求：

• Python 3.10 或更高版本
• pip 包管理器
• 至少一个 LLM API Key（OpenAI、Anthropic 等）

检查 Python 版本：

python3 --version
# 输出应为 Python 3.10.x 或更高

2.2 创建虚拟环境

推荐使用虚拟环境来管理依赖：

# 创建项目目录
mkdir crewai-project
cd crewai-project

# 创建虚拟环境
python3 -m venv venv

# 激活虚拟环境
# macOS/Linux:
source venv/bin/activate
# Windows:
# venv\Scripts\activate

2.3 安装 CrewAI

CrewAI 提供了简便的安装方式：

# 安装 CrewAI 核心包
pip install crewai

# 安装包含所有工具的完整版
pip install 'crewai[tools]'

# 或者使用 crewai 命令行工具创建项目
pip install crewai-tools

验证安装：

python3 -c "import crewai; print(f'CrewAI version: {crewai.__version__}')"

2.4 配置 API Key

CrewAI 支持多种 LLM 提供商。以下是配置方式：

方式一：环境变量（推荐）

# OpenAI
export OPENAI_API_KEY="your-openai-api-key"

# 或者使用其他提供商
export ANTHROPIC_API_KEY="your-anthropic-api-key"

方式二：.env 文件

在项目根目录创建 .env 文件：

OPENAI_API_KEY=your-openai-api-key
# OPENAI_BASE_URL=https://api.openai.com/v1  # 可选，自定义 API 端点

然后在代码中加载：

from dotenv import load_dotenv
load_dotenv()

方式三：直接在代码中指定模型

from crewai import Agent

agent = Agent(
    role="研究员",
    goal="深入研究指定主题",
    backstory="你是一位经验丰富的研究员",
    llm="gpt-4o"  # 或 "anthropic/claude-3-5-sonnet"
)

2.5 使用本地模型

如果你想使用本地部署的模型（如通过 Ollama），可以这样配置：

from crewai import Agent, LLM

# 使用 Ollama 本地模型
ollama_llm = LLM(
    model="ollama/qwen2.5:14b",
    base_url="http://localhost:11434"
)

agent = Agent(
    role="助手",
    goal="帮助用户解决问题",
    backstory="你是一个乐于助人的AI助手",
    llm=ollama_llm
)

2.6 验证环境

创建一个简单的测试脚本来验证环境配置：

# test_env.py
from crewai import Agent, Task, Crew

# 创建一个简单的 Agent
test_agent = Agent(
    role="测试助手",
    goal="验证环境配置是否正确",
    backstory="你负责测试 CrewAI 环境是否正常工作",
    verbose=True
)

# 创建一个简单的任务
test_task = Task(
    description="请用一句话介绍你自己",
    expected_output="一句话的自我介绍",
    agent=test_agent
)

# 创建 Crew 并执行
crew = Crew(
    agents=[test_agent],
    tasks=[test_task],
    verbose=True
)

result = crew.kickoff()
print(f"测试结果: {result}")

运行测试：

python3 test_env.py

如果看到正常的输出，说明环境配置成功。

---

3. Agent 智能体详解

3.1 Agent 的基本属性

Agent 是 CrewAI 的核心构建块。每个 Agent 都需要定义以下关键属性：

from crewai import Agent

researcher = Agent(
    role="高级研究员",           # 角色名称
    goal="发现并分析最新的行业趋势",  # 目标
    backstory="""                  # 背景故事
        你是一位拥有10年经验的高级研究员，
        专注于科技行业趋势分析。你善于从海量信息中
        提取关键洞察，并以清晰的方式呈现给决策者。
    """,
    verbose=True,                  # 是否输出详细日志
    allow_delegation=True,         # 是否允许委派任务
    max_iter=25,                   # 最大迭代次数
    max_rpm=10,                    # 每分钟最大请求数
    memory=True,                   # 是否启用记忆
)

3.2 角色设计原则

设计 Agent 角色时，遵循以下原则可以让协作效果更好：

明确性原则

角色描述应该具体而明确，避免模糊的定义。

# 不好的例子
bad_agent = Agent(
    role="助手",
    goal="帮忙做事",
    backstory="你是一个助手"
)

# 好的例子
good_agent = Agent(
    role="资深数据分析师",
    goal="通过数据挖掘发现业务增长机会",
    backstory="""
        你是一位在电商领域有8年经验的数据分析师。
        擅长使用 Python 进行数据清洗、统计分析和可视化。
        你特别关注用户行为数据，善于发现转化率优化的机会。
        你的分析风格是数据驱动、逻辑清晰，善于用图表说话。
    """
)

专业性原则

每个 Agent 应该有明确的专业领域和技能边界。

# 代码审查专家
code_reviewer = Agent(
    role="代码审查专家",
    goal="确保代码质量、安全性和最佳实践",
    backstory="""
        你是一位资深的软件工程师，拥有15年的代码审查经验。
        你精通 Python、JavaScript、Go 等主流语言，
        熟悉 OWASP 安全标准和常见漏洞模式。
        你注重代码的可读性、可维护性和性能优化。
        你的审查严格但建设性，总是提供具体的改进建议。
    """
)

3.3 委派机制

CrewAI 的委派机制允许 Agent 将子任务分配给其他 Agent：

# 允许委派的 Agent
manager = Agent(
    role="项目经理",
    goal="协调团队完成项目目标",
    backstory="你是一位经验丰富的项目经理，善于任务分配和进度管理",
    allow_delegation=True  # 启用委派
)

# 不允许委派的 Agent（专注于自己的任务）
writer = Agent(
    role="技术文档撰写者",
    goal="撰写清晰、准确的技术文档",
    backstory="你是一位专业的技术写作专家",
    allow_delegation=False  # 禁用委派
)

当 allow_delegation=True 时，Agent 可以：

• 将不适合自己处理的子任务委派给其他 Agent
• 请求其他 Agent 提供专业意见
• 在执行过程中动态调整任务分配

3.4 Agent 配置文件管理

对于复杂项目，建议将 Agent 配置集中管理：

# agents.yaml
# 可以使用 YAML 文件来管理 Agent 配置

# agents_config.py
AGENTS_CONFIG = {
    "researcher": {
        "role": "高级研究员",
        "goal": "深入研究并分析指定主题",
        "backstory": """
            你是一位资深研究员，拥有跨学科的研究能力。
            你善于查阅学术论文、行业报告和技术文档，
            能够从多维度分析问题并提出独到见解。
        """
    },
    "writer": {
        "role": "内容创作者",
        "goal": "将研究成果转化为高质量的文章",
        "backstory": """
            你是一位经验丰富的技术作家，
            擅长将复杂的技术概念用通俗易懂的语言表达。
            你的文章结构清晰、逻辑严密、可读性强。
        """
    },
    "reviewer": {
        "role": "质量审查员",
        "goal": "确保输出内容的准确性和质量",
        "backstory": """
            你是一位严谨的质量控制专家，
            对细节有极高的要求。你擅长发现逻辑漏洞、
            事实错误和表达不清的地方，并提供改进建议。
        """
    }
}

使用配置创建 Agent：

from crewai import Agent

def create_agent(agent_type, llm=None, **kwargs):
    config = AGENTS_CONFIG[agent_type].copy()
    config.update(kwargs)
    if llm:
        config['llm'] = llm
    return Agent(**config)

# 使用
researcher = create_agent("researcher", verbose=True)
writer = create_agent("writer")
reviewer = create_agent("reviewer")

3.5 多模型支持

CrewAI 允许为不同的 Agent 配置不同的模型：

from crewai import Agent, LLM

# 强大的模型用于复杂推理
gpt4_llm = LLM(model="gpt-4o")

# 快速模型用于简单任务
gpt4_mini_llm = LLM(model="gpt-4o-mini")

# 本地模型用于隐私敏感任务
local_llm = LLM(
    model="ollama/qwen2.5:14b",
    base_url="http://localhost:11434"
)

# 研究员使用强模型
researcher = Agent(
    role="研究员",
    goal="深入分析问题",
    backstory="你是资深研究员",
    llm=gpt4_llm
)

# 写手使用中等模型
writer = Agent(
    role="写手",
    goal="撰写内容",
    backstory="你是专业写手",
    llm=gpt4_mini_llm
)

# 敏感数据处理使用本地模型
privacy_agent = Agent(
    role="隐私数据处理专家",
    goal="安全处理敏感数据",
    backstory="你专注于数据安全",
    llm=local_llm
)

---

4. Task 任务系统

4.1 Task 的基本结构

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

from crewai import Task

research_task = Task(
    description="""
        对"2024年AI行业发展趋势"进行全面调研，包括：
        1. 大语言模型的最新进展
        2. AI Agent 的发展方向
        3. 多模态AI的突破
        4. AI在各行业的应用案例
        5. 主要玩家的竞争格局
    """,
    expected_output="""
        一份结构化的研究报告，包含：
        - 各领域的关键发现
        - 数据支撑和案例分析
        - 未来趋势预测
        - 行动建议
    """,
    agent=researcher,
    output_file="research_report.md",  # 可选：输出到文件
)

4.2 任务描述技巧

好的任务描述应该包含以下要素：

具体性：明确说明要做什么，不要含糊。

# 不好的描述
bad_task = Task(
    description="分析数据",
    expected_output="分析报告",
    agent=analyst
)

# 好的描述
good_task = Task(
    description="""
        分析过去12个月的销售数据（sales_data.csv），具体包括：
        1. 月度销售额趋势分析
        2. 各产品类别占比变化
        3. 客户地域分布特征
        4. 复购率和客户留存分析
        5. 识别Top 10 畅销产品和滞销产品
    """,
    expected_output="""
        一份包含以下内容的数据分析报告：
        - 趋势图表（至少5张）
        - 关键数据指标汇总表
        - 洞察发现（至少3条可执行的建议）
        - 数据质量说明
    """,
    agent=analyst
)

约束条件：明确输出的格式、长度、风格等要求。

blog_task = Task(
    description="""
        撰写一篇关于"AI编程助手"的技术博客文章。
        
        约束条件：
        - 字数：2000-3000字
        - 风格：技术科普，面向有一定编程基础的读者
        - 结构：引言、技术原理、主流工具对比、使用技巧、展望
        - 语言：中文，专业术语需附英文原文
        - 需要包含实际代码示例
    """,
    expected_output="一篇完整的Markdown格式博客文章",
    agent=writer
)

4.3 任务间的依赖关系

在 CrewAI 中，任务可以有上下文依赖关系：

# 任务1：数据收集
collect_task = Task(
    description="收集指定主题的相关资料和数据",
    expected_output="结构化的原始资料集合",
    agent=researcher
)

# 任务2：基于任务1的输出进行分析
analyze_task = Task(
    description="""
        基于收集到的资料，进行深度分析。
        关注以下方面：
        - 关键趋势识别
        - 数据模式发现
        - 因果关系分析
    """,
    expected_output="分析报告",
    agent=analyst,
    context=[collect_task]  # 依赖任务1的输出
)

# 任务3：基于任务1和2的输出撰写报告
write_task = Task(
    description="基于研究资料和分析结果，撰写最终报告",
    expected_output="完整的最终报告",
    agent=writer,
    context=[collect_task, analyze_task]  # 依赖任务1和2的输出
)

4.4 异步任务

对于耗时较长的任务，可以使用异步执行：

async_task = Task(
    description="执行长时间运行的数据处理任务",
    expected_output="处理结果",
    agent=processor,
    async_execution=True  # 异步执行
)

# 其他任务可以并行执行
parallel_task = Task(
    description="同时执行另一个独立任务",
    expected_output="任务结果",
    agent=worker
)

4.5 输出验证

CrewAI 支持对任务输出进行验证：

from crewai import Task
from pydantic import BaseModel

# 定义输出结构
class ResearchOutput(BaseModel):
    title: str
    summary: str
    key_findings: list[str]
    sources: list[str]

validated_task = Task(
    description="执行研究任务",
    expected_output="符合结构要求的研究结果",
    agent=researcher,
    output_pydantic=ResearchOutput  # 输出必须符合此模型
)

---

5. Tool 工具系统

5.1 内置工具

CrewAI 提供了丰富的内置工具：

from crewai_tools import (
    SerperDevTool,          # 网络搜索
    WebsiteSearchTool,      # 网站内容搜索
    FileReadTool,           # 文件读取
    FileWriteTool,          # 文件写入
    DirectoryReadTool,      # 目录读取
    DirectorySearchTool,    # 目录搜索
    CodeInterpreterTool,    # 代码执行
    CSVSearchTool,          # CSV 搜索
    JSONSearchTool,         # JSON 搜索
    MDXSearchTool,          # MDX 搜索
    PDFSearchTool,          # PDF 搜索
    PGSearchTool,           # PostgreSQL 搜索
    ScrapeWebsiteTool,      # 网页抓取
    SeleniumScrapingTool,   # Selenium 抓取
    TXTSearchTool,          # 文本搜索
    XMLSearchTool,          # XML 搜索
    YoutubeChannelSearchTool,  # YouTube 频道搜索
    YoutubeVideoSearchTool,    # YouTube 视频搜索
)

# 创建工具实例
search_tool = SerperDevTool()
file_read_tool = FileReadTool()
website_tool = WebsiteSearchTool()

5.2 为 Agent 分配工具

from crewai import Agent
from crewai_tools import SerperDevTool, FileReadTool, FileWriteTool

# 创建工具
search = SerperDevTool()
file_reader = FileReadTool()
file_writer = FileWriteTool()

# 研究员配备搜索和文件读取工具
researcher = Agent(
    role="研究员",
    goal="深入研究指定主题",
    backstory="你是一位资深研究员",
    tools=[search, file_reader],
    verbose=True
)

# 写手配备文件读写工具
writer = Agent(
    role="写手",
    goal="撰写高质量内容",
    backstory="你是一位专业写手",
    tools=[file_reader, file_writer],
    verbose=True
)

5.3 自定义工具

创建自定义工具是 CrewAI 的强大功能之一：

方式一：使用 @tool 装饰器

from crewai import tool

@tool
def calculate_bmi(weight_kg: float, height_m: float) -> str:
    """计算BMI指数。输入体重（千克）和身高（米），返回BMI值和健康分类。"""
    bmi = weight_kg / (height_m ** 2)
    
    if bmi < 18.5:
        category = "偏瘦"
    elif bmi < 24:
        category = "正常"
    elif bmi < 28:
        category = "偏胖"
    else:
        category = "肥胖"
    
    return f"BMI: {bmi:.1f}，分类：{category}"

# 使用
agent = Agent(
    role="健康顾问",
    goal="提供健康建议",
    backstory="你是健康领域专家",
    tools=[calculate_bmi]
)

方式二：继承 BaseTool 类

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type

class StockPriceInput(BaseModel):
    """股票价格查询的输入参数"""
    symbol: str = Field(description="股票代码，如 AAPL、GOOGL")

class StockPriceTool(BaseTool):
    name: str = "股票价格查询"
    description: str = "查询指定股票的当前价格。输入股票代码，返回最新价格信息。"
    args_schema: Type[BaseModel] = StockPriceInput

    def _run(self, symbol: str) -> str:
        # 实际实现中，这里会调用股票 API
        # 这里用模拟数据演示
        mock_data = {
            "AAPL": {"price": 178.50, "change": "+2.3%"},
            "GOOGL": {"price": 142.80, "change": "+1.5%"},
            "MSFT": {"price": 378.90, "change": "-0.8%"},
        }
        
        if symbol.upper() in mock_data:
            data = mock_data[symbol.upper()]
            return f"{symbol}: ${data['price']} ({data['change']})"
        return f"未找到股票 {symbol} 的数据"

# 使用
stock_tool = StockPriceTool()
agent = Agent(
    role="金融分析师",
    goal="分析股票市场",
    backstory="你是金融分析专家",
    tools=[stock_tool]
)

5.4 工具组合使用

from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool, FileWriteTool

# 创建多种工具
search = SerperDevTool()
file_writer = FileWriteTool()

# 创建一个配备多种工具的 Agent
analyst = Agent(
    role="市场分析师",
    goal="分析市场竞争格局",
    backstory="你是一位资深市场分析师",
    tools=[search, file_writer]
)

# 创建任务
analysis_task = Task(
    description="""
        研究"AI编程助手"市场的竞争格局，包括：
        1. 主要玩家（GitHub Copilot, Cursor, Windsurf等）
        2. 各产品的核心特点和定价
        3. 市场份额估算
        4. 优劣势对比
        
        将分析结果保存到 market_analysis.md
    """,
    expected_output="一份详细的市场分析报告",
    agent=analyst,
    output_file="market_analysis.md"
)

---

6. Crew 团队协作

6.1 Crew 的基本配置

Crew 是将 Agent 和 Task 组织在一起的核心容器：

from crewai import Crew, Process

# 基本 Crew 配置
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    process=Process.sequential,  # 顺序执行
    verbose=True,
    memory=True,                 # 启用记忆系统
    max_rpm=10,                  # 每分钟最大请求数
    share_crew=False,            # 是否共享 Crew 状态
)

6.2 执行模式

CrewAI 支持多种任务执行模式：

顺序执行（Sequential）

任务按照定义的顺序依次执行，前一个任务的输出作为后一个任务的输入。

sequential_crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    process=Process.sequential,
    verbose=True
)

# 执行顺序：research_task → write_task → review_task
result = sequential_crew.kickoff()

层级执行（Hierarchical）

由一个管理 Agent 协调其他 Agent 的工作：

from crewai import Crew, Process

manager_crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    process=Process.hierarchical,
    manager_llm="gpt-4o",  # 管理者使用的模型
    verbose=True
)

# 管理者会自动决定任务分配和执行顺序
result = manager_crew.kickoff()

6.3 输入与输出

Crew 支持灵活的输入输出机制：

# 带输入参数的 Crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    verbose=True
)

# 传递输入参数
result = crew.kickoff(
    inputs={
        "topic": "2024年人工智能发展趋势",
        "depth": "深度分析",
        "language": "中文"
    }
)

# 输出结构
print(f"最终结果: {result.raw}")
print(f"Token 使用: {result.token_usage}")
print(f"任务输出列表: {result.tasks_output}")

6.4 回调机制

from crewai import Crew

def on_task_complete(task_output):
    """任务完成时的回调"""
    print(f"任务完成: {task_output.description[:50]}...")
    print(f"输出长度: {len(task_output.raw)} 字符")

def on_crew_complete(crew_output):
    """Crew 完成时的回调"""
    print(f"Crew 执行完成!")
    print(f"总 Token 使用: {crew_output.token_usage}")

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    task_callback=on_task_complete,
    step_callback=lambda step: print(f"步骤: {step}"),
    verbose=True
)

---

7. 流程控制与记忆系统

7.1 记忆系统

CrewAI 的记忆系统让 Agent 能够记住之前的交互和发现：

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    memory=True,           # 启用短期记忆
    embedder={             # 自定义嵌入模型
        "provider": "openai",
        "config": {
            "model": "text-embedding-3-small"
        }
    }
)

记忆系统包含三个层次：

• 短期记忆：当前执行过程中的上下文
• 长期记忆：跨执行的历史信息
• 实体记忆：关于特定实体的知识

7.2 缓存机制

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    cache=True,             # 启用工具调用缓存
    max_rpm=10,             # 限制 API 调用频率
    verbose=True
)

7.3 错误处理

from crewai import Crew
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def safe_crew_execution(crew, inputs=None, max_retries=3):
    """带重试机制的安全 Crew 执行"""
    for attempt in range(max_retries):
        try:
            result = crew.kickoff(inputs=inputs or {})
            return result
        except Exception as e:
            logger.error(f"第 {attempt + 1} 次执行失败: {e}")
            if attempt == max_retries - 1:
                raise
            logger.info(f"正在重试...")

# 使用
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    verbose=True
)

result = safe_crew_execution(crew, {"topic": "AI发展"})

7.4 LangTrace 可观测性

集成 LangTrace 进行执行过程的可视化监控：

# 安装: pip install langtrace-python-sdk
from langtrace_python_sdk import langtrace

langtrace.init(api_key='your_langtrace_api_key')

# 之后的 CrewAI 执行都会自动记录到 LangTrace
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    verbose=True
)

result = crew.kickoff()

---

8. 实战项目一：自动化研究报告生成

8.1 项目目标

构建一个自动化研究系统，能够：

• 自动搜索和收集指定主题的资料
• 分析和整理收集到的信息
• 生成结构化的研究报告
• 对报告进行质量审查

8.2 完整代码实现

# research_crew.py
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileReadTool, FileWriteTool, WebsiteSearchTool
from dotenv import load_dotenv
import os

load_dotenv()

# ===== 工具准备 =====
search_tool = SerperDevTool()
file_write_tool = FileWriteTool()
website_tool = WebsiteSearchTool()

# ===== Agent 定义 =====

# 1. 首席研究员
lead_researcher = Agent(
    role="首席研究员",
    goal="制定研究计划并协调研究工作",
    backstory="""
        你是一位经验丰富的首席研究员，在科技行业有超过15年的研究经验。
        你善于制定系统性的研究计划，能够快速识别关键问题和研究方向。
        你的研究方法严谨，注重数据支撑和多方验证。
    """,
    tools=[search_tool, website_tool],
    verbose=True,
    allow_delegation=True,
    max_iter=30
)

# 2. 数据分析师
data_analyst = Agent(
    role="数据分析师",
    goal="从收集的数据中提取有价值的洞察",
    backstory="""
        你是一位专业的数据分析师，擅长从大量非结构化数据中
        提取关键信息和模式。你善于使用统计方法验证发现，
        并用数据可视化的方式呈现分析结果。
    """,
    tools=[search_tool, file_write_tool],
    verbose=True,
    allow_delegation=False
)

# 3. 技术写作专家
tech_writer = Agent(
    role="技术写作专家",
    goal="将研究成果转化为高质量的报告",
    backstory="""
        你是一位资深的技术写作专家，拥有将复杂技术概念
        转化为通俗易懂内容的特殊能力。你的报告结构清晰、
        逻辑严密、数据准确，深受读者好评。
    """,
    tools=[file_write_tool],
    verbose=True,
    allow_delegation=False
)

# 4. 质量审查员
quality_reviewer = Agent(
    role="质量审查员",
    goal="确保报告的准确性、完整性和可读性",
    backstory="""
        你是一位严格的质量审查专家，对细节有极高的要求。
        你擅长发现报告中的逻辑漏洞、数据错误、表述不清等问题，
        并能提供具体的修改建议。你的审查标准包括：
        - 事实准确性
        - 逻辑一致性
        - 数据可靠性
        - 表达清晰度
        - 结构合理性
    """,
    tools=[file_read_tool],
    verbose=True,
    allow_delegation=False
)

# ===== Task 定义 =====

# 任务1：制定研究计划
planning_task = Task(
    description="""
        针对主题"{topic}"制定详细的研究计划，包括：
        
        1. 研究范围界定
           - 核心问题定义
           - 子主题划分
           - 边界条件设定
        
        2. 信息源规划
           - 需要搜索的关键词列表
           - 目标网站和数据源
           - 需要验证的关键信息
        
        3. 时间和优先级
           - 各子任务的优先级排序
           - 预计工作量分配
        
        请将研究计划保存到文件 research_plan.md
    """,
    expected_output="一份详细的研究计划文档",
    agent=lead_researcher,
    output_file="research_plan.md"
)

# 任务2：执行研究和数据收集
research_task = Task(
    description="""
        按照研究计划，执行以下研究任务：
        
        1. 广度搜索
           - 使用多个关键词组合进行搜索
           - 覆盖技术文档、行业报告、新闻资讯、学术论文等多种来源
           - 记录每个来源的URL和关键信息
        
        2. 深度挖掘
           - 对重要发现进行深入调查
           - 访问相关网页获取详细信息
           - 交叉验证关键数据
        
        3. 信息整理
           - 将收集到的信息分类整理
           - 标记信息的可靠程度
           - 识别信息之间的关联
        
        主题：{topic}
    """,
    expected_output="结构化的原始研究资料，包含分类整理的信息和来源",
    agent=lead_researcher,
    context=[planning_task]
)

# 任务3：数据分析
analysis_task = Task(
    description="""
        对收集到的研究资料进行深入分析：
        
        1. 趋势分析
           - 识别主要发展趋势
           - 分析发展动因
           - 预测未来走向
        
        2. 对比分析
           - 主要参与者/产品对比
           - 技术方案对比
           - 市场表现对比
        
        3. 洞察提取
           - 提炼关键洞察
           - 识别潜在机会和风险
           - 总结核心发现
        
        请将分析结果保存到 analysis_result.md
    """,
    expected_output="深入的分析报告，包含趋势、对比和洞察",
    agent=data_analyst,
    context=[planning_task, research_task],
    output_file="analysis_result.md"
)

# 任务4：撰写报告
writing_task = Task(
    description="""
        基于研究计划、研究资料和分析结果，撰写完整的研究报告：
        
        报告结构要求：
        1. 执行摘要（200-300字）
        2. 研究背景与目的
        3. 方法论说明
        4. 主要发现（分章节详细展开）
        5. 深度分析与洞察
        6. 趋势预测与建议
        7. 结论
        8. 参考资料
        
        写作要求：
        - 语言：专业但易懂的中文
        - 数据：所有关键数据需要注明来源
        - 图表：描述性文字中需要建议插入图表的位置
        - 字数：5000-8000字
        
        主题：{topic}
    """,
    expected_output="一份完整、专业的研究报告",
    agent=tech_writer,
    context=[planning_task, research_task, analysis_task],
    output_file="final_report.md"
)

# 任务5：质量审查
review_task = Task(
    description="""
        对最终报告进行全面质量审查：
        
        审查维度：
        1. 事实准确性
           - 关键数据是否准确
           - 引用是否正确
           - 是否存在过时信息
        
        2. 逻辑一致性
           - 论证逻辑是否严密
           - 各章节是否连贯
           - 结论是否有数据支撑
        
        3. 内容完整性
           - 是否覆盖了所有重要方面
           - 是否有遗漏的关键信息
           - 参考资料是否充分
        
        4. 表达质量
           - 语言是否清晰准确
           - 结构是否合理
           - 是否有错别字或语法错误
        
        输出审查报告，包含发现的问题和修改建议。
    """,
    expected_output="详细的审查报告，列出所有问题和改进建议",
    agent=quality_reviewer,
    context=[writing_task]
)

# ===== Crew 组装 =====
research_crew = Crew(
    agents=[lead_researcher, data_analyst, tech_writer, quality_reviewer],
    tasks=[planning_task, research_task, analysis_task, writing_task, review_task],
    process=Process.sequential,
    memory=True,
    cache=True,
    max_rpm=10,
    verbose=True
)

# ===== 执行 =====
if __name__ == "__main__":
    topic = "2024年大语言模型技术发展趋势与应用前景"
    
    result = research_crew.kickoff(
        inputs={"topic": topic}
    )
    
    print("\n" + "="*60)
    print("研究报告生成完成！")
    print("="*60)
    print(f"\n审查结果：\n{result.raw}")
    print(f"\nToken 使用统计：{result.token_usage}")

8.3 运行与输出

# 运行研究报告生成系统
python3 research_crew.py

运行后，系统会在当前目录生成以下文件：

• research_plan.md - 研究计划
• analysis_result.md - 分析结果
• final_report.md - 最终报告

---

9. 实战项目二：智能内容创作团队

9.1 项目目标

构建一个内容创作团队，能够：

• 根据热点话题生成内容创意
• 撰写多平台适配的内容
• 进行SEO优化
• 内容质量审核

9.2 完整代码实现

# content_crew.py
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileWriteTool, WebsiteSearchTool
from dotenv import load_dotenv
from datetime import datetime

load_dotenv()

# ===== 工具 =====
search = SerperDevTool()
writer_tool = FileWriteTool()
web_search = WebsiteSearchTool()

# ===== Agent 定义 =====

trend_spotter = Agent(
    role="趋势发现者",
    goal="发现和分析当前热门话题和内容趋势",
    backstory="""
        你是一位敏锐的趋势分析师，擅长在社交媒体、
        新闻和行业论坛中发现正在崛起的话题。
        你对内容传播规律有深刻理解，
        能够预测哪些话题会成为爆款。
    """,
    tools=[search, web_search],
    verbose=True
)

content_strategist = Agent(
    role="内容策略师",
    goal="制定有吸引力的内容策略和创意方案",
    backstory="""
        你是一位创意内容策略师，拥有丰富的营销和内容创作经验。
        你善于将热门话题转化为有吸引力的内容创意，
        并能根据不同平台的特点制定差异化的内容策略。
    """,
    tools=[search],
    verbose=True
)

blog_writer = Agent(
    role="博客写手",
    goal="撰写高质量、有深度的博客文章",
    backstory="""
        你是一位资深的技术博客写手，文章风格深入浅出，
        既有技术深度又有可读性。你善于用故事和案例
        来解释复杂概念，让读者容易理解和记忆。
    """,
    tools=[writer_tool],
    verbose=True
)

social_media_writer = Agent(
    role="社交媒体文案",
    goal="创作适合社交媒体传播的短内容",
    backstory="""
        你是一位社交媒体内容专家，精通各平台的内容特点：
        - 微博：简短有力，带话题标签
        - 知乎：专业深入，引发讨论
        - 小红书：生活化，图文并茂描述
        你的文案总能获得高互动率。
    """,
    tools=[writer_tool],
    verbose=True
)

seo_specialist = Agent(
    role="SEO优化专家",
    goal="优化内容的搜索引擎可见性",
    backstory="""
        你是一位SEO专家，精通搜索引擎优化策略。
        你能够分析关键词竞争度、搜索意图，
        并提供标题、元描述、关键词布局等优化建议。
    """,
    tools=[search],
    verbose=True
)

# ===== Task 定义 =====

trend_research = Task(
    description="""
        调研当前"{niche}"领域的热门话题，找出3-5个值得创作内容的方向。
        
        调研维度：
        1. 搜索引擎热门查询
        2. 社交媒体热议话题
        3. 行业最新动态
        4. 用户痛点和需求
        
        对每个方向评估：
        - 热度指数（高/中/低）
        - 竞争程度（高/中/低）
        - 内容缺口（是否存在优质内容）
        - 推荐指数（1-5星）
    """,
    expected_output="3-5个经过评估的内容方向，附带分析依据",
    agent=trend_spotter
)

content_planning = Task(
    description="""
        基于趋势调研结果，制定详细的内容创作计划：
        
        1. 选择最有价值的2-3个话题
        2. 为每个话题设计：
           - 核心观点和角度
           - 目标受众画像
           - 内容大纲
           - 平台适配方案（博客/社交媒体）
           - 预期效果评估
        
        3. 制定发布时间建议
    """,
    expected_output="详细的内容创作计划，包含话题选择、大纲和平台策略",
    agent=content_strategist,
    context=[trend_research]
)

blog_writing = Task(
    description="""
        根据内容计划，撰写一篇高质量的博客文章。
        
        写作要求：
        - 字数：3000-5000字
        - 结构：引言、正文（3-5个主要部分）、总结
        - 风格：专业但易读，有案例和数据支撑
        - 需要包含代码示例或实操步骤（如果适用）
        - 使用Markdown格式
        
        需要包含的内容：
        - 引人入胜的开头
        - 清晰的小标题结构
        - 实用的信息和建议
        - 有说服力的结论
        
        将文章保存为 blog_article.md
    """,
    expected_output="一篇完整的、可直接发布的博客文章",
    agent=blog_writer,
    context=[content_planning],
    output_file="blog_article.md"
)

social_media_content = Task(
    description="""
        根据博客文章内容，创作社交媒体版本：
        
        1. 微博文案（3条，每条不超过140字）
           - 包含相关话题标签
           - 风格各异（提问式、数据式、故事式）
        
        2. 知乎回答（1条，500-800字）
           - 专业深入的风格
           - 引用博客中的核心观点
           - 适合知乎社区调性
        
        3. 小红书笔记（1条）
           - 生活化表达
           - 包含emoji和分段
           - 适合小红书用户习惯
        
        保存为 social_media_content.md
    """,
    expected_output="各平台的社交媒体内容",
    agent=social_media_writer,
    context=[blog_writing],
    output_file="social_media_content.md"
)

seo_optimization = Task(
    description="""
        对博客文章进行SEO优化审查：
        
        1. 关键词分析
           - 主关键词建议（搜索量、竞争度）
           - 长尾关键词建议（5-10个）
           - 关键词密度检查
        
        2. 标题优化
           - SEO友好的标题建议（3个备选）
           - 元描述建议（150字以内）
        
        3. 内容优化建议
           - 内链/外链建议
           - 图片Alt标签建议
           - H1/H2/H3标签优化
           - URL slug建议
        
        4. 竞争分析
           - 同类内容的排名情况
           - 差异化建议
        
        保存为 seo_report.md
    """,
    expected_output="详细的SEO优化报告",
    agent=seo_specialist,
    context=[blog_writing],
    output_file="seo_report.md"
)

# ===== Crew =====
content_crew = Crew(
    agents=[trend_spotter, content_strategist, blog_writer, 
            social_media_writer, seo_specialist],
    tasks=[trend_research, content_planning, blog_writing, 
           social_media_content, seo_optimization],
    process=Process.sequential,
    memory=True,
    verbose=True
)

if __name__ == "__main__":
    result = content_crew.kickoff(
        inputs={"niche": "AI编程助手与开发工具"}
    )
    
    print("\n" + "="*60)
    print("内容创作完成！")
    print("="*60)
    print(f"生成的文件：")
    print(f"  - blog_article.md (博客文章)")
    print(f"  - social_media_content.md (社交媒体内容)")
    print(f"  - seo_report.md (SEO优化报告)")

---

10. 实战项目三：代码审查与优化系统

10.1 项目目标

构建一个自动化代码审查系统，能够：

• 分析代码质量和潜在问题
• 检查安全漏洞
• 提供性能优化建议
• 生成审查报告

10.2 完整代码实现

# code_review_crew.py
from crewai import Agent, Task, Crew, Process
from crewai_tools import FileReadTool, FileWriteTool, DirectoryReadTool
from dotenv import load_dotenv

load_dotenv()

# ===== 工具 =====
file_read = FileReadTool()
file_write = FileWriteTool()
dir_read = DirectoryReadTool()

# ===== Agent 定义 =====

code_analyst = Agent(
    role="代码分析师",
    goal="深入分析代码结构、质量和潜在问题",
    backstory="""
        你是一位拥有12年经验的高级代码分析师。
        你精通多种编程语言（Python、JavaScript、Java、Go等），
        擅长代码静态分析、架构评估和复杂度分析。
        你总是能发现别人忽略的代码问题。
    """,
    tools=[file_read, dir_read],
    verbose=True
)

security_expert = Agent(
    role="安全专家",
    goal="识别代码中的安全漏洞和风险",
    backstory="""
        你是一位资深的应用安全专家，
        精通OWASP Top 10、常见漏洞模式和安全编码实践。
        你曾为多家知名公司进行安全审计，
        能够准确识别SQL注入、XSS、CSRF等安全风险。
    """,
    tools=[file_read],
    verbose=True
)

performance_engineer = Agent(
    role="性能工程师",
    goal="分析代码性能瓶颈并提供优化建议",
    backstory="""
        你是一位性能优化专家，
        擅长识别算法复杂度问题、内存泄漏、
        数据库查询优化和并发处理等性能相关问题。
        你的优化建议总是有具体的数据支撑。
    """,
    tools=[file_read],
    verbose=True
)

review_reporter = Agent(
    role="审查报告撰写者",
    goal="整合各专家意见，生成清晰的审查报告",
    backstory="""
        你是一位技术文档专家，
        擅长将复杂的技术审查结果转化为
        清晰、可操作的报告。你的报告总是
        结构清晰、优先级明确、建议具体。
    """,
    tools=[file_write],
    verbose=True
)

# ===== Task =====

code_analysis_task = Task(
    description="""
        对指定目录的代码进行全面分析：
        
        1. 代码结构分析
           - 目录结构合理性
           - 模块划分清晰度
           - 依赖关系复杂度
        
        2. 代码质量分析
           - 命名规范性
           - 注释质量
           - 代码重复度
           - 函数/方法复杂度
        
        3. 设计模式检查
           - 是否遵循SOLID原则
           - 设计模式使用是否恰当
           - 是否有代码异味
        
        分析目录：{code_directory}
    """,
    expected_output="详细的代码质量分析报告",
    agent=code_analyst
)

security_audit_task = Task(
    description="""
        对代码进行安全审查：
        
        1. 输入验证
           - 是否有适当的输入校验
           - 是否存在注入风险（SQL、NoSQL、命令注入）
        
        2. 认证与授权
           - 认证机制是否安全
           - 权限控制是否完善
           - Session管理是否安全
        
        3. 数据保护
           - 敏感数据是否加密
           - 是否有硬编码的密钥/密码
           - 日志是否泄露敏感信息
        
        4. 常见漏洞
           - XSS风险
           - CSRF防护
           - 文件上传安全
           - 依赖包安全
        
        检查目录：{code_directory}
    """,
    expected_output="安全审查报告，列出所有发现的安全问题和风险等级",
    agent=security_expert,
    context=[code_analysis_task]
)

performance_analysis_task = Task(
    description="""
        对代码进行性能分析：
        
        1. 算法复杂度
           - 识别O(n²)或更高复杂度的代码
           - 循环嵌套问题
           - 递归深度问题
        
        2. 资源使用
           - 内存使用模式
           - 连接池管理
           - 文件句柄管理
        
        3. 数据库相关
           - N+1查询问题
           - 索引使用建议
           - 查询优化建议
        
        4. 并发处理
           - 是否有竞态条件
           - 线程安全问题
           - 异步处理建议
        
        检查目录：{code_directory}
    """,
    expected_output="性能分析报告，包含具体的优化建议和预期效果",
    agent=performance_engineer,
    context=[code_analysis_task]
)

report_generation_task = Task(
    description="""
        整合所有审查结果，生成最终的代码审查报告：
        
        报告结构：
        1. 执行摘要
           - 总体评价（1-10分）
           - 主要发现概述
           - 关键风险提示
        
        2. 问题清单
           - 严重问题（必须修复）
           - 重要问题（建议修复）
           - 改进建议（可选优化）
        
        3. 详细分析
           - 代码质量详细分析
           - 安全问题详细说明
           - 性能问题详细说明
        
        4. 优化路线图
           - 短期修复建议（1-2周）
           - 中期改进计划（1-3个月）
           - 长期优化方向
        
        5. 最佳实践建议
        
        保存为 code_review_report.md
    """,
    expected_output="完整、专业的代码审查报告",
    agent=review_reporter,
    context=[code_analysis_task, security_audit_task, performance_analysis_task],
    output_file="code_review_report.md"
)

# ===== Crew =====
review_crew = Crew(
    agents=[code_analyst, security_expert, performance_engineer, review_reporter],
    tasks=[code_analysis_task, security_audit_task, 
           performance_analysis_task, report_generation_task],
    process=Process.sequential,
    memory=True,
    verbose=True
)

if __name__ == "__main__":
    result = review_crew.kickoff(
        inputs={"code_directory": "./src"}
    )
    
    print("\n" + "="*60)
    print("代码审查完成！")
    print(f"审查报告已保存到 code_review_report.md")

---

11. 高级技巧与最佳实践

11.1 角色设计最佳实践

明确的角色边界

每个 Agent 应该有清晰的职责边界，避免角色重叠：

# 好的设计：职责清晰
researcher = Agent(
    role="市场研究员",
    goal="收集和分析市场数据",
    backstory="专注市场调研，不负责写作",
    allow_delegation=False
)

writer = Agent(
    role="内容写手",
    goal="基于研究结果撰写内容",
    backstory="专注内容创作，不负责数据收集",
    allow_delegation=False
)

互补的技能组合

# 互补的团队组合
analyst = Agent(
    role="定量分析师",
    goal="通过数据分析发现问题",
    backstory="擅长数字和统计分析"
)

psychologist = Agent(
    role="用户心理专家",
    goal="理解用户行为背后的心理动因",
    backstory="擅长用户研究和心理学分析"
)

# 两者结合可以提供更全面的用户洞察

11.2 任务设计最佳实践

渐进式任务分解

# 从宏观到微观的渐进式任务
overview_task = Task(
    description="从宏观角度概述主题全貌",
    expected_output="主题概述",
    agent=researcher
)

deep_dive_task = Task(
    description="基于概述，对关键领域进行深入研究",
    expected_output="深度分析报告",
    agent=researcher,
    context=[overview_task]
)

actionable_task = Task(
    description="基于深度分析，制定具体可执行的建议",
    expected_output="行动建议清单",
    agent=strategist,
    context=[overview_task, deep_dive_task]
)

任务描述的SMART原则

# Specific（具体）
# Measurable（可衡量）
# Achievable（可实现）
# Relevant（相关）
# Time-bound（有时限）

smart_task = Task(
    description="""
        在30分钟内，分析过去7天的GitHub Trending页面，
        识别出Star数增长最快的5个AI相关项目。
        对每个项目提供：
        - 项目简介（100字以内）
        - 核心技术栈
        - 为什么受欢迎的分析（200字以内）
        - 是否值得关注的建议
    """,
    expected_output="5个项目的结构化分析报告",
    agent=researcher
)

11.3 性能优化

减少API调用

# 使用缓存减少重复调用
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    cache=True,  # 启用缓存
    max_rpm=10   # 控制调用频率
)

# 使用更小的模型处理简单任务
from crewai import LLM

fast_llm = LLM(model="gpt-4o-mini")  # 简单任务用小模型
strong_llm = LLM(model="gpt-4o")     # 复杂任务用强模型

simple_agent = Agent(
    role="格式化助手",
    goal="格式化输出内容",
    backstory="负责内容格式化",
    llm=fast_llm  # 使用快速模型
)

complex_agent = Agent(
    role="高级分析师",
    goal="进行深度分析",
    backstory="负责复杂推理",
    llm=strong_llm  # 使用强模型
)

异步执行

# 独立任务使用异步执行
task1 = Task(
    description="独立任务1",
    expected_output="结果1",
    agent=agent1,
    async_execution=True  # 异步执行
)

task2 = Task(
    description="独立任务2",
    expected_output="结果2",
    agent=agent2,
    async_execution=True  # 异步执行
)

task3 = Task(
    description="依赖任务1和2的结果",
    expected_output="最终结果",
    agent=agent3,
    context=[task1, task2]  # 等待前两个任务完成
)

11.4 调试与监控

import logging

# 配置详细日志
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# 自定义回调来监控执行
class CrewMonitor:
    def __init__(self):
        self.task_times = {}
        self.token_usage = {}
    
    def on_task_start(self, task):
        import time
        self.task_times[task.description[:30]] = time.time()
        print(f"▶ 任务开始: {task.description[:50]}...")
    
    def on_task_complete(self, task_output):
        import time
        task_key = task_output.description[:30]
        if task_key in self.task_times:
            duration = time.time() - self.task_times[task_key]
            print(f"✓ 任务完成: {task_key}... (耗时: {duration:.1f}秒)")
        print(f"  输出长度: {len(task_output.raw)} 字符")

monitor = CrewMonitor()

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    step_callback=lambda x: monitor.on_task_complete(x) if hasattr(x, 'raw') else None,
    verbose=True
)

11.5 错误处理与恢复

from crewai import Crew
import json
from datetime import datetime

class CrewExecutor:
    def __init__(self, crew, checkpoint_file="crew_checkpoint.json"):
        self.crew = crew
        self.checkpoint_file = checkpoint_file
    
    def save_checkpoint(self, completed_tasks, current_result):
        checkpoint = {
            "timestamp": datetime.now().isoformat(),
            "completed_tasks": completed_tasks,
            "last_result": current_result
        }
        with open(self.checkpoint_file, 'w') as f:
            json.dump(checkpoint, f, ensure_ascii=False, indent=2)
    
    def execute_with_recovery(self, inputs=None, max_retries=3):
        for attempt in range(max_retries):
            try:
                result = self.crew.kickoff(inputs=inputs or {})
                return result
            except Exception as e:
                print(f"执行失败 (尝试 {attempt + 1}/{max_retries}): {e}")
                if attempt == max_retries - 1:
                    print("已达到最大重试次数")
                    raise
                print("正在重试...")

---

12. 常见问题与排错指南

12.1 API Key 相关问题

问题：API Key 未设置或无效

错误信息: AuthenticationError: Incorrect API key provided

解决方案：

# 检查环境变量
import os
print(f"OPENAI_API_KEY: {'已设置' if os.getenv('OPENAI_API_KEY') else '未设置'}")

# 确保正确加载 .env 文件
from dotenv import load_dotenv
load_dotenv(override=True)  # override=True 确保覆盖已有变量

12.2 模型选择问题

问题：某些模型不可用

# 检查模型是否可用
from openai import OpenAI
client = OpenAI()

try:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "test"}],
        max_tokens=5
    )
    print("GPT-4o 可用")
except Exception as e:
    print(f"GPT-4o 不可用: {e}")

12.3 执行超时问题

问题：任务执行时间过长

# 设置合理的超时和迭代限制
agent = Agent(
    role="研究员",
    goal="研究主题",
    backstory="你是研究员",
    max_iter=15,     # 限制最大迭代次数
    max_rpm=5        # 限制每分钟请求数
)

# 使用更具体、更聚焦的任务描述
task = Task(
    description="请在5分钟内完成以下分析...",  # 添加时间约束
    expected_output="简洁的分析结果",
    agent=agent
)

12.4 输出质量问题

问题：Agent 输出不够详细或准确

# 1. 提供更详细的背景故事
agent = Agent(
    role="分析师",
    goal="提供深入的分析",
    backstory="""
        你是一位资深分析师，拥有10年经验。
        你的分析特点：
        - 数据驱动，所有结论都有数据支撑
        - 逻辑严密，每一步推理都有依据
        - 结论明确，不使用模糊表述
        - 建议具体，每条建议都可执行
    """
)

# 2. 使用更具体的期望输出描述
task = Task(
    description="分析市场趋势",
    expected_output="""
        请提供以下内容：
        1. 市场规模数据（最近3年）
        2. 增长率计算
        3. Top 3 驱动因素
        4. Top 3 风险因素
        5. 未来12个月预测（含置信区间）
        
        所有数据需要注明来源。
    """,
    agent=agent
)

# 3. 使用验证确保输出质量
from pydantic import BaseModel

class AnalysisOutput(BaseModel):
    summary: str
    key_findings: list[str]
    data_sources: list[str]
    confidence_level: str

validated_task = Task(
    description="执行分析",
    expected_output="符合结构要求的分析结果",
    agent=agent,
    output_pydantic=AnalysisOutput
)

12.5 委派机制问题

问题：Agent 之间的委派不生效

# 确保至少有一个 Agent 允许委派
manager = Agent(
    role="项目经理",
    goal="协调团队工作",
    backstory="你是项目经理",
    allow_delegation=True  # 必须设置为 True
)

# 确保 Crew 中有多个 Agent 供委派
crew = Crew(
    agents=[manager, worker1, worker2],  # 需要至少2个 Agent
    tasks=[task1, task2],
    process=Process.hierarchical  # 层级模式更适合委派
)

12.6 内存和性能问题

问题：长时间运行后变慢

# 1. 定期清理不需要的文件
import os
import glob

def cleanup_temp_files():
    for f in glob.glob("temp_*"):
        os.remove(f)

# 2. 使用缓存减少重复计算
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    cache=True  # 启用工具缓存
)

# 3. 控制并发和请求频率
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    max_rpm=10  # 控制每分钟请求数
)

---

总结

CrewAI 是一个强大而灵活的多智能体协作框架，通过本教程，你已经学习了：

1. 核心概念：Agent、Task、Tool、Crew 的基本概念和使用方法
2. 环境搭建：从零开始配置 CrewAI 开发环境
3. Agent 设计：如何设计专业的、互补的 Agent 角色
4. 任务编排：如何设计和编排复杂的任务流程
5. 工具集成：如何使用内置工具和创建自定义工具
6. 实战项目：三个完整的实战项目展示了 CrewAI 的实际应用
7. 最佳实践：性能优化、调试监控、错误处理等高级技巧

下一步学习建议

1. 实践为主：尝试修改本教程中的示例，添加自己的 Agent 和 Task
2. 探索社区：关注 CrewAI GitHub 仓库，了解最新功能
3. 结合其他工具：将 CrewAI 与 LangChain、向量数据库等工具结合使用
4. 生产部署：学习如何将 CrewAI 应用部署到生产环境

参考资源

• CrewAI 官方文档：https://docs.crewai.com
• CrewAI GitHub：https://github.com/crewAIInc/crewAI
• CrewAI 示例库：https://github.com/crewAIInc/crewAI-examples
• CrewAI Discord 社区：https://discord.com/invite/crewai

---

本教程由 AI 辅助生成，内容仅供学习参考。如有疑问或建议，欢迎交流讨论。

最后更新：2026年5月