feat: add an OpenAI-compatible model adapter example

examples/README.md

@@ -9,3 +9,21 @@ npm run demo:command
 ```
 
 Replace its deterministic response with a local model call or another inference client while preserving the same request and chunk protocol.
+
+## OpenAI-compatible servers
+
+`openai-compatible-model.mjs` connects Relay Chat to any server with a streaming `/chat/completions` endpoint. It defaults to [Impossible Inference](https://github.com/impossibleG/impossible-inference) on `http://127.0.0.1:8080/v1`:
+
+```sh
+relay-chat -- node examples/openai-compatible-model.mjs
+```
+
+Configure it with environment variables:
+
+| Variable                | Default                        |
+| ----------------------- | ------------------------------ |
+| `RELAY_OPENAI_BASE_URL` | `http://127.0.0.1:8080/v1`     |
+| `RELAY_OPENAI_MODEL`    | `qwen2.5-0.5b-instruct-q4-k-m` |
+| `RELAY_OPENAI_API_KEY`  | none; sent as a bearer token   |
+
+A model set with `--model` or `/model` takes precedence over `RELAY_OPENAI_MODEL`. For Ollama, set `RELAY_OPENAI_BASE_URL=http://127.0.0.1:11434/v1` and choose an installed model.

examples/openai-compatible-model.mjs

@@ -0,0 +1,97 @@
+// Relay Chat command adapter for OpenAI-compatible chat-completions servers,
+// such as Impossible Inference (the default) or Ollama.
+import readline from 'node:readline';
+
+const baseUrl = (process.env.RELAY_OPENAI_BASE_URL ?? 'http://127.0.0.1:8080/v1').replace(
+  /\/+$/,
+  '',
+);
+const defaultModel = process.env.RELAY_OPENAI_MODEL ?? 'qwen2.5-0.5b-instruct-q4-k-m';
+const apiKey = process.env.RELAY_OPENAI_API_KEY;
+
+const write = (chunk) => process.stdout.write(`${JSON.stringify(chunk)}\n`);
+
+const errorMessage = async (response) => {
+  const body = await response.text();
+  try {
+    const parsed = JSON.parse(body);
+    const message = parsed?.error?.message ?? parsed?.error;
+    if (typeof message === 'string') return `${response.status}: ${message}`;
+  } catch {
+    // Fall through to the raw body.
+  }
+  return `${response.status}: ${body.slice(0, 500) || response.statusText}`;
+};
+
+async function* serverEvents(body) {
+  const decoder = new TextDecoder();
+  let buffer = '';
+  for await (const bytes of body) {
+    buffer += decoder.decode(bytes, {stream: true});
+    let boundary;
+    while ((boundary = buffer.search(/\r?\n\r?\n/)) !== -1) {
+      const block = buffer.slice(0, boundary);
+      buffer = buffer.slice(boundary).replace(/^\r?\n\r?\n/, '');
+      const data = block
+        .split(/\r?\n/)
+        .filter((line) => line.startsWith('data:'))
+        .map((line) => line.slice(5).replace(/^ /, ''))
+        .join('\n');
+      if (data !== '') yield data;
+    }
+  }
+}
+
+const answer = async (request) => {
+  const model = request.model ?? defaultModel;
+  const messages = [
+    ...(request.systemPrompt ? [{role: 'system', content: request.systemPrompt}] : []),
+    ...request.messages,
+  ];
+  let response;
+  try {
+    response = await fetch(`${baseUrl}/chat/completions`, {
+      method: 'POST',
+      headers: {
+        'content-type': 'application/json',
+        accept: 'text/event-stream',
+        ...(apiKey ? {authorization: `Bearer ${apiKey}`} : {}),
+      },
+      body: JSON.stringify({model, messages, stream: true}),
+    });
+  } catch (error) {
+    return write({type: 'error', message: `Could not reach ${baseUrl}: ${error.message}`});
+  }
+  if (!response.ok || response.body === null) {
+    return write({type: 'error', message: await errorMessage(response)});
+  }
+
+  let reportedModel = false;
+  for await (const data of serverEvents(response.body)) {
+    if (data === '[DONE]') break;
+    const chunk = JSON.parse(data);
+    if (chunk.error !== undefined) {
+      return write({type: 'error', message: chunk.error.message ?? JSON.stringify(chunk.error)});
+    }
+    if (!reportedModel && typeof chunk.model === 'string') {
+      write({type: 'model', model: chunk.model});
+      reportedModel = true;
+    }
+    const text = chunk.choices?.[0]?.delta?.content;
+    if (typeof text === 'string' && text !== '') write({type: 'delta', text});
+    if (chunk.usage) {
+      const {prompt_tokens: inputTokens, completion_tokens: outputTokens} = chunk.usage;
+      write({
+        type: 'usage',
+        usage: {
+          ...(typeof inputTokens === 'number' ? {inputTokens} : {}),
+          ...(typeof outputTokens === 'number' ? {outputTokens} : {}),
+        },
+      });
+    }
+  }
+  write({type: 'done'});
+};
+
+const input = readline.createInterface({input: process.stdin, crlfDelay: Infinity});
+for await (const line of input) await answer(JSON.parse(line));

tests/openai-compatible-adapter.test.ts

@@ -0,0 +1,93 @@
+import {createServer, type IncomingMessage, type Server} from 'node:http';
+import type {AddressInfo} from 'node:net';
+import {fileURLToPath} from 'node:url';
+import {afterEach, describe, expect, it} from 'vitest';
+import {CommandTransport, type ChatChunk, type ChatRequest} from '../src/transport/index.js';
+
+const adapter = fileURLToPath(new URL('../examples/openai-compatible-model.mjs', import.meta.url));
+const servers: Server[] = [];
+afterEach(async () => {
+  await Promise.all(
+    servers.splice(0).map((server) => new Promise((resolve) => server.close(resolve))),
+  );
+});
+
+const readBody = async (request: IncomingMessage): Promise<unknown> => {
+  let body = '';
+  for await (const chunk of request) body += String(chunk);
+  return JSON.parse(body);
+};
+
+const serve = async (handler: Parameters<typeof createServer>[1]): Promise<string> => {
+  const server = createServer(handler);
+  servers.push(server);
+  await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve));
+  return `http://127.0.0.1:${String((server.address() as AddressInfo).port)}/v1`;
+};
+
+const run = async (baseUrl: string, request: ChatRequest): Promise<ChatChunk[]> => {
+  const transport = new CommandTransport({
+    command: [process.execPath, adapter],
+    env: {...process.env, RELAY_OPENAI_BASE_URL: baseUrl},
+  });
+  const chunks: ChatChunk[] = [];
+  for await (const chunk of transport.stream(request, new AbortController().signal)) {
+    chunks.push(chunk);
+  }
+  return chunks;
+};
+
+describe('OpenAI-compatible example adapter', () => {
+  it('streams chat-completion deltas, model, and usage', async () => {
+    let received: unknown;
+    const baseUrl = await serve((request, response) => {
+      void readBody(request).then((body) => {
+        received = body;
+        const model = 'qwen2.5-0.5b-instruct-q4-k-m';
+        const event = (value: unknown): string => `data: ${JSON.stringify(value)}\n\n`;
+        response.writeHead(200, {'content-type': 'text/event-stream'});
+        response.write(': keepalive\n\n');
+        response.write(event({model, choices: [{index: 0, delta: {content: 'Hel'}}]}));
+        response.write(event({model, choices: [{index: 0, delta: {content: 'lo'}}]}));
+        response.write(event({model, choices: [{index: 0, delta: {}, finish_reason: 'stop'}]}));
+        response.write(
+          event({model, choices: [], usage: {prompt_tokens: 9, completion_tokens: 2}}),
+        );
+        response.end('data: [DONE]\n\n');
+      });
+    });
+
+    const chunks = await run(baseUrl, {
+      conversationId: 'c1',
+      systemPrompt: 'Be brief.',
+      messages: [{role: 'user', content: 'Say hello.'}],
+    });
+
+    expect(received).toEqual({
+      model: 'qwen2.5-0.5b-instruct-q4-k-m',
+      messages: [
+        {role: 'system', content: 'Be brief.'},
+        {role: 'user', content: 'Say hello.'},
+      ],
+      stream: true,
+    });
+    expect(chunks).toEqual([
+      {type: 'model', model: 'qwen2.5-0.5b-instruct-q4-k-m'},
+      {type: 'delta', text: 'Hel'},
+      {type: 'delta', text: 'lo'},
+      {type: 'usage', usage: {inputTokens: 9, outputTokens: 2}},
+      {type: 'done'},
+    ]);
+  });
+
+  it('reports the server error message', async () => {
+    const baseUrl = await serve((_request, response) => {
+      response.writeHead(503, {'content-type': 'application/json'});
+      response.end(JSON.stringify({error: {code: 'not_ready', message: 'Model is not ready.'}}));
+    });
+
+    await expect(
+      run(baseUrl, {conversationId: 'c1', messages: [{role: 'user', content: 'hi'}]}),
+    ).rejects.toThrow('503: Model is not ready.');
+  });
+});