openvibers/openvibe.actor Commits
Route hosted Actor turns through Exchange (#429)
Changes
@@@ -41,8 +41,12 @@ ACTOR_EVENTS_URL=# ACTOR_AI_ROUTER=1 every model call is a run of ai.complete on AI, which picks the offer by price, latency and# policy and says which model ran; Actor then needs NO provider key below. The person a task belongs to rides as# on_behalf_of, so their allowance and their Balance pay; the run is theirs.--# Production sets this; a self-hosted Actor leaves it off and keeps its own model below.++# Hosted production defaults on; a self-hosted Actor leaves it off and keeps its own model below.ACTOR_AI_ROUTER=++# Hosted turns use Exchange's compatible API with Actor's existing service token. A slow class or++# targeted run still uses AI's runs API; 0 rolls ordinary turns back to the runs API.++# ACTOR_EXCHANGE_ROUTER=0++# ACTOR_EXCHANGE_URL=https://api.openvibe.exchange# The token is Actor's service token for audience openvibe.ai (ai.run.create, namespace ai.complete).# OV_AI_URL=https://ai.openvibe.services# ACTOR_AI_TIMEOUT_MS=180000@@@ -100,9 +100,13 @@ function load(env = process.env) {timeoutMs: Math.max(5000, int(env.ACTOR_AI_TIMEOUT_MS, 180_000)),waitMs: Math.max(1000, int(env.ACTOR_AI_WAIT_MS, 120_000)),// The AI layer of every product: with this on, the tiers below run on AI's router and the operator's-- // provider keys (ACTOR_DEEPSEEK_API_KEY, ACTOR_OPENAI_API_KEY, …) are not used. Off by default so a-- // self-hosted Actor keeps its own model choices; production sets ACTOR_AI_ROUTER=1.-- router: bool(env.ACTOR_AI_ROUTER),++ // provider keys (ACTOR_DEEPSEEK_API_KEY, ACTOR_OPENAI_API_KEY, …) are not used. Hosted++ // Actor uses this by default; self-hosted Actor keeps its own model choices.++ router: env.ACTOR_AI_ROUTER == null ? isProduction && !selfHosted : bool(env.ACTOR_AI_ROUTER),++ // Hosted chat completions use Exchange's compatible API. Slow classes and targeted runs++ // keep the runs API, which carries their extra controls. Set 0 to use the runs API for all.++ exchangeRouter: env.ACTOR_EXCHANGE_ROUTER == null ? isProduction && !selfHosted : bool(env.ACTOR_EXCHANGE_ROUTER),++ exchangeUrl: trim(env.ACTOR_EXCHANGE_URL || 'https://api.openvibe.exchange'),// Each tier's policy on the router: floor and ceiling in one name (economy = cheapest healthy, fast =// quickest, balanced = price and latency, auto = the router's own, private = OpenVibe-run/local only).// An exact id (<org>/<model>@<revision>) pins a model when AI's catalog is on.@@@ -543,6 +543,56 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =};}++ /** A normal hosted turn through Exchange, with the same service token and person attribution. */++ async function chatViaExchange({ tier, messages, tools, json, maxTokens, signal, onDelta, subject, policy }) {++ const c = routerClient();++ if (!c) throw new ModelError('OpenVibe.Exchange is not configured on this server', { code: 'actor.model.unavailable' });++ const body = {++ model: (currentChoice() && currentChoice().model) || routerModel(tier),++ messages,++ ...(tools && tools.length ? { tools, tool_choice: 'auto' } : {}),++ ...(json ? { response_format: { type: 'json_object' } } : {}),++ max_tokens: Math.max(1, Math.min(Number(maxTokens) || 1500, 16000)),++ temperature: 0.2,++ ...(ai.maxCostUsd != null ? { max_cost_usd: ai.maxCostUsd } : {}),++ ...(subject ? { on_behalf_of: { type: 'user', id: String(subject) } } : {}),++ ...(policy === 'uncensored' ? { policy: 'uncensored' } : {}),++ ...(onDelta ? { stream: true, stream_options: { include_usage: true } } : {}),++ };++ const sig = signal ? AbortSignal.any([signal, AbortSignal.timeout(ai.timeoutMs || 180_000)]) : AbortSignal.timeout(ai.timeoutMs || 180_000);++ let res;++ try {++ res = await fetchImpl(`${trimSlash(ai.exchangeUrl || 'https://api.openvibe.exchange')}/v1/chat/completions`, {++ method: 'POST',++ headers: { authorization: `Bearer ${await c.tokens.getToken()}`, 'content-type': 'application/json', accept: onDelta ? 'text/event-stream' : 'application/json' },++ body: JSON.stringify(body), signal: sig,++ });++ } catch (err) {++ if (signal && signal.aborted) throw new ProviderError('cancelled', { code: 'actor.task.cancelled' });++ throw new ProviderError(err && err.name === 'TimeoutError' ? 'OpenVibe.Exchange did not answer in time' : 'OpenVibe.Exchange could not be reached', { code: 'actor.provider.unreachable' });++ }++ if (!res.ok) {++ const answer = await res.json().catch(() => null);++ const err = (answer && answer.error) || {};++ throw new ProviderError(`OpenVibe.Exchange HTTP ${res.status}: ${clip(err.message || err.code || 'request failed', 300)}`, { status: res.status, code: res.status === 429 ? 'actor.provider.rate_limited' : 'actor.provider.error' });++ }++ const answer = onDelta ? await readStream(res, onDelta, { signal, who: 'OpenVibe.Exchange' }) : await res.json().catch(() => null);++ if (!answer || !Array.isArray(answer.choices)) throw new ProviderError('OpenVibe.Exchange answered without a completion');++ const first = answer.choices[0] || {};++ const message = first.message || { role: 'assistant', content: '' };++ const u = answer.usage || {};++ const x = answer.x_openvibe || {};++ return {++ message, text: typeof message.content === 'string' ? message.content : '',++ toolCalls: Array.isArray(message.tool_calls) ? message.tool_calls : [],++ finish: first.finish_reason || null,++ model: x.resolved_model || answer.model || body.model,++ usage: { input: u.prompt_tokens || 0, cached: 0, output: u.completion_tokens || 0 },++ cost_usd: Number(x.cost_usd) || 0,++ latency_class: null,++ };++ }++/** The router's streamed run (POST /api/v1/runs?stream=1): `delta` events as they come, then the run. */async function routerStream(c, body, { onDelta, signal }) {const { response } = await c.client.request({@@@ -592,7 +642,10 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =}// The router (the AI layer): the tier names a policy on AI and the person's allowance pays. No provider key.if (routerOn() || (choice && choice.model && !choice.provider)) {-- const r = await chatViaRouter({ tier, messages, tools, json, maxTokens, signal, onDelta, subject: subject || (choice && choice.subject), target: choice && choice.target, latency, policy: policyOf });++ const args = { tier, messages, tools, json, maxTokens, signal, onDelta, subject: subject || (choice && choice.subject), target: choice && choice.target, latency, policy: policyOf };++ const r = routerOn() && ai.exchangeRouter && !args.target && !args.latency++ ? await chatViaExchange(args)++ : await chatViaRouter(args);return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER };}await warm();nnew file mode 100644@@@ -0,0 +1,48 @@++'use strict';++const assert = require('assert');++const { load } = require('../server/config');++const { createModels } = require('../server/runtime/models');++++const subject = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ';++const env = { NODE_ENV: 'production', OV_NETWORK_INTERNAL_URL: 'https://network.test', OV_OAUTH_CLIENT_SECRET: 'fixture-secret' };++const calls = [];++const fetchImpl = async (url, opts = {}) => {++ if (String(url).endsWith('/oauth/token')) return new Response(JSON.stringify({ access_token: 'fixture-token', expires_in: 300 }), { status: 200, headers: { 'content-type': 'application/json' } });++ calls.push({ url: String(url), headers: opts.headers, body: JSON.parse(opts.body) });++ if (JSON.parse(opts.body).stream) {++ const chunks = [++ { model: 'local/model', choices: [{ index: 0, delta: { content: 'canary ' } }] },++ { choices: [{ index: 0, delta: { content: 'ok' }, finish_reason: 'stop' }] },++ { usage: { prompt_tokens: 8, completion_tokens: 3 }, x_openvibe: { cost_usd: 0.00002, receipt_id: 'rcpt_test' } },++ ];++ return new Response(chunks.map((chunk) => `data: ${JSON.stringify(chunk)}\n\n`).join('') + 'data: [DONE]\n\n', { status: 200, headers: { 'content-type': 'text/event-stream' } });++ }++ return new Response(JSON.stringify({ model: 'local/model', choices: [{ message: { role: 'assistant', content: 'canary ok' }, finish_reason: 'stop' }], usage: { prompt_tokens: 8, completion_tokens: 3 }, x_openvibe: { cost_usd: 0.00002, receipt_id: 'rcpt_test' } }), { status: 200, headers: { 'content-type': 'application/json' } });++};++++(async () => {++ const config = load(env);++ assert.strictEqual(config.ai.router, true);++ assert.strictEqual(config.ai.exchangeRouter, true);++ assert.strictEqual(load({ ...env, ACTOR_SELF_HOSTED: '1' }).ai.router, false);++ assert.strictEqual(load({ ...env, ACTOR_EXCHANGE_ROUTER: '0' }).ai.exchangeRouter, false);++ const models = createModels({ config, fetchImpl, log: { warn() {} } });++ const result = await models.chat({ tier: 'smart', subject, messages: [{ role: 'user', content: 'Reply exactly: canary ok' }], tools: [{ type: 'function', function: { name: 'ping', parameters: { type: 'object' } } }] });++ assert.strictEqual(calls.length, 1);++ assert.strictEqual(calls[0].url, 'https://api.openvibe.exchange/v1/chat/completions');++ assert.strictEqual(calls[0].headers.authorization, 'Bearer fixture-token');++ assert.deepStrictEqual(calls[0].body.on_behalf_of, { type: 'user', id: subject });++ assert.strictEqual(calls[0].body.model, 'openvibe/balanced');++ assert.strictEqual(calls[0].body.tools[0].function.name, 'ping');++ assert.strictEqual(result.text, 'canary ok');++ assert.strictEqual(result.model, 'local/model');++ assert.strictEqual(result.cost_usd, 0.00002);++ assert.deepStrictEqual(result.usage, { input: 8, cached: 0, output: 3 });++ const pieces = [];++ const streamed = await models.chat({ tier: 'fast', subject, messages: [{ role: 'user', content: 'Stream' }], onDelta: (piece) => pieces.push(piece) });++ assert.deepStrictEqual(pieces, ['canary ', 'ok']);++ assert.strictEqual(streamed.text, 'canary ok');++ assert.strictEqual(streamed.cost_usd, 0.00002);++ assert.strictEqual(calls[1].body.stream, true);++ console.log('exchange-router: all passed');++})().catch((err) => { console.error(err); process.exitCode = 1; });
@@ -41,8 +41,12 @@ ACTOR_EVENTS_URL=
@@ -41,8 +41,12 @@ ACTOR_EVENTS_URL=
# ACTOR_AI_ROUTER=1 every model call is a run of ai.complete on AI, which picks the offer by price, latency and
# ACTOR_AI_ROUTER=1 every model call is a run of ai.complete on AI, which picks the offer by price, latency and
# policy and says which model ran; Actor then needs NO provider key below. The person a task belongs to rides as
# policy and says which model ran; Actor then needs NO provider key below. The person a task belongs to rides as
# on_behalf_of, so their allowance and their Balance pay; the run is theirs.
# on_behalf_of, so their allowance and their Balance pay; the run is theirs.
-# Production sets this; a self-hosted Actor leaves it off and keeps its own model below.
+# Hosted production defaults on; a self-hosted Actor leaves it off and keeps its own model below.
ACTOR_AI_ROUTER=
ACTOR_AI_ROUTER=
+# Hosted turns use Exchange's compatible API with Actor's existing service token. A slow class or
+# targeted run still uses AI's runs API; 0 rolls ordinary turns back to the runs API.
+# ACTOR_EXCHANGE_ROUTER=0
+# ACTOR_EXCHANGE_URL=https://api.openvibe.exchange
# The token is Actor's service token for audience openvibe.ai (ai.run.create, namespace ai.complete).
# The token is Actor's service token for audience openvibe.ai (ai.run.create, namespace ai.complete).
# OV_AI_URL=https://ai.openvibe.services
# OV_AI_URL=https://ai.openvibe.services
# ACTOR_AI_TIMEOUT_MS=180000
# ACTOR_AI_TIMEOUT_MS=180000
@@ -100,9 +100,13 @@ function load(env = process.env) {
@@ -100,9 +100,13 @@ function load(env = process.env) {
timeoutMs: Math.max(5000, int(env.ACTOR_AI_TIMEOUT_MS, 180_000)),
timeoutMs: Math.max(5000, int(env.ACTOR_AI_TIMEOUT_MS, 180_000)),
waitMs: Math.max(1000, int(env.ACTOR_AI_WAIT_MS, 120_000)),
waitMs: Math.max(1000, int(env.ACTOR_AI_WAIT_MS, 120_000)),
// The AI layer of every product: with this on, the tiers below run on AI's router and the operator's
// The AI layer of every product: with this on, the tiers below run on AI's router and the operator's
- // provider keys (ACTOR_DEEPSEEK_API_KEY, ACTOR_OPENAI_API_KEY, …) are not used. Off by default so a
- // self-hosted Actor keeps its own model choices; production sets ACTOR_AI_ROUTER=1.
- router: bool(env.ACTOR_AI_ROUTER),
+ // provider keys (ACTOR_DEEPSEEK_API_KEY, ACTOR_OPENAI_API_KEY, …) are not used. Hosted
+ // Actor uses this by default; self-hosted Actor keeps its own model choices.
+ router: env.ACTOR_AI_ROUTER == null ? isProduction && !selfHosted : bool(env.ACTOR_AI_ROUTER),
+ // Hosted chat completions use Exchange's compatible API. Slow classes and targeted runs
+ // keep the runs API, which carries their extra controls. Set 0 to use the runs API for all.
+ exchangeRouter: env.ACTOR_EXCHANGE_ROUTER == null ? isProduction && !selfHosted : bool(env.ACTOR_EXCHANGE_ROUTER),
+ exchangeUrl: trim(env.ACTOR_EXCHANGE_URL || 'https://api.openvibe.exchange'),
// Each tier's policy on the router: floor and ceiling in one name (economy = cheapest healthy, fast =
// Each tier's policy on the router: floor and ceiling in one name (economy = cheapest healthy, fast =
// quickest, balanced = price and latency, auto = the router's own, private = OpenVibe-run/local only).
// quickest, balanced = price and latency, auto = the router's own, private = OpenVibe-run/local only).
// An exact id (<org>/<model>@<revision>) pins a model when AI's catalog is on.
// An exact id (<org>/<model>@<revision>) pins a model when AI's catalog is on.
@@ -543,6 +543,56 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -543,6 +543,56 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
};
};
}
}
+ /** A normal hosted turn through Exchange, with the same service token and person attribution. */
+ async function chatViaExchange({ tier, messages, tools, json, maxTokens, signal, onDelta, subject, policy }) {
+ const c = routerClient();
+ if (!c) throw new ModelError('OpenVibe.Exchange is not configured on this server', { code: 'actor.model.unavailable' });
+ const body = {
+ model: (currentChoice() && currentChoice().model) || routerModel(tier),
+ messages,
+ ...(tools && tools.length ? { tools, tool_choice: 'auto' } : {}),
+ ...(json ? { response_format: { type: 'json_object' } } : {}),
+ max_tokens: Math.max(1, Math.min(Number(maxTokens) || 1500, 16000)),
+ temperature: 0.2,
+ ...(ai.maxCostUsd != null ? { max_cost_usd: ai.maxCostUsd } : {}),
+ ...(subject ? { on_behalf_of: { type: 'user', id: String(subject) } } : {}),
+ ...(policy === 'uncensored' ? { policy: 'uncensored' } : {}),
+ ...(onDelta ? { stream: true, stream_options: { include_usage: true } } : {}),
+ };
+ const sig = signal ? AbortSignal.any([signal, AbortSignal.timeout(ai.timeoutMs || 180_000)]) : AbortSignal.timeout(ai.timeoutMs || 180_000);
+ let res;
+ try {
+ res = await fetchImpl(`${trimSlash(ai.exchangeUrl || 'https://api.openvibe.exchange')}/v1/chat/completions`, {
+ method: 'POST',
+ headers: { authorization: `Bearer ${await c.tokens.getToken()}`, 'content-type': 'application/json', accept: onDelta ? 'text/event-stream' : 'application/json' },
+ body: JSON.stringify(body), signal: sig,
+ });
+ } catch (err) {
+ if (signal && signal.aborted) throw new ProviderError('cancelled', { code: 'actor.task.cancelled' });
+ throw new ProviderError(err && err.name === 'TimeoutError' ? 'OpenVibe.Exchange did not answer in time' : 'OpenVibe.Exchange could not be reached', { code: 'actor.provider.unreachable' });
+ }
+ if (!res.ok) {
+ const answer = await res.json().catch(() => null);
+ const err = (answer && answer.error) || {};
+ throw new ProviderError(`OpenVibe.Exchange HTTP ${res.status}: ${clip(err.message || err.code || 'request failed', 300)}`, { status: res.status, code: res.status === 429 ? 'actor.provider.rate_limited' : 'actor.provider.error' });
+ }
+ const answer = onDelta ? await readStream(res, onDelta, { signal, who: 'OpenVibe.Exchange' }) : await res.json().catch(() => null);
+ if (!answer || !Array.isArray(answer.choices)) throw new ProviderError('OpenVibe.Exchange answered without a completion');
+ const first = answer.choices[0] || {};
+ const message = first.message || { role: 'assistant', content: '' };
+ const u = answer.usage || {};
+ const x = answer.x_openvibe || {};
+ return {
+ message, text: typeof message.content === 'string' ? message.content : '',
+ toolCalls: Array.isArray(message.tool_calls) ? message.tool_calls : [],
+ finish: first.finish_reason || null,
+ model: x.resolved_model || answer.model || body.model,
+ usage: { input: u.prompt_tokens || 0, cached: 0, output: u.completion_tokens || 0 },
+ cost_usd: Number(x.cost_usd) || 0,
+ latency_class: null,
+ };
+ }
+
/** The router's streamed run (POST /api/v1/runs?stream=1): `delta` events as they come, then the run. */
/** The router's streamed run (POST /api/v1/runs?stream=1): `delta` events as they come, then the run. */
async function routerStream(c, body, { onDelta, signal }) {
async function routerStream(c, body, { onDelta, signal }) {
const { response } = await c.client.request({
const { response } = await c.client.request({
@@ -592,7 +642,10 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -592,7 +642,10 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
}
}
// The router (the AI layer): the tier names a policy on AI and the person's allowance pays. No provider key.
// The router (the AI layer): the tier names a policy on AI and the person's allowance pays. No provider key.
if (routerOn() || (choice && choice.model && !choice.provider)) {
if (routerOn() || (choice && choice.model && !choice.provider)) {
- const r = await chatViaRouter({ tier, messages, tools, json, maxTokens, signal, onDelta, subject: subject || (choice && choice.subject), target: choice && choice.target, latency, policy: policyOf });
+ const args = { tier, messages, tools, json, maxTokens, signal, onDelta, subject: subject || (choice && choice.subject), target: choice && choice.target, latency, policy: policyOf };
+ const r = routerOn() && ai.exchangeRouter && !args.target && !args.latency
+ ? await chatViaExchange(args)
+ : await chatViaRouter(args);
return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER };
return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER };
}
}
await warm();
await warm();
new file mode 100644
new file mode 100644
@@ -0,0 +1,48 @@
@@ -0,0 +1,48 @@
+'use strict';
+const assert = require('assert');
+const { load } = require('../server/config');
+const { createModels } = require('../server/runtime/models');
+
+const subject = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ';
+const env = { NODE_ENV: 'production', OV_NETWORK_INTERNAL_URL: 'https://network.test', OV_OAUTH_CLIENT_SECRET: 'fixture-secret' };
+const calls = [];
+const fetchImpl = async (url, opts = {}) => {
+ if (String(url).endsWith('/oauth/token')) return new Response(JSON.stringify({ access_token: 'fixture-token', expires_in: 300 }), { status: 200, headers: { 'content-type': 'application/json' } });
+ calls.push({ url: String(url), headers: opts.headers, body: JSON.parse(opts.body) });
+ if (JSON.parse(opts.body).stream) {
+ const chunks = [
+ { model: 'local/model', choices: [{ index: 0, delta: { content: 'canary ' } }] },
+ { choices: [{ index: 0, delta: { content: 'ok' }, finish_reason: 'stop' }] },
+ { usage: { prompt_tokens: 8, completion_tokens: 3 }, x_openvibe: { cost_usd: 0.00002, receipt_id: 'rcpt_test' } },
+ ];
+ return new Response(chunks.map((chunk) => `data: ${JSON.stringify(chunk)}\n\n`).join('') + 'data: [DONE]\n\n', { status: 200, headers: { 'content-type': 'text/event-stream' } });
+ }
+ return new Response(JSON.stringify({ model: 'local/model', choices: [{ message: { role: 'assistant', content: 'canary ok' }, finish_reason: 'stop' }], usage: { prompt_tokens: 8, completion_tokens: 3 }, x_openvibe: { cost_usd: 0.00002, receipt_id: 'rcpt_test' } }), { status: 200, headers: { 'content-type': 'application/json' } });
+};
+
+(async () => {
+ const config = load(env);
+ assert.strictEqual(config.ai.router, true);
+ assert.strictEqual(config.ai.exchangeRouter, true);
+ assert.strictEqual(load({ ...env, ACTOR_SELF_HOSTED: '1' }).ai.router, false);
+ assert.strictEqual(load({ ...env, ACTOR_EXCHANGE_ROUTER: '0' }).ai.exchangeRouter, false);
+ const models = createModels({ config, fetchImpl, log: { warn() {} } });
+ const result = await models.chat({ tier: 'smart', subject, messages: [{ role: 'user', content: 'Reply exactly: canary ok' }], tools: [{ type: 'function', function: { name: 'ping', parameters: { type: 'object' } } }] });
+ assert.strictEqual(calls.length, 1);
+ assert.strictEqual(calls[0].url, 'https://api.openvibe.exchange/v1/chat/completions');
+ assert.strictEqual(calls[0].headers.authorization, 'Bearer fixture-token');
+ assert.deepStrictEqual(calls[0].body.on_behalf_of, { type: 'user', id: subject });
+ assert.strictEqual(calls[0].body.model, 'openvibe/balanced');
+ assert.strictEqual(calls[0].body.tools[0].function.name, 'ping');
+ assert.strictEqual(result.text, 'canary ok');
+ assert.strictEqual(result.model, 'local/model');
+ assert.strictEqual(result.cost_usd, 0.00002);
+ assert.deepStrictEqual(result.usage, { input: 8, cached: 0, output: 3 });
+ const pieces = [];
+ const streamed = await models.chat({ tier: 'fast', subject, messages: [{ role: 'user', content: 'Stream' }], onDelta: (piece) => pieces.push(piece) });
+ assert.deepStrictEqual(pieces, ['canary ', 'ok']);
+ assert.strictEqual(streamed.text, 'canary ok');
+ assert.strictEqual(streamed.cost_usd, 0.00002);
+ assert.strictEqual(calls[1].body.stream, true);
+ console.log('exchange-router: all passed');
+})().catch((err) => { console.error(err); process.exitCode = 1; });