CrewAI 多智能体

用纯 Python 实现一个极简的多智能体协作框架，演示 CrewAI 的核心概念： Agent（角色）、Task（任务）、Crew（编队）、Process（执行模式）。 支持顺序（sequential）与层级（hierarchical，含委派）两种执行模式。 LLM 以可注入回调形式提供，便于离线测试。真实场景可替换为 CrewAI 库。 对应教程：第9卷《多智能体》第2章 协作与通信、第3章 分工与协调。

文件路径: examples/volume-09-multi-agent\crewai_demo.py

---

源代码

"""多智能体协作示例（CrewAI 概念实现）。

用纯 Python 实现一个极简的多智能体协作框架，演示 CrewAI 的核心概念：
**Agent（角色）**、**Task（任务）**、**Crew（编队）**、**Process（执行模式）**。
支持顺序（sequential）与层级（hierarchical，含委派）两种执行模式。

LLM 以可注入回调形式提供，便于离线测试。真实场景可替换为 CrewAI 库。

对应教程：第9卷《多智能体》第2章 协作与通信、第3章 分工与协调。
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any

# LLM 回调：接收 prompt 返回输出
LLMCallable = Callable[[str], str]


def _default_llm(prompt: str) -> str:
    """默认（无外部 LLM 时）的占位响应。"""
    return f"[规则引擎] 已处理：{prompt[:40]}..."


@dataclass
class Agent:
    """智能体。

    Attributes:
        role: 角色名（如"研究员"）。
        goal: 目标。
        backstory: 背景描述。
        llm: 可注入的 LLM 回调。
        allow_delegation: 是否允许委派任务给其他智能体。
    """

    role: str
    goal: str
    backstory: str = ""
    llm: LLMCallable = _default_llm
    allow_delegation: bool = False

    def perform(self, task_description: str, context: str = "") -> str:
        """执行一项任务。

        Args:
            task_description: 任务描述。
            context: 上游传递的上下文。

        Returns:
            执行结果文本。
        """
        prompt = (
            f"角色：{self.role}\n目标：{self.goal}\n"
            f"任务：{task_description}\n上下文：{context or '（无）'}\n请输出结果。"
        )
        return self.llm(prompt)


@dataclass
class Task:
    """任务。

    Attributes:
        description: 任务描述。
        agent: 负责的智能体。
        expected_output: 期望输出说明。
        context_from: 依赖的前序任务索引列表（用于拼接上下文）。
    """

    description: str
    agent: Agent
    expected_output: str = ""
    context_from: list[int] = field(default_factory=list)


@dataclass
class CrewResult:
    """编队执行结果。

    Attributes:
        tasks: 每个任务的输出列表。
        final: 最终汇总输出。
    """

    tasks: list[str]
    final: str


@dataclass
class Crew:
    """智能体编队。

    Attributes:
        agents: 成员智能体列表。
        tasks: 任务列表（按顺序执行）。
        process: 执行模式，``sequential`` 或 ``hierarchical``。
        manager: 层级模式下的管理者智能体。
    """

    agents: list[Agent]
    tasks: list[Task]
    process: str = "sequential"
    manager: Agent | None = None

    def __post_init__(self) -> None:
        if self.process not in ("sequential", "hierarchical"):
            raise ValueError(f"未知 process: {self.process}")
        if self.process == "hierarchical" and self.manager is None:
            raise ValueError("层级模式需要指定 manager")

    def _gather_context(self, task_index: int, outputs: list[str]) -> str:
        """收集指定任务依赖的上下文。"""
        parts: list[str] = []
        for dep in self.tasks[task_index].context_from:
            if 0 <= dep < len(outputs):
                parts.append(outputs[dep])
        return "\n---\n".join(parts)

    def kickoff(self, initial_context: str = "") -> CrewResult:
        """启动编队执行。

        Args:
            initial_context: 初始上下文。

        Returns:
            ``CrewResult``。
        """
        if self.process == "sequential":
            return self._run_sequential(initial_context)
        return self._run_hierarchical(initial_context)

    def _run_sequential(self, initial_context: str) -> CrewResult:
        outputs: list[str] = []
        for i, task in enumerate(self.tasks):
            ctx = initial_context if i == 0 else self._gather_context(i, outputs)
            if i > 0 and not task.context_from:
                # 默认接上一任务输出
                ctx = outputs[i - 1]
            outputs.append(task.agent.perform(task.description, ctx))
        final = outputs[-1] if outputs else ""
        return CrewResult(tasks=outputs, final=final)

    def _run_hierarchical(self, initial_context: str) -> CrewResult:
        """层级模式：manager 统筹，逐个把任务委派给对应 agent。"""
        assert self.manager is not None
        outputs: list[str] = []
        plan_prompt = (
            f"角色：{self.manager.role}\n"
            f"目标：制定执行计划并委派任务。\n"
            f"任务数：{len(self.tasks)}\n上下文：{initial_context or '（无）'}"
        )
        plan = self.manager.llm(plan_prompt)
        outputs.append(plan)
        for i, task in enumerate(self.tasks):
            ctx = self._gather_context(i, outputs) or plan
            delegated = (
                task.agent.perform(task.description, ctx)
                if task.agent.allow_delegation or task.agent is not self.manager
                else self.manager.perform(task.description, ctx)
            )
            outputs.append(delegated)
        final = self.manager.llm(
            f"角色：{self.manager.role}\n请汇总以下结果并给出最终答案：\n" + "\n---\n".join(outputs)
        )
        return CrewResult(tasks=outputs, final=final)


def build_research_crew(llm: LLMCallable | None = None) -> Crew:
    """构建一个研究-写作的示例编队（顺序模式）。

    Args:
        llm: 可选的共享 LLM 回调。

    Returns:
        ``Crew``。
    """
    shared = llm or _default_llm
    researcher = Agent("研究员", "收集并整理资料", llm=shared)
    writer = Agent("撰稿人", "把资料写成通俗文章", llm=shared)
    tasks = [
        Task(description="调研 AI Agent 的核心能力", agent=researcher, expected_output="要点列表"),
        Task(
            description="撰写一段科普短文", agent=writer, expected_output="短文", context_from=[0]
        ),
    ]
    return Crew(agents=[researcher, writer], tasks=tasks, process="sequential")


def main() -> None:
    """演示多智能体协作。"""
    # 用脚本化 LLM 让输出可预测
    script = iter(
        [
            "[研究] Agent 具备感知、决策、行动能力。",
            "[撰稿] AI Agent 就像能自己思考并动手的智能助手。",
        ]
    )
    crew = build_research_crew(llm=lambda _p: next(script))
    result = crew.kickoff("主题：AI Agent")
    print("各任务输出：")
    for i, t in enumerate(result.tasks):
        print(f"  [{i}] {t}")
    print(f"\n最终：{result.final}")


if __name__ == "__main__":
    main()