《互联网产品的 AI 架构·模型接入与能力边界》5.3 流式下的错误与重试语义

// file: src/stream/errors.ts
// tier: L1
// verify: node --test test/stream/errors.test.ts

import { normalizeError } from "../core/errors.ts";
import type { StreamEvent } from "../protocol/stream.ts";

/**
 * 流式重试的分界线是“首帧有没有发出去”，不是“已经生成了多少文本”。
 * 首帧前用户什么都没看到，重试是安全的；首帧后重试会把同一段话再说一遍。
 */
export type StreamPhase = "pre-first-frame" | "in-flight" | "completed";

export interface RetryPolicy {
  /** 总尝试次数，包含第一次；默认 2 表示“首次 + 一次重试”。 */
  maxAttempts: number;
}

export const DEFAULT_RETRY_POLICY: RetryPolicy = { maxAttempts: 2 };

/** 与 core/errors 的 AiError 解耦，便于把协议里的 error 事件直接转成决策输入。 */
export interface StreamFailure {
  kind: string;
  message: string;
  retryable: boolean;
}

export interface RetryDecision {
  retry: boolean;
  /** 重试时客户端应从第几个字符之后续接；0 表示从头再来。 */
  resumeFromChars: number;
  reason: string;
}

export function streamPhase(emittedChars: number, completed: boolean): StreamPhase {
  if (completed) {
    return "completed";
  }
  return emittedChars === 0 ? "pre-first-frame" : "in-flight";
}

export function failureFromError(error: unknown): StreamFailure {
  const normalized = normalizeError(error);
  return {
    kind: normalized.kind,
    message: normalized.message,
    retryable: normalized.retryable,
  };
}

export function decideRetry(input: {
  phase: StreamPhase;
  emittedChars: number;
  failure: StreamFailure;
  /** 刚刚失败的这一次是第几次尝试，从 1 开始。 */
  attempt: number;
  policy?: RetryPolicy;
}): RetryDecision {
  const policy = input.policy ?? DEFAULT_RETRY_POLICY;

  if (!input.failure.retryable) {
    return {
      retry: false,
      resumeFromChars: input.emittedChars,
      reason: `${input.failure.kind} 不可重试，重复请求只会重复失败`,
    };
  }

  if (input.phase === "completed") {
    return {
      retry: false,
      resumeFromChars: input.emittedChars,
      reason: "流已经正常结束，没有需要重试的失败",
    };
  }

  if (input.phase === "in-flight") {
    return {
      retry: false,
      resumeFromChars: input.emittedChars,
      reason: "首帧已经发出，透明重试会让用户看到重复文本；只能续传或让用户重新生成",
    };
  }

  if (input.attempt >= policy.maxAttempts) {
    return {
      retry: false,
      resumeFromChars: 0,
      reason: `已达到重试上限 ${policy.maxAttempts}，继续重试只会扩大故障面`,
    };
  }

  return {
    retry: true,
    resumeFromChars: 0,
    reason: "首帧之前失败，客户端没有看到任何文本，可以安全重试",
  };
}

/** 三种收尾方式必须分开：正常结束、协议内错误事件、连接被断开。 */
export type TerminationSource = "done" | "error-event" | "connection-drop";

export interface StreamTermination {
  source: TerminationSource;
  emittedChars: number;
  finishReason?: "stop" | "length" | "tool_call" | "content_filter";
  failure?: StreamFailure;
}

export function classifyTermination(events: readonly StreamEvent[]): StreamTermination {
  let emittedChars = 0;
  let done: Extract<StreamEvent, { type: "done" }> | null = null;
  let errorEvent: Extract<StreamEvent, { type: "error" }> | null = null;

  for (const event of events) {
    if (event.type === "delta") {
      emittedChars += event.text.length;
    } else if (event.type === "done") {
      done = event;
    } else if (event.type === "error") {
      errorEvent = event;
    }
  }

  if (done !== null) {
    return { source: "done", emittedChars, finishReason: done.finishReason };
  }

  if (errorEvent !== null) {
    return {
      source: "error-event",
      emittedChars,
      failure: {
        kind: errorEvent.kind,
        message: errorEvent.message,
        retryable: errorEvent.retryable,
      },
    };
  }

  return {
    source: "connection-drop",
    emittedChars,
    failure: {
      kind: "unavailable",
      message: "连接在结束帧之前断开",
      retryable: true,
    },
  };
}

/** 把收尾分类直接接到重试决策上，避免调用方自己拼 phase。 */
export function retryForTermination(
  termination: StreamTermination,
  attempt: number,
  policy: RetryPolicy = DEFAULT_RETRY_POLICY,
): RetryDecision {
  if (termination.source === "done") {
    return {
      retry: false,
      resumeFromChars: termination.emittedChars,
      reason: "流已经正常结束，没有需要重试的失败",
    };
  }

  const failure = termination.failure ?? {
    kind: "provider",
    message: "unknown stream failure",
    retryable: true,
  };

  return decideRetry({
    phase: streamPhase(termination.emittedChars, false),
    emittedChars: termination.emittedChars,
    failure,
    attempt,
    policy,
  });
}

node --test test/stream/errors.test.ts

✔ 首帧前后由已输出字符数划分 (1.005125ms)
✔ 核心错误被翻译成可重试标记 (0.125791ms)
✔ 首帧前可重试，且从头开始 (0.149125ms)
✔ 首帧后不透明重试，保留已输出位置 (0.139917ms)
✔ 不可重试错误即使发生在首帧前也不重试 (0.082625ms)
✔ 达到重试上限后停止 (0.069041ms)
✔ 正常结束的流不触发重试 (0.133875ms)
✔ 协议内错误事件保留原始 kind 与 retryable (0.438ms)
✔ 连接断开被归为 unavailable，与错误事件分工明确 (0.095959ms)
✔ 错误事件在首帧前失败时可以重试，断连在首帧前也可以重试 (0.124334ms)
ℹ tests 10
ℹ suites 0
ℹ pass 10
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 195.000417