Node.js 用 fetch 接入 Tikway 流式输出：正确解析 SSE 与中文分块

$env:TIKWAY_API_KEY = "替换为你的 Tikway API Key"

export TIKWAY_API_KEY='替换为你的 Tikway API Key'

import { pathToFileURL } from "node:url";

const ENDPOINT = "https://api.tikway.ai/v1/chat/completions";
const MAX_EVENT_CHARS = 1024 * 1024;

export async function* readSSE(body) {
  const reader = body.getReader();
  const decoder = new TextDecoder("utf-8", { fatal: true });
  let pending = "";
  let dataLines = [];
  let eventName = "";
  let eventChars = 0;

  function acceptLine(line) {
    if (line === "") {
      const event = dataLines.length
        ? { event: eventName || "message", data: dataLines.join("\n") }
        : null;
      dataLines = [];
      eventName = "";
      eventChars = 0;
      return event;
    }
    eventChars += line.length;
    if (eventChars > MAX_EVENT_CHARS) throw new Error("SSE 事件过大。");
    if (line.startsWith(":")) return null; // 注释或心跳

    const colon = line.indexOf(":");
    const field = colon === -1 ? line : line.slice(0, colon);
    let value = colon === -1 ? "" : line.slice(colon + 1);
    if (value.startsWith(" ")) value = value.slice(1);
    if (field === "data") dataLines.push(value);
    if (field === "event") eventName = value;
    return null; // id/retry 等字段在本例中不使用
  }

  try {
    while (true) {
      const { value, done } = await reader.read();
      pending += done ? decoder.decode() : decoder.decode(value, { stream: true });

      while (true) {
        const match = /\r\n|\r|\n/.exec(pending);
        if (!match) break;
        // 末尾的 CR 可能与下一块开头的 LF 组成同一个换行。
        if (!done && match[0] === "\r" && match.index === pending.length - 1) break;
        const event = acceptLine(pending.slice(0, match.index));
        pending = pending.slice(match.index + match[0].length);
        if (event) yield event;
      }
      if (pending.length + eventChars > MAX_EVENT_CHARS) {
        throw new Error("SSE 事件过大。");
      }
      if (done) return; // 没有空行结尾的残缺事件不作为完整事件发出
    }
  } finally {
    try { await reader.cancel(); } catch { /* 清理连接 */ }
    reader.releaseLock();
  }
}

export async function main() {
  const apiKey = process.env.TIKWAY_API_KEY?.trim();
  if (!apiKey) throw new Error("请先设置 TIKWAY_API_KEY。");

  const response = await fetch(ENDPOINT, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${apiKey}`,
      "Content-Type": "application/json",
      Accept: "text/event-stream",
    },
    body: JSON.stringify({
      model: process.env.TIKWAY_MODEL?.trim() || "openai/gpt-5.4-mini",
      messages: [{ role: "user", content: "请用三句话说明 SSE 的用途。" }],
      stream: true,
    }),
    signal: AbortSignal.timeout(90_000),
  });

  const requestId = response.headers.get("x-request-id") || "未提供";
  if (!response.ok) {
    let detail = "请查看错误码文档。";
    try {
      const body = await response.json();
      const apiError = body?.error;
      if (apiError && typeof apiError === "object") {
        detail = `${apiError.code || apiError.type || "unknown"}: `
          + String(apiError.message || "").slice(0, 300);
      }
    } catch { /* 错误响应也可能不是 JSON */ }
    throw new Error(`HTTP ${response.status}；request_id=${requestId}；${detail}`);
  }

  const contentType = response.headers.get("content-type") || "";
  if (!contentType.toLowerCase().includes("text/event-stream") || !response.body) {
    if (response.body) await response.body.cancel();
    throw new Error(`未收到 SSE 响应；request_id=${requestId}`);
  }

  let completed = false;
  for await (const event of readSSE(response.body)) {
    if (event.data.trim() === "[DONE]") {
      completed = true;
      break;
    }
    let payload;
    try { payload = JSON.parse(event.data); }
    catch { throw new Error(`SSE data 不是有效 JSON；request_id=${requestId}`); }

    if (event.event === "error" || payload?.type === "error" || payload?.error) {
      const apiError = payload?.error;
      const detail = apiError && typeof apiError === "object"
        ? `${apiError.code || apiError.type || "unknown"}: `
          + String(apiError.message || "stream interrupted").slice(0, 300)
        : "stream interrupted";
      throw new Error(`流式错误；request_id=${requestId}；${detail}`);
    }

    const choices = Array.isArray(payload?.choices) ? payload.choices : [];
    const choice = choices.find((item) => item?.index === 0) || choices[0];
    const text = choice?.delta?.content;
    if (typeof text === "string") process.stdout.write(text);
    // finish_reason 或 usage 可以先到；继续读取直到 [DONE]。
  }
  if (!completed) throw new Error(`流提前结束，未收到 [DONE]；request_id=${requestId}`);
  process.stdout.write("\n");
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
  main().catch((err) => {
    console.error(`调用未完成：${err.message}`);
    process.exitCode = 1;
  });
}

node tikway_stream.mjs

data: {"choices":[{"index":0,"delta":{"content":"你好"},"finish_reason":null}]}

data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

node 02-sse-offline-demo.mjs

你好，SSE
Local SSE parsing completed (no API calls).