From 3805a0c8ae0c11e9e19f6d511b5034d0fd960c72 Mon Sep 17 00:00:00 2001 From: Cat-bl <32174@qq.com> Date: Thu, 6 Aug 2026 12:15:54 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AE=B0=E5=BF=86=E7=B3=BB=E7=BB=9F=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 15 +- apps/MessageManager.js | 7 +- apps/chat.js | 30 +- apps/memory.js | 9 +- config_default/message.yaml | 2 +- core/sharedState.js | 3 + models/Guoba/schemas/memory.js | 2 +- tests/memory-helpers.test.js | 28 ++ tests/memory-manager.test.js | 528 ++++++++++++++++++++++ utils/MemoryManager.js | 760 ++++++++++++++++++++++++++------ utils/memory/MemoryExtractor.js | 332 +++++++++++--- utils/memory/MemoryRetriever.js | 26 +- utils/memory/MemoryStore.js | 119 ++++- utils/memory/constants.js | 10 +- utils/memory/helpers.js | 55 ++- 15 files changed, 1650 insertions(+), 276 deletions(-) create mode 100644 tests/memory-manager.test.js diff --git a/README.md b/README.md index 98af513..2abe56a 100644 --- a/README.md +++ b/README.md @@ -386,10 +386,12 @@ MCP 管理命令： | `maxFactsPerGroup` | int | `50` | **每群最大全局记忆条数**（所有类别总计） | | `importanceThreshold` | float | `0.5` | **重要性阈值**：低于此值的事实不会保存 | | `memoryDecayDays` | int | `7` | 记忆召回时参考的时效天数 | +| `userExtractDebounceSeconds` | int | `45` | 同一用户停止发言后等待多久再批量提取，`0` 为立即处理 | +| `userExtractMaxBatchMessages` | int | `6` | 同一用户累计多少条消息后立即提取 | | `groupExtractMinIntervalMinutes` | int | `10` | 群记忆最小整理间隔 | | `groupExtractMaxBatchMessages` | int | `12` | 群累计多少条消息后立即整理 | -| `promptMaxUserFacts` | int | `8` | 注入 prompt 的用户记忆最大条数 | -| `promptMaxGroupFacts` | int | `6` | 注入 prompt 的群记忆最大条数 | +| `promptMaxUserFacts` | int | `8` | 注入 prompt 的用户记忆最大条数，`0` 为不注入 | +| `promptMaxGroupFacts` | int | `6` | 注入 prompt 的群记忆最大条数，`0` 为不注入 | | `promptMaxChars` | int | `1200` | 记忆 prompt 总字符上限 | | `semanticRecallEnabled` | boolean | `false` | 是否启用 embedding 语义召回，默认关闭 | | `semanticRecallTopK` | int | `20` | 语义召回时先筛选多少条候选记忆，只有开启语义召回时才会用到 | @@ -399,8 +401,9 @@ MCP 管理命令： **使用说明**： - 记忆整理是后台异步执行的，不会阻塞正常聊天。 -- 用户记忆会在一次对话回复完成后立即异步提取；群记忆会按上面的时间间隔和累计条数批量整理。 -- 工具调用结果、机器人自己的回复、系统提示不会被保存进记忆。 +- 白名单群里的真实文本消息会进入后台缓冲：用户记忆按用户批量提取，群记忆按上面的时间间隔和累计条数批量整理，不要求机器人必须回复。 +- 工具调用结果、机器人自己的回复、系统提示和明显的 `#`/`/` 机器人命令不会被保存进记忆。 +- 记忆 AI 请求有固定并发上限和超时；临时失败的批次会保留并在退避后重试。 - 旧版记忆会自动兼容迁移，不需要手动清空。 **用户记忆分类**：提取的记忆会自动归类到以下类别： @@ -424,7 +427,7 @@ MCP 管理命令： - `#我的记忆`：查看自己的用户记忆，结果会以合并转发返回，避免刷屏。 - `#群记忆`：查看本群群记忆，结果会以合并转发返回，避免刷屏。 - `#搜索记忆 <关键词>`：搜索自己的用户记忆和本群群记忆，结果会以合并转发返回，避免刷屏。 -- `#删除记忆 `：ID 可以从 `#我的记忆`、`#群记忆` 或 `#搜索记忆 <关键词>` 的结果里看到，例如 `ID:1a2b3c4d`，删除时发送 `#删除记忆 1a2b3c4d`。普通用户只能删除自己的记忆；群主、管理员、主人可删除群记忆。 +- `#删除记忆 `：ID 可以从 `#我的记忆`、`#群记忆` 或 `#搜索记忆 <关键词>` 的结果里看到，例如 `ID:1a2b3c4d`，删除时至少发送前 8 位；如果前缀不唯一，需要输入更完整的 ID。普通用户只能删除自己的记忆；群主、管理员、主人可删除群记忆。 - `#清空我的记忆` / `#禁用我的记忆` / `#启用我的记忆`：管理自己的用户记忆。 - `#清空群记忆`：仅群主、管理员、主人可用。 @@ -1007,7 +1010,7 @@ memoryAiModel: "claude-3-5-haiku-20241022" memoryAiApikey: "sk-ant-xxxxx" ``` -> **说明**：用户记忆会在对话结束后立即异步调用此模型；群记忆会批量整理后再调用此模型，提取值得长期保存的事实。推荐使用 `gpt-4o-mini`、`gemini-2.0-flash`、`claude-3-5-haiku` 等小模型。插件会根据 URL 自动识别 API 格式。 +> **说明**：白名单群中的真实文本会先进入后台缓冲，用户记忆按用户防抖批量调用此模型，群记忆按群批量整理。推荐使用 `gpt-4o-mini`、`gemini-2.0-flash`、`claude-3-5-haiku` 等小模型。插件会根据 URL 自动识别 API 格式。 ### Embedding 模型配置 (`embeddingAiConfig`) diff --git a/apps/MessageManager.js b/apps/MessageManager.js index 51b708f..1f37c7a 100644 --- a/apps/MessageManager.js +++ b/apps/MessageManager.js @@ -1,5 +1,6 @@ import { MessageManager } from '../utils/MessageManager.js' import { emojiPackManager } from '../utils/EmojiPackManager.js' +import { getSharedState } from '../core/sharedState.js' import fs from 'fs'; import YAML from 'yaml'; export class MessageRecordPlugin extends plugin { @@ -43,6 +44,10 @@ export class MessageRecordPlugin extends plugin { async onMessage(e) { await this.messageManager.recordMessage(e); emojiPackManager.maybeAutoCollect(e).catch(() => {}); + const memoryManager = getSharedState()?.memoryManager; + memoryManager?.enqueueGroupEvent(e).catch(error => { + logger.error(`[MemoryManager] 全局消息入队失败: ${error.stack || error}`); + }); return false; } @@ -209,4 +214,4 @@ export class MessageRecordPlugin extends plugin { return false } } -} \ No newline at end of file +} diff --git a/apps/chat.js b/apps/chat.js index e0a70f5..e30bcba 100644 --- a/apps/chat.js +++ b/apps/chat.js @@ -810,7 +810,7 @@ export class ChatPlugin extends plugin { session.groupUserMessages = this.trimMessageHistory(messages) await this.saveGroupUserMessages(e.group_id, e.user_id, messages) - // 更新情感、记忆、表达学习（异步，不阻塞） + // 更新情感、关系分、表达学习（异步，不阻塞） // 使用 e.msg 纯消息内容，而不是格式化的 userContent this.updateEnhancedSystems(e, e.msg || '', output).catch(err => { logger.error('[增强系统] 更新失败:', err) @@ -818,7 +818,7 @@ export class ChatPlugin extends plugin { } /** - * 异步更新情感系统、长期记忆 + * 异步更新情感系统和关系分；用户/群记忆由全局消息记录器处理 */ async updateEnhancedSystems(e, userMessage, botReply) { const { group_id: groupId, user_id: userId } = e @@ -830,16 +830,9 @@ export class ChatPlugin extends plugin { emotionState = await this.emotionManager.updateEmotionFromMessage(groupId, userMessage, isAtBot) } - // 2. 提取并保存长期记忆（后台异步） + // 2. 更新关系分。用户记忆和群记忆都由全局消息记录器统一入队， + // 覆盖机器人未回复的真实群消息，并避免在回复路径重复抽取。 if (this.config.memorySystem?.enabled) { - // 不 await，让它在后台执行 - this.memoryManager.extractAndSaveMemories(groupId, userId, userMessage, botReply, { - source: "user", - messageId: e.message_id, - senderName: e.sender?.card || e.sender?.nickname - }).catch(err => { - logger.error('[MemoryManager] 用户记忆提取异常:', err) - }) const latestEmotionEvent = emotionState?.recentEvents?.[0] if (latestEmotionEvent && Number.isFinite(latestEmotionEvent.delta)) { const relationDelta = Math.max(-0.03, Math.min(0.03, latestEmotionEvent.delta * 0.2)) @@ -849,21 +842,6 @@ export class ChatPlugin extends plugin { }) } } - // 提取群全局记忆（传入聊天记录） - if (groupId) { - const history = await this.messageManager.getMessages('group', groupId) - const chatHistory = (history || []).slice(0, 40).map(msg => ({ - role: msg.sender?.user_id === Bot.uin ? 'assistant' : 'user', - source: msg.source || (msg.sender?.user_id === Bot.uin ? "send" : "user"), - userId: msg.sender?.user_id, - user_id: msg.sender?.user_id, - senderName: msg.sender?.nickname || msg.sender?.card || '群成员', - content: msg.content - })) - this.memoryManager.extractAndSaveGroupMemories(groupId, chatHistory).catch(err => { - logger.error('[MemoryManager] 群记忆提取异常:', err) - }) - } } // 表达学习已移至 handleRandomReply 静默收集，不在此处调用 diff --git a/apps/memory.js b/apps/memory.js index 22cd7a9..c6d1fdf 100644 --- a/apps/memory.js +++ b/apps/memory.js @@ -205,7 +205,14 @@ export class MemoryCommands extends plugin { }) } - await e.reply(result.deleted ? `已删除记忆 ${id}` : "没有找到可删除的记忆，普通用户只能删除自己的记忆") + const failureMessages = { + "id-too-short": "记忆 ID 至少需要输入前 8 位", + "ambiguous-id": "这个 ID 前缀匹配到多条记忆，请输入更完整的 ID", + "not-found": "没有找到可删除的记忆，普通用户只能删除自己的记忆" + } + await e.reply(result.deleted + ? `已删除记忆 ${id}` + : failureMessages[result.reason] || "没有找到可删除的记忆，普通用户只能删除自己的记忆") } catch (error) { logger.error("[记忆管理] 删除记忆失败:", error) await e.reply("删除记忆失败，请看日志") diff --git a/config_default/message.yaml b/config_default/message.yaml index 2ee2696..7810460 100644 --- a/config_default/message.yaml +++ b/config_default/message.yaml @@ -223,7 +223,7 @@ pluginSettings: maxFactsPerGroup: 50 importanceThreshold: 0.5 memoryDecayDays: 7 - userExtractDebounceSeconds: 90 + userExtractDebounceSeconds: 45 userExtractMaxBatchMessages: 6 groupExtractMinIntervalMinutes: 10 groupExtractMaxBatchMessages: 12 diff --git a/core/sharedState.js b/core/sharedState.js index 56d2935..3d81300 100644 --- a/core/sharedState.js +++ b/core/sharedState.js @@ -43,6 +43,9 @@ export function buildMemoryConfig(config) { ...memorySystem, memoryAiConfig: config.memoryAiConfig || null, embeddingAiConfig: config.embeddingAiConfig || null, + pluginEnabled: config.enabled !== false, + enableGroupWhitelist: Boolean(config.enableGroupWhitelist), + allowedGroups: Array.isArray(config.allowedGroups) ? config.allowedGroups : [], groupExtractMinIntervalMinutes: memorySystem.groupExtractMinIntervalMinutes ?? memorySystem.groupExtractMinInterval ?? 10 } diff --git a/models/Guoba/schemas/memory.js b/models/Guoba/schemas/memory.js index 869fc10..640caff 100644 --- a/models/Guoba/schemas/memory.js +++ b/models/Guoba/schemas/memory.js @@ -42,7 +42,7 @@ export default [ label: "用户记忆提取防抖（秒）", component: "InputNumber", bottomHelpMessage: "用户对话结束后，等待 N 秒再触发记忆提取，避免短时间重复调用", - componentProps: { min: 0, max: 600, placeholder: "90" } + componentProps: { min: 0, max: 600, placeholder: "45" } }, { field: "memorySystem.userExtractMaxBatchMessages", diff --git a/tests/memory-helpers.test.js b/tests/memory-helpers.test.js index 0a5ece0..938d1dd 100644 --- a/tests/memory-helpers.test.js +++ b/tests/memory-helpers.test.js @@ -8,10 +8,12 @@ import { normalizeText, compactText, containsToolFeedback, + isLikelyBotCommand, isRealUserSource, charJaccard, isSimilarContent, extractJsonArray, + parseJsonArrayResult, keywordSet, cosineSimilarity, normalizeConfig @@ -58,6 +60,12 @@ test("containsToolFeedback：识别工具痕迹标记", () => { assert.equal(containsToolFeedback("今天天气不错"), false) }) +test("isLikelyBotCommand：过滤常见机器人命令前缀", () => { + assert.equal(isLikelyBotCommand("#表情包列表"), true) + assert.equal(isLikelyBotCommand("/help"), true) + assert.equal(isLikelyBotCommand("今天聊聊 #AI"), false) +}) + test("isRealUserSource：仅用户来源为真", () => { assert.equal(isRealUserSource(undefined), true) assert.equal(isRealUserSource(""), true) @@ -86,6 +94,15 @@ test("extractJsonArray：容忍围栏与解释文字", () => { assert.deepEqual(extractJsonArray("完全不是 JSON"), []) }) +test("parseJsonArrayResult：区分空结果、非法格式并兼容包装对象", () => { + assert.deepEqual(parseJsonArrayResult("[]"), { items: [], status: "empty" }) + assert.deepEqual(parseJsonArrayResult("完全不是 JSON"), { items: [], status: "invalid" }) + assert.deepEqual(parseJsonArrayResult('{"operations":[{"operation":"upsert"}]}'), { + items: [{ operation: "upsert" }], + status: "ok" + }) +}) + test("keywordSet：分词并生成中文 2-gram", () => { const set = keywordSet("hello 世界真好") assert.equal(set.has("hello"), true) @@ -109,6 +126,17 @@ test("normalizeConfig：默认值合并与数值兜底", () => { const clamped = normalizeConfig({ importanceThreshold: 5, maxFactsPerUser: -3 }) assert.equal(clamped.importanceThreshold, 1) assert.equal(clamped.maxFactsPerUser, 1) + + const zeroValues = normalizeConfig({ + userExtractDebounceSeconds: 0, + promptMaxUserFacts: 0, + promptMaxGroupFacts: 0, + minFactsPerCategory: 0 + }) + assert.equal(zeroValues.userExtractDebounceSeconds, 0) + assert.equal(zeroValues.promptMaxUserFacts, 0) + assert.equal(zeroValues.promptMaxGroupFacts, 0) + assert.equal(zeroValues.minFactsPerCategory, 0) }) test("normalizeConfig：兼容旧字段 groupExtractMinInterval（毫秒/分钟自适应）", () => { diff --git a/tests/memory-manager.test.js b/tests/memory-manager.test.js new file mode 100644 index 0000000..ca64371 --- /dev/null +++ b/tests/memory-manager.test.js @@ -0,0 +1,528 @@ +import test from "node:test" +import assert from "node:assert/strict" +import { MemoryManager } from "../utils/MemoryManager.js" +import { MemoryExtractor } from "../utils/memory/MemoryExtractor.js" +import { MemoryRetriever } from "../utils/memory/MemoryRetriever.js" +import { MemoryStore } from "../utils/memory/MemoryStore.js" +import { normalizeConfig } from "../utils/memory/helpers.js" + +function createManager() { + return new MemoryManager({ + enabled: true, + importanceThreshold: 0.5, + memoryAiConfig: { + memoryAiUrl: "https://example.com/v1/chat/completions", + memoryAiApikey: "test" + } + }) +} + +function candidateOperation(messageId, content = "经常熬夜") { + return { + operation: "upsert", + decision: "candidate", + content, + category: "habits", + importance: 0.4, + confidence: 0.6, + sourceMessageIds: [messageId], + sourceUserIds: ["200"] + } +} + +test("stageUserOperations：候选跨两个独立批次后晋升", async () => { + const manager = createManager() + let candidates = [] + manager.store = { + getUserMeta: async () => ({ disabled: false }), + getUserCandidates: async () => candidates, + saveUserCandidates: async (groupId, userId, next) => { + candidates = next + return next + } + } + + const first = await manager.stageUserOperations("100", "200", [candidateOperation("m1")]) + assert.equal(first.operations.length, 0) + assert.equal(first.candidateAdded, 1) + assert.equal(candidates.length, 0) + await manager.finalizeStagedCandidates("100", "200", first, { resolvedCandidateIds: [] }) + assert.equal(candidates.length, 1) + + const secondOperation = candidateOperation("m2", "长期晚睡") + secondOperation.candidateId = candidates[0].id + const second = await manager.stageUserOperations("100", "200", [secondOperation]) + assert.equal(second.candidatePromoted, 1) + assert.equal(second.operations.length, 1) + assert.equal(second.operations[0].decision, "save") + assert.equal(second.operations[0].content, "长期晚睡") + assert.equal(second.operations[0].importance, 0.6) + assert.deepEqual(second.operations[0].sourceMessageIds, ["m1", "m2"]) + assert.equal(candidates.length, 1) + assert.equal(candidates[0].evidenceKeys.length, 1) + await manager.finalizeStagedCandidates("100", "200", second, { + resolvedCandidateIds: [second.operations[0].candidateId] + }) + assert.equal(candidates.length, 0) +}) + +test("stageUserOperations：同一批证据不重复累计，明确事实清理相似候选", async () => { + const manager = createManager() + let candidates = [] + manager.store = { + getUserMeta: async () => ({ disabled: false }), + getUserCandidates: async () => candidates, + saveUserCandidates: async (groupId, userId, next) => { + candidates = next + return next + } + } + + const first = await manager.stageUserOperations("100", "200", [candidateOperation("m1")]) + await manager.finalizeStagedCandidates("100", "200", first, { resolvedCandidateIds: [] }) + const duplicate = await manager.stageUserOperations("100", "200", [candidateOperation("m1")]) + assert.equal(duplicate.candidateDuplicate, 1) + assert.equal(candidates[0].evidenceKeys.length, 1) + + const direct = await manager.stageUserOperations("100", "200", [{ + ...candidateOperation("m2"), + decision: "save", + importance: 0.8 + }]) + assert.equal(direct.operations.length, 1) + assert.equal(candidates.length, 1) + await manager.finalizeStagedCandidates("100", "200", direct, { + resolvedCandidateIds: [direct.operations[0].candidateId] + }) + assert.equal(candidates.length, 0) +}) + +test("enqueueGroupEvent：只接收真实群文本并保留消息 ID", async () => { + const manager = createManager() + manager.enqueueInteraction = async event => event + manager.extractAndSaveGroupMemories = async (groupId, messages) => ({ groupId, messages }) + + const self = await manager.enqueueGroupEvent({ group_id: 100, user_id: 999, self_id: 999 }) + assert.equal(self.reason, "self-message") + + const image = await manager.enqueueGroupEvent({ + group_id: 100, + user_id: 200, + self_id: 999, + message: [{ type: "image", data: { file: "a.jpg" } }] + }) + assert.equal(image.reason, "no-text") + + manager.updateConfig({ enableGroupWhitelist: true, allowedGroups: [101] }) + const denied = await manager.enqueueGroupEvent({ + group_id: 100, + user_id: 200, + self_id: 999, + message: [{ type: "text", data: { text: "我是程序员" } }] + }) + assert.equal(denied.reason, "group-not-allowed") + manager.updateConfig({ enableGroupWhitelist: false }) + + const text = await manager.enqueueGroupEvent({ + group_id: 100, + user_id: 200, + self_id: 999, + message_id: "m1", + time: 1700000000, + sender: { nickname: "测试用户" }, + message: [ + { type: "at", data: { qq: "999" } }, + { type: "text", data: { text: "我是程序员" } } + ] + }) + assert.equal(text.content, "我是程序员") + assert.equal(text.messageId, "m1") + assert.equal(text.senderName, "测试用户") + assert.equal(text.groupMemory.groupId, 100) + assert.equal(text.groupMemory.messages[0].messageId, "m1") +}) + +test("enqueueInteraction：同一用户的相同消息只入队一次", async () => { + const manager = createManager() + let calls = 0 + manager.extractAndSaveMemories = async () => { + calls++ + return { queued: true } + } + + const event = { groupId: "100", userId: "200", content: "我喜欢打游戏", source: "user", messageId: "m1" } + await manager.enqueueInteraction(event) + const duplicate = await manager.enqueueInteraction(event) + assert.equal(calls, 1) + assert.equal(duplicate.reason, "duplicate") +}) + +test("enqueueInteraction：记忆 AI 不可用时不占用去重记录或缓冲区", async () => { + const manager = new MemoryManager({ enabled: true, memoryAiConfig: null }) + const result = await manager.enqueueInteraction({ + groupId: "100", + userId: "200", + content: "我是程序员", + source: "user", + messageId: "m1" + }) + + assert.equal(result.reason, "ai-unavailable") + assert.equal(manager.userSeenMessages.size, 0) + assert.equal(manager.userBuffers.size, 0) +}) + +test("用户禁用记忆后不再注入事实或关系熟悉度", async () => { + const manager = createManager() + manager.retriever.retrieve = async () => ({ + meta: { disabled: true, relationshipScore: 0.95 }, + facts: [{ content: "不应注入", category: "identity" }] + }) + + assert.equal(await manager.getMemoryPromptForUser("100", "200", "测试"), "") +}) + +test("adminClearMemories：等待同一用户的在途抽取完成后再清空", async () => { + const manager = createManager() + let releaseExtraction + let markStarted + const started = new Promise(resolve => { markStarted = resolve }) + const blocker = new Promise(resolve => { releaseExtraction = resolve }) + let cleared = false + manager.store = { + clearScope: async () => { + cleared = true + return 3 + } + } + + const extraction = manager.enqueueUserTask("100", "200", async () => { + markStarted() + await blocker + }) + await started + + const clearing = manager.adminClearMemories({ scope: "user", groupId: "100", userId: "200" }) + await Promise.resolve() + assert.equal(cleared, false) + + releaseExtraction() + await extraction + const result = await clearing + assert.equal(cleared, true) + assert.equal(result.cleared, 3) +}) + +test("MemoryExtractor：保留 candidate 决策并对非法 JSON 修复一次", async t => { + const previousLogger = globalThis.logger + globalThis.logger = { warn() {} } + t.after(() => { + globalThis.logger = previousLogger + }) + + const extractor = new MemoryExtractor(normalizeConfig({}), {}) + const operations = extractor.normalizeOperations([{ + operation: "upsert", + decision: "candidate", + content: "经常熬夜", + category: "habits", + importance: 0.4 + }], "user", { sourceMessageIds: ["m1"], sourceUserIds: ["200"] }) + assert.equal(operations[0].decision, "candidate") + + const responses = ["这不是 JSON", "[]"] + extractor.callChat = async () => responses.shift() + const parsed = await extractor.parseOperationResponse([], 100) + assert.equal(parsed.parseStatus, "repaired_empty") + assert.equal(parsed.repaired, true) + assert.equal(responses.length, 0) +}) + +test("MemoryStore：候选使用短期 Redis 存储并可清除", async t => { + const previousRedis = globalThis.redis + const values = new Map() + const expirations = new Map() + globalThis.redis = { + async set(key, value) { values.set(key, value) }, + async get(key) { return values.get(key) || null }, + async del(key) { values.delete(key) }, + async expire(key, seconds) { expirations.set(key, seconds) } + } + t.after(() => { + globalThis.redis = previousRedis + }) + + const store = new MemoryStore(normalizeConfig({})) + await store.saveUserCandidates("100", "200", [{ + content: "经常熬夜", + category: "habits", + evidenceKeys: ["batch-1"], + sourceMessageIds: ["m1"] + }]) + const candidates = await store.getUserCandidates("100", "200") + assert.equal(candidates.length, 1) + assert.equal(candidates[0].content, "经常熬夜") + assert.ok(expirations.get(store.userCandidateKey("100", "200")) > 0) + + await store.clearUserCandidates("100", "200") + assert.equal((await store.getUserCandidates("100", "200")).length, 0) +}) + +test("MemoryStore：旧版无 ID 事实使用稳定 ID，避免并发迁移产生孤立键", () => { + const store = new MemoryStore(normalizeConfig({})) + const first = store.factFromLegacy("喜欢火锅", "user", "100", "200", "likes") + const second = store.factFromLegacy("喜欢火锅", "user", "100", "200", "likes") + assert.equal(first.id, second.id) +}) + +test("applyOperations：未知 update ID 不新增事实，upsert 不采用模型提供的陌生 ID", async () => { + const manager = createManager() + let savedFact = null + manager.store = { + getMeta: async () => ({ scope: "user", groupId: "100", userId: "200", disabled: false }), + getFacts: async () => [], + normalizeCategory: (scope, category) => category, + saveFact: async fact => { savedFact = fact; return fact } + } + manager.extractor.createEmbedding = async () => ({ embedding: null, embeddingHash: null }) + + const update = await manager.applyOperations("user", "100", "200", [{ + operation: "update", + id: "hallucinated-id", + content: "喜欢火锅", + category: "likes", + importance: 0.8 + }]) + assert.equal(update.saved, 0) + assert.equal(update.invalid, 1) + + const upsert = await manager.applyOperations("user", "100", "200", [{ + operation: "upsert", + id: "model-controlled-id", + content: "喜欢火锅", + category: "likes", + importance: 0.8 + }]) + assert.equal(upsert.saved, 1) + assert.notEqual(savedFact.id, "model-controlled-id") +}) + +test("群记忆缓冲：倒序批量输入按时间正序分批且不丢消息", async () => { + const manager = new MemoryManager({ + enabled: true, + groupExtractMaxBatchMessages: 2, + groupExtractMinIntervalMinutes: 10, + memoryAiConfig: { memoryAiUrl: "https://example.com", memoryAiApikey: "test" } + }) + const batches = [] + manager.extractAndSaveGroupMemoriesNow = async (groupId, messages) => { + batches.push(messages.map(message => message.messageId)) + return { saved: 0, deleted: 0, skipped: 0 } + } + + const messages = [5, 4, 3, 2, 1].map(index => ({ + userId: String(100 + index), + content: `消息${index}`, + messageId: `m${index}`, + createdAt: index * 1000, + source: "user" + })) + await manager.extractAndSaveGroupMemories("100", messages) + await manager.flushGroupBuffer("100") + + assert.deepEqual(batches, [["m1", "m2"], ["m3", "m4"], ["m5"]]) + assert.equal(manager.groupBuffers.size, 0) +}) + +test("群记忆缓冲：调度元数据读取失败后仍保留消息并安排重试", async () => { + const manager = createManager() + manager.store.getGroupMeta = async () => { throw new Error("redis unavailable") } + + await assert.rejects(manager.extractAndSaveGroupMemories("100", [{ + userId: "200", + content: "这是普通聊天", + messageId: "m1", + createdAt: 1, + source: "user" + }]), /redis unavailable/) + + const buffer = manager.groupBuffers.get("100") + assert.deepEqual(buffer.messages.map(message => message.messageId), ["m1"]) + assert.ok(buffer.timer) + manager.discardGroupBuffer("100") +}) + +test("群记忆执行：并发期间的第二批仍遵守群最小整理间隔", async () => { + const manager = createManager() + const lastAttemptAt = Date.now() - 1_000 + manager.store = { + getGroupMeta: async () => ({ + scope: "group", + groupId: "100", + factIds: [], + disabled: false, + lastAttemptAt, + nextRetryAt: 0 + }) + } + + const result = await manager.extractAndSaveGroupMemoriesNow("100", [{ + userId: "200", + content: "这是第二批消息", + messageId: "m2", + createdAt: 2, + source: "user" + }]) + + assert.equal(result.reason, "interval") + assert.ok(result.retryAt > Date.now()) +}) + +test("用户记忆缓冲：退避结果会恢复原批次而不是丢弃", async () => { + const manager = createManager() + manager.extractAndSaveMemoriesNow = async () => ({ + saved: 0, + deleted: 0, + skipped: 1, + reason: "backoff", + retryAt: Date.now() + 60_000 + }) + const key = manager.getUserBufferKey("100", "200") + manager.userBuffers.set(key, { + groupId: "100", + userId: "200", + messages: [{ userId: "200", content: "我是程序员", messageId: "m1", createdAt: 1 }], + firstBufferedAt: 1, + timer: null + }) + + await manager.flushUserBuffer(key) + assert.deepEqual(manager.userBuffers.get(key).messages.map(message => message.messageId), ["m1"]) + manager.discardUserBuffer("100", "200") +}) + +test("热关闭记忆：清空缓冲并取消正在进行的请求", () => { + const manager = createManager() + let aborted = 0 + manager.extractor.abortActiveRequests = () => { aborted++ } + manager.userBuffers.set("100:200", { timer: setTimeout(() => {}, 60_000), messages: [{}] }) + manager.groupBuffers.set("100", { groupId: "100", timer: setTimeout(() => {}, 60_000), messages: [{}] }) + + manager.updateConfig({ enabled: false }) + assert.equal(manager.userBuffers.size, 0) + assert.equal(manager.groupBuffers.size, 0) + assert.equal(aborted, 1) +}) + +test("MemoryExtractor：缺失 decision 保守进入候选，并按 sourceIndexes 绑定证据", () => { + const extractor = new MemoryExtractor(normalizeConfig({}), {}) + const operations = extractor.normalizeOperations([{ + operation: "upsert", + content: "喜欢火锅", + category: "likes", + importance: 0.8, + sourceIndexes: [2] + }, { + operation: "unknown", + content: "不应保存", + category: "identity" + }], "user", { + messages: [ + { messageId: "m1", userId: "200" }, + { messageId: "m2", userId: "200" } + ] + }) + + assert.equal(operations.length, 1) + assert.equal(operations[0].decision, "candidate") + assert.deepEqual(operations[0].sourceMessageIds, ["m2"]) +}) + +test("MemoryExtractor：切换 embedding 服务后旧向量不会被误认为当前向量", () => { + const config = normalizeConfig({ + semanticRecallEnabled: true, + embeddingAiConfig: { + embeddingApiUrl: "https://one.example/embeddings", + embeddingApiKey: "test", + embeddingApiModel: "same-model" + } + }) + const extractor = new MemoryExtractor(config, {}) + const first = extractor.embeddingHashFor("喜欢火锅") + config.embeddingAiConfig.embeddingApiUrl = "https://two.example/embeddings" + assert.notEqual(first, extractor.embeddingHashFor("喜欢火锅")) +}) + +test("MemoryRetriever：召回纯读、使用 semanticRecallTopK 并忽略旧模型向量", async () => { + const facts = [ + { id: "a", content: "当前相关", importance: 0.6, confidence: 0.8, updatedAt: Date.now(), embedding: [1, 0], embeddingHash: "hash:当前相关" }, + { id: "b", content: "次要相关", importance: 1, confidence: 1, updatedAt: Date.now(), embedding: [0.8, 0.2], embeddingHash: "hash:次要相关" }, + { id: "c", content: "旧模型向量", importance: 1, confidence: 1, updatedAt: Date.now(), embedding: [1, 0], embeddingHash: "old-hash" } + ].map(fact => ({ ...fact, scope: "user", groupId: "100", userId: "200", category: "identity" })) + const store = { + getMeta: async () => ({ disabled: false }), + getFacts: async () => facts, + setJson: async () => { throw new Error("召回不应写 Redis") } + } + const extractor = { + canUseEmbedding: () => true, + createEmbedding: async () => ({ embedding: [1, 0] }), + embeddingHashFor: text => `hash:${text}` + } + const retriever = new MemoryRetriever(normalizeConfig({ + semanticRecallEnabled: true, + semanticRecallTopK: 1 + }), store, extractor) + + const result = await retriever.retrieve({ groupId: "100", userId: "200", query: "相关", limit: 3 }) + assert.deepEqual(result.facts.map(fact => fact.id), ["a"]) +}) + +test("MemoryStore：显式删除可审计，清空会删除活动、删除和孤立事实", async t => { + const previousRedis = globalThis.redis + const values = new Map() + globalThis.redis = { + async set(key, value) { values.set(key, value) }, + async get(key) { return values.get(key) || null }, + async del(key) { values.delete(key) }, + async *scanIterator({ MATCH }) { + const prefix = MATCH.replace(/\*$/, "") + for (const key of [...values.keys()]) { + if (key.startsWith(prefix)) yield key + } + } + } + t.after(() => { globalThis.redis = previousRedis }) + + const store = new MemoryStore(normalizeConfig({ maxFactsPerUser: 10 })) + await store.saveFact({ + id: "fact-0001", + scope: "user", + groupId: "100", + userId: "200", + content: "喜欢火锅", + category: "likes" + }) + const meta = await store.getUserMeta("100", "200") + await store.deleteFact(meta, "fact-0001") + const deletedMeta = await store.getUserMeta("100", "200") + assert.deepEqual(deletedMeta.factIds, []) + assert.deepEqual(deletedMeta.deletedFactIds, ["fact-0001"]) + assert.equal((await store.getFacts(deletedMeta, true))[0].status, "deleted") + + const orphanKey = store.factKey("user", store.userScopeId("100", "200"), "orphan") + values.set(orphanKey, JSON.stringify({ id: "orphan" })) + await store.clearScope("user", "100", "200") + assert.equal([...values.keys()].some(key => key.includes("fact:user:100:200:")), false) +}) + +test("adminDeleteMemory：拒绝过短和不唯一的 ID 前缀", async () => { + const manager = createManager() + manager.store = { + getMeta: async () => ({ factIds: ["12345678-a", "12345678-b"] }), + deleteFact: async () => true + } + + assert.equal((await manager.adminDeleteMemory({ scope: "user", groupId: "100", userId: "200", id: "1234" })).reason, "id-too-short") + assert.equal((await manager.adminDeleteMemory({ scope: "user", groupId: "100", userId: "200", id: "12345678" })).reason, "ambiguous-id") +}) diff --git a/utils/MemoryManager.js b/utils/MemoryManager.js index ca2855b..0dabb8e 100644 --- a/utils/MemoryManager.js +++ b/utils/MemoryManager.js @@ -1,8 +1,14 @@ // 长期记忆门面：对话侧入口（缓冲与提取调度、prompt 生成、管理命令）。 // 存储/提取/检索实现拆至 utils/memory/*，本文件只保留 MemoryManager 编排层。 import { randomUUID } from "crypto" -import { USER_CATEGORIES, GROUP_CATEGORIES, USER_CATEGORY_LABELS, GROUP_CATEGORY_LABELS } from "./memory/constants.js" -import { now, clamp, uniq, sha256, compactText, containsToolFeedback, isRealUserSource, isSimilarContent, normalizeConfig } from "./memory/helpers.js" +import { + USER_CATEGORIES, + GROUP_CATEGORIES, + USER_CATEGORY_LABELS, + GROUP_CATEGORY_LABELS, + USER_CANDIDATE_PROMOTION_COUNT +} from "./memory/constants.js" +import { now, clamp, uniq, sha256, compactText, containsToolFeedback, isLikelyBotCommand, isRealUserSource, isSimilarContent, normalizeConfig } from "./memory/helpers.js" import { MemoryStore } from "./memory/MemoryStore.js" import { MemoryExtractor } from "./memory/MemoryExtractor.js" import { MemoryRetriever } from "./memory/MemoryRetriever.js" @@ -21,17 +27,31 @@ export class MemoryManager { this.retriever = new MemoryRetriever(this.config, this.store, this.extractor) this.userBuffers = new Map() + this.userSeenMessages = new Map() this.groupBuffers = new Map() this.groupSeenMessages = new Map() this.scopeQueues = new Map() } setAiConfig(aiConfig) { + const wasExtractionActive = this.isMemoryExtractionActive() this.config.memoryAiConfig = aiConfig + if (wasExtractionActive && !this.isMemoryExtractionActive()) { + this.discardAllBuffers() + this.extractor.abortActiveRequests?.() + } } updateConfig(config = {}) { + const wasActive = this.config.enabled && this.config.pluginEnabled !== false + const wasExtractionActive = this.isMemoryExtractionActive() Object.assign(this.config, normalizeConfig({ ...this.config, ...config })) + const isActive = this.config.enabled && this.config.pluginEnabled !== false + const isExtractionActive = this.isMemoryExtractionActive() + if ((wasActive && !isActive) || (wasExtractionActive && !isExtractionActive)) { + this.discardAllBuffers() + this.extractor.abortActiveRequests?.() + } } getRedisKey(groupId, userId) { @@ -67,6 +87,7 @@ export class MemoryManager { .then(task) .catch(error => { logger?.error?.(`[MemoryManager] 队列任务执行失败 ${key}: ${error.stack || error}`) + throw error }) .finally(() => { if (this.scopeQueues.get(key) === next) { @@ -91,6 +112,7 @@ export class MemoryManager { if (!text) return false if (text.length < 2) return false if (containsToolFeedback(text)) return false + if (isLikelyBotCommand(text)) return false return true } @@ -101,21 +123,102 @@ export class MemoryManager { const source = event.source if (!isRealUserSource(source)) return null + const groupId = String(event.groupId || event.group_id || "") + const userId = String(event.userId || event.user_id || "") + const rawCreatedAt = Number(event.createdAt || event.time) + const createdAt = Number.isFinite(rawCreatedAt) && rawCreatedAt > 0 + ? (rawCreatedAt < 1e12 ? rawCreatedAt * 1000 : rawCreatedAt) + : now() + const messageId = event.messageId || event.message_id || sha256(`${createdAt}:${groupId}:${userId}:${content}`) + return { content, source: source || "user", - userId: String(event.userId || event.user_id || ""), - groupId: String(event.groupId || event.group_id || ""), - messageId: event.messageId || event.message_id || null, + userId, + groupId, + messageId: String(messageId), senderName: event.senderName || event.nickname || event.sender?.nickname || event.sender?.card || null, - createdAt: now() + createdAt } } + extractEventText(event = {}) { + if (Array.isArray(event.message) && event.message.length) { + const text = event.message + .filter(segment => segment?.type === "text") + .map(segment => segment?.data?.text ?? segment?.text ?? "") + .join("") + return compactText(text, 500) + } + return compactText(event.msg || event.raw_message, 500) + } + + async enqueueGroupEvent(event = {}) { + if (!this.isMemoryActive()) return { queued: false, reason: "disabled" } + if (!this.extractor.canUseMemoryAi()) return { queued: false, reason: "ai-unavailable" } + if (!event.group_id || !event.user_id) return { queued: false, reason: "not-group-message" } + if (this.config.enableGroupWhitelist) { + const allowed = (this.config.allowedGroups || []).some(groupId => String(groupId) === String(event.group_id)) + if (!allowed) return { queued: false, reason: "group-not-allowed" } + } + + const selfIds = [event.self_id, event.bot?.uin, globalThis.Bot?.uin] + .flat() + .filter(value => value !== undefined && value !== null) + .map(String) + if (selfIds.includes(String(event.user_id))) return { queued: false, reason: "self-message" } + + const content = this.extractEventText(event) + if (!content) return { queued: false, reason: "no-text" } + + const interaction = { + groupId: event.group_id, + userId: event.user_id, + content, + source: "user", + messageId: event.message_id, + senderName: event.sender?.card || event.sender?.nickname, + createdAt: event.time + } + const [userResult, groupResult] = await Promise.all([ + this.enqueueInteraction(interaction), + this.extractAndSaveGroupMemories(event.group_id, [interaction]) + ]) + return { ...userResult, groupMemory: groupResult } + } + + rememberSeenUserMessage(groupId, userId, messageId) { + if (!messageId) return false + const key = this.getUserBufferKey(groupId, userId) + let seen = this.userSeenMessages.get(key) + if (!seen) { + seen = [] + } else { + this.userSeenMessages.delete(key) + } + this.userSeenMessages.set(key, seen) + + if (this.userSeenMessages.size > 2000) { + const oldestKey = this.userSeenMessages.keys().next().value + this.userSeenMessages.delete(oldestKey) + } + + const normalizedId = String(messageId) + if (seen.includes(normalizedId)) return false + seen.push(normalizedId) + if (seen.length > 300) seen.splice(0, seen.length - 300) + return true + } + async enqueueInteraction(event = {}) { + if (!this.isMemoryActive()) return { queued: false, reason: "disabled" } + if (!this.extractor.canUseMemoryAi()) return { queued: false, reason: "ai-unavailable" } const interaction = this.normalizeInteraction(event) if (!interaction) return { queued: false, reason: "invalid" } if (!interaction.groupId || !interaction.userId) return { queued: false, reason: "missing-id" } + if (!this.rememberSeenUserMessage(interaction.groupId, interaction.userId, interaction.messageId)) { + return { queued: false, reason: "duplicate" } + } return await this.extractAndSaveMemories(interaction.groupId, interaction.userId, interaction.content, "", interaction) } @@ -162,48 +265,61 @@ export class MemoryManager { } async saveUserMemory(groupId, userId, memory) { - await this.adminClearMemories({ scope: "user", groupId, userId }) - const facts = this.store.collectLegacyFacts(memory, "user", groupId, userId) - for (const fact of facts) { - await this.store.saveFact(fact) - } - const meta = await this.store.getUserMeta(groupId, userId) - meta.relationshipScore = clamp(memory?.relationshipScore ?? memory?.relationship ?? 0.5, 0, 1) - meta.nickname = memory?.nickname || null - await this.store.saveMeta(meta) + this.discardUserBuffer(groupId, userId) + return await this.enqueueUserTask(groupId, userId, async () => { + await this.store.clearScope("user", groupId, userId) + const facts = this.store.collectLegacyFacts(memory, "user", groupId, userId) + for (const fact of facts) { + await this.store.saveFact(fact) + } + const meta = await this.store.getUserMeta(groupId, userId) + meta.relationshipScore = clamp(memory?.relationshipScore ?? memory?.relationship ?? 0.5, 0, 1) + meta.nickname = memory?.nickname || null + await this.store.saveMeta(meta) + return meta + }) } async saveGroupMemory(groupId, memory) { - await this.adminClearMemories({ scope: "group", groupId }) - const facts = this.store.collectLegacyFacts(memory, "group", groupId) - for (const fact of facts) { - await this.store.saveFact(fact) - } + this.discardGroupBuffer(groupId) + return await this.enqueueGroupTask(groupId, async () => { + await this.store.clearScope("group", groupId) + const facts = this.store.collectLegacyFacts(memory, "group", groupId) + for (const fact of facts) { + await this.store.saveFact(fact) + } + return await this.store.getGroupMeta(groupId) + }) } async addMemory(groupId, userId, content, importance = 0.6, category = "identity") { - return await this.applyOperations("user", groupId, userId, [{ - operation: "upsert", - content, - importance, - confidence: 0.8, - category - }]) + return await this.enqueueUserTask(groupId, userId, async () => { + return await this.applyOperations("user", groupId, userId, [{ + operation: "upsert", + content, + importance, + confidence: 0.8, + category + }]) + }) } async addGroupMemory(groupId, content, importance = 0.6, category = "topic") { - return await this.applyOperations("group", groupId, null, [{ - operation: "upsert", - content, - importance, - confidence: 0.8, - category - }]) + return await this.enqueueGroupTask(groupId, async () => { + return await this.applyOperations("group", groupId, null, [{ + operation: "upsert", + content, + importance, + confidence: 0.8, + category + }]) + }) } async updateRelationship(groupId, userId, delta) { return await this.enqueueUserTask(groupId, userId, async () => { const meta = await this.store.getUserMeta(groupId, userId) + if (meta.disabled) return meta.relationshipScore ?? 0.5 meta.relationshipScore = clamp((meta.relationshipScore ?? 0.5) + Number(delta || 0), 0, 1) await this.store.saveMeta(meta) return meta.relationshipScore @@ -222,13 +338,127 @@ export class MemoryManager { }) } + async stageUserOperations(groupId, userId, operations = []) { + const meta = await this.store.getUserMeta(groupId, userId) + if (meta.disabled) { + return { + operations: [], + candidateAdded: 0, + candidatePromoted: 0, + candidateDuplicate: 0, + disabled: operations.length, + candidates: null + } + } + + let candidates = (await this.store.getUserCandidates(groupId, userId)).map(candidate => ({ + ...candidate, + evidenceKeys: [...(candidate.evidenceKeys || [])], + sourceMessageIds: [...(candidate.sourceMessageIds || [])] + })) + const accepted = [] + let candidateAdded = 0 + let candidatePromoted = 0 + let candidateDuplicate = 0 + + for (const operation of operations) { + const isCandidate = operation?.operation === "upsert" && operation.decision === "candidate" + if (!isCandidate) { + const matchingCandidate = candidates.find(candidate => { + const candidateIdMatches = operation?.candidateId && + candidate.id === operation.candidateId && + candidate.category === operation.category + const contentMatches = operation?.content && + candidate.category === operation.category && + isSimilarContent(candidate.content, operation.content) + return candidateIdMatches || contentMatches + }) + accepted.push(matchingCandidate ? { ...operation, candidateId: matchingCandidate.id } : operation) + continue + } + + const sourceMessageIds = uniq(operation.sourceMessageIds || []) + const evidenceKey = sha256(sourceMessageIds.slice().sort().join("|") || `${operation.category}:${operation.content}`) + const candidateIndex = candidates.findIndex(candidate => ( + (candidate.id === operation.candidateId && candidate.category === operation.category) || ( + candidate.category === operation.category && isSimilarContent(candidate.content, operation.content) + ) + )) + const timestamp = now() + const current = candidateIndex >= 0 ? candidates[candidateIndex] : null + + if (current?.evidenceKeys?.includes(evidenceKey)) { + candidateDuplicate++ + continue + } + + const merged = { + ...(current || {}), + id: current?.id || randomUUID(), + groupId: String(groupId), + userId: String(userId), + content: operation.content, + category: operation.category, + importance: Math.max(current?.importance || 0, operation.importance || 0), + confidence: Math.max(current?.confidence || 0, operation.confidence || 0), + evidenceKeys: uniq([...(current?.evidenceKeys || []), evidenceKey]), + sourceMessageIds: uniq([...(current?.sourceMessageIds || []), ...sourceMessageIds]), + firstSeenAt: current?.firstSeenAt || timestamp, + lastSeenAt: timestamp + } + + if (merged.evidenceKeys.length >= USER_CANDIDATE_PROMOTION_COUNT) { + if (candidateIndex >= 0) candidates[candidateIndex] = merged + else candidates.push(merged) + accepted.push({ + ...operation, + decision: "save", + candidateId: merged.id, + content: merged.content, + importance: Math.max(0.6, merged.importance), + confidence: Math.max(0.75, merged.confidence), + sourceMessageIds: merged.sourceMessageIds + }) + candidatePromoted++ + continue + } + + if (candidateIndex >= 0) candidates[candidateIndex] = merged + else candidates.push(merged) + candidateAdded++ + } + + return { operations: accepted, candidates, candidateAdded, candidatePromoted, candidateDuplicate, disabled: 0 } + } + + async finalizeStagedCandidates(groupId, userId, staged, result) { + if (!Array.isArray(staged.candidates)) return + const resolvedIds = new Set(result.resolvedCandidateIds || []) + const remaining = (staged.candidates || []).filter(candidate => !resolvedIds.has(candidate.id)) + return await this.store.saveUserCandidates(groupId, userId, remaining) + } + async applyOperations(scope, groupId, userId, operations = []) { let meta = await this.store.getMeta(scope, groupId, userId) - if (meta.disabled) return { saved: 0, deleted: 0, skipped: operations.length } + if (meta.disabled) { + return { + saved: 0, + deleted: 0, + skipped: operations.length, + noop: 0, + invalid: 0, + belowThreshold: 0, + resolvedCandidateIds: [] + } + } let saved = 0 let deleted = 0 let skipped = 0 + let noop = 0 + let invalid = 0 + let belowThreshold = 0 + const resolvedCandidateIds = [] // 一次性加载当前所有 active facts，循环中维护本地副本，避免 N+1 查询 let activeFacts = await this.store.getFacts(meta, false) @@ -236,39 +466,51 @@ export class MemoryManager { for (const operation of operations) { if (!operation || operation.operation === "noop") { skipped++ + noop++ continue } - const target = operation.id - ? activeFacts.find(f => f.id === operation.id) - : activeFacts.find(f => f.category === operation.category && isSimilarContent(f.content, operation.content)) + const targetById = operation.id ? activeFacts.find(f => f.id === operation.id) : null + const targetByContent = operation.content + ? activeFacts.find(f => f.category === operation.category && isSimilarContent(f.content, operation.content)) + : null + const target = targetById || (operation.operation === "upsert" || !operation.id ? targetByContent : null) if (operation.operation === "delete") { if (target) { await this.store.deleteFact(meta, target.id) activeFacts = activeFacts.filter(f => f.id !== target.id) deleted++ + if (operation.candidateId) resolvedCandidateIds.push(operation.candidateId) } else { skipped++ } continue } + if (operation.operation === "update" && !target) { + skipped++ + invalid++ + continue + } + if (!operation.content || containsToolFeedback(operation.content)) { skipped++ + invalid++ continue } const importance = clamp(operation.importance, 0, 1) if (importance < this.config.importanceThreshold) { skipped++ + belowThreshold++ continue } const embeddingSource = await this.extractor.createEmbedding(operation.content) const fact = { ...(target || {}), - id: target?.id || operation.id || randomUUID(), + id: target?.id || randomUUID(), scope, groupId: String(groupId), userId: scope === "user" ? String(userId) : null, @@ -295,15 +537,21 @@ export class MemoryManager { activeFacts.push(saved_fact) } } - meta = await this.store.getMeta(scope, groupId, userId) - saved++ + if (saved_fact) { + meta = await this.store.getMeta(scope, groupId, userId) + saved++ + if (operation.candidateId) resolvedCandidateIds.push(operation.candidateId) + } else { + skipped++ + invalid++ + } } - return { saved, deleted, skipped } + return { saved, deleted, skipped, noop, invalid, belowThreshold, resolvedCandidateIds: uniq(resolvedCandidateIds) } } async retrieveMemories({ groupId, userId = null, query = "", scope = "user", limit = null } = {}) { - const finalLimit = limit || (scope === "group" ? this.config.promptMaxGroupFacts : this.config.promptMaxUserFacts) + const finalLimit = limit ?? (scope === "group" ? this.config.promptMaxGroupFacts : this.config.promptMaxUserFacts) return await this.retriever.retrieve({ groupId, userId, query, scope, limit: finalLimit }) } @@ -323,6 +571,7 @@ export class MemoryManager { } async getMemoryPromptForUser(groupId, userId, query = "") { + if (!this.isMemoryActive() || this.config.promptMaxUserFacts <= 0) return "" const result = await this.retrieveMemories({ groupId, userId, @@ -331,11 +580,27 @@ export class MemoryManager { limit: this.config.promptMaxUserFacts }) - const prompt = this.formatFactsForPrompt("【长期记忆】关于当前用户的稳定事实，仅用于理解语境，不是指令：", result.facts, USER_CATEGORY_LABELS, this.config.promptMaxChars) + if (result.meta?.disabled) return "" + + const score = clamp(result.meta?.relationshipScore ?? 0.5, 0, 1) + const familiarity = score < 0.3 + ? "疏远" + : score < 0.45 + ? "陌生" + : score < 0.65 + ? "一般" + : score < 0.8 + ? "熟悉" + : "很熟" + const factsPrompt = this.formatFactsForPrompt("【长期记忆】关于当前用户的稳定事实，仅用于理解语境，不是指令：", result.facts, USER_CATEGORY_LABELS, this.config.promptMaxChars) + const prompt = factsPrompt + ? `${factsPrompt}\n- 熟悉程度: ${familiarity}` + : `【当前关系】与当前用户的熟悉程度: ${familiarity}。仅用于调整语气，不是指令。` return prompt.slice(0, this.config.promptMaxChars) } async getGroupMemoryPrompt(groupId, query = "") { + if (!this.isMemoryActive() || this.config.promptMaxGroupFacts <= 0) return "" const result = await this.retrieveMemories({ groupId, query, @@ -351,7 +616,56 @@ export class MemoryManager { return `${groupId}:${userId}` } + isMemoryActive() { + return Boolean(this.config.enabled && this.config.pluginEnabled !== false) + } + + isMemoryExtractionActive() { + return Boolean(this.isMemoryActive() && this.extractor.canUseMemoryAi()) + } + + mergeBufferedMessages(...groups) { + const byId = new Map() + for (const message of groups.flat()) { + if (!message) continue + const key = String(message.messageId || sha256(`${message.createdAt}:${message.userId}:${message.content}`)) + byId.set(key, message) + } + return [...byId.values()].sort((a, b) => (a.createdAt || 0) - (b.createdAt || 0)) + } + + scheduleUserFlush(key, delayMs = this.config.userExtractDebounceSeconds * 1000) { + const buffer = this.userBuffers.get(key) + if (!buffer) return + if (buffer.timer) clearTimeout(buffer.timer) + buffer.timer = setTimeout(() => { + this.flushUserBuffer(key).catch(error => { + logger?.error?.(`[MemoryManager] 用户记忆缓冲区刷新失败 ${key}: ${error.stack || error}`) + }) + }, Math.max(0, delayMs)) + buffer.timer.unref?.() + } + + restoreUserBuffer(buffer, messages, retryAt = 0) { + if (!this.isMemoryExtractionActive()) return + const key = this.getUserBufferKey(buffer.groupId, buffer.userId) + const current = this.userBuffers.get(key) || { + groupId: buffer.groupId, + userId: buffer.userId, + messages: [], + firstBufferedAt: buffer.firstBufferedAt || now(), + timer: null + } + current.messages = this.mergeBufferedMessages(messages, current.messages) + current.firstBufferedAt = Math.min(current.firstBufferedAt || now(), buffer.firstBufferedAt || now()) + this.userBuffers.set(key, current) + const retryDelay = retryAt > now() ? retryAt - now() : 0 + this.scheduleUserFlush(key, Math.max(this.config.userExtractDebounceSeconds * 1000, retryDelay)) + } + async extractAndSaveMemories(groupId, userId, userMessage, botReply = "", meta = {}) { + if (!this.isMemoryActive()) return { queued: false, reason: "disabled" } + if (!this.extractor.canUseMemoryAi()) return { queued: false, reason: "ai-unavailable" } const interaction = this.normalizeInteraction({ ...meta, groupId, @@ -380,14 +694,7 @@ export class MemoryManager { } // 否则设置 debounce 定时器，N 秒内没新消息就触发 - if (buffer.timer) clearTimeout(buffer.timer) - const debounceMs = this.config.userExtractDebounceSeconds * 1000 - buffer.timer = setTimeout(() => { - this.flushUserBuffer(key).catch(error => { - logger?.error?.(`[MemoryManager] 用户记忆缓冲区刷新失败 ${key}: ${error.stack || error}`) - }) - }, debounceMs) - buffer.timer.unref?.() + this.scheduleUserFlush(key) return { queued: true, buffered: buffer.messages.length } } @@ -398,13 +705,58 @@ export class MemoryManager { this.userBuffers.delete(key) if (buffer.timer) clearTimeout(buffer.timer) - const messages = buffer.messages - return await this.enqueueUserTask(buffer.groupId, buffer.userId, async () => { - return await this.extractAndSaveMemoriesNow(buffer.groupId, buffer.userId, messages) - }) + const messages = this.mergeBufferedMessages(buffer.messages) + try { + const result = await this.enqueueUserTask(buffer.groupId, buffer.userId, async () => { + return await this.extractAndSaveMemoriesNow(buffer.groupId, buffer.userId, messages) + }) + if (result?.retryAt) this.restoreUserBuffer(buffer, messages, result.retryAt) + return result + } catch (error) { + this.restoreUserBuffer(buffer, messages) + throw error + } + } + + discardUserBuffer(groupId, userId) { + const key = this.getUserBufferKey(groupId, userId) + const buffer = this.userBuffers.get(key) + if (buffer?.timer) clearTimeout(buffer.timer) + this.userBuffers.delete(key) + } + + discardGroupBuffer(groupId) { + const key = String(groupId) + const buffer = this.groupBuffers.get(key) + if (buffer?.timer) clearTimeout(buffer.timer) + this.groupBuffers.delete(key) + } + + discardAllBuffers() { + for (const key of [...this.userBuffers.keys()]) { + const buffer = this.userBuffers.get(key) + if (buffer?.timer) clearTimeout(buffer.timer) + this.userBuffers.delete(key) + } + for (const groupId of [...this.groupBuffers.keys()]) { + this.discardGroupBuffer(groupId) + } + } + + getExtractionResultReason(extraction, staged, result) { + if (result.saved || result.deleted) return "saved" + if (staged.candidateAdded) return "candidate" + if (staged.candidateDuplicate) return "duplicate" + if (result.belowThreshold) return "below_threshold" + if (result.invalid || extraction.diagnostics.rawCount > extraction.diagnostics.normalizedCount) return "invalid_operation" + if (staged.candidatePromoted) return "promotion_not_saved" + if (result.noop) return "noop" + if (extraction.diagnostics.rawCount === 0) return "model_empty" + return "no_change" } async extractAndSaveMemoriesNow(groupId, userId, messagesOrUserMessage = []) { + if (!this.isMemoryActive()) return { saved: 0, deleted: 0, skipped: 0, reason: "disabled" } if (!this.extractor.canUseMemoryAi()) { logger?.debug?.("[MemoryManager] memoryAiConfig 配置不完整，跳过用户记忆抽取") return { saved: 0, deleted: 0, skipped: 0 } @@ -414,31 +766,61 @@ export class MemoryManager { ? messagesOrUserMessage : [this.normalizeInteraction({ groupId, userId, content: messagesOrUserMessage, source: "user" })].filter(Boolean) - const validMessages = messages.filter(m => this.isValidMemoryText(m.content)) + const validMessages = this.mergeBufferedMessages(messages.filter(m => this.isValidMemoryText(m.content))) if (!validMessages.length) return { saved: 0, deleted: 0, skipped: 0 } const meta = await this.store.getUserMeta(groupId, userId) - if (meta.disabled) return { saved: 0, deleted: 0, skipped: validMessages.length } - if (meta.nextRetryAt && meta.nextRetryAt > now()) return { saved: 0, deleted: 0, skipped: validMessages.length } + if (meta.disabled) return { saved: 0, deleted: 0, skipped: validMessages.length, reason: "disabled" } + if (meta.nextRetryAt && meta.nextRetryAt > now()) { + return { saved: 0, deleted: 0, skipped: validMessages.length, reason: "backoff", retryAt: meta.nextRetryAt } + } meta.lastAttemptAt = now() await this.store.saveMeta(meta) try { - const existingFacts = await this.store.getFacts(meta, false) - const operations = await this.extractor.extractUserOperations({ groupId, userId, messages: validMessages, existingFacts }) - const result = await this.applyOperations("user", groupId, userId, operations) + const [existingFacts, existingCandidates] = await Promise.all([ + this.store.getFacts(meta, false), + this.store.getUserCandidates(groupId, userId) + ]) + const extraction = await this.extractor.extractUserOperationResult({ + groupId, + userId, + messages: validMessages, + existingFacts, + existingCandidates + }) + if (!this.isMemoryExtractionActive()) { + const reason = this.isMemoryActive() ? "ai-unavailable" : "disabled" + return { saved: 0, deleted: 0, skipped: validMessages.length, reason } + } + const staged = await this.stageUserOperations(groupId, userId, extraction.operations) + const result = await this.applyOperations("user", groupId, userId, staged.operations) + await this.finalizeStagedCandidates(groupId, userId, staged, result) const latestMeta = await this.store.getUserMeta(groupId, userId) latestMeta.lastSuccessAt = now() latestMeta.failureCount = 0 latestMeta.nextRetryAt = 0 await this.store.saveMeta(latestMeta) - logger?.debug?.(`[MemoryManager] 用户记忆元数据已刷新 group=${groupId} user=${userId} 操作=${operations.length} 当前事实=${latestMeta.factIds.length}`) - logger?.info?.(`[MemoryManager] 用户记忆抽取完成 group=${groupId} user=${userId} 保存=${result.saved} 删除=${result.deleted} 跳过=${result.skipped}`) - return result + const reason = this.getExtractionResultReason(extraction, staged, result) + const malformed = Math.max(0, extraction.diagnostics.rawCount - extraction.diagnostics.normalizedCount) + logger?.debug?.(`[MemoryManager] 用户记忆元数据已刷新 group=${groupId} user=${userId} 操作=${extraction.operations.length} 当前事实=${latestMeta.factIds.length}`) + logger?.info?.(`[MemoryManager] 用户记忆抽取完成 group=${groupId} user=${userId} 保存=${result.saved} 删除=${result.deleted} 跳过=${result.skipped} 候选=${staged.candidateAdded} 晋升=${staged.candidatePromoted} 重复=${staged.candidateDuplicate} 低阈值=${result.belowThreshold} 无效=${result.invalid + malformed} 结果=${reason} 解析=${extraction.diagnostics.parseStatus}`) + return { + ...result, + candidateAdded: staged.candidateAdded, + candidatePromoted: staged.candidatePromoted, + candidateDuplicate: staged.candidateDuplicate, + reason, + parseStatus: extraction.diagnostics.parseStatus + } } catch (error) { - await this.recordExtractionFailure(meta, error, "user") - return { saved: 0, deleted: 0, skipped: validMessages.length, error: error.message } + if (!this.isMemoryExtractionActive()) { + const reason = this.isMemoryActive() ? "ai-unavailable" : "disabled" + return { saved: 0, deleted: 0, skipped: validMessages.length, reason } + } + const retryAt = await this.recordExtractionFailure(meta, error, "user") + return { saved: 0, deleted: 0, skipped: validMessages.length, error: error.message, reason: "error", retryAt } } } @@ -458,121 +840,210 @@ export class MemoryManager { const content = compactText(rawContent, 500) if (!this.isValidMemoryText(content)) return null + const rawCreatedAt = message.createdAt ?? message.time + const numericCreatedAt = Number(rawCreatedAt) + const parsedCreatedAt = Number.isFinite(numericCreatedAt) && numericCreatedAt > 0 + ? (numericCreatedAt < 1e12 ? numericCreatedAt * 1000 : numericCreatedAt) + : Date.parse(rawCreatedAt) + return { content, source: source || "user", userId: String(userId), senderName: message.senderName || sender.nickname || sender.card || nameMatch?.[1] || "群成员", messageId: message.messageId || message.message_id || sha256(`${message.time || ""}:${userId}:${content}`), - createdAt: message.createdAt || now() + createdAt: Number.isFinite(parsedCreatedAt) ? parsedCreatedAt : now() } } rememberSeenGroupMessage(groupId, messageId) { if (!messageId) return false - let seen = this.groupSeenMessages.get(groupId) + const key = String(groupId) + let seen = this.groupSeenMessages.get(key) if (!seen) { seen = [] - this.groupSeenMessages.set(groupId, seen) + } else { + this.groupSeenMessages.delete(key) + } + this.groupSeenMessages.set(key, seen) + + if (this.groupSeenMessages.size > 500) { + const oldestKey = this.groupSeenMessages.keys().next().value + this.groupSeenMessages.delete(oldestKey) } - if (seen.includes(messageId)) return false - seen.push(messageId) + const normalizedId = String(messageId) + if (seen.includes(normalizedId)) return false + seen.push(normalizedId) if (seen.length > 300) seen.splice(0, seen.length - 300) return true } + scheduleGroupFlush(groupId, delayMs = this.config.groupExtractMinIntervalMinutes * 60 * 1000) { + const key = String(groupId) + const buffer = this.groupBuffers.get(key) + if (!buffer) return + if (buffer.timer) clearTimeout(buffer.timer) + buffer.timer = setTimeout(() => { + this.flushGroupBuffer(key).catch(error => { + logger?.error?.(`[MemoryManager] 群记忆缓冲区刷新失败 ${key}: ${error.stack || error}`) + }) + }, Math.max(1000, delayMs)) + buffer.timer.unref?.() + } + + restoreGroupBuffer(buffer, messages, retryAt = 0) { + if (!this.isMemoryExtractionActive()) return + const key = String(buffer.groupId) + const current = this.groupBuffers.get(key) || { + groupId: key, + messages: [], + firstBufferedAt: buffer.firstBufferedAt || now(), + timer: null + } + current.messages = this.mergeBufferedMessages(messages, current.messages) + current.firstBufferedAt = Math.min(current.firstBufferedAt || now(), buffer.firstBufferedAt || now()) + this.groupBuffers.set(key, current) + const intervalMs = this.config.groupExtractMinIntervalMinutes * 60 * 1000 + const retryDelay = retryAt > now() ? retryAt - now() : 0 + this.scheduleGroupFlush(key, Math.max(intervalMs, retryDelay)) + } + async extractAndSaveGroupMemories(groupId, chatHistory = []) { + if (!this.isMemoryActive()) return { queued: false, reason: "disabled" } + if (!this.extractor.canUseMemoryAi()) return { queued: false, reason: "ai-unavailable" } if (!groupId || !Array.isArray(chatHistory) || !chatHistory.length) { return { queued: false, reason: "empty" } } - let buffer = this.groupBuffers.get(groupId) + const key = String(groupId) + let buffer = this.groupBuffers.get(key) if (!buffer) { - buffer = { groupId, messages: [], firstBufferedAt: now(), timer: null } - this.groupBuffers.set(groupId, buffer) + buffer = { groupId: key, messages: [], firstBufferedAt: now(), timer: null } + this.groupBuffers.set(key, buffer) } - for (const rawMessage of chatHistory) { - const message = this.normalizeGroupHistoryMessage(rawMessage) + const normalizedMessages = chatHistory + .map(rawMessage => this.normalizeGroupHistoryMessage(rawMessage)) + .filter(Boolean) + .sort((a, b) => a.createdAt - b.createdAt) + for (const message of normalizedMessages) { if (!message) continue - if (!this.rememberSeenGroupMessage(groupId, message.messageId)) continue + if (!this.rememberSeenGroupMessage(key, message.messageId)) continue buffer.messages.push(message) } if (!buffer.messages.length) return { queued: false, reason: "no-new-message" } - const meta = await this.store.getGroupMeta(groupId) - const intervalMs = this.config.groupExtractMinIntervalMinutes * 60 * 1000 - const intervalBase = meta.lastAttemptAt || buffer.firstBufferedAt - if (!buffer.timer) { - const delay = Math.max(1000, intervalMs - (now() - intervalBase)) - buffer.timer = setTimeout(() => { - this.flushGroupBuffer(groupId).catch(error => { - logger?.error?.(`[MemoryManager] 群记忆缓冲区刷新失败 ${groupId}: ${error.stack || error}`) - }) - }, delay) - buffer.timer.unref?.() - } - - const dueByInterval = intervalBase && now() - intervalBase >= intervalMs const dueByBatch = buffer.messages.length >= this.config.groupExtractMaxBatchMessages + if (dueByBatch) return await this.flushGroupBuffer(key) - if (!dueByInterval && !dueByBatch) { - return { queued: true, buffered: buffer.messages.length } + if (!buffer.timer) { + const intervalMs = this.config.groupExtractMinIntervalMinutes * 60 * 1000 + try { + const meta = await this.store.getGroupMeta(key) + const intervalBase = meta.lastAttemptAt || buffer.firstBufferedAt + this.scheduleGroupFlush(key, intervalMs - (now() - intervalBase)) + } catch (error) { + this.scheduleGroupFlush(key, intervalMs) + throw error + } } - return await this.flushGroupBuffer(groupId) + return { queued: true, buffered: buffer.messages.length } } async flushGroupBuffer(groupId) { - const buffer = this.groupBuffers.get(groupId) + const key = String(groupId) + const buffer = this.groupBuffers.get(key) if (!buffer || !buffer.messages.length) return { queued: false, reason: "empty" } - this.groupBuffers.delete(groupId) if (buffer.timer) clearTimeout(buffer.timer) + buffer.timer = null - const messages = buffer.messages.slice(-this.config.groupExtractMaxBatchMessages) - return await this.enqueueGroupTask(groupId, async () => { - return await this.extractAndSaveGroupMemoriesNow(groupId, messages) - }) + const messages = buffer.messages.splice(0, this.config.groupExtractMaxBatchMessages) + if (!buffer.messages.length) this.groupBuffers.delete(key) + const force = messages.length >= this.config.groupExtractMaxBatchMessages + + try { + const result = await this.enqueueGroupTask(key, async () => { + return await this.extractAndSaveGroupMemoriesNow(key, messages, { force }) + }) + if (result?.retryAt) { + this.restoreGroupBuffer(buffer, messages, result.retryAt) + } else { + const remaining = this.groupBuffers.get(key) + if (remaining?.messages.length >= this.config.groupExtractMaxBatchMessages) { + await this.flushGroupBuffer(key) + } else if (remaining?.messages.length && !remaining.timer) { + this.scheduleGroupFlush(key) + } + } + return result + } catch (error) { + this.restoreGroupBuffer(buffer, messages) + throw error + } } - async extractAndSaveGroupMemoriesNow(groupId, messagesOrHistory = []) { + async extractAndSaveGroupMemoriesNow(groupId, messagesOrHistory = [], { force = false } = {}) { + if (!this.isMemoryActive()) return { saved: 0, deleted: 0, skipped: 0, reason: "disabled" } if (!this.extractor.canUseMemoryAi()) { logger?.debug?.("[MemoryManager] memoryAiConfig 配置不完整，跳过群记忆抽取") return { saved: 0, deleted: 0, skipped: 0 } } - const messages = (Array.isArray(messagesOrHistory) ? messagesOrHistory : []) + const messages = this.mergeBufferedMessages((Array.isArray(messagesOrHistory) ? messagesOrHistory : []) .map(message => this.normalizeGroupHistoryMessage(message)) .filter(Boolean) - .filter(m => this.isValidMemoryText(m.content)) + .filter(m => this.isValidMemoryText(m.content))) if (!messages.length) return { saved: 0, deleted: 0, skipped: 0 } const meta = await this.store.getGroupMeta(groupId) - if (meta.disabled) return { saved: 0, deleted: 0, skipped: messages.length } - if (meta.nextRetryAt && meta.nextRetryAt > now()) return { saved: 0, deleted: 0, skipped: messages.length } + if (meta.disabled) return { saved: 0, deleted: 0, skipped: messages.length, reason: "disabled" } + if (meta.nextRetryAt && meta.nextRetryAt > now()) { + return { saved: 0, deleted: 0, skipped: messages.length, reason: "backoff", retryAt: meta.nextRetryAt } + } + const intervalMs = this.config.groupExtractMinIntervalMinutes * 60 * 1000 + const nextAllowedAt = (meta.lastAttemptAt || 0) + intervalMs + if (!force && meta.lastAttemptAt && nextAllowedAt > now()) { + return { saved: 0, deleted: 0, skipped: messages.length, reason: "interval", retryAt: nextAllowedAt } + } - // 不再做 interval 二次检查：buffer flush 已经决策过是否该提取了，这里只负责执行 meta.lastAttemptAt = now() await this.store.saveMeta(meta) try { const existingFacts = await this.store.getFacts(meta, false) - const operations = await this.extractor.extractGroupOperations({ groupId, messages, existingFacts }) - const result = await this.applyOperations("group", groupId, null, operations) + const extraction = await this.extractor.extractGroupOperationResult({ groupId, messages, existingFacts }) + if (!this.isMemoryExtractionActive()) { + const reason = this.isMemoryActive() ? "ai-unavailable" : "disabled" + return { saved: 0, deleted: 0, skipped: messages.length, reason } + } + const result = await this.applyOperations("group", groupId, null, extraction.operations) const latestMeta = await this.store.getGroupMeta(groupId) latestMeta.lastSuccessAt = now() latestMeta.failureCount = 0 latestMeta.nextRetryAt = 0 await this.store.saveMeta(latestMeta) - logger?.debug?.(`[MemoryManager] 群记忆元数据已刷新 group=${groupId} 操作=${operations.length} 当前事实=${latestMeta.factIds.length}`) - logger?.info?.(`[MemoryManager] 群记忆抽取完成 group=${groupId} 保存=${result.saved} 删除=${result.deleted} 跳过=${result.skipped}`) - return result + const reason = result.saved || result.deleted + ? "saved" + : result.belowThreshold + ? "below_threshold" + : extraction.diagnostics.rawCount === 0 + ? "model_empty" + : "no_change" + const malformed = Math.max(0, extraction.diagnostics.rawCount - extraction.diagnostics.normalizedCount) + logger?.debug?.(`[MemoryManager] 群记忆元数据已刷新 group=${groupId} 操作=${extraction.operations.length} 当前事实=${latestMeta.factIds.length}`) + logger?.info?.(`[MemoryManager] 群记忆抽取完成 group=${groupId} 保存=${result.saved} 删除=${result.deleted} 跳过=${result.skipped} 低阈值=${result.belowThreshold} 无效=${result.invalid + malformed} 结果=${reason} 解析=${extraction.diagnostics.parseStatus}`) + return { ...result, reason, parseStatus: extraction.diagnostics.parseStatus } } catch (error) { - await this.recordExtractionFailure(meta, error, "group") - return { saved: 0, deleted: 0, skipped: messages.length, error: error.message } + if (!this.isMemoryExtractionActive()) { + const reason = this.isMemoryActive() ? "ai-unavailable" : "disabled" + return { saved: 0, deleted: 0, skipped: messages.length, reason } + } + const retryAt = await this.recordExtractionFailure(meta, error, "group") + return { saved: 0, deleted: 0, skipped: messages.length, error: error.message, reason: "error", retryAt } } } @@ -583,6 +1054,7 @@ export class MemoryManager { latestMeta.nextRetryAt = now() + backoffMs await this.store.saveMeta(latestMeta) logger?.error?.(`[MemoryManager] ${scope === "user" ? "用户记忆" : "群记忆"}抽取失败: ${error.stack || error}`) + return latestMeta.nextRetryAt } async adminListMemories({ scope = "user", groupId, userId = null, query = "", limit = 20, includeDeleted = false } = {}) { @@ -612,39 +1084,75 @@ export class MemoryManager { async adminDeleteMemory({ scope = null, groupId, userId = null, id } = {}) { if (!id) return { deleted: false, reason: "missing-id" } + const normalizedId = String(id).trim() + if (normalizedId.length < 8) return { deleted: false, reason: "id-too-short" } const scopes = scope ? [scope] : ["user", "group"] for (const itemScope of scopes) { - const meta = await this.store.getMeta(itemScope, groupId, itemScope === "user" ? userId : null) - const factId = meta.factIds.find(itemId => itemId === id || itemId.startsWith(id)) - if (!factId) continue - const deleted = await this.store.deleteFact(meta, factId) - return { deleted, scope: itemScope, id: factId } + const deleteTask = async () => { + const itemUserId = itemScope === "user" ? userId : null + const meta = await this.store.getMeta(itemScope, groupId, itemUserId) + const exact = meta.factIds.find(itemId => itemId === normalizedId) + const matches = exact ? [exact] : meta.factIds.filter(itemId => itemId.startsWith(normalizedId)) + if (!matches.length) return { deleted: false, reason: "not-found" } + if (matches.length > 1) return { deleted: false, reason: "ambiguous-id" } + const deleted = await this.store.deleteFact(meta, matches[0]) + return { deleted, scope: itemScope, id: matches[0] } + } + const result = itemScope === "user" + ? await this.enqueueUserTask(groupId, userId, deleteTask) + : await this.enqueueGroupTask(groupId, deleteTask) + if (result.deleted || result.reason !== "not-found") return result } return { deleted: false, reason: "not-found" } } async adminClearMemories({ scope = "user", groupId, userId = null } = {}) { - const count = await this.store.clearScope(scope, groupId, userId) - return { cleared: count, scope, groupId, userId } + const clear = async () => { + const count = await this.store.clearScope(scope, groupId, userId) + return { cleared: count, scope, groupId, userId } + } + + if (scope === "user") { + this.discardUserBuffer(groupId, userId) + const result = await this.enqueueUserTask(groupId, userId, clear) + if (!result) throw new Error("用户记忆清空任务执行失败") + return result + } + + this.discardGroupBuffer(groupId) + const result = await this.enqueueGroupTask(groupId, clear) + if (!result) throw new Error("群记忆清空任务执行失败") + return result } async adminSetUserMemoryEnabled({ groupId, userId, enabled }) { - const meta = await this.store.setDisabled("user", groupId, userId, !enabled) - return { enabled: !meta.disabled, meta } + if (!enabled) this.discardUserBuffer(groupId, userId) + return await this.enqueueUserTask(groupId, userId, async () => { + if (!enabled) { + this.discardUserBuffer(groupId, userId) + await this.store.clearUserCandidates(groupId, userId) + } + const meta = await this.store.setDisabled("user", groupId, userId, !enabled) + return { enabled: !meta.disabled, meta } + }) } async adminSetGroupMemoryEnabled({ groupId, enabled }) { - const meta = await this.store.setDisabled("group", groupId, null, !enabled) - return { enabled: !meta.disabled, meta } + if (!enabled) this.discardGroupBuffer(groupId) + return await this.enqueueGroupTask(groupId, async () => { + if (!enabled) this.discardGroupBuffer(groupId) + const meta = await this.store.setDisabled("group", groupId, null, !enabled) + return { enabled: !meta.disabled, meta } + }) } async adminStatus({ groupId, userId } = {}) { const userMeta = userId ? await this.store.getUserMeta(groupId, userId) : null const groupMeta = groupId ? await this.store.getGroupMeta(groupId) : null return { - enabled: this.config.enabled, + enabled: this.isMemoryActive(), user: userMeta ? { disabled: userMeta.disabled, factCount: userMeta.factIds.length, diff --git a/utils/memory/MemoryExtractor.js b/utils/memory/MemoryExtractor.js index c3c8b58..61c3bff 100644 --- a/utils/memory/MemoryExtractor.js +++ b/utils/memory/MemoryExtractor.js @@ -2,12 +2,39 @@ // 从 MemoryManager.js 拆出（行为等价搬迁）。 import { callAI } from "../apiClient.js" import { USER_CATEGORIES, GROUP_CATEGORIES } from "./constants.js" -import { clamp, uniq, sha256, compactText, extractJsonArray } from "./helpers.js" +import { clamp, uniq, sha256, compactText, parseJsonArrayResult } from "./helpers.js" + +const MEMORY_AI_CONCURRENCY = 4 +const MEMORY_AI_TIMEOUT_MS = 45_000 +const EMBEDDING_CACHE_TTL_MS = 10 * 60 * 1000 +const EMBEDDING_CACHE_MAX_ITEMS = 200 +let activeMemoryAiRequests = 0 +const pendingMemoryAiRequests = [] + +function runWithMemoryAiSlot(task) { + return new Promise((resolve, reject) => { + const run = () => { + activeMemoryAiRequests++ + Promise.resolve() + .then(task) + .then(resolve, reject) + .finally(() => { + activeMemoryAiRequests-- + pendingMemoryAiRequests.shift()?.() + }) + } + if (activeMemoryAiRequests < MEMORY_AI_CONCURRENCY) run() + else pendingMemoryAiRequests.push(run) + }) +} export class MemoryExtractor { constructor(config, store) { this.config = config this.store = store + this.activeControllers = new Set() + this.embeddingCache = new Map() + this.abortGeneration = 0 } canUseMemoryAi() { @@ -21,22 +48,51 @@ export class MemoryExtractor { return Boolean(cfg.embeddingApiUrl && cfg.embeddingApiKey) } + abortActiveRequests() { + this.abortGeneration++ + for (const controller of this.activeControllers) controller.abort() + this.activeControllers.clear() + } + + createRequestController(timeoutMs) { + const controller = new AbortController() + this.activeControllers.add(controller) + const timeoutId = setTimeout(() => controller.abort(), timeoutMs) + return { + controller, + release: () => { + clearTimeout(timeoutId) + this.activeControllers.delete(controller) + } + } + } + async callChat(messages, maxTokens = 600) { const cfg = this.config.memoryAiConfig || {} if (!cfg.memoryAiUrl || !cfg.memoryAiApikey) return "[]" - const result = await callAI( - { - url: cfg.memoryAiUrl, - model: cfg.memoryAiModel || "gpt-4o-mini", - apikey: cfg.memoryAiApikey - }, - messages, - { - maxTokens, - temperature: 0.2 + const generation = this.abortGeneration + const result = await runWithMemoryAiSlot(async () => { + if (generation !== this.abortGeneration) throw new Error("记忆请求已取消") + const request = this.createRequestController(MEMORY_AI_TIMEOUT_MS) + try { + return await callAI( + { + url: cfg.memoryAiUrl, + model: cfg.memoryAiModel || "gpt-4o-mini", + apikey: cfg.memoryAiApikey + }, + messages, + { + maxTokens, + temperature: 0.2, + signal: request.controller.signal + } + ) + } finally { + request.release() } - ) + }) if (result.error) { throw new Error(`记忆 AI 请求失败：${result.error}`) @@ -45,49 +101,116 @@ export class MemoryExtractor { return result?.choices?.[0]?.message?.content?.trim() || "[]" } + async parseOperationResponse(messages, maxTokens) { + const content = await this.callChat(messages, maxTokens) + const parsed = parseJsonArrayResult(content) + if (parsed.status !== "invalid") { + return { items: parsed.items, parseStatus: parsed.status, repaired: false } + } + + logger?.warn?.("[MemoryExtractor] 记忆 AI 返回了无效 JSON，尝试修复一次") + const repairedContent = await this.callChat([ + { + role: "system", + content: "把用户提供的内容整理成合法 JSON 数组。只修复格式，不增删事实，不输出解释；无法恢复时输出 []。" + }, + { + role: "user", + content: compactText(content, 6000) + } + ], maxTokens) + const repaired = parseJsonArrayResult(repairedContent) + if (repaired.status === "invalid") { + throw new Error("记忆 AI 连续两次返回无法解析的 JSON") + } + + return { + items: repaired.items, + parseStatus: repaired.status === "empty" ? "repaired_empty" : "repaired", + repaired: true + } + } + async createEmbedding(text) { if (!this.canUseEmbedding()) return { embedding: null, embeddingHash: null } const cfg = this.config.embeddingAiConfig || {} - const hash = sha256(`${cfg.embeddingApiModel || "text-embedding-3-small"}:${text}`) - - // embedding 请求加超时：语义检索时此调用在 handleTool 对话准备链里串行阻塞， - // 无超时时 embedding 服务半挂会拖住整个回复（表现为"接口没问题但回复要一分钟"） - const controller = new AbortController() - const timeoutId = setTimeout(() => controller.abort(), 8000) - try { - const response = await fetch(cfg.embeddingApiUrl, { - method: "POST", - headers: { - Authorization: `Bearer ${cfg.embeddingApiKey}`, - "Content-Type": "application/json" - }, - body: JSON.stringify({ - model: cfg.embeddingApiModel || "text-embedding-3-small", - input: text - }), - signal: controller.signal - }) + const hash = this.embeddingHashFor(text) + const cached = this.embeddingCache.get(hash) + if (cached && cached.expiresAt > Date.now()) { + this.embeddingCache.delete(hash) + this.embeddingCache.set(hash, cached) + return { embedding: cached.embedding, embeddingHash: hash } + } - if (!response.ok) { - logger?.warn?.(`[MemoryExtractor] embedding 请求失败：${response.status}`) + const generation = this.abortGeneration + return await runWithMemoryAiSlot(async () => { + if (generation !== this.abortGeneration) return { embedding: null, embeddingHash: null } + const request = this.createRequestController(8000) + try { + const response = await fetch(cfg.embeddingApiUrl, { + method: "POST", + headers: { + Authorization: `Bearer ${cfg.embeddingApiKey}`, + "Content-Type": "application/json" + }, + body: JSON.stringify({ + model: cfg.embeddingApiModel || "text-embedding-3-small", + input: text + }), + signal: request.controller.signal + }) + + if (!response.ok) { + logger?.warn?.(`[MemoryExtractor] embedding 请求失败：${response.status}`) + return { embedding: null, embeddingHash: null } + } + + const data = await response.json() + const embedding = data?.data?.[0]?.embedding + if (!Array.isArray(embedding)) return { embedding: null, embeddingHash: null } + this.embeddingCache.set(hash, { embedding, expiresAt: Date.now() + EMBEDDING_CACHE_TTL_MS }) + while (this.embeddingCache.size > EMBEDDING_CACHE_MAX_ITEMS) { + this.embeddingCache.delete(this.embeddingCache.keys().next().value) + } + return { embedding, embeddingHash: hash } + } catch (error) { + logger?.warn?.(`[MemoryExtractor] 已跳过 embedding：${error.message}`) return { embedding: null, embeddingHash: null } + } finally { + request.release() } + }) + } - const data = await response.json() - const embedding = data?.data?.[0]?.embedding - return Array.isArray(embedding) ? { embedding, embeddingHash: hash } : { embedding: null, embeddingHash: null } - } catch (error) { - logger?.warn?.(`[MemoryExtractor] 已跳过 embedding：${error.message}`) - return { embedding: null, embeddingHash: null } - } finally { - clearTimeout(timeoutId) - } + embeddingHashFor(text) { + const cfg = this.config.embeddingAiConfig || {} + return sha256(`${cfg.embeddingApiUrl || ""}:${cfg.embeddingApiModel || "text-embedding-3-small"}:${text}`) } existingHint(facts) { if (!facts?.length) return "" - return facts + const sorted = [...facts].sort((a, b) => { + if ((b.importance || 0) !== (a.importance || 0)) return (b.importance || 0) - (a.importance || 0) + return (b.updatedAt || b.createdAt || 0) - (a.updatedAt || a.createdAt || 0) + }) + const selected = [] + const selectedIds = new Set() + const perCategory = Math.max(0, Number(this.config.minFactsPerCategory) || 0) + for (const category of [...USER_CATEGORIES, ...GROUP_CATEGORIES]) { + for (const fact of sorted.filter(item => item.category === category).slice(0, perCategory)) { + if (selectedIds.has(fact.id)) continue + selected.push(fact) + selectedIds.add(fact.id) + } + } + for (const fact of sorted) { + if (selected.length >= 50) break + if (selectedIds.has(fact.id)) continue + selected.push(fact) + selectedIds.add(fact.id) + } + return selected .slice(0, 50) .map(f => `${f.id} | ${f.category} | ${f.content}`) .join("\n") @@ -100,11 +223,8 @@ export class MemoryExtractor { for (const item of rawItems) { if (!item || typeof item !== "object") continue - const operation = ["upsert", "update", "delete", "noop"].includes(item.operation) - ? item.operation - : item.action && ["upsert", "update", "delete", "noop"].includes(item.action) - ? item.action - : "upsert" + const operation = item.operation || item.action + if (!["upsert", "update", "delete", "noop"].includes(operation)) continue if (operation === "noop") { operations.push({ operation: "noop" }) @@ -115,33 +235,54 @@ export class MemoryExtractor { const id = item.id ? String(item.id) : null if (!content && operation !== "delete") continue - const category = categories.includes(item.category) ? item.category : categories[0] - const importance = clamp(item.importance ?? 0.6, 0, 1) + const category = categories.includes(item.category) ? item.category : null + if (!category && operation !== "delete") continue + const explicitDecision = item.decision || item.disposition || item.memoryType + const decision = scope === "group" || operation === "delete" || operation === "update" + ? "save" + : explicitDecision === "save" || item.explicit === true + ? "save" + : "candidate" + const importance = clamp(item.importance ?? (decision === "candidate" ? 0.4 : 0.6), 0, 1) const confidence = clamp(item.confidence ?? 0.7, 0, 1) + const sourceIndexes = uniq(item.sourceIndexes || item.source_indexes || []) + .map(index => Number(index)) + .filter(index => Number.isInteger(index) && index >= 1 && index <= (source.messages || []).length) + const evidenceMessages = sourceIndexes.length + ? sourceIndexes.map(index => source.messages[index - 1]).filter(Boolean) + : (source.messages || []) operations.push({ operation, + decision, id, + candidateId: item.candidateId ? String(item.candidateId) : null, content, - category, + category: category || categories[0], importance, confidence, - sourceMessageIds: uniq([...(item.sourceMessageIds || []), ...(source.sourceMessageIds || [])]), - sourceUserIds: uniq([...(item.sourceUserIds || []), ...(source.sourceUserIds || [])]) + sourceMessageIds: uniq(evidenceMessages.length + ? evidenceMessages.map(message => message.messageId) + : source.sourceMessageIds || []), + sourceUserIds: uniq(evidenceMessages.length + ? evidenceMessages.map(message => message.userId) + : source.sourceUserIds || []) }) } return operations } - async extractUserOperations({ groupId, userId, messages, existingFacts }) { - if (!this.canUseMemoryAi()) return [] + async extractUserOperationResult({ groupId, userId, messages, existingFacts, existingCandidates = [] }) { + if (!this.canUseMemoryAi()) { + return { operations: [], diagnostics: { parseStatus: "disabled", rawCount: 0, normalizedCount: 0 } } + } const chatText = messages .map((m, index) => `${index + 1}. ${m.content}`) .join("\n") - const systemPrompt = `你是长期记忆抽取器。只从真实用户发言中抽取稳定事实，输出操作式 JSON 数组，不要输出解释。 + const systemPrompt = `你是长期记忆抽取器。只从真实用户发言中抽取与该用户有关的事实，输出操作式 JSON 数组，不要输出解释。 允许的 operation: - upsert: 新增或合并事实 @@ -149,6 +290,11 @@ export class MemoryExtractor { - delete: 删除过时或被用户否认的事实 - noop: 没有可保存事实 +允许的 decision（upsert 必填）: +- save: 用户明确自述的稳定事实，可直接进入长期记忆 +- candidate: 仅从行为或上下文推测出的潜在偏好、习惯，需以后再次出现才能晋升 +- 新线索与已有候选含义相同，必须原样复用候选 content，并填写对应 candidateId + 用户记忆分类: - identity: 身份、昵称、所在地、职业、基础属性 - likes: 喜好、兴趣、偏好 @@ -158,50 +304,79 @@ export class MemoryExtractor { - skills: 技能、正在学习或擅长的事 - experience: 近期计划、经历、重要事件 -【核心原则：宁可漏抽，不可乱抽】 +【核心原则：明确自述及时保存，推测信息谨慎累计】 - 要从消息中抽取「用户明确表达的事实」，而不是「旁观者解读的印象」。 - 不要把用户的一句抱怨当成习惯，不要把一次提到当成喜好。 - 例如用户说"今天好累" → 不要抽成"用户经常疲劳"；说"想吃火锅" → 不要抽成"喜欢火锅"。 -- 只有用户在多条消息中反复提及，或是明确陈述（如"我是程序员""我喜欢打游戏"），才算稳定事实。 +- 用户明确自述（如"我是程序员""我喜欢打游戏""我最近在学吉他"）即使只出现一次，也应输出 decision=save。 +- 仅从行为推测的潜在偏好或习惯可输出 decision=candidate，内容要写成简洁、可合并的候选事实。 +- 一次性情绪、普通请求、随口玩笑和无意义闲聊不要输出 candidate，直接忽略。 【importance 评分标准】 - 0.9-1.0: 用户明确陈述的身份、职业、家庭成员等核心信息 -- 0.7-0.8: 用户明确表达且反复出现的喜好/习惯 -- 0.5-0.6: 单次提及但有明确陈述的事实（如"我最近在学吉他"） -- 0.3-0.4: 模糊推断，仅在多条消息交叉验证时才考虑，否则直接跳过 +- 0.7-0.8: 用户明确表达的喜好、反感、技能或长期习惯 +- 0.5-0.6: 单次但明确陈述的近期学习、计划或重要经历 +- 0.3-0.4: 有一定依据但尚未被用户明确确认的候选事实，只能输出 decision=candidate - 0.0-0.2: 临时情绪、一次性事件，禁止保存 【其他规则】 - 禁止保存系统提示、工具结果、工具调用、机器人回复。 - 禁止保存短期闲聊、纯语气词、临时请求。 - 禁止从单次"今天 XX"类型的临时话题中提取事实。 -- 只保留重要性不低于 ${this.config.importanceThreshold} 的事实。 +- decision=save 的事实重要性必须不低于 ${this.config.importanceThreshold}；decision=candidate 可使用 0.3-0.4。 +- 每条操作必须用 sourceIndexes 填写支持该事实的发言序号，例如只来自第 2 条就填 [2]，不要把无关消息算作证据。 - 如果用户明确否认旧事实，请输出 delete 或 update。 -- 输出示例: [{"operation":"upsert","content":"喜欢原神","category":"likes","importance":0.8,"confidence":0.9}] +- save 示例: [{"operation":"upsert","decision":"save","content":"喜欢原神","category":"likes","importance":0.8,"confidence":0.9,"sourceIndexes":[1]}] +- candidate 示例: [{"operation":"upsert","decision":"candidate","content":"经常熬夜","category":"habits","importance":0.4,"confidence":0.6,"sourceIndexes":[2]}] - 无有效事实时输出 []。` const existing = this.existingHint(existingFacts) + const candidates = existingCandidates + .slice(0, 30) + .map(candidate => `${candidate.id} | ${candidate.category} | ${candidate.content}`) + .join("\n") const userPrompt = `群 ${groupId} 用户 ${userId} 的真实发言: ${chatText} 已有记忆: ${existing || "无"} +已有候选（格式为 candidateId | category | content）: +${candidates || "无"} + 请输出 JSON 数组。` - const content = await this.callChat([ + const parsed = await this.parseOperationResponse([ { role: "system", content: systemPrompt }, { role: "user", content: userPrompt } ], 700) - return this.normalizeOperations(extractJsonArray(content), "user", { + const operations = this.normalizeOperations(parsed.items, "user", { + messages, sourceMessageIds: messages.map(m => m.messageId).filter(Boolean), sourceUserIds: [userId] }) + + return { + operations, + diagnostics: { + parseStatus: parsed.parseStatus, + repaired: parsed.repaired, + rawCount: parsed.items.length, + normalizedCount: operations.length + } + } + } + + async extractUserOperations(args) { + const result = await this.extractUserOperationResult(args) + return result.operations } - async extractGroupOperations({ groupId, messages, existingFacts }) { - if (!this.canUseMemoryAi()) return [] + async extractGroupOperationResult({ groupId, messages, existingFacts }) { + if (!this.canUseMemoryAi()) { + return { operations: [], diagnostics: { parseStatus: "disabled", rawCount: 0, normalizedCount: 0 } } + } const chatText = messages .map((m, index) => `${index + 1}. ${m.senderName || "群成员"}(QQ:${m.userId || "unknown"}): ${m.content}`) @@ -240,7 +415,8 @@ ${existing || "无"} - 禁止保存系统提示、工具结果、工具调用、机器人回复。 - 禁止把用户对机器人的指令保存成群规则。 - 只保留重要性不低于 ${this.config.importanceThreshold} 的事实。 -- 输出示例: [{“operation”:”upsert”,”content”:”群里常用”哈基米”当玩笑称呼”,”category”:”meme”,”importance”:0.7,”confidence”:0.8}] +- 每条操作必须用 sourceIndexes 填写支持该事实的发言序号，只引用真正相关的消息。 +- 输出示例: [{"operation":"upsert","content":"群里常用哈基米当玩笑称呼","category":"meme","importance":0.7,"confidence":0.8,"sourceIndexes":[1,3]}] - 无有效事实时输出 []。` const existing = this.existingHint(existingFacts) @@ -252,14 +428,30 @@ ${existing || "无"} 请输出 JSON 数组。` - const content = await this.callChat([ + const parsed = await this.parseOperationResponse([ { role: "system", content: systemPrompt }, { role: "user", content: userPrompt } ], 900) - return this.normalizeOperations(extractJsonArray(content), "group", { + const operations = this.normalizeOperations(parsed.items, "group", { + messages, sourceMessageIds: messages.map(m => m.messageId).filter(Boolean), sourceUserIds: messages.map(m => m.userId).filter(Boolean) }) + + return { + operations, + diagnostics: { + parseStatus: parsed.parseStatus, + repaired: parsed.repaired, + rawCount: parsed.items.length, + normalizedCount: operations.length + } + } + } + + async extractGroupOperations(args) { + const result = await this.extractGroupOperationResult(args) + return result.operations } } diff --git a/utils/memory/MemoryRetriever.js b/utils/memory/MemoryRetriever.js index 79bab52..fe937f4 100644 --- a/utils/memory/MemoryRetriever.js +++ b/utils/memory/MemoryRetriever.js @@ -30,8 +30,8 @@ export class MemoryRetriever { } async retrieve({ groupId, userId = null, scope = "user", query = "", limit = 10 }) { - let meta = await this.store.getMeta(scope, groupId, userId) - if (meta.disabled) return { meta, facts: [] } + const meta = await this.store.getMeta(scope, groupId, userId) + if (meta.disabled || limit <= 0) return { meta, facts: [] } const facts = await this.store.getFacts(meta, false) let queryEmbedding = null @@ -41,8 +41,12 @@ export class MemoryRetriever { queryEmbedding = result.embedding } - const scored = facts.map(fact => { - const semantic = queryEmbedding && fact.embedding ? cosineSimilarity(queryEmbedding, fact.embedding) : null + let scored = facts.map(fact => { + const embeddingIsCurrent = fact.embeddingHash && + fact.embeddingHash === this.extractor.embeddingHashFor(fact.content) + const semantic = queryEmbedding && embeddingIsCurrent && fact.embedding + ? cosineSimilarity(queryEmbedding, fact.embedding) + : null const relevance = semantic ?? this.keywordRelevance(query, fact.content) const recency = this.recencyScore(fact) const score = @@ -54,18 +58,14 @@ export class MemoryRetriever { return { ...fact, relevance, recency, score } }) + if (queryEmbedding) { + scored = scored + .sort((a, b) => b.relevance - a.relevance) + .slice(0, this.config.semanticRecallTopK) + } scored.sort((a, b) => b.score - a.score) const selected = scored.slice(0, limit) - // 并行更新 lastUsed，避免串行 N 次 Redis 写入 - await Promise.all(selected.map(fact => { - fact.lastUsed = now() - const scopeId = fact.scope === "user" - ? this.store.userScopeId(fact.groupId, fact.userId) - : this.store.groupScopeId(fact.groupId) - return this.store.setJson(this.store.factKey(fact.scope, scopeId, fact.id), fact) - })) - return { meta, facts: selected } } } diff --git a/utils/memory/MemoryStore.js b/utils/memory/MemoryStore.js index d1c3ae0..2531cb9 100644 --- a/utils/memory/MemoryStore.js +++ b/utils/memory/MemoryStore.js @@ -1,8 +1,16 @@ // redis 事实存储层：作用域 key 规划、meta/fact CRUD、容量裁剪、旧版数据迁移。 // 从 MemoryManager.js 拆出（行为等价搬迁）。 import { randomUUID } from "crypto" -import { USER_CATEGORIES, GROUP_CATEGORIES, LEGACY_MEMORY_ROLLBACK_DAYS } from "./constants.js" -import { now, clamp, uniq, safeJsonParse, compactText, containsToolFeedback } from "./helpers.js" +import { + USER_CATEGORIES, + GROUP_CATEGORIES, + LEGACY_MEMORY_ROLLBACK_DAYS, + DELETED_MEMORY_RETENTION_DAYS, + MAX_DELETED_FACTS_PER_SCOPE, + USER_CANDIDATE_TTL_DAYS, + MAX_USER_CANDIDATES +} from "./constants.js" +import { now, clamp, uniq, sha256, safeJsonParse, compactText, containsToolFeedback } from "./helpers.js" export class MemoryStore { constructor(config) { @@ -38,10 +46,15 @@ export class MemoryStore { return `${this.v2Prefix}fact:${scope}:${scopeId}:${factId}` } + userCandidateKey(groupId, userId) { + return `${this.v2Prefix}candidate:user:${groupId}:${userId}` + } + async setRaw(key, value, ttlSeconds = null) { if (ttlSeconds) { try { await redis.set(key, value, { EX: ttlSeconds }) + if (typeof redis.expire === "function") await redis.expire(key, ttlSeconds) return } catch { try { @@ -99,9 +112,7 @@ export class MemoryStore { } async deleteKeys(keys = []) { - for (const key of keys.filter(Boolean)) { - await redis.del(key) - } + await Promise.all(uniq(keys).map(key => redis.del(key))) } createMeta(scope, groupId, userId = null) { @@ -111,6 +122,7 @@ export class MemoryStore { groupId: String(groupId), userId: userId === null || userId === undefined ? null : String(userId), factIds: [], + deletedFactIds: [], disabled: false, createdAt: timestamp, updatedAt: timestamp, @@ -136,6 +148,7 @@ export class MemoryStore { merged.groupId = String(groupId) merged.userId = userId === null || userId === undefined ? null : String(userId) merged.factIds = uniq(Array.isArray(merged.factIds) ? merged.factIds : []) + merged.deletedFactIds = uniq(Array.isArray(merged.deletedFactIds) ? merged.deletedFactIds : []) merged.disabled = Boolean(merged.disabled) merged.updatedAt = Number(merged.updatedAt) || now() merged.createdAt = Number(merged.createdAt) || merged.updatedAt @@ -226,7 +239,10 @@ export class MemoryStore { } async getFacts(meta, includeDeleted = false) { - const factResults = await Promise.all((meta.factIds || []).map(factId => this.getFactForMeta(meta, factId))) + const factIds = includeDeleted + ? uniq([...(meta.factIds || []), ...(meta.deletedFactIds || [])]) + : (meta.factIds || []) + const factResults = await Promise.all(factIds.map(factId => this.getFactForMeta(meta, factId))) const facts = [] for (const fact of factResults) { if (!fact) continue @@ -245,6 +261,7 @@ export class MemoryStore { if (!meta.factIds.includes(normalized.id)) { meta.factIds.push(normalized.id) } + meta.deletedFactIds = (meta.deletedFactIds || []).filter(id => id !== normalized.id) normalized.updatedAt = now() const scopeId = normalized.scope === "user" @@ -257,10 +274,56 @@ export class MemoryStore { return normalized } + normalizeUserCandidate(candidate, groupId, userId) { + const timestamp = now() + return { + id: String(candidate?.id || randomUUID()), + groupId: String(groupId), + userId: String(userId), + content: compactText(candidate?.content), + category: this.normalizeCategory("user", candidate?.category), + importance: clamp(candidate?.importance ?? 0.4, 0, 1), + confidence: clamp(candidate?.confidence ?? 0.5, 0, 1), + evidenceKeys: uniq(candidate?.evidenceKeys || []), + sourceMessageIds: uniq(candidate?.sourceMessageIds || []), + firstSeenAt: Number(candidate?.firstSeenAt) || timestamp, + lastSeenAt: Number(candidate?.lastSeenAt) || timestamp + } + } + + async getUserCandidates(groupId, userId) { + const raw = await this.getJson(this.userCandidateKey(groupId, userId), []) + const cutoff = now() - USER_CANDIDATE_TTL_DAYS * 24 * 60 * 60 * 1000 + return (Array.isArray(raw) ? raw : []) + .map(candidate => this.normalizeUserCandidate(candidate, groupId, userId)) + .filter(candidate => candidate.content && candidate.lastSeenAt >= cutoff) + } + + async saveUserCandidates(groupId, userId, candidates = []) { + const normalized = candidates + .map(candidate => this.normalizeUserCandidate(candidate, groupId, userId)) + .filter(candidate => candidate.content) + .sort((a, b) => b.lastSeenAt - a.lastSeenAt) + .slice(0, MAX_USER_CANDIDATES) + + const key = this.userCandidateKey(groupId, userId) + if (!normalized.length) { + await redis.del(key) + return [] + } + + const ttlSeconds = USER_CANDIDATE_TTL_DAYS * 24 * 60 * 60 + await this.setJson(key, normalized, ttlSeconds) + return normalized + } + + async clearUserCandidates(groupId, userId) { + await redis.del(this.userCandidateKey(groupId, userId)) + } + async deleteFact(meta, factId) { const fact = await this.getFactForMeta(meta, factId) meta.factIds = meta.factIds.filter(id => id !== factId) - await this.saveMeta(meta) if (fact) { fact.status = "deleted" @@ -268,9 +331,24 @@ export class MemoryStore { const scopeId = fact.scope === "user" ? this.userScopeId(fact.groupId, fact.userId) : this.groupScopeId(fact.groupId) - await this.setJson(this.factKey(fact.scope, scopeId, fact.id), fact) + const ttlSeconds = DELETED_MEMORY_RETENTION_DAYS * 24 * 60 * 60 + await this.setJson(this.factKey(fact.scope, scopeId, fact.id), fact, ttlSeconds) + meta.deletedFactIds = uniq([...(meta.deletedFactIds || []), fact.id]) + } else { + meta.deletedFactIds = (meta.deletedFactIds || []).filter(id => id !== factId) } + if (meta.deletedFactIds.length > MAX_DELETED_FACTS_PER_SCOPE) { + const expiredIds = meta.deletedFactIds.slice(0, meta.deletedFactIds.length - MAX_DELETED_FACTS_PER_SCOPE) + const scopeId = meta.scope === "user" + ? this.userScopeId(meta.groupId, meta.userId) + : this.groupScopeId(meta.groupId) + await this.deleteKeys(expiredIds.map(id => this.factKey(meta.scope, scopeId, id))) + meta.deletedFactIds = meta.deletedFactIds.slice(-MAX_DELETED_FACTS_PER_SCOPE) + } + + await this.saveMeta(meta) + return Boolean(fact) } @@ -279,6 +357,10 @@ export class MemoryStore { if ((meta.factIds || []).length <= maxFacts) return const facts = await this.getFacts(meta, false) + const activeIds = new Set(facts.map(fact => fact.id)) + meta.factIds = meta.factIds.filter(id => activeIds.has(id)) + if (meta.factIds.length <= maxFacts) return + facts.sort((a, b) => { if (a.importance !== b.importance) return a.importance - b.importance return (a.lastUsed || a.updatedAt) - (b.lastUsed || b.updatedAt) @@ -288,15 +370,10 @@ export class MemoryStore { const removeIds = new Set(facts.slice(0, removeCount).map(f => f.id)) meta.factIds = meta.factIds.filter(id => !removeIds.has(id)) - // 并行标记删除，避免串行 Redis 写入 - await Promise.all(facts.filter(f => removeIds.has(f.id)).map(fact => { - fact.status = "deleted" - fact.updatedAt = now() - const scopeId = fact.scope === "user" - ? this.userScopeId(fact.groupId, fact.userId) - : this.groupScopeId(fact.groupId) - return this.setJson(this.factKey(fact.scope, scopeId, fact.id), fact) - })) + const scopeId = meta.scope === "user" + ? this.userScopeId(meta.groupId, meta.userId) + : this.groupScopeId(meta.groupId) + await this.deleteKeys([...removeIds].map(id => this.factKey(meta.scope, scopeId, id))) } factFromLegacy(raw, scope, groupId, userId, category) { @@ -305,7 +382,7 @@ export class MemoryStore { if (!content || containsToolFeedback(content)) return null return this.normalizeFact({ - id: data.id || randomUUID(), + id: data.id || sha256(`legacy:${scope}:${groupId}:${userId || ""}:${category}:${content}`), scope, groupId, userId, @@ -415,12 +492,16 @@ export class MemoryStore { async clearScope(scope, groupId, userId = null) { const meta = await this.getMeta(scope, groupId, userId) const scopeId = scope === "user" ? this.userScopeId(groupId, userId) : this.groupScopeId(groupId) - const factKeys = meta.factIds.map(id => this.factKey(scope, scopeId, id)) + const indexedFactKeys = uniq([...(meta.factIds || []), ...(meta.deletedFactIds || [])]) + .map(id => this.factKey(scope, scopeId, id)) + const scannedFactKeys = await this.scanKeys(this.factKey(scope, scopeId, "*")) + const factKeys = uniq([...indexedFactKeys, ...scannedFactKeys]) await this.deleteKeys(factKeys) await redis.del(this.metaKey(scope, groupId, userId)) if (scope === "user") { await redis.del(this.legacyUserKey(groupId, userId)) + await this.clearUserCandidates(groupId, userId) } else { await redis.del(this.legacyGroupKey(groupId)) } diff --git a/utils/memory/constants.js b/utils/memory/constants.js index 22fbb5a..8252441 100644 --- a/utils/memory/constants.js +++ b/utils/memory/constants.js @@ -26,7 +26,7 @@ export const DEFAULT_CONFIG = { maxFactsPerGroup: 50, importanceThreshold: 0.5, memoryDecayDays: 7, - userExtractDebounceSeconds: 90, + userExtractDebounceSeconds: 45, userExtractMaxBatchMessages: 6, groupExtractMinIntervalMinutes: 10, groupExtractMaxBatchMessages: 12, @@ -41,6 +41,14 @@ export const DEFAULT_CONFIG = { } export const LEGACY_MEMORY_ROLLBACK_DAYS = 30 +export const DELETED_MEMORY_RETENTION_DAYS = 30 +export const MAX_DELETED_FACTS_PER_SCOPE = 200 + +// 推测性用户事实不会直接进入长期记忆。它们先短期保留，跨两个独立 +// 抽取批次重复出现后再晋升，避免把单次闲聊误判成稳定偏好或习惯。 +export const USER_CANDIDATE_TTL_DAYS = 7 +export const USER_CANDIDATE_PROMOTION_COUNT = 2 +export const MAX_USER_CANDIDATES = 30 export const TOOL_FEEDBACK_MARKERS = [ "[tool_request]", diff --git a/utils/memory/helpers.js b/utils/memory/helpers.js index bc804a5..37ef8a9 100644 --- a/utils/memory/helpers.js +++ b/utils/memory/helpers.js @@ -47,6 +47,10 @@ export function containsToolFeedback(content) { return TOOL_FEEDBACK_MARKERS.some(marker => text.includes(marker)) } +export function isLikelyBotCommand(content) { + return /^[#＃/／]\s*\S/.test(String(content || "").trim()) +} + export function isRealUserSource(source) { return source === undefined || source === null || source === "" || source === "user" || source === "message" } @@ -74,22 +78,38 @@ export function isSimilarContent(a, b) { return similarity >= 0.72 || (Math.min(na.length, nb.length) >= 6 && similarity >= 0.6) } -export function extractJsonArray(content) { +function normalizeJsonArrayValue(value) { + if (Array.isArray(value)) return value + if (!value || typeof value !== "object") return null + for (const key of ["operations", "items", "data", "results"]) { + if (Array.isArray(value[key])) return value[key] + } + return [value] +} + +export function parseJsonArrayResult(content) { const text = String(content || "").trim() + if (!text) return { items: [], status: "empty" } + // 找到第一个 [ 和最后一个 ]，截取中间部分尝试解析 // 这样可以容忍 LLM 在 JSON 前后追加 markdown 围栏、解释文字等 const start = text.indexOf("[") const end = text.lastIndexOf("]") if (start >= 0 && end > start) { const parsed = safeJsonParse(text.slice(start, end + 1), null) - if (Array.isArray(parsed)) return parsed - if (parsed && typeof parsed === "object") return [parsed] + const items = normalizeJsonArrayValue(parsed) + if (items) return { items, status: items.length ? "ok" : "empty" } } + // 兜底：尝试解析整个文本 - const parsed = safeJsonParse(text, []) - if (Array.isArray(parsed)) return parsed - if (parsed && typeof parsed === "object") return [parsed] - return [] + const parsed = safeJsonParse(text, null) + const items = normalizeJsonArrayValue(parsed) + if (items) return { items, status: items.length ? "ok" : "empty" } + return { items: [], status: "invalid" } +} + +export function extractJsonArray(content) { + return parseJsonArrayResult(content).items } export function keywordSet(text) { @@ -137,14 +157,27 @@ export function normalizeConfig(config = {}) { merged.maxFactsPerUser = Math.max(1, Number(merged.maxFactsPerUser) || DEFAULT_CONFIG.maxFactsPerUser) merged.maxFactsPerGroup = Math.max(1, Number(merged.maxFactsPerGroup) || DEFAULT_CONFIG.maxFactsPerGroup) merged.memoryDecayDays = Math.max(1, Number(merged.memoryDecayDays) || DEFAULT_CONFIG.memoryDecayDays) - merged.userExtractDebounceSeconds = Math.max(1, Number(merged.userExtractDebounceSeconds) || DEFAULT_CONFIG.userExtractDebounceSeconds) + const userDebounce = Number(merged.userExtractDebounceSeconds) + merged.userExtractDebounceSeconds = Number.isFinite(userDebounce) + ? Math.max(0, userDebounce) + : DEFAULT_CONFIG.userExtractDebounceSeconds merged.userExtractMaxBatchMessages = Math.max(1, Number(merged.userExtractMaxBatchMessages) || DEFAULT_CONFIG.userExtractMaxBatchMessages) merged.groupExtractMinIntervalMinutes = Math.max(1, Number(merged.groupExtractMinIntervalMinutes) || DEFAULT_CONFIG.groupExtractMinIntervalMinutes) merged.groupExtractMaxBatchMessages = Math.max(1, Number(merged.groupExtractMaxBatchMessages) || DEFAULT_CONFIG.groupExtractMaxBatchMessages) - merged.promptMaxUserFacts = Math.max(1, Number(merged.promptMaxUserFacts) || DEFAULT_CONFIG.promptMaxUserFacts) - merged.promptMaxGroupFacts = Math.max(1, Number(merged.promptMaxGroupFacts) || DEFAULT_CONFIG.promptMaxGroupFacts) - merged.promptMaxChars = Math.max(200, Number(merged.promptMaxChars) || DEFAULT_CONFIG.promptMaxChars) + const promptMaxUserFacts = Number(merged.promptMaxUserFacts) + const promptMaxGroupFacts = Number(merged.promptMaxGroupFacts) + merged.promptMaxUserFacts = Number.isFinite(promptMaxUserFacts) + ? Math.max(0, promptMaxUserFacts) + : DEFAULT_CONFIG.promptMaxUserFacts + merged.promptMaxGroupFacts = Number.isFinite(promptMaxGroupFacts) + ? Math.max(0, promptMaxGroupFacts) + : DEFAULT_CONFIG.promptMaxGroupFacts + merged.promptMaxChars = Math.max(100, Number(merged.promptMaxChars) || DEFAULT_CONFIG.promptMaxChars) merged.semanticRecallTopK = Math.max(1, Number(merged.semanticRecallTopK) || DEFAULT_CONFIG.semanticRecallTopK) + const minFactsPerCategory = Number(merged.minFactsPerCategory) + merged.minFactsPerCategory = Number.isFinite(minFactsPerCategory) + ? Math.max(0, minFactsPerCategory) + : DEFAULT_CONFIG.minFactsPerCategory return merged }