openvibers/openvibe.ai Commits

Measure prompt-cache reuse per offer; cache only the stable system block (#414)

OP OpenVibers committed 4 days ago · d31b8d3943 · 6 files, +140 −12

* Measure prompt-cache reuse per offer, and cache only the stable system block

The Exchange catalog now reports cache_hit_share_7d per offer and per
model: of the prompt tokens provider attempts reported in the window, the
share the provider served from its own prompt cache, from the cached token
counts each attempt already records (requests.tokens_cached). It needs the
same 20 real attempts as uptime and is null, never 0, when nothing was
measured; a cached answer made no call and is not a sample. The field is
documented in openapi.json.

The workflow engine now sends a template's stable rules as one system block
with the cache marker and the person's preferences as a second, unmarked
block after it, so a breakpoint on the rules covers the stable prefix and
their edits no longer invalidate it. Anthropic's cache_control goes on the
last stable block; OpenAI-compatible endpoints join the blocks as before.

* Measure prompt cache from real traffic and preserve preferences

* Merge current market fixtures and workflow fixes

Changes

ddiff --git a/server/exchange/catalog.js b/server/exchange/catalog.js
iindex bb55176..895b1e6 100644
---- a/server/exchange/catalog.js
++++ b/server/exchange/catalog.js
@@@ -261,7 +261,7 @@ function createCatalog({ db, registry, pool, env = process.env, clock = { now: (
// What those offers actually did: real runs (cached answers counted apart), TTFT where recorded,
// else the answering attempt's total latency.
const measured = await seen.read({ days });
-- observed.fold(measured, drafts);
++ const measuredStats = observed.fold(measured, drafts);
// The router's own measurements win once they have 20 samples; until then the observed runs' numbers stay.
const performance = await db.prepare('SELECT * FROM provider_metrics').all();
for (const d of drafts) {
@@@ -285,6 +285,17 @@ function createCatalog({ db, registry, pool, env = process.env, clock = { now: (
m.requests_today = m.offers.reduce((n, o) => n + o.requests_today, 0);
m.cached_7d = m.offers.reduce((n, o) => n + o.cached_7d, 0);
m.cached_today = m.offers.reduce((n, o) => n + o.cached_today, 0);
++ // The model's own prompt-cache share over the same attempts, across its offers (null when no offer has
++ // enough attempts to measure one).
++ let tokensIn = 0, tokensCached = 0;
++ for (const d of drafts) {
++ if (!m.offers.includes(d.offer)) continue;
++ const s = measuredStats.get(d);
++ tokensIn += s.tokens_in;
++ tokensCached += s.tokens_cached;
++ }
++ m.cache_hit_share_7d = tokensIn > 0 && m.offers.some((o) => o.cache_hit_share_7d != null)
++ ? Math.min(1, Math.max(0, Math.round((tokensCached / tokensIn) * 10000) / 10000)) : null;
const activity = new Map();
for (const d of daysByModel.filter(d=>m.offers.some(o=>o.model===d.model_key))) activity.set(d.activity_day,(activity.get(d.activity_day)||0)+Number(d.requests));
m.activity = [...activity].sort(([a],[b])=>a.localeCompare(b)).map(([day,requests])=>({day,requests}));
ddiff --git a/server/exchange/observed.js b/server/exchange/observed.js
iindex 2275890..80f9608 100644
---- a/server/exchange/observed.js
++++ b/server/exchange/observed.js
@@@ -82,6 +82,7 @@ function createObserved({ db, clock = { now: () => Date.now() } }) {
JOIN runs r ON r.id = q.run_id
WHERE q.status IN ('ok', 'error', 'timeout')
AND r.status IN ('succeeded', 'failed')
++ AND r.synthetic = 0
AND r.created_at::timestamptz >= ?::timestamptz AND r.created_at::timestamptz <= ?::timestamptz
GROUP BY 1, 2, 3, 4`);
// A cached answer called no provider; it follows the run it reused (`cached_from`).
@@@ -92,9 +93,22 @@ function createObserved({ db, clock = { now: () => Date.now() } }) {
COUNT(*)::bigint AS n
FROM runs r
LEFT JOIN runs src ON src.id = r.cached_from
-- WHERE r.status = 'cached'
++ WHERE r.status = 'cached' AND r.synthetic = 0
AND r.created_at::timestamptz >= ?::timestamptz AND r.created_at::timestamptz <= ?::timestamptz
GROUP BY 1, 2, 3, 4`);
++ // The prompt tokens provider attempts reported, and how many of them the provider served from its own prompt
++ // cache (server/providers/*.js record usage.cached per attempt). Only attempts that reached the provider count;
++ // a cached answer made no call and is not a sample here.
++ const tokenRows = db.prepare(`SELECT q.provider_key AS provider_key, q.model_key AS model_key,
++ SUM(COALESCE(q.tokens_in, 0))::bigint AS tokens_in,
++ SUM(COALESCE(q.tokens_cached, 0))::bigint AS tokens_cached
++ FROM requests q
++ JOIN runs r ON r.id = q.run_id
++ WHERE q.status = 'ok'
++ AND r.status IN ('succeeded', 'failed')
++ AND r.synthetic = 0
++ AND r.created_at::timestamptz >= ?::timestamptz AND r.created_at::timestamptz <= ?::timestamptz
++ GROUP BY 1, 2`);
const latencyRows = db.prepare(`SELECT q.provider_key AS provider_key,
q.model_key AS model_key,
q.latency_ms, rc.body->>'ttft_ms' AS ttft_ms
@@@ -103,6 +117,7 @@ function createObserved({ db, clock = { now: () => Date.now() } }) {
LEFT JOIN ai_receipts rc ON rc.run_id = r.id
WHERE q.status = 'ok' AND q.latency_ms IS NOT NULL
AND r.status IN ('succeeded', 'failed')
++ AND r.synthetic = 0
AND (r.error_code IS NULL OR r.error_code <> 'output.empty')
AND NOT (q.operation IN ('chat', 'generate', 'summarize', 'classify', 'extract', 'enrich', 'complete', 'messages')
AND COALESCE(q.tokens_out, 0) = 0
@@@ -115,16 +130,16 @@ function createObserved({ db, clock = { now: () => Date.now() } }) {
const span = Math.max(1, Number(days) || 7);
const from = new Date(now - span * DAY_MS).toISOString();
const to = new Date(now).toISOString();
-- const [attempts, cached, latencies] = await Promise.all([
-- attemptRows.all(from, to), cachedRows.all(from, to), latencyRows.all(from, to),
++ const [attempts, cached, latencies, tokens] = await Promise.all([
++ attemptRows.all(from, to), cachedRows.all(from, to), latencyRows.all(from, to), tokenRows.all(from, to),
]);
-- return { from, to, today: dayOf(now), runs: [...attempts, ...cached], latencies };
++ return { from, to, today: dayOf(now), runs: [...attempts, ...cached], latencies, tokens };
}
return { read };
}
--const EMPTY = () => ({ succeeded: 0, failed: 0, cached: 0, succeeded_today: 0, failed_today: 0, cached_today: 0, p50: null, p95: null });
++const EMPTY = () => ({ succeeded: 0, failed: 0, cached: 0, succeeded_today: 0, failed_today: 0, cached_today: 0, tokens_in: 0, tokens_cached: 0, p50: null, p95: null });
/**
* Fold the measured rows into the catalog's offers. `drafts` are `{ providerKey, modelKey, offer }`,
@@@ -169,6 +184,13 @@ function fold(observed, drafts) {
const value = Number.isFinite(ttft) && ttft >= 0 ? ttft : (Number.isFinite(total) && total >= 0 ? total : null);
if (value != null) samples.get(d).push(value);
}
++ for (const row of observed.tokens || []) {
++ const d = pick(row.provider_key, row.model_key);
++ if (!d) continue;
++ const s = stats.get(d);
++ s.tokens_in += Number(row.tokens_in) || 0;
++ s.tokens_cached += Number(row.tokens_cached) || 0;
++ }
for (const [d, values] of samples) {
const s = stats.get(d);
s.p50 = percentile(values, 0.5);
@@@ -185,6 +207,10 @@ function fold(observed, drafts) {
d.offer.requests_today = s.succeeded_today + s.failed_today;
d.offer.cached_7d = s.cached;
d.offer.cached_today = s.cached_today;
++ // The share of the prompt tokens these attempts reported that the provider served from its own prompt cache
++ // (0.4 is 40 %), from the cached token counts each attempt records. null below 20 attempts and when no
++ // attempt reported input tokens: a missing measurement is not zero.
++ d.offer.cache_hit_share_7d = enough && s.tokens_in > 0 ? Math.min(1, Math.max(0, Math.round((s.tokens_cached / s.tokens_in) * 10000) / 10000)) : null;
}
return stats;
}
ddiff --git a/server/openapi.js b/server/openapi.js
iindex 65098d6..e647c55 100644
---- a/server/openapi.js
++++ b/server/openapi.js
@@@ -449,6 +449,7 @@ const exchangeOffer = {
requests_7d: integer({ minimum: 0 }), requests_today: integer({ minimum: 0 }),
cached_7d: integer({ minimum: 0, description: 'Cached answers (no provider call) in the window, counted apart from requests_7d' }),
cached_today: integer({ minimum: 0 }),
++ cache_hit_share_7d: nullable('number', { minimum: 0, maximum: 1, description: 'Of the prompt tokens this offer\'s attempts reported in the window, the share the provider served from its own prompt cache (0.4 is 40 %). Null below 20 attempts or with no reported input tokens, never 0.' }),
data_policy: nullable('object', { properties: { trains: nullable('boolean'), retention_days: nullable('integer'), region: nullable('string'), anonymized: nullable('boolean', { description: 'The provider says the request is anonymised before the model sees it' }) } }),
refusal_rate: nullable('number', { minimum: 0, maximum: 1, description: 'The share of the published probe set this offer refused (0.75 is three probes out of four). null when nobody measured this offer, never 0.' }),
refusal_measured_on: nullable('string', { description: 'The day the refusal rate was measured' }),
@@@ -465,6 +466,7 @@ const exchangeModel = {
modalities: arrayOf(str()), description: nullable('string'),
requests_7d: integer({ minimum: 0 }), requests_today: integer({ minimum: 0 }),
cached_7d: integer({ minimum: 0 }), cached_today: integer({ minimum: 0 }), community: { type: 'boolean' },
++ cache_hit_share_7d: nullable('number', { minimum: 0, maximum: 1, description: 'The prompt-cache share over this model\'s offers in the window (cached prompt tokens / reported prompt tokens). Null when no offer has enough attempts to measure one.' }),
offers: arrayOf(exchangeOffer),
},
};
ddiff --git a/server/workflows/engine.js b/server/workflows/engine.js
iindex a66e382..69d6efe 100644
---- a/server/workflows/engine.js
++++ b/server/workflows/engine.js
@@@ -188,7 +188,16 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
const vars = prep.vars || {};
// The person's ai.preferences (runs on their behalf; server/user-modules.js), after the template's own rules.
const asked = preferenceLines(ctx && ctx.preferences);
-- const system = [render(template.system_prompt, vars).trim(), asked].filter(Boolean).join('\n\n');
++ const rules = render(template.system_prompt, vars).trim();
++ const system = [rules, asked].filter(Boolean).join('\n\n');
++ // Two system blocks, not one: the template's rules are the same for every run of this version, while a
++ // person's preferences change per person and whenever they edit them. The cache breakpoint sits on the rules
++ // block only (Anthropic's cache_control on the last stable block; tools → system → messages, so everything
++ // before it is the cached prefix), and an OpenAI-compatible endpoint joins the blocks back into one message.
++ // Anthropic, "Prompt caching": a breakpoint goes on the last block that stays identical, because a block
++ // that changes every request writes a fresh entry and never hits.
++ // https://platform.claude.com/docs/en/build-with-claude/prompt-caching
++ const systemBlocks = [...(rules ? [{ text: rules, cache: true }] : []), ...(asked ? [{ text: `${rules ? "\n\n" : ""}${asked}` }] : [])];
const user = render(template.user_prompt, vars);
const routeKey = step.route || template.default_route || wf.default_route || 'default.chat';
const route = await routeFor(routeKey);
@@@ -204,7 +213,7 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
const image = step.image && input[step.image] ? await fetcher.loadImage({ ...input[step.image], max_width: input[step.image].max_width || step.image_max_width || 1024 }, { signal: ctx.signal }) : null;
const p = { ...(route.options || {}), ...(step.params || {}), ...(prep.params || {}) };
const req = {
-- system: system ? [{ text: system, cache: true }] : [],
++ system: systemBlocks,
messages: [{ role: 'user', content: user }],
image,
json: jsonSchema ? { name: (template.metadata && template.metadata.json_name) || 'result', schema: jsonSchema, strict: !prep.jsonSchema } : null,
ddiff --git a/test/exchange-observed.test.js b/test/exchange-observed.test.js
iindex 549f8be..e96da42 100644
---- a/test/exchange-observed.test.js
++++ b/test/exchange-observed.test.js
@@@ -23,9 +23,9 @@ async function insertRun({ id = uid('run'), status = 'succeeded', provider = nul
return id;
}
async function insertRequest({ runId, provider = 'deepseek', model = 'deepseek-flash', status = 'ok', operation = 'chat',
-- tokensOut = 7, latency = 100, created = at(0), seqNo = 0 }) {
-- await h.db.prepare(`INSERT INTO requests (run_id, seq, operation, provider_key, model_key, status, tokens_out, latency_ms, created_at)
-- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`).run(runId, seqNo, operation, provider, model, status, tokensOut, latency, created);
++ tokensOut = 7, tokensIn = 0, tokensCached = 0, latency = 100, created = at(0), seqNo = 0 }) {
++ await h.db.prepare(`INSERT INTO requests (run_id, seq, operation, provider_key, model_key, status, tokens_in, tokens_out, tokens_cached, latency_ms, created_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`).run(runId, seqNo, operation, provider, model, status, tokensIn, tokensOut, tokensCached, latency, created);
}
async function reset() {
await h.db.prepare('DELETE FROM requests').run();
@@@ -124,4 +124,68 @@ t.test('8 successes and 0 real failures are 100% only once 20 samples exist', as
assert.deepStrictEqual(offer.speed, { p50_ms: 100, p95_ms: 100 });
});
++t.test('the prompt-cache share comes from the cached token counts of real attempts', async () => {
++ await reset();
++ // 20 attempts of 1,000 prompt tokens; half reported 500 of them served from the provider's prompt cache.
++ for (let i = 0; i < 20; i++) {
++ const id = await insertRun();
++ await insertRequest({ runId: id, tokensIn: 1000, tokensCached: i < 10 ? 500 : 0 });
++ }
++ const offer = flash(await catalog());
++ assert.strictEqual(offer.cache_hit_share_7d, 0.25, '5,000 cached of 20,000 prompt tokens');
++ assert.strictEqual(offer.cached_7d, 0, 'a provider-side cache hit is not a cached answer');
++ const model = (await catalog()).models.find((x) => x.slug === 'deepseek/deepseek-flash');
++ assert.strictEqual(model.cache_hit_share_7d, 0.25);
++});
++
++t.test('below 20 attempts, or with no reported input tokens, the share is null, never 0', async () => {
++ await reset();
++ for (let i = 0; i < 5; i++) {
++ const id = await insertRun();
++ await insertRequest({ runId: id, tokensIn: 100, tokensCached: 100 });
++ }
++ assert.strictEqual(flash(await catalog()).cache_hit_share_7d, null, '5 attempts is below the floor');
++
++ await reset();
++ for (let i = 0; i < 20; i++) {
++ const id = await insertRun();
++ await insertRequest({ runId: id });
++ }
++ assert.strictEqual(flash(await catalog()).cache_hit_share_7d, null, 'no attempt reported input tokens');
++});
++
++t.test('synthetic attempts neither meet the sample floor nor contribute tokens', async () => {
++ await reset();
++ for (let i = 0; i < 20; i++) {
++ const id = await insertRun();
++ await h.db.prepare('UPDATE runs SET synthetic = 1 WHERE id = ?').run(id);
++ await insertRequest({ runId: id, tokensIn: 1000, tokensCached: 1000 });
++ }
++ for (let i = 0; i < 5; i++) {
++ const id = await insertRun();
++ await insertRequest({ runId: id, tokensIn: 1000, tokensCached: 500 });
++ }
++ let offer = flash(await catalog());
++ assert.strictEqual(offer.requests_7d, 5);
++ assert.strictEqual(offer.cache_hit_share_7d, null);
++ for (let i = 0; i < 15; i++) {
++ const id = await insertRun();
++ await insertRequest({ runId: id, tokensIn: 1000, tokensCached: 500 });
++ }
++ offer = flash(await catalog());
++ assert.strictEqual(offer.requests_7d, 20);
++ assert.strictEqual(offer.cache_hit_share_7d, 0.5);
++});
++
++t.test('anomalous cached token counts cannot exceed a whole prompt', async () => {
++ await reset();
++ for (let i = 0; i < 20; i++) {
++ const id = await insertRun();
++ await insertRequest({ runId: id, tokensIn: 100, tokensCached: 200 });
++ }
++ const body = await catalog();
++ assert.strictEqual(flash(body).cache_hit_share_7d, 1);
++ assert.strictEqual(body.models.find(x => x.slug === 'deepseek/deepseek-flash').cache_hit_share_7d, 1);
++});
++
t.run();
ddiff --git a/test/user-modules.test.js b/test/user-modules.test.js
iindex 2c56649..de93cfe 100644
---- a/test/user-modules.test.js
++++ b/test/user-modules.test.js
@@@ -56,6 +56,13 @@ t.test("a run on A's behalf follows A's preferences; history off keeps no input
assert.strictEqual(r.body.run.status, 'succeeded', r.text);
const sys = systemOf();
assert.ok(sys.includes('casual style') && sys.includes('Keep it short.') && sys.includes('a night-shift worker'), sys);
++ // The template's rules are one cached block; the person's preferences follow in a block of their own, after the
++ // cache breakpoint, so editing them leaves the cached prompt prefix intact.
++ const blocks = seam.last.system;
++ assert.ok(Array.isArray(blocks) && blocks.length === 2, `two system blocks, got ${JSON.stringify(blocks).slice(0, 200)}`);
++ assert.strictEqual(blocks[0].cache, true, 'the stable rules block carries the cache marker');
++ assert.ok(!blocks[0].text.includes('casual style'), 'the preferences are not inside the cached block');
++ assert.ok(!blocks[1].cache && blocks[1].text.includes('casual style'), 'the preferences are a separate, uncached block after it');
const row = await h.db.prepare('SELECT input, cache_key, options FROM runs WHERE id = ?').get(r.body.run.id);
assert.strictEqual(row.input, 'null'); assert.strictEqual(row.cache_key, null);
assert.strictEqual(JSON.parse(row.options).history, false);
@@@ -89,6 +96,15 @@ t.test('ai.usage_summary: 30-day runs and tokens per person, written only when c
assert.strictEqual(await mods.summarize('usr_01JAB2C3D4E5F6G7H8J9K0MNPC'), null);
});
--t.test('shutdown', async () => { await h.stop(); await seam.close(); network.close(); });
++t.test('preferences survive a template with no system rules', async () => {
++ await h.registry.createTemplateVersion('ai.summarize', { system_prompt: '' });
++ const r = await summarize({ on_behalf_of: A });
++ assert.strictEqual(r.body.run.status, 'succeeded', r.text);
++ assert.strictEqual(seam.last.system.length, 1);
++ assert.ok(seam.last.system[0].text.includes('casual style'));
++ assert.ok(!seam.last.system[0].cache);
++});
++
++t.test('shutdown' , async () => { await h.stop(); await seam.close(); network.close(); });
t.run();