《互联网产品的 AI 架构·多能力编排与任务闭环》5.3 编排级重试预算与熔断

单步 13.02% 的失败上限落到编排层,就变成全局重试预算与节点熔断。本节给出按成功率增量分配预算的确定性算法,用三能力样本对比无重试、本地重试、全局预算与熔断四种策略的成功率、每次成功交付成本与尾部成本,并说明重试风暴的尾部成本与幂等边界。

从单步预算到编排预算

卷二给出过一个单步判据:三步链路、每步 1 次重试、目标成功率 95% 时,单步失败率上限是 13.02%。这个数字回答的是「每个节点最多能坏到什么程度」,但没有回答另一个问题:三次重试的额度应该怎么在节点之间分配?

文中数字都来自 ai-lab 的示例样本运行,节点成本沿用卷二的三能力样本:检索 1,055、摘要 1,643、回复 1,088 微元,失败率分别为 1%、3%、2%。

预算该给谁

预算单位是「重试次数」,所以分配目标不是「谁更便宜」,而是「多一次重试能多救回多少成功率」。第 r+1 次重试的增量收益是:

Δsuccess = f^(r+1) × (1 - f)

它随重试次数指数衰减。第一次重试的收益是 f × (1-f),第二次只有 f² × (1-f)。因此预算永远先给「还没重试过、失败率又高」的必需节点,而不是给已经重试两次的节点再加一次。

熔断与重试风暴

全局预算限制的是一次请求内的重试总量,熔断限制的是一个节点在连续失败后还能不能被调用。两者解决不同的问题:预算防「一次请求烧太多」,熔断防「一个坏节点拖垮所有请求」。

熔断器按连续失败计数打开,打开后先拒绝若干次调用,再放一个探测请求。探测成功则关闭,失败则重新打开。这里用「被拒绝的调用数」而不是时钟推进冷却,是为了让同一串调用序列在测试里永远得到同一组状态。

重试风暴的尾部成本可以用一个上界描述:把所有可用重试分配给单次成本最高的节点。示例样本里,本地无上限策略的最坏情况是 9 次尝试、11,358 微元,是基线 3,786 的 3 倍;全局预算 3 加熔断阈值 2 把最坏情况压到 6 次尝试、7,572 微元,正好 2 倍。

实现

下面的模块做三件事:按成功率增量分配全局预算、计算整链成功率与每次成功交付成本、给出尾部成本上界。熔断器与预算分配在同一个模块里,由同一组测试覆盖。

// file: src/orch/retry-budget.ts
// tier: L1
// verify: node --test test/orch/failure.test.ts

import { AiError } from "../core/errors.ts";

/** 参与预算分配的节点:成本、延迟与失败率都按单次尝试计。 */
export interface RetryNode {
  id: string;
  failureRate: number;
  attemptCostMicros: number;
  latencyMs: number;
  /** 本地允许的最大重试次数。 */
  maxRetries: number;
  /** 必需节点失败会拖垮整链;可选节点失败只影响部分结果。 */
  required: boolean;
}

export interface BudgetPolicy {
  /** 一次编排请求内所有节点共享的重试次数池。 */
  globalRetryBudget: number;
  /** 连续失败达到该值后打开熔断;1 表示首次失败即熔断。 */
  consecutiveFailureThreshold: number;
  /** 熔断打开后先拒绝多少次调用,再放一个探测请求。 */
  halfOpenAfterCalls: number;
}

export interface NodeAllocation {
  nodeId: string;
  retries: number;
  retryCap: number;
  effectiveFailureRate: number;
  expectedAttempts: number;
  successProbability: number;
  expectedCostMicros: number;
}

export interface BudgetPlan {
  allocations: NodeAllocation[];
  retriesUsed: number;
  chainSuccess: number;
  expectedCostMicros: number;
  costPerSuccessMicros: number;
  expectedLatencyMs: number;
  /** 期望尝试次数除以节点数,衡量重试把请求放大了多少。 */
  retryAmplification: number;
  /** 可选节点至少失败一个的概率,对应用户会看到部分结果。 */
  partialProbability: number;
  /** 被熔断阈值压低本地重试上限的节点。 */
  breakerCapped: string[];
}

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

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

function validateNodes(nodes: readonly RetryNode[]): void {
  if (nodes.length === 0) fail("至少需要一个节点");
  const ids = new Set<string>();
  for (const node of nodes) {
    if (node.id.trim() === "") fail("node.id 不能为空");
    if (ids.has(node.id)) fail(`重复的 node.id ${node.id}`);
    ids.add(node.id);
    if (!(node.failureRate >= 0 && node.failureRate < 1)) fail(`${node.id}.failureRate 必须落在 [0, 1)`);
    integer(node.attemptCostMicros, `${node.id}.attemptCostMicros`);
    if (!Number.isFinite(node.latencyMs) || node.latencyMs < 0) fail(`${node.id}.latencyMs 必须 >= 0`);
    integer(node.maxRetries, `${node.id}.maxRetries`);
  }
}

function validatePolicy(policy: BudgetPolicy): void {
  integer(policy.globalRetryBudget, "globalRetryBudget");
  const threshold = policy.consecutiveFailureThreshold;
  if (threshold !== Number.POSITIVE_INFINITY && (!Number.isInteger(threshold) || threshold < 1)) {
    fail("consecutiveFailureThreshold 必须是 >= 1 的整数或 Infinity");
  }
  integer(policy.halfOpenAfterCalls, "halfOpenAfterCalls", 1);
}

/** 闭合上限:本地重试次数与「熔断前允许的连续尝试」取小。 */
export function retryCap(node: RetryNode, policy: BudgetPolicy): number {
  const breakerCap = policy.consecutiveFailureThreshold === Number.POSITIVE_INFINITY
    ? node.maxRetries
    : Math.max(0, policy.consecutiveFailureThreshold - 1);
  return Math.min(node.maxRetries, breakerCap);
}

function expectedAttempts(failureRate: number, retries: number): number {
  if (failureRate === 0) return 1;
  return (1 - Math.pow(failureRate, retries + 1)) / (1 - failureRate);
}

/**
 * 分配目标是「多一次重试能多救回多少成功率」,不是「谁更便宜」。
 * 必需节点先于可选节点:可选节点再高的失败率也不影响整链是否成功。
 */
export function planRetryBudget(
  nodes: readonly RetryNode[],
  policy: BudgetPolicy,
): BudgetPlan {
  validateNodes(nodes);
  validatePolicy(policy);
  const caps = nodes.map((node) => retryCap(node, policy));
  const retries = nodes.map(() => 0);
  let remaining = policy.globalRetryBudget;

  while (remaining > 0) {
    let pick = -1;
    let bestGain = 0;
    let bestCost = Number.POSITIVE_INFINITY;

    // 必需节点优先:只有必需节点没有可分配额度时才考虑可选节点。
    for (const requiredPass of [true, false]) {
      nodes.forEach((node, index) => {
        if (node.required !== requiredPass) return;
        if (retries[index] >= caps[index] || node.failureRate === 0) return;
        const current = retries[index];
        // 第 current+1 次重试把失败率从 f^(current+1) 降到 f^(current+2)。
        const gain = Math.pow(node.failureRate, current + 1) * (1 - node.failureRate);
        const cost = node.attemptCostMicros * Math.pow(node.failureRate, current + 1);
        if (gain > bestGain || (gain === bestGain && cost < bestCost)) {
          pick = index;
          bestGain = gain;
          bestCost = cost;
        }
      });
      if (pick >= 0) break;
    }

    if (pick < 0) break;
    retries[pick] += 1;
    remaining -= 1;
  }

  let chainSuccess = 1;
  let optionalSuccess = 1;
  let expectedCost = 0;
  let attemptsTotal = 0;
  let expectedLatency = 0;
  const allocations: NodeAllocation[] = nodes.map((node, index) => {
    const retryCount = retries[index];
    const attempts = expectedAttempts(node.failureRate, retryCount);
    const success = 1 - Math.pow(node.failureRate, retryCount + 1);
    const cost = node.attemptCostMicros * attempts;
    expectedCost += cost;
    attemptsTotal += attempts;
    expectedLatency += node.latencyMs * attempts;
    if (node.required) chainSuccess *= success;
    else optionalSuccess *= success;
    return {
      nodeId: node.id,
      retries: retryCount,
      retryCap: caps[index],
      effectiveFailureRate: Number(Math.pow(node.failureRate, retryCount + 1).toFixed(8)),
      expectedAttempts: Number(attempts.toFixed(6)),
      successProbability: Number(success.toFixed(8)),
      expectedCostMicros: Math.round(cost),
    };
  });

  const totalCost = Math.round(expectedCost);
  return {
    allocations,
    retriesUsed: retries.reduce((sum, value) => sum + value, 0),
    chainSuccess: Number(chainSuccess.toFixed(8)),
    expectedCostMicros: totalCost,
    costPerSuccessMicros: chainSuccess === 0
      ? Number.POSITIVE_INFINITY
      : Math.round(totalCost / chainSuccess),
    expectedLatencyMs: Math.round(expectedLatency),
    retryAmplification: Number((attemptsTotal / nodes.length).toFixed(4)),
    partialProbability: Number((1 - optionalSuccess).toFixed(8)),
    breakerCapped: nodes
      .filter((node, index) => caps[index] < node.maxRetries)
      .map((node) => node.id),
  };
}

/**
 * 尾部上界:把可用重试全部给单次成本最高的节点。
 * 期望值回答「平时多少钱」,尾部上界回答「一次风暴最多烧多少」。
 */
export function worstCaseRetryCost(
  nodes: readonly RetryNode[],
  policy: BudgetPolicy,
): { maxAttempts: number; maxCostMicros: number; amplification: number } {
  validateNodes(nodes);
  validatePolicy(policy);
  const caps = nodes.map((node) => retryCap(node, policy));
  const baseCost = nodes.reduce((sum, node) => sum + node.attemptCostMicros, 0);
  const pool = Math.min(policy.globalRetryBudget, caps.reduce((sum, cap) => sum + cap, 0));
  const retryCosts = nodes
    .flatMap((node, index) => Array.from({ length: caps[index] }, () => node.attemptCostMicros))
    .sort((left, right) => right - left);
  const maxCost = baseCost + retryCosts.slice(0, pool).reduce((sum, cost) => sum + cost, 0);
  return {
    maxAttempts: nodes.length + pool,
    maxCostMicros: maxCost,
    amplification: baseCost === 0 ? 0 : Number((maxCost / baseCost).toFixed(4)),
  };
}

export type BreakerState = "closed" | "open" | "half-open";

export interface BreakerOptions {
  failureThreshold: number;
  halfOpenAfterCalls: number;
}

/**
 * 计数式熔断器:不用时钟,冷却按「被拒绝的调用数」推进,
 * 因而同一串调用序列永远得到同一组状态,适合放进确定性测试。
 */
export class NodeCircuitBreaker {
  readonly nodeId: string;
  readonly options: BreakerOptions;
  private current: BreakerState = "closed";
  private consecutiveFailures = 0;
  private rejectedSinceOpen = 0;

  constructor(nodeId: string, options: BreakerOptions) {
    if (nodeId.trim() === "") fail("nodeId 不能为空");
    if (!Number.isInteger(options.failureThreshold) || options.failureThreshold < 1) {
      fail("failureThreshold 必须是 >= 1 的整数");
    }
    if (!Number.isInteger(options.halfOpenAfterCalls) || options.halfOpenAfterCalls < 1) {
      fail("halfOpenAfterCalls 必须是 >= 1 的整数");
    }
    this.nodeId = nodeId;
    this.options = options;
  }

  get state(): BreakerState {
    return this.current;
  }

  /** 调用前询问;false 表示本次被短路,不产生模型成本。 */
  allow(): boolean {
    if (this.current === "closed" || this.current === "half-open") return true;
    if (this.rejectedSinceOpen < this.options.halfOpenAfterCalls) {
      this.rejectedSinceOpen += 1;
      return false;
    }
    this.current = "half-open";
    return true;
  }

  record(success: boolean): BreakerState {
    if (success) {
      this.consecutiveFailures = 0;
      this.rejectedSinceOpen = 0;
      this.current = "closed";
      return this.current;
    }
    this.consecutiveFailures += 1;
    if (this.current === "half-open" || this.consecutiveFailures >= this.options.failureThreshold) {
      this.current = "open";
      this.rejectedSinceOpen = 0;
    }
    return this.current;
  }
}

export interface BreakerTrace {
  allowed: number;
  rejected: number;
  attempts: number;
  finalState: BreakerState;
  states: BreakerState[];
}

/** 回放一串「若被调用则成功/失败」的脚本;被拒绝的调用不消耗脚本结果。 */
export function traceBreaker(
  options: BreakerOptions,
  calls: readonly boolean[],
): BreakerTrace {
  const breaker = new NodeCircuitBreaker("trace", options);
  let allowed = 0;
  let rejected = 0;
  let attempts = 0;
  const states: BreakerState[] = [];
  for (const wouldSucceed of calls) {
    if (breaker.allow()) {
      allowed += 1;
      attempts += 1;
      breaker.record(wouldSucceed);
    } else {
      rejected += 1;
    }
    states.push(breaker.state);
  }
  return { allowed, rejected, attempts, finalState: breaker.state, states };
}

/** 卷二三能力样本:检索 1,055、摘要 1,643、回复 1,088 微元。 */
export const ECHO_RETRY_NODES: readonly RetryNode[] = [
  { id: "search.rerank", failureRate: 0.01, attemptCostMicros: 1_055, latencyMs: 160, maxRetries: 2, required: true },
  { id: "content.summarize", failureRate: 0.03, attemptCostMicros: 1_643, latencyMs: 620, maxRetries: 2, required: true },
  { id: "chat.reply", failureRate: 0.02, attemptCostMicros: 1_088, latencyMs: 420, maxRetries: 2, required: true },
];

export const ECHO_BUDGET_POLICY: BudgetPolicy = {
  globalRetryBudget: 3,
  consecutiveFailureThreshold: 2,
  halfOpenAfterCalls: 3,
};

retryCap 把本地重试上限与熔断阈值合成一个闭合上限:阈值是 2 时,节点最多尝试 2 次,也就是只能重试 1 次。预算分配在这个上限内进行,因此熔断不是事后补救,而是提前改变了可分配的额度。

示例样本上的运行结果

四种策略在同一组节点上的对比如下:

策略重试次数整链成功率期望成本每次成功交付最坏尝试最坏成本
无重试094.11%3,7864,02333,786
本地重试、无全局上限699.9964%3,8703,870911,358
全局预算 3 + 熔断阈值 2399.86%3,8683,87367,572
全局预算 6、无熔断699.9964%3,8703,870911,358

全局预算 3 刚好覆盖三步各一次重试,成功率 99.86%,每次成功交付 3,873 微元。把预算加到 6,成功率只多 0.14 个百分点,成本却多了 2 微元,最坏成本从 7,572 涨到 11,358。第二次重试买到的成功率非常有限,这是指数衰减的直接结果。

预算降到 2 时,示例样本的成功率是 98.87%,每次成功交付 3,901 微元;分配会放弃失败率最低的检索。若三步失败率都贴着 13.02% 的上限,预算 2 只能守住 84.06%,预算 3 才刚好达到 95.0001%。这就是「三步各一次重试」必须对应全局预算 3 的原因:单步上限是针对这个预算算出来的,预算缩水,上限就失效。

熔断器的示例回放:6 次调用、前 2 次失败、阈值 2、冷却 3 次调用。结果是允许 3 次、拒绝 3 次;被拒绝的调用不产生模型成本。若按摘要节点 1,643 微元估算,这 3 次拒绝省下 4,929 微元,代价是这几次请求转为部分结果。

幂等边界

重试预算只对幂等节点开放。发布、下单、发消息这类最终动作,即使错误码可重试,也不能自动重放;它们应该走 5.1 的中止分支,进入对账队列。

需要重试的写操作必须带业务幂等键。幂等键让「重放」在服务端变成「查询上一次结果」,此时重试才是安全的。没有幂等键的写操作,重试次数越多,重复副作用的风险越大。

小结

编排级重试预算把单步失败上限变成一个可分配的资源。分配依据是每次重试的成功率增量,先必需节点、后可选节点;示例样本里预算 3 恰好覆盖三步各一次,成功率 99.86%,每次成功交付 3,873 微元。

熔断限制坏节点的连续调用,全局预算限制单次请求的重试总量,两者一起把最坏成本从 11,358 压到 7,572 微元。预算不是越多越好:从 3 加到 6,成功率只多 0.14 个百分点,尾部成本却回到 3 倍。

阅读导航:上一节:5.2 部分可用的验收标准 · 下一节:6.1 并行分支的收益与竞争 。

常见错误与边界

  • 每个节点各自重试、没有全局池。一次请求的重试次数会随节点数线性增长。
  • 预算缩水后仍沿用 13.02% 的单步上限。上限与预算绑定,预算 2 时它不再成立。
  • 只算期望成本、不算尾部成本。风暴发生在尾部,平均值看不见。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「books」更多文章

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