From d06ea277f4ba04c7c3575b84ebdb693bd1b06054 Mon Sep 17 00:00:00 2001 From: john Date: Tue, 29 Sep 2026 06:19:56 +0000 Subject: [PATCH] fix(embeddings): keep the semantic layer working on a slow CPU model MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 线上 seed-SER9 上 bge-m3 跑在繁忙服务器的 CPU 上，每条约 3.5 秒：一批 16 条要 60 秒，超过 15 秒超时，后台补齐每次都失败， 向量一直是 0。 - 后台补齐改为每批 4 条，超时按条数放宽（每条 30 秒），新技能优先 - 查询向量单独用较短的超时（ASH_EMBED_QUERY_TIMEOUT_MS，默认 8 秒），超时就只用文本比对，ash suggest 不会被慢模型拖住 - 回归测试：慢速假向量服务 + 很短的超时，旧代码下补不上向量 Co-Authored-By: Claude Opus 5.5 --- server/embeddings.js | 19 ++++++++++++------ server/tests/semantic-agents.test.js | 30 ++++++++++++++++++++++++---- 2 files changed, 39 insertions(+), 10 deletions(-) diff --git a/server/embeddings.js b/server/embeddings.js index f89dc52..dbac0f8 100644 --- a/server/embeddings.js +++ b/server/embeddings.js @@ -5,6 +5,10 @@ // ASH_EMBED_URL Ollama 地址（如 http://127.0.0.1:11434），或 OpenAI 兼容接口（以 /v1 结尾） // ASH_EMBED_MODEL 模型名，默认 bge-m3（多语言；Ollama 上 ollama pull bge-m3） // ASH_EMBED_KEY OpenAI 兼容接口的 API key（可选） +// ASH_EMBED_QUERY_TIMEOUT_MS 查询向量（ash suggest）最多等多久，默认 8000；超时就只用文本比对 +// +// 模型可能跑在繁忙服务器的 CPU 上（实测 seed-SER9 上 bge-m3 每条约 3.5 秒）：后台补齐按小批次进行、 +// 超时随批次大小放宽，新技能优先；查询向量单独设较短的超时，找技能不会被慢模型拖住。 // // 语义分数只在文本比对看不见的地方补位（见 skillIndex）：同语言的同系列技能语义上很像却不是重复， // 所以相似度只对主语言不同的两个技能用语义；按任务找技能时取文本、语义两者较高的相关度。 @@ -17,9 +21,11 @@ const config = () => ({ model: process.env.ASH_EMBED_MODEL || 'bge-m3', key: process.env.ASH_EMBED_KEY || '', timeout: Number(process.env.ASH_EMBED_TIMEOUT_MS) || 15000, + queryTimeout: Number(process.env.ASH_EMBED_QUERY_TIMEOUT_MS) || 8000, }); const enabled = () => Boolean(config().url); -const BATCH = 16; +const BATCH = 4; +const PER_TEXT_MS = 30000; // 后台补齐时每条文本最多给这么久（慢 CPU 上一条要好几秒） let lastError = null; let lastErrorLogged = 0; @@ -48,8 +54,8 @@ function cosine(a, b) { return dot; } -async function embed(texts) { - const { url, model, key, timeout } = config(); +async function embed(texts, timeout = config().timeout) { + const { url, model, key } = config(); const openai = /\/v1$/.test(url); const res = await fetch(openai ? `${url}/embeddings` : `${url}/api/embed`, { method: 'POST', @@ -90,7 +96,8 @@ function ensureVectors(userId, docs) { // 留下一个永远不清除的已完成标记，后来的调用会无限地「等它结束再查一遍」 await null; const have = vectorsFor(docs); - const missing = docs.filter((d) => !have.has(d.skill.id)); + // 新技能优先：刚推送的技能要先有向量，才能参与查重 + const missing = docs.filter((d) => !have.has(d.skill.id)).sort((a, b) => b.skill.id - a.skill.id); const upsert = db.prepare(` INSERT INTO skill_embeddings (skill_id, text_hash, vector, updated_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(skill_id) DO UPDATE SET text_hash = excluded.text_hash, vector = excluded.vector, updated_at = CURRENT_TIMESTAMP @@ -98,7 +105,7 @@ function ensureVectors(userId, docs) { try { for (let i = 0; i < missing.length; i += BATCH) { const batch = missing.slice(i, i + BATCH); - const vectors = await embed(batch.map((d) => d.embedText)); + const vectors = await embed(batch.map((d) => d.embedText), Math.max(config().timeout, PER_TEXT_MS * batch.length)); db.transaction(() => batch.forEach((d, j) => { // 算向量期间技能可能被删了：外键约束会拒绝写入，跳过即可 try { upsert.run(d.skill.id, textHash(d.embedText), toBlob(vectors[j])); } catch { /* 已删除 */ } @@ -134,7 +141,7 @@ async function embedQuery(text) { const key = textHash(text); if (queryCache.has(key)) return queryCache.get(key); try { - const [vec] = await embed([text]); + const [vec] = await embed([text], config().queryTimeout); queryCache.set(key, vec); if (queryCache.size > 200) queryCache.delete(queryCache.keys().next().value); return vec; diff --git a/server/tests/semantic-agents.test.js b/server/tests/semantic-agents.test.js index 5b829b1..93d10c7 100644 --- a/server/tests/semantic-agents.test.js +++ b/server/tests/semantic-agents.test.js @@ -79,18 +79,18 @@ const server = require('http').createServer((req, res) => { if (req.url === '/calls') return res.end(String(calls)); let body = ''; req.on('data', (c) => { body += c; }); - req.on('end', () => { + req.on('end', () => setTimeout(() => { calls += 1; const { input } = JSON.parse(body || '{}'); const vectors = (Array.isArray(input) ? input : [input]).map((t) => [...CONCEPTS.map((re) => (String(t).match(re) || []).length), 0.3]); res.setHeader('content-type', 'application/json'); res.end(JSON.stringify({ embeddings: vectors })); - }); + }, Number(process.env.DELAY_MS || 0))); }); server.listen(0, '127.0.0.1', () => console.log(server.address().port)); `; -async function startFakeEmbeddings() { - const child = spawn(process.execPath, ['-e', FAKE_EMBEDDINGS], { stdio: ['ignore', 'pipe', 'inherit'] }); +async function startFakeEmbeddings({ delayMs = 0 } = {}) { + const child = spawn(process.execPath, ['-e', FAKE_EMBEDDINGS], { stdio: ['ignore', 'pipe', 'inherit'], env: { ...process.env, DELAY_MS: String(delayMs) } }); const port = await new Promise((resolve) => child.stdout.once('data', (d) => resolve(Number(String(d).trim())))); const url = `http://127.0.0.1:${port}`; return { @@ -217,3 +217,25 @@ test('semantic layer, pins, multi-agent installs and ash doctor end to end', asy assert.doesNotMatch(out, /pin-old|历史版本/); }); }); + +// 线上实测：繁忙服务器的 CPU 上 bge-m3 每条要好几秒。后台补齐不能因为「整批超时」永远补不上， +// 找技能也不能被慢模型拖住 +test('a slow embedding model still gets every vector, and never slows down suggest', async (t) => { + const fake = await startFakeEmbeddings({ delayMs: 800 }); + const server = await startServer({ ASH_EMBED_URL: fake.url, ASH_EMBED_MODEL: 'fake', ASH_EMBED_TIMEOUT_MS: '300', ASH_EMBED_QUERY_TIMEOUT_MS: '300' }); + t.after(async () => { await server.stop(); await fake.close(); }); + const detail = await (await fetch(`${server.origin}/api/skills/systematic-debugging`)).json(); + assert.ok(detail.id); + const Database = require('better-sqlite3'); + const count = () => { + const db = new Database(path.join(server.env.DATA_DIR, 'another-skillhub.db'), { readonly: true }); + try { return db.prepare('SELECT COUNT(*) AS n FROM skill_embeddings').get().n; } finally { db.close(); } + }; + for (let i = 0; i < 40 && count() < 3; i += 1) await new Promise((r) => setTimeout(r, 250)); + assert.equal(count(), 3, 'batches get a per-text allowance instead of the short request timeout'); + const started = Date.now(); + const res = await fetch(`${server.origin}/api/agent/suggest?q=${encodeURIComponent('测试失败找根因')}&format=text`); + assert.equal(res.status, 200); + assert.match(await res.text(), /systematic-debugging/, 'falls back to text matching'); + assert.ok(Date.now() - started < 700, `suggest does not wait for the slow model (${Date.now() - started}ms)`); +});