feat: add OpenAI-compatible vision inference server with tracing

- Add Express server exposing OpenAI-compatible chat completions API
- Integrate @huggingface/transformers for vision-language model inference
- Support multimodal user messages with data URL images
- Add Langfuse OpenTelemetry tracing for observability
- Include Swagger UI for API documentation
- Add package.json and package-lock.json with dependencies

package.json

@@ -12,6 +12,7 @@
     "@huggingface/transformers": "next",
     "express": "^4.21.0",
     "onnxruntime-node": "^1.21.0",
-    "swagger-ui-express": "^5.0.0"
+    "swagger-ui-express": "^5.0.0",
+    "sharp": "^0.34.5"
   }
 }

server.js

@@ -3,8 +3,11 @@ import swaggerUi from "swagger-ui-express";
 import {
   AutoProcessor,
   Qwen3_5ForConditionalGeneration,
+  RawImage,
 } from "@huggingface/transformers";
 import crypto from "crypto";
+import { Readable } from "node:stream";
+import { pipeline } from "node:stream/promises";
 import { NodeSDK } from "@opentelemetry/sdk-node";
 import { LangfuseSpanProcessor } from "@langfuse/otel";
 import { startActiveObservation } from "@langfuse/tracing";
@@ -13,6 +16,10 @@ const app = express();
 const PORT = 7860;
 const MODEL_ID = "huggingworld/Qwen3.5-0.8B-ONNX";
 const API_KEY = process.env.API_KEY;
+const NVIDIA_CHAT_URL =
+  process.env.NVIDIA_CHAT_URL || "https://integrate.api.nvidia.com/v1/chat/completions";
+const NVIDIA_API_KEY = process.env.NVIDIA_API_KEY;
+const NVIDIA_FETCH_TIMEOUT_MS = parseInt(process.env.NVIDIA_FETCH_TIMEOUT_MS || "300000", 10);
 const LANGFUSE_PUBLIC_KEY = process.env.LANGFUSE_PUBLIC_KEY;
 const LANGFUSE_SECRET_KEY = process.env.LANGFUSE_SECRET_KEY;
 const LANGFUSE_BASE_URL = process.env.LANGFUSE_BASE_URL || "https://cloud.langfuse.com";
@@ -105,6 +112,12 @@ async function withPromptTrace(req, prompt, maxTokens, handler) {
   );
 }
 
+function queueInference(runner) {
+  const task = inferenceQueue.then(() => runner());
+  inferenceQueue = task.catch(() => {});
+  return task;
+}
+
 async function runTextInference(prompt, maxTokens) {
   const conversation = [
     {
@@ -134,17 +147,320 @@ async function runTextInference(prompt, maxTokens) {
 }
 
 function queueTextInference(prompt, maxTokens) {
-  const task = inferenceQueue.then(() => runTextInference(prompt, maxTokens));
-  inferenceQueue = task.catch(() => {});
-  return task;
+  return queueInference(() => runTextInference(prompt, maxTokens));
+}
+
+function decodeDataUrl(url) {
+  const trimmed = url.trim();
+  const m = /^data:([^;,]+)?;base64,(.+)$/i.exec(trimmed);
+  if (!m) return null;
+  const mime = (m[1] || "application/octet-stream").split(";")[0].trim();
+  try {
+    return { mime, buffer: Buffer.from(m[2], "base64") };
+  } catch {
+    return null;
+  }
+}
+
+async function rawImageFromDataUrlOnly(url) {
+  const data = decodeDataUrl(url);
+  if (!data) {
+    return null;
+  }
+  return RawImage.fromBlob(new Blob([data.buffer], { type: data.mime }));
+}
+
+async function runVisionInference(userText, rawImages, maxTokens) {
+  const content = [];
+  for (let i = 0; i < rawImages.length; i++) {
+    content.push({ type: "image" });
+  }
+
+  const textLine = (userText ?? "").trim() || "What is in this image?";
+  content.push({ type: "text", text: textLine });
+
+  const conversation = [
+    {
+      role: "user",
+      content,
+    },
+  ];
+
+  const text = processor.apply_chat_template(conversation, {
+    add_generation_prompt: true,
+    chat_template_kwargs: { enable_thinking: false },
+  });
+
+  const visionArg = rawImages.length === 1 ? rawImages[0] : rawImages;
+  const inputs = await processor(text, visionArg);
+  const output = await model.generate({
+    ...inputs,
+    max_new_tokens: maxTokens,
+    do_sample: false,
+  });
+
+  const promptLength = inputs.input_ids.dims.at(-1);
+  const decoded = processor.batch_decode(output.slice(null, [promptLength, null]), {
+    skip_special_tokens: true,
+  });
+  return decoded[0];
+}
+
+function getLastUserMessage(messages) {
+  if (!Array.isArray(messages)) return null;
+  for (let i = messages.length - 1; i >= 0; i--) {
+    if (messages[i]?.role === "user") return messages[i];
+  }
+  return null;
+}
+
+function flattenUserMultimodalContent(userMessage) {
+  if (!userMessage) return { segments: [] };
+  const c = userMessage.content;
+  /** @type {{ type: string, text?: string, url?: string }[]} */
+  const segments = [];
+
+  if (typeof c === "string") {
+    segments.push({ type: "text", text: c });
+    return { segments };
+  }
+
+  if (!Array.isArray(c)) return { segments };
+
+  for (const part of c) {
+    if (!part || typeof part !== "object") continue;
+    if (part.type === "text" && typeof part.text === "string") {
+      segments.push({ type: "text", text: part.text });
+    } else if (part.type === "image_url") {
+      const imgUrl = part.image_url?.url;
+      if (typeof imgUrl === "string" && imgUrl.length > 0) {
+        segments.push({ type: "image_url", url: imgUrl });
+      }
+    }
+  }
+
+  return { segments };
+}
+
+/** @returns {Promise<boolean>} true if the response was fully sent (do not send fallback) */
+async function proxyNvidiaChatCompletions(req, body, streamRequested, res) {
+  if (!NVIDIA_API_KEY) {
+    log("info", "nvidia_skipped", { request_id: req.requestId, reason: "no_nvidia_api_key" });
+    return false;
+  }
+
+  const controller = new AbortController();
+  const timeoutId = setTimeout(() => controller.abort(), NVIDIA_FETCH_TIMEOUT_MS);
+  try {
+    const upstream = await fetch(NVIDIA_CHAT_URL, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        Authorization: `Bearer ${NVIDIA_API_KEY}`,
+        Accept: streamRequested ? "text/event-stream" : "application/json",
+      },
+      body: JSON.stringify(body),
+      signal: controller.signal,
+    });
+
+    if (!upstream.ok) {
+      let preview = "";
+      try {
+        preview = (await upstream.text()).slice(0, 800);
+      } catch {
+        /* ignore */
+      }
+      log("warn", "nvidia_upstream_rejected", {
+        request_id: req.requestId,
+        status: upstream.status,
+        body_preview: preview,
+      });
+      return false;
+    }
+
+    const ct = upstream.headers.get("content-type") || "";
+
+    if (upstream.body != null && streamRequested && ct.includes("text/event-stream")) {
+      res.status(200);
+      res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
+      res.setHeader("Cache-Control", "no-cache");
+      res.setHeader("Connection", "keep-alive");
+      try {
+        await pipeline(
+          Readable.fromWeb(/** @type {any} */ (upstream.body)),
+          res,
+        );
+      } catch (pipeErr) {
+        log("error", "nvidia_stream_pipe_failed", {
+          request_id: req.requestId,
+          error: pipeErr.message,
+        });
+        if (!res.headersSent) {
+          return false;
+        }
+        if (!res.writableEnded) {
+          try {
+            res.destroy();
+          } catch {
+            /* ignore */
+          }
+        }
+      }
+      return true;
+    }
+
+    const buf = await upstream.text();
+    if (streamRequested && ct.includes("text/event-stream")) {
+      res.status(200);
+      res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
+      res.send(buf);
+      return true;
+    }
+    res.status(200).type(ct || "application/json").send(buf);
+    return true;
+  } catch (err) {
+    log("warn", "nvidia_fetch_failed", {
+      request_id: req.requestId,
+      error: err.message,
+      name: err.name,
+    });
+    return false;
+  } finally {
+    clearTimeout(timeoutId);
+  }
+}
+
+function openAiCompletionJson(assistantText) {
+  const id = `chatcmpl-${crypto.randomUUID()}`;
+  return {
+    id,
+    object: "chat.completion",
+    created: Math.floor(Date.now() / 1000),
+    model: MODEL_ID,
+    choices: [
+      {
+        index: 0,
+        message: { role: "assistant", content: assistantText },
+        finish_reason: "stop",
+      },
+    ],
+  };
+}
+
+function writeFallbackSseChunks(res, assistantText) {
+  const id = `chatcmpl-${crypto.randomUUID()}`;
+  const base = () => ({
+    id,
+    object: "chat.completion.chunk",
+    created: Math.floor(Date.now() / 1000),
+    model: MODEL_ID,
+  });
+
+  res.write(`data: ${JSON.stringify({
+    ...base(),
+    choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }],
+  })}\n\n`);
+
+  res.write(`data: ${JSON.stringify({
+    ...base(),
+    choices: [{ index: 0, delta: { content: assistantText }, finish_reason: null }],
+  })}\n\n`);
+
+  res.write(`data: ${JSON.stringify({
+    ...base(),
+    choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
+  })}\n\n`);
+
+  res.write("data: [DONE]\n\n");
+}
+
+async function fallbackLocalChatCompletion(req, res, body, streamRequested) {
+  if (!model || !processor) {
+    log("error", "fallback_model_unavailable", { request_id: req.requestId });
+    return res.status(503).json({
+      error: { message: "Local ONNX model is not loaded." },
+      source: "local_onnx_unavailable",
+    });
+  }
+
+  const rawMax = typeof body.max_tokens === "number" ? body.max_tokens : 512;
+  const maxTokens =
+    typeof rawMax === "number" && rawMax > 0 ? Math.min(Math.floor(rawMax), 16384) : 512;
+
+  const lastUser = getLastUserMessage(body.messages);
+  const { segments } = flattenUserMultimodalContent(lastUser);
+
+  const textParts = segments
+    .filter((s) => s.type === "text")
+    .map((s) => s.text ?? "")
+    .join("\n");
+  const imageUrls = segments.filter((s) => s.type === "image_url");
+
+  let assistantText;
+  try {
+    if (imageUrls.length > 0) {
+      const unsupported = imageUrls.find((s) => !decodeDataUrl(s.url));
+      if (unsupported) {
+        return res.status(400).json({
+          error: {
+            message:
+              "Local ONNX fallback only accepts embedded images as data:image/...;base64,... URIs.",
+          },
+          source: "local_onnx_fallback",
+        });
+      }
+      const rawImages = await Promise.all(imageUrls.map((s) => rawImageFromDataUrlOnly(s.url)));
+      assistantText = await queueInference(() =>
+        runVisionInference(textParts || "What is in this image?", rawImages, maxTokens),
+      );
+    } else if ((textParts ?? "").trim().length > 0) {
+      assistantText = await queueInference(() => runTextInference(textParts.trim(), maxTokens));
+    } else {
+      return res.status(400).json({
+        error: {
+          message:
+            "Local fallback needs text or image parts on the last user message.",
+        },
+        source: "local_onnx_fallback",
+      });
+    }
+
+    log("info", "local_fallback_completed", {
+      request_id: req.requestId,
+      chars: assistantText?.length ?? 0,
+      had_images: imageUrls.length > 0,
+    });
+
+    if (streamRequested) {
+      res.status(200);
+      res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
+      res.setHeader("Cache-Control", "no-cache");
+      res.setHeader("Connection", "keep-alive");
+      writeFallbackSseChunks(res, assistantText);
+      return res.end();
+    }
+
+    return res.json(openAiCompletionJson(assistantText));
+  } catch (err) {
+    log("error", "local_fallback_failed", {
+      request_id: req.requestId,
+      error: err.message,
+      stack: err.stack,
+    });
+    return res.status(500).json({
+      error: { message: err.message || "Local ONNX inference failed." },
+      source: "local_onnx_fallback",
+    });
+  }
 }
 
 const swaggerDoc = {
   openapi: "3.0.0",
   info: {
     title: "Qwen3.5-0.8B Text API (ONNX)",
-    version: "1.0.0",
-    description: "Text inference API using Qwen3.5-0.8B ONNX with transformers.js",
+    version: "1.1.0",
+    description:
+      "/prompt serves local text. POST /v1/chat/completions tries NVIDIA integrate.api (NVIDIA_API_KEY) then falls back to local ONNX (vision + text) on upstream errors or timeouts.",
   },
   components: {
     securitySchemes: {
@@ -196,6 +512,18 @@ const swaggerDoc = {
         security: [{ ApiKeyAuth: [] }],
       },
     },
+    "/v1/chat/completions": {
+      post: {
+        summary: "OpenAI-compatible chat completions; NVIDIA proxy with local ONNX fallback",
+        responses: {
+          200: { description: "Completion JSON or streamed events" },
+          400: { description: "Invalid input" },
+          401: { description: "Invalid or missing API key" },
+          503: { description: "Model not loaded during fallback" },
+        },
+        security: [{ ApiKeyAuth: [] }],
+      },
+    },
   },
 };
 
@@ -236,9 +564,15 @@ app.use((req, res, next) => {
   next();
 });
 
+const chatJsonParser = express.json({ limit: "50mb" });
+
 app.get("/", (req, res) => {
   log("info", "root_status", { request_id: req.requestId });
-  res.json({ status: "ok", model: MODEL_ID });
+  res.json({
+    status: "ok",
+    model: MODEL_ID,
+    nvidia_chat_completions_configured: Boolean(NVIDIA_API_KEY),
+  });
 });
 
 app.get("/health", (req, res) => {
@@ -246,9 +580,41 @@ app.get("/health", (req, res) => {
     request_id: req.requestId,
     model_loaded: model !== null && processor !== null,
   });
-  res.json({ status: "healthy", model_loaded: model !== null });
+  res.json({
+    status: "healthy",
+    model_loaded: model !== null,
+    nvidia_chat_completions_configured: Boolean(NVIDIA_API_KEY),
+  });
 });
 
+app.post(
+  "/v1/chat/completions",
+  requireApiKey,
+  chatJsonParser,
+  async (req, res) => {
+    const body = req.body;
+    if (!body?.messages || !Array.isArray(body.messages)) {
+      return res.status(400).json({
+        error: { message: "messages (array) is required." },
+      });
+    }
+
+    const streamRequested = Boolean(body.stream);
+    log("info", "chat_completions_request", {
+      request_id: req.requestId,
+      stream: streamRequested,
+      model: body.model,
+      nvidia_configured: Boolean(NVIDIA_API_KEY),
+    });
+
+    const proxied = await proxyNvidiaChatCompletions(req, body, streamRequested, res);
+    if (proxied || res.headersSent) return;
+
+    log("info", "chat_completions_fallback_local", { request_id: req.requestId });
+    return fallbackLocalChatCompletion(req, res, body, streamRequested);
+  },
+);
+
 app.post("/prompt", requireApiKey, express.json(), async (req, res) => {
   const prompt = req.body.prompt;
   const maxTokens = parseInt(req.body.max_tokens) || 256;
@@ -317,6 +683,8 @@ Promise.all([setupTracing(), loadModel()]).then(() => {
       port: PORT,
       model_id: MODEL_ID,
       langfuse_enabled: LANGFUSE_ENABLED,
+      nvidia_chat_url: NVIDIA_CHAT_URL,
+      nvidia_configured: Boolean(NVIDIA_API_KEY),
     });
 
     const shutdown = async (signal) => {