AI 工作流里的 DAG: 把一条长 prompt 拆成一张可缓存、可重试的图

上一篇云服务测试的 AI 转型里,用例生成、失败分诊都是"一句 prompt 干一件事"。任务一复杂——比如"分析日志、分类错误、排查嫌疑提交、再写报告"——直觉做法是把所有材料塞进一条长 prompt。这篇文章解释为什么更好的做法是把它拆成一张 DAG(有向无环图),并给出一个 66 行、已实际跑通验证的最小执行器。

结论先放在前面

  1. DAG 不神秘:节点 = 一次调用(LLM 或普通函数),边 = 数据依赖,按依赖拓扑顺序执行。你写的每一个"先 A 再 B 最后 C"的 shell 脚本,手工画出来就是一张 DAG;

  2. 它解决一条长 prompt 的四个具体毛病:错误无法定位、改一处要全量重跑、无依赖的步骤被迫串行、过程不可审计;

  3. 最小 demo 是一个真实问题:CI 构建失败 → 自动产出"错误分类 + 嫌疑提交"分析报告。实测:改一个节点的实现,只有那个节点重跑,其余全部命中缓存;

  4. 边界:流程已知、步骤固定的任务用 DAG;流程未知的开放式探索(比如"帮我调查这个诡异 bug")该用 Agent 循环,DAG 套不住它。

从一个真实痛点说起

任务:CI 构建失败,要产出一份分析报告——错误有哪些、各是什么类型、最可能是哪个提交引起的。

直觉做法是一条长 prompt:

阅读下面的构建日志和提交列表, 输出一份分析报告: 错误分类统计 + 嫌疑提交。
日志: ...(800 行)
提交: ...(20 条)

能用,但有四个毛病,而且随任务复杂度放大:

毛病

后果

错误无法定位

报告里"嫌疑提交"排错了,你不知道是模型读错了日志、分类错了类型,还是关联逻辑有问题——一个黑盒,里面三段推理纠缠在一起

改一处全量重跑

只想调整报告格式?整条 prompt 重新调 LLM,日志重新读、分类重新做,钱和时间全白花

无依赖也串行

"错误分类"和"排查嫌疑提交"互不依赖,一条 prompt 里它们却必须同时做——单次输出的注意力被摊薄,质量反而下降

过程不可审计

中间结论没有独立产物,事后无法复查"当时模型到底看到了什么、判断了什么"

DAG 就是逐个对症下药的结构:拆开(可定位)→ 缓存(局部重跑)→ 拓扑排序(并行)→ 中间产物落盘(可审计)。

DAG 是什么

DAG(Directed Acyclic Graph,有向无环图):节点是计算步骤,边是"前者的产物作为后者的输入"的依赖关系,且不允许成环——不允许 A 依赖 B、B 又依赖 A。

上面的 CI 分析任务画出来就是这样:

        flowchart LR
    L[构建日志] --> E[提取错误]
    L --> C[提交列表]
    E --> F[错误分类]
    E --> B[排查提交]
    C --> B
    F --> R[分析报告]
    B --> R

"错误分类"和"排查提交"是两个无依赖的分支,汇入"分析报告"。环被禁止不是洁癖,是终止性保证:执行顺序按拓扑排序展开,图无环 ⇒ 执行必然结束——这是它和 Agent 循环(while not done 那种)最本质的区别。

四种形态放一起对比:

形态

例子

优点

致命伤

单条 prompt

一次性问答

零成本上手

复杂任务四毛病俱全

线性链

prompt chaining,A 输出喂 B

步骤可复用

无并行;中途失败全丢;只能串行重试

Agent 循环

ReAct:思考→行动→观察→再思考

流程自适应,能探索

可能死循环;行为不可预测;难以复现

DAG

本文

可定位、可缓存、可并行、必终止

流程必须事先知道,不适合开放探索

实践中成熟的做法是分层:探索阶段用 Agent 循环摸清流程,固化下来的流程改成 DAG 进生产。

解决了什么问题: 四个收益逐个验证

以下数据全部来自文末 demo 的真实运行(通过给每个节点注入执行探针统计):

1. 局部重跑:改一个节点,只重跑一个节点

== 第 1 次(冷缓存) ==   提取错误 → 排查提交 → 错误分类 → 分析报告   (4 个节点全执行)
== 第 2 次(热缓存) ==   (零执行, 全部命中缓存)
== 第 3 次(只改了"分析报告"的实现) ==   (只有"分析报告"执行, 上游三个节点全部复用)

第 3 次运行后报告标题变成新版、内容数据不变——这正是"调 prompt"的日常:迭代报告模板时,上游昂贵的 LLM 调用一次都不浪费。

2. 并行:无依赖的分支同时跑

冷启动的日志里,"排查提交"先于"错误分类"打印完成——两个节点同属一批就绪节点,被线程池同时调度。分支越多,省的时间越多;这条在 LLM 节点上价值翻倍:三个 30 秒的调用并行,总时长还是 30 秒。

3. 可观测:中间产物是落盘的一等公民

每个节点的产物以 JSON 形式存在 .dag-cache/ 目录里,命名即节点名。报告出错了,先打开上游节点的产物文件,逐环检查"提取的错误行对不对→分类统计对不对",而不是盯着一坨最终输出反推。

4. 必终止 + 失败兜底

环会被执行器直接拦下(存在循环依赖, 无法继续),从机制上杜绝死循环;单个节点的瞬时失败由节点级重试兜底(demo 实测:前两次调用抛异常、第三次成功,retries=2 共尝试 3 次)。这对 LLM 节点格外重要——API 超时、限流是常态,重试应该长在框架里,不该由每个调用点自己操心。

最小 Demo: 66 行执行器 + CI 失败分析

三个文件。dag.py 是通用执行器,dag_demo.py 用真实问题搭一张四节点图,ci.log 是模拟的构建日志。

dag.py —— 拓扑排序 + 线程并行 + 节点级缓存 + 失败重试

"""DAG 执行器: 拓扑排序 + 线程并行 + 节点级缓存 + 失败重试"""
import hashlib
import inspect
import json
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path


class Dag:
    def __init__(self, workdir=".dag-cache", retries=2):
        self.nodes = {}          # 节点名 -> {fn, deps, cache}
        self.workdir = Path(workdir)
        self.retries = retries

    def node(self, name, deps=(), cache=True):
        """注册节点: fn 的返回值就是节点的产物"""
        def deco(fn):
            self.nodes[name] = dict(fn=fn, deps=list(deps), cache=cache)
            return fn
        return deco

    def _cache_key(self, name, spec, results):
        # 缓存键 = 节点名 + 节点代码 + 上游产物: 改代码或改输入都会自动失效
        try:
            code = inspect.getsource(spec["fn"])
        except OSError:                       # lambda / 交互式定义的函数没有源码文件
            code = repr(spec["fn"])
        payload = json.dumps([name, code, {d: results[d] for d in spec["deps"]}],
                             sort_keys=True, default=str, ensure_ascii=False)
        return hashlib.sha256(payload.encode()).hexdigest()[:16]

    def run(self, inputs):
        self.workdir.mkdir(exist_ok=True)
        results = dict(inputs)   # inputs 是"没有上游的源节点"
        pending = set(self.nodes)
        with ThreadPoolExecutor(max_workers=8) as pool:
            while pending:
                # 依赖全部就绪的节点, 本批一起跑
                ready = [n for n in pending
                         if all(d in results for d in self.nodes[n]["deps"])]
                assert ready, f"存在循环依赖, 无法继续: {pending}"
                futures = {}
                for n in ready:
                    spec = self.nodes[n]
                    cache_file = self.workdir / f"{n}-{self._cache_key(n, spec, results)}.json"
                    if spec["cache"] and cache_file.exists():
                        results[n] = json.loads(cache_file.read_text(encoding="utf-8"))
                        continue
                    futures[pool.submit(self._run_with_retry, spec["fn"],
                                        *[results[d] for d in spec["deps"]])] = (n, cache_file)
                for fut, (n, cache_file) in futures.items():
                    results[n] = fut.result()
                    cache_file.write_text(json.dumps(results[n], ensure_ascii=False),
                                          encoding="utf-8")
                pending -= set(ready)
        return results

    def _run_with_retry(self, fn, *args):
        for attempt in range(self.retries + 1):
            try:
                return fn(*args)
            except Exception:
                if attempt == self.retries:
                    raise
                time.sleep(1)

dag_demo.py —— 四节点的真实问题

"""真实问题: CI 构建失败, 自动产出一份"错误分类 + 嫌疑提交"的分析报告"""
from dag import Dag

dag = Dag()


@dag.node("提取错误", deps=["构建日志"])
def extract_errors(log: str) -> list:
    return [line for line in log.splitlines()
            if "ERROR" in line or "FAIL" in line]


@dag.node("错误分类", deps=["提取错误"])
def classify(errors: list) -> dict:
    table = {"编译错误": ["cannot find symbol", "error:"],
             "测试失败": ["AssertionError", "Tests run"],
             "部署错误": ["connection refused", "timeout"]}
    stats = {kind: 0 for kind in table}
    for line in errors:
        for kind, keywords in table.items():
            if any(k in line for k in keywords):
                stats[kind] += 1
                break
    return stats


@dag.node("排查提交", deps=["提取错误", "提交列表"])
def blame(errors: list, commits: list) -> list:
    # 最小 demo 用关键词匹配; 换成 AI 后这里是一次 LLM 调用
    hits = [c for c in commits if any(k in c for k in ("UserService", "AuthFilter", "config"))]
    return hits if errors else []


@dag.node("分析报告", deps=["错误分类", "排查提交"])
def report(stats: dict, suspects: list) -> str:
    lines = ["# CI 失败分析报告", "", "## 错误分布"]
    lines += [f"- {kind}: {count} 条" for kind, count in stats.items() if count]
    lines += ["", "## 建议优先排查的提交"] + [f"- {c}" for c in suspects]
    return "\n".join(lines)


if __name__ == "__main__":
    log = open("ci.log", encoding="utf-8").read()
    commits = ["a1b2c3 重构 UserService 的鉴权逻辑",
               "d4e5f6 升级 AuthFilter 依赖",
               "07f8a9 更新 README"]
    results = dag.run({"构建日志": log, "提交列表": commits})
    print(results["分析报告"])

ci.log —— 模拟的构建日志

[INFO] 开始构建 user-service v2.3.1
[INFO] 编译 214 个源文件
[ERROR] /src/main/java/com/acme/UserService.java:[88,23] cannot find symbol: method verifyToken()
[ERROR] /src/main/java/com/acme/AuthFilter.java:[41,9] cannot find symbol: class TokenParser
[INFO] 编译失败, 重试一次
[ERROR] 编译重试仍失败, 终止构建
[INFO] 回滚到上一个可用版本
[ERROR] connection refused: 10.0.3.15:8443, rollback timeout after 30s

运行

python3 dag_demo.py

输出:

# CI 失败分析报告

## 错误分布
- 编译错误: 2 条
- 部署错误: 1 条

## 建议优先排查的提交
- a1b2c3 重构 UserService 的鉴权逻辑
- d4e5f6 升级 AuthFilter 依赖

再把上文"四个收益"里的三次运行实验跑一遍(给节点函数加一行 print 作探针即可观察),就能亲眼看到局部重跑和并行生效。

把节点换成真正的 AI 调用

demo 里节点是普通函数,换成 LLM 调用只需一个约定:节点函数 = 一次 kimi -p 调用 + JSON 输出。

import json, subprocess

def ask_kimi(prompt: str, payload: str):
    """节点函数的统一封装: 调用 kimi, 解析 JSON 输出"""
    out = subprocess.run(
        ["kimi", "-p", f"{prompt}\n输入:\n{payload}\n只输出 JSON, 不要 markdown 代码块"],
        capture_output=True, text=True, timeout=600)
    text = out.stdout.strip()
    return json.loads(text[text.find("{"): text.rfind("}") + 1])

于是"排查提交"节点从关键词匹配升级成真正的推理:

@dag.node("排查提交", deps=["提取错误", "提交列表"])
def blame(errors, commits):
    return ask_kimi(
        "你是构建失败分析专家。结合错误日志和提交列表找出嫌疑提交, "
        "输出 {\"suspects\": [\"提交说明\"]}",
        "错误:\n" + "\n".join(errors) + "\n提交:\n" + "\n".join(commits),
    )["suspects"]

替换之后,前面验证过的四件事——局部重跑、并行、缓存、重试——原样生效,因为它们是执行器的性质,与节点内部是不是 AI 无关。这正是 DAG 作为 AI 工程结构的价值:AI 只住在节点里,图的纪律由框架保证。

警告

节点并发调 LLM 时注意服务商的限流:max_workers=8 对本地函数无所谓,对 API 可能触发 429。生产上按账户配额给线程池设上限,并善用执行器自带的重试。

DAG 其实一直在你身边

  • Airflow:数据管道的事实标准,任务依赖即 DAG——AI 工作流的 DAG 热是同一思想在 LLM 时代的重演;

  • LangGraph / Dify / Coze / n8n:把节点和边做成可视化画布,拖拽生成 AI 工作流,底层都是 DAG 执行器;

  • 本目录的 shell 脚本:结对编程的 pair.sh、"开发→提交→审查→修复"的循环、多模型辩论的 debate.sh、"提案→批评→修订"——手工写的串行流程,画出来都是两三层的小 DAG。手写的版本没有缓存、没有并行、环靠人脑检查;流程稳定后,值得用执行器固化下来。

常见坑

坑

现象

处理

缓存键不含代码版本

改了节点逻辑,重跑却拿到旧结果

缓存键必须含节点代码(如 inspect.getsource)——本文 demo 初版就踩过这个坑,已修复并写进 _cache_key

任务拆得太碎

节点间传递的数据比计算本身还大,编排开销超过收益

合并过小的节点;判断标准:每个节点是否都有独立可评审的产物

用 DAG 硬套探索性任务

流程走不通,反复回头改图

探索期用 Agent 循环,流程固化后再 DAG 化

中间产物不落盘

出错只能对着最终输出反推

节点产物一律 JSON 落盘,命名即节点名

下游不处理上游失败

一个节点挂了,整图白跑

生产级执行器要标记失败节点、跳过其下游并报告;最小版选择直接抛错

并发打爆 API 限额

批量跑 LLM 节点大面积 429

线程池按配额设限 + 重试退避

总结

  • DAG = 节点(一次调用)+ 依赖边 + 拓扑执行,它把"一条长 prompt"的四个毛病逐个拆掉:错误可定位、改动可局部重跑、无依赖可并行、无环必终止;

  • 66 行执行器已经包含生产框架的四个核心机制:拓扑调度、节点缓存、并发执行、失败重试——看懂它,再看 Airflow、LangGraph 都是同一套语言;

  • AI 只住在节点里,图的纪律由框架保证;流程未知的任务交给 Agent 循环,流程固化的任务交给 DAG;

  • 你手头那些"先 A 再 B 最后 C"的 shell 脚本,就是还没画出来的 DAG。

参考资料