用数学拓扑结构编排 AI 工作流

第一层：有向无环图（成熟）

1.1 工作流就是一张依赖图

数学上，这是一个有向无环图（DAG）：节点是任务，箭头是"必须先完成"。"无环"保证任务能按某种顺序全部完成。

1.2 拓扑排序：谁先谁后、谁能并行

拓扑排序给出一个满足所有依赖的执行顺序。把它按"层"展开，同一层的任务互不依赖，可以交给不同的 Agent 并行执行。

关键路径是从开始到结束耗时最长的那条依赖链，它决定了整个流程的最短完成时间。要加速整个流程，只有缩短关键路径上的任务才有用。

环意味着死锁：Agent A 在等 B 的结果，B 又在等 A 的结果。

1.3 代码

Python 标准库的 graphlib 就能完成这些：

from graphlib import TopologicalSorter, CycleError

workflow = {   # 键 = 任务，值 = 它依赖的任务
    "取数":       set(),
    "清洗":       {"取数"},
    "因子A":      {"清洗"},
    "因子B":      {"清洗"},
    "因果检验":   {"因子A", "因子B"},
    "回测":       {"因子A", "因子B"},
    "风控审查":   {"回测", "因果检验"},
    "报告":       {"风控审查"},
}
cost = {"取数": 3, "清洗": 2, "因子A": 5, "因子B": 8, "因果检验": 4, "回测": 6, "风控审查": 2, "报告": 1}

ts = TopologicalSorter(workflow); ts.prepare()
layer = 0
while ts.is_active():
    ready = list(ts.get_ready())               # 这一批互不依赖，可以并行交给不同 Agent
    print(f"第 {layer} 层可并行:", ready)
    ts.done(*ready); layer += 1

order = list(TopologicalSorter(workflow).static_order())
finish, prev = {}, {}
for t in order:
    deps = workflow[t]
    start = max((finish[d] for d in deps), default=0)
    prev[t] = max(deps, key=lambda d: finish[d]) if deps else None
    finish[t] = start + cost[t]
path, t = [], "报告"
while t: path.append(t); t = prev[t]
print("关键路径:", " → ".join(reversed(path)), " 总耗时:", finish["报告"])

try:
    list(TopologicalSorter({"A": {"B"}, "B": {"A"}}).static_order())
except CycleError as e:
    print("发现环（死锁）:", e.args[1])

输出：

第 0 层可并行: ['取数']
第 1 层可并行: ['清洗']
第 2 层可并行: ['因子A', '因子B']
第 3 层可并行: ['因果检验', '回测']
第 4 层可并行: ['风控审查']
第 5 层可并行: ['报告']
关键路径: 取数 → 清洗 → 因子B → 回测 → 风控审查 → 报告  总耗时: 22
发现环（死锁）: ['A', 'B', 'A']

（同一层内的任务顺序可能因运行而不同。）

1.4 工程意义

风控审查被设计成"回测"和"因果检验"的共同后继：结构上保证任何结果在进入报告之前都必须经过风控。这是把安全规则写进拓扑结构本身，而不是依赖 Agent 记得去做（参见《Agent交易系统的安全边界》）。

第二层：持续同调——看见 Agent 输出的"形状"（研究基础 + 构想）

2.1 问题

五个 Agent 分别独立分析同一份数据，给出结论。它们是：

• 一致的？
• 分成了几派？
• 有没有一个"离群者"？

只看平均意见会掩盖结构。拓扑数据分析（TDA）中的持续同调正好用来描述数据的"形状"在不同尺度下如何变化。

2.2 直觉

把每个 Agent 的输出（比如对一组股票的打分向量，或结论文本的嵌入向量）看作空间中的一个点。逐渐增大一个距离阈值 $\varepsilon$，把距离小于 $\varepsilon$ 的点连起来：

• $\varepsilon$ 很小时，每个点各自一团；
• $\varepsilon$ 增大，点逐渐合并成几个团；
• 某个团存活了很长的 $\varepsilon$ 区间才和别的团合并，说明它是一个稳定的结构（一个真正的"派别"）；
• 某个点在很大的 $\varepsilon$ 下才并入，说明它是离群者。

记录每个团"出生"和"消亡"的 $\varepsilon$，就得到持续条形码。0 维持续同调描述的是"连通分支"（团）；1 维描述的是"环"（比如 Agent 之间循环的、无法收敛的意见）。

2.3 代码：0 维持续同调

import numpy as np
rng = np.random.default_rng(0)
emb = np.vstack([rng.normal(0, 0.3, (5, 2)), rng.normal(3, 0.3, (5, 2)), [[8, 8]]])  # 两组共识 + 一个离群 Agent
n = len(emb); parent = list(range(n))
def find(i):
    while parent[i] != i: parent[i] = parent[parent[i]]; i = parent[i]
    return i
edges = sorted((np.linalg.norm(emb[i] - emb[j]), i, j) for i in range(n) for j in range(i + 1, n))
deaths = []
for d, i, j in edges:
    a, b = find(i), find(j)
    if a != b: parent[a] = b; deaths.append(round(float(d), 2))
print("连通分支消亡的距离（持续条形码）:", deaths)

输出：

连通分支消亡的距离（持续条形码）: [0.11, 0.17, 0.24, 0.25, 0.32, 0.36, 0.42, 0.43, 3.27, 6.76]

2.4 怎么读

• 前 8 个消亡距离都很小（0.1–0.4）：组内的 Agent 很快合并，说明组内意见一致；
• 3.27：两个派别直到距离 3.27 才合并，说明存在两个稳定的、明显不同的意见群；
• 6.76：最后一个合并的是离群 Agent。

在投研流程中，这可以变成一条规则：条形码里出现长寿命的分支时，不要直接平均，而是把分歧交给人或"仲裁 Agent"（参见《解构主义与任务流》 中"分歧是有价值的信号"）。

2.5 已有研究基础

• 持续同调由 Edelsbrunner、Carlsson 等人在 2000 年代发展起来，是拓扑数据分析的核心工具；
• 金融领域已有研究用持续同调分析股票收益的相关性网络，发现在市场危机（如 2000 年互联网泡沫破裂、2008 年金融危机）前后，拓扑特征出现显著变化，被提出可作为市场结构转变的预警信号之一；
• 将其用于多 Agent 意见结构的实时监控，是本文提出的构想，尚需实验验证。

第三层：范畴论——工作流的可组合性（理论构想）

3.1 直觉

范畴论研究的是"东西"（对象）和"东西之间的变换"（态射），以及变换如何组合。

3.2 构想中的用途

• 类型检查：在组装工作流时，自动检查"Agent X 的输出能否直接交给 Agent Y"，防止把"因子值"错接成"收益率"；
• 可替换性：两个输入输出类型相同的 Agent 可以互换，用于 A/B 测试不同模型；
• 并行组合（单子范畴中的张量积）：两条互不依赖的流水线可以"并排"组合，对应 DAG 中的并行层；
• 函数式编程中的管道、单子等概念，本身就来自范畴论，已经在软件工程中广泛使用。

3.3 诚实的评估

范畴论在这里主要提供的是思考框架和设计约束，而不是新的算法。在规模不大的工作流中，类型系统和 DAG 已经足够；当工作流变得非常大、由 Agent 动态生成时，形式化的组合规则可能更有价值。

总结