《互联网产品的 AI 架构·推理性能与成本工程》3.1 上下文窗口的成本结构

// file: src/perf/context.ts
// tier: L2
// verify: node --test test/perf/context.test.ts

import { AiError } from "../core/errors.ts";
import { estimateTokens } from "../prompt/context-budget.ts";
import type { ModelProfile } from "../core/types.ts";

export const CONTEXT_ROLES = ["system", "history", "retrieval", "tool", "question"] as const;
export type ContextRole = (typeof CONTEXT_ROLES)[number];

/** 拼装前的上下文片段。role 决定成本归属，不影响发送顺序。 */
export interface ContextPiece {
  id: string;
  role: ContextRole;
  text: string;
}

export interface RoleRow {
  role: ContextRole;
  pieces: number;
  tokens: number;
  /** 占全部输入 token 的比例，0 到 1。 */
  share: number;
}

export interface ContextComposition {
  totalTokens: number;
  rows: RoleRow[];
  byRole: Record<ContextRole, number>;
}

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

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

function emptyByRole(): Record<ContextRole, number> {
  return { system: 0, history: 0, retrieval: 0, tool: 0, question: 0 };
}

/**
 * 输入 token 的构成。行按固定顺序输出，零片段也保留，
 * 这样两次统计可以逐行对比，不会因为缺行错位。
 */
export function composeContext(pieces: readonly ContextPiece[]): ContextComposition {
  if (pieces.length === 0) fail("至少需要一个上下文片段");
  const seen = new Set<string>();
  const tokens = emptyByRole();
  const counts = emptyByRole();
  for (const piece of pieces) {
    if (piece.id.trim() === "") fail("片段 id 不能为空");
    if (seen.has(piece.id)) fail(`片段 id 重复：${piece.id}`);
    seen.add(piece.id);
    if (!CONTEXT_ROLES.includes(piece.role)) fail(`未知角色：${String(piece.role)}`);
    tokens[piece.role] += estimateTokens(piece.text);
    counts[piece.role] += 1;
  }
  const totalTokens = CONTEXT_ROLES.reduce((sum, role) => sum + tokens[role], 0);
  return {
    totalTokens,
    byRole: tokens,
    rows: CONTEXT_ROLES.map((role) => ({
      role,
      pieces: counts[role],
      tokens: tokens[role],
      share: totalTokens === 0 ? 0 : tokens[role] / totalTokens,
    })),
  };
}

export interface GrowthInput {
  /** 不含历史的固定部分：system、检索、工具结果与当前问题。 */
  base: readonly ContextPiece[];
  initialHistoryTokens: number;
  /** 每轮新增的用户消息与助手回复合计 token。 */
  perTurnHistoryTokens: number;
  turns: number;
}

export interface GrowthRow {
  turn: number;
  historyTokens: number;
  totalTokens: number;
  historyShare: number;
}

/** 历史按轮次线性增长，但它在输入里的占比会随轮次上升。 */
export function projectGrowth(input: GrowthInput): GrowthRow[] {
  requireInteger(input.initialHistoryTokens, "initialHistoryTokens", 0);
  requireInteger(input.perTurnHistoryTokens, "perTurnHistoryTokens", 0);
  requireInteger(input.turns, "turns", 1);
  const base = composeContext(input.base).totalTokens;
  const rows: GrowthRow[] = [];
  for (let turn = 1; turn <= input.turns; turn += 1) {
    const historyTokens = input.initialHistoryTokens + input.perTurnHistoryTokens * (turn - 1);
    const totalTokens = base + historyTokens;
    rows.push({
      turn,
      historyTokens,
      totalTokens,
      historyShare: totalTokens === 0 ? 0 : historyTokens / totalTokens,
    });
  }
  return rows;
}

/** 成本沿用网关的整数微元口径，向上取整，避免一次请求被算成零成本。 */
export function inputCostMicros(model: ModelProfile, inputTokens: number): number {
  requireInteger(inputTokens, "inputTokens", 0);
  requireInteger(model.inputMicrosPerMTok, "inputMicrosPerMTok", 0);
  return Math.ceil((inputTokens * model.inputMicrosPerMTok) / 1_000_000);
}

export function outputCostMicros(model: ModelProfile, outputTokens: number): number {
  requireInteger(outputTokens, "outputTokens", 0);
  requireInteger(model.outputMicrosPerMTok, "outputMicrosPerMTok", 0);
  return Math.ceil((outputTokens * model.outputMicrosPerMTok) / 1_000_000);
}

/**
 * 首字延迟的参考模型：固定开销 + 线性预填充 + 注意力二次项。
 * 系数是示例值，不是任何提供方的实测曲线，只用于比较相对代价。
 */
export interface PrefillModel {
  fixedMs: number;
  perKTokMs: number;
  quadraticMsPerKTok2: number;
}

export const SAMPLE_PREFILL: PrefillModel = { fixedMs: 40, perKTokMs: 8, quadraticMsPerKTok2: 0.06 };

export function estimatePrefillMs(tokens: number, model: PrefillModel = SAMPLE_PREFILL): number {
  requireInteger(tokens, "tokens", 0);
  const k = tokens / 1000;
  return Math.round((model.fixedMs + model.perKTokMs * k + model.quadraticMsPerKTok2 * k * k) * 10) / 10;
}

export interface CurveRow {
  tokens: number;
  costMicros: number;
  prefillMs: number;
  /** 相对上一行的增量；首行为 0。 */
  marginalCostMicros: number;
  marginalMs: number;
  marginalMsPerKTok: number;
}

/** 同一份提示在不同长度下的成本与首字延迟，用来暴露边际代价的变化。 */
export function contextCurve(
  model: ModelProfile,
  lengths: readonly number[],
  prefill: PrefillModel = SAMPLE_PREFILL,
): CurveRow[] {
  if (lengths.length === 0) fail("lengths 不能为空");
  const sorted = [...lengths].sort((left, right) => left - right);
  return sorted.map((tokens, index) => {
    requireInteger(tokens, "lengths[]", 0);
    const costMicros = inputCostMicros(model, tokens);
    const prefillMs = estimatePrefillMs(tokens, prefill);
    if (index === 0) {
      return { tokens, costMicros, prefillMs, marginalCostMicros: 0, marginalMs: 0, marginalMsPerKTok: 0 };
    }
    const previous = sorted[index - 1];
    if (tokens === previous) fail("lengths 不能包含重复值");
    const marginalMs = Math.round((prefillMs - estimatePrefillMs(previous, prefill)) * 10) / 10;
    return {
      tokens,
      costMicros,
      prefillMs,
      marginalCostMicros: costMicros - inputCostMicros(model, previous),
      marginalMs,
      marginalMsPerKTok: Math.round((marginalMs / ((tokens - previous) / 1000)) * 10) / 10,
    };
  });
}

export interface ConversationInput {
  model: ModelProfile;
  /** 每轮都要重发的固定部分：system、检索、工具、当前问题。 */
  fixedTokens: number;
  initialHistoryTokens: number;
  perTurnHistoryTokens: number;
  turns: number;
  outputTokensPerTurn: number;
}

export interface ConversationRow {
  turn: number;
  inputTokens: number;
  costMicros: number;
  cumulativeMicros: number;
}

export interface ConversationCost {
  rows: ConversationRow[];
  totalInputTokens: number;
  totalOutputTokens: number;
  totalMicros: number;
  /** 把历史当成“只发一次”的错误口径，仅用于展示差额。 */
  noResendMicros: number;
}

/** 多轮对话每轮都重发固定部分与历史，累计成本近似二次增长。 */
export function conversationCost(input: ConversationInput): ConversationCost {
  requireInteger(input.fixedTokens, "fixedTokens", 0);
  requireInteger(input.initialHistoryTokens, "initialHistoryTokens", 0);
  requireInteger(input.perTurnHistoryTokens, "perTurnHistoryTokens", 0);
  requireInteger(input.turns, "turns", 1);
  requireInteger(input.outputTokensPerTurn, "outputTokensPerTurn", 0);

  const rows: ConversationRow[] = [];
  let cumulative = 0;
  let totalInputTokens = 0;
  for (let turn = 1; turn <= input.turns; turn += 1) {
    const historyTokens = input.initialHistoryTokens + input.perTurnHistoryTokens * (turn - 1);
    const inputTokens = input.fixedTokens + historyTokens;
    const costMicros =
      inputCostMicros(input.model, inputTokens) + outputCostMicros(input.model, input.outputTokensPerTurn);
    cumulative += costMicros;
    totalInputTokens += inputTokens;
    rows.push({ turn, inputTokens, costMicros, cumulativeMicros: cumulative });
  }
  const totalOutputTokens = input.outputTokensPerTurn * input.turns;
  const sentOnce = input.fixedTokens + input.initialHistoryTokens + input.perTurnHistoryTokens * input.turns;
  return {
    rows,
    totalInputTokens,
    totalOutputTokens,
    totalMicros: cumulative,
    noResendMicros: inputCostMicros(input.model, sentOnce) + outputCostMicros(input.model, totalOutputTokens),
  };
}

node --test test/perf/context.test.ts

ℹ tests 8
ℹ pass 8
ℹ fail 0
ℹ duration_ms 177.843209