《互联网产品的 AI 架构·模型接入与能力边界》12.1 一次 AI 请求的最小观测字段

// file: src/observe/fields.ts
// tier: L1
// verify: node --test test/observe/fields.test.ts

import { createHash } from "node:crypto";
import type { AiErrorKind } from "../core/errors.ts";
import type { DegradeReason, Usage } from "../core/types.ts";

export const OBSERVATION_SCHEMA_VERSION = 1;

/** 耗时必须拆段记录，否则只能看到总数，无法定位是哪一段变慢。 */
export interface TimingBreakdown {
  queueMs: number;
  routeMs: number;
  providerMs: number;
  postprocessMs: number;
  /** 端到端总耗时，必须不小于各分段之和。 */
  totalMs: number;
}

export type ObservationOutcome = "ok" | "degraded" | "error";
/** 采样分层：错误全留，降级按比例留，正常请求低频留。 */
export type SampleTier = "error" | "degraded" | "normal";

export interface AiObservation {
  schemaVersion: number;
  traceId: string;
  requestId: string;
  tenantId: string;
  capability: string;
  providerId: string;
  modelId: string;
  region: string;
  timing: TimingBreakdown;
  usage: Usage;
  outcome: ObservationOutcome;
  errorKind: AiErrorKind | null;
  degradeReason: DegradeReason | null;
  sampleTier: SampleTier;
}

export interface ValidationIssue {
  path: string;
  message: string;
}

const TRACE_ID = /^[0-9a-f]{16,32}$/;
const NAME = /^[a-z][a-z0-9._:-]*$/;
const TEXT_FIELDS = ["requestId", "tenantId", "capability", "providerId", "modelId", "region"];
const TIMING_FIELDS = ["queueMs", "routeMs", "providerMs", "postprocessMs", "totalMs"];

function isRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}

function checkName(value: unknown, path: string, issues: ValidationIssue[]): void {
  if (typeof value !== "string" || value.trim() === "") {
    issues.push({ path, message: "must be a non-empty string" });
  } else if (!NAME.test(value)) {
    issues.push({ path, message: "must match /^[a-z][a-z0-9._:-]*$/" });
  }
}

function checkCount(value: unknown, path: string, issues: ValidationIssue[]): void {
  if (typeof value !== "number" || !Number.isInteger(value) || value < 0) {
    issues.push({ path, message: "must be a non-negative integer" });
  }
}

/**
 * 观测字段的运行时校验。类型系统在边界处不可信：
 * 日志可能来自旧版本客户端，也可能被手工拼接。
 */
export function validateObservation(value: unknown): ValidationIssue[] {
  if (!isRecord(value)) return [{ path: "$", message: "observation must be an object" }];
  const issues: ValidationIssue[] = [];

  if (value.schemaVersion !== OBSERVATION_SCHEMA_VERSION) {
    issues.push({ path: "schemaVersion", message: `must equal ${OBSERVATION_SCHEMA_VERSION}` });
  }
  if (typeof value.traceId !== "string" || !TRACE_ID.test(value.traceId)) {
    issues.push({ path: "traceId", message: "must be 16-32 lowercase hex characters" });
  }
  for (const field of TEXT_FIELDS) checkName(value[field], field, issues);

  if (!isRecord(value.timing)) {
    issues.push({ path: "timing", message: "must be an object" });
  } else {
    for (const field of TIMING_FIELDS) checkCount(value.timing[field], `timing.${field}`, issues);
    const sum = TIMING_FIELDS.slice(0, 4).reduce((total, field) => {
      const segment = value.timing[field];
      return total + (typeof segment === "number" ? segment : 0);
    }, 0);
    const total = value.timing.totalMs;
    if (typeof total === "number" && total < sum) {
      issues.push({ path: "timing.totalMs", message: `must be >= sum of segments (${sum})` });
    }
  }

  if (!isRecord(value.usage)) {
    issues.push({ path: "usage", message: "must be an object" });
  } else {
    for (const field of ["inputTokens", "outputTokens", "costMicros"]) {
      checkCount(value.usage[field], `usage.${field}`, issues);
    }
  }

  if (value.outcome !== "ok" && value.outcome !== "degraded" && value.outcome !== "error") {
    issues.push({ path: "outcome", message: "must be ok, degraded or error" });
  }
  if (value.errorKind !== null && typeof value.errorKind !== "string") {
    issues.push({ path: "errorKind", message: "must be a string or null" });
  }
  if (value.degradeReason !== null && typeof value.degradeReason !== "string") {
    issues.push({ path: "degradeReason", message: "must be a string or null" });
  }
  if (value.outcome === "ok") {
    if (value.errorKind !== null) issues.push({ path: "errorKind", message: "must be null when outcome is ok" });
    if (value.degradeReason !== null) issues.push({ path: "degradeReason", message: "must be null when outcome is ok" });
  }
  if (value.outcome === "degraded" && value.degradeReason === null) {
    issues.push({ path: "degradeReason", message: "is required when outcome is degraded" });
  }
  if (value.outcome === "error" && value.errorKind === null) {
    issues.push({ path: "errorKind", message: "is required when outcome is error" });
  }
  if (value.sampleTier !== "error" && value.sampleTier !== "degraded" && value.sampleTier !== "normal") {
    issues.push({ path: "sampleTier", message: "must be error, degraded or normal" });
  }
  return issues;
}

export interface RedactedText {
  /** 原文字符数，用于发现异常长的输入，但不暴露内容。 */
  chars: number;
  /** 稳定摘要，便于把同一次输入的多条日志关联起来。 */
  sha256: string;
}

/** 提示词与用户文本只保留长度与摘要，原文一律不进入观测链路。 */
export function redactText(text: string): RedactedText {
  return {
    chars: Array.from(text).length,
    sha256: createHash("sha256").update(text, "utf8").digest("hex"),
  };
}

const METADATA_ALLOWLIST = new Set(["surface", "locale", "clientVersion", "experiment", "route"]);
const SENSITIVE_VALUE = /(@|\+?\d[\d\s-]{6,}\d)/;

/**
 * metadata 采用白名单而不是黑名单：业务方随手塞进来的字段默认全部丢弃。
 * 即使键名合法，值里出现邮箱或长数字串也按敏感信息处理。
 */
export function redactMetadata(metadata: Record<string, string>): Record<string, string> {
  const safe: Record<string, string> = {};
  for (const [key, value] of Object.entries(metadata)) {
    if (!METADATA_ALLOWLIST.has(key)) continue;
    if (SENSITIVE_VALUE.test(value)) continue;
    safe[key] = Array.from(value).slice(0, 64).join("");
  }
  return safe;
}

export interface ObservationLogInput {
  observation: AiObservation;
  /** 原始内容仅用于生成摘要，函数返回前不会保留原文。 */
  content?: string[];
  metadata?: Record<string, string>;
}

export interface ObservationLogRecord extends AiObservation {
  contentDigests: RedactedText[];
  metadata: Record<string, string>;
}

/** 把一次请求收敛成可落盘的记录：原文被摘要替换，metadata 被白名单过滤。 */
export function toLogRecord(input: ObservationLogInput): ObservationLogRecord {
  return {
    ...input.observation,
    contentDigests: (input.content ?? []).map(redactText),
    metadata: redactMetadata(input.metadata ?? {}),
  };
}

export interface SamplingPolicy {
  /** 正常请求的保留比例，0 到 1。 */
  normalRate: number;
  /** 降级请求的保留比例。 */
  degradedRate: number;
  /** 错误是否无条件保留。 */
  keepAllErrors: boolean;
}

export const DEFAULT_SAMPLING: SamplingPolicy = {
  normalRate: 0.02,
  degradedRate: 0.5,
  keepAllErrors: true,
};

function fnv1a(text: string): number {
  let hash = 0x811c9dc5;
  for (let index = 0; index < text.length; index += 1) {
    hash ^= text.charCodeAt(index);
    hash = Math.imul(hash, 0x01000193) >>> 0;
  }
  return hash;
}

/** 按 traceId 做确定性分桶，同一个请求重放多少次都落在同一个桶。 */
export function sampleBucket(traceId: string): number {
  return fnv1a(traceId) % 10_000;
}

export interface SampleDecision {
  tier: SampleTier;
  bucket: number;
  emit: boolean;
}

function withinRate(bucket: number, rate: number): boolean {
  if (rate <= 0) return false;
  if (rate >= 1) return true;
  return bucket < Math.round(rate * 10_000);
}

/**
 * 分层采样：错误全量保留，降级按较高比例保留，正常请求低频保留。
 * 分桶只看 traceId，避免随机抖动把刚出问题的样本抽没。
 */
export function decideSample(
  observation: AiObservation,
  policy: SamplingPolicy = DEFAULT_SAMPLING,
): SampleDecision {
  const tier: SampleTier =
    observation.outcome === "error" ? "error" : observation.outcome === "degraded" ? "degraded" : "normal";
  const bucket = sampleBucket(observation.traceId);
  if (tier === "error") return { tier, bucket, emit: policy.keepAllErrors };
  const rate = tier === "degraded" ? policy.degradedRate : policy.normalRate;
  return { tier, bucket, emit: withinRate(bucket, rate) };
}

node --test test/observe/fields.test.ts

ℹ tests 11
ℹ pass 11
ℹ fail 0
ℹ duration_ms 143.776333