CrewAI 多智能体框架实战：角色扮演与团队协作

Crew（团队）
├── Agent 1: 研究员（搜索资料）
├── Agent 2: 分析师（分析数据）
├── Agent 3: 写作者（生成报告）
├── Task 1: 搜索相关资料
├── Task 2: 分析搜索结果
├── Task 3: 撰写总结报告
└── Process: 顺序执行 / 层级管理

from crewai import Agent, Task, Crew, Process

# 1. 定义 Agent
researcher = Agent(
    role="资深研究员",
    goal="找到关于 {topic} 的最新、最准确的信息",
    backstory="你是一位有 20 年经验的研究员，擅长从海量信息中提取关键洞察。",
    verbose=True,
    llm="qwen3-72b",
    tools=[search_tool, scrape_tool],
)

analyst = Agent(
    role="数据分析师",
    goal="深入分析研究结果，发现趋势和模式",
    backstory="你是前麦肯锡分析师，擅长数据驱动的战略分析。",
    verbose=True,
    llm="qwen3-72b",
)

writer = Agent(
    role="技术写作专家",
    goal="将分析结果转化为清晰、有洞见的报告",
    backstory="你是前 Wired 杂志编辑，擅长把复杂技术概念讲得通俗易懂。",
    verbose=True,
    llm="qwen3-72b",
)

# 2. 定义 Task
research_task = Task(
    description="研究 {topic} 的最新进展，包括技术突破、市场动态和未来趋势。",
    expected_output="一份包含 10 个关键发现的研究简报",
    agent=researcher,
)

analysis_task = Task(
    description="基于研究结果，分析 {topic} 对行业的影响和机会。",
    expected_output="一份包含 SWOT 分析和 3 个战略建议的分析报告",
    agent=analyst,
    context=[research_task],  # 依赖研究任务的输出
)

writing_task = Task(
    description="将分析和研究整合为一份面向高管的执行摘要。",
    expected_output="一份 2000 字的执行摘要，包含图表建议",
    agent=writer,
    context=[research_task, analysis_task],
)

# 3. 组建 Crew
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,  # 顺序执行
    verbose=True,
)

# 4. 执行
result = crew.kickoff(inputs={"topic": "AI Agent 框架"})

# 任务按顺序执行，前者输出作为后者输入
crew = Crew(
    agents=[agent_a, agent_b, agent_c],
    tasks=[task_1, task_2, task_3],
    process=Process.sequential,
)
# task_1 → task_2 → task_3

# 经理 Agent 负责分配任务
manager = Agent(
    role="项目经理",
    goal="协调团队高效完成任务",
    backstory="你是经验丰富的 PM，擅长任务分解和团队管理。",
    allow_delegation=True,
)

crew = Crew(
    agents=[researcher, analyst, writer, manager],
    tasks=[complex_task],
    process=Process.hierarchical,
    manager_llm="qwen3-72b",
)
# manager 分配子任务给其他 Agent

from crewai import Pipeline

# 自定义流程：研究→分析→如果需要→再研究
pipeline = Pipeline(
    stages=[
        research_crew,    # 第一阶段：研究
        review_crew,      # 第二阶段：审查
        revision_crew,    # 第三阶段：修订（如果审查不通过）
    ]
)

# Agent 可以委派任务给其他 Agent
researcher = Agent(
    role="研究员",
    goal="研究给定主题",
    allow_delegation=False,  # 不允许委派
)

manager = Agent(
    role="项目经理",
    goal="管理项目进度",
    allow_delegation=True,   # 允许委派
)

# 委派示例
# manager: "我需要数据分析，请 analyst 处理"
# analyst: "收到，开始分析..."
# analyst: "分析完成，结果如下..."
# manager: "收到，请 writer 基于分析结果写报告"

from crewai_tools import (
    SerperDevTool,       # Google 搜索
    ScrapeWebsiteTool,    # 网页抓取
    PDFSearchTool,        # PDF 搜索
    CSVSearchTool,        # CSV 搜索
    DirectoryReadTool,    # 目录读取
    FileReadTool,         # 文件读取
)

# 预置工具
search = SerperDevTool()
scraper = ScrapeWebsiteTool()

# 自定义工具
from crewai_tools import BaseTool

class DatabaseQueryTool(BaseTool):
    name: str = "数据库查询"
    description: str = "查询业务数据库，返回销售数据"
    
    def _run(self, query: str) -> str:
        results = db.execute(query)
        return json.dumps(results)

db_tool = DatabaseQueryTool()

researcher = Agent(
    role="数据研究员",
    tools=[search, scraper, db_tool],
)

# CrewAI 内置记忆系统
crew = Crew(
    agents=[agent_a, agent_b],
    tasks=[task_1, task_2],
    memory=True,                    # 启用记忆
    embedder={                      # 配置嵌入模型
        "provider": "ollama",
        "config": {"model": "nomic-embed-text"},
    },
)

# 记忆类型：
# 1. 短期记忆：当前任务上下文
# 2. 长期记忆：跨任务的知识存储
# 3. 实体记忆：记住关键实体（人名、公司等）

# 案例：调研某个技术方向

# Agent 定义
tech_researcher = Agent(
    role="技术研究员",
    goal="深入调研 {tech} 的技术原理、优缺点和适用场景",
    backstory="你是前 Google 研究员，擅长技术深度分析。",
    tools=[search, scraper, github_tool],
)

market_analyst = Agent(
    role="市场分析师",
    goal="分析 {tech} 的市场采用情况、竞品对比和投资机会",
    backstory="你是 a16z 的技术投资人，关注新兴技术趋势。",
    tools=[search, crunchbase_tool],
)

architect = Agent(
    role="解决方案架构师",
    goal="设计基于 {tech} 的技术方案和实施路径",
    backstory="你是 AWS 首席架构师，擅长将技术转化为方案。",
)

# 任务定义
tasks = [
    Task(
        description="调研 {tech} 的核心技术原理、开源实现和性能基准",
        agent=tech_researcher,
        expected_output="技术调研报告（含代码示例和性能对比）",
    ),
    Task(
        description="分析 {tech} 的市场格局、主要玩家和投资趋势",
        agent=market_analyst,
        expected_output="市场分析报告（含竞品矩阵和投资建议）",
        context=[tasks[0]],
    ),
    Task(
        description="基于调研结果，设计一个 POC 技术方案",
        agent=architect,
        expected_output="技术方案文档（含架构图和实施计划）",
        context=[tasks[0], tasks[1]],
    ),
]

crew = Crew(agents=[tech_researcher, market_analyst, architect], tasks=tasks)
result = crew.kickoff(inputs={"tech": "MCP 协议"})

best_practices = {
    "角色定义": "backstory 越具体，Agent 表现越好",
    "任务粒度": "每个任务应该有明确的输入和输出",
    "工具选择": "给 Agent 它需要的工具，不要给多余的",
    "LLM 选择": "规划用大模型（72B），执行用小模型（7B）",
    "成本控制": "设置 max_iter 和 max_rpm 限制",
    "结果验证": "添加 review_task 让另一个 Agent 验证结果",
}