fix: harden conversation memory extraction flow
This commit is contained in:
@@ -1,5 +1,5 @@
|
|||||||
import crypto from 'node:crypto';
|
import crypto from 'node:crypto';
|
||||||
import { Agent, fetch as undiciFetch } from 'undici';
|
import { fetch as undiciFetch } from 'undici';
|
||||||
import { jsonrepair } from 'jsonrepair';
|
import { jsonrepair } from 'jsonrepair';
|
||||||
import {
|
import {
|
||||||
decryptSecret,
|
decryptSecret,
|
||||||
@@ -7,10 +7,6 @@ import {
|
|||||||
resolveChatCompletionsUrl,
|
resolveChatCompletionsUrl,
|
||||||
} from './llm-providers.mjs';
|
} from './llm-providers.mjs';
|
||||||
|
|
||||||
const insecureDispatcher = new Agent({
|
|
||||||
connect: { rejectUnauthorized: false },
|
|
||||||
});
|
|
||||||
|
|
||||||
const MAX_MESSAGE_TEXT = 12000;
|
const MAX_MESSAGE_TEXT = 12000;
|
||||||
const MAX_ANALYSIS_MESSAGES = 8;
|
const MAX_ANALYSIS_MESSAGES = 8;
|
||||||
const MAX_MEMORY_TEXT = 1000;
|
const MAX_MEMORY_TEXT = 1000;
|
||||||
@@ -264,7 +260,6 @@ export function createConversationMemoryService(pool, options = {}) {
|
|||||||
temperature: 0,
|
temperature: 0,
|
||||||
...(row.relay_provider ? { provider: row.relay_provider } : {}),
|
...(row.relay_provider ? { provider: row.relay_provider } : {}),
|
||||||
}),
|
}),
|
||||||
dispatcher: url.startsWith('https://') ? insecureDispatcher : undefined,
|
|
||||||
});
|
});
|
||||||
const text = await upstream.text().catch(() => '');
|
const text = await upstream.text().catch(() => '');
|
||||||
if (!upstream.ok) return null;
|
if (!upstream.ok) return null;
|
||||||
@@ -317,16 +312,20 @@ export function createConversationMemoryService(pool, options = {}) {
|
|||||||
const messages = await loadUnanalyzedUserMessages(userId);
|
const messages = await loadUnanalyzedUserMessages(userId);
|
||||||
if (!messages.length) return { ok: true, analyzed: 0, memories: 0 };
|
if (!messages.length) return { ok: true, analyzed: 0, memories: 0 };
|
||||||
let memories = null;
|
let memories = null;
|
||||||
|
let extractionSucceeded = false;
|
||||||
if (llmEnabled()) {
|
if (llmEnabled()) {
|
||||||
try {
|
try {
|
||||||
memories = await extractWithLlm(messages);
|
memories = await extractWithLlm(messages);
|
||||||
|
extractionSucceeded = Array.isArray(memories);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
console.warn('[conversation-memory] LLM extraction failed:', err instanceof Error ? err.message : err);
|
console.warn('[conversation-memory] LLM extraction failed:', err instanceof Error ? err.message : err);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if (!memories) memories = fallbackMemoriesFromMessages(messages);
|
if (!memories) memories = fallbackMemoriesFromMessages(messages);
|
||||||
const stored = await storeMemories(userId, messages, memories);
|
const stored = await storeMemories(userId, messages, memories);
|
||||||
await markAnalyzed(messages.map((message) => message.id));
|
if (!llmEnabled() || extractionSucceeded || stored > 0) {
|
||||||
|
await markAnalyzed(messages.map((message) => message.id));
|
||||||
|
}
|
||||||
return { ok: true, analyzed: messages.length, memories: stored };
|
return { ok: true, analyzed: messages.length, memories: stored };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -150,3 +150,54 @@ test('saveAndAnalyze stores messages and fallback memories', async () => {
|
|||||||
if (previous == null) delete process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED;
|
if (previous == null) delete process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED;
|
||||||
else process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED = previous;
|
else process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED = previous;
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('saveAndAnalyze leaves messages unanalyzed when extraction fails and no memory is stored', async () => {
|
||||||
|
const previous = process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED;
|
||||||
|
process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED = '1';
|
||||||
|
const pool = createPool();
|
||||||
|
const service = createConversationMemoryService(pool, {
|
||||||
|
now: () => 2000,
|
||||||
|
fetch: async () => {
|
||||||
|
throw new Error('upstream unavailable');
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
const result = await service.saveAndAnalyze('session-2', 'user-2', [
|
||||||
|
{
|
||||||
|
id: 'm3',
|
||||||
|
role: 'user',
|
||||||
|
content: [{ type: 'text', text: '今天下雨了。' }],
|
||||||
|
metadata: { userVisible: true },
|
||||||
|
},
|
||||||
|
]);
|
||||||
|
|
||||||
|
assert.equal(result.saved, 1);
|
||||||
|
assert.equal(result.memories, 0);
|
||||||
|
assert.equal(pool.state.messages.find((item) => item.message_key === 'm3')?.analyzed_at, null);
|
||||||
|
|
||||||
|
if (previous == null) delete process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED;
|
||||||
|
else process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED = previous;
|
||||||
|
});
|
||||||
|
|
||||||
|
test('saveAndAnalyze still marks messages analyzed when llm extraction is disabled and fallback stores nothing', async () => {
|
||||||
|
const previous = process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED;
|
||||||
|
process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED = '0';
|
||||||
|
const pool = createPool();
|
||||||
|
const service = createConversationMemoryService(pool, { now: () => 3000 });
|
||||||
|
|
||||||
|
const result = await service.saveAndAnalyze('session-3', 'user-3', [
|
||||||
|
{
|
||||||
|
id: 'm4',
|
||||||
|
role: 'user',
|
||||||
|
content: [{ type: 'text', text: '今天天气不错。' }],
|
||||||
|
metadata: { userVisible: true },
|
||||||
|
},
|
||||||
|
]);
|
||||||
|
|
||||||
|
assert.equal(result.saved, 1);
|
||||||
|
assert.equal(result.memories, 0);
|
||||||
|
assert.equal(pool.state.messages.find((item) => item.message_key === 'm4')?.analyzed_at, 3000);
|
||||||
|
|
||||||
|
if (previous == null) delete process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED;
|
||||||
|
else process.env.USER_CONVERSATION_MEMORY_LLM_ENABLED = previous;
|
||||||
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user