Agent 任务编排的 DAG 化与可重放执行工程 2026

一、问题的提出：从线性 chain 到并行 DAG

截至 2026 年 8 月，主流 LLM agent 框架（LangGraph、CrewAI、AutoGen、OpenAI Agents SDK、Claude Agent SDK）默认都把"任务编排"建模为线性链式调用：prompt → tool call → 观察结果 → 下一轮 prompt → 下一轮 tool call。这种模型在小任务下很自然，但一旦任务深度超过 10 轮、或者任务天然存在并行分支（如"同时查询 5 个数据库再合并"），链式执行就暴露出三个根本问题：

第一，延迟是串行累加的。一个典型的 research agent 链路——搜索 5 个源 + 抽取实体 + 交叉验证 + 生成报告——如果全部串行，需要约 35 秒；如果识别出"搜索 5 个源"是无依赖的并行子任务，并行执行可以压缩到 9 秒左右。这是 4 倍的 P95 延迟差距，对生产 SLA 是不可接受的。

第二，重试的颗粒度太粗。链式模型下，任意一步失败都意味着从头来过。但生产环境里"第 3 步失败、第 4 步还没启动"和"第 4 步执行到一半失败"在拓扑上是完全不同的失败——前者可以"重放第 3 步"，后者必须"补偿第 4 步的部分执行"。链式模型无法区分这两种语义。

第三，可观测性是扁平的。trace 是一条直线，无法回答"为什么这一步等了 12 秒才返回""哪个上游任务的延迟传导到了下游"这种典型的 SRE 问题。

这篇文章要解决的核心问题是：如何把 agent 的任务编排从线性 chain 升级为 DAG（有向无环图），并在此基础上构建可重放、可调试、可观测的执行引擎。我们会从拓扑形式化、执行引擎工程、event sourcing 与 checkpoint、replay 与 time-travel debugging、生产踩坑、与现有框架对比、可观测性清单 7 个维度系统展开。

我们采用学术风格但重工程实战的写法：每个核心概念会给出精确定义和工程取舍，每个抽象会落到具体的代码或伪代码，最后给出生产可用的可观测性指标清单。我们假设读者熟悉 async/await、LLM tool calling、trace 系统的基本概念。

二、为什么必须 DAG 化：任务的拓扑结构与因果序

任务 DAG 不是新概念——数据库的查询计划、编译器 IR、CI/CD pipeline、Airflow 都已经使用了 DAG。但 LLM agent 的任务 DAG 有三个独有特征：节点的产出高度依赖 LLM、节点之间的依赖关系可能动态变化、节点的执行有显著的非确定性。

我们先把任务图严格定义如下。设 $T = (V, E)$ 是一个有向无环图，其中每个节点 $v \in V$ 是一个原子操作（一次 LLM call、一次 tool call、或者一次纯函数），每条边 $e = (v_s, v_d) \in E$ 表示因果依赖：$v_d$ 的执行必须等待 $v_s$ 的成功完成。关键约束：$T$ 必须是 DAG，不允许环。环意味着死循环或需要外部干预的动态依赖。

为什么必须是 DAG？三个理由：

1. 可终止性保证：DAG 的拓扑序是有限的，因此执行必然终止。带环的图需要靠"重试上限"或"超时"来强制终止，这在生产里是反模式——它把终止保证从结构层降级到了运行时检查层。
2. 可重放性基础：DAG 的拓扑序是确定的，给定相同的输入和事件日志，可以 deterministic 地复现执行轨迹。如果有环，相同的输入在不同执行下可能走完全不同的环路径，replay 就退化为"再跑一次"的同义词。
3. 可并行性识别：DAG 的 topological sort 天然给出最大并行度——同一层的节点可以并行，不同层的必须串行。这是并行执行的基础。

agent 任务里常见的 4 类拓扑结构是：

• Fan-out：一个节点的产出扇出到多个独立下游节点。比如"读完论文后，分别抽取方法、实验、引用"。拓扑上是一个根节点 + 多个叶子。
• Fan-in：多个独立上游节点的产出汇聚到一个聚合节点。比如"查询 5 个数据源后合并去重"。拓扑上是多个根 + 一个聚合节点。
• Diamond：fan-out 后接 fan-in，形成菱形。比如"查询多个数据库 → 各自抽取 → 交叉验证 → 报告"。这是最常见的复杂拓扑，也是最容易在 LLM agent 里出错的——交叉验证节点的输入是动态生成的。
• Conditional branch：基于某个节点的输出来决定下游走哪个分支。这是 DAG 化的难点，因为"分支"在执行时才会确定。工程上我们通常用"条件节点 + skip 标记"来处理：未选中的分支节点标记为 skipped，不执行。

为什么 LLM 自己规划容易漏依赖？ 因为 LLM 的"任务分解"是基于自然语言理解的，它对"哪些任务可以并行、哪些必须串行"的判断是概率性的而非结构性的。实测中，我们发现 LLM 倾向于过度串行——它会把"可以并行查询 3 个 API"的任务写成 3 步串行，因为串行更符合自然语言叙述习惯。结构化的 DAG 化把"哪些可以并行"从 LLM 的判断中抽离出来，由调度器根据拓扑结构自动决定。

三、task graph 形式化：节点、边、状态机的统一建模

这一节我们给出 task graph 的工程化定义。一个完整的 task graph 由三部分组成：节点定义、边定义、状态机。

@dataclass
class TaskNode:
    id: str                              # 唯一标识，UUID 或 nanoid
    op: Callable[[Context], Awaitable[Any]]  # 节点操作，可能是 async
    deps: list[str]                      # 依赖的上游节点 id 列表
    retry_policy: RetryPolicy            # 重试策略（max_attempts, backoff, jitter）
    timeout: float                       # 单次执行超时（秒）
    idempotency_key: Optional[str]       # 用于检测重复执行
    metadata: dict                       # 任意元数据，用于可观测性
    
    # 运行时动态填充
    status: NodeStatus = NodeStatus.PENDING  # PENDING/RUNNING/SUCCESS/FAILED/SKIPPED
    attempts: int = 0
    result: Optional[Any] = None
    error: Optional[Exception] = None
    started_at: Optional[float] = None
    finished_at: Optional[float] = None

@dataclass
class TaskEdge:
    src: str        # 源节点 id
    dst: str        # 目标节点 id
    port: str       # 端口名（用于 fan-in 时按端口合并上游输出）

关键设计决策：

• op 是 Callable，不只是字符串：很多框架把 op 存为字符串（tool name），运行时再 dispatch。我们的设计直接存 Callable，这样可以在编译期做类型检查，也可以在测试里直接 mock。
• deps 是 list 不是 set：保留顺序，便于 deterministic 序列化与 replay。
• retry_policy 与 timeout 是显式字段：不放 metadata 里，是为了在调度器里能直接读取而无需解 metadata。
• idempotency_key 是 Optional：只有"对外部世界有副作用"的节点（如发送邮件、扣款）才需要；纯计算节点不需要。

状态机方面，每个节点有 5 个状态：PENDING → RUNNING → {SUCCESS, FAILED, SKIPPED}。转移规则：

• PENDING → RUNNING：当且仅当所有 deps 节点都进入 SUCCESS 状态。
• RUNNING → SUCCESS：op 正常返回。
• RUNNING → FAILED：op 抛异常（且已耗尽 retry_policy）。
• PENDING → SKIPPED：节点本身或其上游节点被 conditional branch 标记为不执行。
• RUNNING → FAILED：超时（timeout 触发）或显式 cancel。

注意我们不允许 FAILED → RUNNING 的自动恢复——一旦 FAILED，必须由外部显式触发 retry。这避免了"自我修复"的隐性状态污染。

与 Temporal/Cadence/Airflow 的对比：Temporal 用的是 workflow + activity 的双层模型，activity 是节点，workflow 是节点编排（包含分支、循环、子工作流）。Cadence 类似。Airflow 用的是 DAG 文件 + Operator 的模式，更适合 ETL 而非交互式 agent。我们的 task graph 比 Temporal 更轻量（不引入 namespace/versioning 这些企业级特性），但比 Airflow 更动态（节点拓扑可以在执行时变化）。

重要工程细节：port 字段。fan-in 节点需要把多个上游的输出合并，但合并逻辑往往不是简单的 concatenate——例如"3 个查询结果合并去重"和"3 个查询结果取第一个非空"是两种完全不同的合并。port 字段让每个上游输出走不同的"槽位"，fan-in 节点的 op 可以按 port 读取。我们的工程经验是：fan-in 节点的 op 应该是纯函数，不接受"任意 dict"，而是显式列出它需要的 ports。这是减少 bug 的关键约束。

四、execution engine 工程：调度器、reactor、并发控制

执行引擎是 task graph 的大脑。它负责四件事：节点调度、并发控制、超时传播、取消传播。

class ExecutionEngine:
    def __init__(self, graph: TaskGraph, max_concurrency: int = 16):
        self.graph = graph
        self.sem = asyncio.Semaphore(max_concurrency)
        self.event_log = EventLog()
    
    async def run(self) -> TaskGraph:
        # 初始化：所有节点都是 PENDING
        await self.event_log.append(EngineStartEvent(graph_id=self.graph.id))
        
        # 调度循环
        while not self._is_terminal():
            ready_nodes = self._ready_nodes()  # deps 全 SUCCESS 且自身 PENDING
            if not ready_nodes and self._has_running():
                await asyncio.sleep(0.01)  # 等运行中的节点
                continue
            if not ready_nodes and not self._has_running():
                break  # 死锁：没有 ready 也没有 running
            
            # 限制并发
            async def _run_with_sem(node):
                async with self.sem:
                    await self._run_node(node)
            
            await asyncio.gather(*[_run_with_sem(n) for n in ready_nodes])
        
        await self.event_log.append(EngineEndEvent(graph_id=self.graph.id))
        return self.graph

调度策略的选择：我们用"拓扑序 + 优先级"——同一层的节点按 metadata.priority 排序，不同层必须串行。优先级让"关键路径"上的节点先调度，但要注意避免优先级反转（高优先级节点等低优先级节点的输出）。

concurrency limit 是生产必加的。否则一个 fan-out=100 的节点会瞬间把 LLM API 打挂，或者把下游数据库压垮。工程经验：max_concurrency 设为"你的下游 API rate limit / 单节点预期 QPS"，向上取整。我们的经验值是 16-64。

backpressure：当所有 worker 都在跑（sem 已满），新 ready 的节点会自然等 sem 释放。这是天然的 backpressure，无需额外实现。不要为了"更快"而无限拉高 max_concurrency——这会把故障域扩散到所有节点。

超时传播：节点的 timeout 是"单次执行超时"。但还有"工作流总 deadline"的概念——例如用户的 SLA 是 30 秒，整个工作流必须在 30 秒内完成。实现上用一个全局 deadline timer，定期检查"如果现在就跑最快路径，能否赶上 deadline"，如果赶不上就主动 skip 非关键节点。

async def _run_node(self, node: TaskNode):
    node.status = NodeStatus.RUNNING
    node.started_at = time.time()
    node.attempts += 1
    await self.event_log.append(NodeStartEvent(node_id=node.id, attempt=node.attempts))
    
    try:
        # 检查全局 deadline
        remaining = self.deadline - time.time() if self.deadline else None
        per_node_timeout = min(node.timeout, remaining) if remaining else node.timeout
        
        # 执行（带超时）
        result = await asyncio.wait_for(node.op(self._build_context()), timeout=per_node_timeout)
        
        node.status = NodeStatus.SUCCESS
        node.result = result
        node.finished_at = time.time()
        await self.event_log.append(NodeSuccessEvent(node_id=node.id, result=result))
    
    except asyncio.TimeoutError:
        node.status = NodeStatus.FAILED
        node.error = TimeoutError(f"node {node.id} timeout after {per_node_timeout}s")
        node.finished_at = time.time()
        await self.event_log.append(NodeFailEvent(node_id=node.id, error=node.error))
    
    except Exception as e:
        if node.attempts < node.retry_policy.max_attempts:
            backoff = node.retry_policy.backoff * (2 ** (node.attempts - 1))
            await asyncio.sleep(backoff)
            node.status = NodeStatus.PENDING  # 重试
        else:
            node.status = NodeStatus.FAILED
            node.error = e
            await self.event_log.append(NodeFailEvent(node_id=node.id, error=e))

取消传播：当用户取消整个 agent run 时，所有 RUNNING 节点必须收到 cancel 信号。我们用 asyncio.CancelledError 实现：父任务 cancel 后，所有 asyncio.gather 的子任务都会收到 cancel，节点的 op 必须正确处理这个异常（清理临时文件、回滚数据库事务、释放锁）。

reactor 模式：上面的伪代码是简化版。生产实现通常用一个"reactor loop"——一个持续运行的 async 循环，从 ready queue 取节点、spawn worker、更新状态。reactor 模式的优势是所有状态都在一个地方，便于可观测性和故障恢复。

五、event sourcing 与 checkpoint：可重放的真相之源

可重放性是 DAG 化最大的工程回报。要让 agent 可重放，唯一真相之源必须是事件日志，而非节点状态。这就是 event sourcing 模式。

event log 的设计：

@dataclass
class EngineStartEvent:
    timestamp: float
    graph_id: str
    graph_definition: bytes  # 整个 TaskGraph 的序列化（pickle/protobuf/JSON）

@dataclass  
class NodeStartEvent:
    timestamp: float
    node_id: str
    attempt: int
    input_snapshot: bytes  # 节点输入的 deep copy

@dataclass
class NodeSuccessEvent:
    timestamp: float
    node_id: str
    attempt: int
    output: bytes  # 节点输出的序列化

@dataclass
class NodeFailEvent:
    timestamp: float
    node_id: str
    attempt: int
    error_type: str
    error_message: str
    traceback: str

@dataclass
class EngineEndEvent:
    timestamp: float
    graph_id: str
    final_status: str  # SUCCESS/PARTIAL/FAILED

为什么把 graph_definition 也存进日志？ 因为重放时必须知道"当时的 graph 长什么样"——拓扑可能在两次执行之间改了（比如某个 tool 的 schema 升级了）。只存事件而不存 graph，重放时用的是新版 graph，结果会与原版不一致。

input_snapshot 必存。这是"该节点拿什么输入执行的"的真相。否则 replay 时如果上游节点的实现变了，输入计算会不同，重放就失去意义。input_snapshot 必须是 deep copy——尤其是 LLM 输入的 messages 列表，里面嵌套的 tool_calls 引用必须深拷贝。

output 必存但可压缩。纯计算的 output 可以无损压缩（gzip）；LLM 输出如果是大段文本也可以压缩；如果包含二进制（如 tool 返回的 PDF bytes）就存 hash + 引用。

checkpoint 的时机：理想情况下每个事件都写日志，但实际生产中我们做"周期性 checkpoint + 关键事件 checkpoint"：

• 周期性：每 N 个事件或每 T 秒做一次 snapshot，snapshot 包含当前所有节点状态。snapshot 后的事件可以从 snapshot 恢复，避免从头 replay。
• 关键事件：EngineStart/EngineEnd、每个 NodeStart/Success/Fail 都必须写。
• 关键节点：有副作用的节点（写数据库、发邮件）的 SuccessEvent 必须是 fsync 才能继续，避免崩溃后状态不一致。

resume from checkpoint 的流程：

async def resume(checkpoint_id: str) -> TaskGraph:
    checkpoint = await self.event_log.load_checkpoint(checkpoint_id)
    events_after = await self.event_log.events_after(checkpoint_id)
    
    # 重建 graph 状态
    graph = checkpoint.graph  # 从 snapshot 反序列化
    for event in events_after:
        graph.apply_event(event)  # 把事件 replay 到 graph 上
    
    # 找到当前应该 ready 但还没跑的节点，继续执行
    return await ExecutionEngine(graph).run_from_state()

关键工程取舍：resume 是"接着跑"而不是"从头跑"。但生产中我们发现，"接着跑"经常导致状态不一致——因为节点的非确定性（如 LLM temperature > 0）使得"接着跑"和"从头跑"的结果不同。生产实践：critical 任务宁可从头 replay，也不要 resume；non-critical 任务可以 resume 但必须在结果里标记 "resumed from checkpoint X" 以便审计。

序列化格式：我们用 protobuf 而非 JSON。三个理由：1) protobuf 有 schema 校验，graph 改版时可以 forward/backward compatibility；2) protobuf 序列化/反序列化比 JSON 快 3-5 倍；3) protobuf 的 binary 形式天然支持 deep copy 的 byte-equality 比较。

六、replay 与 time-travel debugging：把 agent 变成可调试系统

可重放的最终价值是time-travel debugging——像调试器一样"回到过去看看某一步发生了什么"。

replay 模式分三类：

1. Full replay：用同一份事件日志，从头跑一遍整个 graph。前提是 graph 的 op 实现是 deterministic 的（同输入同输出）。注意：LLM 默认不是 deterministic 的（temperature > 0）。要做 full replay，必须用 temperature=0 或者 stub 掉 LLM call（用原事件的 output 替换）。
2. Partial replay：从某个 checkpoint 开始 replay（用 checkpoint snapshot + 之后的事件）。比 full replay 快，适合"只想重放后半段"的场景。
3. Stub replay：把所有 LLM call 和 tool call stub 掉，用原事件日志里的 output 替代。这是确定性最强的模式——给定事件日志，输出完全确定。

Stub replay 的工程实现：

class StubReplayEngine(ExecutionEngine):
    def __init__(self, graph: TaskGraph, event_log: EventLog):
        super().__init__(graph)
        self.event_log = event_log
        self.stub_outputs = self._index_outputs(event_log)
    
    async def _run_node(self, node: TaskNode):
        # 找原事件的 output
        original_output = self.stub_outputs.get(node.id)
        if original_output is None:
            raise ReplayError(f"no stub for node {node.id}")
        
        node.result = original_output
        node.status = NodeStatus.SUCCESS
        # 不真的调 op，直接用 stub 输出
        await self.event_log.append(StubReplayEvent(node_id=node.id))

为什么 stub replay 是 SRE 排查的杀手锏？ 因为 production 问题的复现路径往往是"同样的输入在生产环境跑出奇怪结果，本地怎么都复现不出来"。stub replay 让 SRE 可以把生产的事件日志下载到本地，然后用 stub 模式跑——所有外部依赖（LLM API、tool、DB）都被 stub 掉，只有调度逻辑是真实的。如果 stub replay 能复现问题，那就是调度逻辑 bug；如果不能，就是外部依赖问题。两种情况的修复路径完全不同。

time-travel debugger 的 UX：我们给 SRE 提供一个 CLI：

$ agent-debug replay --run-id=abc123 --from-node=node_5 --mode=stub
[10:23:01] node_5 SUCCESS (stub from event 245)
[10:23:01] node_6 SUCCESS (stub from event 246)
[10:23:01] node_7 SUCCESS (stub from event 247)
[10:23:01] node_7 had downstream error in original run, see event 248

调试器可以"跳到"任意节点，查看它的 input_snapshot、output、依赖链、所有相关事件。这是传统 log-based debugging 给不了的体验。

LLM determinism 挑战：即使 temperature=0，LLM 在不同时间点的输出也可能不同（模型服务升级、API 路径变化）。我们的工程实践：critical replay 必须额外 stub LLM，禁止假设 LLM 是 deterministic 的。这是反复踩过的坑——某次复盘一个 production bug，我们用 temperature=0 replay，结果与原版不一致，最后发现是 LLM 服务商在前一天升级了模型版本。

七、production 实战踩坑：7 类典型失败与防御

把 DAG 化落地到生产，我们总结了 7 类最常见的失败模式：

1. 死锁（Diamond 反模式）。最常见的死锁是"两个节点互相等待对方的输出"。表面上看拓扑是 DAG，但运行时因为条件分支把 diamond 折叠成环。防御：在调度器启动前，跑一次"反环检测 + 反自依赖检测"；运行时如果检测到 _ready_nodes() = [] and _has_running() = False，立即报 DEADLOCK 并列出所有 PENDING 节点及其 deps。

2. 级联超时（雪崩）。一个上游节点超时，它的下游节点全部取消，但因为状态机设计 bug，下游节点没有正确处理 CancelledError，导致资源（数据库连接、临时文件）泄漏。防御：所有节点的 op 必须用 try/finally 或 async context manager 包装资源；单元测试必须包含"上游 cancel 后下游资源是否回收"的检查。

3. 状态漂移（idempotency 失配）。节点 A 发邮件后 SUCCESS，但 retry 时实际又发了一次（因为 idempotency_key 没正确传递到下游服务）。防御：每个有副作用的节点必须有 idempotency_key，downstream service 必须检查；event log 里记录 idempotency_key + 下游的 ack。

4. Checkpoint 膨胀。事件日志没清理，几个月后单个 run 的事件日志达到 GB 级别，重放一次要 10 分钟。防御：定期归档旧 run（保留 summary，丢弃 detailed events）；或者按重要程度分层（critical run 保留详细事件，routine run 只保留 summary）。

5. Replay 不确定（LLM 非确定性）。前面提过的：LLM 默认不是 deterministic 的，replay 结果可能与原版不同。防御：所有 replay 必须用 stub mode；不允许"裸 replay"（不 stub LLM）。

6. Tool side-effect 重放。即使节点本身是 idempotent，tool 的副作用可能不是。比如"调用外部 API 退款"——节点代码检查了 idempotency_key，但下游退款 API 不支持 idempotency，导致重放时实际退了两次款。防御：external tool 必须独立做 idempotency（用 order_id 等业务键）；事件日志里记录 tool call 的外部 ID 而不仅是节点 ID。

7. Diamond 的输入合并 bug。fan-in 节点的 op 假设了"上游一定按顺序 a/b/c 完成"，但实际可能 a/b/c 完成的顺序是 b/a/c，op 把 b 当成 a 处理。防御：fan-in 节点的 op 必须按 port 读取输入，不能依赖上游完成的顺序；port 名字必须是稳定的（不要用 "result_0"、"result_1" 这种 index-based 命名）。

踩坑的整体教训——"实现 DAG 不难，难的是把 DAG 化思维贯彻到团队每一层代码"。很多团队的失败不是因为调度器本身有 bug，而是因为某个工程师写了"图外的逻辑"（绕过调度器直接调 op），破坏了 event sourcing 的完整性。我们的工程纪律是：任何节点 op 不能 import event_log 之外的副作用通道——所有外部副作用必须通过 event_log 记录，否则视为违反 event sourcing 纪律，code review 直接 reject。

八、与现有框架对比：LangGraph / Temporal / CrewAI / AutoGen 的工程取舍

这一节我们把上面的工程模式映射到 4 个主流框架，给出选型建议。

LangGraph（LangChain 生态）：把 graph 定义为 StateGraph + Nodes + Edges。DAG 支持较好，有 built-in 的 conditional edges 和循环（cycle）。Checkpoint 方面，LangGraph v0.2+ 引入了 CheckpointSaver，可以存到内存、SQLite、Postgres。Replay 能力相对弱——主要靠 thread_id 重放，但不支持 stub mode（直接重放会真的再调 LLM）。工程取舍：适合中等复杂度的 agent（20-50 个节点），需要快速原型；不适合需要 stub replay 的高 SLA 场景。

Temporal（workflow-as-code 平台）：工业级的 workflow 引擎，比我们的 task graph 重得多。优势：强大的 versioning、namespace、worker fleet 管理、跨语言的 workflow（Go/Python/TypeScript/Java）。劣势：引入 Temporal cluster 的运维成本；对 LLM 的非确定性没有原生支持（需要自己写 activity 来 stub LLM）。工程取舍：适合企业级、跨团队、需要长跑工作流（天/周级）的场景；不适合个人项目或快速迭代。

CrewAI：把 agent 建模为"角色（role）+ 任务（task）+ 团队（crew）"。底层是 sequential 为主，虽然支持 hierarchical 但不是真正的 DAG。Checkpoint 和 replay 能力较弱。工程取舍：适合"角色扮演 + 任务链"的场景（研究、内容生成）；不适合需要复杂拓扑或 replay 的场景。

AutoGen（Microsoft）：基于 actor model（ConversableAgent + GroupChat）。天然的并发支持（多个 agent 并行对话），但缺乏显式 DAG——拓扑是"对话流"而非"任务图"。Replay 能力有限。工程取舍：适合 multi-agent 协作的研究场景；不适合需要明确拓扑控制的工程化场景。

OpenAI Agents SDK / Claude Agent SDK：相对较新（2025-2026），主打"轻量 + handoff + guardrails"。DAG 支持较弱，更接近 linear chain 加 handoff 分支。适合简单场景；复杂拓扑需要自己包装。

总体建议：

• 小项目（< 10 节点，单人开发）：直接用 LangGraph 或裸写（参考我们第四节的执行引擎伪代码，~300 行 Python）。
• 中等项目（10-50 节点，需要 replay 和 checkpoint）：用 LangGraph + 自己加 stub replay 层，或者直接用我们这套 task graph 模式。
• 大项目（> 50 节点，跨团队，需要 enterprise 特性）：用 Temporal，把 LLM 调用包装成 activity，自己实现 stub replay。
• Multi-agent 协作（research 场景）：用 AutoGen 或 CrewAI，但接受"放弃严格 DAG 控制"的 trade-off。
• 生产高 SLA + 需要 time-travel debugging：必须用 event sourcing + stub replay（我们这套模式或 Temporal 的 event history）。

对 LLM 非确定性的处理 是选型的关键分水岭。如果你需要"同样的输入产生同样的输出"（用于测试、A/B、debug），就必须有 stub replay 能力——LangGraph/CrewAI/AutoGen 都不原生支持，需要自己包装。Temporal 通过 worker 的 versioning 和 activity retry policy 间接支持，但配置较复杂。我们的工程经验：除非项目特别小，否则event sourcing + stub replay 是必选项，不是可选项。

九、给 SRE 的可观测性清单：4 类必采集信号

最后给 SRE 一份可观测性清单。我们把 agent 的可观测性信号分成 4 类，每类都有具体的指标和告警规则。

第一类：DAG 结构 metrics。这些指标反映整个工作流的健康度。

• graph_run_total{status="success|partial|failed"}：工作流总次数，按状态分。partial 是我们引入的状态——表示部分节点 SUCCESS 部分 FAILED 且未触发整体重试。partial 率应 < 5%；过高说明 retry policy 太激进或节点太脆弱。
• graph_run_duration_seconds{quantile="0.5|0.95|0.99"}：工作流总延迟分位数。P95 应该稳定在 SLA 目标内；如果 P95 突增 2 倍以上，立即告警。
• node_count{graph_id}：每个 graph 的节点数。超过 100 的 graph 标记为 "complex"，需要单独 review。

第二类：per-node latency metrics。

• node_duration_seconds{node_id, status, quantile}：单个节点的执行延迟分位数，按状态分（success vs failed）。failed 节点通常更快（fail fast），如果 success 节点 P95 突增，往往是上游 LLM 服务降级。
• node_retry_total{node_id}：节点重试次数。某个节点 retry 率 > 30% 是危险信号——要么 retry policy 太宽松，要么节点本身不稳。
• node_dependency_wait_seconds{node_id, quantile}：节点等待上游完成的耗时。这是诊断"为什么这个节点这么慢"的关键——很多时候不是 op 慢，是上游慢。P95 wait > 5s 通常意味着上游是 LLM call 阻塞了。

第三类：replay hit rate。这是我们独有的指标。

• replay_run_total{mode="full|partial|stub"}：replay 次数按模式分。stub 应该占 > 80%（说明大多数 debug 用 stub replay 而非真的再跑一遍）。
• replay_outcome_match_total{mode, matched="true|false"}：replay 结果与原版是否一致。matched=false 必须告警——这意味着我们的 stub 不完整，或者 graph 在 replay 期间被改了。
• event_log_size_bytes{run_id, quantile}：单个 run 的事件日志大小。P95 > 100MB 告警。

第四类：LLM determinism score。这是我们为了监测 LLM 服务稳定性引入的。

• llm_determinism_score{provider, model}：在 stub 模式下，相同输入的 LLM 输出与原版的相似度（用 embedding cosine 或简单的 normalized edit distance）。理想是 1.0；< 0.95 告警。
• llm_version_drift{provider, model}：LLM 服务的版本号。版本变化时（如 OpenAI 升级 gpt-4o-2024-08 到 gpt-4o-2024-12），determinism score 通常会下降，记录版本变化的时间点和 drift 大小。

告警规则示例（PromQL）：

- alert: AgentGraphFailureSpike
  expr: rate(graph_run_total{status="failed"}[5m]) > 0.1
  for: 2m
  annotations:
    summary: "Agent 工作流失败率突增：{{ $value }}"

- alert: AgentNodeRetryAbuse
  expr: node_retry_total / node_attempt_total > 0.3
  for: 5m
  annotations:
    summary: "节点 {{ $labels.node_id }} retry 率过高"

- alert: AgentLLMDrift
  expr: llm_determinism_score < 0.95
  for: 10m
  annotations:
    summary: "LLM {{ $labels.model }} 输出 drift，可能版本变化"

最后一点——trace 与 metric 的关联。可观测性最强的能力是把"一个 graph run"的所有事件、metrics、logs 关联起来。我们用 run_id 作为统一标识：所有 metric 打 run_id label，所有 log 行带 run_id，所有 event 属于某个 run。这样 SRE 拿到一个失败的 run_id，可以一键跳转到它的所有 timeline、所有节点、所有 LLM call、所有 tool call。这是 Datadog APM、OpenTelemetry 都支持的 pattern，但要工程纪律——每个 metric/log/event 必须强制带 run_id，不能跳。

至此，我们完整展开了 agent 任务 DAG 化的工程全景。从拓扑形式化、执行引擎、event sourcing、replay、踩坑、框架对比、到可观测性，我们建立了一套完整的方法论。DAG 化的最大价值不是性能提升，而是把 agent 从"难以复现的黑盒"变成"可调试、可重放、可观测的系统"。这是 LLM agent 从 demo 走向 production 的必经之路。

参考文献

1. Lamport, L. (1978). Time, Clocks, and the Ordering of Events in a Distributed System. Communications of the ACM, 21(7), 558-565.
2. Hewitt, C., Bishop, P., & Steiger, R. (1973). A Universal Modular ACTOR Formalism for Artificial Intelligence. IJCAI.
3. Marz, N. (2013). Storm: Distributed and Fault-Tolerant Realtime Computation. Twitter Engineering Blog.
4. Thusoo, A., et al. (2010). Hive: A Warehousing Solution Over a Map-Reduce Framework. VLDB.
5. Kulkarni, S., et al. (2011). Twitter Heron: Stream Processing at Scale. SIGMOD.
6. Carbone, P., et al. (2015). Apache Flink: Stream and Batch Processing in a Single Engine. IEEE Data Eng. Bull.
7. Bergmann, G., et al. (2014). An Introduction to Acumulo: The Storage Layer of Apache Flink. Datenbanksysteme für Business, Technologie und Web.
8. Akidau, T., et al. (2015). The Dataflow Model: A Practical Approach to Balancing Correctness, Latency, and Cost. VLDB.
9. Foundation, A. S. (2024). Apache Airflow Documentation. https://airflow.apache.org/docs/
10. Tech, T. (2024). Temporal Workflow Documentation. https://temporal.io/blog/temporal-explained
11. Cadence Team, U. (2024). Cadence Documentation. https://cadenceworkflow.io/
12. Chase, N. (2024). LangGraph Documentation. https://langchain-ai.github.io/langgraph/
13. Microsoft Research. (2024). AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation. https://arxiv.org/abs/2308.08155
14. CrewAI Inc. (2024). CrewAI Documentation. https://docs.crewai.com/
15. OpenAI. (2025). Agents SDK Documentation. https://openai.github.io/openai-agents-python/
16. Anthropic. (2025). Claude Agent SDK Overview. https://docs.anthropic.com/en/docs/agents-sdk/overview
17. Fowler, M. (2005). Event Sourcing. https://martinfowler.com/eaaDev/EventSourcing.html
18. Vernon, V. (2013). Implementing Domain-Driven Design. Addison-Wesley, Chapter 8 (Event Sourcing).

一句话摘要：把 agent 任务编排从线性 chain 升级为 DAG，配合 event sourcing 与 stub replay，把 LLM agent 从难以复现的黑盒变成可调试、可重放、可观测的工程系统——这是 LLM agent 从 demo 走向 production 的必经之路。