Crew 与 Flow 双模式编排

学习目标

理解 CrewAI 如何通过 Crew 和 Flow 两种编排模式覆盖从简单任务分配到复杂业务流程的全部场景，并掌握何时选择哪种模式的决策方法。

前置知识

• 多 Agent 路由与隔离 — Agent 隔离与委托架构
• 多 Agent 编排与置信度过滤 — Orchestrator 编排模式

下文假设你已理解上述概念，直接聚焦 CrewAI 的具体实现。

---

项目实践

Crew 模式：角色团队编排

Crew 是 CrewAI 最基础的编排抽象。一个 Crew 包含一组 Agent 和一组 Task，通过 process 参数控制执行方式。

# 伪代码
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,  # 或 Process.hierarchical
)
result = crew.kickoff()

Sequential 执行：按 tasks 列表顺序执行，每个任务的输出作为后续任务的上下文。代码路径：kickoff() → _run_sequential_process() → _execute_tasks()。

Hierarchical 执行：创建 Manager Agent 来协调任务分配。Manager 使用 LLM 决定哪个 Agent 执行哪个任务，通过 DelegateWorkTool 实现委派。

Crew 的生命周期：

流式输出：设置 stream=True 时，kickoff() 返回 CrewStreamingOutput（同步/异步迭代器），允许在任务执行过程中消费中间结果。

并行执行：kickoff_for_each_async() 使用 asyncio.gather 并行执行多个 crew 副本（例如同一任务应用于不同输入数据）。

Flow 模式：事件驱动状态机

Flow 是比 Crew 更底层的编排抽象。它使用装饰器定义方法的执行顺序和依赖关系。

四个核心装饰器：

装饰器	作用	示例
@start()	标记入口点	@start() 或 @start("method_name")
@listen(condition)	监听方法完成事件	@listen("process_data")
@router(condition)	动态路由，返回值决定下一步	@router("check_status")
and_() / or_()	组合条件	@listen(and_("a", "b"))

# 伪代码
class ReviewFlow(Flow[ReviewState]):
    @start()
    def submit_code(self):
        self.state.review_request = gather_changes()

    @listen("submit_code")
    def run_analysis(self):
        self.state.analysis = static_analysis(self.state.review_request)

    @router("run_analysis")
    def route_by_severity(self):
        if self.state.analysis.critical_count > 0:
            return "BLOCK"
        return "APPROVE"

    @listen("BLOCK")
    def block_merge(self):
        send_notification("PR blocked: critical issues found")

    @listen("APPROVE")
    def approve_merge(self):
        send_notification("PR approved")

Flow 的并发安全：self.state 通过 StateProxy + threading.Lock 保证并行 listener 对状态的写操作安全。Sync 方法通过 asyncio.to_thread() 在线程池执行，防止阻塞 event loop。

---

问题与规避

Crew 的 Sequential 模式下任务间无共享状态

• 任务间数据传递依赖 TaskOutput 链式传递，没有共享的 self.state
• 对策：如果需要在任务间共享复杂状态，使用 Flow 模式或将状态序列化到 Task 的 context 参数

Flow 的嵌套 AND/OR 条件复杂度

• 多层嵌套 and_(or_("a", "b"), "c") 随方法数量指数增长
• 对策：将复杂条件拆分为多个中间方法，每个方法只负责一个简单的 AND/OR 判断

Crew 嵌入 Flow 时的状态同步

• Crew 的 CrewOutput 需要手动映射到 Flow 的 self.state 字段
• 对策：在 Flow 方法中调用 crew.kickoff() 后，立即将结果写入 self.state

---

设计取舍

Crew vs Flow 的选择不是互斥的

• Crew 专注于”谁做什么”（Agent 任务分配），Flow 专注于”什么之后做什么”（事件驱动控制流）
• Crew 继承自 FlowTrackable，可以被 Flow 追踪和嵌入
• 实际项目中可以组合使用：Flow 控制整体流程，Crew 负责具体的多 Agent 协作阶段

装饰器 DSL vs 声明式 YAML/JSON

• CrewAI 选择 Python 装饰器而非 YAML DSL
• 优势：类型安全、IDE 支持、可调试
• 代价：复杂条件嵌套时代码可读性下降

---

参考来源

• CrewAI 官方文档 — Crew 与 Flow 的概念介绍