《互联网产品的 AI 架构·推理性能与成本工程》7.1 在线与离线的边界

ceil(50,000 / (636 - 429)) = 242

// file: src/perf/batch.ts
// tier: L1
// verify: node --test test/perf/batch.test.ts

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

export type Placement = "online" | "offline" | "hybrid";
export type Consumption = "blocking" | "deferred";

/** 作业画像：描述的是“一批条目”，不是“一次用户请求”。 */
export interface TaskProfile {
  id: string;
  items: number;
  /** 单条结果最多能等多久；不是整个作业的截止时间。 */
  itemToleranceMs: number;
  /** 结果的新鲜度上限，超过就必须重算。 */
  freshnessMaxAgeMs: number;
  /** 本次作业允许占用的窗口长度。 */
  windowMs: number;
  partialSuccessAllowed: boolean;
  consumption: Consumption;
  /** 必须赶在窗口前产出的紧急条数，例如当天投放的商品。 */
  urgentItems: number;
}

export interface PlacementPolicy {
  /** 成本盈亏平衡条数，由 breakEvenItems 在给定成本结构下算出。 */
  minItems: number;
  /** 当前窗口内可处理的条数。 */
  capacityPerWindow: number;
  /** 全有或全无时为重跑预留的容量比例。 */
  rerunReserveRatio: number;
}

export interface PlacementDecision {
  taskId: string;
  placement: Placement;
  reasons: string[];
  onlineItems: number;
  offlineItems: number;
  /** 作业条数 / 有效窗口容量；大于 1 表示一个窗口做不完。 */
  windowUtilization: number;
}

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

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

/** 离线不是“低优先级在线”：判定必须同时看消费方式、时效、容量与失败单位。 */
export function classifyTask(task: TaskProfile, policy: PlacementPolicy): PlacementDecision {
  requireCount(task.items, "items", 1);
  requireCount(task.urgentItems, "urgentItems", 0);
  requireCount(task.itemToleranceMs, "itemToleranceMs", 0);
  requireCount(task.freshnessMaxAgeMs, "freshnessMaxAgeMs", 0);
  requireCount(task.windowMs, "windowMs", 0);
  requireCount(policy.minItems, "minItems", 0);
  requireCount(policy.capacityPerWindow, "capacityPerWindow", 0);
  if (!Number.isFinite(policy.rerunReserveRatio) || policy.rerunReserveRatio < 0 || policy.rerunReserveRatio >= 1) {
    fail("rerunReserveRatio must be in [0, 1)");
  }
  if (task.urgentItems > task.items) fail("urgentItems cannot exceed items");

  // 全有或全无时，窗口容量必须留出一次完整重跑，否则失败就必然错过窗口。
  const capacity = task.partialSuccessAllowed
    ? policy.capacityPerWindow
    : Math.floor(policy.capacityPerWindow * (1 - policy.rerunReserveRatio));
  const utilization = capacity === 0 ? Number.POSITIVE_INFINITY : task.items / capacity;
  const base = { taskId: task.id, onlineItems: 0, offlineItems: 0, windowUtilization: utilization };

  if (task.consumption === "blocking") {
    return { ...base, placement: "online", onlineItems: task.items, reasons: ["结果进入用户等待路径"] };
  }
  if (task.items < policy.minItems) {
    return {
      ...base,
      placement: "online",
      onlineItems: task.items,
      reasons: [`低于成本盈亏平衡点 ${policy.minItems} 条`],
    };
  }
  if (task.urgentItems === task.items) {
    return {
      ...base,
      placement: "online",
      onlineItems: task.items,
      reasons: ["全部条目都要在窗口前产出"],
    };
  }
  if (task.windowMs > task.itemToleranceMs || task.windowMs > task.freshnessMaxAgeMs) {
    return {
      ...base,
      placement: "online",
      onlineItems: task.items,
      reasons: ["窗口晚于结果可接受的新鲜度"],
    };
  }

  const deferred = task.items - task.urgentItems;
  if (deferred <= capacity) {
    if (task.urgentItems === 0) {
      return {
        ...base,
        placement: "offline",
        offlineItems: deferred,
        reasons: ["允许延后消费，窗口容量足够"],
      };
    }
    return {
      ...base,
      placement: "hybrid",
      onlineItems: task.urgentItems,
      offlineItems: deferred,
      reasons: ["紧急子集走在线，其余进入窗口"],
    };
  }

  const online = task.items - capacity;
  return {
    ...base,
    placement: "hybrid",
    onlineItems: online,
    offlineItems: capacity,
    reasons: [`窗口容量不足，${online} 条回落在在线路径`],
  };
}

/** 成本结构：在线按请求付固定开销，离线按作业付一次固定开销。 */
export interface CostShape {
  tokenMicrosPerItem: number;
  onlineCallFixedMicros: number;
  offlineJobFixedMicros: number;
  /** 在线按整次请求重试，失败会重复输入与固定开销。 */
  onlineRetryRate: number;
  /** 离线按条目重试，已完成的条目不会重算。 */
  offlineRetryRate: number;
}

export interface CostComparison {
  onlineMicros: number;
  plannedMicros: number;
  savingMicros: number;
  savingShare: number;
}

function requireRate(value: number, field: string): number {
  if (typeof value !== "number" || !Number.isFinite(value) || value < 0 || value > 1) {
    fail(`${field} must be in [0, 1]`);
  }
  return value;
}

export function onlineItemMicros(shape: CostShape): number {
  const base = requireCount(shape.tokenMicrosPerItem, "tokenMicrosPerItem") +
    requireCount(shape.onlineCallFixedMicros, "onlineCallFixedMicros");
  return Math.ceil(base * (1 + requireRate(shape.onlineRetryRate, "onlineRetryRate")));
}

export function offlineItemMicros(shape: CostShape): number {
  const base = requireCount(shape.tokenMicrosPerItem, "tokenMicrosPerItem");
  return Math.ceil(base * (1 + requireRate(shape.offlineRetryRate, "offlineRetryRate")));
}

/** 离线固定开销 / 每条节省额；低于这个条数，离线一定更贵。 */
export function breakEvenItems(shape: CostShape): number {
  const delta = onlineItemMicros(shape) - offlineItemMicros(shape);
  if (delta <= 0) return Number.POSITIVE_INFINITY;
  return Math.ceil(requireCount(shape.offlineJobFixedMicros, "offlineJobFixedMicros") / delta);
}

/** 与“全部走在线”比较；混合方案只对实际离线的那部分计入作业固定开销。 */
export function compareCost(decision: PlacementDecision, shape: CostShape): CostComparison {
  const onlineCount = decision.onlineItems;
  const offlineCount = decision.offlineItems;
  const onlineAll = (onlineCount + offlineCount) * onlineItemMicros(shape);
  const planned = onlineCount * onlineItemMicros(shape) +
    (offlineCount > 0 ? requireCount(shape.offlineJobFixedMicros, "offlineJobFixedMicros") +
      offlineCount * offlineItemMicros(shape) : 0);
  return {
    onlineMicros: onlineAll,
    plannedMicros: planned,
    savingMicros: onlineAll - planned,
    savingShare: onlineAll === 0 ? 0 : (onlineAll - planned) / onlineAll,
  };
}

node --test test/perf/batch.test.ts

ℹ tests 32
ℹ suites 0
ℹ pass 32
ℹ fail 0
ℹ duration_ms 621.1925