diff --git a/examples/README.md b/examples/README.md index 7696e66..08e3956 100644 --- a/examples/README.md +++ b/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. diff --git a/examples/openai-compatible-model.mjs b/examples/openai-compatible-model.mjs new file mode 100644 index 0000000..864e8ac --- /dev/null +++ b/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)); diff --git a/tests/openai-compatible-adapter.test.ts b/tests/openai-compatible-adapter.test.ts new file mode 100644 index 0000000..6f5fcb0 --- /dev/null +++ b/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 => { + let body = ''; + for await (const chunk of request) body += String(chunk); + return JSON.parse(body); +}; + +const serve = async (handler: Parameters[1]): Promise => { + const server = createServer(handler); + servers.push(server); + await new Promise((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 => { + 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.'); + }); +});