aiartstudio/server/utils/openai.ts

// server/utils/openai.ts - OpenAI/NewAPI 调用工具：文本、视觉、Responses 流式和 Chat Completions 流式生图。
import OpenAI from "openai";
import type { BaseOptions, IImageGenerateData } from "#shared/types/openai";

/** 图像生成改走 Chat Completions 流式接口，通常能更快拿到中转平台返回的图片地址 */
const CHAT_COMPLETIONS_URL = "https://api.qflink.xyz/v1/chat/completions";
const IMAGE_GENERATION_MODEL = "gpt-image-2";

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

/** 聚合后的流式生图上游响应，供服务端入库排查使用，不包含 API Key */
interface IImageStreamUpstreamResponse {
  content: string;
  imageUrl?: string;
  chunks: Array<{
    id?: string;
    created?: number;
    model?: string;
    content?: string;
    finishReason?: string | null;
    metadata?: unknown;
    hasUsage: boolean;
  }>;
  usage?: unknown;
}

export interface IAskImageResult extends IImageGenerateData {
  /** 完整上游生图接口返回结果，仅服务端内部保存 */
  upstreamResponse: unknown;
}

/** 通用 AI 调用函数（支持文本 / 图文 / 多模态） */
export const askAI = async ({
  apiKey,
  model,
  baseURL,
  input
}: BaseOptions & {
  input: OpenAI.Responses.ResponseCreateParams["input"];
}) => {
  const client = new OpenAI({ apiKey, baseURL });

  return await client.responses.create({
    model,
    input
  });
};

/** 调用文本模型 */
export const askText = async (
  options: BaseOptions & {
    text: string;
  }
) => {
  const res = await askAI({
    ...options,
    input: options.text
  });

  return res.output_text;
};

/**
 * 调用视觉模型：输入「文本 + 图片」，输出文本
 *
 * 分辨率参数说明：
 *
 * | 值   | 含义                     |
 * |------|--------------------------|
 * | low  | 低分辨率（更快、更省成本） |
 * | high | 高分辨率（更精准）         |
 * | auto | 自动选择                 |
 */
export const askVision = async (
  options: BaseOptions & {
    text: string;
    image: string;
    detail?: "low" | "high" | "auto";
  }
) => {
  // 如果图片不是 URL，也不是 data URL，就当成 base64 处理，自动补全 data URL 前缀
  const normalizeImage = (img: string) => {
    if (img.startsWith("http")) return img;
    if (!img.startsWith("data:")) {
      return `data:image/png;base64,${img}`;
    }
    return img;
  };

  const res = await askAI({
    ...options,
    input: [
      {
        role: "user",
        content: [
          {
            type: "input_text",
            text: options.text
          },
          {
            type: "input_image",
            image_url: normalizeImage(options.image),
            detail: options.detail || "auto"
          }
        ]
      }
    ]
  });

  return res.output_text;
};

/** 调用流式图片生成接口，完整 API Key 只在服务端使用 */
export const askImgStream = async ({
  apiKey,
  prompt
}: {
  apiKey: string;
  prompt: string;
}): Promise<IAskImageResult> => {
  const response = await fetch(CHAT_COMPLETIONS_URL, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${apiKey}`,
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      model: IMAGE_GENERATION_MODEL,
      stream: true,
      messages: [
        {
          role: "user",
          content: prompt
        }
      ]
    })
  });

  if (!response.ok) {
    const message = await response.text().catch(() => "");
    throw new Error(message || `图片生成失败：${response.status}`);
  }

  if (!response.body) {
    throw new Error("图片生成失败：上游没有返回流");
  }

  const upstreamResponse = await readChatCompletionStream(response.body);
  const imageUrl = extractImageUrlFromStreamText(upstreamResponse.content);

  return {
    imageUrl,
    revisedPrompt: undefined,
    upstreamResponse: {
      ...upstreamResponse,
      imageUrl
    }
  };
};

/** 从流式累积文本中提取图片地址，兼容 Markdown 图片、常见图片 URL 和无扩展下载链接 */
export const extractImageUrlFromStreamText = (text: string): string => {
  const markdownImageMatch = text.match(/!\[[^\]]*]\((https?:\/\/[^)\s]+)\)/i);
  if (markdownImageMatch?.[1]) {
    return cleanupImageUrl(markdownImageMatch[1]);
  }

  const imageUrlMatch = text.match(
    /https?:\/\/[^\s)>'"]+(?:\.(?:png|jpe?g|webp|gif)(?:\?[^\s)>'"]*)?|\/file_download\/[^\s)>'"]+)/i
  );
  if (imageUrlMatch?.[0]) {
    return cleanupImageUrl(imageUrlMatch[0]);
  }

  const urlMatch = text.match(/https?:\/\/[^\s)>'"]+/i);
  if (urlMatch?.[0]) {
    return cleanupImageUrl(urlMatch[0]);
  }

  throw new Error("图片生成失败：未找到图片地址");
};

/** 调用 Responses API 流式接口，保留给其他文本/多模态场景复用 */
export const askStream = async (
  options: BaseOptions & {
    input: OpenAI.Responses.ResponseCreateParams["input"];
  }
) => {
  const client = new OpenAI({
    apiKey: options.apiKey,
    baseURL: options.baseURL
  });

  const stream = await client.responses.stream({
    model: options.model,
    input: options.input
  });

  return stream;
};

/** 读取 Chat Completions SSE 流，累积 delta.content 并提取 usage/chunk 元信息 */
const readChatCompletionStream = async (
  stream: ReadableStream<Uint8Array>
): Promise<IImageStreamUpstreamResponse> => {
  const reader = stream.getReader();
  const decoder = new TextDecoder();
  // TODO: 这里会把完整 SSE 文本和 chunk 元数据都累计到当前 Node 进程内存里，
  // 高并发或超长响应下可能带来明显的事件循环抖动；后续可改为限制缓存窗口或只保留摘要。
  const chunks: IImageStreamUpstreamResponse["chunks"] = [];
  let buffer = "";
  let content = "";
  let usage: unknown;

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

    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split(/\r?\n/);
    buffer = lines.pop() ?? "";

    for (const line of lines) {
      const chunk = parseSseDataLine(line);
      if (!chunk) continue;

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

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

      chunks.push({
        id: chunk.id,
        created: chunk.created,
        model: chunk.model,
        content: deltaContent || undefined,
        finishReason: chunk.choices?.find((choice) => choice.finish_reason)
          ?.finish_reason,
        metadata: chunk.metadata,
        hasUsage: Boolean(chunk.usage)
      });
    }
  }

  const finalText = buffer + decoder.decode();
  for (const line of finalText.split(/\r?\n/)) {
    const chunk = parseSseDataLine(line);
    if (!chunk) continue;

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

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

    chunks.push({
      id: chunk.id,
      created: chunk.created,
      model: chunk.model,
      content: deltaContent || undefined,
      finishReason: chunk.choices?.find((choice) => choice.finish_reason)
        ?.finish_reason,
      metadata: chunk.metadata,
      hasUsage: Boolean(chunk.usage)
    });
  }

  return {
    content,
    chunks,
    usage
  };
};

/** 解析单行 SSE data，跳过空行和 [DONE] */
const parseSseDataLine = (line: string): IChatCompletionStreamChunk | null => {
  const trimmed = line.trim();
  if (!trimmed.startsWith("data:")) return null;

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

  // TODO: JSON.parse 仍运行在主线程；如果后续接入更大的 SSE chunk，需要增加单行大小保护，
  // 避免异常大包在单个 tick 内长时间占用事件循环。
  return JSON.parse(data) as IChatCompletionStreamChunk;
};

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

/** 清理模型文本里 URL 后面可能粘上的句末标点 */
const cleanupImageUrl = (url: string) => {
  return url.replace(/[,.!?，。！？]+$/u, "");
};