openvibers/openvibe.chat Commits
Vibe's answers stream: each turn is read from AI as it is written (runs?stream=1) and the chat sees room_ai_delta with the text so far; the waiting bubble renders it as Markdown until the finished message replaces it (an AI without the stream still answers) (#182)
Changes
@@@ -138,6 +138,9 @@ body.oc-app-body { overflow: hidden; }@keyframes vbDot { 0%, 80%, 100% { opacity: .35; transform: translateY(0); } 40% { opacity: 1; transform: translateY(-3px); } }@media (prefers-reduced-motion: reduce) { .vb-dots i { animation: none; opacity: .7; } }++/* An answer as it is written (vibe.js streamText): the text, then the dots, small. */++.vb-wait .vb-streaming + .vb-dots { margin-top: 6px; transform: scale(.8); transform-origin: left center; }++/* ── A new chat ───────────────────────────────────────────────────────────── */.vb-hello { max-width: var(--vb-col); margin: 0 auto; min-height: 100%; display: flex; flex-direction: column; align-items: center; justify-content: center; text-align: center; padding: 24px 0 8px; }.vb-hello .vb-mark { width: 52px; height: 52px; margin-bottom: 14px; }@@@ -148,6 +148,26 @@// An answer takes up to a minute with tools; after two the dots give way to a way forward.waitTimer = setTimeout(function () { clearWait(); notice('Vibe did not answer this time. Send your message again, or try again in a minute.'); }, 120000);}++ // The answer as it is written (room_ai_delta: the whole text so far): the waiting bubble shows it, rendered like the++ // finished answer, until that message arrives and takes its place. Screen readers hear the finished one only.++ function streamText(text) {++ showWait();++ var w = document.getElementById('vb-wait'); if (!w) return;++ var stick = nearBottom();++ var body = w.querySelector('.vb-body');++ var md = body.querySelector('.vb-md');++ if (!md) {++ body.textContent = '';++ md = el('div', 'vb-md vb-streaming'); md.setAttribute('aria-hidden', 'true');++ body.appendChild(md);++ var dots = el('span', 'vb-dots'); dots.setAttribute('aria-label', 'Vibe is writing');++ dots.appendChild(el('i')); dots.appendChild(el('i')); dots.appendChild(el('i'));++ body.appendChild(dots);++ }++ if (MD) md.innerHTML = MD.render(String(text || '')); else md.textContent = String(text || '');++ if (waitTimer) { clearTimeout(waitTimer); waitTimer = setTimeout(function () { clearWait(); notice('Vibe did not answer this time. Send your message again, or try again in a minute.'); }, 120000); }++ if (stick) toBottom();++ }function notice(text) {var old = scroll.querySelector('.vb-live-notice'); if (old) old.remove();var p = el('p', 'oc-notice oc-error vb-live-notice', text); p.setAttribute('role', 'alert');@@@ -187,6 +207,7 @@if (m.type === 'room_joined') catchUp();else if (m.type === 'room_message' && m.room === room) add(m.message);else if (m.type === 'room_ai_typing' && m.room === room) { showWait(); if (nearBottom()) toBottom(); }++ else if (m.type === 'room_ai_delta' && m.room === room) streamText(m.text);else if (m.type === 'room_message_deleted' && m.room === room) { var x = have(m.id); if (x) x.remove(); }else if (m.type === 'auth_revoked') notice('You were signed out. Sign in again to keep chatting.');};@@@ -12,6 +12,7 @@ const config = require('../config');let _ai = null;let _http = null;++let _tokens = null;/** The HTTP client behind ai() (Chat's token for openvibe.ai), for calls the SDK's AI client has no method for. */function http() { ai(); return _http; }@@@ -29,6 +30,7 @@ function ai() {audience: 'openvibe.ai',scope: config.ai.scope,});++ _tokens = tokenClient;_http = createClient({ tokenProvider: tokenClient, timeoutMs: config.ai.waitMs + 10000 });_ai = createAiClient(_http, { baseUrl: config.ai.internalUrl });return _ai;@@@ -71,4 +73,55 @@ async function structured(workflow, input, opts = {}) {}}--module.exports = { structured, enabled, http, _setClient(c) { _ai = c; }, _setHttp(h) { _http = h; }, _reset() { _ai = null; _http = null; } };++/**++ * structured(), with the answer's text as it is written: AI's POST /api/v1/runs?stream=1 sends `delta` events, and++ * onText(textSoFar) hears each one; the run's output comes back the same way (null on any failure). An AI without the++ * stream answers JSON, which is read as structured() reads it. A test's fake client (_setClient) has no stream: it goes++ * through structured().++ */++async function stream(workflow, input, opts = {}, onText = () => {}) {++ if (!(opts.force ? config.oauth.clientSecret : enabled())) return null;++ if (_ai && !_tokens) return structured(workflow, input, opts);++ try {++ if (!_tokens) ai();++ const token = await _tokens.getToken();++ const res = await (opts.fetchImpl || globalThis.fetch)(`${config.ai.internalUrl}/api/v1/runs?stream=1&wait=${Math.max(0, Math.floor(config.ai.waitMs))}`, {++ method: 'POST',++ headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', Accept: 'text/event-stream', ...(opts.idempotencyKey ? { 'Idempotency-Key': opts.idempotencyKey } : {}) },++ body: JSON.stringify({ workflow, input, ...(opts.credential ? { credential: opts.credential } : {}), ...(opts.attribution !== undefined ? { attribution: opts.attribution } : {}) }),++ signal: AbortSignal.timeout(config.ai.waitMs + 10000),++ });++ if (!res.ok) { console.warn(`[AI] ${workflow}: AI answered ${res.status}`); return null; }++ let run = null;++ if (!/event-stream/.test(String(res.headers.get('content-type') || ''))) {++ const body = await res.json();++ run = body && (body.run || body);++ } else {++ const decoder = new TextDecoder();++ let buffer = '';++ let text = '';++ for await (const chunk of res.body) {++ buffer += decoder.decode(chunk, { stream: true });++ let cut;++ while ((cut = buffer.indexOf('\n\n')) >= 0) {++ const block = buffer.slice(0, cut);++ buffer = buffer.slice(cut + 2);++ const event = (/^event: (\w+)$/m.exec(block) || [])[1];++ const line = (/^data: (.*)$/m.exec(block) || [])[1];++ let data = null;++ try { data = line ? JSON.parse(line) : null; } catch { data = null; }++ if (event === 'delta' && data && typeof data.text === 'string') { text += data.text; try { onText(text); } catch { /* a listener never breaks the answer */ } }++ else if (event === 'run' && data) run = data.run || null;++ else if (event === 'error') console.warn(`[AI] ${workflow}: ${(data && data.message) || 'the stream broke'}`);++ }++ }++ }++ if (!run || run.status !== 'succeeded' || run.synthetic) return null;++ return run.output || null;++ } catch (err) {++ console.warn(`[AI] ${workflow}: ${err && err.message}`);++ return null;++ }++}++++module.exports = { structured, stream, enabled, http, _setClient(c) { _ai = c; }, _setHttp(h) { _http = h; }, _setTokens(t) { _tokens = t; }, _reset() { _ai = null; _http = null; _tokens = null; } };@@@ -48,6 +48,7 @@ const inFlight = new Set(); // room ids with an answer being writtenconst personHour = new Map(); // `${room}:${user}` → [timestamps]const handleOf = (name) => String(name || '').toLowerCase().normalize('NFKD').replace(/[^\p{L}\p{N}]+/gu, '').slice(0, 24);++const DELTA_MS = 120; // the most often a streaming answer goes to the chat's socketsconst clean = (v, max) => String(v == null ? '' : v).replace(/\r\n?/g, '\n').replace(/[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f]/g, '').trim().slice(0, max);class AiMemberError extends Error { constructor(status, code, message) { super(message); this.status = status; this.code = code; } }@@@ -282,7 +283,7 @@ function userPrompt({ member, context, transcript, askedBy, text }) {* The answer through chat.member.turn with tools: up to MAX_TOOL_ROUNDS rounds of tool calls run here, then a turn* with tool_choice none. → the reply text, or null when a turn failed (the caller falls back to chat.member.reply).*/--async function answerWithTools({ room, member, context, transcript, askedBy, text, opts, subject = null, images = [], user = null, messageId = null, memory = null, called = new Set() }) {++async function answerWithTools({ room, member, context, transcript, askedBy, text, opts, subject = null, images = [], user = null, messageId = null, memory = null, called = new Set(), onText = null }) {const tools = aiTools.definitions(member, room, { memory: memory !== null });const vibe = Number(room.assistant) === 1;const prompt = userPrompt({ member, context, transcript, askedBy, text });@@@ -293,8 +294,10 @@ async function answerWithTools({ room, member, context, transcript, askedBy, tex];for (let round = 0; round <= MAX_TOOL_ROUNDS; round++) {const last = round === MAX_TOOL_ROUNDS;-- const out = await aiClient.structured('chat.member.turn', { messages, tools, tool_choice: last ? 'none' : 'auto', max_tokens: vibe ? 1500 : 600, temperature: 0.7 },-- { ...opts, idempotencyKey: `${opts.idempotencyKey}-t${round}` });++ const turn = { messages, tools, tool_choice: last ? 'none' : 'auto', max_tokens: vibe ? 1500 : 600, temperature: 0.7 };++ const turnOpts = { ...opts, idempotencyKey: `${opts.idempotencyKey}-t${round}` };++ // Vibe's answer streams (each round's text so far, to the chat as it is written); other members answer whole.++ const out = onText ? await aiClient.stream('chat.member.turn', turn, turnOpts, onText) : await aiClient.structured('chat.member.turn', turn, turnOpts);const msg = out && Array.isArray(out.choices) && out.choices[0] && out.choices[0].message;if (!msg) return null;const calls = !last && Array.isArray(msg.tool_calls) ? msg.tool_calls.filter((c) => c && c.function && c.function.name).slice(0, MAX_CALLS_PER_ROUND) : [];@@@ -353,9 +356,20 @@ async function afterPost(room, message, user, deps = {}) {const memory = vibe && Number(room.owner_id) === Number(user.id) ? await vibeMemory.forPrompt(user) : null;let text = '';const called = new Set(); // the tools the model called while answering++ // In a Vibe chat the answer shows as it is written: room_ai_delta { text } (the whole text so far, at most every++ // DELTA_MS), until the finished message replaces it.++ let lastDelta = 0;++ let pending = null;++ const onText = vibe ? (sofar) => {++ pending = sofar;++ const now = Date.now();++ if (now - lastDelta < DELTA_MS) return;++ lastDelta = now;++ broadcast({ type: 'room_ai_delta', room: room.slug, name: member.name, text: String(pending).slice(0, 8000) });++ } : null;if (config.ai.memberTools) {const images = await picturesFor(message, history, user).catch(() => []);-- try { text = clean(await answerWithTools({ room, member, context, transcript, askedBy, text: String(message.message).slice(0, 2000), opts, subject: user.subject_id || null, images, user, messageId: message.id, memory, called }) || '', vibe ? 8000 : 2000); } catch { text = ''; }++ try { text = clean(await answerWithTools({ room, member, context, transcript, askedBy, text: String(message.message).slice(0, 2000), opts, subject: user.subject_id || null, images, user, messageId: message.id, memory, called, onText }) || '', vibe ? 8000 : 2000); } catch { text = ''; }}// A plain "remember that …" is kept even when the model only promised it (small models often do).if (memory !== null && text && !called.has('remember')) {nnew file mode 100644@@@ -0,0 +1,79 @@++'use strict';++/**++ * Vibe's answer as it is written (server/ai/client.js stream(), ai-members.js; OpenVibe.AI's POST /api/v1/runs?stream=1):++ * the chat's sockets get room_ai_delta frames with the text so far (at most every DELTA_MS), then the finished answer as++ * a room message; an AI that answers JSON (no stream) still answers; the request carries the run's idempotency key.++ */++const assert = require('assert');++const http = require('http');++const { ids } = require('openvibe-contracts');++const { boot, suite } = require('./helpers');++++const t = suite('vibe stream');++let h, ann, ai, config, aiClient;++const seen = [];++let mode = 'stream';++const pause = (ms) => new Promise((r) => setTimeout(r, ms));++const RUN = (content) => ({ id: 'run_1', status: 'succeeded', output: { choices: [{ index: 0, message: { role: 'assistant', content }, finish_reason: 'stop' }] } });++++t('boot', async () => {++ ai = http.createServer((req, res) => {++ let body = '';++ req.on('data', (c) => { body += c; });++ req.on('end', async () => {++ seen.push({ url: req.url, auth: req.headers.authorization, key: req.headers['idempotency-key'], body: JSON.parse(body) });++ if (mode === 'json') { res.writeHead(201, { 'content-type': 'application/json' }); return res.end(JSON.stringify({ run: RUN('A whole answer.') })); }++ res.writeHead(200, { 'content-type': 'text/event-stream; charset=utf-8' });++ const send = (event, data) => res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);++ send('delta', { text: 'Rice, ' });++ await pause(200);++ send('delta', { text: 'lentils ' });++ await pause(200);++ send('delta', { text: 'and greens.' });++ send('run', { run: RUN('Rice, lentils and greens.') });++ res.end();++ });++ });++ await new Promise((r) => ai.listen(0, '127.0.0.1', r));++ h = await boot({ env: { CHAT_WEB_URL: 'https://openvibe.chat' } });++ config = require('../server/config');++ config.ai.internalUrl = `http://127.0.0.1:${ai.address().port}`;++ aiClient = require('../server/ai/client');++ aiClient._reset();++ aiClient._setTokens({ getToken: async () => 'chat-ai-token' });++ ann = h.addUser('ann', { subject: ids.newId('user') });++ await h.ctx.upsertUser(h.live.users.get(ann.id));++});++++async function ask(text) {++ const vibe = require('../server/rooms/vibe');++ const rooms = require('../server/rooms/rooms');++ const user = h.live.users.get(ann.id);++ const room = await vibe.createRoom(user, text);++ const message = await rooms.post(room, user, text, { ai: false });++ const frames = [];++ const reply = await require('../server/rooms/ai-members').afterPost(room, message, user, { broadcast: (f) => frames.push(f) });++ return { frames, reply, room };++}++++t('the chat sees the answer as it is written, then the finished message', async () => {++ const { frames, reply, room } = await ask('What should I cook tonight?');++ const deltas = frames.filter((f) => f.type === 'room_ai_delta');++ assert.ok(deltas.length >= 2, `streamed in steps (${deltas.length})`);++ assert.deepStrictEqual(deltas.slice(0, 2).map((f) => [f.room, f.name, f.text]), [[room.slug, 'Vibe', 'Rice, '], [room.slug, 'Vibe', 'Rice, lentils ']], 'the whole text so far, each time');++ const types = frames.map((f) => f.type);++ assert.ok(types.indexOf('room_ai_typing') < types.indexOf('room_ai_delta') && types.lastIndexOf('room_ai_delta') < types.lastIndexOf('room_message'), types.join(' '));++ assert.strictEqual(reply.message, 'Rice, lentils and greens.');++ const call = seen[seen.length - 1];++ assert.match(call.url, /^\/api\/v1\/runs\?stream=1&wait=\d+$/);++ assert.deepStrictEqual([call.auth, call.body.workflow, /^chat-ai-\d+-\d+-t0$/.test(call.key)], ['Bearer chat-ai-token', 'chat.member.turn', true]);++});++++t('an AI without the stream answers JSON: the answer still comes, without deltas', async () => {++ mode = 'json';++ const { frames, reply } = await ask('And for dessert?');++ assert.strictEqual(frames.filter((f) => f.type === 'room_ai_delta').length, 0);++ assert.strictEqual(reply.message, 'A whole answer.');++});++++t.run(async () => { aiClient._reset(); if (h && h.close) await h.close(); await new Promise((r) => ai.close(r)); });
@@ -138,6 +138,9 @@ body.oc-app-body { overflow: hidden; }
@@ -138,6 +138,9 @@ body.oc-app-body { overflow: hidden; }
@keyframes vbDot { 0%, 80%, 100% { opacity: .35; transform: translateY(0); } 40% { opacity: 1; transform: translateY(-3px); } }
@keyframes vbDot { 0%, 80%, 100% { opacity: .35; transform: translateY(0); } 40% { opacity: 1; transform: translateY(-3px); } }
@media (prefers-reduced-motion: reduce) { .vb-dots i { animation: none; opacity: .7; } }
@media (prefers-reduced-motion: reduce) { .vb-dots i { animation: none; opacity: .7; } }
+/* An answer as it is written (vibe.js streamText): the text, then the dots, small. */
+.vb-wait .vb-streaming + .vb-dots { margin-top: 6px; transform: scale(.8); transform-origin: left center; }
+
/* ── A new chat ───────────────────────────────────────────────────────────── */
/* ── A new chat ───────────────────────────────────────────────────────────── */
.vb-hello { max-width: var(--vb-col); margin: 0 auto; min-height: 100%; display: flex; flex-direction: column; align-items: center; justify-content: center; text-align: center; padding: 24px 0 8px; }
.vb-hello { max-width: var(--vb-col); margin: 0 auto; min-height: 100%; display: flex; flex-direction: column; align-items: center; justify-content: center; text-align: center; padding: 24px 0 8px; }
.vb-hello .vb-mark { width: 52px; height: 52px; margin-bottom: 14px; }
.vb-hello .vb-mark { width: 52px; height: 52px; margin-bottom: 14px; }
@@ -148,6 +148,26 @@
@@ -148,6 +148,26 @@
// An answer takes up to a minute with tools; after two the dots give way to a way forward.
// An answer takes up to a minute with tools; after two the dots give way to a way forward.
waitTimer = setTimeout(function () { clearWait(); notice('Vibe did not answer this time. Send your message again, or try again in a minute.'); }, 120000);
waitTimer = setTimeout(function () { clearWait(); notice('Vibe did not answer this time. Send your message again, or try again in a minute.'); }, 120000);
}
}
+ // The answer as it is written (room_ai_delta: the whole text so far): the waiting bubble shows it, rendered like the
+ // finished answer, until that message arrives and takes its place. Screen readers hear the finished one only.
+ function streamText(text) {
+ showWait();
+ var w = document.getElementById('vb-wait'); if (!w) return;
+ var stick = nearBottom();
+ var body = w.querySelector('.vb-body');
+ var md = body.querySelector('.vb-md');
+ if (!md) {
+ body.textContent = '';
+ md = el('div', 'vb-md vb-streaming'); md.setAttribute('aria-hidden', 'true');
+ body.appendChild(md);
+ var dots = el('span', 'vb-dots'); dots.setAttribute('aria-label', 'Vibe is writing');
+ dots.appendChild(el('i')); dots.appendChild(el('i')); dots.appendChild(el('i'));
+ body.appendChild(dots);
+ }
+ if (MD) md.innerHTML = MD.render(String(text || '')); else md.textContent = String(text || '');
+ if (waitTimer) { clearTimeout(waitTimer); waitTimer = setTimeout(function () { clearWait(); notice('Vibe did not answer this time. Send your message again, or try again in a minute.'); }, 120000); }
+ if (stick) toBottom();
+ }
function notice(text) {
function notice(text) {
var old = scroll.querySelector('.vb-live-notice'); if (old) old.remove();
var old = scroll.querySelector('.vb-live-notice'); if (old) old.remove();
var p = el('p', 'oc-notice oc-error vb-live-notice', text); p.setAttribute('role', 'alert');
var p = el('p', 'oc-notice oc-error vb-live-notice', text); p.setAttribute('role', 'alert');
@@ -187,6 +207,7 @@
@@ -187,6 +207,7 @@
if (m.type === 'room_joined') catchUp();
if (m.type === 'room_joined') catchUp();
else if (m.type === 'room_message' && m.room === room) add(m.message);
else if (m.type === 'room_message' && m.room === room) add(m.message);
else if (m.type === 'room_ai_typing' && m.room === room) { showWait(); if (nearBottom()) toBottom(); }
else if (m.type === 'room_ai_typing' && m.room === room) { showWait(); if (nearBottom()) toBottom(); }
+ else if (m.type === 'room_ai_delta' && m.room === room) streamText(m.text);
else if (m.type === 'room_message_deleted' && m.room === room) { var x = have(m.id); if (x) x.remove(); }
else if (m.type === 'room_message_deleted' && m.room === room) { var x = have(m.id); if (x) x.remove(); }
else if (m.type === 'auth_revoked') notice('You were signed out. Sign in again to keep chatting.');
else if (m.type === 'auth_revoked') notice('You were signed out. Sign in again to keep chatting.');
};
};
@@ -12,6 +12,7 @@ const config = require('../config');
@@ -12,6 +12,7 @@ const config = require('../config');
let _ai = null;
let _ai = null;
let _http = null;
let _http = null;
+let _tokens = null;
/** The HTTP client behind ai() (Chat's token for openvibe.ai), for calls the SDK's AI client has no method for. */
/** The HTTP client behind ai() (Chat's token for openvibe.ai), for calls the SDK's AI client has no method for. */
function http() { ai(); return _http; }
function http() { ai(); return _http; }
@@ -29,6 +30,7 @@ function ai() {
@@ -29,6 +30,7 @@ function ai() {
audience: 'openvibe.ai',
audience: 'openvibe.ai',
scope: config.ai.scope,
scope: config.ai.scope,
});
});
+ _tokens = tokenClient;
_http = createClient({ tokenProvider: tokenClient, timeoutMs: config.ai.waitMs + 10000 });
_http = createClient({ tokenProvider: tokenClient, timeoutMs: config.ai.waitMs + 10000 });
_ai = createAiClient(_http, { baseUrl: config.ai.internalUrl });
_ai = createAiClient(_http, { baseUrl: config.ai.internalUrl });
return _ai;
return _ai;
@@ -71,4 +73,55 @@ async function structured(workflow, input, opts = {}) {
@@ -71,4 +73,55 @@ async function structured(workflow, input, opts = {}) {
}
}
}
}
-module.exports = { structured, enabled, http, _setClient(c) { _ai = c; }, _setHttp(h) { _http = h; }, _reset() { _ai = null; _http = null; } };
+/**
+ * structured(), with the answer's text as it is written: AI's POST /api/v1/runs?stream=1 sends `delta` events, and
+ * onText(textSoFar) hears each one; the run's output comes back the same way (null on any failure). An AI without the
+ * stream answers JSON, which is read as structured() reads it. A test's fake client (_setClient) has no stream: it goes
+ * through structured().
+ */
+async function stream(workflow, input, opts = {}, onText = () => {}) {
+ if (!(opts.force ? config.oauth.clientSecret : enabled())) return null;
+ if (_ai && !_tokens) return structured(workflow, input, opts);
+ try {
+ if (!_tokens) ai();
+ const token = await _tokens.getToken();
+ const res = await (opts.fetchImpl || globalThis.fetch)(`${config.ai.internalUrl}/api/v1/runs?stream=1&wait=${Math.max(0, Math.floor(config.ai.waitMs))}`, {
+ method: 'POST',
+ headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', Accept: 'text/event-stream', ...(opts.idempotencyKey ? { 'Idempotency-Key': opts.idempotencyKey } : {}) },
+ body: JSON.stringify({ workflow, input, ...(opts.credential ? { credential: opts.credential } : {}), ...(opts.attribution !== undefined ? { attribution: opts.attribution } : {}) }),
+ signal: AbortSignal.timeout(config.ai.waitMs + 10000),
+ });
+ if (!res.ok) { console.warn(`[AI] ${workflow}: AI answered ${res.status}`); return null; }
+ let run = null;
+ if (!/event-stream/.test(String(res.headers.get('content-type') || ''))) {
+ const body = await res.json();
+ run = body && (body.run || body);
+ } else {
+ const decoder = new TextDecoder();
+ let buffer = '';
+ let text = '';
+ for await (const chunk of res.body) {
+ buffer += decoder.decode(chunk, { stream: true });
+ let cut;
+ while ((cut = buffer.indexOf('\n\n')) >= 0) {
+ const block = buffer.slice(0, cut);
+ buffer = buffer.slice(cut + 2);
+ const event = (/^event: (\w+)$/m.exec(block) || [])[1];
+ const line = (/^data: (.*)$/m.exec(block) || [])[1];
+ let data = null;
+ try { data = line ? JSON.parse(line) : null; } catch { data = null; }
+ if (event === 'delta' && data && typeof data.text === 'string') { text += data.text; try { onText(text); } catch { /* a listener never breaks the answer */ } }
+ else if (event === 'run' && data) run = data.run || null;
+ else if (event === 'error') console.warn(`[AI] ${workflow}: ${(data && data.message) || 'the stream broke'}`);
+ }
+ }
+ }
+ if (!run || run.status !== 'succeeded' || run.synthetic) return null;
+ return run.output || null;
+ } catch (err) {
+ console.warn(`[AI] ${workflow}: ${err && err.message}`);
+ return null;
+ }
+}
+
+module.exports = { structured, stream, enabled, http, _setClient(c) { _ai = c; }, _setHttp(h) { _http = h; }, _setTokens(t) { _tokens = t; }, _reset() { _ai = null; _http = null; _tokens = null; } };
@@ -48,6 +48,7 @@ const inFlight = new Set(); // room ids with an answer being written
@@ -48,6 +48,7 @@ const inFlight = new Set(); // room ids with an answer being written
const personHour = new Map(); // `${room}:${user}` → [timestamps]
const personHour = new Map(); // `${room}:${user}` → [timestamps]
const handleOf = (name) => String(name || '').toLowerCase().normalize('NFKD').replace(/[^\p{L}\p{N}]+/gu, '').slice(0, 24);
const handleOf = (name) => String(name || '').toLowerCase().normalize('NFKD').replace(/[^\p{L}\p{N}]+/gu, '').slice(0, 24);
+const DELTA_MS = 120; // the most often a streaming answer goes to the chat's sockets
const clean = (v, max) => String(v == null ? '' : v).replace(/\r\n?/g, '\n').replace(/[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f]/g, '').trim().slice(0, max);
const clean = (v, max) => String(v == null ? '' : v).replace(/\r\n?/g, '\n').replace(/[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f]/g, '').trim().slice(0, max);
class AiMemberError extends Error { constructor(status, code, message) { super(message); this.status = status; this.code = code; } }
class AiMemberError extends Error { constructor(status, code, message) { super(message); this.status = status; this.code = code; } }
@@ -282,7 +283,7 @@ function userPrompt({ member, context, transcript, askedBy, text }) {
@@ -282,7 +283,7 @@ function userPrompt({ member, context, transcript, askedBy, text }) {
* The answer through chat.member.turn with tools: up to MAX_TOOL_ROUNDS rounds of tool calls run here, then a turn
* The answer through chat.member.turn with tools: up to MAX_TOOL_ROUNDS rounds of tool calls run here, then a turn
* with tool_choice none. → the reply text, or null when a turn failed (the caller falls back to chat.member.reply).
* with tool_choice none. → the reply text, or null when a turn failed (the caller falls back to chat.member.reply).
*/
*/
-async function answerWithTools({ room, member, context, transcript, askedBy, text, opts, subject = null, images = [], user = null, messageId = null, memory = null, called = new Set() }) {
+async function answerWithTools({ room, member, context, transcript, askedBy, text, opts, subject = null, images = [], user = null, messageId = null, memory = null, called = new Set(), onText = null }) {
const tools = aiTools.definitions(member, room, { memory: memory !== null });
const tools = aiTools.definitions(member, room, { memory: memory !== null });
const vibe = Number(room.assistant) === 1;
const vibe = Number(room.assistant) === 1;
const prompt = userPrompt({ member, context, transcript, askedBy, text });
const prompt = userPrompt({ member, context, transcript, askedBy, text });
@@ -293,8 +294,10 @@ async function answerWithTools({ room, member, context, transcript, askedBy, tex
@@ -293,8 +294,10 @@ async function answerWithTools({ room, member, context, transcript, askedBy, tex
];
];
for (let round = 0; round <= MAX_TOOL_ROUNDS; round++) {
for (let round = 0; round <= MAX_TOOL_ROUNDS; round++) {
const last = round === MAX_TOOL_ROUNDS;
const last = round === MAX_TOOL_ROUNDS;
- const out = await aiClient.structured('chat.member.turn', { messages, tools, tool_choice: last ? 'none' : 'auto', max_tokens: vibe ? 1500 : 600, temperature: 0.7 },
- { ...opts, idempotencyKey: `${opts.idempotencyKey}-t${round}` });
+ const turn = { messages, tools, tool_choice: last ? 'none' : 'auto', max_tokens: vibe ? 1500 : 600, temperature: 0.7 };
+ const turnOpts = { ...opts, idempotencyKey: `${opts.idempotencyKey}-t${round}` };
+ // Vibe's answer streams (each round's text so far, to the chat as it is written); other members answer whole.
+ const out = onText ? await aiClient.stream('chat.member.turn', turn, turnOpts, onText) : await aiClient.structured('chat.member.turn', turn, turnOpts);
const msg = out && Array.isArray(out.choices) && out.choices[0] && out.choices[0].message;
const msg = out && Array.isArray(out.choices) && out.choices[0] && out.choices[0].message;
if (!msg) return null;
if (!msg) return null;
const calls = !last && Array.isArray(msg.tool_calls) ? msg.tool_calls.filter((c) => c && c.function && c.function.name).slice(0, MAX_CALLS_PER_ROUND) : [];
const calls = !last && Array.isArray(msg.tool_calls) ? msg.tool_calls.filter((c) => c && c.function && c.function.name).slice(0, MAX_CALLS_PER_ROUND) : [];
@@ -353,9 +356,20 @@ async function afterPost(room, message, user, deps = {}) {
@@ -353,9 +356,20 @@ async function afterPost(room, message, user, deps = {}) {
const memory = vibe && Number(room.owner_id) === Number(user.id) ? await vibeMemory.forPrompt(user) : null;
const memory = vibe && Number(room.owner_id) === Number(user.id) ? await vibeMemory.forPrompt(user) : null;
let text = '';
let text = '';
const called = new Set(); // the tools the model called while answering
const called = new Set(); // the tools the model called while answering
+ // In a Vibe chat the answer shows as it is written: room_ai_delta { text } (the whole text so far, at most every
+ // DELTA_MS), until the finished message replaces it.
+ let lastDelta = 0;
+ let pending = null;
+ const onText = vibe ? (sofar) => {
+ pending = sofar;
+ const now = Date.now();
+ if (now - lastDelta < DELTA_MS) return;
+ lastDelta = now;
+ broadcast({ type: 'room_ai_delta', room: room.slug, name: member.name, text: String(pending).slice(0, 8000) });
+ } : null;
if (config.ai.memberTools) {
if (config.ai.memberTools) {
const images = await picturesFor(message, history, user).catch(() => []);
const images = await picturesFor(message, history, user).catch(() => []);
- try { text = clean(await answerWithTools({ room, member, context, transcript, askedBy, text: String(message.message).slice(0, 2000), opts, subject: user.subject_id || null, images, user, messageId: message.id, memory, called }) || '', vibe ? 8000 : 2000); } catch { text = ''; }
+ try { text = clean(await answerWithTools({ room, member, context, transcript, askedBy, text: String(message.message).slice(0, 2000), opts, subject: user.subject_id || null, images, user, messageId: message.id, memory, called, onText }) || '', vibe ? 8000 : 2000); } catch { text = ''; }
}
}
// A plain "remember that …" is kept even when the model only promised it (small models often do).
// A plain "remember that …" is kept even when the model only promised it (small models often do).
if (memory !== null && text && !called.has('remember')) {
if (memory !== null && text && !called.has('remember')) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,79 @@
@@ -0,0 +1,79 @@
+'use strict';
+/**
+ * Vibe's answer as it is written (server/ai/client.js stream(), ai-members.js; OpenVibe.AI's POST /api/v1/runs?stream=1):
+ * the chat's sockets get room_ai_delta frames with the text so far (at most every DELTA_MS), then the finished answer as
+ * a room message; an AI that answers JSON (no stream) still answers; the request carries the run's idempotency key.
+ */
+const assert = require('assert');
+const http = require('http');
+const { ids } = require('openvibe-contracts');
+const { boot, suite } = require('./helpers');
+
+const t = suite('vibe stream');
+let h, ann, ai, config, aiClient;
+const seen = [];
+let mode = 'stream';
+const pause = (ms) => new Promise((r) => setTimeout(r, ms));
+const RUN = (content) => ({ id: 'run_1', status: 'succeeded', output: { choices: [{ index: 0, message: { role: 'assistant', content }, finish_reason: 'stop' }] } });
+
+t('boot', async () => {
+ ai = http.createServer((req, res) => {
+ let body = '';
+ req.on('data', (c) => { body += c; });
+ req.on('end', async () => {
+ seen.push({ url: req.url, auth: req.headers.authorization, key: req.headers['idempotency-key'], body: JSON.parse(body) });
+ if (mode === 'json') { res.writeHead(201, { 'content-type': 'application/json' }); return res.end(JSON.stringify({ run: RUN('A whole answer.') })); }
+ res.writeHead(200, { 'content-type': 'text/event-stream; charset=utf-8' });
+ const send = (event, data) => res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
+ send('delta', { text: 'Rice, ' });
+ await pause(200);
+ send('delta', { text: 'lentils ' });
+ await pause(200);
+ send('delta', { text: 'and greens.' });
+ send('run', { run: RUN('Rice, lentils and greens.') });
+ res.end();
+ });
+ });
+ await new Promise((r) => ai.listen(0, '127.0.0.1', r));
+ h = await boot({ env: { CHAT_WEB_URL: 'https://openvibe.chat' } });
+ config = require('../server/config');
+ config.ai.internalUrl = `http://127.0.0.1:${ai.address().port}`;
+ aiClient = require('../server/ai/client');
+ aiClient._reset();
+ aiClient._setTokens({ getToken: async () => 'chat-ai-token' });
+ ann = h.addUser('ann', { subject: ids.newId('user') });
+ await h.ctx.upsertUser(h.live.users.get(ann.id));
+});
+
+async function ask(text) {
+ const vibe = require('../server/rooms/vibe');
+ const rooms = require('../server/rooms/rooms');
+ const user = h.live.users.get(ann.id);
+ const room = await vibe.createRoom(user, text);
+ const message = await rooms.post(room, user, text, { ai: false });
+ const frames = [];
+ const reply = await require('../server/rooms/ai-members').afterPost(room, message, user, { broadcast: (f) => frames.push(f) });
+ return { frames, reply, room };
+}
+
+t('the chat sees the answer as it is written, then the finished message', async () => {
+ const { frames, reply, room } = await ask('What should I cook tonight?');
+ const deltas = frames.filter((f) => f.type === 'room_ai_delta');
+ assert.ok(deltas.length >= 2, `streamed in steps (${deltas.length})`);
+ assert.deepStrictEqual(deltas.slice(0, 2).map((f) => [f.room, f.name, f.text]), [[room.slug, 'Vibe', 'Rice, '], [room.slug, 'Vibe', 'Rice, lentils ']], 'the whole text so far, each time');
+ const types = frames.map((f) => f.type);
+ assert.ok(types.indexOf('room_ai_typing') < types.indexOf('room_ai_delta') && types.lastIndexOf('room_ai_delta') < types.lastIndexOf('room_message'), types.join(' '));
+ assert.strictEqual(reply.message, 'Rice, lentils and greens.');
+ const call = seen[seen.length - 1];
+ assert.match(call.url, /^\/api\/v1\/runs\?stream=1&wait=\d+$/);
+ assert.deepStrictEqual([call.auth, call.body.workflow, /^chat-ai-\d+-\d+-t0$/.test(call.key)], ['Bearer chat-ai-token', 'chat.member.turn', true]);
+});
+
+t('an AI without the stream answers JSON: the answer still comes, without deltas', async () => {
+ mode = 'json';
+ const { frames, reply } = await ask('And for dessert?');
+ assert.strictEqual(frames.filter((f) => f.type === 'room_ai_delta').length, 0);
+ assert.strictEqual(reply.message, 'A whole answer.');
+});
+
+t.run(async () => { aiClient._reset(); if (h && h.close) await h.close(); await new Promise((r) => ai.close(r)); });