56. AI Agent 进阶——CrewAI 多 Agent 协作¶

单 Agent 模式：
  你 → 目标 → 一个 AI 做所有事 → 结果（容易出错、角色混乱）

多 Agent 协作模式：
  你 → 目标 → 管理者拆任务 → 搜索员找资料
                              → 分析师看数据
                              → 写手出报告
                            → 整合结果 → 高质量输出

宏基因组分析 Crew（团队）：

Agent 1 "文献调研员"：搜索 PubMed，总结 T2D 肠道菌群研究进展
Agent 2 "分析方案师"：根据文献设计分析 pipeline（工具选型、参数推荐）
Agent 3 "代码工程师"：按方案写出可执行的分析脚本

任务流：文献调研 → 方案设计（依赖文献结果） → 代码编写（依赖方案）

                    ┌─────────────────────────────────┐
                    │           Crew（团队）            │
                    │  ┌─────────┐  ┌─────────┐       │
                    │  │ Agent 1 │  │ Agent 2 │  ...  │
                    │  │ (角色)   │  │ (角色)   │       │
                    │  │ + Tools │  │ + Tools │       │
                    │  └─────────┘  └─────────┘       │
                    │                                   │
                    │  Process（执行流程）               │
                    │  sequential / hierarchical        │
                    │                                   │
                    │  Memory（记忆系统）                │
                    │  短期 / 长期 / 实体记忆            │
                    │                                   │
                    │  Task 1 → Task 2 → Task 3        │
                    │  (任务链)                          │
                    └─────────────────────────────────┘

from crewai import Agent

# 定义一个"文献调研员"Agent
researcher = Agent(
    role="生物信息学文献调研员",           # 岗位：告诉 LLM 它是什么角色
    goal="找到与课题最相关的 5 篇高质量论文",  # 目标：Agent 的核心任务方向
    backstory="你有 10 年生信研究经验，擅长在 PubMed 检索并筛选高影响因子论文。",  # 背景：帮 LLM 进入角色
    verbose=True,            # 打印详细执行过程（调试时开启）
    allow_delegation=False,  # 是否允许把任务委托给其他 Agent
    max_iter=20,             # 最大迭代次数（防止死循环，默认 20）
    llm="gpt-4o-mini",      # 使用的 LLM 模型
)

from crewai import Task

# 定义一个文献搜索任务
search_task = Task(
    description="""
    搜索 2023-2025 年关于'肠道菌群与 2 型糖尿病'的最新 Meta-analysis 和队列研究。
    重点关注：
    1. 哪些菌属与 T2D 显著相关
    2. 分析方法（16S vs 宏基因组）
    3. 样本量和研究质量
    """,                                       # 任务描述：越详细越好
    expected_output="包含 5 篇论文摘要的结构化列表，每篇含标题、期刊、年份、关键发现",  # 期望输出
    agent=researcher,                          # 分配给哪个 Agent
    output_file="literature_review.md",        # 输出保存到文件
)

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, analyst, coder],   # 团队成员列表
    tasks=[search_task, design_task, code_task],  # 任务列表（按顺序排）
    process=Process.sequential,  # 执行模式：顺序执行
    verbose=True,                # 打印详细日志
    memory=True,                 # 开启记忆（Agent 之间共享上下文）
)

# 启动团队！
result = crew.kickoff(inputs={"topic": "T2D gut microbiome"})
print(result)

from crewai.tools import tool

@tool("PubMed搜索")  # 工具名称
def search_pubmed(query: str) -> str:
    """搜索 PubMed 数据库获取论文信息。

    Args:
        query: 搜索关键词，如 'gut microbiome T2D metagenomics'
    """
    # 这里实际调用 PubMed API（Biopython 的 Entrez）
    from Bio import Entrez
    Entrez.email = "your@email.com"
    handle = Entrez.esearch(db="pubmed", term=query, retmax=5)
    results = Entrez.read(handle)
    return f"找到 {results['Count']} 篇论文，Top IDs: {results['IdList']}"

# 顺序执行（最常用，适合线性流程）
crew = Crew(
    agents=[a1, a2, a3],
    tasks=[t1, t2, t3],
    process=Process.sequential,  # t1 做完 → t2 做完 → t3
)

# 层级执行（需要一个"经理"LLM）
crew = Crew(
    agents=[a1, a2, a3],
    tasks=[t1, t2, t3],
    process=Process.hierarchical,    # 经理 Agent 动态分配任务
    manager_llm="gpt-4o",           # 经理用的 LLM（必须指定）
)

from crewai import Crew, Memory

# 方式 1：简单开启（用默认配置）
crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,  # 一行开启记忆
)

# 方式 2：自定义记忆配置（调整权重和衰减速度）
memory = Memory(
    recency_weight=0.3,        # 时间越近的记忆权重越高
    semantic_weight=0.5,        # 语义相似度权重
    importance_weight=0.2,      # 重要性权重
    recency_half_life_days=30,  # 记忆"半衰期"（30天后权重减半）
)
crew = Crew(agents=[...], tasks=[...], memory=memory)

你的需求是什么？

├── 快速搭建角色分工明确的多 Agent 流程
│   └── → 选 CrewAI（开箱即用，学习成本最低）
│
├── 需要复杂的分支/循环/人工审批节点
│   └── → 选 LangGraph（最灵活，但代码量大）
│
├── 需要多个 Agent 互相讨论得出结论
│   └── → 选 AutoGen（对话式协作的首选）
│
└── 面试时怎么说？
    └── → 三个都了解，项目中用 CrewAI 快速搭建原型
         需要复杂流程控制时用 LangGraph

# 创建虚拟环境（推荐）
conda create -n crewai_env python=3.12 -y
conda activate crewai_env

# 安装 CrewAI（v1.14.4，最新版）
pip install crewai            # 核心包
pip install 'crewai[tools]'   # 包含常用工具（crewai-tools）

# 验证安装
python -c "import crewai; print(crewai.__version__)"
# 输出: 1.14.4（或更新版本）

# 配置 API Key（二选一）
# 方式1: 环境变量
export OPENAI_API_KEY="sk-your-key-here"

# 方式2: .env 文件
echo 'OPENAI_API_KEY=sk-your-key-here' > .env

"""
文件: crew_minimal.py
功能: CrewAI 最小可运行示例——两个 Agent 协作写报告
说明: 04 入门篇只展示了基础 Agent 定义，这里展示完整的任务依赖和流程控制
"""

from crewai import Agent, Task, Crew, Process  # 导入四个核心类

# ============ 第 1 步：定义 Agent（定义"谁来干活"） ============

# 研究员 Agent：负责搜集信息
researcher = Agent(
    role="Senior Researcher",                    # 角色
    goal="研究并总结给定主题的最新进展",              # 目标
    backstory="你是一位资深研究员，擅长快速定位并总结学术领域的最新进展。",
    verbose=True,                                # 打印执行过程
)

# 报告员 Agent：负责写报告
reporter = Agent(
    role="Report Writer",
    goal="将研究发现撰写为结构清晰的中文报告",
    backstory="你是一位经验丰富的科技记者，擅长将复杂研究转化为通俗易懂的报告。",
    verbose=True,
)

# ============ 第 2 步：定义 Task（定义"干什么活"） ============

# 任务 1：研究任务
research_task = Task(
    description="调研 {topic} 的最新进展（2024-2025 年），总结 3 个关键发现。",
    expected_output="包含 3 个关键发现的列表，每个发现包含来源和简要说明",
    agent=researcher,  # 分配给研究员
)

# 任务 2：报告任务（依赖任务 1 的输出）
report_task = Task(
    description="基于研究发现，撰写一份 300 字的中文报告。",
    expected_output="一份结构为'背景→发现→结论'的中文报告",
    agent=reporter,          # 分配给报告员
    context=[research_task], # 关键！声明依赖：报告员能看到研究员的输出
    output_file="report.md", # 自动保存到文件
)

# ============ 第 3 步：组建 Crew 并运行 ============

crew = Crew(
    agents=[researcher, reporter],         # 团队成员
    tasks=[research_task, report_task],    # 任务列表
    process=Process.sequential,           # 顺序执行：研究 → 报告
    verbose=True,                         # 打印执行日志
)

# 启动！{topic} 会替换 Task description 中的占位符
result = crew.kickoff(inputs={"topic": "AI 在宏基因组分析中的应用"})

print("\n" + "=" * 60)
print("最终输出：")
print("=" * 60)
print(result.raw)                    # 获取原始文本输出
print(f"\nToken 用量: {result.token_usage}")  # 查看 Token 消耗

my_bioinfo_crew/
├── src/
│   └── my_bioinfo_crew/
│       ├── config/
│       │   ├── agents.yaml    # Agent 角色定义
│       │   └── tasks.yaml     # 任务定义
│       ├── crew.py            # Crew 组装逻辑
│       └── main.py            # 入口
├── pyproject.toml
└── .env                       # API Key

# Agent 角色定义（白话：员工花名册）
literature_researcher:
  role: >
    {topic} 领域文献调研专家
  goal: >
    找到该领域最新、最权威的研究论文并提取关键发现
  backstory: >
    你是一位有 10 年经验的生物信息学研究员，发表过 50+ 篇 SCI 论文，
    擅长在 PubMed 上高效检索和筛选高质量文献。

pipeline_designer:
  role: >
    生信分析 Pipeline 设计师
  goal: >
    根据文献信息设计一套完整、可执行的分析流程
  backstory: >
    你精通宏基因组分析流程，熟悉 MetaPhlAn4、HUMAnN3、Kraken2 等工具，
    能根据数据特点选择最合适的分析策略。

# 任务定义（白话：工作单）
literature_search:
  description: >
    搜索 2023-2025 年关于 {topic} 的最新研究。
    重点关注 Meta-analysis 和大规模队列研究。
    找出 5 篇核心论文，提取关键发现。
  expected_output: >
    一份结构化文献列表，每篇包含：标题、期刊、年份、样本量、关键发现。
  agent: literature_researcher

pipeline_design:
  description: >
    基于文献调研结果，为 {topic} 设计一个完整的分析 pipeline。
    需要包含：数据质控、物种注释、功能分析、统计检验每一步的工具和参数。
  expected_output: >
    一份分析流程文档，格式为：步骤 → 工具 → 命令 → 参数 → 预期输出。
  agent: pipeline_designer
  context:
    - literature_search

# CrewAI 提供了 CLI 工具快速生成项目骨架
crewai create crew my_bioinfo_crew

# 生成的目录结构就是上面推荐的 YAML 结构
# 直接修改 agents.yaml 和 tasks.yaml 即可

# 运行项目
cd my_bioinfo_crew
crewai run

"""
文件: bioinfo_crew.py
功能: 宏基因组分析的三 Agent 协作团队
场景: 给定课题方向，自动完成文献调研→分析方案→代码编写

注意: 需要设置 OPENAI_API_KEY 环境变量
安装: pip install crewai 'crewai[tools]'
"""

import os
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool  # 自定义工具的装饰器

# ============================================================
# 第 1 部分：定义自定义工具
# （白话：给 Agent 配备专业"装备"）
# ============================================================

@tool("PubMed论文搜索")
def search_pubmed_tool(query: str) -> str:
    """搜索 PubMed 数据库获取相关论文。

    Args:
        query: 搜索关键词，如 'gut microbiome type 2 diabetes metagenomics'
    """
    # 生产环境应使用 Biopython 的 Entrez 模块
    # 这里用模拟数据演示完整流程
    mock_results = f"""
PubMed 搜索结果 (query: {query})：

1. "Multi-omics analysis reveals gut microbiome signatures in T2D"
   - 期刊: Nature Medicine, 2024
   - 样本量: 2,000 人（1000 T2D + 1000 对照）
   - 关键发现: Faecalibacterium 和 Roseburia 在 T2D 中显著降低

2. "Metagenomic profiling of diabetic gut reveals functional shifts"
   - 期刊: Cell Host & Microbe, 2024
   - 样本量: 800 人
   - 关键发现: 丁酸合成通路（butanoate metabolism）在 T2D 中减弱

3. "Machine learning on gut metagenome predicts T2D onset"
   - 期刊: Gut, 2025
   - 样本量: 5,000 人纵向队列
   - 关键发现: 基于 50 个菌种丰度的随机森林模型 AUC=0.89

4. "Probiotic intervention alters gut microbiome in T2D: RCT"
   - 期刊: Lancet Microbe, 2024
   - 样本量: 300 人 RCT
   - 关键发现: 12 周干预后 HbA1c 平均降低 0.4%

5. "Strain-level analysis of Akkermansia in metabolic syndrome"
   - 期刊: Science, 2024
   - 样本量: 1,500 人
   - 关键发现: A. muciniphila 特定菌株与胰岛素敏感性正相关
"""
    return mock_results


@tool("生信工具查询")
def query_bioinfo_tool(tool_name: str) -> str:
    """查询生信工具的最新版本、用法和推荐参数。

    Args:
        tool_name: 工具名称，如 'MetaPhlAn4', 'HUMAnN3', 'Kraken2'
    """
    tool_db = {
        "MetaPhlAn4": "MetaPhlAn 4.1 (2024)：基于 marker gene 的物种注释，数据库 mpa_vOct22_CHOCOPhlAnSGB_202403",
        "HUMAnN3": "HUMAnN 3.9 (2024)：功能通路分析，搭配 UniRef90 和 MetaCyc",
        "Kraken2": "Kraken2 v2.1.3 (2024)：k-mer 物种分类，搭配 PlusPF 数据库",
        "fastp": "fastp v1.3.3 (2025)：质控+过滤+去接头，推荐 -q 20 -l 50",
    }
    return tool_db.get(tool_name, f"未找到 {tool_name} 的信息，请检查工具名称")


# ============================================================
# 第 2 部分：定义三个 Agent
# ============================================================

# Agent 1：文献调研员
literature_agent = Agent(
    role="宏基因组文献调研专家",
    goal="找到与研究课题最相关的 5 篇高质量论文，提取关键发现和方法学信息",
    backstory="""你是一位有 10 年经验的生物信息学研究员，发表过 30+ 篇 SCI 论文。
你擅长在 PubMed 高效检索，能快速判断论文质量（看期刊影响因子、样本量、研究设计）。
你特别关注方法学细节——用了什么测序平台、什么分析工具、什么统计方法。""",
    tools=[search_pubmed_tool],  # 配备 PubMed 搜索工具
    verbose=True,
    max_iter=15,  # 最多迭代 15 次
)

# Agent 2：分析方案设计师
pipeline_agent = Agent(
    role="宏基因组分析 Pipeline 架构师",
    goal="根据文献信息设计一套科学合理、可执行的宏基因组分析流程",
    backstory="""你是一位资深生信工程师，精通宏基因组分析全流程。
你熟悉 MetaPhlAn4、HUMAnN3、Kraken2、LEfSe、MaAsLin2 等主流工具。
你设计的 pipeline 注重：可复现性、参数合理性、步骤完整性。
你会根据数据特点（测序类型、样本量、研究问题）选择最合适的工具组合。""",
    tools=[query_bioinfo_tool],  # 配备生信工具查询工具
    verbose=True,
)

# Agent 3：代码工程师
coder_agent = Agent(
    role="生信分析脚本工程师",
    goal="将分析方案转化为可直接运行的 Shell/Python 脚本",
    backstory="""你是一位全栈生信工程师，精通 Shell、Python 和 R。
你写的脚本特点：每行都有中文注释、有错误处理、有进度提示、有参数验证。
你遵循 best practice：使用 conda 环境管理、使用相对路径、脚本模块化。""",
    verbose=True,
)

# ============================================================
# 第 3 部分：定义三个任务（含依赖关系）
# ============================================================

# 任务 1：文献调研（无依赖，最先执行）
task_literature = Task(
    description="""
    搜索 2023-2025 年关于'{topic}'的最新研究论文。

    要求：
    1. 找到 5 篇高质量论文（优先 Nature/Science/Cell 子刊）
    2. 每篇提取：标题、期刊、年份、样本量、核心发现、使用的分析方法
    3. 总结该领域的共识发现和尚存争议
    4. 特别关注使用了宏基因组测序（而非 16S）的研究
    """,
    expected_output="""一份结构化文献综述，包含：
    - 5 篇论文的详细信息
    - 领域共识（至少 3 点）
    - 争议/未解决问题（至少 2 点）
    - 推荐的分析方法""",
    agent=literature_agent,
)

# 任务 2：分析方案设计（依赖任务 1 的文献调研结果）
task_pipeline = Task(
    description="""
    基于文献调研的结果，为'{topic}'的宏基因组分析设计一套完整的 pipeline。

    要求：
    1. 覆盖从原始数据到最终结果的完整流程
    2. 每一步列出：步骤名 → 推荐工具 → 关键参数 → 输入/输出
    3. 包含质控、物种注释、功能分析、差异分析、可视化 5 大模块
    4. 参考文献中使用的分析方法，但可以根据最新工具版本调整
    """,
    expected_output="""一份分析 pipeline 文档，格式为：
    步骤编号 | 模块 | 工具 | 版本 | 命令模板 | 关键参数 | 输入 | 输出""",
    agent=pipeline_agent,
    context=[task_literature],  # 关键：依赖文献调研的输出
)

# 任务 3：代码编写（依赖任务 2 的分析方案）
task_code = Task(
    description="""
    根据分析 pipeline 方案，编写一套可直接运行的分析脚本。

    要求：
    1. 用 Shell 脚本实现主流程（质控→比对→注释）
    2. 用 Python 脚本实现下游统计分析和可视化
    3. 每行代码都加中文注释
    4. 包含错误处理（set -euo pipefail）
    5. 使用 conda 环境管理依赖
    6. 脚本开头有使用说明
    """,
    expected_output="完整的分析脚本代码（Shell + Python），可直接复制运行",
    agent=coder_agent,
    context=[task_pipeline],  # 关键：依赖分析方案的输出
    output_file="bioinfo_pipeline_scripts.md",  # 自动保存
)

# ============================================================
# 第 4 部分：组建 Crew 并运行
# ============================================================

bioinfo_crew = Crew(
    agents=[literature_agent, pipeline_agent, coder_agent],
    tasks=[task_literature, task_pipeline, task_code],
    process=Process.sequential,  # 顺序执行：文献→方案→代码
    verbose=True,
    memory=True,  # 开启记忆，Agent 间共享上下文
)

if __name__ == "__main__":
    # 启动 Crew！inputs 中的变量会替换任务描述中的 {topic}
    result = bioinfo_crew.kickoff(
        inputs={"topic": "肠道菌群与2型糖尿病的宏基因组分析"}
    )

    print("\n" + "=" * 60)
    print("三 Agent 协作完成！最终输出：")
    print("=" * 60)
    print(result.raw)

    # 查看每个任务的输出
    for i, task_output in enumerate(result.tasks_output):
        print(f"\n--- 任务 {i+1} 输出摘要 ---")
        print(task_output.summary)

    # 查看 Token 消耗
    print(f"\n总 Token 用量: {result.token_usage}")

执行流程：

[Agent 1: 文献调研专家] 正在执行...
  → 调用 PubMed搜索 工具
  → 搜索到 5 篇相关论文
  → 总结领域共识和争议
  → 输出文献综述

[Agent 2: Pipeline 架构师] 正在执行...
  → 接收到文献调研结果（自动通过 context 传递）
  → 调用 生信工具查询 工具
  → 设计 5 大模块分析流程
  → 输出 pipeline 文档

[Agent 3: 脚本工程师] 正在执行...
  → 接收到 pipeline 方案
  → 编写 Shell 质控脚本
  → 编写 Python 统计分析脚本
  → 输出可运行代码
  → 自动保存到 bioinfo_pipeline_scripts.md

from crewai import Task, TaskOutput

def validate_has_references(output: TaskOutput):
    """验证输出是否包含文献引用"""
    if "参考文献" not in output.raw and "Reference" not in output.raw:
        return (False, "输出缺少参考文献引用，请补充至少 3 篇文献")  # 不通过，返回反馈
    return (True, output.raw)  # 通过

task = Task(
    description="写一篇文献综述",
    expected_output="包含参考文献的综述",
    agent=researcher,
    guardrail=validate_has_references,    # 添加保护栏
    guardrail_max_retries=3,              # 最多重试 3 次
)

from pydantic import BaseModel
from typing import List

class PaperInfo(BaseModel):
    """论文信息的结构化模型"""
    title: str           # 标题
    journal: str         # 期刊
    year: int            # 年份
    key_finding: str     # 关键发现
    sample_size: int     # 样本量

class LiteratureReview(BaseModel):
    """文献综述的结构化模型"""
    papers: List[PaperInfo]    # 论文列表
    consensus: List[str]       # 共识
    controversies: List[str]   # 争议

task = Task(
    description="搜索并总结 5 篇论文",
    expected_output="结构化的文献综述",
    agent=researcher,
    output_pydantic=LiteratureReview,  # 输出必须符合这个模型
)

crew = Crew(
    agents=[...],
    tasks=[...],
    checkpoint=True,  # 开启断点保存（保存在 .checkpoints/ 目录）
)

# 从断点恢复
crew = Crew.from_checkpoint(".checkpoints/latest.json")
crew.kickoff()

# ===== Agent 速查 =====
Agent(
    role="角色名",              # 必填
    goal="目标",                # 必填
    backstory="背景",           # 必填
    llm="gpt-4o-mini",         # 可选，默认 gpt-4
    tools=[tool1, tool2],      # 可选，工具列表
    verbose=True,              # 可选，打印日志
    allow_delegation=False,    # 可选，是否可委托
    memory=True,               # 可选，是否启用记忆
    max_iter=20,               # 可选，最大迭代次数
    reasoning=True,            # 可选，是否启用推理模式
)

# ===== Task 速查 =====
Task(
    description="任务描述",           # 必填
    expected_output="期望输出",       # 必填
    agent=my_agent,                  # 可选（hierarchical 模式可不指定）
    context=[other_task],            # 可选，依赖的前置任务
    tools=[tool1],                   # 可选，任务级工具（覆盖 Agent 工具）
    output_file="result.md",         # 可选，输出保存路径
    output_pydantic=MyModel,         # 可选，结构化输出
    guardrail=my_validator,          # 可选，输出验证函数
    guardrail_max_retries=3,         # 可选，验证失败重试次数
    async_execution=False,           # 可选，是否异步执行
)

# ===== Crew 速查 =====
Crew(
    agents=[a1, a2, a3],             # 必填，Agent 列表
    tasks=[t1, t2, t3],             # 必填，Task 列表
    process=Process.sequential,      # 可选，sequential|hierarchical
    memory=True,                     # 可选，开启记忆
    verbose=True,                    # 可选，打印日志
    manager_llm="gpt-4o",           # hierarchical 模式必填
    checkpoint=True,                 # 可选，断点保存
)

# ===== 运行 =====
result = crew.kickoff(inputs={"key": "value"})  # 启动
print(result.raw)                # 原始输出
print(result.token_usage)       # Token 消耗
print(result.tasks_output)      # 各任务输出

# 安装
pip install crewai 'crewai[tools]'

# CLI 创建项目
crewai create crew my_project

# 运行项目
crewai run

# 查看任务日志
crewai log-tasks-outputs

# 从断点恢复
crewai replay -t <task_id>

# 重置记忆
crewai reset-memories