cento-engine / src /semantic.js
LJTSG's picture
Cento v0.1 — bounded recombinant-memory engine
8494d00 verified
Raw
History Blame Contribute Delete
10.7 kB
// semantic.js — node side of the vector space: builds/loads the fragment
// embedding cache (python MiniLM bridge), embeds queries, ranks by cosine.
'use strict';
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const { spawnSync, spawn } = require('child_process');
const EMBED_PY = path.join(__dirname, 'embed.py');
// PERSISTENT EMBED SERVER: spawn python ONCE (model stays loaded), feed queries
// over stdin, read vectors over stdout. Turns the 7.3s-per-query Python
// cold-start into ~30ms. Async + FIFO (requests serialized in order). Separate
// process + async pipes — NOT the in-process-spawnSync shape that deadlocked.
let _proc = null, _buf = '', _queue = [], _ready = null;
function ensureServer() {
if (_proc) return _ready;
_proc = spawn('python', [EMBED_PY, 'serve'], { stdio: ['pipe', 'pipe', 'ignore'] });
_proc.stdout.setEncoding('utf8');
_ready = new Promise(res => {
_proc.stdout.on('data', d => {
_buf += d;
let nl;
while ((nl = _buf.indexOf('\n')) >= 0) {
const line = _buf.slice(0, nl); _buf = _buf.slice(nl + 1);
if (line.trim() === 'READY') { res(true); continue; }
const r = _queue.shift();
if (r) { try { r(line.trim() === 'null' ? null : Float32Array.from(JSON.parse(line))); } catch (_) { r(null); } }
}
});
});
_proc.on('exit', () => { _proc = null; const q = _queue; _queue = []; for (const r of q) r(null); });
return _ready;
}
async function embedQueryAsync(text) {
await ensureServer();
return new Promise(resolve => { _queue.push(resolve); _proc.stdin.write(String(text).replace(/[\r\n]+/g, ' ') + '\n'); });
}
function shutdownEmbed() { if (_proc) { try { _proc.kill(); } catch (_) {} _proc = null; } }
function textsHash(texts) {
const h = crypto.createHash('sha1');
for (const t of texts) h.update(t + '\n');
return h.digest('hex').slice(0, 12);
}
// generic embedding cache for any text list (fragments, prompts, ...)
function ensureEmbeddings(texts, cacheDir, tag) {
fs.mkdirSync(cacheDir, { recursive: true });
const prefix = path.join(cacheDir, tag + '-' + textsHash(texts));
const metaPath = prefix + '.meta.json';
if (!fs.existsSync(metaPath)) {
const textsPath = prefix + '.texts.json';
fs.writeFileSync(textsPath, JSON.stringify(texts));
process.stderr.write(`[semantic] embedding ${texts.length} ${tag} (one-time, CPU)...\n`);
const r = spawnSync('python', [EMBED_PY, 'cache', textsPath, prefix], { encoding: 'utf8', timeout: 900000 });
if (r.status !== 0) { process.stderr.write('[semantic] cache build failed: ' + (r.stderr || '').slice(-300) + '\n'); return null; }
}
const meta = JSON.parse(fs.readFileSync(metaPath, 'utf8'));
const buf = fs.readFileSync(prefix + '.f32');
return { vectors: new Float32Array(buf.buffer, buf.byteOffset, meta.n * meta.d), n: meta.n, d: meta.d };
}
function ensureFragmentEmbeddings(store, cacheDir) {
// retrieval key ≠ surface form: embed with full original context
// (diary headers etc.) while the composer emits only clean text
return ensureEmbeddings(store.fragments.map(f => f.embedText || f.text), cacheDir, 'frags');
}
function ensurePromptEmbeddings(store, cacheDir) {
if (!store.prompts || !store.prompts.length) return null;
return ensureEmbeddings(store.prompts, cacheDir, 'prompts');
}
// DISK-BACKED QUERY-EMBEDDING CACHE (R94): torch CPU mean-pooling sums in a
// thread-scheduling-dependent order, so the SAME text yields subtly different
// vectors across runs as box load changes the effective parallelism — non-null,
// so the nullEmb guard misses it. That drift fed compose different anchors and
// silently moved eval results run-to-run with ZERO code change (R89/R93: garden
// 1.0→0.802, identical semTopic, different text). Caching the first-computed
// vector per text makes the bridge bit-reproducible: query+fragment vectors stay
// in ONE numerical regime (fragments are already cached on disk), so the eval is
// finally deterministic across runs. Delete cache/qembed-cache.json to re-baseline.
const QCACHE_PATH = path.join(__dirname, '..', 'cache', 'qembed-cache.json');
let _qcache = null;
function _loadQCache() {
if (_qcache) return _qcache;
try { _qcache = JSON.parse(fs.readFileSync(QCACHE_PATH, 'utf8')); } catch (_) { _qcache = {}; }
return _qcache;
}
function _qkey(text) { return crypto.createHash('sha1').update(String(text)).digest('hex').slice(0, 16); }
function embedQuery(text) {
const cache = _loadQCache();
const key = _qkey(text);
if (cache[key]) return Float32Array.from(cache[key]);
// Under heavy box load a fresh python spawn can time out or fail; a silent null
// drops compose to keyword-only anchoring. Retry transient failures (R88), then
// cache the result so every later run reuses the EXACT vector (R94).
for (let attempt = 0; attempt < 3; attempt++) {
const r = spawnSync('python', [EMBED_PY, 'query', text], { encoding: 'utf8', timeout: 120000 });
if (r.status === 0) {
try {
const v = Float32Array.from(JSON.parse(r.stdout.trim().split('\n').pop()));
cache[key] = Array.from(v);
try { fs.mkdirSync(path.dirname(QCACHE_PATH), { recursive: true }); fs.writeFileSync(QCACHE_PATH, JSON.stringify(cache)); } catch (_) {}
return v;
} catch (_) { /* retry */ }
}
}
return null;
}
// cosine of query vs fragment i (vectors are L2-normalized -> dot product)
function sim(emb, i, q) {
const d = emb.d;
let s = 0;
const off = i * d;
for (let k = 0; k < d; k++) s += emb.vectors[off + k] * q[k];
return s;
}
// top-k semantic matches -> Map(index -> normalized 0..1 score)
function semanticRank(emb, q, k) {
const scores = [];
for (let i = 0; i < emb.n; i++) scores.push([i, sim(emb, i, q)]);
scores.sort((a, b) => b[1] - a[1]);
const top = scores.slice(0, k || 80);
const max = top[0][1], min = top[top.length - 1][1];
const m = new Map();
for (const [i, s] of top) m.set(i, max > min ? (s - min) / (max - min) : 1);
// ABSOLUTE confidence (raw best cosine), preserved alongside the normalized
// map — min-max normalization erases it, but the floor-miss detector needs to
// know whether ANYTHING in the corpus actually addresses the query (R3's law:
// a channel's authority scales with absolute confidence, never relative rank).
m.confidence = max;
return m;
}
// ---------------- TRAINED ANSWER PROJECTION (R167) ----------------
// A low-rank projection P (d x r), trained contrastively on the corpus's own
// (prompt -> answer-fragment) pairs, so a QUERY lands near the fragments that
// ANSWERED similar prompts instead of the ones that merely ECHO its shape.
// Fully bounded: P only reshapes the RANKING of verbatim fragments; it never
// invents text and never calls an LLM. Gated by availability — corpora with no
// <tag>.proj.json get null and retrieval is byte-identical to before.
const _projCache = new Map();
function loadProjection(tag, baseDir) {
if (!tag) return null;
if (_projCache.has(tag)) return _projCache.get(tag);
const p = path.join(baseDir || path.join(__dirname, '..', 'retrieval-train'), tag + '.proj.json');
let proj = null;
try {
if (fs.existsSync(p)) {
const j = JSON.parse(fs.readFileSync(p, 'utf8'));
// flatten W (d x r) to a typed array for fast projection
const W = new Float32Array(j.d * j.r);
for (let a = 0; a < j.d; a++) for (let b = 0; b < j.r; b++) W[a * j.r + b] = j.W[a][b];
proj = { d: j.d, r: j.r, W };
}
} catch (_) { proj = null; }
_projCache.set(tag, proj);
return proj;
}
// project + L2-normalize one vector (vec offset `off` into a flat array)
function _projectVec(vec, off, proj) {
const { d, r, W } = proj;
const out = new Float32Array(r);
for (let k = 0; k < r; k++) { let s = 0; for (let j = 0; j < d; j++) s += vec[off + j] * W[j * r + k]; out[k] = s; }
let n = 0; for (let k = 0; k < r; k++) n += out[k] * out[k]; n = Math.sqrt(n) || 1;
for (let k = 0; k < r; k++) out[k] /= n;
return out;
}
// answer channel: rank fragments by projected-query x projected-fragment cosine.
// Returns Map(fragmentIndex -> normalized 0..1), with .confidence = best raw sim.
function answerRank(emb, q, proj, k) {
if (!proj || !emb) return null;
const qp = _projectVec(q, 0, proj);
const r = proj.r;
const scores = [];
for (let i = 0; i < emb.n; i++) {
const fp = _projectVec(emb.vectors, i * emb.d, proj);
let v = 0; for (let t = 0; t < r; t++) v += fp[t] * qp[t];
scores.push([i, v]);
}
scores.sort((a, b) => b[1] - a[1]);
const top = scores.slice(0, k || 80);
const max = top[0][1], min = top[top.length - 1][1];
const m = new Map();
for (const [i, s] of top) m.set(i, max > min ? (s - min) / (max - min) : 1);
m.confidence = max;
return m;
}
// stimulus channel: score each FRAGMENT by how similar its parent prompt was
// to the current query — the conversational reflex, made geometric.
// Returns { map, confidence } — confidence is the ABSOLUTE best cosine, so a
// corpus that has never seen a stimulus like this admits it instead of
// shouting noise. (Rank-normalization alone erased this and broke R3 v1.)
function stimulusRank(promptEmb, q, fragments, k) {
if (!promptEmb) return null;
const pScore = [];
for (let i = 0; i < promptEmb.n; i++) pScore.push([i, sim(promptEmb, i, q)]);
pScore.sort((a, b) => b[1] - a[1]);
const top = pScore.slice(0, Math.min(k || 12, pScore.length));
const confidence = top[0][1];
const max = top[0][1], min = top[top.length - 1][1];
const pNorm = new Map();
for (const [i, s] of top) pNorm.set(i, max > min ? (s - min) / (max - min) : 1);
const m = new Map();
fragments.forEach((f, fi) => {
if (f.promptIdx >= 0 && pNorm.has(f.promptIdx)) m.set(fi, pNorm.get(f.promptIdx));
});
return { map: m, confidence };
}
// is the query a life-event SHARE (statement about their world) vs a question?
function eventness(query) {
const q = query.trim();
if (/\?\s*$/.test(q)) return 0.2;
const pastShare = /\b(i|we|my|me)\b[^?]*\b(finished|talked|did|went|got|made|had|found|fixed|broke|lost|won|started|quit|saw|met|built|wrote)\b/i.test(q);
return pastShare ? 0.85 : 0.45;
}
// cosine between two cached fragment vectors (both L2-normalized)
function pairSim(emb, i, j) {
const d = emb.d;
let s = 0;
const oi = i * d, oj = j * d;
for (let k = 0; k < d; k++) s += emb.vectors[oi + k] * emb.vectors[oj + k];
return s;
}
module.exports = { ensureFragmentEmbeddings, ensurePromptEmbeddings, embedQuery, embedQueryAsync, shutdownEmbed, semanticRank, stimulusRank, eventness, pairSim, loadProjection, answerRank };