books

《互联网产品的 AI 架构·多能力编排与任务闭环》7.3 回放与可复现调试

同一条失败记录，重跑三次得到三个不同结果，日志却只有一句「answer failed」。本节给出一次链路的最小回放包：请求、节点输入输出摘要、模型档位、提示版本与随机种子，用规范化序列化和逐节点对比把分歧定位到具体节点，并演示一次提示版本导致的重放差异。

Leeting Yan 8 分钟阅读 3534 字

同一条失败，重跑三次三个结果

7.2 把失败归因到节点。值班同学拿着「answer 节点模型失败」的记录去复现：在测试环境重跑同一条输入，第一次成功，第二次引用错了文档，第三次超时。三次都「合理」，但没有一次能解释线上那次为什么失败。

原因不神秘：模型档位悄悄从 standard 换成了 economy，提示从 v3 升到了 v4，采样温度是 0.7 且没有固定种子，上游检索结果也在变。任何一项变化，都足以让重跑走出不同路径。不可复现不是因为模型「玄学」，而是因为复现时没有冻结当时的环境。

回放要解决的就是这件事：把一次链路执行压缩成一个足够小、又足够完整的包，让同一个包在任何时候都能重建同一条执行路径。文中数字来自 ai-lab 的示例回放运行结果，不是生产实测。

最小回放包

回放包不是全量日志。全量日志又大又含隐私，真正决定路径的只有几个字段：

字段记录什么缺了会怎样
request能力名、租户、原始输入、截止时间无法重建首节点输入
workflowVersion编排图版本节点顺序或分支条件可能已经变了
modelTierfrontier / standard / economy换档位会直接改变输出分布
promptVersion提示工件版本提示改版是常见的静默变量
sampling温度、随机种子、输出上限没有种子只能得到「另一条合理路径」
nodes每个节点的输入摘要、输出摘要、状态、延迟无法定位从哪一步开始分歧
result结果类别、总成本、端到端延迟无法判断重放是否复现了失败

主场景的成功路径成本是 3,786 微元、延迟 1,269ms，这些值进包；高失败变体每次成功交付 4,139 微元，说明同一个包结构也能承载失败与重试成本。

只存必要信息

请求原文必须存，否则首节点无法重建；节点之间只存摘要，不存完整输入输出。摘要用 SHA-256 前 16 位十六进制，足以判断「是不是同一份内容」，又不会把合同、订单号、用户原话复制到调试系统里。

代价是摘要不能直接阅读。排查时先靠摘要定位分歧节点，再按 runId 从受控存储取该节点的短期原文。这个设计把「可复现」和「少存隐私」放在一起，而不是二选一。

规范化序列化

回放包要能比较，序列化就必须确定：对象键按字典序排列，数组保持原顺序，固定 schemaVersion，末尾保留一个换行。同样的包无论序列化多少次，字节都相同；不同的包才会产生差异。

没有规范化，两个语义相同的包会因为键顺序不同而比较失败，回放对比会退化成噪音。规范化也让包可以直接做快照测试和内容寻址存储。

实现

replay.ts 提供三件事：摘要、规范序列化/反序列化、逐字段对比。校验放在序列化与反序列化两侧，坏包不会进入对比。

// file: src/orch/replay.ts
// tier: L2
// verify: node --test test/orch/eval.test.ts

import { createHash } from "node:crypto";
import { AiError } from "../core/errors.ts";

export type ReplayOutcome = "success" | "partial" | "failure" | "timeout" | "unauthorized";
export type ModelTier = "frontier" | "standard" | "economy";
export type ReplayNodeStatus = "ok" | "failed" | "timeout" | "skipped" | "denied";

export interface ReplayRequest {
  capability: string;
  tenantId: string;
  /** 原始请求文本；回放包要能重建首节点输入，不能只存摘要。 */
  input: string;
  deadlineMs: number;
}

export interface ReplayNode {
  nodeId: string;
  inputDigest: string;
  outputDigest: string;
  status: ReplayNodeStatus;
  latencyMs: number;
}

export interface ReplaySampling {
  temperature: number;
  /** 固定随机种子；没有它，重放只能得到“另一条合理路径”。 */
  seed: number;
  maxOutputTokens: number;
}

/** 一条链路的最小回放包：请求、节点输入输出、模型档位、提示版本、采样参数。 */
export interface ReplayPackage {
  schemaVersion: 1;
  runId: string;
  workflow: string;
  workflowVersion: string;
  request: ReplayRequest;
  modelTier: ModelTier;
  promptVersion: string;
  sampling: ReplaySampling;
  nodes: readonly ReplayNode[];
  outcome: ReplayOutcome;
  costMicros: number;
  latencyMs: number;
}

export type ReplayDifferenceCode =
  | "request"
  | "workflow-version"
  | "model-tier"
  | "prompt-version"
  | "sampling"
  | "node-order"
  | "node-input"
  | "node-output"
  | "node-status"
  | "outcome"
  | "cost"
  | "latency";

export interface ReplayDifference {
  code: ReplayDifferenceCode;
  path: string;
  baseline: string;
  replayed: string;
}

export interface ReplayComparison {
  runId: string;
  identical: boolean;
  /** 第一个发生输入或输出分歧的节点；包级差异全部一致时才有值。 */
  firstDivergenceNode?: string;
  differences: ReplayDifference[];
}

const OUTCOMES: readonly ReplayOutcome[] = [
  "success",
  "partial",
  "failure",
  "timeout",
  "unauthorized",
];
const TIERS: readonly ModelTier[] = ["frontier", "standard", "economy"];
const STATUSES: readonly ReplayNodeStatus[] = ["ok", "failed", "timeout", "skipped", "denied"];

function fail(message: string): never {
  throw new AiError("invalid-output", `replay: ${message}`);
}

function text(value: unknown, field: string): string {
  if (typeof value !== "string" || value.trim() === "") fail(`${field} 必须是非空字符串`);
  return value;
}

function integer(value: unknown, field: string): number {
  if (typeof value !== "number" || !Number.isInteger(value) || value < 0) {
    fail(`${field} 必须是 >= 0 的整数`);
  }
  return value;
}

function finite(value: unknown, field: string): number {
  if (typeof value !== "number" || !Number.isFinite(value) || value < 0) {
    fail(`${field} 必须是 >= 0 的有限数`);
  }
  return value;
}

/** 节点摘要用 SHA-256 前 16 位十六进制；足以区分内容，又不会把原文写进包。 */
export function digestText(content: string): string {
  return createHash("sha256").update(content, "utf8").digest("hex").slice(0, 16);
}

/** 递归按键名排序；数组保持原顺序，因为节点顺序本身是回放语义的一部分。 */
function canonical(value: unknown): unknown {
  if (Array.isArray(value)) return value.map((item) => canonical(item));
  if (value !== null && typeof value === "object") {
    const source = value as Record<string, unknown>;
    const sorted: Record<string, unknown> = {};
    for (const key of Object.keys(source).sort()) {
      sorted[key] = canonical(source[key]);
    }
    return sorted;
  }
  return value;
}

export function validateReplay(pkg: ReplayPackage): void {
  if (pkg.schemaVersion !== 1) fail(`不支持的 schemaVersion ${String(pkg.schemaVersion)}`);
  text(pkg.runId, "runId");
  text(pkg.workflow, "workflow");
  text(pkg.workflowVersion, "workflowVersion");
  text(pkg.request.capability, "request.capability");
  text(pkg.request.tenantId, "request.tenantId");
  text(pkg.request.input, "request.input");
  integer(pkg.request.deadlineMs, "request.deadlineMs");
  if (!TIERS.includes(pkg.modelTier)) fail(`未知的 modelTier ${String(pkg.modelTier)}`);
  text(pkg.promptVersion, "promptVersion");
  finite(pkg.sampling.temperature, "sampling.temperature");
  if (pkg.sampling.temperature > 2) fail("sampling.temperature 必须落在 [0, 2]");
  integer(pkg.sampling.seed, "sampling.seed");
  integer(pkg.sampling.maxOutputTokens, "sampling.maxOutputTokens");
  if (pkg.nodes.length === 0) fail("回放包至少需要一个节点");
  const seen = new Set<string>();
  for (const node of pkg.nodes) {
    text(node.nodeId, "node.nodeId");
    if (seen.has(node.nodeId)) fail(`重复的 nodeId ${node.nodeId}`);
    seen.add(node.nodeId);
    text(node.inputDigest, `${node.nodeId}.inputDigest`);
    text(node.outputDigest, `${node.nodeId}.outputDigest`);
    if (!STATUSES.includes(node.status)) fail(`未知的节点状态 ${String(node.status)}`);
    finite(node.latencyMs, `${node.nodeId}.latencyMs`);
  }
  if (!OUTCOMES.includes(pkg.outcome)) fail(`未知的 outcome ${String(pkg.outcome)}`);
  integer(pkg.costMicros, "costMicros");
  integer(pkg.latencyMs, "latencyMs");
}

/** 序列化是规范化的：同样的包永远得到同样的字节，便于做快照与差异比较。 */
export function serializeReplay(pkg: ReplayPackage): string {
  validateReplay(pkg);
  return `${JSON.stringify(canonical(pkg))}\n`;
}

export function deserializeReplay(serialized: string): ReplayPackage {
  let raw: unknown;
  try {
    raw = JSON.parse(serialized);
  } catch (error) {
    throw new AiError("invalid-output", "replay: 回放包不是合法 JSON", { cause: error });
  }
  if (raw === null || typeof raw !== "object" || Array.isArray(raw)) {
    fail("回放包必须是 JSON 对象");
  }
  const pkg = raw as ReplayPackage;
  validateReplay(pkg);
  return pkg;
}

function sameJson(left: unknown, right: unknown): boolean {
  return JSON.stringify(canonical(left)) === JSON.stringify(canonical(right));
}

/**
 * 对比基线包与重放包。节点级分歧优先记录，firstDivergenceNode 指向第一个
 * 输入或输出摘要变化的节点，用来回答“从哪一步开始不一样”。
 */
export function compareReplay(baseline: ReplayPackage, replayed: ReplayPackage): ReplayComparison {
  validateReplay(baseline);
  validateReplay(replayed);
  const differences: ReplayDifference[] = [];
  let firstDivergenceNode: string | undefined;

  const push = (code: ReplayDifferenceCode, path: string, before: unknown, after: unknown): void => {
    differences.push({
      code,
      path,
      baseline: typeof before === "string" ? before : JSON.stringify(before),
      replayed: typeof after === "string" ? after : JSON.stringify(after),
    });
  };

  if (!sameJson(baseline.request, replayed.request)) push("request", "request", baseline.request, replayed.request);
  if (baseline.workflowVersion !== replayed.workflowVersion) {
    push("workflow-version", "workflowVersion", baseline.workflowVersion, replayed.workflowVersion);
  }
  if (baseline.modelTier !== replayed.modelTier) {
    push("model-tier", "modelTier", baseline.modelTier, replayed.modelTier);
  }
  if (baseline.promptVersion !== replayed.promptVersion) {
    push("prompt-version", "promptVersion", baseline.promptVersion, replayed.promptVersion);
  }
  if (!sameJson(baseline.sampling, replayed.sampling)) {
    push("sampling", "sampling", baseline.sampling, replayed.sampling);
  }

  const nodeCount = Math.min(baseline.nodes.length, replayed.nodes.length);
  for (let index = 0; index < nodeCount; index += 1) {
    const before = baseline.nodes[index];
    const after = replayed.nodes[index];
    if (before.nodeId !== after.nodeId) {
      push("node-order", `nodes[${index}].nodeId`, before.nodeId, after.nodeId);
      if (firstDivergenceNode === undefined) firstDivergenceNode = before.nodeId;
      continue;
    }
    if (before.inputDigest !== after.inputDigest) {
      push("node-input", `nodes[${index}].inputDigest`, before.inputDigest, after.inputDigest);
      if (firstDivergenceNode === undefined) firstDivergenceNode = before.nodeId;
    }
    if (before.outputDigest !== after.outputDigest) {
      push("node-output", `nodes[${index}].outputDigest`, before.outputDigest, after.outputDigest);
      if (firstDivergenceNode === undefined) firstDivergenceNode = before.nodeId;
    }
    if (before.status !== after.status) {
      push("node-status", `nodes[${index}].status`, before.status, after.status);
      if (firstDivergenceNode === undefined) firstDivergenceNode = before.nodeId;
    }
  }
  if (baseline.nodes.length !== replayed.nodes.length) {
    push("node-order", "nodes.length", baseline.nodes.length, replayed.nodes.length);
  }

  if (baseline.outcome !== replayed.outcome) push("outcome", "outcome", baseline.outcome, replayed.outcome);
  if (baseline.costMicros !== replayed.costMicros) {
    push("cost", "costMicros", baseline.costMicros, replayed.costMicros);
  }
  if (baseline.latencyMs !== replayed.latencyMs) {
    push("latency", "latencyMs", baseline.latencyMs, replayed.latencyMs);
  }

  return {
    runId: baseline.runId,
    identical: differences.length === 0,
    firstDivergenceNode,
    differences,
  };
}

compareReplay 先比包级字段，再按节点顺序比输入摘要、输出摘要与状态。firstDivergenceNode 只在节点级出现分歧时才有值，它回答的是「从哪一步开始不一样」，而不是「哪个字段不一样」。

一次重放定位

基线包记录提示 v3、standard 档、温度 0、种子 424242，结果是成功，成本 3,786 微元、延迟 1,269ms。有人把提示升到 v4 后，同一条请求失败，成本 1,842 微元。

把两次执行分别序列化后对比，差异有四项：promptVersion 从 v3 变成 v4，answer 节点的输出摘要变化，outcome 从 success 变成 failure，costMicros 变化。firstDivergenceNode 是 answer，它之前的 retrieve 输入输出完全一致。

结论很直接：分歧不是检索造成的，也不是模型档位或种子造成的，而是提示改版。把 promptVersion 改回 v3、其余字段不变，重放包与基线逐字节相同，对比结果 identical: true。这一步不是「相信直觉」，而是用包把变量逐个排除。

三种重放结论

第一种，完全相同。包的输入、档位、提示、种子和节点输出都一致，说明这次失败没有被包捕获：可能是外部系统状态、并发竞争或包外的人工节点。此时应该扩包，而不是继续重跑。

第二种，输入相同、输出不同。档位、提示、种子都一致，某节点输出摘要却变了，说明模型侧发生了漂移：提供方更新了模型、温度没有真正生效，或者采样参数被网关覆盖。修复动作是版本锁定与参数透传检查。

第三种，失败节点之前的输入摘要就变了。说明上游产出在两次执行间不同，包没有冻结上游状态；常见于检索索引更新、缓存过期或人工节点。此时要先补上游快照，再看失败节点。

保留窗口与隐私

回放包按 7 天保留请求原文与节点摘要，30 天保留包级结果与摘要；完整节点输出默认不存，只对抽样失败开启短期留存。租户级删除请求要能级联删除回放包，否则调试系统会变成隐私死角。

回放包不是审计日志。审计要记录「谁在什么时候做了什么」，回放要记录「这次执行为什么走这条路径」。两者的字段有重叠，但保留策略与访问权限应该分开。

小结

不可复现通常不是模型不可解释，而是复现时没有冻结变量。最小回放包记录请求、节点输入输出摘要、模型档位、提示版本、采样参数与结果；规范化序列化保证同包同字节；逐节点对比用 firstDivergenceNode 把分歧定位到具体节点。示例中提示 v3 升到 v4，分歧落在 answer，改回 v3 后重放包与基线完全一致。

7.1 定义任务级验收，7.2 把失败拆到节点，7.3 让同一条失败可以被重放和排除。这三件事合起来，编排质量才从「看板上的一个百分比」变成可复核的证据。

阅读导航：上一节：7.2 失败归因：哪个节点造成的 · 下一节：8.1 把预算分到节点 。

常见错误与边界

  • 只记录最终错误。没有节点输入输出摘要，无法判断分歧从哪一步开始。
  • 回放时用当前提示与当前档位。环境没有冻结，重跑得到的是另一个任务。
  • 不记录随机种子。温度为 0 也不代表所有提供方都确定，种子仍要显式保存。
  • 把完整节点输出都存下来。体积与隐私风险都会失控，摘要加受控原文更合适。
  • 序列化不排序键。语义相同的包会因为键顺序不同而比较失败。
  • 把 identical: true 当成「不是我们的问题」。它也可能说明包没有捕获外部状态。
  • 回放包无限期保留。请求原文与节点摘要都应按窗口清理，并支持租户级删除。
  • 用回放包替代审计日志。两者目的、字段与访问权限不同，不能混用。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《互联网产品的 AI 架构·平台化与治理》附录 C 平台化评审清单
  2. 《互联网产品的 AI 架构·平台化与治理》附录 B 能力目录与配额字段速查
  3. 《互联网产品的 AI 架构·平台化与治理》附录 A 治理角色与职责速查