OCS_service/server/utils/openai.ts

// server/utils/openai.ts - OpenAI Chat Completions 流式调用工具
import { serverEnv } from "~~/server/utils/env";

/** 旧项目的 `OPENAI_API_BASE` 语义是完整 base，例如 `https://api.openai.com/v1` */
const CHAT_COMPLETIONS_URL = `${serverEnv.openAiApiBase}/chat/completions`;
/** 复用编码器计算 SSE data 字节长度，避免循环里频繁创建对象 */
const textEncoder = new TextEncoder();

/** Chat Completions SSE 每个 `data:` chunk 的最小结构 */
interface ChatCompletionStreamChunk {
  id?: string;
  object?: string;
  created?: number;
  model?: string;
  choices?: Array<{
    delta?: {
      content?: string;
      role?: string;
    };
    finish_reason?: string | null;
    index?: number;
  }>;
  usage?: unknown;
}

/** 读取完整 SSE 后汇总出的信息；只在服务端日志或内部状态中使用 */
interface ChatCompletionStreamResult {
  /** 累积的 `choices[].delta.content` 文本 */
  content: string;
  /** content 字符长度，方便排查模型是否返回了异常长文本 */
  contentLength: number;
  /** 实际解析到的 SSE JSON chunk 数量 */
  chunkCount: number;
  /** 模型返回的结束原因，例如 stop、length；可能不存在 */
  finishReason?: string | null;
  /** 上游如果返回 usage，则保留下来供内部排查 */
  usage?: unknown;
}

/** 答题调用结果：对外只用 answer，upstreamResponse 留给服务端排查 */
export interface AskAnswerStreamResult {
  answer: string;
  upstreamResponse: ChatCompletionStreamResult;
}

/**
 * 通过 Chat Completions 流式接口生成答案
 *
 * 对 OCS 客户端仍返回普通 JSON；流式只发生在服务端到 OpenAI 之间
 * 这样既能更早读取上游内容，又不破坏 AnswererWrapper 的 handler 契约
 */
export const askAnswerStream = async ({
  prompt,
  systemPrompt
}: {
  prompt: string;
  systemPrompt: string;
}): Promise<AskAnswerStreamResult> => {
  if (!serverEnv.openAiApiKey) {
    throw new Error("OPENAI_API_KEY is not set.");
  }

  // 这里不要记录 request body：其中包含完整题目和可能的选项文本
  const response = await fetch(CHAT_COMPLETIONS_URL, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${serverEnv.openAiApiKey}`,
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      model: serverEnv.openAiModel,
      temperature: serverEnv.temperature,
      max_tokens: serverEnv.maxTokens,
      stream: true,
      messages: [
        {
          role: "system",
          content: systemPrompt
        },
        {
          role: "user",
          content: prompt
        }
      ]
    })
  });

  if (!response.ok) {
    throw new Error(`OpenAI request failed with status ${response.status}.`);
  }

  if (!response.body) {
    throw new Error("OpenAI stream body is empty.");
  }

  // 将 SSE delta 读完后再返回给 API handler，由 handler 统一做答案清洗和缓存
  const upstreamResponse = await readChatCompletionStream(response.body);

  return {
    answer: upstreamResponse.content.trim(),
    upstreamResponse
  };
};

/**
 * 读取 Chat Completions SSE 流
 *
 * 上游返回形如：
 * `data: {"choices":[{"delta":{"content":"..."}}]}`
 * `data: [DONE]`
 *
 * 网络 chunk 不一定按行对齐，所以用 buffer 保存尚未拼完整的一行
 */
export const readChatCompletionStream = async (
  stream: ReadableStream<Uint8Array>
): Promise<ChatCompletionStreamResult> => {
  const reader = stream.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  let content = "";
  let chunkCount = 0;
  let finishReason: string | null | undefined;
  let usage: unknown;

  /** 把一个已解析 JSON chunk 合并进最终结果 */
  const collectChunk = (chunk: ChatCompletionStreamChunk) => {
    chunkCount += 1;

    const deltaContent = collectDeltaContent(chunk);
    if (deltaContent) {
      content += deltaContent;
    }

    if (chunk.usage) {
      usage = chunk.usage;
    }

    finishReason =
      chunk.choices?.find((choice) => choice.finish_reason)?.finish_reason ??
      finishReason;
  };

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    // stream: true 可以正确处理跨 chunk 的多字节字符，例如中文
    buffer += decoder.decode(value, { stream: true });
    assertSseBufferSize(buffer, serverEnv.maxSseLineBytes);

    // 只处理已经遇到换行的完整 SSE 行，最后一段留到下次 chunk 再拼
    const lines = buffer.split(/\r?\n/u);
    buffer = lines.pop() ?? "";

    for (const line of lines) {
      const chunk = parseSseDataLine(line, serverEnv.maxSseLineBytes);
      if (chunk) collectChunk(chunk);
    }
  }

  // 处理流结束时仍留在 buffer 中的最后一行
  const finalText = buffer + decoder.decode();
  for (const line of finalText.split(/\r?\n/u)) {
    const chunk = parseSseDataLine(line, serverEnv.maxSseLineBytes);
    if (chunk) collectChunk(chunk);
  }

  return {
    content,
    contentLength: content.length,
    chunkCount,
    finishReason,
    usage
  };
};

/**
 * 解析单行 SSE data
 *
 * 空行、event 行和 `[DONE]` 都不包含业务内容，直接跳过
 * 超过大小限制的 data 行视为异常，防止上游或代理故障导致内存压力
 */
export const parseSseDataLine = (
  line: string,
  maxDataLineBytes: number
): ChatCompletionStreamChunk | null => {
  const trimmed = line.trim();
  if (!trimmed.startsWith("data:")) return null;

  const data = trimmed.slice("data:".length).trim();
  if (!data || data === "[DONE]") return null;

  if (textEncoder.encode(data).byteLength > maxDataLineBytes) {
    throw new Error("OpenAI stream data line is too large.");
  }

  return JSON.parse(data) as ChatCompletionStreamChunk;
};

/** 收集一个 chunk 中所有 choice 的增量文本 */
export const collectDeltaContent = (chunk: ChatCompletionStreamChunk) => {
  return (
    chunk.choices
      ?.map((choice) => choice.delta?.content || "")
      .filter(Boolean)
      .join("") || ""
  );
};

/**
 * 限制尚未出现换行的 SSE 缓冲区大小
 *
 * 正常 SSE 会频繁换行；如果一直没有换行且持续变大，说明可能不是合法 SSE
 * 或上游返回了异常长单行，应该尽早失败
 */
export const assertSseBufferSize = (
  buffer: string,
  maxDataLineBytes: number
) => {
  if (
    !buffer.includes("\n") &&
    textEncoder.encode(buffer).byteLength > maxDataLineBytes
  ) {
    throw new Error("OpenAI stream buffer is too large.");
  }
};