Add support OpenAI Threads (/v1/threads) via proxy server with payloads transform to Chat (/v1/chat/completions)

[zmnv]

Hello. I am currently developing a VS Code extension that interacts with LLMs via API requests to:
https://llama-cpp-python.readthedocs.io/en/latest/server/

I have noticed that there is no /v1/threads support in the llama.cpp server.

To address this, I have created a simple express.js server to handle that route,
and I would like to share it with you as it is (it's not a perfect).

Perhaps someone can add this route to llama-cpp[server] based on this information.
I have not found any information about the /v1/threads route.

This code consists of:

• code only for the streaming mode
• /v1/threads/runs – payloads are converted from /v1/threads/runs to /v1/chat/completions and vice versa.
• /v1/threads/:threadId/runs/:runId/cancel – returns mocked data
• mocked thread_id and run_id values

You can help me with:

• handle with not streaming mode
• add other API routes for threads...
• add storage for messages history by thread id...

const express = require("express");
const axios = require("axios");
var cors = require("cors");

const app = express();

app.use(
  cors({
    origin: "*",
    optionsSuccessStatus: 200, // some legacy browsers (IE11, various SmartTVs) choke on 204
  })
);

const HOST = "http://localhost:8000";
const OPENAI_API_KEY = "sk-XXX";
const CHAT_COMPLETION_ADDRESS = `${HOST}/v1/chat/completions`;
const threadId = "thread_id_callback";
const runId = "run_id_callback";

app.use(express.json());

app.post("/v1/threads/runs", async (req, res) => {
  const { assistant_id, thread, stream, model } = req.body;

  const messages = thread.messages.map((message) => ({
    role: message.role,
    content: message.content,
  }));

  try {
    const response = await axios.post(
      CHAT_COMPLETION_ADDRESS,
      {
        model,
        messages,
        stream: true,
      },
      {
        headers: {
          "Content-Type": "application/json",
          Authorization: `Bearer ${OPENAI_API_KEY}`,
        },
        responseType: "stream",
      }
    );

    res.setHeader("Content-Type", "text/event-stream");
    res.setHeader("Cache-Control", "no-cache");
    res.setHeader("Connection", "keep-alive");

    res.write(
      `event: thread.created\ndata: ${JSON.stringify({
        id: threadId,
        object: "thread",
        created_at: Date.now(),
        metadata: {},
      })}\n\n`
    );

    res.write(
      `event: thread.run.created\ndata: ${JSON.stringify({
        id: runId,
        object: "thread.run",
        created_at: Date.now(),
        assistant_id: assistant_id,
        thread_id: threadId,
        status: "queued",
        model,
        metadata: {},
      })}\n\n`
    );

    res.write(
      `event: thread.run.queued\ndata: ${JSON.stringify({
        id: runId,
        object: "thread.run",
        created_at: Date.now(),
        assistant_id: assistant_id,
        thread_id: threadId,
        status: "queued",
      })}\n\n`
    );

    res.write(
      `event: thread.run.in_progress\ndata: ${JSON.stringify({
        id: runId,
        object: "thread.run",
        created_at: Date.now(),
        assistant_id: assistant_id,
        thread_id: threadId,
        status: "in_progress",
      })}\n\n`
    );

    response.data.on("data", (chunk) => {
      const data = chunk.toString().split("\n").filter(Boolean);
      data.forEach((line) => {
        if (line.startsWith("data:")) {
          const jsonData = line.replace("data: ", "");
          if (jsonData === "[DONE]") {
            res.write("event: done\ndata: [DONE]\n\n");
            res.end();
          } else {
            try {
              const json = JSON.parse(jsonData);
              if (json.choices && json.choices.length > 0) {
                const content = json.choices[0].delta.content;
                if (content) {
                  res.write(
                    `event: thread.message.delta\ndata: ${JSON.stringify({
                      id: "msg_001",
                      object: "thread.message.delta",
                      delta: {
                        content: [
                          { index: 0, type: "text", text: { value: content } },
                        ],
                      },
                    })}\n\n`
                  );
                }
              }
            } catch (error) {
              console.error("Error parsing JSON:", error);
            }
          }
        }
      });
    });

    response.data.on("end", () => {
      if (!res.headersSent) {
        res.write("event: done\ndata: [DONE]\n\n");
        res.end();
      }
    });

    response.data.on("error", (error) => {
      console.error("Stream error:", error);
      res.status(500).send("Stream Error");
    });
  } catch (error) {
    console.error("Error:", error);
    res.status(500).send("Internal Server Error");
  }
});

app.post("/v1/threads/:threadId/runs/:runId/cancel", (req, res) => {
  const { threadId, runId } = req.params;
  console.log("STOP", params);

  const response = {
    id: runId,
    object: "thread.run",
    created_at: Date.now(),
    assistant_id: "asst_abc123",
    thread_id: threadId,
    status: "cancelling",
    started_at: Date.now() - 60000,
    expires_at: Date.now() + 60000,
    cancelled_at: null,
    failed_at: null,
    completed_at: null,
    last_error: null,
    model: "gpt-4o",
    instructions: "You summarize books.",
    tools: [
      {
        type: "file_search",
      },
    ],
    tool_resources: {
      file_search: {
        vector_store_ids: ["vs_123"],
      },
    },
    metadata: {},
    usage: null,
    temperature: 1.0,
    top_p: 1.0,
    response_format: "auto",
  };

  res.json(response);
});

const PORT = process.env.PORT || 8001;
app.listen(PORT, () => {
  console.log(`/v1/threads Backward Compatibility Proxy ${PORT}`);
});

I use something like that to communicate with

async function invokeMessage(...) {
 try {
    // custom_response is any function with fetching data to llama.cpp API
    // result of `fetch('.../v1/threads/runs', ...)`
    const response = custom_response;

    const reader = response?.body?.getReader();
    const decoder = new TextDecoder();
    let done = false;
    let partialData = "";

    assistantMessageRef.current = "";
    const prevMessages: Message[] = getMessages();

    while (!done) {
      const { value, done: doneReading } = (await reader?.read()) || {};
      done = !!doneReading;
      const chunk = decoder.decode(value);

      const lines = (partialData + chunk)
        .split("\n")
        .filter((line) => line.trim() !== "");
      partialData = "";

      for (const line of lines) {
        if (line.startsWith("data: ")) {
          const json = line.replace("data: ", "");

          if (json === "[DONE]") {
            setError(null);
            setLoading(false);
            if (onDone) {
              onDone();
            }
            return;
          }

          partialData += json;

          try {
            const parsedData = JSON.parse(partialData);
            partialData = ""; 

            if (parsedData?.run_id || parsedData?.thread_id) {
              const runId = parsedData.run_id;
              const threadId = parsedData.thread_id;
              setRunId?.(runId, threadId); // Сохраняем run_id
            }

            const deltaContent = parsedData?.delta?.content;
            if (deltaContent && deltaContent.length > 0) {
              for (const content of deltaContent) {
                const token = content?.text?.value || "";

                if (!token && !assistantMessageRef.current) {
                  continue;
                }

                assistantMessageRef.current += token;

                const lastMessage = prevMessages[prevMessages.length - 1];
                if (lastMessage?.role === "assistant" && !lastMessage?._type) {
                  lastMessage.content = assistantMessageRef.current;
                  setMessages([...prevMessages.slice(0, -1), lastMessage]);
                }

                setMessages([
                  ...prevMessages,
                  { role: "assistant", content: assistantMessageRef.current },
                ]);
              }
            }
          } catch (error: any) {
            if (error instanceof SyntaxError) {
              console.warn("Partial JSON data, waiting for more chunks...");
            } else {
              console.error("Parsing error:", error);
              partialData = ""; 
            }
          }
        }
      }
    }
  } catch (error: any) {
    if (error.name !== "AbortError") {
      console.error("Error communicating with OpenAI:", error);
      setError(new Error(error).message);
    }
    setLoading(false);
    throw error;
  }
}