系列导读：本系列共 7 篇，带你从零到一构建完整的 RAG + 多智能体项目。

• 第 1 篇：最小 RAG 实现，纯 numpy，无任何 AI 框架
• 第 2 篇：接入 Ollama 本地大模型，实现真实语义检索
• 第 3 篇：接入 ChromaDB 持久化向量数据库
• 第 4 篇：用 LangChain 重构 + 多轮对话
• 第 5 篇：LangGraph 多步推理工作流
• 第 6 篇：MCP 工具调用协议集成
• 第 7 篇（本文）：多智能体 RAG（CrewAI 核心思想）

---

一、为什么需要多智能体

前 6 篇是「单智能体」模式：一个 LLM 完成所有工作。

但复杂问题需要分工：

用户：我要出差3天，请假和差旅报销同时申请，流程是什么？

单智能体：一个 LLM 又检索又分析又回答，角色混乱，质量不稳定

多智能体：
  - 检索专家：专注找相关文档
  - 分析师：专注提炼关键信息
  - 回答专家：专注生成友好回答
  每个 Agent 专注一件事，整体质量更高

这就是 CrewAI 的核心思想：像组建公司团队一样组建 AI 团队。

---

二、CrewAI 是什么

CrewAI 是 2024 年增长最快的多智能体框架，核心概念只有 4 个：

Agent（智能体）= 角色 + 目标 + 工具
Task（任务）   = 做什么 + 谁来做 + 期望输出
Crew（团队）   = Agent 列表 + Task 列表 + 执行模式
Process（模式）= sequential（顺序）或 hierarchical（层级）

真实的 CrewAI 安装：

pip install crewai   # 要求 Python ≥ 3.10

本文手写等价实现（Python 3.8 兼容），帮你理解底层原理。

---

三、核心类：Agent

class Agent:
    """
    智能体：有角色和目标的 AI 个体

    对应 CrewAI 的：
    Agent(role="...", goal="...", backstory="...", tools=[...])
    """
    def __init__(self, role: str, goal: str, tools: list = None):
        self.role  = role
        self.goal  = goal
        self.tools = tools or []

    def execute(self, task_description: str, context: str = "") -> str:
        """
        执行任务：用角色设定约束 LLM 的行为
        """
        system_prompt = f"""你是{self.role}。
你的核心目标：{self.goal}

请严格按照你的角色和目标完成以下任务。"""

        context_section = f"\n\n【前序工作成果】\n{context}" if context else ""

        full_prompt = f"""{system_prompt}{context_section}

【当前任务】
{task_description}

【你的回答】"""

        return ollama_chat(full_prompt)

关键设计：role 和 goal 注入 System Prompt，让同一个 LLM 表现出不同的"身份"和"专注点"。

---

四、核心类：Task

class Task:
    """
    任务：分配给某个 Agent 完成的具体工作

    对应 CrewAI 的：
    Task(description="...", agent=..., expected_output="...")
    """
    def __init__(self, description: str, agent: Agent, expected_output: str = ""):
        self.description     = description
        self.agent           = agent
        self.expected_output = expected_output
        self.output          = None

    def run(self, context: str = "") -> str:
        """执行任务，context 是上一个 Task 的输出"""
        result = self.agent.execute(
            task_description=self.description,
            context=context,
        )
        self.output = result
        return result

关键设计：Task 的 context 参数接受前一个 Task 的输出，实现信息接力。

---

五、核心类：Crew（两种模式）

Sequential（顺序模式）

def _run_sequential(self) -> str:
    """
    顺序模式：
      Task1 → 输出A
      Task2（收到A）→ 输出B
      Task3（收到B）→ 输出C（最终返回）
    """
    last_output = ""
    for task in self.tasks:
        last_output = task.run(context=last_output)
    return last_output

流程图：

用户问题
   ↓
[检索专家] → 找到相关文档
   ↓（把文档传给下一个）
[分析师] → 提炼关键要点
   ↓（把要点传给下一个）
[回答专家] → 生成最终回答
   ↓
用户看到回答

Hierarchical（层级模式）

def _run_hierarchical(self, inputs: dict) -> str:
    """
    层级模式：
      1. Manager 分析问题，制定计划（调用哪些 Worker）
      2. Worker 各自执行
      3. Manager 综合结果，生成最终回答
    """
    manager = self.agents[0]
    workers = {agent.role: agent for agent in self.agents[1:]}

    # Manager 制定计划
    plan = manager.execute(f"分析问题，决定调用哪些专家：{inputs['question']}")

    # Worker 执行
    results = [worker.execute(task) for worker in workers.values()]

    # Manager 综合
    return manager.execute(f"综合以下成果，给出最终回答：{results}")

流程图：

用户问题
   ↓
[Manager: HR总监] ←── 制定计划
   ↓ 分配任务
┌──────────────────────────────┐
│  [考勤专家]  [财务专家]  [政策专家]  │ ← 并行/顺序执行
└──────────────────────────────┘
   ↓ 上交成果
[Manager: HR总监] ──→ 综合最终回答

---

六、搭建企业知识库的 Crew

顺序模式（最常用）

def build_rag_crew() -> Crew:
    # ── 3 个 Agent ──
    retrieval_agent = Agent(
        role="企业知识库检索专家",
        goal="精准找到与用户问题最相关的知识库内容，不遗漏关键信息",
    )
    analysis_agent = Agent(
        role="信息分析师",
        goal="深度分析检索结果，提炼出回答用户问题所需的核心要点",
    )
    answer_agent = Agent(
        role="用户服务专家",
        goal="基于分析结果，用友好、清晰、专业的语言回答用户问题",
    )

    # ── 3 个 Task（顺序执行）──
    task_retrieve = Task(
        description="用户问题：{question}\n\n知识库搜索结果：\n" + search_results,
        agent=retrieval_agent,
        expected_output="检索到的相关知识库内容",
    )
    task_analyze = Task(
        description="根据检索内容，提炼3-5个关键要点",
        agent=analysis_agent,
        expected_output="关键要点列表",
    )
    task_answer = Task(
        description="用户问题：{question}\n\n基于分析成果，生成友好专业的回答",
        agent=answer_agent,
        expected_output="最终用户回答",
    )

    return Crew(
        agents=[retrieval_agent, analysis_agent, answer_agent],
        tasks=[task_retrieve, task_analyze, task_answer],
        process="sequential",
    )

层级模式（复杂问题）

def build_hierarchical_crew() -> Crew:
    manager = Agent(role="HR 总监", goal="综合各专家意见，给出权威完整的解答")
    attendance_expert = Agent(role="考勤专家", goal="专精于请假、年假、加班政策")
    finance_expert    = Agent(role="财务专家", goal="专精于报销、差旅政策")
    policy_expert     = Agent(role="政策解读专家", goal="将复杂制度转化为易懂指南")

    return Crew(
        agents=[manager, attendance_expert, finance_expert, policy_expert],
        tasks=[],      # 层级模式 tasks 由 Manager 动态创建
        process="hierarchical",
    )

---

七、运行效果

Sequential 模式：

🚀 Crew 启动 | 模式：sequential | 任务数：3

  [步骤 1/3]
  >> 执行任务 [企业知识库检索专家]
     任务：用户问题：我想请假，需要提前几天申请？...
  ✅ 完成：根据检索结果，找到了关于请假流程的相关文档...

  [步骤 2/3]
  >> 执行任务 [信息分析师]
     任务：根据检索到的知识库内容，分析以下几点...
  ✅ 完成：关键要点：1. 需提前3个工作日提交...

  [步骤 3/3]
  >> 执行任务 [用户服务专家]
     任务：用户原始问题：我想请假...
  ✅ 完成：根据公司规定，请假申请需要...

  🤖 最终回答：
  您好！根据公司HR手册，请假申请需要提前3个工作日通过OA系统提交...

---

八、与 LangGraph（Step 5）的对比

维度	LangGraph（Step 5）	Multi-Agent CrewAI（Step 7）
核心抽象	状态图（State + Node + Edge）	角色团队（Agent + Task + Crew）
设计思路	工作流驱动，显式定义流转	角色驱动，Agent 自主决策
适合场景	流程确定、步骤清晰的任务	需要专业分工、角色协作的任务
可控程度	高（每条路径都显式定义）	中（依赖 LLM 角色扮演）
代码复杂度	较高（需定义 State + 所有边）	较低（描述角色和任务即可）
典型用例	RAG 流程控制、质量评估重试	研究助手、内容创作、多专家咨询

选择建议：

• 流程可以画成流程图？→ 用 LangGraph
• 问题需要不同专家协作？→ 用 CrewAI/Multi-Agent

---

九、真实 CrewAI 的对比

真实 CrewAI（Python ≥ 3.10）写法：

from crewai import Agent, Task, Crew, Process

# Agent 定义几乎一样
retrieval_agent = Agent(
    role="企业知识库检索专家",
    goal="精准找到与用户问题最相关的知识库内容",
    backstory="你有10年的文档检索经验",  # backstory 是额外的背景设定
    tools=[search_tool],
    llm=llm,
)

# Task 定义几乎一样
task = Task(
    description="搜索关于 {question} 的相关文档",
    agent=retrieval_agent,
    expected_output="检索结果列表",
)

# Crew 定义几乎一样
crew = Crew(
    agents=[retrieval_agent, analysis_agent, answer_agent],
    tasks=[task_retrieve, task_analyze, task_answer],
    process=Process.sequential,  # 枚举类
    verbose=True,
)

result = crew.kickoff(inputs={"question": "我想请假怎么申请？"})

与本文手写版的差异：

• 多了 backstory 参数（补充背景故事，提升角色扮演质量）
• tools 参数直接集成工具调用（类似 Step 6 的 MCP）
• 支持真正的并行执行、内存共享、回调等高级功能

本文手写版帮你理解核心机制，接入真实 CrewAI 只需调整 import 和枚举类。

---

十、主流多智能体框架横向对比

框架	定位	Python版本	核心特色
CrewAI	角色协作	≥3.10	API 最简洁，入门最快
AutoGen	对话驱动	≥3.8	微软出品，Agent 互相对话
LangGraph	工作流图	≥3.9	流程可控，适合复杂 RAG
LlamaIndex	RAG专精	≥3.8	RAG 生态最完善
Dify	低代码平台	无需编码	有 Web UI，企业快速部署

2024年趋势：CrewAI GitHub stars 从 0 增长到 2.5万+，是增长最快的框架。

---

总结

本文手写了 CrewAI 的核心机制：

1. Agent：role + goal 注入 System Prompt，让 LLM 扮演特定角色
2. Task：context 参数实现信息接力，每步输出是下步输入
3. Crew：两种执行模式
   • Sequential：检索 → 分析 → 回答，适合大多数 RAG 场景
   • Hierarchical：Manager 动态调度多个专家，适合复杂综合问题

至此，整个系列完整覆盖了企业 AI 助手的所有核心技术：

步骤	技术	解决的问题
Step 1	numpy RAG	理解 RAG 原理
Step 2	Ollama Embedding	真实语义检索
Step 3	ChromaDB	向量持久化
Step 4	LangChain LCEL	代码规范化 + 多轮对话
Step 5	LangGraph	工作流条件控制
Step 6	MCP	工具调用能力
Step 7	Multi-Agent	角色分工协作