多 Agent 协作：CrewAI 框架

为什么需要多 Agent

单个 Agent 面对复杂任务时存在瓶颈：上下文窗口被单一 LLM 实例的历史占满、单一角色无法同时具备所有专业知识、复杂任务难以在单线程中可靠完成。多 Agent 系统的核心思想是：分而治之，专业分工。

核心概念

CrewAI 协作流程图

构建研究团队：完整示例

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool
from langchain_openai import ChatOpenAI

# ── 定义共用的 LLM ────────────────────────────────────────
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
search_tool = SerperDevTool()

# ── 定义 Agents ───────────────────────────────────────────
researcher = Agent(
    role="高级市场研究员",
    goal="从可靠来源收集准确、全面的市场数据和技术信息",
    backstory="""你是一位拥有10年经验的市场研究专家，擅长识别趋势、
    验证数据来源、综合多维度信息。你注重数据准确性，
    会主动标注信息的可靠程度。""",
    tools=[search_tool, WebsiteSearchTool()],
    llm=llm,
    verbose=True,
    allow_delegation=False,  # 不允许将子任务委派给其他 Agent
    max_iter=5,              # 最大迭代次数
    memory=True              # 启用跨任务记忆
)

analyst = Agent(
    role="数据分析师",
    goal="将原始数据转化为可操作的商业洞见，量化风险和机会",
    backstory="""你是量化分析领域的专家，善用统计方法和数据可视化。
    你的分析总是基于证据，并清晰地标注假设和局限性。""",
    tools=[],  # 分析师不需要外部工具，依赖研究员提供的数据
    llm=llm,
    verbose=True
)

writer = Agent(
    role="技术写作专家",
    goal="将技术分析转化为清晰易懂、结构完整的专业报告",
    backstory="""你擅长用简洁的语言传达复杂概念，
    你的报告结构清晰，有执行摘要、详细分析和行动建议。""",
    tools=[],
    llm=llm,
    verbose=True
)

# ── 定义 Tasks ────────────────────────────────────────────
research_task = Task(
    description="""研究 {topic} 的最新发展现状：
    1. 市场规模和增长率（2024-2025数据）
    2. 主要参与者和技术方案
    3. 关键挑战和机遇
    请引用具体数据来源。""",
    expected_output="""包含以下部分的Markdown报告：
    - 市场概览（含数据来源）
    - 主要竞争者列表（含市场份额）
    - 技术成熟度评估
    - 数据来源列表""",
    agent=researcher,
    output_file="research_raw.md"  # 自动保存到文件
)

analysis_task = Task(
    description="""基于研究员提供的数据，深入分析 {topic}：
    1. SWOT 分析
    2. 竞争态势评分（1-10分，含理由）
    3. 未来12个月的发展预测
    4. 主要风险因素（按影响程度排序）""",
    expected_output="""结构化的分析报告，包含：
    - SWOT 矩阵
    - 竞争评分表
    - 预测结论（含置信度）
    - 风险清单""",
    agent=analyst,
    context=[research_task]  # 依赖研究任务的输出
)

report_task = Task(
    description="""整合研究和分析结果，创作完整的行业报告。
    目标读者：企业决策者（非技术背景）。
    报告必须：专业但易读、有清晰结论、提供具体行动建议。""",
    expected_output="""完整的Markdown格式报告，包含：
    - 执行摘要（200字）
    - 市场分析
    - 竞争格局
    - 风险评估
    - 战略建议（3-5条）
    - 附录：数据来源""",
    agent=writer,
    context=[research_task, analysis_task],
    output_file="final_report.md"
)

# ── 组建 Crew ─────────────────────────────────────────────
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.sequential,  # 串行：研究→分析→写作
    verbose=2,                  # 详细日志
    memory=True,                # 跨任务记忆
    max_rpm=10                  # 限速：每分钟最多10次API调用
)

# ── 启动执行 ──────────────────────────────────────────────
result = crew.kickoff(inputs={"topic": "2025年 AI Agent 框架市场"})
print(result.raw)         # 最终输出
print(result.usage_metrics) # Token 使用统计

人机协作（Human-in-the-Loop）

CrewAI 支持在任务执行过程中暂停，等待人工输入或确认：

# 方案1：Task 级别的人工审核
approval_task = Task(
    description="""草拟一封给客户的道歉邮件，解释服务中断原因。
    邮件需要：专业、诚恳、提供补偿方案。""",
    expected_output="完整的电子邮件正文",
    agent=writer,
    human_input=True  # 执行后暂停，等待人工确认/修改
)

# 方案2：工具级别的人工确认（适合高风险操作）
from crewai.tools import BaseTool

class SendEmailTool(BaseTool):
    name: str = "send_email"
    description: str = "发送电子邮件（需要人工确认）"

    def _run(self, to: str, subject: str, body: str) -> str:
        print(f"\n⚠️  即将发送邮件：")
        print(f"收件人: {to}")
        print(f"主题: {subject}")
        print(f"正文:\n{body}")

        confirm = input("\n确认发送？(yes/no): ")
        if confirm.lower() == "yes":
            # 实际发送逻辑
            return f"邮件已发送到 {to}"
        return "用户取消发送，请告知用户需要修改哪里"

层级流程与 Manager Agent

# 层级流程：由 Manager 自动分配任务
crew_hierarchical = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.hierarchical,

    # Manager Agent（框架自动创建，也可自定义）
    manager_llm=ChatOpenAI(model="gpt-4o", temperature=0),

    # 或者使用自定义 Manager
    # manager_agent=Agent(role="项目经理", goal="高效协调团队完成目标", ...)

    verbose=2
)

# 支持并行任务（层级模式下）
research_us = Task(description="研究美国市场...", agent=researcher)
research_cn = Task(description="研究中国市场...", agent=researcher)
merge_task  = Task(
    description="整合美国和中国市场研究...",
    agent=analyst,
    context=[research_us, research_cn]
)

CrewAI vs LangGraph：选型对比