CrewAI 协作模式：顺序、层级、异步

CrewAI 协作模式顺序、层级、异步一、概念速查Process 类型速查模式调度方式依赖表达输出聚合Process.sequentialTask 列表按声明顺序逐个串行执行context参数声明前驱自动拼接各 Task 输出Process.hierarchicalManager Agent 运行时动态分配 Task 并协调Manager 自行决策Manager 收集后统一返回Process.consensual(已弃用)基于讨论的共识路由0.30 移除——CrewAI 在 0.30 中重构了 Process 层废弃 consensual 模式保留三种活跃模式sequential、hierarchical 和新引入的async。Sequential 模式顺序模式是最简单可靠的协作方式。Task 按照在列表中声明的顺序逐个执行前一个 Task 的输出可以通过context参数传递给后一个 Task。这种模式适合具有明确流水线关系的任务场景比如先调研后撰写或者先采集数据后分析。每个 Task 的 Agent 执行结果会按序写入共享结果池后续 Task 可以引用。fromcrewaiimportAgent,Task,Crew,Process# crewai0.30.0researcherAgent(role研究员,goal收集数据,backstory资深研究背景)writerAgent(role写手,goal撰写报告,backstory技术写作背景)researchTask(description调研 CrewAI 三种 Process 的性能差异,expected_output性能对比表格,agentresearcher,)writeTask(description基于调研结果撰写分析报告,expected_output1500 字报告文档,agentwriter,context[research],# 声明依赖write 依赖 research 的输出)crewCrew(agents[researcher,writer],tasks[research,write],processProcess.sequential,)resultcrew.kickoff()Hierarchical 模式层级模式引入一个 Manager Agent 来分配任务。Manager 负责理解任务描述、评估各 Agent 的能力与角色匹配度然后动态分配任务并监控执行过程。这种方式适合任务分配方案不确定或者需要根据中间结果动态调整的场景。Manager Agent 可以自动生成也可以由用户自定义注入。fromcrewaiimportAgent,Task,Crew,Process# crewai0.30.0managerAgent(role项目经理,goal协调团队高效完成任务,backstory你是一位资深技术经理擅长分配任务和把控质量,allow_delegationTrue,)coderAgent(role工程师,goal编写代码,backstory全栈开发背景)testerAgent(role测试,goal验证代码质量,backstoryQA 背景)dev_taskTask(description实现用户登录功能,expected_output可运行的登录模块代码,agentcoder,)test_taskTask(description为登录功能编写单元测试,expected_output测试用例通过的报告,agenttester,)crewCrew(agents[coder,tester],tasks[dev_task,test_task],processProcess.hierarchical,manager_agentmanager,# 显式指定 Manager# 不传 manager_agent 时由框架自动生成)Async 模式异步模式并非独立的 Process 类型而是 Task 级别的并行执行开关。当多个 Task 没有依赖关系时将它们标记为async_executionTrue可以让这些 Task 同时执行缩短总耗时。注意 async Task 之间不能相互依赖否则会触发依赖环检测异常。这个模式特别适合数据采集场景——同时从多个数据源抓取信息最后统一合并。fromcrewaiimportAgent,Task,Crew,Process# crewai0.30.0scraperAgent(role爬虫,goal抓取数据,backstory数据采集专家)analyzerAgent(role分析师,goal分析趋势,backstory数据分析师)fetch_aTask(description抓取电商平台 A 的手机价格,expected_output价格表,agentscraper,async_executionTrue,)fetch_bTask(description抓取电商平台 B 的手机价格,expected_output价格表,agentscraper,async_executionTrue,)mergeTask(description合并两份价格数据并输出比价报告,expected_output比价报告,agentanalyzer,context[fetch_a,fetch_b],)crewCrew(agents[scraper,analyzer],tasks[fetch_a,fetch_b,merge],processProcess.sequential,# async 由 Task 级 async_execution 控制)二、底层原理Process 引擎调度流程三种 Process 共享同一内核引擎区别在于Task 出队策略和分配决策主体。sequentialhierarchicalasync否是是否crew.kickoff()ProcessEngine解析 Task 列表Process 类型按声明顺序单线程出队Manager Agent生成分配计划标记 async_execution的 Task 同时出队检查 Task.context所有前驱已完成Manager 指派Agent 下发指令await asyncio.gather()并行执行阻塞等待前驱完成Agent 执行ReAct 循环写入 TaskOutput到共享结果池还有未完成的Task聚合 TaskOutput返回 CrewOutputSequential 的 Task 依赖链sequential 模式的核心机制是Task.context参数组成的有向无环图DAG。ProcessEngine 在 kickoff 时遍历所有 Task收集每个 Task 的context引用的前驱 Task 列表构建依赖图拓扑排序后按序执行当 Task 的context为空或所有前驱已完成该 Task 立即出队执行当前驱未完成时引擎挂起当前 Task切换到下一个就绪 Task这意味着 sequential 并非死板的列表顺序 —— 只要依赖允许引擎会跳跃执行可并行的小批量。例如 Task[A, B, C] 中 B 依赖 A 但 C 无依赖实际执行顺序是 A→C→B 或 A→(B|C) 并发。Hierarchical 的 Manager Agenthierarchical 模式引入一个调度 AgentManager其职责不是执行具体 Task而是解读当前 Crew 目标和 Task 描述为每个 Task 选择最合适的 Agent基于 Agent 的role、goal、backstoryembedding将 Task 描述改写为具体指令下发给目标 Agent检查 Agent 输出质量不合格时要求重做或换人Manager Agent 有两种来源方式代码行为自动生成Crew(processProcess.hierarchical)框架创建一个默认 Manager使用与 Crew 相同的 LLMgoal 和 backstory 硬编码自定义注入Crew(processProcess.hierarchical, manager_agentmy_manager)传入自定义 Agent 实例自由控制 Manager 的 LLM、工具、人格自定义 Manager 的意义在于生产环境中 Manager 应使用更强模型如 GPT-4 而非 GPT-4o-mini因为分配决策的质量直接决定整个 Crew 的产出上限。Manager 的allow_delegation必须为True否则 hierarchical 模式无法工作。Async 的并行与聚合async 模式不是独立的 Process 类型而是Task 级别的执行开关。在 sequential 或 hierarchical 框架下将单个 Task 的async_executionTrue即可触发并行引擎使用asyncio.gather()/ 线程池同时启动所有标记为async_executionTrue的 Task这些 Task 通常分配给同一个 Agent如爬虫同时抓取多个 URL或分配给不同 Agent所有 async Task 完成后引擎将它们的结果写入共享结果池后续依赖这些结果的其他 Task 读取聚合关键约束async Task 之间不能互相依赖。所有 async Task 在出队时必须是互不依赖的叶子节点。引擎在 kickoff 阶段会做依赖环检测如果检测到 async Task 之间存在context交叉引用抛出CrewDependencyError。输出聚合机制无论哪种 Process最终产物都是CrewOutput对象CrewOutput ├── tasks_output: List[TaskOutput] # 每个 Task 的原始输出 │ ├── task_output.description # Task 描述 │ ├── task_output.raw # Agent 生成的原始文本 │ ├── task_output.agent # 执行 Agent 名称 │ └── task_output.summary # LLM 生成的摘要 ├── raw: str # 全部 TaskOutput.raw 的拼接 └── json_dict: dict # 结构化输出若 Task 指定了 JSONsequential 模式下自动拼接所有 Task 的输出字符串hierarchical 模式下 Manager 可能对最终输出做额外整合async 模式下引擎按 Task 声明顺序排列输出而非执行完成顺序。三、架构设计原则Process 选择指南条件推荐模式理由Task 数量少10且依赖关系明确sequential调度开销接近零行为可预测Task 数量多且执行时间不均sequential async_execution让长 Task 与短 Task 重叠执行Task 分配策略需要动态决策hierarchicalManager 根据中间结果调整人力分配Agent 能力差异大需质检把关hierarchical 自定义 ManagerManager 可要求重做或重新分配纯数据并行同一 Agent 处理多来源sequential 多 async Task爬虫/扫描类场景最适用首次搭建不确定怎么选sequential下限最高refactor 到 hierarchical 成本低性能对比维度sequentialhierarchicalasync额外 LLM 调用次数0每个 Task 至少 1 次Manager 决策0端到端延迟∑(各 Task 耗时)∑(各 Task 耗时) Manager 开销max(async Task 耗时) 后续 Task 耗时确定性完全确定依赖 Manager 模型可能不稳定完全确定给定输入和种子Task 间数据传递context 显式传递Manager 上下文隐式携带context 显式传递调试难度低中需追踪 Manager 决策链中并发日志交错错误恢复Task 级重试Manager 可重新分配Task 级重试混杂模式的最佳实践顺序 异步的组合是实际项目中最常见的模式主体流程用 sequential 保证依赖正确性将 I/O 密集型子任务标记为 async 并行执行。fromcrewaiimportAgent,Task,Crew,Process# crewai0.30.0# 爬虫 Agent 并行抓取多个源scraperAgent(role爬虫,goal抓取网页内容,backstory数据采集专家)writerAgent(role编辑,goal整合信息产出报告,backstory资深编辑)tasks[]sources[https://source-a.com,https://source-b.com,https://source-c.com]fori,urlinenumerate(sources):tasks.append(Task(descriptionf抓取{url}的内容,expected_output网页正文,agentscraper,async_executionTrue,))tasks.append(Task(description整合所有来源的内容撰写一份综合报告,expected_output综合报告,agentwriter,contexttasks[:3],# 依赖所有爬虫任务完成))crewCrew(agents[scraper,writer],taskstasks,processProcess.sequential,# 主体串行子任务并行)避坑清单Manager 与 Worker 用同一个 LLMManager 决策失误会拖累所有 Worker。自定义 Manager 时至少比 Worker 用一个级别的模型。不设 max_iterAgent 在复杂 Task 上可能陷入无限推理循环导致 Token 消耗失控。async Task 过多并发数受 LLM API 速率限制和 Agent 的 tool 调用瓶颈约束建议上限 5-8 个并行 Task。忽略 context 顺序多个context中 Task 的执行结果按tasks_output中的索引顺序排列不是按完成时间。hierarchical 模式下使用 memoryFalseManager 需要跨 Task 记忆上下文才能做出连贯分配决策关闭 memory 会严重降低分配质量。