数据截至 (上游 commit 2ebb9390fe07)
03 · Flow(事件驱动的精确编排)
本章讲什么: Flow 是 Crew 之外的第二套范式。你用装饰器声明「方法之间谁触发谁」,Flow 引擎据此调度。它适合需要确定性 走向、分支路由、并行、持久化的工作流——比 Crew 的「自治队伍」更可控。
3.1 它要解决的小问题
Crew 把过程交给 Agent 自治,但很多生产场景你想要精确控制:第一步算个值,据此分支走 A 或 B,B 完成后并行触发 C 和 D,全程可中断恢复。把这些写成一堆 if/await 会很乱。Flow 让你用声明式装饰器表达这张「事件图」,引擎负责调度。
3.2 用起来什么样
# 示意,非源码:一个最小 Flow
from crewai.flow.flow import Flow, start, listen, router
class Guide(Flow):
@start() # 入口方法
def fetch(self):
return "raw data"
@router(fetch) # fetch 完成后,路由器决定走向
def decide(self, data):
return "big" if len(data) > 100 else "small"
@listen("small") # 监听路由器吐出的标签 "small"
def handle_small(self):
return "handled small"
Guide().kickoff()
关键直觉:方法的「完成」就是一个事件,@listen(X) 订阅这个事件。@router 特殊在它的返回值变成新的触发标签,从而改变后续走向。
3.3 装饰器 DSL:四件套
| 装饰器 | 含义 | 实现 |
|---|---|---|
@start() | 标记 Flow 的入口方法(可多个,并行启动) | flow/dsl/_start.py |
@listen(cond) | 当条件满足时执行;cond 可是方法引用、路由标签、或 or_/and_ | flow/dsl/_listen.py (listen) |
@router(cond) | 像 listen,但返回值成为新触发事件,用于分支 | flow/dsl/_router.py |
or_(...) / and_(...) | 多触发条件:任一满足 / 全部满足 | flow/dsl/_conditions.py |
这些装饰器不直接执行逻辑,而是把方法包成一个带 FlowMethodDefinition 的 wrapper。看 listen 的实现(flow/dsl/_listen.py:45-55):
# _listen.py:45-55(摘)
def decorator(func):
wrapper = ListenMethod(func)
_merge_flow_method_definition(
wrapper,
FlowMethodDefinition(
do=_method_action(func),
listen=_to_definition_condition(condition), # 把触发条件标准化
),
)
return wrapper
Flow 的元类 FlowMeta(flow/runtime/__init__.py:381)在类定义时扫描这些 wrapper,建立「触发条件 → 方法」的注册表。
3.4 执行 模型:谁触发谁
怎么读: kickoff 先跑所有 @start 方法;每个方法完成后,引擎查「谁在听这个方法」,先跑路由器(改走向),再并行跑普通监听器。
kickoff
│ 跑所有 @start 方法
▼
方法 M 完成 ──→ _execute_listeners(M, result)
│
① 先反复跑被 M 触发的 @router(顺序、可链式)
│ router 的返回值 → 新触发标签,可再触发别的 router
▼
② 收集全部触发标签 [M, router结果...]
▼
③ 找出被触发的普通 @listen 方法
→ asyncio.gather 并行执行
→ 每个完成后又递归 _execute_listeners
入口 Flow.kickoff(flow/runtime/__init__.py:2006)其实是 kickoff_async(:1934)的同步包装——「所有状态初始化和事件发射都在 async 方法里处理」。
调度核心是 _execute_listeners(flow/runtime/__init__.py:3036)。它的两段式很关键(见其 docstring :2722-2726):
Routers are executed sequentially to maintain flow control. Normal listeners are executed in parallel for efficiency.
先跑路由器(顺序、可链式)。 while True 循环不断找被当前触发标签触发的 router(router_only=True),执行后把 router 的返回值转成新的触发标签,可能再触发下一个 router(:2738-2778)。这让路由可以链式收窄走向。
再并行跑监听器。 对每个触发标签,_find_triggered_methods(..., router_only=False) 找出监听者,然后 asyncio.gather(*tasks) 并行执行(:2812-2821)。
3.5 or_ / and_ 条件
@listen(or_(a, b)) 表示 a 或 b 任一完成就触发;and_(a, b) 表示两者都完成才触发。and_ 的实现靠累积已见事件:_condition_met 把每个到达的触发标签记进 _pending_events,直到集合满足条件才放行(flow/runtime/__init__.py:3169-3180)。
or_ 有个细节叫**「已触发」去重**:多事件 or_ 监听器触发一次后会被记进 _fired_or_listeners,避免同一轮被重复触发(:2860-2869);循环结构里又有 _rearm_or_listeners(重新武装)机制,让循环流程中 or_ 监听器能被再次触发(:1001-1028、:2787-2788)。这是支持有环 Flow(循环工作流)的基础。
3.6 并行与竞速
除了普通的 asyncio.gather 并行,Flow 还支持竞速组(racing group):当多个监听器为同一个 or_ 条件竞争「谁先到」时,_execute_racing_listeners 处理这种「先到先得」语义,同时让非竞速的监听器照常并行(flow/runtime/__init__.py:3128-3143、_execute_racing_listeners :1105)。
3.7 状态、持久化与重放
Flow 有一个结构化状态对象 FlowState(flow/runtime/__init__.py:366),带一个 id。状态可以是字典或 Pydantic 模型,由 _initialize_state(:1688)从 inputs 初始化。
两套独立的「记住进度」机制(kickoff docstring 明确区分,:1864-1869):
| 机制 | 触发 | 用途 |
|---|---|---|
@persist + restore_from_state_id | 给方法/Flow 标 @persist,kickoff 传 restore_from_state_id | 从某次历史快照恢复状态(新 run 用新 state.id,不污染原历史) |
Checkpoint(from_checkpoint) | kickoff 传 from_checkpoint 配置 | 检查点恢复 |
两者不能同时用,同传会 raise ValueError(:1864-1869)。
3.8 巧妙之处:Crew 与 Flow 互相嵌套
Flow 的方法里可以直接 SomeCrew().kickoff(),把「自治队伍」当成流程图里的一个节点;反过来,第 1 章已说过,新版 Agent 执行器 AgentExecutor 本身就是 Flow 的子类。所以这两套范式不是对立的,而是可组合的——你用 Flow 画确定性骨架,在需要开放式推理的节点塞一个 Crew。这正是 README「combine precise workflow control ... and native support for Crews」的工程兑现。
→ 下一章:支撑这一切的横切设施(LLM 层、记忆、事件总线)与边界。