john
00a00a1f69
- Free stale Memind listeners on 8081 before startup and exit cleanly on EADDRINUSE - Backfill owned sessions missing from Goose via h5_conversation_messages summaries - Strip Memind task orchestration prefixes from user-facing chat text - Repair missing public docx links before release MindSpace link checks Co-authored-by: Cursor <cursoragent@cursor.com>
489 lines
16 KiB
JavaScript
489 lines
16 KiB
JavaScript
import crypto from 'node:crypto';
|
|
import { Agent, fetch as undiciFetch } from 'undici';
|
|
import { jsonrepair } from 'jsonrepair';
|
|
import {
|
|
decryptSecret,
|
|
normalizeApiUrl,
|
|
resolveChatCompletionsUrl,
|
|
} from './llm-providers.mjs';
|
|
import { deriveUserFacingText } from './conversation-display.mjs';
|
|
|
|
const httpsDispatcher = new Agent({
|
|
connect: { rejectUnauthorized: false },
|
|
});
|
|
|
|
const MAX_MESSAGE_TEXT = 12000;
|
|
const MAX_ANALYSIS_MESSAGES = 8;
|
|
const MAX_MEMORY_TEXT = 1000;
|
|
const LABELS = new Set([
|
|
'preference',
|
|
'habit',
|
|
'interest',
|
|
'goal',
|
|
'fact',
|
|
'experience',
|
|
'knowledge',
|
|
]);
|
|
|
|
function resolveEncryptionKey(explicitKey) {
|
|
return (
|
|
explicitKey ??
|
|
process.env.H5_SETTINGS_ENCRYPTION_KEY ??
|
|
process.env.TKMIND_SERVER__SECRET_KEY ??
|
|
'local-dev-secret'
|
|
);
|
|
}
|
|
|
|
function stableId(...parts) {
|
|
return crypto
|
|
.createHash('sha256')
|
|
.update(parts.map((part) => String(part ?? '')).join('\0'))
|
|
.digest('hex');
|
|
}
|
|
|
|
function parseJsonObject(text) {
|
|
const source = String(text ?? '').trim();
|
|
if (!source) return null;
|
|
try {
|
|
return JSON.parse(source);
|
|
} catch {
|
|
try {
|
|
return JSON.parse(jsonrepair(source));
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
}
|
|
|
|
function extractJsonObject(text) {
|
|
const direct = parseJsonObject(text);
|
|
if (direct && typeof direct === 'object') return direct;
|
|
const source = String(text ?? '').trim();
|
|
const start = source.indexOf('{');
|
|
const end = source.lastIndexOf('}');
|
|
if (start < 0 || end <= start) return null;
|
|
return parseJsonObject(source.slice(start, end + 1));
|
|
}
|
|
|
|
export function extractConversationMessageText(message) {
|
|
const content = Array.isArray(message?.content) ? message.content : [];
|
|
const chunks = [];
|
|
for (const item of content) {
|
|
if (item?.type === 'text' && typeof item.text === 'string') chunks.push(item.text);
|
|
if (item?.type === 'image' && item.url) chunks.push(`[image] ${item.url}`);
|
|
}
|
|
const fromContent = chunks.join('\n').trim();
|
|
if (fromContent) return fromContent.slice(0, MAX_MESSAGE_TEXT);
|
|
const displayText = message?.metadata?.displayText;
|
|
return typeof displayText === 'string' ? displayText.trim().slice(0, MAX_MESSAGE_TEXT) : '';
|
|
}
|
|
|
|
function normalizeMessage(sessionId, userId, message, index, now) {
|
|
const role = (String(message?.role ?? '').trim() || 'unknown').toLowerCase().slice(0, 32);
|
|
const text = extractConversationMessageText(message);
|
|
if (!text) return null;
|
|
const rawMessageId = String(message?.id ?? message?.message_id ?? '').trim();
|
|
const messageKey = (rawMessageId || `idx:${index}:${stableId(role, text).slice(0, 16)}`).slice(0, 128);
|
|
return {
|
|
id: stableId(userId, sessionId, messageKey),
|
|
userId,
|
|
sessionId,
|
|
messageKey,
|
|
sequenceNo: index,
|
|
role,
|
|
text,
|
|
rawJson: JSON.stringify(message),
|
|
createdAt: Number(message?.created_at ?? message?.createdAt ?? now) || now,
|
|
updatedAt: now,
|
|
};
|
|
}
|
|
|
|
function buildMemoryPrompt(messages) {
|
|
const transcript = messages
|
|
.map((message, index) => {
|
|
const text = String(message.text ?? '').replace(/\s+/g, ' ').trim().slice(0, 1500);
|
|
return `${index + 1}. [${message.role}] ${text}`;
|
|
})
|
|
.join('\n');
|
|
return [
|
|
'你是 TKMind 的用户长期记忆提取器。',
|
|
'请从下面的用户对话中提取可以长期复用的个人记忆。',
|
|
'只记录用户明确表达或强证据支持的信息，不要猜测，不要记录临时闲聊。',
|
|
'不要记录密码、密钥、身份证、手机号、银行卡等敏感信息。',
|
|
'只返回 JSON 对象，不要 Markdown，不要解释。',
|
|
'格式：{"memories":[{"label":"preference|habit|interest|goal|fact|experience|knowledge","text":"一句完整中文记忆","confidence":0.0-1.0}]}',
|
|
'最多返回 5 条。没有值得记录的内容时返回 {"memories":[]}',
|
|
'',
|
|
transcript,
|
|
].join('\n');
|
|
}
|
|
|
|
function normalizeMemoryItem(item, fallbackLabel = 'fact') {
|
|
const text = String(item?.text ?? item?.memory ?? item?.content ?? '').trim();
|
|
if (!text) return null;
|
|
const labelRaw = String(item?.label ?? item?.category ?? fallbackLabel).trim().toLowerCase();
|
|
const label = LABELS.has(labelRaw) ? labelRaw : fallbackLabel;
|
|
const confidence = Math.max(0, Math.min(1, Number(item?.confidence ?? 0.6)));
|
|
return {
|
|
label,
|
|
text: text.slice(0, MAX_MEMORY_TEXT),
|
|
confidence: Number.isFinite(confidence) ? confidence : 0.6,
|
|
};
|
|
}
|
|
|
|
function fallbackMemoriesFromMessages(messages) {
|
|
const memories = [];
|
|
const patterns = [
|
|
{ label: 'preference', re: /(?:我喜欢|我偏好|我更喜欢|以后.*(?:用|叫|按)|不要再|别再)(.+)/ },
|
|
{ label: 'interest', re: /(?:我对|我关注|我感兴趣|我想了解)(.+)/ },
|
|
{ label: 'goal', re: /(?:我的目标是|我想要|我希望|我打算|我计划|我想去|打算去|准备去)(.+)/ },
|
|
{ label: 'habit', re: /(?:我通常|我习惯|我一般)(.+)/ },
|
|
{ label: 'fact', re: /(?:我是|我叫|我来自|我在)(.{2,40})/ },
|
|
{ label: 'experience', re: /(?:我们|我).{0,8}(?:去|到|在).{2,40}(?:玩|旅游|旅行|出差|度假)/ },
|
|
];
|
|
for (const message of messages) {
|
|
const text = String(message.text ?? '').replace(/\s+/g, ' ').trim();
|
|
for (const { label, re } of patterns) {
|
|
const match = text.match(re);
|
|
if (!match?.[0]) continue;
|
|
memories.push({
|
|
label,
|
|
text: `用户${match[0].replace(/[。！？\s]+$/g, '')}`,
|
|
confidence: 0.45,
|
|
});
|
|
if (memories.length >= 3) return memories;
|
|
}
|
|
}
|
|
return memories;
|
|
}
|
|
|
|
function resolveMemoryExtractionModelConfig(env = process.env) {
|
|
const providerKeyId = String(
|
|
env?.USER_CONVERSATION_MEMORY_MODEL_PROVIDER_KEY_ID
|
|
?? env?.MEMORY_EXTRACTION_MODEL_PROVIDER_KEY_ID
|
|
?? env?.MEMIND_CHAT_ROUTER_MODEL_PROVIDER_KEY_ID
|
|
?? '',
|
|
).trim() || null;
|
|
const model = String(
|
|
env?.USER_CONVERSATION_MEMORY_MODEL
|
|
?? env?.MEMORY_EXTRACTION_MODEL
|
|
?? env?.MEMIND_CHAT_ROUTER_MODEL
|
|
?? '',
|
|
).trim() || null;
|
|
return { providerKeyId, model };
|
|
}
|
|
|
|
function parseExtractionMemories(rawText) {
|
|
const parsed = extractJsonObject(rawText);
|
|
const rawItems = Array.isArray(parsed?.memories) ? parsed.memories : [];
|
|
return rawItems.map((item) => normalizeMemoryItem(item)).filter(Boolean);
|
|
}
|
|
|
|
export function createConversationMemoryService(pool, options = {}) {
|
|
const now = options.now ?? (() => Date.now());
|
|
const fetchImpl = options.fetch ?? undiciFetch;
|
|
const llmProviderService = options.llmProviderService ?? null;
|
|
const resolveEffectiveEnv = options.getEffectiveEnv ?? (async () => process.env);
|
|
const encryptionKey = options.encryptionKey;
|
|
const llmWarningIntervalMs = Math.max(
|
|
0,
|
|
Number(options.llmWarningIntervalMs ?? process.env.USER_CONVERSATION_MEMORY_LLM_WARN_INTERVAL_MS ?? 60000) || 0,
|
|
);
|
|
let lastLlmExtractionWarningAt = 0;
|
|
|
|
function warnLlmExtractionFailed(err) {
|
|
const ts = now();
|
|
if (
|
|
lastLlmExtractionWarningAt > 0 &&
|
|
llmWarningIntervalMs > 0 &&
|
|
ts - lastLlmExtractionWarningAt < llmWarningIntervalMs
|
|
) return;
|
|
lastLlmExtractionWarningAt = ts;
|
|
console.warn('[conversation-memory] LLM extraction failed:', err instanceof Error ? err.message : err);
|
|
}
|
|
|
|
function isEnabled() {
|
|
return process.env.USER_CONVERSATION_MEMORY_ENABLED !== '0';
|
|
}
|
|
|
|
function llmEnabled() {
|
|
return process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED !== '0';
|
|
}
|
|
|
|
async function saveConversationMessages(sessionId, userId, messages = []) {
|
|
if (!isEnabled() || !pool || !sessionId || !userId || !Array.isArray(messages)) return [];
|
|
const ts = now();
|
|
const normalized = messages
|
|
.map((message, index) => normalizeMessage(sessionId, userId, message, index, ts))
|
|
.filter(Boolean);
|
|
if (!normalized.length) return [];
|
|
|
|
await pool.query(
|
|
`INSERT INTO h5_conversation_messages
|
|
(id, user_id, agent_session_id, message_key, sequence_no, role, text, raw_json,
|
|
created_at, updated_at)
|
|
VALUES ?
|
|
ON DUPLICATE KEY UPDATE
|
|
sequence_no = VALUES(sequence_no),
|
|
role = VALUES(role),
|
|
text = VALUES(text),
|
|
raw_json = VALUES(raw_json),
|
|
updated_at = VALUES(updated_at)`,
|
|
[
|
|
normalized.map((message) => [
|
|
message.id,
|
|
message.userId,
|
|
message.sessionId,
|
|
message.messageKey,
|
|
message.sequenceNo,
|
|
message.role,
|
|
message.text,
|
|
message.rawJson,
|
|
message.createdAt,
|
|
message.updatedAt,
|
|
]),
|
|
],
|
|
);
|
|
return normalized;
|
|
}
|
|
|
|
async function loadUnanalyzedUserMessages(userId, limit = MAX_ANALYSIS_MESSAGES) {
|
|
const [rows] = await pool.query(
|
|
`SELECT id, user_id, agent_session_id, message_key, sequence_no, role, text, created_at
|
|
FROM h5_conversation_messages
|
|
WHERE user_id = ? AND role = 'user' AND analyzed_at IS NULL
|
|
ORDER BY created_at ASC, sequence_no ASC
|
|
LIMIT ?`,
|
|
[userId, Math.max(1, limit)],
|
|
);
|
|
return rows;
|
|
}
|
|
|
|
async function markAnalyzed(messageIds) {
|
|
if (!messageIds.length) return;
|
|
await pool.query(
|
|
`UPDATE h5_conversation_messages SET analyzed_at = ? WHERE id IN (?)`,
|
|
[now(), messageIds],
|
|
);
|
|
}
|
|
|
|
async function selectedProvider() {
|
|
const [rows] = await pool.query(
|
|
`SELECT * FROM h5_llm_provider_keys WHERE is_selected = 1 AND status = 'active' LIMIT 1`,
|
|
);
|
|
return rows[0] ?? null;
|
|
}
|
|
|
|
async function extractWithLlm(messages) {
|
|
const prompt = buildMemoryPrompt(messages);
|
|
const effectiveEnv = await resolveEffectiveEnv();
|
|
const { providerKeyId, model } = resolveMemoryExtractionModelConfig(effectiveEnv);
|
|
|
|
if (llmProviderService?.createChatCompletion) {
|
|
const completion = await llmProviderService.createChatCompletion({
|
|
providerKeyId: providerKeyId || undefined,
|
|
model: model || undefined,
|
|
temperature: 0,
|
|
messages: [{ role: 'user', content: prompt }],
|
|
});
|
|
if (!completion?.ok) return null;
|
|
const memories = parseExtractionMemories(completion.reply);
|
|
return memories.length ? memories : [];
|
|
}
|
|
|
|
const row = await selectedProvider();
|
|
if (!row) return null;
|
|
const apiUrl = normalizeApiUrl(row.api_url);
|
|
const url = resolveChatCompletionsUrl(apiUrl);
|
|
if (!url) return null;
|
|
let apiKey;
|
|
try {
|
|
apiKey = decryptSecret(
|
|
{
|
|
ciphertext: row.api_key_ciphertext,
|
|
iv: row.api_key_iv,
|
|
tag: row.api_key_tag,
|
|
},
|
|
resolveEncryptionKey(encryptionKey),
|
|
);
|
|
} catch {
|
|
return null;
|
|
}
|
|
|
|
const upstream = await fetchImpl(url, {
|
|
method: 'POST',
|
|
headers: {
|
|
'Content-Type': 'application/json',
|
|
Authorization: `Bearer ${apiKey}`,
|
|
},
|
|
body: JSON.stringify({
|
|
model: model || row.default_model,
|
|
messages: [{ role: 'user', content: prompt }],
|
|
stream: false,
|
|
temperature: 0,
|
|
...(row.relay_provider ? { provider: row.relay_provider } : {}),
|
|
}),
|
|
dispatcher: url.startsWith('https://') ? httpsDispatcher : undefined,
|
|
});
|
|
const text = await upstream.text().catch(() => '');
|
|
if (!upstream.ok) return null;
|
|
const payload = extractJsonObject(text);
|
|
const content = payload?.choices?.[0]?.message?.content ?? payload?.content ?? text;
|
|
return parseExtractionMemories(content);
|
|
}
|
|
|
|
async function storeMemories(userId, sourceMessageIds, memories, rawJson = null) {
|
|
const rows = [];
|
|
const ts = now();
|
|
const sourceSessionId = sourceMessageIds[0]?.agent_session_id ?? null;
|
|
const evidenceMessageId = sourceMessageIds[0]?.id ?? null;
|
|
for (const item of memories) {
|
|
const normalized = normalizeMemoryItem(item);
|
|
if (!normalized) continue;
|
|
const hash = stableId(userId, normalized.label, normalized.text);
|
|
rows.push([
|
|
stableId('memory', hash),
|
|
userId,
|
|
normalized.label,
|
|
hash,
|
|
normalized.text,
|
|
evidenceMessageId,
|
|
sourceSessionId,
|
|
normalized.confidence,
|
|
rawJson ? JSON.stringify(rawJson) : null,
|
|
ts,
|
|
ts,
|
|
]);
|
|
}
|
|
if (!rows.length) return 0;
|
|
await pool.query(
|
|
`INSERT INTO h5_user_memory_items
|
|
(id, user_id, label, memory_hash, memory_text, evidence_message_id,
|
|
source_session_id, confidence, raw_json, created_at, updated_at)
|
|
VALUES ?
|
|
ON DUPLICATE KEY UPDATE
|
|
confidence = GREATEST(confidence, VALUES(confidence)),
|
|
updated_at = VALUES(updated_at)`,
|
|
[rows],
|
|
);
|
|
return rows.length;
|
|
}
|
|
|
|
async function analyzeUser(userId) {
|
|
if (!isEnabled() || !pool || !userId) return { ok: false, reason: 'disabled' };
|
|
const messages = await loadUnanalyzedUserMessages(userId);
|
|
if (!messages.length) return { ok: true, analyzed: 0, memories: 0 };
|
|
let memories = null;
|
|
if (llmEnabled()) {
|
|
try {
|
|
memories = await extractWithLlm(messages);
|
|
} catch (err) {
|
|
warnLlmExtractionFailed(err);
|
|
}
|
|
}
|
|
if (!memories) memories = fallbackMemoriesFromMessages(messages);
|
|
const stored = await storeMemories(userId, messages, memories);
|
|
// Always mark the batch processed after one pass so transient LLM failures
|
|
// do not leave messages permanently stuck in the analyze queue.
|
|
await markAnalyzed(messages.map((message) => message.id));
|
|
return { ok: true, analyzed: messages.length, memories: stored };
|
|
}
|
|
|
|
async function saveAndAnalyze(sessionId, userId, messages = []) {
|
|
const saved = await saveConversationMessages(sessionId, userId, messages);
|
|
if (!saved.length) return { saved: 0, analyzed: 0, memories: 0 };
|
|
const result = await analyzeUser(userId);
|
|
return {
|
|
saved: saved.length,
|
|
analyzed: result.analyzed ?? 0,
|
|
memories: result.memories ?? 0,
|
|
};
|
|
}
|
|
|
|
async function listMemories(userId, { limit = 50 } = {}) {
|
|
const [rows] = await pool.query(
|
|
`SELECT id, label, memory_text, confidence, source_session_id, created_at, updated_at
|
|
FROM h5_user_memory_items
|
|
WHERE user_id = ? AND status = 'active'
|
|
ORDER BY updated_at DESC
|
|
LIMIT ?`,
|
|
[userId, Math.max(1, Math.min(200, Number(limit) || 50))],
|
|
);
|
|
return rows.map((row) => ({
|
|
id: row.id,
|
|
label: row.label,
|
|
text: row.memory_text,
|
|
confidence: Number(row.confidence ?? 0),
|
|
sourceSessionId: row.source_session_id ?? null,
|
|
createdAt: Number(row.created_at),
|
|
updatedAt: Number(row.updated_at),
|
|
}));
|
|
}
|
|
|
|
async function loadSessionListSummaries(userId, sessionIds = []) {
|
|
if (!isEnabled() || !pool || !userId || !Array.isArray(sessionIds) || sessionIds.length === 0) {
|
|
return new Map();
|
|
}
|
|
const ids = [...new Set(sessionIds.filter(Boolean))];
|
|
if (!ids.length) return new Map();
|
|
const placeholders = ids.map(() => '?').join(', ');
|
|
const [rows] = await pool.query(
|
|
`SELECT agent_session_id, role, text, raw_json, sequence_no, created_at, updated_at
|
|
FROM h5_conversation_messages
|
|
WHERE user_id = ? AND agent_session_id IN (${placeholders})
|
|
ORDER BY sequence_no ASC, created_at ASC`,
|
|
[userId, ...ids],
|
|
);
|
|
const buckets = new Map();
|
|
for (const row of rows) {
|
|
const sessionId = String(row.agent_session_id ?? '').trim();
|
|
if (!sessionId) continue;
|
|
if (!buckets.has(sessionId)) {
|
|
buckets.set(sessionId, { messages: [], maxUpdatedAt: 0 });
|
|
}
|
|
const bucket = buckets.get(sessionId);
|
|
bucket.messages.push(row);
|
|
bucket.maxUpdatedAt = Math.max(
|
|
bucket.maxUpdatedAt,
|
|
Number(row.updated_at ?? row.created_at ?? 0),
|
|
);
|
|
}
|
|
const summaries = new Map();
|
|
for (const sessionId of ids) {
|
|
const bucket = buckets.get(sessionId);
|
|
if (!bucket || bucket.messages.length === 0) continue;
|
|
const firstUser = bucket.messages.find((row) => String(row.role ?? '') === 'user');
|
|
let title = '';
|
|
if (firstUser) {
|
|
let text = String(firstUser.text ?? '').trim();
|
|
if (firstUser.raw_json) {
|
|
try {
|
|
const parsed = JSON.parse(firstUser.raw_json);
|
|
text = extractConversationMessageText(parsed) || text;
|
|
} catch {
|
|
// keep text column fallback
|
|
}
|
|
}
|
|
title = deriveUserFacingText(text).slice(0, 80);
|
|
}
|
|
summaries.set(sessionId, {
|
|
title: title || sessionId,
|
|
messageCount: bucket.messages.length,
|
|
updatedAt: bucket.maxUpdatedAt
|
|
? new Date(bucket.maxUpdatedAt).toISOString()
|
|
: undefined,
|
|
});
|
|
}
|
|
return summaries;
|
|
}
|
|
|
|
return {
|
|
isEnabled,
|
|
saveConversationMessages,
|
|
analyzeUser,
|
|
saveAndAnalyze,
|
|
listMemories,
|
|
loadSessionListSummaries,
|
|
};
|
|
}
|