From f44b3b0949c5c5f9fc9e3b112b64480474e06cda Mon Sep 17 00:00:00 2001 From: autoantohaki Date: Wed, 21 Oct 2026 20:09:29 -0300 Subject: [PATCH] feat: add an Impossible ReAct Agent adapter example --- examples/README.md | 12 +++++ examples/react-agent-model.mjs | 62 ++++++++++++++++++++++ tests/react-agent-adapter.test.ts | 87 +++++++++++++++++++++++++++++++ 3 files changed, 161 insertions(+) create mode 100644 examples/react-agent-model.mjs create mode 100644 tests/react-agent-adapter.test.ts diff --git a/examples/README.md b/examples/README.md index 08e3956..61061bb 100644 --- a/examples/README.md +++ b/examples/README.md @@ -27,3 +27,15 @@ Configure it with environment variables: | `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. + +## Impossible ReAct Agent + +`react-agent-model.mjs` connects Relay Chat to the streaming HTTP API of [Impossible ReAct Agent](https://github.com/impossibleG/impossible-react-agent). Start the agent with `npm start -- --serve`, then run: + +```sh +relay-chat -- node examples/react-agent-model.mjs +``` + +Set `RELAY_AGENT_URL` when the agent is not on `http://127.0.0.1:4141`. + +The agent keeps its own memory for each thread, so the adapter sends only the newest user message and uses the Relay Chat conversation ID as the thread ID. Regenerating or editing a turn does not remove the earlier attempt from the agent's memory; start a new conversation for a clean thread. diff --git a/examples/react-agent-model.mjs b/examples/react-agent-model.mjs new file mode 100644 index 0000000..f0fe524 --- /dev/null +++ b/examples/react-agent-model.mjs @@ -0,0 +1,62 @@ +// Relay Chat command adapter for the Impossible ReAct Agent streaming HTTP API. +import readline from 'node:readline'; + +const baseUrl = (process.env.RELAY_AGENT_URL ?? 'http://127.0.0.1:4141').replace(/\/+$/, ''); + +const write = (chunk) => process.stdout.write(`${JSON.stringify(chunk)}\n`); + +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/, ''); + let event = 'message'; + const data = []; + for (const line of block.split(/\r?\n/)) { + if (line.startsWith('event:')) event = line.slice(6).trim(); + if (line.startsWith('data:')) data.push(line.slice(5).replace(/^ /, '')); + } + if (data.length > 0) yield {event, data: JSON.parse(data.join('\n'))}; + } + } +} + +const answer = async (request) => { + // The agent keeps its own thread memory, so only the newest user message is sent. + const input = request.messages.findLast((message) => message.role === 'user')?.content ?? ''; + let response; + try { + response = await fetch(`${baseUrl}/v1/chat/stream`, { + method: 'POST', + headers: {'content-type': 'application/json', accept: 'text/event-stream'}, + body: JSON.stringify({input, threadId: request.conversationId}), + }); + } catch (error) { + return write({type: 'error', message: `Could not reach ${baseUrl}: ${error.message}`}); + } + if (!response.ok || response.body === null) { + const body = await response.text(); + let message = body.slice(0, 500) || response.statusText; + try { + const parsed = JSON.parse(body); + if (typeof parsed.error === 'string') message = parsed.error; + } catch { + // Keep the raw body. + } + return write({type: 'error', message: `${response.status}: ${message}`}); + } + + for await (const {event, data} of serverEvents(response.body)) { + if (event === 'text' && typeof data.text === 'string') write({type: 'delta', text: data.text}); + if (event === 'error') return write({type: 'error', message: String(data.error)}); + if (event === 'done') break; + } + write({type: 'done'}); +}; + +const lines = readline.createInterface({input: process.stdin, crlfDelay: Infinity}); +for await (const line of lines) await answer(JSON.parse(line)); diff --git a/tests/react-agent-adapter.test.ts b/tests/react-agent-adapter.test.ts new file mode 100644 index 0000000..3e8d166 --- /dev/null +++ b/tests/react-agent-adapter.test.ts @@ -0,0 +1,87 @@ +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} from '../src/transport/index.js'; + +const adapter = fileURLToPath(new URL('../examples/react-agent-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 (events: string[]): Promise<{url: string; received: unknown[]}> => { + const received: unknown[] = []; + const server = createServer((request, response) => { + void readBody(request).then((body) => { + received.push({path: request.url, body}); + response.writeHead(200, {'content-type': 'text/event-stream'}); + for (const event of events) response.write(event); + response.end(); + }); + }); + servers.push(server); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + return {url: `http://127.0.0.1:${String((server.address() as AddressInfo).port)}`, received}; +}; + +const event = (type: string, data: unknown): string => + `event: ${type}\ndata: ${JSON.stringify(data)}\n\n`; + +const run = async (url: string): Promise => { + const transport = new CommandTransport({ + command: [process.execPath, adapter], + env: {...process.env, RELAY_AGENT_URL: url}, + }); + const chunks: ChatChunk[] = []; + for await (const chunk of transport.stream( + { + conversationId: 'c1', + messages: [ + {role: 'user', content: 'first'}, + {role: 'assistant', content: 'reply'}, + {role: 'user', content: 'What time is it?'}, + ], + }, + new AbortController().signal, + )) { + chunks.push(chunk); + } + return chunks; +}; + +describe('ReAct agent example adapter', () => { + it('sends the newest user message on the conversation thread and streams text', async () => { + const server = await serve([ + event('text', {type: 'text', text: 'It is '}), + event('text', {type: 'text', text: 'noon.'}), + event('done', {type: 'done', threadId: 'c1'}), + ]); + + expect(await run(server.url)).toEqual([ + {type: 'delta', text: 'It is '}, + {type: 'delta', text: 'noon.'}, + {type: 'done'}, + ]); + expect(server.received).toEqual([ + {path: '/v1/chat/stream', body: {input: 'What time is it?', threadId: 'c1'}}, + ]); + }); + + it('fails the turn on a streamed error event', async () => { + const server = await serve([ + event('text', {type: 'text', text: 'partial'}), + event('error', {error: 'Step limit reached.'}), + ]); + + await expect(run(server.url)).rejects.toThrow('Step limit reached.'); + }); +});