谁决定下一步:AI 工作流的 DAG 与事件驱动取舍

AI 工作流最容易出问题的地方,往往不是模型效果,而是"下一步谁说了算"。同样三步——抽取、判定、落库,写成 DAG 和写成事件订阅,半年后的运维成本能差一个量级。差别不在编排引擎的功能清单,而在控制权归属:决定下一步的那段逻辑,到底活在编排定义里,还是活在运行时的消息处理里。

控制权的两种归属

DAG 编排把决策前移:分支、依赖、并发度在提交时就写死在图里,调度器只负责按拓扑序推进。控制权属于图的作者,运行期不应出现图外分支。

事件驱动把决策后移:生产者只发事实,消费者按订阅关系各自决定要不要处理、处理完再发什么。控制权被拆散到每个消费者的代码里,实际路径由消息序列决定。

两者的本质交换是:DAG 用设计期的完备性换运行期的确定性,事件驱动用运行期的灵活性换设计期的简单。前者要求你写代码时就能穷举分支,后者要求你出问题时能还原路径。

判断该用哪个,不看引擎能力,看下面三个具体问题。

判据一:分支能不能在设计期穷举

这是最硬的判据。核心问法不是"AI 会不会返回意外结果",而是"意外结果的集合能不能收敛成有限枚举"。

工具调用型 AI 步骤属于可枚举:模型只能调用你注册的那几个工具,分支集合就是工具集合。这种用 DAG 完全够,做法是把不确定性关进一个决策节点,节点输出结构化 decision,边由编排器按白名单映射,模型无权直连任意后继:

DECISION_EDGES = {
    "extract":  {"ok": "validate", "empty": "human_review", "schema_error": "retry_extract"},
    "validate": {"pass": "persist", "fail": "human_review"},
}

def route(run_id: str, node: str, decision: dict) -> str:
    allowed = DECISION_EDGES[node]
    key = decision.get("next")
    if key not in allowed:                       # 模型输出越界即失败,不猜
        raise ValueError(f"unknown edge {node}->{key}, run={run_id}")
    return allowed[key]

真正不该用 DAG 的是另一类:分支取决于外部系统何时回调、取决于某个人什么时候审批、取决于上游数据什么时候到齐。这些分支在设计期不可枚举,硬塞进 DAG 只会得到两种坏结果——要么画出一个永远在改的巨型 switch 图,要么在某个节点内部又实现一遍消费循环,于是你同时拥有了两套状态机,还都不知道谁是真源。

一个可操作的判断:如果这张流程图在过去一个月里改过三次以上,或者你无法给出一个稳定版本的图,说明分支不在设计期,别用 DAG 硬编码。

判据二:补偿与重放是不是常态

AI 工作流的重放频率远高于普通业务流:调 prompt 要重跑、换模型版本要回归、评测集更新要复算。所以重放不是异常路径,是日常操作,必须按日常操作的成本来设计。

DAG 的重放是节点级的:重跑某个节点,前序节点结果直接复用。代价是你必须做两件事——把已产出的模型结果落库缓存(否则每次重跑都重新付费调用),以及给每个节点的副作用加幂等键。副作用才是难点:LLM 调用本身不可撤销,能补偿的只有它引发的外部写入。

事件驱动的重放是流级的:要么回退消费位点,要么把历史事件重投。这要求事件日志有足够保留期、消费位点可管理、每个消费者都幂等。成本明显更高,但换来的是"新增一个下游无需改主干"。

幂等键的粒度建议落在 run_id + node_id + effect_key 上,用唯一约束兜底,而不是靠应用层判重:

CREATE TABLE ai_side_effect (
  run_id     TEXT NOT NULL,
  node_id    TEXT NOT NULL,
  effect_key TEXT NOT NULL,   -- 例如 "write_case:case-8891"
  attempt    INT  NOT NULL,
  payload    JSONB,
  PRIMARY KEY (run_id, node_id, effect_key)
);
-- 写入用 ON CONFLICT DO NOTHING,重放时自然幂等

补偿动作要绑定到"副作用"而不是"步骤"。一个 AI 步骤可能产生三个副作用(写库、发通知、调外部接口),只有前两个可补偿,第三个只能登记人工处理。把这个映射显式写在节点定义里,别留给重试框架去猜。

判据三:可观测粒度决定你能不能被叫起来

DAG 给的是节点级执行记录,天然有甘特图和节点耗时,缺点是节点内部是黑盒——一次 LLM 调用里发生了什么,图上看不见。

事件驱动给的是消息级轨迹,粒度更细,但链路极易断:某个消费者忘了透传 header,整条链就断在这里,而这恰恰是事件驱动最高频的故障形态。

统一做法是三个 ID 各司其职,不允许缺省:

• run_id:一次业务意图,贯穿全流程,是排查入口;
• step_id:一次节点执行或一次消息消费,用于重放与去重;
• trace_id:跨服务调用链,只在跨进程边界时新增。

所有日志、LLM 调用记录、副作用写入都必须带 run_id + step_id。每个节点至少暴露四个指标:成功率、P95 延迟、token 成本、重试次数;事件驱动的链路额外加一个死信堆积量,并按消费者维度拆分——死信按消费者拆开,才能定位到"是谁不吞消息"。

混用时的边界:外层 DAG,内层事件

现实中绝大多数 AI 工作流是混用的:主干是 DAG(工单、审批、多阶段处理),其中扇出部分用事件——比如把一万份文档丢进队列做抽取。混用本身没问题,出问题的是状态一致性没划清。

三条硬约束:

1. 状态只由编排器写。事件消费者只上报结果事件,绝不直接改流程状态。消费者一旦能推进状态机,你就有了两个真源。
2. 迟到与孤儿事件必须显式拒绝。节点超时或已被重试覆盖后,旧结果不能再推进流程:

def on_step_result(evt):
    run = store.get_run(evt.run_id)
    if run is None or run.current_step != evt.step_id or run.state != "RUNNING":
        store.dead_letter(evt, reason="stale_or_orphan")   # 入死信,不丢也不生效
        return
    store.advance(evt.run_id, evt.step_id, evt.result)

3. 死信边界要写进文档,而不是写进某个人脑子里。建议只让三类消息进死信:反序列化失败(毒消息)、业务校验失败(重试无意义)、超过重试上限。第一类和第二类不应自动重投,直接进人工队列;只有第三类允许有限次重投,并记录每次重投的 attempt。

配套要求:每个死信队列有明确 owner 和响应时限。没有 owner 的死信队列等于数据黑洞,一周就能堆到没人敢碰。

落地建议

1. 先画控制权图再选引擎:把流程里每个"决定下一步"的点标出来,写清决策所需信息来自设计期还是运行期。运行期决策点超过三成的,别用纯 DAG。
2. 把不确定性收敛到一个决策节点:模型只输出结构化 decision,后继边由编排器按白名单映射,越界直接失败而不是猜。这一条能挡掉大部分"AI 乱跳流程"的事故。
3. 上线前先落幂等表:run_id + node_id + effect_key 唯一约束 + 副作用登记表,重放能力才有意义。没有幂等的"可重跑"是负债,不是特性。
4. 统一三个 ID 并强制透传:在消息 header、日志上下文、LLM 调用记录里都带上 run_id 和 step_id,消费者侧做透传校验,缺失即拒收。
5. 给死信定 owner 和时限:三类入死信条件、重投上限、人工队列 SLA 写进运行手册,并在监控上按消费者维度展示堆积量。