流式测试：正确断言chunk结构而非最终字符串

大多数流式测试的断言逻辑是拼接每个 delta 后与预期语句比较，但这测的是模型而非你的代码。你真正能控制的一切都在 chunk 的结构里，而拼接操作把这些结构信息全部丢弃了。

为什么拼接字符串是错误的主题

两个问题，各自指向相反的方向。拼接后的字符串对你无法控制的部分过于严格——模型可以在不同运行之间、不同版本之间、以及同一版本的非零温度下改写答案，所以精确比对会因为与你的改动无关的原因而失败。而对你能够控制的部分，它又过于宽松：一个流如果把每个 token 放在一个 chunk 里传出、次序错误、第一个 delta 重复、缺少终止帧、或者缓冲了九秒后才发出，拼出来的字符串和一个正常的流完全一样。

用 toContain 或正则来放宽精确比对并不能解决任何一个问题。它只是通过断言几乎什么都没测来解决了不稳定性，但仍然没有说明顺序、帧结构或终止情况。出路在于停止对文本本身做断言，转而对文本到达时的结构做断言。

记录结构的收集器

写一个辅助函数来消费流并返回断言可能需要的一切信息。这样做一次意味着下面的每个断言都是单独的一行，这正是让写多个断言变得可以忍受的原因。

type Collected = {
  events: any[];        // every parsed data payload, in arrival order
  text: string;         // the concatenation, for the few cases that need it
  deltaCount: number;   // frames that carried content
  finishReasons: (string | null)[];
  sawTerminal: boolean;
};

async function collect(res: Response): Promise<Collected> {
  const out: Collected = {
    events: [], text: "", deltaCount: 0, finishReasons: [], sawTerminal: false,
  };
  let buffer = "";
  const reader = res.body!.pipeThrough(new TextDecoderStream()).getReader();
  for (;;) {
    const { value, done } = await reader.read();
    if (done) break;
    buffer += value;
    let i: number;
    while ((i = buffer.indexOf("\n\n")) !== -1) {
      const frame = buffer.slice(0, i);
      buffer = buffer.slice(i + 2);
      for (const line of frame.split("\n")) {
        if (!line.startsWith("data:")) continue;      // skip ": " comments and event: lines
        const payload = line.slice(5).trimStart();
        if (payload === "[DONE]") { out.sawTerminal = true; continue; }
        const event = JSON.parse(payload);
        out.events.push(event);
        const choice = event.choices?.[0];
        if (typeof choice?.delta?.content === "string" && choice.delta.content !== "") {
          out.text += choice.delta.content;
          out.deltaCount++;
        }
        if (choice) out.finishReasons.push(choice.finish_reason);
      }
    }
  }
  return out;
}

payload === "[DONE]" 分支必须放在 JSON.parse 之前。OpenAI 的终止行是字面量 [DONE]，对这个 schema 来说不是合法 JSON，所以先解析的收集器会在每个成功流的最后一帧上抛出异常。Anthropic 的 Messages API 根本没有这样的哨兵值，所以同一个收集器在那里需要不同的终止条件——它那个叫 message_stop 的命名事件。

对 chunk 能做出的最强断言是它符合 provider 文档中的 schema，因为这适用于每一个 prompt，并且能捕获 provider 改变其传输格式的情况。验证每一个事件，而不是只验证第一个。

import { z } from "zod";

const Chunk = z.object({
  id: z.string(),
  object: z.literal("chat.completion.chunk"),
  model: z.string(),
  choices: z.array(
    z.object({
      index: z.number().int(),
      delta: z.object({
        role: z.literal("assistant").optional(),
        content: z.string().nullable().optional(),
        tool_calls: z.array(z.any()).optional(),
      }),
      finish_reason: z
        .enum(["stop", "length", "tool_calls", "content_filter", "function_call"])
        .nullable(),
    }),
  ),
  usage: z.any().nullable().optional(),
});

test("every chunk matches the documented shape", async () => {
  const { events } = await collect(await callEndpoint("hello"));
  for (const [i, event] of events.entries()) {
    const parsed = Chunk.safeParse(event);
    expect(parsed.success, "chunk " + i + ": " + JSON.stringify(event)).toBe(true);
  }
});

注意这个 enum 列出了每一个有文档记录的 finish_reason 值，而不仅仅是 stop。一个只接受愉快路径值的 schema 会把合法的长度截断变成 schema 失败，你会为此花上半天。在失败消息中包含 index：六十个 chunk 的情况下，单独的"expected true, got false"几乎毫无用处。

顺序才是真正不变量所在的地方，因为这些是 provider 保证的、你的代理可能破坏的属性。对于 OpenAI 形状的流：

delta.role 出现在第一个 chunk 上，之后不会再出现。断言 events.filter((e) => e.choices[0].delta.role).length 为 1，且匹配在 index 0。

恰好一个非 null 的 finish_reason，且它在有 choices 条目的最后一个 chunk 上。断言 finishReasons.filter(Boolean) 长度为 1。

带有 finish_reason 的 chunk 之后不会收到任何 content delta。这是能捕获代理重新排序或重发帧的断言。

带有 stream_options: {"include_usage": true} 时，usage 在一个额外的 chunk 上到达，其 choices 数组为空；之前所有 chunk 都带 usage: null。断言携带 usage 的 chunk 是最后一个，且没有 chunk 同时携带 content 和 usage。

Tool calls 增加了一条额外的顺序规则，很容易漏掉。当模型在流式输出时调用工具，delta 携带一个 tool_calls 数组，其条目有一个 index，且对于给定 index 只有第一个 delta 携带 call id 和函数名；该 index 后续每个 delta 只携带 function.arguments 的片段，别的什么都没有。所以不变量是：对于每个 index，恰好有一个 delta 提供 name，且它是第一个，没有 delta 两次提供 name。一个用覆盖而非拼接的累加器会通过纯文本测试但在这里失败，这就是为什么这个断言应该放在同一个文件里而不是 tool-calling 测试套件中。

对于 Anthropic 形状的流，等价的不变量是关于事件名的：content_block_start 在同一 index 的任何 delta 之前出现，每个启动的 block 都由 content_block_stop 关闭，message_stop 是最后一个。有用的单一断言是事件类型的序列，过滤掉 ping 后，与类型名上的一个正则表达式匹配——这用一行表达了整个语法，且失败时会产生可读的 diff。

Chunk 数量很诱人，但大多数时候是个陷阱。一个 provider 为给定答案发出多少个 chunk 取决于 token 化、推理服务器内部的批处理和网络合并；这不是任何公开发布的契约的一部分，断言 deltaCount === 14 会给你一个在好日子里也会失败的测试。可以断言的是：

针对固定装置，精确计数没问题。当上游是你自己的 ReadableStream 时，数量完全由你决定，断言它能捕获你的代理合并或拆分帧。

针对真实 provider，断言一个下界。expect(deltaCount).toBeGreaterThan(1) 是区分流式和非流式的断言，且它是稳定的。

断言关系而非数量。text 等于按顺序拼接的 deltas，也等于同一次请求在温度为零时的非流式响应。这两个端点之间的这个关系是一个真正的不变量，比任何计数都更有价值。

同样的推理适用于 token 总数：只将流报告的 usage 数值与你自己统计的做一个近似比较，因为两者是由不同代码路径产生的，可能合法地不同——详见 token count mismatch。你断言的是 usage 数值确实到达了、恰好一次，且在 provider 文档指定的帧上。

一个看起来像计数但不是计数的断言：随时间推移的增量性。记录每个帧到达时的时间戳及其内容，并断言第一个帧有意义地早于最后一个帧到达。针对有意的入队间隔的固定装置，这是确定性的，并且它是这组中断言中唯一一个在代理收集整个响应然后一次性 flush 时会失败的断言。它只花一个数组，是区分"你的端点返回了正确的字节"和"你的端点在流式传输它们"之间的差别。

Testing Reassembly of Streamed Tokens Into a Full Message

Testing That a Stream Closes Cleanly on the Happy Path

Contract Tests for Streaming Chunk Format