角色制团队 CrewAI:角色、任务、流程

角色制团队 CrewAI:角色、任务、流程

本节摘要:多智能体框架的同一道墙——「自主协作」在 demo 里很美,但客户报 bug 时没法确定性地重放,财务算不出单次运行的 LLM 成本,值班工程师在凌晨三点查不出哪个 Agent 卡住了。CrewAI 的回答很诚实,把两类形态分得很清:Crew 是 LLM 驱动的自主协作(角色 + 任务 + 流程),适合研究、草拟、头脑风暴这种「路径本身就是答案」的探索性工作;Flow 是代码拥有的事件驱动图(@start 入口、@listen(topic) 步骤),适合生产,可观测、可测、可重放。CrewAI 2026 年文档的生产建议很直白:「任何生产级应用,都从 Flow 开始」,把 Crew 作为 Crew.kickoff() 折进 Flow 步骤里。本节吃透四个原语(Agent / Task / Crew / Process)、三种流程(Sequential / Hierarchical / Consensus)、Crew 与 Flow 的取舍、@tool 与 BaseTool 两种工具接法、四种记忆类型,并用标准库从零实现一个三 Agent 团队(研究员、撰稿人、编辑)产出简报,跑过 Sequential 与 Flow 两种形态。读完本节,你应能识别 CrewAI 的三大失败模式(提示膨胀、管理者 LLM 税、脆性交接),并为生产选择正确的形态。

对应原课程:Phase 14 · Lesson 15 · crewai-role-based-crews(原英文 phases/14-agent-engineering/15-crewai-role-based-crews/docs/en.md)。前置:第 12 节(工作流模式)、第 14 节(Actor 模型)。

学习目标

阅读完本节,你应当能够:

1. 说出 CrewAI 的四个原语(Agent、Task、Crew、Process)以及各自拥有什么。
2. 区分 Sequential、Hierarchical、Consensus(规划中) 三种流程,并按工作负载选一种。
3. 区分 Crew(自主角色制) 与 Flow(事件驱动确定性),并解释文档为何把生产推荐留给 Flow。
4. 用 @tool 装饰器与 BaseTool 子类接工具,并在结构化输出与自由文本间取舍。
5. 说出 CrewAI 的四种记忆类型及其各自何时回本。
6. 用标准库实现一个三 Agent 团队(研究员、撰稿人、编辑),产出一份简报。
7. 识别 CrewAI 的三大失败模式:提示膨胀、管理者 LLM 税、脆性交接。

一、问题与直觉

采用多智能体框架的团队,几乎都会撞上同一道墙。

「自主协作」在 demo 里听起来很棒。然后,一个客户提交了 bug,你需要确定性地重放那一次运行;财务问这一次 LLM 路由的团队单次跑下来多少钱;值班同事在凌晨三点需要知道,到底是哪个 Agent 卡住了。

自由形式的 LLM 路由团队,这三个问题一个都答不干净。纯 DAG(有向无环图)能全部答上,却丢掉了头脑风暴 Agent 所需要的探索性形状。

CrewAI 的切分诚实面对了这道取舍:Crew 给协作的、角色制的、探索性的工作;Flow 给事件驱动的、代码拥有的、可审计的生产工作。同一个框架,两种形态,按界面挑。

四个原语

CrewAI 的表面很小。背下这四个,剩下都是配置。

• Agent(智能体) —— role + goal + backstory + tools +(可选)llm。其中 backstory(背景故事)是承重的:它塑造语气、判断、何时停止。tools 是 Agent 可以调用的函数。
• Task(任务) —— description + expected_output + agent +(可选)context +(可选)output_pydantic。一个可复用的工作单元。expected_output 是契约;context 列出上游任务,其输出会被传进来;output_pydantic 强制结构化形状。
• Crew(团队) —— 容器。拥有 agents 列表、tasks 列表、process,以及可选的 memory + verbose + manager_llm 设置。
• Process(流程) —— 执行策略。Sequential、Hierarchical、Consensus(规划中)。它决定这一次运行的形状。

关键心智模型:Agent 之间彼此看不见。Task 引用 Agent,Crew 把 Task 排序,Process 决定谁来挑下一个 Task。整个模型到此为止。

💡 版本说明:本节内容针对 CrewAI 0.86(2026-05) 验证。新版本可能重命名或合并流程类型;依赖某个具体形态前,请查阅 CrewAI Processes 文档。

三种流程

• Sequential(顺序) —— 任务按声明顺序运行,任务 N 的输出作为 context 传给任务 N+1。成本最低、最可预测。顺序固定时用它。
• Hierarchical(分层) —— 一个管理者 Agent(单独的 LLM 调用)在专家之间路由。CrewAI 会从你的 manager_llm 配置或默认值生成管理者;管理者每轮挑下一个任务,可以拒绝或改路。当专家≥4 个且顺序真实地依赖前序输出时用它。
• Consensus(共识) —— 规划中,公共 API 尚未实现。文档为未来基于投票的流程保留了这名字,今天别依赖它。

Hierarchical 在每次专家调用之上加了一次管理者 LLM 调用。一个五步运行的 token 成本可能翻三倍。只在确实需要路由时才付这笔钱。

Crew 与 Flow

这是 2026 年文档开篇就强调的框定。

• Crew —— LLM 驱动的自主性。框架在运行时决定形状。适合:研究、头脑风暴、初稿、任何「路径本身就是答案一部分」的场景。难重放、难测试、原型便宜。
• Flow —— 你拥有的事件驱动图。@start 标记入口;@listen(topic) 标记一个在某步骤发出该 topic 时触发的步骤;每个步骤是普通 Python(内部可以调一个 Crew)。适合:生产。可观测、可测、确定性。

文档 2026 年的生产建议:从 Flow 开始。当自主性确实赚回成本时,把 Crew 作为 Flow 步骤里的 Crew.kickoff() 调用折进去。Flow 给你审计轨迹,Crew 给你探索。组合,而不是二选一。

工具集成

给 Agent 一个工具有三种方式,挑最简单且合用的那个。

1. @tool 装饰器 —— 纯函数变工具。函数签名即模式,docstring 即 LLM 看到的描述。适合一次性辅助函数。

   from crewai.tools import tool
   
   @tool("Search the web")
   def search(query: str) -> str:
       """Return top results for the query."""
       return run_search(query)
   ​

2. BaseTool 子类 —— 带显式参数模式、异步支持、重试的类式工具。工具带有状态(客户端、缓存)或需要结构化参数时用它。

   from crewai.tools import BaseTool
   from pydantic import BaseModel
   
   class SearchArgs(BaseModel):
       query: str
       limit: int = 10
   
   class SearchTool(BaseTool):
       name = "web_search"
       description = "Search the web and return top results."
       args_schema = SearchArgs
   
       def _run(self, query: str, limit: int = 10) -> str:
           return self.client.search(query, limit=limit)
   ​

3. 内置工具包 —— CrewAI 自带一等适配器:SerperDevTool、FileReadTool、DirectoryReadTool、CodeInterpreterTool、RagTool、WebsiteSearchTool,一个 import 即接好。

结构化输出用 Pydantic。在 Task 上传 output_pydantic=MyModel,CrewAI 会把 LLM 响应按该校验,要么强制要么重试。配合一个紧凑的 expected_output 字符串一起用。自由文本输出适合草稿;结构化输出才是下游 Flow 能消费的东西。

记忆钩子

CrewAI 开箱提供四种记忆类型,可组合(一个 Crew 可以同时开四种)。

• Short-term(短期) —— 单次运行内的对话缓冲,运行结束即清空。
• Long-term(长期) —— 跨运行持久化。存在向量库里(默认 Chroma,可换),按与当前任务的相似度检索。
• Entity(实体) —— 按实体存的事实。「客户 X 在企业版套餐」。按实体键索引,而非按相似度,跨运行存活。
• Contextual(上下文) —— 装配时检索。在 Agent 正好需要的瞬间拉相关记忆,而非预加载。

在 Crew 上用 memory=True 或按类型配置即可开启,背后由你配置的 embeddings 提供者驱动(默认 OpenAI,可换本地)。这是 CrewAI 相对更薄框架真正赚回成本的地方——纯 LangGraph 要求你自己把每一种都接上。

角色制团队何时合适

• 三到六个带命名角色、有协作工作流的 Agent。草拟、评审、规划、头脑风暴。
• LLM 对下一步的判断本身是价值一部分的路由场景(Hierarchical)。
• 团队读 role + goal + backstory 比读图定义更开心的任何地方。

何时不合适

• 确定性 DAG,顺序严格 —— 用 LangGraph(第 13 节)。图形状是正确的抽象,CrewAI 的角色框定在这里是摩擦。
• 亚秒级延迟预算 —— Hierarchical 加往返;即便 Sequential 也把含背景故事与前序输出的提示串行化。
• 单 Agent 循环 —— 跳过框架。一个 Agent 循环(第 01 节)加一个工具注册表更短。

二、从零实现

原课程 code/main.py 用标准库实现了两种形态加一个三 Agent 团队:

• Agent、Task 数据类,对齐 CrewAI 的表面。
• SequentialCrew.kickoff(inputs) 按声明顺序跑任务,把输出作为 context 穿线。
• HierarchicalCrew.kickoff(topic) 加一个管理者 Agent,每轮挑下一个专家,在 "done" 时停。
• Flow 带 @start 和 @listen(topic) 装饰器、一个迷你事件循环、一份轨迹。
• tool(name) 装饰器镜像 CrewAI 的 @tool 形状。
• Memory 带 short_term、long_term、entity 三仓;相似度用 numpy 模拟。
• Mock LLM 响应是按角色加输入前缀硬编码的字符串,无网络、确定性。

具体演示:研究员、撰稿人、编辑三 Agent,产出一份关于「agent engineering 2026」的简报。研究员拉(mock 的)来源,撰稿人起草,编辑收紧。同一个团队再跑过一遍 Flow,以展示确定性形状。

Step 1:Agent 与 Task 数据类

@dataclass
class Agent:
    role: str
    goal: str
    backstory: str          # 承重:塑造语气与判断
    tools: list = field(default_factory=list)

@dataclass
class Task:
    description: str
    expected_output: str    # 契约
    agent: Agent
    context: list = field(default_factory=list)   # 上游 Task
    output_pydantic: type = None
​

Step 2:Sequential Crew 把输出穿线

class SequentialCrew:
    def __init__(self, agents, tasks):
        self.agents, self.tasks = agents, tasks

    def kickoff(self, inputs):
        outputs = {}
        for t in self.tasks:                       # 按声明顺序
            ctx = [outputs[c.name] for c in t.context if c.name in outputs]
            prompt = f"角色:{t.agent.role}\n目标:{t.agent.goal}\n" \
                     f"背景:{t.agent.backstory}\n任务:{t.description}\n" \
                     f"契约:{t.expected_output}\n上游:{ctx}\n输入:{inputs}"
            outputs[t.name] = mock_llm(t.agent, prompt)   # 按角色 mock
        return outputs
​

Step 3:Hierarchical Crew 加管理者

class HierarchicalCrew:
    def __init__(self, agents, tasks, manager_llm):
        self.agents, self.tasks, self.manager = agents, tasks, manager_llm

    def kickoff(self, topic):
        trace, done = [], False
        while not done:
            pick = manager_llm(self.tasks, trace, topic)   # 管理者挑下一步
            if pick == "done": break
            trace.append(run_task(pick))                   # 多一次 LLM 调用
        return trace
​

Step 4:Flow 事件驱动(确定性)

class Flow:
    def __init__(self): self.listeners = {}
    def start(self, fn):   self.listeners.setdefault("@start", []).append(fn); return fn
    def listen(self, topic):
        def deco(fn): self.listeners.setdefault(topic, []).append(fn); return fn
        return deco
    def run(self, inputs):
        bus, trace = [{"topic": "@start", "data": inputs}], []
        while bus:
            ev = bus.pop(0)
            for fn in self.listeners.get(ev["topic"], []):
                out = fn(ev["data"]); trace.append((ev["topic"], out))
                if isinstance(out, dict) and "topic" in out:
                    bus.append(out)            # 步骤发新 topic
        return trace                           # 形状固定
​

💡 设计要点:Crew 的轨迹是流式的(管理者原则上可以重排);Flow 的轨迹是固定的。这个选择本身就是本节的课。Crew 的审计门在 Flow 这一圈外面。

运行 python3 code/main.py 会打印:Sequential crew 把输出经 context 穿线、Hierarchical crew 由管理者挑选(研究员→撰稿人→编辑→"done")、Flow 跑同样三步但 topic 显式(researched/drafted/edited)、@tool 路由的工具调用、长期记忆跨两次 kickoff 存活。

三、框架对比

CrewAI 坐在「协作式角色制」这一角。选型矩阵的简版:

依赖形状:与 LangChain 独立;Python 3.10~3.13;用 uv 管理。AWS Bedrock 集成有文档;厂商基准报告相对 LangGraph 在 QA 工作负载上有显著提速,但方法论(数据集、硬件、评估指标)未公开,故把框架厂商的数字仅当方向性参考。

四、可复用产物

原课程 outputs/skill-crew-or-flow.md:为给定任务在 Crew 与 Flow 间做选择,并脚手架最小实现。硬拒绝三种情况——Crew 不带 backstory、Flow 不带显式 topic、Hierarchical 专家少于 3 个。

五、练习

1. (Easy) 把 Sequential crew 改写成 Flow。数一数可变性下降的接触点,记下可读性在哪下降了。
2. (Easy) 给团队加实体记忆:关于某个客户的事实跨 kickoff 持久化,验证检索拉对了实体。
3. (Medium) 实现一个 Hierarchical 流程,管理者在撰稿人输出不足三段前拒绝路由给编辑。追踪重试。
4. (Medium) 为(mock 的)网页搜索接一个 BaseTool 子类,对比它与 @tool 装饰器版本的轨迹形状。
5. (Hard) 给编辑任务加 output_pydantic=Brief(Brief 含 title、summary、sections)。让撰稿人任务输出一次畸形的 JSON,验证 CrewAI 在轨迹里的重试行为。
6. (Hard) 读 CrewAI 文档简介,把玩具移植到真实的 crewai API。标准库版本跳过了哪些保证?
7. (Hard) 把 AgentOps 或 Langfuse(第 24 节)接到一次真实运行上。你在标准库版本里漏掉了哪些轨迹?

本节要点回顾

1. 四原语:Agent(role+goal+backstory+tools)、Task(description+expected_output+agent)、Crew(容器)、Process(执行策略),Agent 之间彼此看不见。
2. backstory 是承重的:塑造语气、判断、何时停止;不要当装饰。
3. 三种流程:Sequential(最便宜可预测)、Hierarchical(管理者每轮多一次调用,token 可翻三倍)、Consensus(规划中,别依赖)。
4. Crew vs Flow 是核心取舍:Crew 给探索(LLM 驱动、难重放),Flow 给生产(事件驱动、可审计)。
5. 2026 生产建议:从 Flow 开始,把 Crew 作为 Crew.kickoff() 折进 Flow 步骤——组合,不二选一。
6. 工具三接法:@tool(最简)、BaseTool 子类(带状态/结构化参数)、内置工具包。
7. 结构化输出用 Pydantic:output_pydantic=MyModel + 紧凑 expected_output;自由文本只适合草稿。
8. 四种记忆:短期(运行内)、长期(向量库跨运行)、实体(按实体键)、上下文(装配时检索);纯 LangGraph 要自己全接上。
9. 三大失败模式:提示膨胀(backstory<200 字)、管理者 LLM 税(不需要路由就退 Sequential)、脆性交接(用 output_pydantic 让下游读类型对象而非自由文本)。
10. 不该用 CrewAI 时:确定性严格 DAG(用 LangGraph)、亚秒延迟、单 Agent 循环。

下一节,我们进入 OpenAI Agents SDK——把 handoff(智能体交接)、guardrail(护栏)与 tracing(追踪)作为一等公民的生产框架,看它如何用厂商原生的工具调用与结构化输出,把本节的「角色 + 任务 + 流程」压缩成更短的代码。