《互联网产品的 AI 架构·模型接入与能力边界》8.1 端到端延迟预算怎么拆

// file: src/cost/latency-budget.ts
// tier: L1
// verify: node --test test/cost/latency-budget.test.ts

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

/** 端到端预算的五个阶段，顺序就是一次请求的经过顺序。 */
export type LatencyStage = "network" | "queue" | "first-token" | "generation" | "postprocess";

export const LATENCY_STAGES: LatencyStage[] = [
  "network",
  "queue",
  "first-token",
  "generation",
  "postprocess",
];

export interface LatencyBudgetSpec {
  /** 业务承诺的端到端上限，单位毫秒。 */
  totalMs: number;
  /** 留给抖动与收尾的余量，不参与阶段分配。 */
  reserveMs: number;
  /** 阶段权重，必须是正整数；按最大余数法分配，保证总和精确。 */
  weights: Record<LatencyStage, number>;
}

export interface StageBudget {
  stage: LatencyStage;
  budgetMs: number;
  /** 从请求开始算起的累计截止偏移，不是本阶段耗时。 */
  deadlineMs: number;
}

export interface LatencyPlan {
  totalMs: number;
  usableMs: number;
  reserveMs: number;
  stages: StageBudget[];
  /** 首字出现前的累计预算：network + queue + first-token。 */
  firstTokenBudgetMs: number;
}

export interface StageTiming {
  stage: LatencyStage;
  durationMs: number;
}

export interface StageOverrun {
  stage: LatencyStage;
  budgetMs: number;
  actualMs: number;
  overMs: number;
}

export interface LatencyAudit {
  totalMs: number;
  firstTokenMs: number;
  overruns: StageOverrun[];
  withinBudget: boolean;
  firstTokenWithinBudget: boolean;
}

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

function requireInteger(value: number, field: string, min: number): void {
  if (!Number.isInteger(value) || value < min) {
    fail(`${field} must be an integer >= ${min}`);
  }
}

/**
 * 按权重把可用预算分到五个阶段。最大余数法保证各阶段之和精确等于
 * totalMs - reserveMs，余数相同时按阶段顺序分配，结果可被测试逐毫秒断言。
 */
export function planLatency(spec: LatencyBudgetSpec): LatencyPlan {
  requireInteger(spec.totalMs, "totalMs", 1);
  requireInteger(spec.reserveMs, "reserveMs", 0);
  if (spec.reserveMs >= spec.totalMs) {
    fail("reserveMs must be smaller than totalMs");
  }
  for (const stage of LATENCY_STAGES) {
    requireInteger(spec.weights[stage], `weights.${stage}`, 1);
  }

  const usableMs = spec.totalMs - spec.reserveMs;
  const totalWeight = LATENCY_STAGES.reduce((sum, stage) => sum + spec.weights[stage], 0);

  const shares = LATENCY_STAGES.map((stage, index) => {
    const exact = (usableMs * spec.weights[stage]) / totalWeight;
    return {
      stage,
      index,
      budgetMs: Math.floor(exact),
      remainder: exact - Math.floor(exact),
    };
  });

  let left = usableMs - shares.reduce((sum, share) => sum + share.budgetMs, 0);
  const byRemainder = [...shares].sort(
    (left, right) => right.remainder - left.remainder || left.index - right.index,
  );
  for (const share of byRemainder) {
    if (left === 0) break;
    share.budgetMs += 1;
    left -= 1;
  }

  let cursor = 0;
  const stages = shares.map((share) => {
    cursor += share.budgetMs;
    return { stage: share.stage, budgetMs: share.budgetMs, deadlineMs: cursor };
  });

  const firstTokenBudgetMs = stages
    .slice(0, 3)
    .reduce((sum, stage) => sum + stage.budgetMs, 0);

  return {
    totalMs: spec.totalMs,
    usableMs,
    reserveMs: spec.reserveMs,
    stages,
    firstTokenBudgetMs,
  };
}

export function stageDeadlineMs(plan: LatencyPlan, stage: LatencyStage): number {
  const found = plan.stages.find((item) => item.stage === stage);
  if (!found) fail(`unknown stage "${stage}"`);
  return found.deadlineMs;
}

/** 距离本阶段截止还剩多少毫秒；已经超时返回 0，不抛错，便于观测。 */
export function remainingMs(plan: LatencyPlan, stage: LatencyStage, elapsedMs: number): number {
  requireInteger(elapsedMs, "elapsedMs", 0);
  return Math.max(0, stageDeadlineMs(plan, stage) - elapsedMs);
}

/**
 * 计算可以传给下一跳的超时。前面阶段省下的时间允许借用，但绝不允许
 * 借用 reserveMs；余额为零时直接按 timeout 失败，而不是传 0 让下游空转。
 */
export function nextTimeoutMs(plan: LatencyPlan, stage: LatencyStage, elapsedMs: number): number {
  const left = remainingMs(plan, stage, elapsedMs);
  if (left <= 0) {
    throw new AiError("timeout", `latency-budget: stage "${stage}" has no time left at ${elapsedMs}ms`);
  }
  return left;
}

export function elapsedSince(startedAtMs: number, clock: Clock): number {
  const elapsed = clock.now() - startedAtMs;
  return Math.max(0, Math.trunc(elapsed));
}

/** 用一次完整样本核对预算；overruns 按阶段顺序输出，便于直接写进日志。 */
export function auditLatency(plan: LatencyPlan, timings: StageTiming[]): LatencyAudit {
  const seen = new Map<LatencyStage, number>();
  for (const timing of timings) {
    if (!LATENCY_STAGES.includes(timing.stage)) fail(`unknown stage "${timing.stage}"`);
    if (seen.has(timing.stage)) fail(`duplicate timing for stage "${timing.stage}"`);
    requireInteger(timing.durationMs, `durationMs for "${timing.stage}"`, 0);
    seen.set(timing.stage, timing.durationMs);
  }

  const overruns: StageOverrun[] = [];
  for (const budget of plan.stages) {
    const actualMs = seen.get(budget.stage);
    if (actualMs === undefined) continue;
    if (actualMs > budget.budgetMs) {
      overruns.push({
        stage: budget.stage,
        budgetMs: budget.budgetMs,
        actualMs,
        overMs: actualMs - budget.budgetMs,
      });
    }
  }

  const totalMs = [...seen.values()].reduce((sum, value) => sum + value, 0);
  const firstTokenMs = (["network", "queue", "first-token"] as LatencyStage[]).reduce(
    (sum, stage) => sum + (seen.get(stage) ?? 0),
    0,
  );

  return {
    totalMs,
    firstTokenMs,
    overruns,
    withinBudget: totalMs <= plan.usableMs,
    firstTokenWithinBudget: firstTokenMs <= plan.firstTokenBudgetMs,
  };
}

node --test test/cost/latency-budget.test.ts

✔ 预算按权重精确拆分，总和等于可用预算 (1.561208ms)
✔ 最大余数法在余数相同时按阶段顺序分配 (0.138667ms)
✔ 累计截止偏移单调递增，首字预算只含前三段 (0.115917ms)
✔ 审计同时发现单阶段超支与首字超支 (0.182083ms)
✔ 前面阶段省下的时间可以借给生成阶段 (0.090625ms)
✔ 余额耗尽时抛 timeout，而不是把 0 传给下游 (0.282667ms)
✔ 注入时钟只用于观测耗时，逻辑本身保持纯函数 (0.077792ms)
✔ 非法权重与过大的余量被拒绝 (0.140209ms)
ℹ tests 8
ℹ suites 0
ℹ pass 8
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 191.230417