CrewAI 多智能体协作框架完全教程

一、CrewAI框架概述：多Agent协作的简洁方案

在大语言模型（LLM）应用开发领域，单一Agent的能力已经相当强大，但面对复杂任务时，单个Agent往往力不从心。想象一个真实的企业场景：完成一份市场研究报告，需要有人负责数据收集、有人负责分析、有人负责撰写报告、有人负责审校——这正是多智能体协作的核心思想。

CrewAI 是一个开源的 Python 框架，专门用于编排多个 AI Agent 协同工作。它的设计理念可以用一句话概括：像组建一个真实团队一样组建 AI 团队。与 AutoGen、LangGraph 等框架相比，CrewAI 的最大优势在于其极简的 API 设计和直觉化的角色定义方式。

为什么选择 CrewAI？

1. 简洁优雅的 API：用极少的代码就能定义复杂的多Agent工作流
2. 角色驱动设计：每个 Agent 都有明确的角色、目标和背景故事
3. 灵活的编排模式：支持顺序执行、层级管理等多种协作模式
4. 丰富的工具生态：内置大量工具，并支持自定义工具扩展
5. 记忆系统：支持短期、长期和实体记忆，让 Agent 具备学习能力
6. 人类介入：天然支持 Human-in-the-loop，适合需要人工审核的场景

与其他框架的对比

二、核心概念：Agent、Task、Crew、Process

CrewAI 的核心由四个概念构成，理解它们是掌握整个框架的基础。

2.1 Agent（智能体）

Agent 是 CrewAI 中的基本执行单元。每个 Agent 都是一个具有特定角色的 AI 实体，它拥有：

• role（角色）：Agent 的身份标识，如"资深数据分析师"
• goal（目标）：Agent 需要达成的目标
• backstory（背景故事）：为 Agent 提供上下文和行为动机
• tools（工具）：Agent 可以使用的工具列表
• llm：Agent 使用的语言模型
• memory（记忆）：是否启用记忆系统
• allow_delegation（允许委派）：是否可以将子任务委派给其他 Agent

from crewai import Agent

researcher = Agent(
    role="资深市场研究员",
    goal="深入分析目标市场的最新趋势和竞争格局",
    backstory="""你是一位拥有10年市场研究经验的资深分析师。
    你擅长从海量数据中提取关键洞察，对科技行业有深刻理解。
    你的分析报告曾多次帮助企业做出正确的战略决策。""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, scrape_tool],
    llm=my_llm
)

2.2 Task（任务）

Task 是分配给 Agent 的具体工作单元。每个 Task 包含：

• description（描述）：任务的详细说明
• expected_output（期望输出）：对输出结果的期望描述
• agent（执行者）：负责执行此任务的 Agent
• context（上下文）：来自其他 Task 的输出作为输入
• output_file（输出文件）：结果保存的文件路径

from crewai import Task

research_task = Task(
    description="""对2024年全球AI芯片市场进行全面调研，
    包括主要厂商的市场份额、技术路线、产品布局等。
    重点关注NVIDIA、AMD、Intel以及新兴玩家的动态。""",
    expected_output="""一份结构清晰的市场调研报告，包含：
    1. 市场规模与增长预测
    2. 主要厂商竞争分析
    3. 技术趋势与创新方向
    4. 投资建议""",
    agent=researcher,
    output_file="market_research.md"
)

2.3 Crew（团队）

Crew 是 Agent 和 Task 的容器，负责协调整个工作流程：

• agents：团队成员列表
• tasks：任务列表
• process：执行流程类型
• verbose：是否输出详细日志
• memory：是否启用团队记忆

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,
    verbose=True,
    memory=True
)

2.4 Process（流程）

Process 定义了任务的执行方式：

• Process.sequential：任务按顺序依次执行，前一个任务的输出作为后一个的输入
• Process.hierarchical：由一个 Manager Agent 协调任务分配，支持动态任务规划

三、Agent角色设计：目标、背景、工具、记忆

设计好的 Agent 角色是 CrewAI 项目成功的关键。一个角色定义良好的 Agent，其表现可以远超通用的 LLM 调用。

3.1 角色设计原则

原则一：角色要具体

# ❌ 不好的角色定义
agent = Agent(
    role="助手",
    goal="帮助用户",
    backstory="你是一个AI助手"
)

# ✅ 好的角色定义
agent = Agent(
    role="Python性能优化专家",
    goal="通过代码分析和重构，将Python应用的性能提升至少50%",
    backstory="""你是一位在Python性能优化领域深耕15年的工程师。
    你熟悉CPython内部机制、GIL的影响、内存管理和各种性能分析工具。
    你曾为多家知名科技公司优化核心服务，将响应时间从秒级降低到毫秒级。
    你信奉'测量先行'的原则，绝不凭直觉优化。"""
)

原则二：目标要可衡量

模糊的目标会导致 Agent 行为不确定。给 Agent 明确的成功标准：

analyst = Agent(
    role="财务数据分析师",
    goal="从提供的财务数据中识别出至少3个关键风险指标，并给出量化评估",
    # ...
)

原则三：背景故事要提供上下文

背景故事不是装饰，它为 Agent 提供了决策框架：

editor = Agent(
    role="科技媒体主编",
    goal="确保文章质量达到发表标准",
    backstory="""你是一位严格的科技媒体主编，拥有计算机科学博士学位。
    你坚持技术准确性第一的原则，对代码示例要求能直接运行。
    你的编辑风格是：去掉所有废话，让每个词都有价值。
    你特别关注：数据来源是否可靠、技术细节是否准确、逻辑是否自洽。""",
    # ...
)

3.2 高级角色配置

多模型混合使用：不同 Agent 可以使用不同的 LLM，根据任务特点选择最合适的模型。

from langchain_openai import ChatOpenAI

# 用GPT-4处理复杂推理
analyst = Agent(
    role="战略分析师",
    llm=ChatOpenAI(model="gpt-4o"),
    # ...
)

# 用GPT-3.5处理信息整理
organizer = Agent(
    role="信息整理员",
    llm=ChatOpenAI(model="gpt-3.5-turbo"),
    # ...
)

委派机制：允许 Agent 将子任务委派给团队中的其他 Agent。

manager = Agent(
    role="项目经理",
    goal="确保项目按时高质量完成",
    allow_delegation=True,  # 允许委派
    # ...
)

四、Task定义与依赖管理

4.1 Task 的高级用法

上下文传递：Task 之间可以通过 context 参数传递信息。

research = Task(
    description="调研AI Agent框架市场现状",
    expected_output="详细的市场调研报告",
    agent=researcher
)

analysis = Task(
    description="基于调研结果，分析各框架的优劣势",
    expected_output="对比分析表和推荐方案",
    agent=analyst,
    context=[research]  # 将research任务的输出作为输入
)

report = Task(
    description="撰写最终的技术选型报告",
    expected_output="完整的Markdown格式报告",
    agent=writer,
    context=[research, analysis]  # 可以接收多个任务的输出
)

4.2 异步执行 Task

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

task = Task(
    description="分析大规模数据集",
    expected_output="分析报告",
    agent=analyst,
    async_execution=True  # 异步执行
)

4.3 条件执行与回调

def task_callback(task_output):
    """Task完成后的回调函数"""
    print(f"任务完成，输出长度: {len(task_output.raw)}")
    # 可以在这里做日志记录、数据持久化等

task = Task(
    description="生成报告摘要",
    expected_output="简洁的摘要文本",
    agent=writer,
    callback=task_callback,
    output_file="summary.md"
)

4.4 Task 输出类型

CrewAI 支持多种输出格式：

from crewai import Task

# 默认输出
task = Task(
    description="分析数据",
    expected_output="分析结果",
    agent=analyst
)

# JSON输出
task = Task(
    description="提取结构化信息",
    expected_output="JSON格式的结构化数据",
    agent=analyst,
    output_json=True
)

# Pydantic模型输出
from pydantic import BaseModel

class AnalysisResult(BaseModel):
    summary: str
    key_findings: list[str]
    risk_level: str

task = Task(
    description="分析风险",
    expected_output="结构化的风险分析结果",
    agent=analyst,
    output_pydantic=AnalysisResult
)

五、Crew编排：顺序执行、层级管理、共识机制

5.1 顺序执行（Sequential Process）

最简单的编排模式，适合有明确先后顺序的工作流：

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, analyst, writer, reviewer],
    tasks=[research_task, analysis_task, writing_task, review_task],
    process=Process.sequential,
    verbose=True
)

# 执行流程：researcher → analyst → writer → reviewer
result = crew.kickoff()

适用场景：

• 内容生产流水线（调研 → 写作 → 编辑 → 审核）
• 数据处理管道（采集 → 清洗 → 分析 → 可视化）
• 代码开发流程（需求分析 → 编码 → 测试 → 部署）

5.2 层级管理（Hierarchical Process）

由一个 Manager Agent 自动协调任务分配：

from crewai import Crew, Process

manager = Agent(
    role="项目总监",
    goal="高效协调团队完成项目目标",
    backstory="你是一位经验丰富的项目经理...",
    allow_delegation=True,
    verbose=True
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4o"),
    verbose=True
)

在层级模式下，Manager Agent 会：

1. 分析所有任务
2. 决定任务执行顺序
3. 将任务分配给最合适的 Agent
4. 监控执行进度
5. 必要时重新分配或调整

5.3 共识机制

CrewAI 支持通过设计实现简单的共识机制：

# 设计一个需要多方确认的流程
validator1 = Agent(
    role="技术审核员",
    goal="从技术角度验证方案的可行性",
    # ...
)

validator2 = Agent(
    role="业务审核员",
    goal="从业务角度验证方案的合理性",
    # ...
)

consensus_task = Task(
    description="综合技术审核和业务审核的意见，给出最终结论",
    expected_output="综合评估报告",
    agent=decision_maker,
    context=[tech_review_task, biz_review_task]
)

六、工具集成：自定义工具、第三方工具库

6.1 使用 CrewAI 内置工具

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

from crewai_tools import (
    SerperDevTool,           # 搜索工具
    ScrapeWebsiteTool,       # 网页抓取
    FileReadTool,            # 文件读取
    FileWriteTool,           # 文件写取
    DirectoryReadTool,       # 目录读取
    CodeInterpreterTool,     # 代码执行
    CSVSearchTool,           # CSV搜索
    PDFSearchTool,           # PDF搜索
    WebsiteSearchTool,       # 网站搜索
    YoutubeVideoSearchTool,  # YouTube搜索
)

# 创建搜索工具
search_tool = SerperDevTool()

# 创建网页抓取工具
scrape_tool = ScrapeWebsiteTool()

# 创建文件读取工具
file_tool = FileReadTool(file_path="data/report.txt")

6.2 自定义工具

方式一：使用 @tool 装饰器

from crewai_tools import tool

@tool("股票价格查询工具")
def get_stock_price(symbol: str) -> str:
    """查询指定股票的实时价格。
    
    Args:
        symbol: 股票代码，如 AAPL, GOOGL
    Returns:
        股票价格信息
    """
    # 这里可以调用真实的股票API
    import yfinance as stock_ticker
    stock = stock_ticker.Ticker(symbol)
    price = stock.info.get("currentPrice", "N/A")
    return f"{symbol} 当前价格: ${price}"

# 使用自定义工具
agent = Agent(
    role="投资分析师",
    tools=[get_stock_price],
    # ...
)

方式二：继承 BaseTool 类

from crewai_tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type

class DatabaseQueryInput(BaseModel):
    query: str = Field(..., description="SQL查询语句")
    database: str = Field(default="main", description="数据库名称")

class DatabaseQueryTool(BaseTool):
    name: str = "数据库查询工具"
    description: str = "执行SQL查询并返回结果"
    args_schema: Type[BaseModel] = DatabaseQueryInput

    def _run(self, query: str, database: str = "main") -> str:
        # 实际的数据库查询逻辑
        import sqlite3
        conn = sqlite3.connect(f"{database}.db")
        cursor = conn.execute(query)
        results = cursor.fetchall()
        conn.close()
        return str(results)

db_tool = DatabaseQueryTool()

6.3 集成第三方工具

# 集成LangChain工具
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

wikipedia_tool = WikipediaQueryRun(
    api_wrapper=WikipediaAPIWrapper()
)

# 将LangChain工具适配到CrewAI
from crewai_tools import tool

@tool("Wikipedia搜索")
def search_wikipedia(query: str) -> str:
    """在Wikipedia上搜索相关信息"""
    return wikipedia_tool.run(query)

七、记忆与学习：短期、长期、实体记忆

CrewAI 的记忆系统是其最强大的特性之一，让 Agent 能够从历史交互中学习。

7.1 记忆系统架构

CrewAI 的记忆系统包含三个层次：

1. 短期记忆（Short-term Memory）：当前任务执行中的临时记忆
2. 长期记忆（Long-term Memory）：跨任务、跨会话的持久化记忆
3. 实体记忆（Entity Memory）：对特定实体（人物、组织、概念）的记忆

7.2 启用记忆系统

from crewai import Crew

crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    memory=True,  # 启用记忆系统
    verbose=True
)

7.3 自定义记忆存储

from crewai import Crew
from crewai.memory import LongTermMemory, ShortTermMemory, EntityMemory
from crewai.memory.storage import (
    SQLiteStorage,
    RAGStorage
)

# 使用SQLite存储长期记忆
long_term_storage = SQLiteStorage(db_path="./memory/long_term.db")

# 使用RAG存储实体记忆（基于向量数据库）
entity_storage = RAGStorage(
    embedder_config={
        "provider": "openai",
        "config": {
            "model": "text-embedding-3-small"
        }
    }
)

crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    memory=True,
    long_term_memory=LongTermMemory(storage=long_term_storage),
    entity_memory=EntityMemory(storage=entity_storage),
    verbose=True
)

7.4 记忆的实际应用

# 场景：多轮研究任务
# 第一轮：研究AI行业
crew1 = Crew(
    agents=[researcher],
    tasks=[ai_research_task],
    memory=True,
    verbose=True
)
result1 = crew1.kickoff()

# 第二轮：基于第一轮记忆，研究云计算行业
# Agent会记住之前的研究方法和发现
crew2 = Crew(
    agents=[researcher],
    tasks=[cloud_research_task],
    memory=True,
    verbose=True
)
result2 = crew2.kickoff()
# researcher会利用之前学到的研究方法和分析框架

八、与LangChain/LlamaIndex集成

8.1 与 LangChain 集成

CrewAI 底层使用 LangChain 作为 LLM 接口，因此天然支持 LangChain 生态。

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_community.tools import DuckDuckGoSearchRun

# 使用不同的LLM provider
openai_llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
anthropic_llm = ChatAnthropic(model="claude-3-opus-20240229")

# 为不同Agent分配不同模型
researcher = Agent(
    role="研究员",
    llm=openai_llm,
    # ...
)

writer = Agent(
    role="撰稿人",
    llm=anthropic_llm,
    # ...
)

# 使用LangChain工具
ddg_search = DuckDuckGoSearchRun()

@tool("网页搜索")
def search_web(query: str) -> str:
    """搜索网页获取最新信息"""
    return ddg_search.run(query)

8.2 与 LlamaIndex 集成

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from crewai_tools import tool

# 创建LlamaIndex查询引擎
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

@tool("知识库查询")
def query_knowledge_base(question: str) -> str:
    """从本地知识库中查询信息"""
    response = query_engine.query(question)
    return str(response)

# 将知识库工具分配给Agent
analyst = Agent(
    role="知识分析师",
    tools=[query_knowledge_base],
    # ...
)

8.3 自定义LLM Provider

from langchain_openai import ChatOpenAI

# 使用本地部署的模型
local_llm = ChatOpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama",
    model="qwen2.5:72b",
    temperature=0.3
)

# 使用vLLM部署的模型
vllm_llm = ChatOpenAI(
    base_url="http://localhost:8000/v1",
    api_key="token-abc123",
    model="meta-llama/Llama-3-70b-chat-hf",
    temperature=0.5
)

九、人类介入机制（Human-in-the-loop）

9.1 为什么需要人类介入？

在很多场景中，完全自动化是不够的：

• 关键决策：投资建议、合同审核等需要人工确认
• 质量控制：发布前的内容审核
• 异常处理：当 Agent 不确定时请求人类指导

9.2 启用人类介入

from crewai import Agent, Task, Crew

# 在Agent级别启用人类介入
reviewer = Agent(
    role="内容审核员",
    goal="确保内容质量",
    human_input=True,  # 启用人类输入
    # ...
)

# 在Task级别启用人类介入
final_review = Task(
    description="对最终报告进行审核",
    expected_output="审核意见",
    agent=reviewer,
    human_input=True  # 此任务需要人类确认
)

9.3 实现审批流程

from crewai import Crew, Task, Agent

# 定义一个需要多级审批的流程
writer = Agent(role="撰稿人", goal="撰写高质量文章", ...)
editor = Agent(role="编辑", goal="确保文章质量", ...)
approver = Agent(role="总编", goal="做出最终发布决定", human_input=True, ...)

draft_task = Task(
    description="撰写初稿",
    expected_output="文章初稿",
    agent=writer
)

edit_task = Task(
    description="编辑和修改文章",
    expected_output="修改后的文章",
    agent=editor,
    context=[draft_task]
)

approval_task = Task(
    description="审核文章是否可以发布",
    expected_output="发布决定和修改建议",
    agent=approver,
    context=[edit_task],
    human_input=True  # 需要人类总编确认
)

crew = Crew(
    agents=[writer, editor, approver],
    tasks=[draft_task, edit_task, approval_task],
    verbose=True
)

9.4 人类反馈的处理

# CrewAI会在human_input为True时暂停执行，等待输入
# 人类的反馈会被注入到Agent的上下文中，影响后续行为
result = crew.kickoff()
# 执行流程：
# 1. writer撰写初稿
# 2. editor编辑文章
# 3. 暂停，等待人类输入审核意见
# 4. approver根据人类意见做出决定

十、性能优化与调试技巧

10.1 Token 使用优化

# 1. 使用更小的模型处理简单任务
from langchain_openai import ChatOpenAI

fast_llm = ChatOpenAI(model="gpt-3.5-turbo", max_tokens=500)
smart_llm = ChatOpenAI(model="gpt-4o", max_tokens=2000)

formatter = Agent(
    role="格式化助手",
    llm=fast_llm,  # 简单任务用小模型
    # ...
)

strategist = Agent(
    role="战略顾问",
    llm=smart_llm,  # 复杂任务用大模型
    # ...
)

# 2. 控制输出长度
task = Task(
    description="用不超过500字总结报告要点",
    expected_output="500字以内的摘要",
    agent=writer
)

10.2 调试技巧

# 1. 启用详细日志
crew = Crew(
    agents=[...],
    tasks=[...],
    verbose=True  # 输出详细的执行过程
)

# 2. 使用step_callback监控每一步
def step_callback(step_output):
    """每步执行后的回调"""
    print(f"Step输出: {step_output}")
    # 可以记录到文件或监控系统

crew = Crew(
    agents=[...],
    tasks=[...],
    step_callback=step_callback
)

# 3. 使用LangChain的调试模式
import langchain
langchain.debug = True

10.3 错误处理

import logging

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

try:
    result = crew.kickoff()
except Exception as e:
    logger.error(f"Crew执行失败: {e}")
    # 可以实现重试逻辑
    for attempt in range(3):
        try:
            result = crew.kickoff()
            break
        except Exception as retry_error:
            logger.warning(f"重试 {attempt + 1} 失败: {retry_error}")

10.4 并发与异步优化

import asyncio
from crewai import Crew

async def run_crew_async():
    """异步执行Crew"""
    crew = Crew(
        agents=[researcher, analyst],
        tasks=[research_task, analysis_task],
        verbose=True
    )
    result = await crew.kickoff_async()
    return result

# 并行执行多个独立的Crew
async def run_parallel_crews():
    tasks = [
        run_crew_async(),
        run_crew_async(),
        run_crew_async()
    ]
    results = await asyncio.gather(*tasks)
    return results

十一、实战项目：构建自动化研究分析团队

11.1 项目概述

我们将构建一个完整的自动化研究分析团队，能够：

1. 自动搜索和收集指定主题的信息
2. 对收集到的数据进行深度分析
3. 生成结构化的研究报告
4. 对报告进行质量审核

11.2 完整代码实现

"""
CrewAI 自动化研究分析团队
功能：自动完成从信息收集到报告生成的全流程
"""

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

# ==================== 环境配置 ====================
os.environ["OPENAI_API_KEY"] = "your-api-key"
os.environ["SERPER_API_KEY"] = "your-serper-key"

# ==================== 工具准备 ====================
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
file_write_tool = FileWriteTool()

# ==================== LLM配置 ====================
# 使用GPT-4o处理复杂分析
smart_llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
# 使用GPT-3.5处理信息整理
fast_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.5)

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

# 信息收集专家
collector = Agent(
    role="信息收集专家",
    goal="高效收集指定主题的全面、准确、最新的信息",
    backstory="""你是一位资深的信息收集专家，擅长使用各种搜索工具和数据源。
    你有极强的信息筛选能力，能快速识别高质量、可靠的信息来源。
    你特别擅长：
    - 使用精确的搜索关键词找到最相关的信息
    - 从大量网页中提取关键数据点
    - 识别信息的可信度和时效性
    - 整理和归类收集到的信息""",
    tools=[search_tool, scrape_tool],
    llm=smart_llm,
    verbose=True,
    allow_delegation=False,
    max_iter=15
)

# 数据分析师
analyst = Agent(
    role="高级数据分析师",
    goal="从收集到的信息中提取深度洞察和趋势分析",
    backstory="""你是一位拥有统计学和计算机科学双学位的高级数据分析师。
    你擅长从非结构化数据中提取结构化信息，发现隐藏的模式和趋势。
    你的分析方法包括：
    - SWOT分析
    - 趋势分析
    - 竞争格局分析
    - 技术路线图分析
    - 市场规模预测""",
    llm=smart_llm,
    verbose=True,
    allow_delegation=False
)

# 报告撰写专家
writer = Agent(
    role="专业报告撰写专家",
    goal="将分析结果转化为清晰、专业、有说服力的研究报告",
    backstory="""你是一位资深的技术写作者，曾为多家顶级咨询公司撰写研究报告。
    你的写作风格：清晰、简洁、有逻辑、数据驱动。
    你的报告特点：
    - 结构清晰，使用标准的研究报告框架
    - 数据准确，每个结论都有数据支撑
    - 观点明确，给出可操作的建议
    - 语言专业但不晦涩""",
    tools=[file_write_tool],
    llm=smart_llm,
    verbose=True,
    allow_delegation=False
)

# 质量审核员
reviewer = Agent(
    role="质量审核专家",
    goal="确保报告的准确性、完整性和专业性",
    backstory="""你是一位严谨的质量审核专家，拥有多年学术论文和商业报告的审核经验。
    你对报告的要求极为严格：
    - 逻辑链条必须完整
    - 数据必须有来源
    - 结论必须有依据
    - 格式必须规范
    - 语言必须专业""",
    llm=smart_llm,
    verbose=True,
    allow_delegation=False,
    human_input=True  # 审核结果需要人类确认
)

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

def create_research_tasks(topic: str):
    """根据研究主题创建完整的任务链"""
    
    # 任务1：信息收集
    collection_task = Task(
        description=f"""对以下主题进行全面的信息收集：
        
        研究主题：{topic}
        
        要求：
        1. 搜索至少5个不同的信息来源
        2. 收集的信息应包括：背景概述、最新发展、主要参与者、技术趋势、市场数据
        3. 标注每个信息的来源和可信度
        4. 整理成结构化的信息汇编""",
        expected_output="""一份结构化的信息汇编，包含：
        - 背景概述（500字以上）
        - 最新发展动态（至少5条）
        - 主要参与者列表及简介
        - 关键数据点汇总
        - 信息来源标注""",
        agent=collector
    )
    
    # 任务2：深度分析
    analysis_task = Task(
        description=f"""基于收集到的信息，对"{topic}"进行深度分析：
        
        分析维度：
        1. 市场/行业现状分析
        2. 竞争格局分析
        3. 技术发展趋势
        4. 机遇与挑战分析（SWOT）
        5. 未来发展趋势预测""",
        expected_output="""一份深度分析报告，包含：
        - 市场现状概述
        - 竞争格局图表描述
        - 技术演进路线
        - SWOT分析矩阵
        - 未来3-5年发展预测""",
        agent=analyst,
        context=[collection_task]
    )
    
    # 任务3：报告撰写
    writing_task = Task(
        description=f"""将分析结果撰写成一份完整的研究报告：
        
        主题：{topic}
        
        报告结构要求：
        1. 执行摘要（200-300字）
        2. 研究背景与方法
        3. 市场/行业分析
        4. 深度洞察
        5. 机遇与挑战
        6. 战略建议
        7. 结论与展望
        8. 参考文献
        
        格式要求：
        - 使用Markdown格式
        - 适当使用标题、列表、表格
        - 语言专业、简洁""",
        expected_output="一份完整的、可直接使用的Markdown格式研究报告",
        agent=writer,
        context=[collection_task, analysis_task],
        output_file="research_report.md"
    )
    
    # 任务4：质量审核
    review_task = Task(
        description="对研究报告进行全面质量审核",
        expected_output="""审核报告，包含：
        - 整体评分（1-10分）
        - 优点列表
        - 需改进的地方
        - 最终发布建议（发布/修改后发布/重写）""",
        agent=reviewer,
        context=[writing_task],
        human_input=True
    )
    
    return [collection_task, analysis_task, writing_task, review_task]

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

def run_research_team(topic: str):
    """运行研究团队"""
    
    tasks = create_research_tasks(topic)
    
    crew = Crew(
        agents=[collector, analyst, writer, reviewer],
        tasks=tasks,
        process=Process.sequential,
        memory=True,
        verbose=True
    )
    
    print(f"🚀 开始研究: {topic}")
    print("=" * 60)
    
    result = crew.kickoff()
    
    print("=" * 60)
    print("✅ 研究完成!")
    print(f"📄 报告已保存到: research_report.md")
    
    return result

# ==================== 主程序 ====================

if __name__ == "__main__":
    # 运行研究团队
    topic = "2024年全球人工智能芯片市场发展趋势分析"
    result = run_research_team(topic)
    print(result)

11.3 运行与测试

# 安装依赖
pip install crewai crewai-tools langchain-openai

# 设置环境变量
export OPENAI_API_KEY="your-api-key"
export SERPER_API_KEY="your-serper-key"

# 运行
python research_team.py

11.4 扩展建议

1. 添加更多专业Agent：如行业专家、法务顾问等
2. 集成更多数据源：如数据库、API、PDF文档等
3. 添加缓存机制：避免重复搜索相同内容
4. 实现并行执行：对独立任务使用异步执行
5. 添加Web界面：使用 Streamlit 或 Gradio 构建交互界面

总结

CrewAI 通过其直觉化的角色定义和灵活的编排模式，让多智能体协作变得简单而强大。本教程从核心概念到实战项目，系统地介绍了 CrewAI 的方方面面。

关键要点回顾：

1. Agent 设计是核心：好的角色定义决定了团队的表现
2. Task 依赖管理很重要：合理设计任务间的数据流
3. 工具是 Agent 的手脚：根据任务需求选择合适的工具
4. 记忆让 Agent 成长：启用记忆系统提升长期表现
5. 人类介入保质量：关键环节加入人工审核

CrewAI 正在快速发展，新特性不断推出。建议持续关注官方文档和社区动态，将最新的最佳实践应用到你的项目中。

本教程由 AI 教程助手生成，内容基于 CrewAI 框架的核心设计理念和实践经验编写。