16 | 实战——用 CrewAI 构建多 Agent 协作系统

python -m venv crewai-demo
cd crewai-demo
# Windows
.venv\Scripts\activate
# macOS/Linux
source .venv/bin/activate

pip install crewai crewai-tools

import crewai
print(crewai.__version__)
# 输出: 0.105.2

export OPENAI_API_KEY="sk-your-key-here"
# Windows PowerShell
$env:OPENAI_API_KEY="sk-your-key-here"

export ANTHROPIC_API_KEY="sk-ant-your-key-here"

from crewai import Agent
from crewai_tools import SerperDevTool

# 搜索工具（后面 Step 3 会详细讲）
search_tool = SerperDevTool()

# 研究员 Agent
researcher = Agent(
    role="行业研究员",
    goal="针对 {industry} 行业，搜集最新的市场数据、技术进展、主要公司和行业趋势",
    backstory="""你是一位资深行业研究员，在麦肯锡和贝恩有超过 10 年的咨询经验。
你擅长从海量信息中快速识别关键数据点，尤其关注市场规模（具体数字）、
增长率（CAGR）、技术成熟度曲线和头部公司的最新动态。
你的输出以事实和数据为主，不做主观判断。""",
    tools=[search_tool],
    llm="gpt-4o",
    allow_delegation=False,
    verbose=True,
)

# 分析师 Agent
analyst = Agent(
    role="竞争分析师",
    goal="基于研究员提供的原始数据，提炼 {industry} 行业的竞争格局、关键洞察和投资价值",
    backstory="""你是一位严谨的商业分析师，专注于技术行业的竞争分析。
你擅长使用波特五力、SWOT 等分析框架，能从杂乱的数据中提取结构化洞察。
你特别关注：市场份额分布、技术壁垒、融资轮次和估值趋势。
你的输出需要有明确的结论和量化支撑。""",
    llm="gpt-4o",
    allow_delegation=False,
    verbose=True,
)

# 写作员 Agent
writer = Agent(
    role="行业报告撰写员",
    goal="将分析师的结构化分析转化为一份专业、可读性强的 {industry} 行业研究报告",
    backstory="""你是一位经验丰富的商业报告撰写专家，长期为投资机构和企业管理层撰写行业白皮书。
你的写作风格专业但不晦涩，善用数据和案例支撑观点。
报告结构清晰：行业概况 → 竞争分析 → 投资建议，每个部分都有小结。
你注重报告的实用价值——读完这份报告的人，应该能做出更明智的投资决策。""",
    llm="gpt-4o",
    allow_delegation=False,
    verbose=True,
)

# 太模糊——输出会很泛
backstory="你是一位行业研究员"

# 有细节——输出会更有针对性和专业感
backstory="你在麦肯锡有 10 年经验，擅长从海量信息中识别关键数据点，关注市场规模和 CAGR..."

from crewai_tools import SerperDevTool

search_tool = SerperDevTool(
    n_results=10,  # 每次搜索返回 10 条结果
)

export SERPER_API_KEY="your-serper-api-key"

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


class IndustryReportInput(BaseModel):
    """自定义工具的输入 schema"""
    industry: str = Field(description="行业名称，如'量子计算'、'新能源'")


class IndustryReportTool(BaseTool):
    name: str = "行业报告摘要工具"
    description: str = "根据行业名称，返回该行业的近期重要报告摘要和数据来源"
    args_schema: type[BaseModel] = IndustryReportInput

    def _run(self, industry: str) -> str:
        # 实际项目中这里会调用数据库或 API
        # 这里用模拟数据演示
        reports = {
            "量子计算": (
                "1. McKinsey 2025 量子技术报告：全球量子计算市场预计 2030 年达 280 亿美元\n"
                "2. BCG 量子优势分析：量子纠错技术进入实用化阶段\n"
                "3. PitchBook 数据：2025 年量子计算领域融资总额达 42 亿美元"
            ),
            "新能源": (
                "1. IEA 2025 全球能源展望：可再生能源装机量首次超过化石燃料\n"
                "2. Bloomberg NEF：2025 年清洁能源投资达 1.8 万亿美元\n"
                "3. 储能技术成本在过去 3 年下降 60%"
            ),
        }
        return reports.get(industry, f"暂未找到 {industry} 的行业报告数据")

report_tool = IndustryReportTool()

# 把它加到研究员的工具列表
researcher = Agent(
    # ... 其他参数同上
    tools=[search_tool, report_tool],  # 现在有两个工具
)

from crewai import Task

# 研究员的任务：搜集原始数据
research_task = Task(
    description="""
    针对 {industry} 行业进行全面的信息搜集。

    你需要搜集以下信息：
    1. 市场规模（当前值和预测值，带具体数字）
    2. 主要公司（至少 5 家，包括名称、核心业务、融资情况）
    3. 技术趋势（当前主流技术路线和未来方向）
    4. 行业关键事件（最近 12 个月的重要新闻）

    输出要求：以结构化的方式呈现原始数据，每条信息标注来源。
    """,
    expected_output="一份包含市场规模、主要公司、技术趋势和关键事件的结构化数据报告，至少 1500 字",
    agent=researcher,
)

# 分析师的任务：做竞争分析
analysis_task = Task(
    description="""
    基于研究员提供的 {industry} 行业原始数据，进行深度分析。

    你需要完成：
    1. 竞争格局分析：使用表格对比主要公司的市场份额、核心技术、融资阶段
    2. 行业关键洞察：提炼 3-5 个最重要的趋势或发现
    3. SWOT 分析：该行业整体的优势、劣势、机会、威胁
    4. 投资价值评估：给出明确的投资建议（看好/谨慎/观望），并说明理由

    输出要求：结论明确，每个观点都有数据支撑。
    """,
    expected_output="一份包含竞争分析表格、关键洞察列表、SWOT 分析和投资建议的结构化分析报告",
    agent=analyst,
    context=[research_task],  # 依赖研究员的输出
)

# 写作员的任务：撰写最终报告
writing_task = Task(
    description="""
    将分析师的 {industry} 行业分析报告转化为一份面向投资者的专业行业报告。

    报告结构：
    1. 执行摘要（200 字以内，概括核心结论）
    2. 行业概况（市场规模、增长驱动力、技术成熟度）
    3. 竞争分析（用表格呈现主要玩家对比）
    4. 关键趋势与洞察（3-5 个要点）
    5. 投资建议（明确的建议和风险提示）

    写作要求：
    - 语言专业但易读，避免过度使用行业黑话
    - 所有数据标注来源
    - 每个章节有小标题和简要小结
    - 报告总长度 2000-3000 字
    """,
    expected_output="一份完整的行业研究报告，包含执行摘要、行业概况、竞争分析表格、趋势洞察和投资建议，2000-3000 字",
    agent=writer,
    context=[analysis_task],  # 依赖分析师的输出
    output_file="industry_report.md",  # 自动保存为文件
)

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,  # 流水线模式：按 tasks 列表顺序依次执行
    verbose=True,
    memory=True,  # 开启记忆功能
    max_rpm=10,  # 每分钟最多 10 次 LLM 调用
)

result = crew.kickoff(inputs={"industry": "量子计算"})

print("=" * 50)
print("最终报告：")
print(result)

# Agent: 行业研究员
## Task: 针对 量子计算 行业进行全面的信息搜集...

# Agent: 行业研究员
## Thought: 我需要搜索量子计算行业的最新市场数据和技术进展
## Using tool: SerperDevTool
## Tool input: {"search_query": "量子计算 市场规模 2025 2026"}
## Tool output: [搜索结果列表...]

# Agent: 行业研究员
## Thought: 我已经搜集到足够的市场数据，现在搜索主要公司的信息
## Using tool: SerperDevTool
## Tool input: {"search_query": "量子计算 主要公司 融资 2025"}
## Tool output: [搜索结果列表...]

# ... 更多搜索步骤 ...

# Agent: 行业研究员
## Final Answer: [研究员的完整输出，约 1800 字]

---

# Agent: 竞争分析师
## Task: 基于研究员提供的 量子计算 行业原始数据...

# Agent: 竞争分析师
## Thought: 研究员提供了丰富的数据，我先整理竞争格局...

# Agent: 竞争分析师
## Final Answer: [分析师的完整输出，包含表格和 SWOT]

---

# Agent: 行业报告撰写员
## Task: 将分析师的 量子计算 行业分析报告转化...

# Agent: 行业报告撰写员
## Thought: 我需要把分析师的结构化数据转化为面向投资者的报告...

# Agent: 行业报告撰写员
## Final Answer: [写作员的完整报告，2000-3000 字]

研究员 → 分析师 → 写作员

        经理 Agent
       /    |    \
  研究员  分析师  写作员

researcher = Agent(
    # ... 其他参数同上
    max_iter=5,  # 最多 5 轮 Thought-Action-Observation 循环
)

crew = Crew(
    # ... 其他参数
    max_rpm=10,  # 每分钟最多 10 次 LLM 请求
)

crew = Crew(
    # ... 其他参数
    memory=True,           # 开启短期记忆（Agent 之间共享上下文）
)

def step_callback(step_output):
    """每一步执行后的回调"""
    # 记录到日志系统
    with open("crew_log.txt", "a", encoding="utf-8") as f:
        f.write(f"[{__import__('datetime').datetime.now()}] {step_output}\n")

def task_callback(task_output):
    """每个任务完成后的回调"""
    print(f"[任务完成] 耗时信息已记录")

crew = Crew(
    # ... 其他参数
    step_callback=step_callback,
    task_callback=task_callback,
)

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,
    max_rpm=10,
    step_callback=step_callback,
    task_callback=task_callback,
    share_crew=False,  # 不向 CrewAI 官方分享运行数据
)

# 确保显式设置 context
analysis_task = Task(
    # ...
    context=[research_task],  # 不要省略这行
)

researcher = Agent(
    # ...
    llm="gpt-4o",
    # 也可以在 CrewAI 的 LLM 配置中指定 temperature
)

from crewai import LLM

gpt4o = LLM(
    model="gpt-4o",
    temperature=0.3,  # 降低随机性，输出更稳定
)

researcher = Agent(
    # ...
    llm=gpt4o,
)

researcher = Agent(
    # ...
    max_iter=5,       # 最多 5 轮迭代
    max_retry_limit=2, # 失败最多重试 2 次
)

from crewai import LLM

gpt4o = LLM(model="gpt-4o", temperature=0.3)
gpt4o_mini = LLM(model="gpt-4o-mini", temperature=0.3)

# 研究员：需要深度搜索和分析，用大模型
researcher = Agent(
    # ...
    llm=gpt4o,
)

# 分析师：主要是结构化整理，用 mini 够了
analyst = Agent(
    # ...
    llm=gpt4o_mini,
)

# 写作员：需要高质量输出，用大模型
writer = Agent(
    # ...
    llm=gpt4o,
)

"""
CrewAI 多 Agent 行业研究系统
运行方式: python crewai_demo.py
环境变量: OPENAI_API_KEY, SERPER_API_KEY
"""
import datetime
from crewai import Agent, Task, Crew, Process, LLM
from crewai.tools import BaseTool
from crewai_tools import SerperDevTool
from pydantic import BaseModel, Field


# ============ 自定义工具 ============

class IndustryReportInput(BaseModel):
    industry: str = Field(description="行业名称")

class IndustryReportTool(BaseTool):
    name: str = "行业报告摘要工具"
    description: str = "根据行业名称，返回该行业的近期重要报告摘要"
    args_schema: type[BaseModel] = IndustryReportInput

    def _run(self, industry: str) -> str:
        reports = {
            "量子计算": (
                "1. McKinsey 2025: 全球量子计算市场 2030 年预计达 280 亿美元\n"
                "2. BCG: 量子纠错进入实用化阶段\n"
                "3. PitchBook: 2025 年量子计算融资总额 42 亿美元"
            ),
        }
        return reports.get(industry, f"暂未找到 {industry} 的行业报告数据")


# ============ 模型配置 ============

gpt4o = LLM(model="gpt-4o", temperature=0.3)
gpt4o_mini = LLM(model="gpt-4o-mini", temperature=0.3)

# ============ 工具 ============

search_tool = SerperDevTool(n_results=10)
report_tool = IndustryReportTool()


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

researcher = Agent(
    role="行业研究员",
    goal="针对 {industry} 行业，搜集最新的市场数据、技术进展、主要公司和行业趋势",
    backstory="""你是一位资深行业研究员，在麦肯锡和贝恩有超过 10 年的咨询经验。
你擅长从海量信息中快速识别关键数据点，尤其关注市场规模（具体数字）、
增长率（CAGR）、技术成熟度曲线和头部公司的最新动态。
你的输出以事实和数据为主，不做主观判断。""",
    tools=[search_tool, report_tool],
    llm=gpt4o,
    max_iter=5,
    allow_delegation=False,
    verbose=True,
)

analyst = Agent(
    role="竞争分析师",
    goal="基于研究员提供的原始数据，提炼 {industry} 行业的竞争格局、关键洞察和投资价值",
    backstory="""你是一位严谨的商业分析师，专注于技术行业的竞争分析。
你擅长使用波特五力、SWOT 等分析框架，能从杂乱的数据中提取结构化洞察。
你特别关注：市场份额分布、技术壁垒、融资轮次和估值趋势。
你的输出需要有明确的结论和量化支撑。""",
    llm=gpt4o_mini,
    max_iter=3,
    allow_delegation=False,
    verbose=True,
)

writer = Agent(
    role="行业报告撰写员",
    goal="将分析师的结构化分析转化为一份专业、可读性强的 {industry} 行业研究报告",
    backstory="""你是一位经验丰富的商业报告撰写专家，长期为投资机构和企业管理层撰写行业白皮书。
你的写作风格专业但不晦涩，善用数据和案例支撑观点。
报告结构清晰：行业概况 → 竞争分析 → 投资建议，每个部分都有小结。
你注重报告的实用价值——读完这份报告的人，应该能做出更明智的投资决策。""",
    llm=gpt4o,
    max_iter=3,
    allow_delegation=False,
    verbose=True,
)


# ============ 任务定义 ============

research_task = Task(
    description="""
    针对 {industry} 行业进行全面的信息搜集。
    你需要搜集以下信息：
    1. 市场规模（当前值和预测值，带具体数字）
    2. 主要公司（至少 5 家，包括名称、核心业务、融资情况）
    3. 技术趋势（当前主流技术路线和未来方向）
    4. 行业关键事件（最近 12 个月的重要新闻）
    输出要求：以结构化的方式呈现原始数据，每条信息标注来源。
    """,
    expected_output="一份包含市场规模、主要公司、技术趋势和关键事件的结构化数据报告，至少 1500 字",
    agent=researcher,
)

analysis_task = Task(
    description="""
    基于研究员提供的 {industry} 行业原始数据，进行深度分析。
    你需要完成：
    1. 竞争格局分析：表格对比主要公司的市场份额、核心技术、融资阶段
    2. 行业关键洞察：提炼 3-5 个最重要的趋势或发现
    3. SWOT 分析
    4. 投资价值评估：给出明确的投资建议并说明理由
    """,
    expected_output="一份包含竞争分析表格、关键洞察列表、SWOT 分析和投资建议的结构化分析报告",
    agent=analyst,
    context=[research_task],
)

writing_task = Task(
    description="""
    将分析师的 {industry} 行业分析报告转化为一份面向投资者的专业行业报告。
    报告结构：
    1. 执行摘要（200 字以内）
    2. 行业概况
    3. 竞争分析（用表格呈现）
    4. 关键趋势与洞察
    5. 投资建议和风险提示
    """,
    expected_output="一份完整的行业研究报告，2000-3000 字",
    agent=writer,
    context=[analysis_task],
    output_file="industry_report.md",
)


# ============ 回调函数 ============

def step_callback(step_output):
    with open("crew_log.txt", "a", encoding="utf-8") as f:
        f.write(f"[{datetime.datetime.now():%H:%M:%S}] {step_output}\n")


# ============ 组建并运行 Crew ============

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

if __name__ == "__main__":
    result = crew.kickoff(inputs={"industry": "量子计算"})
    print("\n" + "=" * 50)
    print(result)