refactor: Business logic and dependencies updates
- 核心服务代码更新 (db, server, auth, proxy) - Agent 相关模块更新 (mindspace, experience) - 前端组件和 hooks 更新 - 数据库 schema 更新 - 依赖版本更新
This commit is contained in:
john
@@ -0,0 +1,227 @@
|
||||
import crypto from 'node:crypto';
|
||||
|
||||
/**
|
||||
* PostgreSQL + pgvector backed experience store (etat C, polyglot mode).
|
||||
*
|
||||
* Same contract as the MySQL `createExperienceService` (record / search /
|
||||
* reflect) so callers (mindspace-agent-runner) are backend-agnostic. The MySQL
|
||||
* business DB is untouched; only this shared-experience workload lives on PG.
|
||||
*
|
||||
* Semantic search via pgvector is used when an `embed(text) => number[]` function
|
||||
* is supplied; otherwise it degrades to keyword (ILIKE) + recency, identical in
|
||||
* spirit to the MySQL version, so it is useful before an embedding provider is wired.
|
||||
*
|
||||
* The `pg` driver is imported dynamically so this module loads even when the
|
||||
* dependency is not yet installed; it is only required once the service is built.
|
||||
*
|
||||
* @param {object} options
|
||||
* @param {string} options.connectionString - e.g. process.env.EXPERIENCE_PG_URL
|
||||
* @param {(text: string) => Promise<number[]>} [options.embed] - embedding fn
|
||||
* @param {number} [options.embeddingDim=1536] - vector dimension (must match embed)
|
||||
* @param {() => number} [options.now]
|
||||
* @param {number} [options.recencyHalfLifeMs]
|
||||
*/
|
||||
export async function createPgExperienceService(options = {}) {
|
||||
const connectionString = options.connectionString;
|
||||
if (!connectionString) {
|
||||
throw new Error('createPgExperienceService 需要 connectionString（建议用 EXPERIENCE_PG_URL）');
|
||||
}
|
||||
const now = options.now ?? (() => Date.now());
|
||||
const embed = options.embed ?? null;
|
||||
const embeddingDim = Math.max(1, Number(options.embeddingDim ?? 1536));
|
||||
const halfLifeMs = Math.max(
|
||||
60_000,
|
||||
Number(options.recencyHalfLifeMs ?? 30 * 24 * 60 * 60 * 1000),
|
||||
);
|
||||
|
||||
const { default: pg } = await import('pg');
|
||||
const pool = new pg.Pool({ connectionString, max: Number(options.poolMax ?? 10) });
|
||||
|
||||
// Idempotent setup. pgvector must be available on the server (CREATE EXTENSION).
|
||||
await pool.query('CREATE EXTENSION IF NOT EXISTS vector');
|
||||
await pool.query(`
|
||||
CREATE TABLE IF NOT EXISTS h5_experience (
|
||||
id UUID PRIMARY KEY,
|
||||
scope TEXT NOT NULL DEFAULT 'global',
|
||||
kind TEXT NOT NULL DEFAULT 'lesson',
|
||||
title TEXT NOT NULL,
|
||||
body TEXT NOT NULL,
|
||||
tags JSONB NOT NULL DEFAULT '[]'::jsonb,
|
||||
source_session_id TEXT,
|
||||
source_user_id TEXT,
|
||||
use_count BIGINT NOT NULL DEFAULT 0,
|
||||
embedding vector(${embeddingDim}),
|
||||
created_at BIGINT NOT NULL,
|
||||
updated_at BIGINT NOT NULL
|
||||
)
|
||||
`);
|
||||
await pool.query(
|
||||
'CREATE INDEX IF NOT EXISTS idx_h5_experience_scope_updated ON h5_experience (scope, updated_at DESC)',
|
||||
);
|
||||
// Trigram-ish keyword fallback index (btree on lower(title)) is cheap; the
|
||||
// ANN vector index is created lazily once rows have embeddings.
|
||||
await pool.query(
|
||||
"CREATE INDEX IF NOT EXISTS idx_h5_experience_embedding ON h5_experience USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100)",
|
||||
).catch(() => {
|
||||
// ivfflat needs the extension + may warn on empty table; non-fatal.
|
||||
});
|
||||
|
||||
function normalizeTags(tags) {
|
||||
if (!Array.isArray(tags)) return [];
|
||||
return [...new Set(tags.map((t) => String(t).trim()).filter(Boolean))];
|
||||
}
|
||||
|
||||
function toVectorLiteral(vec) {
|
||||
// pgvector accepts a string like '[0.1,0.2,...]'
|
||||
return `[${vec.map((n) => Number(n)).join(',')}]`;
|
||||
}
|
||||
|
||||
function rowToExperience(row) {
|
||||
return {
|
||||
id: row.id,
|
||||
scope: row.scope,
|
||||
kind: row.kind,
|
||||
title: row.title,
|
||||
body: row.body,
|
||||
tags: Array.isArray(row.tags) ? row.tags : [],
|
||||
sourceSessionId: row.source_session_id ?? null,
|
||||
sourceUserId: row.source_user_id ?? null,
|
||||
useCount: Number(row.use_count ?? 0),
|
||||
createdAt: Number(row.created_at),
|
||||
updatedAt: Number(row.updated_at),
|
||||
};
|
||||
}
|
||||
|
||||
async function record(input) {
|
||||
const title = String(input?.title ?? '').trim();
|
||||
const body = String(input?.body ?? '').trim();
|
||||
if (!title) throw experienceError('经验标题不能为空', 'invalid_experience_input');
|
||||
if (!body) throw experienceError('经验内容不能为空', 'invalid_experience_input');
|
||||
const id = crypto.randomUUID();
|
||||
const ts = now();
|
||||
const scope = String(input?.scope ?? 'global').trim() || 'global';
|
||||
const kind = String(input?.kind ?? 'lesson').trim() || 'lesson';
|
||||
const tags = normalizeTags(input?.tags);
|
||||
let embedding = null;
|
||||
if (embed) {
|
||||
try {
|
||||
const vec = await embed(`${title}\n${body}`);
|
||||
if (Array.isArray(vec) && vec.length === embeddingDim) embedding = toVectorLiteral(vec);
|
||||
} catch {
|
||||
embedding = null; // embedding failure must not block recording
|
||||
}
|
||||
}
|
||||
await pool.query(
|
||||
`INSERT INTO h5_experience
|
||||
(id, scope, kind, title, body, tags, source_session_id, source_user_id,
|
||||
use_count, embedding, created_at, updated_at)
|
||||
VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7,$8,0,$9::vector,$10,$11)`,
|
||||
[
|
||||
id,
|
||||
scope,
|
||||
kind,
|
||||
title,
|
||||
body,
|
||||
JSON.stringify(tags),
|
||||
input?.sourceSessionId ?? null,
|
||||
input?.sourceUserId ?? null,
|
||||
embedding,
|
||||
ts,
|
||||
ts,
|
||||
],
|
||||
);
|
||||
return rowToExperience({
|
||||
id,
|
||||
scope,
|
||||
kind,
|
||||
title,
|
||||
body,
|
||||
tags,
|
||||
source_session_id: input?.sourceSessionId ?? null,
|
||||
source_user_id: input?.sourceUserId ?? null,
|
||||
use_count: 0,
|
||||
created_at: ts,
|
||||
updated_at: ts,
|
||||
});
|
||||
}
|
||||
|
||||
async function search(query, { scope = 'global', limit = 5 } = {}) {
|
||||
const text = String(query ?? '').trim();
|
||||
if (!text) return [];
|
||||
const max = Math.max(1, limit);
|
||||
|
||||
// Vector path: cosine distance ordering, recency as tiebreaker.
|
||||
if (embed) {
|
||||
try {
|
||||
const vec = await embed(text);
|
||||
if (Array.isArray(vec) && vec.length === embeddingDim) {
|
||||
const { rows } = await pool.query(
|
||||
`SELECT id, scope, kind, title, body, tags, source_session_id,
|
||||
source_user_id, use_count, created_at, updated_at
|
||||
FROM h5_experience
|
||||
WHERE scope = $1 AND embedding IS NOT NULL
|
||||
ORDER BY embedding <=> $2::vector, updated_at DESC
|
||||
LIMIT $3`,
|
||||
[scope, toVectorLiteral(vec), max],
|
||||
);
|
||||
if (rows.length > 0) {
|
||||
await bumpUseCount(rows.map((r) => r.id));
|
||||
return rows.map(rowToExperience);
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// fall through to keyword path
|
||||
}
|
||||
}
|
||||
|
||||
// Keyword fallback: ILIKE any term, ordered by recency.
|
||||
const terms = tokenize(text);
|
||||
if (terms.length === 0) return [];
|
||||
const likeClauses = terms.map((_, i) => `(title ILIKE $${i + 2} OR body ILIKE $${i + 2})`);
|
||||
const params = [scope, ...terms.map((t) => `%${t}%`), max];
|
||||
const { rows } = await pool.query(
|
||||
`SELECT id, scope, kind, title, body, tags, source_session_id,
|
||||
source_user_id, use_count, created_at, updated_at
|
||||
FROM h5_experience
|
||||
WHERE scope = $1 AND (${likeClauses.join(' OR ')})
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT $${terms.length + 2}`,
|
||||
params,
|
||||
);
|
||||
if (rows.length > 0) await bumpUseCount(rows.map((r) => r.id));
|
||||
return rows.map(rowToExperience);
|
||||
}
|
||||
|
||||
async function bumpUseCount(ids) {
|
||||
if (!ids.length) return;
|
||||
await pool
|
||||
.query('UPDATE h5_experience SET use_count = use_count + 1 WHERE id = ANY($1::uuid[])', [ids])
|
||||
.catch(() => {});
|
||||
}
|
||||
|
||||
function tokenize(query) {
|
||||
return [
|
||||
...new Set(
|
||||
String(query ?? '')
|
||||
.toLowerCase()
|
||||
.split(/[^\p{L}\p{N}]+/u)
|
||||
.map((t) => t.trim())
|
||||
.filter((t) => t.length >= 2),
|
||||
),
|
||||
];
|
||||
}
|
||||
|
||||
async function reflect() {
|
||||
return { ok: false, reason: 'not_implemented' };
|
||||
}
|
||||
|
||||
async function close() {
|
||||
await pool.end();
|
||||
}
|
||||
|
||||
return { record, search, reflect, close };
|
||||
}
|
||||
|
||||
function experienceError(message, code) {
|
||||
return Object.assign(new Error(message), { code });
|
||||
}
|
||||
Reference in New Issue
Block a user