《互联网产品的 AI 架构·模型接入与能力边界》5.1 流式的端到端链路

// file: src/protocol/stream.ts
// tier: L1
// verify: node --test test/core/foundation.test.ts

import type { Usage } from "../core/types.ts";

/**
 * 流式事件是网关对外的协议，不是提供方协议的直出。
 * 增加新事件必须向后兼容：旧客户端忽略未知事件而不是崩溃。
 */
export type StreamEvent =
  | { type: "start"; requestId: string; modelId: string }
  | { type: "delta"; text: string }
  | { type: "usage"; usage: Usage }
  | { type: "done"; finishReason: "stop" | "length" | "tool_call" | "content_filter" }
  | { type: "error"; kind: string; message: string; retryable: boolean };

/** 把事件编码成 SSE 帧；data 只放单行 JSON，避免换行破坏帧边界。 */
export function encodeStreamEvent(event: StreamEvent): string {
  return `data: ${JSON.stringify(event)}\n\n`;
}

/** 解码一帧；空行与注释帧返回 null，格式错误直接抛弃而不是中断整条流。 */
export function decodeStreamEvent(frame: string): StreamEvent | null {
  const line = frame
    .split("\n")
    .map((part) => part.trim())
    .find((part) => part.startsWith("data:"));
  if (!line) {
    return null;
  }
  const payload = line.slice("data:".length).trim();
  if (payload === "" || payload === "[DONE]") {
    return null;
  }
  try {
    return JSON.parse(payload) as StreamEvent;
  } catch {
    return null;
  }
}

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

import type { ChatRequest, ModelProfile, Usage } from "../core/types.ts";
import type { StreamEvent } from "../protocol/stream.ts";
import { estimateCostMicros } from "../protocol/chat.ts";

export type FinishReason = "stop" | "length" | "tool_call" | "content_filter";

/**
 * 提供方增量只描述“上游刚刚发生了什么”，不承担对外协议。
 * 不同厂商的字段名与结束语义都在这一层被抹平。
 */
export type ProviderChunk =
  | { type: "text"; text: string }
  | { type: "usage"; inputTokens: number; outputTokens: number }
  | { type: "finish"; finishReason: FinishReason };

export interface StreamingProvider {
  readonly id: string;
  stream(request: ChatRequest, model: ModelProfile): AsyncIterable<ProviderChunk>;
}

export interface PipelineOptions {
  provider: StreamingProvider;
  request: ChatRequest;
  model: ModelProfile;
}

/**
 * 把提供方增量翻译成统一 StreamEvent。
 * 不变量：start 先于一切；usage 一定在 done 之前；done 只出现一次。
 */
export async function* pipeProviderStream(
  options: PipelineOptions,
): AsyncGenerator<StreamEvent, void, void> {
  const { provider, request, model } = options;

  yield { type: "start", requestId: request.requestId, modelId: model.id };

  let finishReason: FinishReason | null = null;
  let reported: { inputTokens: number; outputTokens: number } | null = null;

  stream: for await (const chunk of provider.stream(request, model)) {
    switch (chunk.type) {
      case "text": {
        if (chunk.text !== "") {
          yield { type: "delta", text: chunk.text };
        }
        break;
      }
      case "usage": {
        reported = {
          inputTokens: chunk.inputTokens,
          outputTokens: chunk.outputTokens,
        };
        break;
      }
      case "finish": {
        finishReason = chunk.finishReason;
        // 结束帧之后的包一律丢弃，避免同一请求出现两个结束原因。
        break stream;
      }
    }
  }

  const inputTokens = reported?.inputTokens ?? 0;
  const outputTokens = reported?.outputTokens ?? 0;
  const usage: Usage = {
    inputTokens,
    outputTokens,
    costMicros: estimateCostMicros(model, inputTokens, outputTokens),
  };

  yield { type: "usage", usage };
  yield { type: "done", finishReason: finishReason ?? "stop" };
}

/** 把事件里的文本拼回整段回复，供测试与非流式兜底复用。 */
export function collectText(events: Iterable<StreamEvent>): string {
  let text = "";
  for (const event of events) {
    if (event.type === "delta") {
      text += event.text;
    }
  }
  return text;
}

node --test test/stream/pipeline.test.ts

✔ start 先于提供方调用被发出 (1.785375ms)
✔ 增量按到达顺序翻译，usage 在 done 之前 (0.265458ms)
✔ 空文本不产生 delta 帧 (0.123084ms)
✔ 结束帧之后的增量被丢弃 (0.126542ms)
✔ 没有结束帧时以 stop 收尾且用量为零 (0.12875ms)
✔ collectText 忽略非 delta 事件 (0.064542ms)
ℹ tests 6
ℹ suites 0
ℹ pass 6
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 152.360375