工作流编排：画布背后的那套内核

"""一个 120 行的工作流引擎：把画布上的节点和连线跑起来。

所有可视化编排平台的内核都是这套东西——节点、边、拓扑排序、
一份在节点之间传递的变量表。看懂这个文件，你就看懂了画布背后
到底发生了什么，换任何平台都不用重学。

纯标准库，直接 python3 dag_engine.py 就能跑。
"""

from collections import deque


class Node:
    """一个节点 = 一个名字 + 一个函数 + 声明它要读哪些变量、写哪个变量。

    平台上那些「开始节点」「大模型节点」「代码节点」，
    差别只在 fn 里干了什么，结构上完全一样。
    """

    def __init__(self, name, fn, reads, writes):
        self.name = name
        self.fn = fn
        self.reads = reads      # 依赖的变量名列表
        self.writes = writes    # 产出的变量名

    def run(self, state):
        # 只把声明过的变量交给节点，防止节点偷偷依赖没声明的东西
        kwargs = {k: state[k] for k in self.reads}
        return self.fn(**kwargs)


class Workflow:
    def __init__(self):
        self.nodes = {}
        self.producer = {}   # 变量名 -> 产出它的节点名

    def add(self, node):
        if node.writes in self.producer:
            raise ValueError(
                "变量 %s 被两个节点写入：%s 和 %s —— "
                "画布上最难查的 bug 就是这种覆盖"
                % (node.writes, self.producer[node.writes], node.name))
        self.nodes[node.name] = node
        self.producer[node.writes] = node.name
        return self

    def _edges(self):
        """边不用手画：一个节点读了谁写的变量，边就自动存在。"""
        deps = {name: set() for name in self.nodes}
        for name, node in self.nodes.items():
            for var in node.reads:
                if var in self.producer:
                    deps[name].add(self.producer[var])
        return deps

    def topo_order(self):
        """拓扑排序：算出节点的执行顺序，顺便检测环。"""
        deps = self._edges()
        indeg = {n: len(d) for n, d in deps.items()}
        # 反向表：某节点跑完后，谁的入度可以减一
        downstream = {n: [] for n in self.nodes}
        for n, d in deps.items():
            for parent in d:
                downstream[parent].append(n)

        q = deque(sorted(n for n, k in indeg.items() if k == 0))
        order = []
        while q:
            n = q.popleft()
            order.append(n)
            for child in sorted(downstream[n]):
                indeg[child] -= 1
                if indeg[child] == 0:
                    q.append(child)

        if len(order) != len(self.nodes):
            stuck = sorted(set(self.nodes) - set(order))
            raise ValueError(
                "工作流里有环，这些节点永远轮不到执行：%s" % stuck)
        return order

    def parallel_layers(self):
        """按层分组：同一层的节点互不依赖，可以并行跑。"""
        deps = self._edges()
        done, layers = set(), []
        while len(done) < len(self.nodes):
            layer = sorted(
                n for n in self.nodes
                if n not in done and deps[n] <= done)
            if not layer:
                raise ValueError("工作流里有环，无法分层")
            layers.append(layer)
            done |= set(layer)
        return layers

    def run(self, initial_state, verbose=True):
        state = dict(initial_state)
        for name in self.topo_order():
            node = self.nodes[name]
            missing = [k for k in node.reads if k not in state]
            if missing:
                raise KeyError(
                    "节点 %s 读不到变量 %s —— "
                    "多半是连线漏了，或者上游节点改了输出名"
                    % (name, missing))
            value = node.run(state)
            state[node.writes] = value
            if verbose:
                print("  [%-10s] %s = %r" % (name, node.writes, value))
        return state


def build_demo():
    """一个真实形状的流程：分类 → 两路并行取数 → 汇总 → 生成。"""
    wf = Workflow()
    wf.add(Node("classify", lambda question: (
        "售后" if "退" in question or "坏" in question else "咨询"),
        reads=["question"], writes="intent"))
    wf.add(Node("retrieve", lambda question: (
        ["文档片段A", "文档片段B"]),
        reads=["question"], writes="docs"))
    wf.add(Node("lookup_order", lambda question, intent: (
        {"order_id": "SO-2026-0918", "status": "已发货"}
        if intent == "售后" else None),
        reads=["question", "intent"], writes="order"))
    wf.add(Node("compose", lambda docs, order, intent: (
        "意图=%s | 依据=%d 条 | 订单=%s"
        % (intent, len(docs), order["status"] if order else "无")),
        reads=["docs", "order", "intent"], writes="prompt"))
    wf.add(Node("generate", lambda prompt: "【回复】" + prompt,
        reads=["prompt"], writes="answer"))
    return wf


def main():
    wf = build_demo()

    print("执行顺序（拓扑排序）：")
    print("  " + " → ".join(wf.topo_order()))

    print("\n可并行的层：")
    for i, layer in enumerate(wf.parallel_layers(), 1):
        tag = "（可并行）" if len(layer) > 1 else ""
        print("  第%d层: %s %s" % (i, "、".join(layer), tag))

    print("\n执行过程：")
    final = wf.run({"question": "我买的杯子坏了，能退吗"})
    print("\n最终输出：%s" % final["answer"])

    # 环检测：把 generate 的输出接回 classify 的输入
    print("\n--- 故意造一个环 ---")
    bad = build_demo()
    bad.nodes["classify"].reads = ["question", "answer"]
    try:
        bad.topo_order()
    except ValueError as e:
        print("  捕获：%s" % e)


if __name__ == "__main__":
    main()

执行顺序（拓扑排序）：
  classify → retrieve → lookup_order → compose → generate

可并行的层：
  第1层: classify、retrieve （可并行）
  第2层: lookup_order 
  第3层: compose 
  第4层: generate 

执行过程：
  [classify  ] intent = '售后'
  [retrieve  ] docs = ['文档片段A', '文档片段B']
  [lookup_order] order = {'order_id': 'SO-2026-0918', 'status': '已发货'}
  [compose   ] prompt = '意图=售后 | 依据=2 条 | 订单=已发货'
  [generate  ] answer = '【回复】意图=售后 | 依据=2 条 | 订单=已发货'

最终输出：【回复】意图=售后 | 依据=2 条 | 订单=已发货

--- 故意造一个环 ---
  捕获：工作流里有环，这些节点永远轮不到执行：['classify', 'compose', 'generate', 'lookup_order']

"""变量作用域：画布上最难查的那类 bug，根源都在这里。

节点之间靠一份共享的变量表通信。这份表是全局的、可被覆盖的、
没有类型检查的 —— 三个特点凑在一起，就构成了编排里最常见的
三种故障：改名断链、同名覆盖、类型漂移。

纯标准库，直接 python3 state_scope.py 就能跑。
"""


class Scope:
    """带写入追踪的变量表：谁写的、写了几次、被谁读过。"""

    def __init__(self):
        self.values = {}
        self.written_by = {}     # 变量 -> [写它的节点, ...]
        self.read_by = {}        # 变量 -> [读它的节点, ...]

    def write(self, node, key, value):
        self.values[key] = value
        self.written_by.setdefault(key, []).append(node)

    def read(self, node, key):
        if key not in self.values:
            raise KeyError(
                "节点 %s 读不到 %s。三种可能：上游改了输出名、"
                "连线漏了、或者这个节点被排到了上游前面" % (node, key))
        self.read_by.setdefault(key, []).append(node)
        return self.values[key]

    def audit(self):
        """跑完之后做一次体检，把三类隐患捞出来。"""
        problems = []

        for key, writers in self.written_by.items():
            if len(writers) > 1:
                problems.append(
                    ("同名覆盖", key,
                     "被 %s 先后写入，后写的把先写的盖掉了" % "、".join(writers)))

        for key in self.written_by:
            if key not in self.read_by:
                problems.append(
                    ("写了没人读", key,
                     "产出后无人使用，要么是废节点，要么是下游连线漏了"))

        return problems


def demo_type_drift():
    """类型漂移：同一个变量在不同分支里是不同类型。"""
    sc = Scope()

    # A 分支：检索节点返回列表
    sc.write("retrieve", "docs", ["片段1", "片段2"])
    n = len(sc.read("compose", "docs"))
    print("  A 分支：docs 是 list，len=%d，一切正常" % n)

    # B 分支：换了个节点，它返回的是拼好的字符串
    sc.write("retrieve_v2", "docs", "片段1\n片段2")
    v = sc.read("compose", "docs")
    print("  B 分支：docs 变成 str，len=%d —— 数字还在，含义已经错了"
          % len(v))
    print("  ⚠ 这类 bug 不报错，只是答案悄悄变差，最难查")


def demo_rename_break():
    """改名断链：上游改了输出名，下游还在读旧名字。"""
    sc = Scope()
    sc.write("retrieve", "chunks", ["片段1"])   # 原来叫 docs，改成了 chunks
    try:
        sc.read("compose", "docs")
    except KeyError as e:
        print("  捕获：%s" % e)


def demo_overwrite():
    """同名覆盖：两个节点都往同一个变量里写。"""
    sc = Scope()
    sc.write("lookup_order", "result", {"status": "已发货"})
    sc.write("lookup_ticket", "result", {"ticket": "T-001"})
    got = sc.read("compose", "result")
    print("  compose 拿到的是 %r —— 订单信息已经被工单覆盖掉了" % got)
    sc.write("dead_node", "unused", 42)
    return sc


def main():
    print("【故障一】类型漂移")
    demo_type_drift()

    print("\n【故障二】改名断链")
    demo_rename_break()

    print("\n【故障三】同名覆盖")
    sc = demo_overwrite()

    print("\n作用域体检：")
    for kind, key, why in sc.audit():
        print("  ❌ %-8s %-10s %s" % (kind, key, why))

    print("\n三条防线：")
    print("  1. 变量名带节点前缀，如 retrieve_docs，天然不会撞名")
    print("  2. 每个节点的输出只写一个变量，别让一个节点写多处")
    print("  3. 上线前跑一次体检，把「写了没人读」的节点清掉")


if __name__ == "__main__":
    main()

【故障一】类型漂移
  A 分支：docs 是 list，len=2，一切正常
  B 分支：docs 变成 str，len=7 —— 数字还在，含义已经错了
  ⚠ 这类 bug 不报错，只是答案悄悄变差，最难查

【故障二】改名断链
  捕获：'节点 compose 读不到 docs。三种可能：上游改了输出名、连线漏了、或者这个节点被排到了上游前面'

【故障三】同名覆盖
  compose 拿到的是 {'ticket': 'T-001'} —— 订单信息已经被工单覆盖掉了

作用域体检：
  ❌ 同名覆盖     result     被 lookup_order、lookup_ticket 先后写入，后写的把先写的盖掉了
  ❌ 写了没人读    unused     产出后无人使用，要么是废节点，要么是下游连线漏了

三条防线：
  1. 变量名带节点前缀，如 retrieve_docs，天然不会撞名
  2. 每个节点的输出只写一个变量，别让一个节点写多处
  3. 上线前跑一次体检，把「写了没人读」的节点清掉

"""节点失败了怎么办：重试、超时、降级三件套。

画布上每个调外部服务的节点都会失败 —— 模型超时、接口限流、
网络抖动。平台一般在节点设置里提供这三个开关，但开关背后的
策略必须你自己想清楚，否则要么雪上加霜，要么整条流程挂掉。

纯标准库，直接 python3 retry_fallback.py 就能跑。
"""

import random
import time

random.seed(20260918)   # 固定种子，保证输出可复现


class NodeFailed(Exception):
    pass


def backoff_delays(max_retries, base=0.4, cap=8.0, jitter=True):
    """指数退避 + 抖动，返回每次重试前要等的秒数。

    不加抖动时，被同一次故障打挂的所有请求会在同一时刻一起重试，
    把刚缓过来的下游再次打死 —— 这叫重试风暴。
    """
    delays = []
    for i in range(max_retries):
        d = min(base * (2 ** i), cap)
        if jitter:
            d = random.uniform(d * 0.5, d)
        delays.append(round(d, 2))
    return delays


# 哪些错误值得重试：只有「再试一次可能就好了」的才重试
RETRYABLE = {"timeout", "rate_limited", "upstream_5xx"}
NON_RETRYABLE = {"bad_request", "auth_failed", "content_filtered"}


def call_with_retry(node_name, fn, max_retries=3, sleep=False):
    """带重试的节点调用。返回 (成功?, 结果或错误, 尝试次数)。"""
    delays = backoff_delays(max_retries)
    for attempt in range(max_retries + 1):
        try:
            return True, fn(attempt), attempt + 1
        except NodeFailed as e:
            kind = str(e)
            if kind in NON_RETRYABLE:
                print("    第%d次失败(%s)：不可重试，立即放弃"
                      % (attempt + 1, kind))
                return False, kind, attempt + 1
            if attempt == max_retries:
                print("    第%d次失败(%s)：重试用尽" % (attempt + 1, kind))
                return False, kind, attempt + 1
            d = delays[attempt]
            print("    第%d次失败(%s)：等待 %.2fs 后重试"
                  % (attempt + 1, kind, d))
            if sleep:
                time.sleep(d)
    return False, "unknown", max_retries + 1


def make_flaky(fail_times, kind="timeout"):
    """造一个前 N 次必失败、之后成功的假节点。"""
    def fn(attempt):
        if attempt < fail_times:
            raise NodeFailed(kind)
        return "第%d次成功拿到结果" % (attempt + 1)
    return fn


def run_with_fallback(node_name, primary, fallback_value, max_retries=3):
    """主路径失败后走降级：宁可给一个保守答案，也别整条流程挂掉。"""
    print("  节点 %s：" % node_name)
    ok, result, tries = call_with_retry(node_name, primary, max_retries)
    if ok:
        print("    ✅ %s（共 %d 次尝试）" % (result, tries))
        return result
    print("    ⚠ 主路径失败(%s)，降级为：%r" % (result, fallback_value))
    return fallback_value


def main():
    print("退避序列（base=0.4s，上限 8s，带抖动）：")
    print("  %s" % backoff_delays(5))
    print("  不带抖动：%s  <- 所有请求会在同一刻齐射"
          % backoff_delays(5, jitter=False))

    print("\n【场景一】抖动两次后恢复")
    run_with_fallback("call_model", make_flaky(2), fallback_value=None)

    print("\n【场景二】一直超时，走降级")
    run_with_fallback("search_web", make_flaky(99),
                      fallback_value="（本次未取到外部资料）")

    print("\n【场景三】参数写错了，重试毫无意义")
    run_with_fallback("call_api", make_flaky(99, "bad_request"),
                      fallback_value="（接口调用被跳过）")

    print("\n三条判断依据：")
    print("  • 可重试：%s" % "、".join(sorted(RETRYABLE)))
    print("  • 不可重试：%s" % "、".join(sorted(NON_RETRYABLE)))
    print("  • 降级值必须是业务能接受的保守答案，不是随便填个空字符串")


if __name__ == "__main__":
    main()

退避序列（base=0.4s，上限 8s，带抖动）：
  [0.38, 0.75, 1.29, 3.2, 6.13]
  不带抖动：[0.4, 0.8, 1.6, 3.2, 6.4]  <- 所有请求会在同一刻齐射

【场景一】抖动两次后恢复
  节点 call_model：
    第1次失败(timeout)：等待 0.27s 后重试
    第2次失败(timeout)：等待 0.72s 后重试
    ✅ 第3次成功拿到结果（共 3 次尝试）

【场景二】一直超时，走降级
  节点 search_web：
    第1次失败(timeout)：等待 0.23s 后重试
    第2次失败(timeout)：等待 0.73s 后重试
    第3次失败(timeout)：等待 0.89s 后重试
    第4次失败(timeout)：重试用尽
    ⚠ 主路径失败(timeout)，降级为：'（本次未取到外部资料）'

【场景三】参数写错了，重试毫无意义
  节点 call_api：
    第1次失败(bad_request)：不可重试，立即放弃
    ⚠ 主路径失败(bad_request)，降级为：'（接口调用被跳过）'

三条判断依据：
  • 可重试：rate_limited、timeout、upstream_5xx
  • 不可重试：auth_failed、bad_request、content_filtered
  • 降级值必须是业务能接受的保守答案，不是随便填个空字符串

"""并行与聚合：扇出去、收回来，以及收不齐时怎么办。

画布上两个节点并排画着，不等于它们真的同时跑。真正决定能不能
并行的是依赖关系 —— 谁也不读谁的输出，才谈得上并行。

难点从来不在「怎么扇出去」，而在「收回来时有一路失败/超时」。
纯标准库（concurrent.futures），直接 python3 parallel_gather.py 就能跑。
"""

import random
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

random.seed(918)


def branch(name, cost, fail=False):
    """造一个耗时 cost 秒的分支任务。"""
    def fn():
        time.sleep(cost)
        if fail:
            raise RuntimeError("%s 调用失败" % name)
        return "%s 的结果" % name
    fn.__name__ = name
    return fn


def gather_all_or_nothing(tasks, timeout):
    """严格模式：任一路失败，整体失败。

    适合「缺了就不能答」的场景，比如必须拿到订单状态才能回复。
    """
    results, errors = {}, {}
    with ThreadPoolExecutor(max_workers=len(tasks)) as pool:
        futs = {pool.submit(fn): name for name, fn in tasks.items()}
        try:
            for fut in as_completed(futs, timeout=timeout):
                name = futs[fut]
                try:
                    results[name] = fut.result()
                except Exception as e:
                    errors[name] = str(e)
        except TimeoutError:
            for fut, name in futs.items():
                if name not in results and name not in errors:
                    errors[name] = "超时未返回"
    ok = not errors
    return ok, results, errors


def gather_best_effort(tasks, timeout, required):
    """尽力模式：关键路必须成功，其余缺了就标注为空。

    适合「多一份资料更好，少一份也能答」的场景。
    """
    ok, results, errors = gather_all_or_nothing(tasks, timeout)
    missing_required = [k for k in required if k not in results]
    usable = not missing_required
    return usable, results, errors, missing_required


def show(title, usable, results, errors, missing=None):
    print("\n【%s】" % title)
    for k, v in sorted(results.items()):
        print("  ✅ %-12s %s" % (k, v))
    for k, v in sorted(errors.items()):
        print("  ❌ %-12s %s" % (k, v))
    if missing:
        print("  ⛔ 关键路缺失：%s" % "、".join(missing))
    print("  → 本次%s继续往下走" % ("可以" if usable else "不能"))


def main():
    # 三路并行：检索最快，订单中等，网页搜索最慢
    tasks = {
        "retrieve": branch("retrieve", 0.20),
        "order": branch("order", 0.35),
        "websearch": branch("websearch", 0.55),
    }

    t0 = time.time()
    ok, res, err = gather_all_or_nothing(tasks, timeout=2.0)
    elapsed = time.time() - t0
    show("三路全成功（严格模式）", ok, res, err)
    print("  耗时 %.2fs —— 串行要 %.2fs，并行只花最慢那一路的时间"
          % (elapsed, 0.20 + 0.35 + 0.55))

    # 网页搜索挂了
    tasks2 = dict(tasks)
    tasks2["websearch"] = branch("websearch", 0.30, fail=True)

    ok, res, err = gather_all_or_nothing(tasks2, timeout=2.0)
    show("一路失败（严格模式）", ok, res, err)

    usable, res, err, missing = gather_best_effort(
        tasks2, timeout=2.0, required=["retrieve", "order"])
    show("一路失败（尽力模式，关键路=retrieve+order）",
         usable, res, err, missing)

    # 关键路自己挂了
    tasks3 = dict(tasks)
    tasks3["order"] = branch("order", 0.20, fail=True)
    usable, res, err, missing = gather_best_effort(
        tasks3, timeout=2.0, required=["retrieve", "order"])
    show("关键路失败（尽力模式）", usable, res, err, missing)

    print("\n收口的三个决定：")
    print("  1. 超时按「最慢那一路」设，不是按各路之和")
    print("  2. 先划出关键路，其余走尽力模式，别让可选项拖垮整条流程")
    print("  3. 缺了的分支要在下游提示词里显式标注，别让模型以为查过了")


if __name__ == "__main__":
    main()

【三路全成功（严格模式）】
  ✅ order        order 的结果
  ✅ retrieve     retrieve 的结果
  ✅ websearch    websearch 的结果
  → 本次可以继续往下走
  耗时 0.55s —— 串行要 1.10s，并行只花最慢那一路的时间

【一路失败（严格模式）】
  ✅ order        order 的结果
  ✅ retrieve     retrieve 的结果
  ❌ websearch    websearch 调用失败
  → 本次不能继续往下走

【一路失败（尽力模式，关键路=retrieve+order）】
  ✅ order        order 的结果
  ✅ retrieve     retrieve 的结果
  ❌ websearch    websearch 调用失败
  → 本次可以继续往下走

【关键路失败（尽力模式）】
  ✅ retrieve     retrieve 的结果
  ✅ websearch    websearch 的结果
  ❌ order        order 调用失败
  ⛔ 关键路缺失：order
  → 本次不能继续往下走

收口的三个决定：
  1. 超时按「最慢那一路」设，不是按各路之和
  2. 先划出关键路，其余走尽力模式，别让可选项拖垮整条流程
  3. 缺了的分支要在下游提示词里显式标注，别让模型以为查过了

"""循环与防护：让流程能回头，但不能永远回头。

纯 DAG 不能表达「改到满意为止」这类需求，所以平台会额外提供
循环/迭代节点。代价是 DAG 的天然保证没了 —— 它不再必然终止。

所以只要画布上出现回边，就必须同时给出三道闸：最大轮次、
预算上限、无进展判定。少一道都可能跑出一条停不下来的流程。

纯标准库，直接 python3 loop_guard.py 就能跑。
"""


class LoopBudget:
    """三道闸门合在一起：轮次、成本、进展。"""

    def __init__(self, max_rounds=5, max_cost=1.0, min_gain=0.02,
                 patience=2):
        self.max_rounds = max_rounds      # 闸一：最多转几圈
        self.max_cost = max_cost          # 闸二：累计花费上限
        self.min_gain = min_gain          # 闸三：单轮最小进步
        self.patience = patience          # 连续几轮没进步就停
        self.round = 0
        self.cost = 0.0
        self.stale = 0
        self.best = None

    def should_stop(self):
        if self.round >= self.max_rounds:
            return "达到最大轮次 %d" % self.max_rounds
        if self.cost >= self.max_cost:
            return "累计成本 %.3f 触达上限 %.3f" % (self.cost, self.max_cost)
        if self.stale >= self.patience:
            return "连续 %d 轮进步不足 %.3f" % (self.stale, self.min_gain)
        return None

    def record(self, score, cost):
        """登记一轮的结果，更新三道闸的状态。"""
        self.round += 1
        self.cost += cost
        if self.best is None:
            gain = score
        else:
            gain = score - self.best
        if self.best is None or score > self.best:
            self.best = score
        if gain < self.min_gain:
            self.stale += 1
        else:
            self.stale = 0
        return gain


def refine_loop(scores, costs, **kw):
    """模拟「生成→打分→不满意就重来」的循环。

    scores/costs 是预先准备好的每轮结果，保证输出可复现。
    """
    budget = LoopBudget(**kw)
    print("  轮次  本轮分  单轮增益  累计成本  状态")
    for score, cost in zip(scores, costs):
        stop = budget.should_stop()
        if stop:
            print("  —— 停止：%s" % stop)
            return budget
        gain = budget.record(score, cost)
        print("  %3d   %.3f   %+.3f     %.3f" %
              (budget.round, score, gain, budget.cost))
    stop = budget.should_stop()
    print("  —— 停止：%s" % (stop or "候选轮次用尽"))
    return budget


def main():
    print("【场景一】三轮就收敛，闸门没被触发")
    refine_loop([0.62, 0.78, 0.86, 0.87, 0.87],
                [0.12, 0.12, 0.12, 0.12, 0.12])

    print("\n【场景二】分数原地踏步，被「无进展」闸拦下")
    refine_loop([0.55, 0.56, 0.565, 0.57, 0.57],
                [0.12, 0.12, 0.12, 0.12, 0.12])

    print("\n【场景三】每轮都在涨，但先撞上成本闸")
    refine_loop([0.40, 0.55, 0.68, 0.79, 0.88],
                [0.30, 0.30, 0.30, 0.30, 0.30], max_cost=0.8)

    print("\n【场景四】只设了轮次闸，没设成本和进展闸")
    b = refine_loop([0.5] * 8, [0.25] * 8,
                    max_rounds=8, max_cost=999, min_gain=0.0)
    print("  结果：转满 %d 轮，花掉 %.2f，分数一点没涨" % (b.round, b.cost))
    print("  ⚠ 这就是「能跑完但白烧钱」的典型形态，靠轮次闸兜不住")

    print("\n只要画布上出现回边，三道闸缺一不可：")
    print("  闸一 最大轮次 —— 保证一定会停")
    print("  闸二 预算上限 —— 保证停之前不会烧穿")
    print("  闸三 无进展判定 —— 保证不做无用功")


if __name__ == "__main__":
    main()

【场景一】三轮就收敛，闸门没被触发
  轮次  本轮分  单轮增益  累计成本  状态
    1   0.620   +0.620     0.120
    2   0.780   +0.160     0.240
    3   0.860   +0.080     0.360
    4   0.870   +0.010     0.480
    5   0.870   +0.000     0.600
  —— 停止：达到最大轮次 5

【场景二】分数原地踏步，被「无进展」闸拦下
  轮次  本轮分  单轮增益  累计成本  状态
    1   0.550   +0.550     0.120
    2   0.560   +0.010     0.240
    3   0.565   +0.005     0.360
  —— 停止：连续 2 轮进步不足 0.020

【场景三】每轮都在涨，但先撞上成本闸
  轮次  本轮分  单轮增益  累计成本  状态
    1   0.400   +0.400     0.300
    2   0.550   +0.150     0.600
    3   0.680   +0.130     0.900
  —— 停止：累计成本 0.900 触达上限 0.800

【场景四】只设了轮次闸，没设成本和进展闸
  轮次  本轮分  单轮增益  累计成本  状态
    1   0.500   +0.500     0.250
    2   0.500   +0.000     0.500
    3   0.500   +0.000     0.750
    4   0.500   +0.000     1.000
    5   0.500   +0.000     1.250
    6   0.500   +0.000     1.500
    7   0.500   +0.000     1.750
    8   0.500   +0.000     2.000
  —— 停止：达到最大轮次 8
  结果：转满 8 轮，花掉 2.00，分数一点没涨
  ⚠ 这就是「能跑完但白烧钱」的典型形态，靠轮次闸兜不住

只要画布上出现回边，三道闸缺一不可：
  闸一 最大轮次 —— 保证一定会停
  闸二 预算上限 —— 保证停之前不会烧穿
  闸三 无进展判定 —— 保证不做无用功

"""把编排结构导出成自己能版本化的格式。

平台的画布文件是私有格式，换平台带不走。但编排的**语义**是通用的：
有哪些节点、每个节点读什么写什么、连线怎么连。把这层语义抽出来
存成自己的 JSON，进 git，就等于给流程上了保险。

这不是为了「以后一键迁移」——没有一键迁移。是为了迁移时
有一份准确的结构说明，而不是对着截图重画。

纯标准库，直接 python3 workflow_export.py 就能跑。
"""

import json
import hashlib

from dag_engine import build_demo


def export(wf):
    """把 Workflow 对象导成与平台无关的结构描述。"""
    nodes = []
    for name in wf.topo_order():
        n = wf.nodes[name]
        nodes.append({
            "id": name,
            "reads": list(n.reads),
            "writes": n.writes,
        })

    # 边由变量依赖推出来，不需要单独维护一份
    edges = []
    for n in nodes:
        for var in n["reads"]:
            src = wf.producer.get(var)
            if src:
                edges.append({"from": src, "to": n["id"], "via": var})

    doc = {
        "schema": "workflow-structure/v1",
        "nodes": nodes,
        "edges": edges,
        "layers": wf.parallel_layers(),
        "entry_vars": sorted({
            v for n in nodes for v in n["reads"]
            if v not in wf.producer
        }),
        "exit_vars": sorted({
            n["writes"] for n in nodes
            if not any(n["writes"] in m["reads"] for m in nodes)
        }),
    }
    return doc


def fingerprint(doc):
    """给结构算一个指纹，用来发现「画布被人改了但没人说」。"""
    payload = json.dumps(
        {"nodes": doc["nodes"], "edges": doc["edges"]},
        sort_keys=True, ensure_ascii=False).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()[:16]


def diff(old, new):
    """比较两份导出，列出结构层面的变化。"""
    o = {n["id"]: n for n in old["nodes"]}
    n_ = {n["id"]: n for n in new["nodes"]}

    changes = []
    for nid in sorted(set(n_) - set(o)):
        changes.append("新增节点 %s" % nid)
    for nid in sorted(set(o) - set(n_)):
        changes.append("删除节点 %s" % nid)
    for nid in sorted(set(o) & set(n_)):
        if o[nid]["reads"] != n_[nid]["reads"]:
            changes.append("节点 %s 的输入变了：%s → %s"
                           % (nid, o[nid]["reads"], n_[nid]["reads"]))
        if o[nid]["writes"] != n_[nid]["writes"]:
            changes.append("节点 %s 的输出变了：%s → %s"
                           % (nid, o[nid]["writes"], n_[nid]["writes"]))
    return changes


def main():
    wf = build_demo()
    doc = export(wf)

    print("导出的结构（节选）：")
    print(json.dumps(
        {k: doc[k] for k in ("schema", "entry_vars", "exit_vars", "layers")},
        ensure_ascii=False, indent=2))

    print("\n节点清单：")
    for n in doc["nodes"]:
        print("  %-14s reads=%-32s writes=%s"
              % (n["id"], ",".join(n["reads"]), n["writes"]))

    print("\n连线（由变量依赖自动推出，共 %d 条）：" % len(doc["edges"]))
    for e in doc["edges"]:
        print("  %-14s --%s--> %s" % (e["from"], e["via"], e["to"]))

    fp1 = fingerprint(doc)
    print("\n结构指纹：%s" % fp1)

    # 模拟有人在画布上偷偷改了一个节点的输入
    wf2 = build_demo()
    wf2.nodes["compose"].reads = ["docs", "intent"]   # 不再读 order
    doc2 = export(wf2)
    fp2 = fingerprint(doc2)

    print("\n有人改了画布之后：")
    print("  新指纹：%s（%s）" % (fp2, "一致" if fp1 == fp2 else "已变化"))
    for c in diff(doc, doc2):
        print("  • %s" % c)

    print("\n把这份 JSON 提交进 git，好处有三：")
    print("  1. 画布任何结构变动都能在代码评审里被看见")
    print("  2. 迁移时照着它重建，而不是对着截图猜")
    print("  3. 出故障时能确认「上线那天的流程到底长什么样」")


if __name__ == "__main__":
    main()

导出的结构（节选）：
{
  "schema": "workflow-structure/v1",
  "entry_vars": [
    "question"
  ],
  "exit_vars": [
    "answer"
  ],
  "layers": [
    [
      "classify",
      "retrieve"
    ],
    [
      "lookup_order"
    ],
    [
      "compose"
    ],
    [
      "generate"
    ]
  ]
}

节点清单：
  classify       reads=question                         writes=intent
  retrieve       reads=question                         writes=docs
  lookup_order   reads=question,intent                  writes=order
  compose        reads=docs,order,intent                writes=prompt
  generate       reads=prompt                           writes=answer

连线（由变量依赖自动推出，共 5 条）：
  classify       --intent--> lookup_order
  retrieve       --docs--> compose
  lookup_order   --order--> compose
  classify       --intent--> compose
  compose        --prompt--> generate

结构指纹：8be845a9d6cc2e97

有人改了画布之后：
  新指纹：cf4c557af96ed3d0（已变化）
  • 节点 compose 的输入变了：['docs', 'order', 'intent'] → ['docs', 'intent']

把这份 JSON 提交进 git，好处有三：
  1. 画布任何结构变动都能在代码评审里被看见
  2. 迁移时照着它重建，而不是对着截图猜
  3. 出故障时能确认「上线那天的流程到底长什么样」