CrewAI Task+Process：多Agent协作

可视化演示

\u5c55\u793a\u6570\u636e\u5e93\u5d29\u6e83\u540e\u901a\u8fc7WAL\u5199\u524d\u65e5\u5fd7\u8fdb\u884c\u6062\u590d\u7684\u5168\u8fc7\u7a0b\uff1a\u5206\u6790\u65e5\u5fd7\u786e\u5b9a\u4e8b\u52a1\u72b6\u6001\uff0c\u5bf9\u5df2\u63d0\u4ea4\u4e8b\u52a1REDO\u91cd\u505a\uff0c\u5bf9\u672a\u5b8c\u6210\u4e8b\u52a1UNDO\u64a4\u9500\u3002

Task 定义与分配

在老牌餐厅里，每一道工序都有一张挂在墙上的「工序卡」：这道工序做什么、由哪个岗位负责、做完长什么样算合格、允许用哪口锅哪把刀，写得明明白白。新来的厨师不用到处问人，看一眼卡片就知道自己该干什么。CrewAI 的 Task 就是这张工序卡。

在 CrewAI 中，Task 是协作的最小执行单元，它把「做什么、谁来做、做成什么样、用什么做」封装成一条结构化指令。一个清晰的 Task 是高效协作的地基：卡片上的信息模糊，后厨就会乱套；卡片清晰，每个 Agent 才能在自己的岗位上稳定输出。

Task 的核心四要素通常是 description、agent、expected_output 和 tools。description 描述目标，越具体 Agent 越不容易跑偏；agent 把任务绑定到某个有对应技能的 Agent；expected_output 相当于验收标准，告诉 Agent 输出的格式与内容；tools 则允许任务调用搜索、代码解释器等外部能力，相当于允许厨师使用特定的锅和刀。

回到后厨：一张好的工序卡，能让新来的帮厨也立刻上手。在设计多 Agent 系统时，花时间把每个 Task 的四要素写清楚，远比堆砌更多 Agent 更有效——工序卡不清晰，再好的厨师也会炒出黑暗料理。

顺序 Process

流水线上，洗好的菜放在传菜口，切菜师傅取走切好，再放到下一个传菜口，炒菜师傅接手……每个工位只做一件事，做完就传给下一位。谁也不会抢跑，也不会有人闲着。这就是后厨的「顺序配合」，也是 CrewAI 中最直观的协作方式。

CrewAI 的 SequentialProcess 正是这种模式：把多个 Task 排成一个列表，按顺序逐个执行。前一个任务的输出会自动成为后续任务的上下文，就像半成品在工位间流动，把信息一层层带下去。组织流水线时，你只需要想清楚任务的前后依赖，并把每一站的「验收标准」写具体，剩下的由框架自动衔接。

这种模式简单、可控、token 开销相对较低。它非常适合流程固定、步骤明确的工作，比如「调研 → 写作 → 审核」的内容生产链。每个 Agent 只拿到它所需要的那部分上下文，不容易被无关信息干扰，产出的稳定性也更高。

流水线的弱点也很明显：如果某一步突然需要拆成更多步骤，整条线就得停下来重新排。所以顺序 Process 适合「菜单早就定好」的常规流程，而面对「客人临时加菜」这类复杂需求，就要请出主厨模式了。

层级 Process

想象一桌婚宴的大订单：主厨不会自己从头炒到尾，而是看一眼菜单，心里快速拆解——凉菜给小李、热菜给老王、汤给阿珍，每人分到一张写着「做什么、几点前完成」的小纸条。做完之后，主厨还要逐盘尝一口，不合格的打回重做。这就是后厨的「主厨模式」。

CrewAI 的 HierarchicalProcess 就是这个模式的软件版：一个 Manager Agent（对应主厨）接收整体目标，动态创建并委派子任务给其他 Agent，然后审查每个返回的结果。要实现它，你需要提供一个 manager_llm 或 manager_agent 来扮演主厨角色，由它来决定「这桌菜怎么拆、派给谁」。

与顺序 Process 最大的不同是：层级模式不要求你预先列出所有子任务——主厨现场拆解。这使它特别适合目标复杂、子任务不确定的场景，比如「写一份商业计划书」这种需要边做边调整的大目标。Manager 还拥有局部决策权，可以决定某个子任务是否需要返工、是否需要调用额外资源。

主厨模式也不是万能的：每拆解一次、审查一次，都要额外花 token 和时间，就像婚宴主厨也要付加班费。当流程已经稳定时，回去用流水线反而更划算——这正是选型时要权衡的关键：你的任务是固定菜单，还是创意大餐？

Agent 通信机制

后厨里没有人大声聊天。配菜切好，往传菜口一放；炒锅师傅瞄一眼进度板，就知道该炒哪道菜；需要问前台，就通过对讲机说一句。信息跟着「半成品」走，而不是靠人围在一起闲聊。

CrewAI 的 Agent 之间也没有「自由闲聊」。协作信息主要通过任务衔接传递：上一个 Task 的输出，作为下一个 Task 的输入（context）出现。同时，Crew 内的共享上下文与记忆机制，让多个 Agent 能感知同一份「进度板」——比如一个 Agent 中途发现的约束条件，可以被后续 Agent 读取。

除此之外，Agent 可以通过工具调用把外部数据写进上下文，比如搜索结果的摘要、数据库查询结果。一次消息本质上是一个「（发送方 → 接收方：内容）」的结构，只是这个「发送」由任务编排自动完成，而不是 Agent 自己发起。这种设计让每一条信息的来源和去处都可追踪、可审计。

这种「传菜口式」通信的好处是清晰、可控——每一条信息都有来源和去处；缺点是不像真人讨论那样灵活。如果你需要 Agent 像圆桌会议一样你一言我一语地碰撞想法，那就要看 AutoGen 的思路了。

结果聚合

各灶台的菜炒好了，不能直接端上桌——需要摆盘师傅统一检查：菜量够不够、温度行不行、酱汁有没有串味，最后合成一道完整的大菜再上桌。结果聚合，就是后厨的「摆盘出品」环节。

在多 Agent 协作中，每个 Task 都会产出结果；但如果直接把这些文本拼在一起，往往会有重复、冲突或格式混乱。聚合阶段要做的三件事是：解析（Parse）、合并（⊕）和校验（Validate）。CrewAI 会在每个 Task 结束后解析输出，并把整个 Crew 的成果汇总为一个统一的输出对象。

解析指把原始输出转成结构化内容，比如解析 JSON 或 Markdown 表格；合并是把多份结果去重、去冲突、按顺序拼装；校验则是拿 expected_output 当验收标准，检查最终结果是否达标。如果某份输出格式不符，聚合层可能直接丢弃或要求重跑——就像摆盘师傅把炒糊的菜退回炒灶。

没有摆盘师傅的后厨，菜再好也上不了台面；没有结果聚合的多 Agent 系统，单看每个结果都很好，合在一起却没法用。设计协作流程时，从一开始就要约定好每份输出的格式——聚合的代价，取决于你在 Task 卡片上有没有写清楚「做成什么样」。

与 AutoGen 对比

如果你走进另一家店，会发现完全不同的风格：他们不设固定流水线，而是让所有厨师围坐在一张大圆桌旁，针对「今天做什么菜」自由发言，聊着聊着决定分工。这家店就是 AutoGen——多智能体的「圆桌会议」模式。

CrewAI 与 AutoGen 最根本的区别在驱动方式：CrewAI 是任务驱动，你事先用 Task+Process 把流程编排好，Agent 按工序执行；AutoGen 是会话驱动，核心是 GroupChat，多个 Agent 在一个会话里轮流发言，对话的走向决定任务如何推进。一句话：CrewAI 像按菜谱执行的后厨，AutoGen 像边聊边定的创意厨房。

各有优劣。CrewAI 的后厨式编排流程显式、可预测、token 开销相对可控，适合生产环境中的确定性流程；AutoGen 的圆桌模式灵活、能自由多轮协商，但结果随机性高、更难审计。选型应该根据任务特征：线性固定流程选顺序 Process；复杂动态目标选层级 Process；需要多 Agent 自由讨论协商则考虑 AutoGen。

回到开头的比喻：你是开一家「固定菜单的流水线餐厅」，还是一家「每天圆桌定菜的创意餐厅」？没有最好的后厨，只有最适合客人需求的模式。作为 AI Agent 架构师，你的职责就是看清每道工序的特征，然后把它们安排到最合理的协作结构中。

AI 应用场景

Python 演示

# 模拟 CrewAI 的 Task + Process 多智能体协作 class Task:    def __init__(self, desc, agent=None, expected_output=""):        self.desc = desc        self.agent = agent        self.expected_output = expected_output        self.output = ""     def run(self):        if self.agent is None:            return ""        self.output = self.agent.execute(self)        return self.output  class Agent:    def __init__(self, name, skill):        self.name = name        self.skill = skill        self.gc_count = 0     def execute(self, task):        self.gc_count += 1        return "[%s] 完成: %s -> %s" % (self.name, task.desc, self.skill)  class ManagerAgent(Agent):    def decompose(self, goal, n):        return ["%s 子任务%d" % (goal, k + 1) for k in range(n)]  def sequential_process(agent_list, task_list):    # 顺序 Process: 依次执行, 前序输出累积为上下文    context = []    for i, task in enumerate(task_list):        if task.agent is None:            task.agent = agent_list[i % len(agent_list)]        task.run()        context.append(task.output)    return context  def hierarchical_process(manager, worker_list, goal, n=3):    # 层级 Process: 主厨拆解目标, 委派工人并回收结果    results = []    for i, desc in enumerate(manager.decompose(goal, n)):        worker = worker_list[i % len(worker_list)]        task = Task(desc, worker)        task.run()        results.append(task.output)    return results  def main():    researcher = Agent("研究员", "资料收集")    writer = Agent("写手", "内容撰写")    editor = Agent("编辑", "质量把关")     print("== 顺序 Process: 流水线协作 ==")    tasks = [        Task("收集 2025 年 AI 芯片市场资料"),        Task("撰写一篇千字市场分析"),        Task("审核并润色文章"),    ]    for out in sequential_process([researcher, writer, editor], tasks):        print(out)     print("\n== 层级 Process: 主厨委派 ==")    boss = ManagerAgent("主编", "任务拆解")    for out in hierarchical_process(boss, [researcher, writer], "发布技术周报", 3):        print(out)  if __name__ == "__main__":    main()

1. self.output = self.agent.execute(self)任务把工序卡交给绑定的 Agent 执行，产出写入任务对象。
2. if task.agent is None: task.agent = agent_list[...]未指定负责人的任务，按顺序自动分配到团队成员。
3. context.append(task.output)前序输出累积进上下文，流水线下一步就能看到半成品。
4. manager.decompose(goal, n)主厨（Manager）把大目标拆解成 n 个子任务。
5. worker_list[i % len(worker_list)]轮询分配工人，让团队负载均衡。
6. task.run()执行单个任务，触发对应 Agent 的 execute 逻辑。

常见误区

面试题