diff --git a/README.md b/README.md index fbf1c23..cd95ab5 100644 --- a/README.md +++ b/README.md @@ -108,6 +108,53 @@ Anthropic direct (Claude) — drop `baseURL`, switch `provider`: `embedding` is optional. When present, `dimensions` must match the Neo4j vector index dimension. For a fresh database, the plugin creates matching indexes during startup. If you change dimensions later, recreate the vector indexes or the Neo4j database. +### Memory decay (forgetting curve) + +Each maintenance cycle scores every active node with a three-factor weighted model (recency + frequency + intrinsic) and bidirectionally transitions nodes across three tiers: `core` / `working` / `peripheral`. Nodes never get `status=deprecated` from decay — only manual deprecate / merge does that. Decay only adjusts `tier`, so all active nodes remain searchable. + +The full formula, field mapping from the reference implementation, default-value rationale, and tuning guide live in **[`docs/decay.md`](docs/decay.md)**. + +Minimal config (all fields optional, defaults shown): + +```json +"decay": { "enabled": true } +``` + +Common overrides — for fuller control see `docs/decay.md` §4: + +```json +"decay": { + "enabled": true, + "recencyHalfLifeDays": 30, + "peripheralCompositeThreshold": 0.15, + "workingAccessThreshold": 3 +} +``` + +### Cron sessions + +Sessions created by OpenClaw scheduled tasks can be configured independently of normal sessions. The host places the cron marker on the **sessionKey** (`sessionId` is a random UUID); real shapes are `cron:`, `agent::cron:`, or `agent::cron::run:`: + +```json +"cron": { + "enabled": true, + "extract": true, + "finalizeAndMaintain": true +} +``` + +| Option | Default | Description | +| --- | --- | --- | +| `enabled` | `true` | Enable graph functionality inside cron sessions (recall injection + message buffering). When `false`, cron sessions skip automatic recall and message persistence; the `gm_*` tools remain available for explicit calls (manual escape hatch). | +| `extract` | `true` | Trigger knowledge extraction (LLM triples) in cron sessions via `afterTurn` / `compact`. When `false`, messages are still buffered and can be backfilled later with `openclaw graph-memory extract`. | +| `finalizeAndMaintain` | `true` | Run finalize (EVENT→SKILL promotion) and graph maintenance (decay / PageRank / communities) when a cron session ends. Disable when frequent cron runs make end-of-session global maintenance too costly. | + +All three options default to **`true`**: cron sessions behave like normal sessions (recall, buffering, extraction, and end-of-session maintenance all enabled) unless explicitly disabled. `enabled: false` is the master switch — even with `extract` / `finalizeAndMaintain` set to `true`, nothing runs. Non-cron sessions are never affected by these options. + +All three sub-options are optional; omitted fields keep the default `true` (e.g. with `"cron": { "extract": false }` only extraction is disabled — recall, buffering, and end-of-session maintenance stay on). + +Caveat: when a cron job sets an explicit custom `sessionKey`, the host does not append the `cron` segment — such sessions cannot be detected and are treated as normal sessions. + ### OAuth login (experimental) ```bash @@ -127,7 +174,7 @@ conversation messages -> GmMessage nodes -> LLM triple extraction -> embeddings -> vector recall + community expansion + GDS PPR -> XML context injection -session end -> dedup -> global PageRank -> communities -> summaries +session end -> decay (forgetting curve) -> dedup -> global PageRank -> communities -> summaries ``` ## Verify diff --git a/README_CN.md b/README_CN.md index eabc330..a67305b 100644 --- a/README_CN.md +++ b/README_CN.md @@ -81,6 +81,28 @@ bash setup-graph-memory-pro.sh --uninstall `embedding` 可选。设置时，`dimensions` 必须与 Neo4j 向量索引维度一致。新数据库会在插件启动时按配置创建索引；更换维度后需要重建向量索引或 Neo4j 数据库。 +### cron 会话行为控制 + +OpenClaw 定时任务创建的会话可以独立配置图谱行为。host 把 cron 标记放在 **sessionKey** 上（`sessionId` 是随机 UUID），实际形状为 `cron:`、`agent::cron:` 或 `agent::cron::run:`： + +```json +"cron": { + "enabled": true, + "extract": true, + "finalizeAndMaintain": true +} +``` + +| 选项 | 默认 | 说明 | +| --- | --- | --- | +| `enabled` | `true` | 是否在 cron 会话内启用图谱功能（召回注入 + 消息入库）。关闭后 cron 会话不自动召回、不自动入库；`gm_*` 工具仍可手动调用（作为显式逃生通道）。 | +| `extract` | `true` | 是否在 cron 会话内触发知识提取（afterTurn / compact 的 LLM 三元组提取）。关闭后消息仍入库缓冲，之后可用 `openclaw graph-memory extract` 手动回填。 | +| `finalizeAndMaintain` | `true` | cron 会话结束时是否执行 finalize（EVENT→SKILL 晋升）和图维护（decay / PageRank / 社区检测）。定时任务频繁时可关闭，避免每次会话结束都跑全局维护。 | + +三个选项**默认全部开启**：cron 会话默认使用图谱，需按需显式关闭。`enabled=false` 是总开关：即使 `extract`/`finalizeAndMaintain` 设为 `true` 也不生效。非 cron 会话不受这些选项影响。三个子项均可省略，未写的字段取默认值 `true`。 + +注意：若 cron 任务显式设置了自定义 `sessionKey`，host 不再附加 `cron` 段，此类会话无法被识别，将按普通会话处理。 + ### OAuth 登录（实验性） ```bash diff --git a/docs/decay.md b/docs/decay.md new file mode 100644 index 0000000..ba57d5f --- /dev/null +++ b/docs/decay.md @@ -0,0 +1,175 @@ +# Memory Decay — 柔性评分模型 + +graph-memory-pro 的衰减机制采用**三因子加权评分 + tier 双向转换**，参考 [memory-lancedb-pro](https://github.com/CortexReach/memory-lancedb-pro) 的设计并映射到本仓库的图模型信号。 + +- **decay 不动 `status`**——只调整 `tier`（`core` / `working` / `peripheral`）。`status=deprecated` 仅由手动弃用（`gm_update mode=deprecate` / merge）触发。 +- 每次 `gm_maintain` 或 `session_end` 维护的第 0 步执行：扫描所有 active 节点 → 评分 → tier 转换 → 写回 `decayScore` / `tier` / `decayComputedAt`。 +- 评分结果可通过 `gm_stats` / CRUD API 查看；外层搜索目前**不读 decayScore 排序**（已由 PageRank + tier 隐含分层）。 + +--- + +## 1. 评分公式 + +``` +composite = wR · recency + wF · frequency + wI · intrinsic +``` + +三个权重默认 `0.4 / 0.3 / 0.3`，**推荐**和为 1。运行时若和≠1 会自动按比例归一化（`wR' = wR / (wR+wF+wI)`），保证 `composite ∈ [0,1]`，避免用户覆盖单个权重导致评分越界。归一化在 `scoreNode()` 内进行，原始 `cfg.*Weight` 值不被修改。 + +### 1.1 Recency（时间衰减，权重 0.4） + +Weibull 拉伸指数： + +``` +recency = exp( −λ · daysSinceLastAccess^β ) + +λ = ln(2) / effectiveHL +effectiveHL = recencyHalfLifeDays · exp( importanceModulation · importance ) +``` + +- **半衰期调制**：重要记忆（高 `importance`）的 `effectiveHL` 更大 → 衰减更慢。对应艾宾浩斯曲线"重要事件保留更久"。 +- **tier-β**：曲线形状随 tier 变化，反馈式调整衰减速度： + + | tier | β | 效果 | + |---|---|---| + | `core` | 0.8 | 尾部衰减缓（核心知识保得久） | + | `working` | 1.0 | 标准指数衰减 | + | `peripheral` | 1.3 | 加速衰减（边缘知识更快被遗忘） | + +### 1.2 Frequency（访问频率，权重 0.3） + +``` +frequency = base · ( 0.5 + 0.5 · recentnessBonus ) + +base = 1 − exp( −validatedCount / 5 ) +recentnessBonus = exp( −avgAccessGapDays / 30 ) # 仅当 validatedCount > 1 +avgAccessGapDays = ( lastAccessedAt − createdAt ) / ( validatedCount − 1 ) +``` + +- 用 `validatedCount`（LLM 重新提取的次数）替代 lancedb-pro 的 `accessCount`（manual recall 触发的次数）。前者是更强的"重新确认"信号。 +- `validatedCount ≤ 1` 时跳过 `recentnessBonus`，只返回 `base`（无法算平均间隔）。 + +### 1.3 Intrinsic（内在价值，权重 0.3） + +``` +intrinsic = importance · confidence + +importance = pagerank / maxPagerank # 每次扫描时按当前批次归一化到 [0,1] +confidence = 1 − 1 / ( 1 + validatedCount ) # 饱和函数，收敛到 1 +``` + +--- + +## 2. 字段映射（lancedb-pro → graph-memory-pro） + +| lancedb-pro 字段 | 本仓库替代 | 说明 | +|---|---|---| +| `accessCount` | `validatedCount` | LLM 重新提取次数（强信号，原为 manual recall 触发） | +| `lastAccessedAt` | `lastAccessedAt` | 由 `upsertNode` 在任意写入路径刷新（重新提取、`gm_record`、`gm_update`、CRUD POST）。`mergeNodes` 故意不刷新（合并 ≠ 用户重新激活） | +| `importance` | `pagerank / maxPagerank` | 图结构重要性，每次扫描归一化 | +| `confidence` | `1 − 1/(1+validatedCount)` | 饱和置信度 | +| `tier` | `tier`（新增字段） | 与 `status` 正交 | + +--- + +## 3. Tier 双向转换 + +| 转换 | 条件 | +|---|---| +| **core → working** | `composite < peripheralCompositeThreshold` **AND** `count < workingAccessThreshold` | +| **working → peripheral** | `composite < peripheralCompositeThreshold` **OR**（`ageDays > peripheralAgeDays` **AND** `count < workingAccessThreshold`） | +| **peripheral → working** | `count >= workingAccessThreshold` **AND** `composite >= workingCompositeThreshold` | +| **working → core** | `count >= coreAccessThreshold` **AND** `composite >= coreCompositeThreshold` **AND** `importance >= coreImportanceThreshold` | + +- 新节点默认 `tier = "working"`。 +- 节点保持 `status = active` 不变；tier 变化时仅更新 `updatedAt`，不改变搜索过滤行为。 +- 不存在的"core→peripheral"和"peripheral→core"由两次相邻转换实现（经过 working）。 + +--- + +## 4. 默认值与调参指南 + +### 4.1 默认配置 + +```json +{ + "decay": { + "enabled": true, + "recencyHalfLifeDays": 30, + "recencyWeight": 0.4, + "importanceModulation": 1.5, + "frequencyWeight": 0.3, + "intrinsicWeight": 0.3, + "betaCore": 0.8, + "betaWorking": 1.0, + "betaPeripheral": 1.3, + "coreAccessThreshold": 10, + "coreCompositeThreshold": 0.7, + "coreImportanceThreshold": 0.8, + "peripheralCompositeThreshold": 0.15, + "peripheralAgeDays": 60, + "workingAccessThreshold": 3, + "workingCompositeThreshold": 0.4 + } +} +``` + +### 4.2 数值来源 + +| 参数 | 默认值 | 来源 | +|---|---|---| +| `recencyHalfLifeDays` | 30 | 艾宾浩斯曲线 ~25% 保留率拐点；同时与 lancedb-pro 的 `recencyHalfLifeDays` + `ACCESS_DECAY_HALF_LIFE_DAYS` 一致 | +| `importanceModulation` | 1.5 | lancedb-pro：`effectiveHL = 30 · exp(1.5 · importance)`，importance=1 时半衰期延长到 ~134 天 | +| `betaCore/Working/Peripheral` | 0.8 / 1.0 / 1.3 | lancedb-pro Weibull 形状参数 | +| 7 个 tier 转换阈值 | — | lancedb-pro `tier-manager` 默认值 | +| `recencyWeight / frequencyWeight / intrinsicWeight` | 0.4 / 0.3 / 0.3 | lancedb-pro 三因子权重，和为 1 | +| `validatedCount` 分母 | 5 | lancedb-pro 的 `1 − exp(−count/5)` 基础频率项（未改） | + +### 4.3 常见调参场景 + +| 想要的效果 | 调整方向 | +|---|---| +| 记忆整体保留更久 | 调高 `recencyHalfLifeDays`（如 60）或调低 `peripheralCompositeThreshold`（更难降级） | +| 更激进遗忘 | 调低 `recencyHalfLifeDays`（如 14）或调高 `peripheralCompositeThreshold` | +| 重要知识显著保得久 | 调高 `importanceModulation`（半衰期调制更强） | +| 核心知识不易降级 | 调低 `betaCore`（更缓的尾部）或调高 `coreCompositeThreshold`（更难升 core，留在 working 也保得久） | +| 单次曝光更易遗忘 | 调高 `workingAccessThreshold`（promote 到 working 需要更多确认） | +| 永久禁用衰减 | `"enabled": false` | + +### 4.4 与原布尔阈值方案的对照（向后兼容） + +旧版本（`maxAgeDays` + `minCalls`）的布尔规则已被这套柔性评分取代。原默认值 `maxAgeDays=30, minCalls=2` 在新模型下大致对应于： + +- 一个 `validatedCount=1`、`tier=working`、低 pagerank 的节点，约 30 天后 `recency` 跌破 0.15 → `composite` 跌破 `peripheralCompositeThreshold` → demote 到 `peripheral`。 +- 关键差别：新模型**不会 deprecate**，只是降到 `peripheral` tier，搜索过滤仍包含它（只是 decayScore 较低）。 + +--- + +## 5. 数据库字段 + +| 字段 | 类型 | 写入者 | 说明 | +|---|---|---|---| +| `tier` | string | `applyDecay` / `upsertNode`（创建时初始化为 `working`） | `core` / `working` / `peripheral` | +| `lastAccessedAt` | int (epoch ms) | `upsertNode`（重新提取时） | decay 评分的时间基准 | +| `decayScore` | float (0~1) | `applyDecay` | 最近一次评分结果 | +| `decayComputedAt` | int (epoch ms) | `applyDecay` | 评分时间戳 | + +旧节点缺这些字段时： +- `tier` 缺失 → 评分按 `working` 处理；首次 `applyDecay` 时自动写入 `working` +- `lastAccessedAt` 缺失 → 回退到 `updatedAt` / `createdAt` +- `decayScore` / `decayComputedAt` 缺失 → 在首次 `applyDecay` 前为 undefined，不影响评分 + +**Backfill 时机**：新字段在第一次 `applyDecay` 运行时为每个 active 节点批量写入。如果部署初始用 `decay.enabled=false`，字段会一直缺失直到切换为 `true` 后的第一次维护周期。在切换前的窗口期，对 raw DB 直接做 `tier` 过滤查询会返回 null/missing 而非 `"working"`——目前搜索路径不读 `tier`，但自定义查询需要留意。 + +--- + +## 6. 实现位置 + +| 文件 | 内容 | +|---|---| +| `src/graph/decay.ts` | 评分函数 + tier 决策 + `applyDecay()` 批处理 | +| `src/types.ts` | `DecayConfig` 接口、`NodeTier` 类型、`GmNode` 新字段、`DEFAULT_CONFIG.decay` | +| `src/store/store.ts` | `toNode` 字段映射、`upsertNode` 初始化 `tier` / `lastAccessedAt` | +| `src/graph/maintenance.ts` | 调用入口（step 0） | +| `test/decay.test.ts` | 评分函数 + tier 决策纯函数单元测试 | +| `openclaw.plugin.json` | 用户可见的配置 schema | diff --git a/index.ts b/index.ts index a5f3c9b..9973e7b 100755 --- a/index.ts +++ b/index.ts @@ -9,7 +9,7 @@ import type { OpenClawPluginApi } from "openclaw/plugin-sdk"; import { Type } from "@sinclair/typebox"; import { getDriver, initSchema, getSession } from "./src/store/db.ts"; import { - saveMessage, getUnextracted, + saveMessage, getUnextracted, getMaxTurnIndex, markExtracted, isTurnExtracted, upsertNode, upsertEdge, findByName, updateNode, deleteNode, deprecateNodeAndDisconnect, @@ -24,7 +24,7 @@ import { Extractor } from "./src/extractor/extract.ts"; import { assembleContext } from "./src/format/assemble.ts"; import { sanitizeToolUseResultPairing } from "./src/format/transcript-repair.ts"; import { runMaintenance } from "./src/graph/maintenance.ts"; -import { DEFAULT_CONFIG, type GmConfig, type RecallResult, type EdgeType } from "./src/types.ts"; +import { DEFAULT_CONFIG, DEFAULT_CRON_CONFIG, isCronSessionKey, type GmConfig, type RecallResult, type EdgeType } from "./src/types.ts"; import { registerCrudRoutes } from "./src/routes/crud.ts"; import { createGraphMemoryCli } from "./src/cli.ts"; @@ -264,6 +264,7 @@ const graphMemoryProPlugin = { pluginId: "graph-memory-pro", pluginConfig: raw as Record | undefined, resolveConfigPath: (p: string) => api.resolvePath?.(p) ?? p, + defaultModel: readDefaultModel(api.config), }), { commands: ["graph-memory"] }, ); @@ -277,6 +278,9 @@ const graphMemoryProPlugin = { const cfg: GmConfig = { ...DEFAULT_CONFIG, ...raw }; if (raw.neo4j) cfg.neo4j = { ...DEFAULT_CONFIG.neo4j, ...raw.neo4j }; + if (raw.decay) cfg.decay = { ...DEFAULT_CONFIG.decay, ...raw.decay }; + if (raw.cron) cfg.cron = { ...DEFAULT_CONFIG.cron, ...raw.cron }; + const cronCfg = cfg.cron ?? DEFAULT_CRON_CONFIG; const providerModel = readDefaultModel(api.config); @@ -350,13 +354,30 @@ const graphMemoryProPlugin = { /** * 每轮结束后直接从原始消息提取知识图谱 * 一轮 = 用户发一条消息 → agent 不管调了多少工具 → 最终回复用户 + * + * compact() 与本函数对同一 session 存在 TOCTOU 竞争：两条路径都先 + * isTurnExtracted/getUnextracted → 调 LLM → 最后 markExtracted，中间窗口 + * 允许另一条路径重复提取同一批消息（重复 LLM 调用 + validatedCount 双递增）。 + * 用 per-session async 互斥锁串行化两条路径的提取体。 */ + const extractLocks = new Map>(); + function withExtractLock(sessionId: string, fn: () => Promise): Promise { + const prev = extractLocks.get(sessionId) ?? Promise.resolve(); + const chain = prev.catch(() => {}); + const result = chain.then(() => fn()); + // 链上只保留"上一轮是否结束"的状态，丢弃返回值并吞掉错误， + // 否则一次失败会永久污染链 → 后续 acquire 直接 reject。 + extractLocks.set(sessionId, result.then(() => undefined, () => undefined)); + return result; + } + async function extractTurnKnowledge(sessionId: string, turnNum: number, rawMessages: any[]): Promise { - try { - if (await isTurnExtracted(driver, sessionId, turnNum)) { - api.logger.info(`[graph-memory-pro] turn ${turnNum}: already extracted (compact), skipping`); - return; - } + return withExtractLock(sessionId, async () => { + try { + if (await isTurnExtracted(driver, sessionId, turnNum)) { + api.logger.info(`[graph-memory-pro] turn ${turnNum}: already extracted (compact), skipping`); + return; + } const existing = (await getBySession(driver, sessionId)).map(n => n.name); const result = await extractor.extract({ messages: rawMessages, @@ -399,10 +420,12 @@ const graphMemoryProPlugin = { } catch (err) { api.logger.error(`[graph-memory-pro] turn ${turnNum} extract failed: ${err}`); } + }); } // ── Session 运行时状态 ────────────────────────────────── const msgSeq = new Map(); + const msgSeqLoaders = new Map>(); const recalled = new Map(); const sessionIdsByKey = new Map(); const pendingSubagentRecall = new Map(); @@ -419,6 +442,24 @@ const graphMemoryProPlugin = { } async function ingestMessage(sessionId: string, message: any): Promise { + if (!msgSeq.has(sessionId)) { + // 插件重启后内存 Map 会丢，必须从 DB 恢复 MAX(turnIndex)，否则下一条消息 + // turnIndex=1 → MERGE 命中旧行 → ON CREATE 被跳过 → 新消息静默丢失。 + // in-flight Promise 去重，避免并发 ingest 同时查询 + 互相覆盖 seq。 + let loader = msgSeqLoaders.get(sessionId); + if (!loader) { + loader = getMaxTurnIndex(driver, sessionId).then(max => { + msgSeq.set(sessionId, max); + msgSeqLoaders.delete(sessionId); + return max; + }).catch(err => { + msgSeqLoaders.delete(sessionId); + throw err; + }); + msgSeqLoaders.set(sessionId, loader); + } + await loader; + } const seq = (msgSeq.get(sessionId) ?? 0) + 1; msgSeq.set(sessionId, seq); await saveMessage(driver, sessionId, seq, message.role ?? "unknown", message); @@ -428,6 +469,9 @@ const graphMemoryProPlugin = { api.on("before_agent_start", async (event: any, ctx: any) => { try { + // cron session 关闭图谱功能时不召回（cron 标记在 sessionKey 上，sessionId 是随机 UUID） + if (isCronSessionKey(typeof ctx?.sessionKey === "string" ? ctx.sessionKey : null) && !cronCfg.enabled) return; + const rawPrompt = typeof event?.prompt === "string" ? event.prompt : ""; const prompt = cleanPrompt(rawPrompt); if (!prompt) return; @@ -467,6 +511,10 @@ const graphMemoryProPlugin = { async ingest({ sessionId, sessionKey, message, isHeartbeat }: { sessionId: string; sessionKey?: string; message: any; isHeartbeat?: boolean }) { if (isHeartbeat) return { ingested: false }; bindSessionIdentity(sessionId, sessionKey); + // cron session 关闭图谱功能：消息不入库 + if (isCronSessionKey(sessionKey) && !cronCfg.enabled) { + return { ingested: false }; + } await ingestMessage(sessionId, message); ingestedSinceTurn.set(sessionId, (ingestedSinceTurn.get(sessionId) ?? 0) + 1); return { ingested: true }; @@ -478,6 +526,21 @@ const graphMemoryProPlugin = { bindSessionIdentity(sessionId, sessionKey); const budget = tokenBudget ?? 128_000; + // cron session 关闭图谱功能：仅做消息裁剪与配对修复，不注入图谱上下文 + if (isCronSessionKey(sessionKey) && !cronCfg.enabled) { + const prepared = prepareAssemblyMessages(messages); + if (prepared.dropped > 0) { + api.logger.info( + `[graph-memory-pro] assemble: ${prepared.messages.length} msgs (~${prepared.tokens} tok), ` + + `dropped ${prepared.dropped} older msgs, graph skipped (cron session)`, + ); + } + return { + messages: prepared.messages, + estimatedTokens: prepared.tokens, + }; + } + const activeNodes = await getBySession(driver, sessionId); const activeEdges: any[] = []; for (const n of activeNodes) { @@ -546,51 +609,60 @@ const graphMemoryProPlugin = { async compact({ sessionId, sessionKey, currentTokenCount }: { sessionId: string; sessionKey?: string; sessionFile: string; tokenBudget?: number; force?: boolean; currentTokenCount?: number }) { bindSessionIdentity(sessionId, sessionKey); - const msgs = await getUnextracted(driver, sessionId, cfg.compactTurnCount * 3); + // cron session 关闭图谱功能或知识提取：不触发 LLM 提取 + if (isCronSessionKey(sessionKey) && !(cronCfg.enabled && cronCfg.extract)) { + return { + ok: true, compacted: false, + reason: cronCfg.enabled ? "cron session extraction disabled" : "cron session graph disabled", + }; + } + return withExtractLock(sessionId, async () => { + const msgs = await getUnextracted(driver, sessionId, cfg.compactTurnCount * 3); - if (!msgs.length) return { ok: true, compacted: false, reason: "no messages" }; + if (!msgs.length) return { ok: true, compacted: false, reason: "no messages" }; - try { - const existing = (await getBySession(driver, sessionId)).map(n => n.name); - const result = await extractor.extract({ messages: msgs, existingNames: existing }); - - const nameToId = new Map(); - for (const nc of result.nodes) { - const { node } = await upsertNode(driver, { - type: nc.type, name: nc.name, - description: nc.description, content: nc.content, - }, sessionId); - nameToId.set(node.name, node.id); - recaller.syncEmbed(node).catch(() => {}); - } + try { + const existing = (await getBySession(driver, sessionId)).map(n => n.name); + const result = await extractor.extract({ messages: msgs, existingNames: existing }); + + const nameToId = new Map(); + for (const nc of result.nodes) { + const { node } = await upsertNode(driver, { + type: nc.type, name: nc.name, + description: nc.description, content: nc.content, + }, sessionId); + nameToId.set(node.name, node.id); + recaller.syncEmbed(node).catch(() => {}); + } - for (const ec of result.edges) { - const fromNode = await findByName(driver, ec.from); - const toNode = await findByName(driver, ec.to); - const fromId = nameToId.get(ec.from) ?? fromNode?.id; - const toId = nameToId.get(ec.to) ?? toNode?.id; - if (fromId && toId) { - await upsertEdge(driver, { - fromId, toId, type: ec.type, - instruction: ec.instruction, condition: ec.condition, sessionId, - }); + for (const ec of result.edges) { + const fromNode = await findByName(driver, ec.from); + const toNode = await findByName(driver, ec.to); + const fromId = nameToId.get(ec.from) ?? fromNode?.id; + const toId = nameToId.get(ec.to) ?? toNode?.id; + if (fromId && toId) { + await upsertEdge(driver, { + fromId, toId, type: ec.type, + instruction: ec.instruction, condition: ec.condition, sessionId, + }); + } } - } - const maxTurn = Math.max(...msgs.map((m: any) => m.turn_index)); - await markExtracted(driver, sessionId, maxTurn); + const maxTurn = Math.max(...msgs.map((m: any) => m.turn_index)); + await markExtracted(driver, sessionId, maxTurn); - return { - ok: true, compacted: true, - result: { - summary: `extracted ${result.nodes.length} nodes, ${result.edges.length} edges`, - tokensBefore: currentTokenCount ?? 0, - }, - }; - } catch (err) { - api.logger.error(`[graph-memory-pro] compact failed: ${err}`); - return { ok: false, compacted: false, reason: String(err) }; - } + return { + ok: true, compacted: true, + result: { + summary: `extracted ${result.nodes.length} nodes, ${result.edges.length} edges`, + tokensBefore: currentTokenCount ?? 0, + }, + }; + } catch (err) { + api.logger.error(`[graph-memory-pro] compact failed: ${err}`); + return { ok: false, compacted: false, reason: String(err) }; + } + }); }, async afterTurn({ sessionId, sessionKey, messages, prePromptMessageCount, isHeartbeat }: { @@ -607,6 +679,12 @@ const graphMemoryProPlugin = { return; } + // cron session 关闭图谱功能：跳过入库回填与知识提取 + if (isCronSessionKey(sessionKey) && !cronCfg.enabled) { + ingestedSinceTurn.delete(sessionId); + return; + } + // Official OpenClaw delivers ingest() and afterTurn() as separate // lifecycle phases. Older downstream builds incorrectly call only // afterTurn(). Persist just the missing suffix so neither host loses @@ -627,6 +705,12 @@ const graphMemoryProPlugin = { api.logger.info(`[graph-memory-pro] afterTurn sid=${sessionId.slice(0, 8)} turn=${turnNum} rawMsgs=${newMessages.length}`); + // cron session 关闭知识提取：消息仅入库缓冲，可稍后用 `graph-memory extract` 手动回填 + if (isCronSessionKey(sessionKey) && !cronCfg.extract) { + api.logger.info("[graph-memory-pro] cron session: extraction skipped (cron.extract=false)"); + return; + } + // 直接用原始消息提取知识图谱（异步，不阻塞） extractTurnKnowledge(sessionId, turnNum, newMessages).catch(err => { api.logger.error(`[graph-memory-pro] extract failed: ${err}`); @@ -647,6 +731,8 @@ const graphMemoryProPlugin = { if (childSessionId) { recalled.delete(childSessionId); msgSeq.delete(childSessionId); + msgSeqLoaders.delete(childSessionId); + extractLocks.delete(childSessionId); ingestedSinceTurn.delete(childSessionId); } sessionIdsByKey.delete(childSessionKey); @@ -655,6 +741,8 @@ const graphMemoryProPlugin = { async dispose() { msgSeq.clear(); + msgSeqLoaders.clear(); + extractLocks.clear(); recalled.clear(); sessionIdsByKey.clear(); pendingSubagentRecall.clear(); @@ -677,6 +765,12 @@ const graphMemoryProPlugin = { : typeof ctx?.sessionKey === "string" ? ctx.sessionKey : undefined; try { + // cron session：图谱功能关闭或明确禁用时，跳过 finalize 与图维护（finally 清理仍执行） + if (isCronSessionKey(sessionKey) && !(cronCfg.enabled && cronCfg.finalizeAndMaintain)) { + api.logger.info(`[graph-memory-pro] cron session ${sid.slice(0, 12)}…: finalize + maintenance skipped (cron config)`); + return; + } + const nodes = await getBySession(driver, sid); if (nodes.length) { // 获取图谱摘要 @@ -731,6 +825,8 @@ const graphMemoryProPlugin = { api.logger.error(`[graph-memory-pro] session_end error: ${err}`); } finally { msgSeq.delete(sid); + msgSeqLoaders.delete(sid); + extractLocks.delete(sid); recalled.delete(sid); ingestedSinceTurn.delete(sid); if (sessionKey && sessionIdsByKey.get(sessionKey) === sid) { @@ -1145,13 +1241,21 @@ const graphMemoryProPlugin = { (_ctx: any) => ({ name: "gm_maintain", label: "Graph Memory Maintenance", - description: "手动触发图维护：去重、PageRank、社区检测。", + description: "手动触发图维护：衰减评分 + tier 转换、去重、PageRank、社区检测。", parameters: Type.Object({}), async execute() { const embedFn = (recaller as any).embed ?? undefined; const result = await runMaintenance(driver, cfg, llm, embedFn); + const t = result.decay.tierTransitions; + const totalTransitions = t.coreToWorking + t.workingToPeripheral + t.peripheralToWorking + t.workingToCore; const text = [ `🔧 图维护完成（${result.durationMs}ms）`, + result.decay.enabled + ? `衰减：扫描 ${result.decay.scanned} 个节点，tier 转换 ${totalTransitions} 次` + + (totalTransitions > 0 + ? `（core→working ${t.coreToWorking}，working→peripheral ${t.workingToPeripheral}，peripheral→working ${t.peripheralToWorking}，working→core ${t.workingToCore}）` + : "") + : `衰减：已禁用`, `去重：${result.dedup.pairs.length} 对相似，合并 ${result.dedup.merged} 对`, ...(result.dedup.pairs.length > 0 ? result.dedup.pairs.slice(0, 5).map(p => ` "${p.nameA}" ≈ "${p.nameB}" (${(p.similarity * 100).toFixed(1)}%)`) @@ -1161,7 +1265,7 @@ const graphMemoryProPlugin = { `PageRank Top 5：`, ...result.pagerank.topK.slice(0, 5).map((n, i) => ` ${i + 1}. ${n.name} (${n.score.toFixed(4)})`), ].join("\n"); - return { content: [{ type: "text", text }], details: { durationMs: result.durationMs, dedupMerged: result.dedup.merged, communities: result.community.count } }; + return { content: [{ type: "text", text }], details: { durationMs: result.durationMs, decayTransitions: totalTransitions, dedupMerged: result.dedup.merged, communities: result.community.count } }; }, }), { name: "gm_maintain" }, diff --git a/openclaw.plugin.json b/openclaw.plugin.json index f54c9d0..8022351 100755 --- a/openclaw.plugin.json +++ b/openclaw.plugin.json @@ -29,6 +29,37 @@ "dedupThreshold": { "type": "number", "default": 0.90 }, "pagerankDamping": { "type": "number", "default": 0.85 }, "pagerankIterations": { "type": "number", "default": 20 }, + "decay": { + "type": "object", + "description": "柔性衰减：三因子加权评分（recency+frequency+intrinsic）+ tier 双向转换（core/working/peripheral）。完整公式与调参指南见 docs/decay.md。recencyWeight + frequencyWeight + intrinsicWeight 推荐和为 1（运行时会自动归一化）。", + "properties": { + "enabled": { "type": "boolean", "default": true, "description": "是否启用自动衰减。关闭后 tier 永久保持初始 working 状态。" }, + "recencyHalfLifeDays": { "type": "number", "default": 30, "description": "Recency 半衰期（天）。effectiveHL = halfLife * exp(importanceModulation * importance)。" }, + "recencyWeight": { "type": "number", "default": 0.4, "description": "Recency 在 composite 中的权重。三个权重推荐和为 1。" }, + "importanceModulation": { "type": "number", "default": 1.5, "description": "半衰期调制系数；越大则高 importance 节点衰减越慢。" }, + "frequencyWeight": { "type": "number", "default": 0.3, "description": "Frequency 在 composite 中的权重。三个权重推荐和为 1。" }, + "intrinsicWeight": { "type": "number", "default": 0.3, "description": "Intrinsic（importance × confidence）在 composite 中的权重。三个权重推荐和为 1。" }, + "betaCore": { "type": "number", "default": 0.8, "description": "core tier 的 Weibull 形状参数；<1 = 缓衰。" }, + "betaWorking": { "type": "number", "default": 1.0, "description": "working tier 的 Weibull 形状参数；=1 = 标准指数衰减。" }, + "betaPeripheral": { "type": "number", "default": 1.3, "description": "peripheral tier 的 Weibull 形状参数；>1 = 加速衰减。" }, + "coreAccessThreshold": { "type": "number", "default": 10, "description": "working→core 所需的最低 validatedCount。" }, + "coreCompositeThreshold": { "type": "number", "default": 0.7, "description": "working→core 所需的最低 composite 分数。" }, + "coreImportanceThreshold": { "type": "number", "default": 0.8, "description": "working→core 所需的最低归一化 importance。" }, + "peripheralCompositeThreshold": { "type": "number", "default": 0.15, "description": "composite 低于此值触发 demote（core→working 或 working→peripheral）。" }, + "peripheralAgeDays": { "type": "number", "default": 60, "description": "working→peripheral 的年龄阈值（同时 validatedCount < workingAccessThreshold 才触发）。" }, + "workingAccessThreshold": { "type": "number", "default": 3, "description": "demote（count 不足时）/ promote（count 充足时）的 access 次数分界。" }, + "workingCompositeThreshold": { "type": "number", "default": 0.4, "description": "peripheral→working 所需的最低 composite 分数。" } + } + }, + "cron": { + "type": "object", + "description": "cron 定时会话的图谱行为开关（host 将 cron 标记放在 sessionKey 上，形如 agent::cron:）。默认全部启用，与普通会话一致。注意：cron 任务若显式设置自定义 sessionKey，则无法被识别为 cron 会话。", + "properties": { + "enabled": { "type": "boolean", "default": true, "description": "是否在 cron session 内启用图谱功能（召回注入 + 消息入库）。关闭后 cron 会话不自动召回、不自动入库；gm_* 工具仍可手动调用。" }, + "extract": { "type": "boolean", "default": true, "description": "是否在 cron session 内触发知识提取（afterTurn / compact 的 LLM 三元组提取）。关闭后消息仍入库缓冲，可用 openclaw graph-memory extract 手动回填。" }, + "finalizeAndMaintain": { "type": "boolean", "default": true, "description": "cron session 结束时是否执行 finalize（EVENT→SKILL 晋升）和图维护（decay/PageRank/社区检测）。频繁的 cron 任务可关闭以避免每次结束都跑全局维护。" } + } + }, "llm": { "type": "object", "properties": { diff --git a/package.json b/package.json index 4ea628f..2c0112d 100755 --- a/package.json +++ b/package.json @@ -24,8 +24,9 @@ "test:watch": "vitest --passWithNoTests" }, "dependencies": { + "@sinclair/typebox": "^0.34.48", "neo4j-driver": "^5.27.0", - "@sinclair/typebox": "^0.34.48" + "opencode-ai": "^1.18.16" }, "devDependencies": { "@types/node": "^20.0.0", diff --git a/setup-graph-memory-pro.sh b/setup-graph-memory-pro.sh index 6307a09..1e9baf8 100644 --- a/setup-graph-memory-pro.sh +++ b/setup-graph-memory-pro.sh @@ -79,6 +79,7 @@ NEO4J_USER="neo4j" NEO4J_URI="" # 留空 → 根据是否自建 Neo4j 自动决定 PLUGIN_REF="" INTERACTIVE=true +PC="" # 嵌入式 provider 选择（1-7）；交互模式由 read 赋值，非交互留空 AUTOSTART_METHODS=() # configure_autostart 写入；卸载与完成提示读取 while [[ $# -gt 0 ]]; do case "$1" in diff --git a/src/cli-extract.ts b/src/cli-extract.ts new file mode 100644 index 0000000..cfe9d18 --- /dev/null +++ b/src/cli-extract.ts @@ -0,0 +1,288 @@ +/** + * graph-memory-pro CLI — `openclaw graph-memory extract` + * + * 对未被提取的会话消息做批量图谱提取，补齐因 compact 未触发、提取失败或 + * 进程退出而残留的 GmMessage。流程镜像 index.ts 的 compact() 路径： + * getUnextracted → extractor.extract → upsertNode + syncEmbed → upsertEdge → markExtracted + * + * 命令在 cli-metadata 模式下运行（register() 早早 return），所以这里必须自行 + * 完成 Neo4j driver / schema / LLM / embedder / Extractor / Recaller 的初始化。 + */ + +import readline from "node:readline/promises"; +import { stdin as input, stdout as output } from "node:process"; + +import type { Driver } from "neo4j-driver"; +import type { GmConfig } from "./types.ts"; +import { getDriver, initSchema, closeDriver } from "./store/db.ts"; +import { + listUnextractedSessions, + getUnextracted, + markExtracted, + upsertNode, + upsertEdge, + findByName, + getBySession, + type UnextractedSessionInfo, +} from "./store/store.ts"; +import { createCompleteFn, resolveProvider } from "./engine/llm.ts"; +import { createEmbedFn } from "./engine/embed.ts"; +import { Recaller } from "./recaller/recall.ts"; +import { Extractor } from "./extractor/extract.ts"; + +const AFFIRMATIVE = new Set(["y", "yes", "yeah", "yep", "ok", "okay", "true", "1", "confirm"]); + +export function isAffirmative(answer: string): boolean { + return AFFIRMATIVE.has(answer.trim().toLowerCase()); +} + +export interface BackfillExtractOptions { + yes?: boolean; + limit?: number; + session?: string; + dryRun?: boolean; +} + +export interface BackfillExtractParams { + cfg: GmConfig; + effectiveModel: string; + options: BackfillExtractOptions; + log?: (msg: string) => void; + prompt?: (question: string) => Promise; +} + +export interface BackfillExtractResult { + sessionsTotal: number; + sessionsProcessed: number; + sessionsSkipped: number; + nodesCreated: number; + edgesCreated: number; + batches: number; + durationMs: number; +} + +const DEFAULT_BATCH_LIMIT_MULTIPLIER = 3; + +function defaultLog(msg: string): void { + console.log(msg); +} + +function formatSessionLine(info: UnextractedSessionInfo, index: number): string { + const created = info.minCreatedAt > 0 + ? new Date(info.minCreatedAt).toISOString().replace("T", " ").slice(0, 19) + : "?"; + return ` ${String(index + 1).padStart(3, " ")}. sid=${info.sessionId.slice(0, 12)}… msgs=${info.messageCount} maxTurn=${info.maxTurn} since=${created}`; +} + +export async function runBackfillExtraction( + params: BackfillExtractParams, +): Promise { + const start = Date.now(); + const log = params.log ?? defaultLog; + const opts = params.options; + const cfg = params.cfg; + + const result: BackfillExtractResult = { + sessionsTotal: 0, + sessionsProcessed: 0, + sessionsSkipped: 0, + nodesCreated: 0, + edgesCreated: 0, + batches: 0, + durationMs: 0, + }; + + if (!cfg.neo4j?.uri) { + throw new Error( + "[graph-memory-pro] extract 需要 neo4j.uri 配置。请在 graph-memory-pro 插件配置中设置 neo4j.uri / neo4j.user / neo4j.password。", + ); + } + + if (!params.effectiveModel) { + throw new Error( + "[graph-memory-pro] extract 需要一个 LLM model。请在 config.llm.model 或 agents.defaults.model 中设置。", + ); + } + + const providerInfo = resolveProvider(cfg.llm); + if (providerInfo.provider === "anthropic" && !cfg.llm?.apiKey) { + throw new Error("[graph-memory-pro] llm.provider=anthropic 但未配 llm.apiKey，无法提取。"); + } + if (providerInfo.provider === "openai" && (!cfg.llm?.apiKey || !cfg.llm?.baseURL)) { + throw new Error("[graph-memory-pro] llm.provider=openai 需要 llm.apiKey + llm.baseURL，无法提取。"); + } + if (providerInfo.provider === "oauth" && !cfg.llm?.oauthPath) { + throw new Error( + "[graph-memory-pro] llm.provider=oauth 但未配 llm.oauthPath。请先运行 `openclaw graph-memory auth login`。", + ); + } + + const driver: Driver = getDriver(cfg.neo4j); + + try { + log("[graph-memory-pro] 正在初始化 Neo4j schema..."); + await initSchema(driver, cfg.embedding); + + log("[graph-memory-pro] 正在初始化 LLM 与 embedder..."); + const llm = createCompleteFn(params.effectiveModel, cfg.llm); + const extractor = new Extractor(llm); + const recaller = new Recaller(driver, cfg); + const embedFn = await createEmbedFn(cfg.embedding); + if (embedFn) { + recaller.setEmbedFn(embedFn); + log("[graph-memory-pro] embedding 已就绪，新节点将同步向量。"); + } else { + log("[graph-memory-pro] 未配置 embedding，跳过向量同步（dual-path recall 会降级为文本搜索）。"); + } + + let sessions = await listUnextractedSessions(driver); + if (opts.session) { + sessions = sessions.filter(s => s.sessionId === opts.session); + if (!sessions.length) { + log(`[graph-memory-pro] --session=${opts.session} 没有匹配到含未提取消息的会话。`); + result.durationMs = Date.now() - start; + return result; + } + } + result.sessionsTotal = sessions.length; + + if (sessions.length === 0) { + log("[graph-memory-pro] 没有需要提取的会话。"); + result.durationMs = Date.now() - start; + return result; + } + + const totalMessages = sessions.reduce((s, info) => s + info.messageCount, 0); + log(`[graph-memory-pro] 发现 ${sessions.length} 个会话共 ${totalMessages} 条未提取消息：`); + sessions.forEach((info, i) => log(formatSessionLine(info, i))); + + if (opts.dryRun) { + log("[graph-memory-pro] --dry-run 模式，未执行提取。"); + result.sessionsSkipped = sessions.length; + result.durationMs = Date.now() - start; + return result; + } + + if (!opts.yes) { + const prompt = params.prompt ?? ((q: string) => defaultPrompt(q)); + const answer = await prompt(`\n将对以上 ${sessions.length} 个会话发起 LLM 提取，继续？[y/N] `); + if (!isAffirmative(answer)) { + log("[graph-memory-pro] 已取消。"); + result.sessionsSkipped = sessions.length; + result.durationMs = Date.now() - start; + return result; + } + } + + const batchLimit = opts.limit && opts.limit > 0 + ? opts.limit + : Math.max(1, cfg.compactTurnCount) * DEFAULT_BATCH_LIMIT_MULTIPLIER; + + log(`\n[graph-memory-pro] 开始提取（每批最多 ${batchLimit} 条消息）...`); + + for (const info of sessions) { + log(`\n[graph-memory-pro] 会话 ${info.sessionId.slice(0, 12)}… (${info.messageCount} 条消息)`); + try { + const processed = await extractSessionLoop(driver, extractor, recaller, info.sessionId, batchLimit, log); + result.nodesCreated += processed.nodes; + result.edgesCreated += processed.edges; + result.batches += processed.batches; + result.sessionsProcessed += 1; + log(` -> 完成：${processed.nodes} 节点 / ${processed.edges} 边 / ${processed.batches} 批`); + } catch (err) { + result.sessionsSkipped += 1; + log(` -> 失败：${err instanceof Error ? err.message : String(err)}`); + } + } + + result.durationMs = Date.now() - start; + + log( + `\n[graph-memory-pro] 提取完成：${result.sessionsProcessed}/${result.sessionsTotal} 会话，` + + `${result.nodesCreated} 节点，${result.edgesCreated} 边，${result.batches} 批，` + + `用时 ${(result.durationMs / 1000).toFixed(1)}s`, + ); + return result; + } finally { + await closeDriver(); + } +} + +interface SessionExtractStats { + nodes: number; + edges: number; + batches: number; +} + +async function extractSessionLoop( + driver: Driver, + extractor: Extractor, + recaller: Recaller, + sessionId: string, + batchLimit: number, + log: (msg: string) => void, +): Promise { + const stats: SessionExtractStats = { nodes: 0, edges: 0, batches: 0 }; + const hardBatchCeiling = 50; + let exhausted = false; + + for (let i = 0; i < hardBatchCeiling; i++) { + const msgs = await getUnextracted(driver, sessionId, batchLimit); + if (!msgs.length) break; + + stats.batches += 1; + const existing = (await getBySession(driver, sessionId)).map(n => n.name); + const extraction = await extractor.extract({ messages: msgs, existingNames: existing }); + + const nameToId = new Map(); + for (const nc of extraction.nodes) { + const { node } = await upsertNode(driver, { + type: nc.type, name: nc.name, + description: nc.description, content: nc.content, + }, sessionId); + nameToId.set(node.name, node.id); + stats.nodes += 1; + void recaller.syncEmbed(node).catch(() => {}); + } + + for (const ec of extraction.edges) { + const fromNode = await findByName(driver, ec.from); + const toNode = await findByName(driver, ec.to); + const fromId = nameToId.get(ec.from) ?? fromNode?.id; + const toId = nameToId.get(ec.to) ?? toNode?.id; + if (fromId && toId) { + await upsertEdge(driver, { + fromId, toId, type: ec.type, + instruction: ec.instruction, condition: ec.condition, sessionId, + }); + stats.edges += 1; + } + } + + const maxTurn = msgs.reduce((m, msg) => Math.max(m, msg.turn_index ?? 0), 0); + await markExtracted(driver, sessionId, maxTurn); + log(` batch ${stats.batches}: ${msgs.length} 消息 -> ${extraction.nodes.length} 节点 / ${extraction.edges.length} 边（累计 ${stats.nodes}/${stats.edges}）`); + + if (msgs.length < batchLimit) break; + if (i === hardBatchCeiling - 1) exhausted = true; + } + + if (exhausted) { + log(` 警告：达到批数上限 ${hardBatchCeiling}，会话 ${sessionId.slice(0, 12)}… 仍有未提取消息，请再次运行。`); + } + + return stats; +} + +async function defaultPrompt(question: string): Promise { + if (!process.stdin.isTTY && process.env.GRAPH_MEMORY_EXTRACT_CONFIRM === undefined) { + return ""; + } + const rl = readline.createInterface({ input, output }); + try { + const answer = await rl.question(question); + return answer; + } finally { + rl.close(); + } +} diff --git a/src/cli.ts b/src/cli.ts index f4c5a25..a7faeaf 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -25,6 +25,8 @@ import { type OAuthProviderId, } from "./engine/oauth.ts"; import type { ReasoningEffort } from "./engine/llm.ts"; +import { runBackfillExtraction } from "./cli-extract.ts"; +import { DEFAULT_CONFIG, type GmConfig } from "./types.ts"; // ─── 最小 Commander 鸭子类型（避免引入 commander 依赖） ─────────── // host 运行时注入真正的 commander.Command 实例，结构兼容此接口即可。 @@ -45,6 +47,7 @@ export interface GraphMemoryCliDeps { pluginId?: string; pluginConfig?: Record | undefined; resolveConfigPath?: (input: string) => string; + defaultModel?: string; oauthTestHooks?: { openUrl?: (url: string) => void | Promise; authorizeUrl?: (url: string) => void | Promise; @@ -375,5 +378,57 @@ export function createGraphMemoryCli(deps: GraphMemoryCliDeps) { throw new Error(`[graph-memory-pro] OAuth login failed: ${message}`); } }); + + root + .command("extract") + .description( + "扫描 Neo4j 中未提取的会话消息，按 compact 流程批量补提知识图谱，并同步节点 embedding", + ) + .option("--yes", "跳过确认提示，直接执行提取", false) + .option("--dry-run", "只列出待提取会话，不调用 LLM", false) + .option("--limit ", "每个会话每批最多提取的消息条数（默认 compactTurnCount * 3）", undefined) + .option("--session ", "仅提取指定 sessionId（默认全部含未提取消息的会话）", undefined) + .option("--model ", "本次提取使用的 LLM 模型（覆盖配置中的 llm.model / agents.defaults.model）", undefined) + .action(async (options: Record) => { + try { + const rawCfg = isPlainObject(deps.pluginConfig) + ? (deps.pluginConfig as Record) + : {}; + const cfg: GmConfig = { + ...DEFAULT_CONFIG, + ...(rawCfg as Partial), + }; + if (isPlainObject(rawCfg.neo4j)) { + cfg.neo4j = { ...DEFAULT_CONFIG.neo4j, ...(rawCfg.neo4j as any) }; + } + + const cfgLlm = isPlainObject(rawCfg.llm) ? (rawCfg.llm as any) : undefined; + const flagModel = typeof options.model === "string" && options.model.trim() + ? options.model.trim() + : undefined; + const effectiveModel = flagModel ?? cfgLlm?.model ?? deps.defaultModel ?? ""; + + const limitFlag = typeof options.limit === "string" + ? Number.parseInt(options.limit, 10) + : (typeof options.limit === "number" ? options.limit : undefined); + + await runBackfillExtraction({ + cfg, + effectiveModel, + options: { + yes: options.yes === true, + dryRun: options.dryRun === true, + session: typeof options.session === "string" ? options.session : undefined, + limit: limitFlag !== undefined && Number.isFinite(limitFlag) && limitFlag > 0 + ? Math.floor(limitFlag) + : undefined, + }, + }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + console.error("[graph-memory-pro] extract 失败：", message); + throw new Error(`[graph-memory-pro] extract failed: ${message}`); + } + }); }; } diff --git a/src/graph/community.ts b/src/graph/community.ts index d5f5dd5..71b021a 100755 --- a/src/graph/community.ts +++ b/src/graph/community.ts @@ -6,12 +6,15 @@ * 保留 summarizeCommunities()（需要 LLM） */ +import { createHash } from "node:crypto"; import type { Driver } from "neo4j-driver"; import { getSession } from "../store/db.ts"; import { clearCommunities, updateCommunities, upsertCommunitySummary, + getCommunitySummary, + getCommunitySummaryBySignature, pruneCommunitySummaries, } from "../store/store.ts"; import { getExistingActiveRelTypes, projectActiveGraph } from "./projection.ts"; @@ -142,6 +145,10 @@ const COMMUNITY_SUMMARY_SYS = `你是知识图谱社区摘要引擎。根据社 - 不要使用"社区"这个词 - 不要加引号或标点以外的格式`; +export function buildCommunityMemberSignature(memberIds: string[]): string { + return createHash("sha1").update([...memberIds].sort().join(",")).digest("hex"); +} + export async function summarizeCommunities( driver: Driver, communities: Map, @@ -154,6 +161,26 @@ export async function summarizeCommunities( for (const [communityId, memberIds] of communities) { if (memberIds.length === 0) continue; + const memberSignature = buildCommunityMemberSignature(memberIds); + + const current = await getCommunitySummary(driver, communityId); + if (current?.memberSignature === memberSignature && current.summary.trim()) { + continue; + } + + const reusable = await getCommunitySummaryBySignature(driver, memberSignature); + if (reusable?.summary.trim()) { + await upsertCommunitySummary( + driver, + communityId, + reusable.summary, + memberIds.length, + reusable.embedding, + memberSignature, + ); + continue; + } + const session = getSession(driver); let members: any[]; try { @@ -204,7 +231,7 @@ export async function summarizeCommunities( } catch {} } - await upsertCommunitySummary(driver, communityId, cleaned, memberIds.length, embedding); + await upsertCommunitySummary(driver, communityId, cleaned, memberIds.length, embedding, memberSignature); generated++; } catch (err) { console.log(` [WARN] community summary failed for ${communityId}: ${err}`); diff --git a/src/graph/decay.ts b/src/graph/decay.ts new file mode 100644 index 0000000..9859c88 --- /dev/null +++ b/src/graph/decay.ts @@ -0,0 +1,268 @@ +/** + * graph-memory-pro — 柔性衰减（三因子加权评分 + tier 双向转换） + * + * 完整公式、字段映射、默认值来源、调参指南见 docs/decay.md。 + * 评分 / tier 决策 / applyDecay 的入口均在本文件。 + * + * 调用时机：runMaintenance 的第 0 步（去重/PageRank/社区之前）。 + * decay 不动 status，只动 tier。 + */ + +import type { Driver } from "neo4j-driver"; +import type { GmConfig, DecayConfig, GmNode, NodeTier } from "../types.ts"; +import { getSession } from "../store/db.ts"; +import { allActiveNodes } from "../store/store.ts"; + +const MS_PER_DAY = 86_400_000; + +export interface CompositeScore { + composite: number; + recency: number; + frequency: number; + intrinsic: number; +} + +export interface TierTransition { + coreToWorking: number; + workingToPeripheral: number; + peripheralToWorking: number; + workingToCore: number; +} + +export interface DecayResult { + enabled: boolean; + scanned: number; + tierTransitions: TierTransition; + durationMs: number; +} + +// ─── 归一化辅助（纯函数，便于单元测试） ────────────────────── + +/** importance ∈ [0,1]：当前批次的 pagerank 归一化值。 */ +export function normalizeImportance(pagerank: number, maxPagerank: number): number { + if (maxPagerank <= 0) return 0; + return Math.min(1, Math.max(0, pagerank / maxPagerank)); +} + +/** confidence ∈ [0,1)：validatedCount 越高越可信，饱和收敛到 1。 */ +export function computeConfidence(validatedCount: number): number { + const c = Math.max(0, validatedCount); + return 1 - 1 / (1 + c); +} + +// ─── 三因子评分（纯函数） ──────────────────────────────────── + +/** 选 lastAccessedAt → updatedAt → createdAt 中第一个 > 0 的，用于回退旧节点缺字段。 */ +function pickLastActive(node: Pick): number { + const la = node.lastAccessedAt ?? 0; + const up = node.updatedAt ?? 0; + if (la > 0) return la; + if (up > 0) return up; + return node.createdAt ?? 0; +} + +/** β 随 tier 变化：core 缓衰、peripheral 促衰。 */ +export function computeBeta(tier: NodeTier, cfg: DecayConfig): number { + switch (tier) { + case "core": return cfg.betaCore; + case "working": return cfg.betaWorking; + case "peripheral": return cfg.betaPeripheral; + } +} + +/** + * Recency 分量：Weibull 拉伸指数衰减。 + * tier 决定 β；importance 调制半衰期（高重要性 → 慢衰减）。 + */ +export function scoreRecency( + node: Pick, + importance: number, + now: number, + cfg: DecayConfig, +): number { + const lastActive = pickLastActive(node); + const daysSince = Math.max(0, (now - lastActive) / MS_PER_DAY); + + const effectiveHL = cfg.recencyHalfLifeDays * Math.exp(cfg.importanceModulation * importance); + const lambda = Math.LN2 / effectiveHL; + const beta = computeBeta(node.tier ?? "working", cfg); + + return Math.exp(-lambda * Math.pow(daysSince, beta)); +} + +/** + * Frequency 分量：基础饱和项 × 平均访问间隔新鲜度。 + * validatedCount ≤ 1 时只返回基础项（无法算平均间隔）。 + */ +export function scoreFrequency( + node: Pick, +): number { + const count = Math.max(0, node.validatedCount); + const base = 1 - Math.exp(-count / 5); + if (count <= 1) return base; + + const lastActive = pickLastActive(node); + const accessSpanDays = Math.max(1, (lastActive - node.createdAt) / MS_PER_DAY); + const avgGapDays = accessSpanDays / Math.max(count - 1, 1); + const recentnessBonus = Math.exp(-avgGapDays / 30); + + return base * (0.5 + 0.5 * recentnessBonus); +} + +/** Intrinsic 分量：importance × confidence。 */ +export function scoreIntrinsic(importance: number, confidence: number): number { + return importance * confidence; +} + +/** 三因子加权汇总。权重和在运行时归一化到 1，避免用户配置偏差导致 composite > 1。 */ +export function scoreNode( + node: Pick, + maxPagerank: number, + now: number, + cfg: DecayConfig, +): CompositeScore { + const importance = normalizeImportance(node.pagerank, maxPagerank); + const confidence = computeConfidence(node.validatedCount); + const recency = scoreRecency(node, importance, now, cfg); + const frequency = scoreFrequency(node); + const intrinsic = scoreIntrinsic(importance, confidence); + + const wSum = cfg.recencyWeight + cfg.frequencyWeight + cfg.intrinsicWeight; + const safeSum = wSum > 0 ? wSum : 1; + const wR = cfg.recencyWeight / safeSum; + const wF = cfg.frequencyWeight / safeSum; + const wI = cfg.intrinsicWeight / safeSum; + + const composite = wR * recency + wF * frequency + wI * intrinsic; + + return { composite, recency, frequency, intrinsic }; +} + +// ─── Tier 转换决策（纯函数） ───────────────────────────────── + +/** + * 决定节点的下一个 tier。返回 null 表示保持不变。 + * importance 已归一化（调用方须先 normalizeImportance）。 + */ +export function decideTierTransition( + node: Pick, + score: CompositeScore, + importance: number, + cfg: DecayConfig, + now: number = Date.now(), +): NodeTier | null { + const current = node.tier ?? "working"; + const count = node.validatedCount; + const ageDays = Math.max(0, (now - node.createdAt) / MS_PER_DAY); + const composite = score.composite; + + if (current === "core" + && composite < cfg.peripheralCompositeThreshold + && count < cfg.workingAccessThreshold) { + return "working"; + } + + if (current === "working") { + if (composite < cfg.peripheralCompositeThreshold) return "peripheral"; + if (ageDays > cfg.peripheralAgeDays && count < cfg.workingAccessThreshold) { + return "peripheral"; + } + } + + if (current === "peripheral" + && count >= cfg.workingAccessThreshold + && composite >= cfg.workingCompositeThreshold) { + return "working"; + } + + if (current === "working" + && count >= cfg.coreAccessThreshold + && composite >= cfg.coreCompositeThreshold + && importance >= cfg.coreImportanceThreshold) { + return "core"; + } + + return null; +} + +// ─── 应用层：扫描 + 评分 + 转换 ────────────────────────────── + +const EMPTY_TRANSITIONS: TierTransition = { + coreToWorking: 0, + workingToPeripheral: 0, + peripheralToWorking: 0, + workingToCore: 0, +}; + +function bumpTransition(transitions: TierTransition, from: NodeTier, to: NodeTier): void { + if (from === "core" && to === "working") transitions.coreToWorking++; + else if (from === "working" && to === "peripheral") transitions.workingToPeripheral++; + else if (from === "peripheral" && to === "working") transitions.peripheralToWorking++; + else if (from === "working" && to === "core") transitions.workingToCore++; +} + +/** + * 扫描所有 active 节点：评分 + tier 转换 + 写回 decayScore / tier。 + * 不动 status（status=deprecated 仅由手动弃用触发）。 + */ +export async function applyDecay(driver: Driver, cfg: Pick): Promise { + const start = Date.now(); + const d = cfg.decay; + if (!d?.enabled) { + return { enabled: false, scanned: 0, tierTransitions: { ...EMPTY_TRANSITIONS }, durationMs: 0 }; + } + + const nodes = await allActiveNodes(driver); + if (nodes.length === 0) { + return { enabled: true, scanned: 0, tierTransitions: { ...EMPTY_TRANSITIONS }, durationMs: 0 }; + } + + const maxPagerank = Math.max(...nodes.map(n => n.pagerank), 0.0001); + + const updates: Array<{ id: string; tier: NodeTier; composite: number; tierChanged: boolean }> = []; + const transitions: TierTransition = { ...EMPTY_TRANSITIONS }; + + for (const node of nodes) { + const score = scoreNode(node, maxPagerank, start, d); + const importance = normalizeImportance(node.pagerank, maxPagerank); + const currentTier = node.tier ?? "working"; + const nextTier = decideTierTransition(node, score, importance, d, start); + const finalTier = nextTier ?? currentTier; + const tierChanged = nextTier !== null; + + if (tierChanged) bumpTransition(transitions, currentTier, finalTier); + + updates.push({ + id: node.id, + tier: finalTier, + composite: score.composite, + tierChanged, + }); + } + + if (updates.length > 0) { + const session = getSession(driver); + try { + await session.run( + `UNWIND $updates AS u + MATCH (n:Task|Skill|Event {id: u.id}) + SET n.tier = u.tier, + n.decayScore = u.composite, + n.decayComputedAt = $now, + n.updatedAt = CASE WHEN u.tierChanged THEN $now ELSE n.updatedAt END`, + { updates, now: start }, + ); + } finally { + await session.close(); + } + } + + return { + enabled: true, + scanned: nodes.length, + tierTransitions: transitions, + durationMs: Date.now() - start, + }; +} diff --git a/src/graph/maintenance.ts b/src/graph/maintenance.ts index 64cd4fa..a2e68b8 100755 --- a/src/graph/maintenance.ts +++ b/src/graph/maintenance.ts @@ -2,7 +2,7 @@ * graph-memory-pro — 图谱维护 * * 调用时机：session_end（finalize 之后） - * 执行顺序：去重 → 全局 PageRank → 社区检测 → 社区描述 + * 执行顺序：衰减 → 去重 → 全局 PageRank → 社区检测 → 社区描述 */ import type { Driver } from "neo4j-driver"; @@ -12,8 +12,10 @@ import type { EmbedFn } from "../engine/embed.ts"; import { computeGlobalPageRank, type GlobalPageRankResult } from "./pagerank.ts"; import { detectCommunities, summarizeCommunities, type CommunityResult } from "./community.ts"; import { dedup, type DedupResult } from "./dedup.ts"; +import { applyDecay, type DecayResult } from "./decay.ts"; export interface MaintenanceResult { + decay: DecayResult; dedup: DedupResult; pagerank: GlobalPageRankResult; community: CommunityResult; @@ -26,6 +28,9 @@ export async function runMaintenance( ): Promise { const start = Date.now(); + // 0. 衰减（柔性评分 + tier 转换）—— 先于其他步骤，让后续基于最新 tier 集合运算 + const decayResult = await applyDecay(driver, cfg); + // 1. 去重 const dedupResult = await dedup(driver, cfg); @@ -44,6 +49,7 @@ export async function runMaintenance( } return { + decay: decayResult, dedup: dedupResult, pagerank: pagerankResult, community: communityResult, diff --git a/src/store/db.ts b/src/store/db.ts index 620441c..d5d7315 100755 --- a/src/store/db.ts +++ b/src/store/db.ts @@ -75,6 +75,7 @@ export async function initSchema(driver: Driver, embedding?: EmbeddingConfig): P // Community await session.run("CREATE CONSTRAINT community_id IF NOT EXISTS FOR (c:Community) REQUIRE c.id IS UNIQUE"); + await session.run("CREATE INDEX community_member_signature IF NOT EXISTS FOR (c:Community) ON (c.memberSignature)"); // Message (temporary extraction buffer) await session.run("CREATE CONSTRAINT gm_msg_id IF NOT EXISTS FOR (m:GmMessage) REQUIRE m.id IS UNIQUE"); diff --git a/src/store/store.ts b/src/store/store.ts index e08787c..b5d6f6a 100755 --- a/src/store/store.ts +++ b/src/store/store.ts @@ -8,7 +8,7 @@ import type { Driver } from "neo4j-driver"; import neo4j from "neo4j-driver"; import { createHash } from "crypto"; -import type { GmNode, GmEdge, EdgeType, NodeType } from "../types.ts"; +import type { GmNode, GmEdge, EdgeType, NodeType, NodeTier } from "../types.ts"; import { NODE_TYPE_TO_LABEL, isValidEdgeDirection } from "../types.ts"; import { getSession } from "./db.ts"; @@ -32,6 +32,8 @@ function toNode(r: any): GmNode { description: n.description ?? "", content: n.content, status: n.status, + tier: (n.tier === "core" || n.tier === "working" || n.tier === "peripheral" + ? n.tier : "working") as NodeTier, validatedCount: toInt(n.validatedCount ?? n.validated_count ?? 1), sourceSessions: typeof n.sourceSessions === "string" ? JSON.parse(n.sourceSessions) @@ -40,6 +42,9 @@ function toNode(r: any): GmNode { pagerank: toFloat(n.pagerank ?? 0), createdAt: toInt(n.createdAt ?? n.created_at ?? 0), updatedAt: toInt(n.updatedAt ?? n.updated_at ?? 0), + lastAccessedAt: toInt(n.lastAccessedAt ?? n.last_accessed_at ?? n.updatedAt ?? n.updated_at ?? n.createdAt ?? 0), + decayScore: typeof n.decayScore === "number" ? n.decayScore : undefined, + decayComputedAt: n.decayComputedAt ? toInt(n.decayComputedAt) : undefined, }; } @@ -164,6 +169,7 @@ export async function upsertNode( THEN n.sourceSessions + $sessionId ELSE n.sourceSessions END, + n.lastAccessedAt = $now, n.updatedAt = $now RETURN n `, { name, content: c.content, description: c.description, sessionId, now: Date.now() }); @@ -180,9 +186,10 @@ export async function upsertNode( CREATE (n:MemoryNode:${label} { id: $id, name: $name, type: $type, description: $description, content: $content, - status: 'active', validatedCount: 1, + status: 'active', tier: 'working', validatedCount: 1, sourceSessions: $sessions, communityId: null, - pagerank: 0.0, createdAt: $now, updatedAt: $now + pagerank: 0.0, createdAt: $now, updatedAt: $now, + lastAccessedAt: $now }) RETURN n `, { @@ -852,6 +859,9 @@ export async function saveMessage( m.content = $content, m.extracted = false, m.createdAt = $now + ON MATCH SET + m.role = $role, + m.content = $content `, { id: uid("m"), sid, @@ -865,6 +875,22 @@ export async function saveMessage( } } +/** 该会话当前最大 turnIndex（无消息返回 0）。用于插件重启后恢复内存 msgSeq； + * 否则 turnIndex 从 1 重计 → MERGE 命中旧行 → ON CREATE 被跳过 → 新消息被静默丢弃。 */ +export async function getMaxTurnIndex(driver: Driver, sid: string): Promise { + const session = getSession(driver); + try { + const result = await session.run( + `MATCH (m:GmMessage {sessionId: $sid}) + RETURN coalesce(max(m.turnIndex), 0) AS maxTurn`, + { sid }, + ); + return toInt(result.records[0].get("maxTurn")); + } finally { + await session.close(); + } +} + export async function getUnextracted(driver: Driver, sid: string, limit: number): Promise { const session = getSession(driver); try { @@ -888,6 +914,37 @@ export async function getUnextracted(driver: Driver, sid: string, limit: number) } } +export interface UnextractedSessionInfo { + sessionId: string; + messageCount: number; + maxTurn: number; + minCreatedAt: number; +} + +export async function listUnextractedSessions(driver: Driver): Promise { + const session = getSession(driver); + try { + const result = await session.run(` + MATCH (m:GmMessage {extracted: false}) + WITH m.sessionId AS sid, + count(*) AS msgCount, + max(m.turnIndex) AS maxTurn, + min(coalesce(m.createdAt, 0)) AS minCreated + WHERE sid IS NOT NULL + RETURN sid, msgCount, maxTurn, minCreated + ORDER BY minCreated ASC, sid ASC + `); + return result.records.map(r => ({ + sessionId: r.get("sid"), + messageCount: toInt(r.get("msgCount")), + maxTurn: toInt(r.get("maxTurn")), + minCreatedAt: toInt(r.get("minCreated")), + })); + } finally { + await session.close(); + } +} + export async function markExtracted(driver: Driver, sid: string, upToTurn: number): Promise { const session = getSession(driver); try { @@ -972,12 +1029,15 @@ export interface CommunitySummary { id: string; summary: string; nodeCount: number; + /** 成员 ID 排序后的 sha1 — 用于识别"成员构成未变"的社区（复用摘要） */ + memberSignature: string | null; createdAt: number; updatedAt: number; } export async function upsertCommunitySummary( - driver: Driver, id: string, summary: string, nodeCount: number, embedding?: number[], + driver: Driver, id: string, summary: string, nodeCount: number, + embedding?: number[], memberSignature?: string, ): Promise { const session = getSession(driver); try { @@ -987,18 +1047,21 @@ export async function upsertCommunitySummary( c.summary = $summary, c.nodeCount = $nodeCount, c.embedding = $embedding, + c.memberSignature = $memberSignature, c.createdAt = $now, c.updatedAt = $now ON MATCH SET c.summary = $summary, c.nodeCount = $nodeCount, c.embedding = CASE WHEN $embedding IS NOT NULL THEN $embedding ELSE c.embedding END, + c.memberSignature = CASE WHEN $memberSignature IS NOT NULL THEN $memberSignature ELSE c.memberSignature END, c.updatedAt = $now `, { id, summary, nodeCount, embedding: embedding ?? null, + memberSignature: memberSignature ?? null, now: Date.now(), }); } finally { @@ -1019,6 +1082,32 @@ export async function getCommunitySummary(driver: Driver, id: string): Promise { + const session = getSession(driver); + try { + const result = await session.run( + "MATCH (c:Community {memberSignature: $memberSignature}) RETURN c ORDER BY c.updatedAt DESC LIMIT 1", + { memberSignature }, + ); + if (result.records.length === 0) return null; + const c = result.records[0].get("c").properties; + return { + id: c.id, + summary: c.summary, + nodeCount: toInt(c.nodeCount), + memberSignature: c.memberSignature ?? null, + embedding: Array.isArray(c.embedding) ? (c.embedding as number[]) : undefined, createdAt: toInt(c.createdAt), updatedAt: toInt(c.updatedAt), }; @@ -1039,6 +1128,7 @@ export async function getAllCommunitySummaries(driver: Driver): PromiseTask, SKILL->Skill, EVENT->Event */ export const NODE_TYPE_TO_LABEL: Record = { TASK: "Task", @@ -24,12 +31,26 @@ export interface GmNode { description: string; content: string; status: NodeStatus; + /** 与 NodeStatus 正交的衰减分层；旧节点/新节点缺省时按 working 处理。 */ + tier?: NodeTier; validatedCount: number; sourceSessions: string[]; communityId: string | null; pagerank: number; createdAt: number; updatedAt: number; + /** + * 最近一次"相关性活动"时间戳（epoch ms），由 upsertNode 在任意写入路径刷新 + * （重新提取、gm_record、gm_update、CRUD POST）。是衰减判定的基准。 + * 与 updatedAt 的区别：updatedAt 在 deprecate/merge 时也会变，不能代表相关性； + * 而 mergeNodes 故意不更新 lastAccessedAt（合并 ≠ 用户重新激活）。 + * 缺省时回退到 updatedAt / createdAt。 + */ + lastAccessedAt?: number; + /** 最近一次 decay 评分（0~1，越大越鲜活/重要）。仅 applyDecay 写入。 */ + decayScore?: number; + /** decayScore 的计算时间戳（epoch ms）。 */ + decayComputedAt?: number; } // ─── 边 ─────────────────────────────────────────────────────── @@ -137,6 +158,55 @@ export interface Neo4jConfig { password: string; } +// ─── 衰减（柔性评分模型）配置 ───────────────────────────────── +// +// 完整公式、字段映射、默认值来源、调参指南见 docs/decay.md。 +// 评分和 tier 转换逻辑实现在 src/graph/decay.ts。 + +export interface DecayConfig { + enabled: boolean; + recencyHalfLifeDays: number; + recencyWeight: number; + importanceModulation: number; + frequencyWeight: number; + intrinsicWeight: number; + betaCore: number; + betaWorking: number; + betaPeripheral: number; + coreAccessThreshold: number; + coreCompositeThreshold: number; + coreImportanceThreshold: number; + peripheralCompositeThreshold: number; + peripheralAgeDays: number; + workingAccessThreshold: number; + workingCompositeThreshold: number; +} + +// ─── cron 会话（定时任务）的图谱行为配置 ───────────────────── + +/** + * 判断是否为 cron 定时会话。host 把 cron 标记放在 sessionKey 上（sessionId 是随机 UUID）， + * 实际形状：cron: / agent::cron: / agent::cron::run:。 + * 按段匹配（split(":") 后包含 "cron"），避免误匹配 "cron-daily" 这类自定义段。 + * 注意：cron 任务若显式设置了自定义 sessionKey，host 不再附加 cron 段，此类会话无法识别（见 README）。 + * cron session 的图谱行为（召回/消息入库、知识提取、结束维护）可由 `cron` 配置独立开关；非 cron session 不受影响。 + */ +export function isCronSessionKey(sessionKey: string | undefined | null): boolean { + return typeof sessionKey === "string" && sessionKey.split(":").includes("cron"); +} + +export interface CronConfig { + enabled: boolean; + extract: boolean; + finalizeAndMaintain: boolean; +} + +export const DEFAULT_CRON_CONFIG: CronConfig = { + enabled: true, + extract: true, + finalizeAndMaintain: true, +}; + // ─── 插件配置 ───────────────────────────────────────────────── export interface GmConfig { @@ -163,6 +233,9 @@ export interface GmConfig { dedupThreshold: number; pagerankDamping: number; pagerankIterations: number; + /** 遗忘曲线衰减配置；未提供时使用 DEFAULT_CONFIG.decay。 */ + decay?: DecayConfig; + cron?: CronConfig; } export const DEFAULT_CONFIG: GmConfig = { @@ -178,4 +251,23 @@ export const DEFAULT_CONFIG: GmConfig = { dedupThreshold: 0.90, pagerankDamping: 0.85, pagerankIterations: 20, + decay: { + enabled: true, + recencyHalfLifeDays: 30, + recencyWeight: 0.4, + importanceModulation: 1.5, + frequencyWeight: 0.3, + intrinsicWeight: 0.3, + betaCore: 0.8, + betaWorking: 1.0, + betaPeripheral: 1.3, + coreAccessThreshold: 10, + coreCompositeThreshold: 0.7, + coreImportanceThreshold: 0.8, + peripheralCompositeThreshold: 0.15, + peripheralAgeDays: 60, + workingAccessThreshold: 3, + workingCompositeThreshold: 0.4, + }, + cron: DEFAULT_CRON_CONFIG, }; diff --git a/test/assemble-context.test.ts b/test/assemble-context.test.ts index e025e90..0c2340c 100644 --- a/test/assemble-context.test.ts +++ b/test/assemble-context.test.ts @@ -11,12 +11,14 @@ function makeNode(overrides: Partial): GmNode { description: "description", content: "content", status: "active", + tier: "working", validatedCount: 1, sourceSessions: ["test"], communityId: null, pagerank: 0, createdAt: now, updatedAt: now, + lastAccessedAt: now, ...overrides, }; } diff --git a/test/cli-extract.test.ts b/test/cli-extract.test.ts new file mode 100644 index 0000000..902016b --- /dev/null +++ b/test/cli-extract.test.ts @@ -0,0 +1,417 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + listUnextractedSessions: vi.fn(async () => [] as any[]), + getUnextracted: vi.fn(async (_d: any, _sid: any, _limit: any) => [] as any[]), + markExtracted: vi.fn(async () => {}), + upsertNode: vi.fn(async (_driver: any, c: any) => ({ + node: { + id: `n-${c.name}`, + type: c.type, + name: c.name, + description: c.description ?? "", + content: c.content, + status: "active", + validatedCount: 1, + sourceSessions: [], + communityId: null, + pagerank: 0, + createdAt: 0, + updatedAt: 0, + }, + isNew: true, + })), + upsertEdge: vi.fn(async () => {}), + findByName: vi.fn(async () => null), + getBySession: vi.fn(async () => [] as any[]), + extract: vi.fn(async () => ({ nodes: [] as any[], edges: [] as any[] })), + initSchema: vi.fn(async () => {}), + closeDriver: vi.fn(async () => {}), +})); + +vi.mock("../src/store/db.ts", () => ({ + getDriver: () => ({}), + initSchema: mocks.initSchema, + getSession: () => ({ close: async () => {} }), + closeDriver: mocks.closeDriver, +})); + +vi.mock("../src/store/store.ts", () => ({ + listUnextractedSessions: mocks.listUnextractedSessions, + getUnextracted: mocks.getUnextracted, + markExtracted: mocks.markExtracted, + upsertNode: mocks.upsertNode, + upsertEdge: mocks.upsertEdge, + findByName: mocks.findByName, + getBySession: mocks.getBySession, +})); + +vi.mock("../src/engine/llm.ts", () => ({ + createCompleteFn: () => async () => "", + resolveProvider: () => ({ provider: "openai", inferred: false }), +})); + +vi.mock("../src/engine/embed.ts", () => ({ + createEmbedFn: async () => null, +})); + +vi.mock("../src/recaller/recall.ts", () => ({ + Recaller: class { + setEmbedFn(): void {} + async syncEmbed(): Promise {} + }, +})); + +vi.mock("../src/extractor/extract.ts", () => ({ + Extractor: class { + async extract() { + return mocks.extract(); + } + }, +})); + +import { isAffirmative, runBackfillExtraction } from "../src/cli-extract.ts"; +import { DEFAULT_CONFIG } from "../src/types.ts"; + +function makeCfg(overrides: Record = {}) { + return { + ...DEFAULT_CONFIG, + neo4j: { uri: "bolt://localhost:7687", user: "neo4j", password: "x" }, + llm: { provider: "openai", apiKey: "k", baseURL: "https://api.openai.com/v1", model: "gpt-test" }, + ...overrides, + } as any; +} + +const SAMPLE_SESSION = { + sessionId: "sid-abc-1234567890", + messageCount: 5, + maxTurn: 5, + minCreatedAt: 1700000000000, +}; + +describe("isAffirmative", () => { + it.each([ + ["y", true], + ["Y", true], + ["yes", true], + ["YES", true], + [" yes ", true], + ["yeah", true], + ["ok", true], + ["confirm", true], + ["1", true], + ["true", true], + ["n", false], + ["no", false], + ["", false], + ["maybe", false], + ["nope", false], + ["0", false], + ])("isAffirmative(%j) -> %s", (input, expected) => { + expect(isAffirmative(input)).toBe(expected); + }); +}); + +describe("runBackfillExtraction", () => { + beforeEach(() => { + mocks.listUnextractedSessions.mockReset(); + mocks.getUnextracted.mockReset(); + mocks.markExtracted.mockReset(); + mocks.upsertNode.mockReset(); + mocks.upsertEdge.mockReset(); + mocks.findByName.mockReset(); + mocks.getBySession.mockReset(); + mocks.extract.mockReset(); + mocks.initSchema.mockReset(); + mocks.closeDriver.mockReset(); + + mocks.initSchema.mockResolvedValue(undefined); + mocks.closeDriver.mockResolvedValue(undefined); + mocks.getBySession.mockResolvedValue([]); + mocks.extract.mockResolvedValue({ nodes: [], edges: [] }); + mocks.upsertNode.mockImplementation(async (_d: any, c: any) => ({ + node: { + id: `n-${c.name}`, + type: c.type, + name: c.name, + description: c.description ?? "", + content: c.content, + status: "active", + validatedCount: 1, + sourceSessions: [], + communityId: null, + pagerank: 0, + createdAt: 0, + updatedAt: 0, + }, + isNew: true, + })); + mocks.upsertEdge.mockResolvedValue(undefined); + mocks.findByName.mockResolvedValue(null); + mocks.markExtracted.mockResolvedValue(undefined); + }); + + it("returns sessionsTotal=0 and skips everything when no unextracted sessions", async () => { + mocks.listUnextractedSessions.mockResolvedValue([]); + const log = vi.fn(); + + const result = await runBackfillExtraction({ + cfg: makeCfg(), + effectiveModel: "gpt-test", + options: {}, + log, + }); + + expect(result.sessionsTotal).toBe(0); + expect(result.sessionsProcessed).toBe(0); + expect(mocks.extract).not.toHaveBeenCalled(); + expect(mocks.closeDriver).toHaveBeenCalledTimes(1); + expect(log).toHaveBeenCalledWith(expect.stringContaining("没有需要提取的会话")); + }); + + it("requires an LLM model and throws a clear error when missing", async () => { + mocks.listUnextractedSessions.mockResolvedValue([]); + await expect( + runBackfillExtraction({ cfg: makeCfg(), effectiveModel: "", options: {}, log: vi.fn() }), + ).rejects.toThrow(/LLM model/); + expect(mocks.closeDriver).not.toHaveBeenCalled(); + }); + + it("requires neo4j.uri and throws a clear error when missing", async () => { + mocks.listUnextractedSessions.mockResolvedValue([]); + await expect( + runBackfillExtraction({ + cfg: { ...makeCfg(), neo4j: { uri: "", user: "", password: "" } } as any, + effectiveModel: "gpt-test", + options: {}, + log: vi.fn(), + }), + ).rejects.toThrow(/neo4j\.uri/); + }); + + it("aborts when the user declines the confirmation prompt", async () => { + mocks.listUnextractedSessions.mockResolvedValue([SAMPLE_SESSION]); + const prompt = vi.fn().mockResolvedValue("n"); + const log = vi.fn(); + + const result = await runBackfillExtraction({ + cfg: makeCfg(), + effectiveModel: "gpt-test", + options: {}, + log, + prompt, + }); + + expect(prompt).toHaveBeenCalledTimes(1); + expect(result.sessionsProcessed).toBe(0); + expect(result.sessionsSkipped).toBe(1); + expect(mocks.extract).not.toHaveBeenCalled(); + expect(mocks.markExtracted).not.toHaveBeenCalled(); + expect(mocks.closeDriver).toHaveBeenCalledTimes(1); + }); + + it("does not prompt when --yes is set and runs extraction", async () => { + mocks.listUnextractedSessions.mockResolvedValue([SAMPLE_SESSION]); + mocks.getUnextracted.mockResolvedValueOnce([ + { role: "user", content: "hello", turn_index: 1 }, + { role: "assistant", content: "hi", turn_index: 2 }, + ]).mockResolvedValueOnce([]); + mocks.extract.mockResolvedValueOnce({ + nodes: [ + { type: "TASK", name: "t1", description: "d", content: "c" }, + { type: "SKILL", name: "s1", description: "d", content: "c" }, + ], + edges: [ + { from: "t1", to: "s1", type: "USED_SKILL", instruction: "i" }, + ], + }); + const prompt = vi.fn(); + const log = vi.fn(); + + const result = await runBackfillExtraction({ + cfg: makeCfg(), + effectiveModel: "gpt-test", + options: { yes: true }, + log, + prompt, + }); + + expect(prompt).not.toHaveBeenCalled(); + expect(result.sessionsProcessed).toBe(1); + expect(result.nodesCreated).toBe(2); + expect(result.edgesCreated).toBe(1); + expect(result.batches).toBe(1); + expect(mocks.extract).toHaveBeenCalledTimes(1); + expect(mocks.markExtracted).toHaveBeenCalledWith(expect.anything(), "sid-abc-1234567890", 2); + expect(mocks.closeDriver).toHaveBeenCalledTimes(1); + }); + + it("filters sessions to the one specified by --session", async () => { + mocks.listUnextractedSessions.mockResolvedValue([ + { ...SAMPLE_SESSION, sessionId: "aaa" }, + { ...SAMPLE_SESSION, sessionId: "bbb" }, + ]); + mocks.getUnextracted.mockResolvedValue([]); + const log = vi.fn(); + + const result = await runBackfillExtraction({ + cfg: makeCfg(), + effectiveModel: "gpt-test", + options: { yes: true, session: "bbb" }, + log, + }); + + expect(result.sessionsTotal).toBe(1); + expect(result.sessionsProcessed).toBe(1); + expect(mocks.getUnextracted).toHaveBeenCalledWith(expect.anything(), "bbb", expect.any(Number)); + }); + + it("exits cleanly when --session matches no sessions", async () => { + mocks.listUnextractedSessions.mockResolvedValue([SAMPLE_SESSION]); + const log = vi.fn(); + + const result = await runBackfillExtraction({ + cfg: makeCfg(), + effectiveModel: "gpt-test", + options: { session: "does-not-exist" }, + log, + }); + + expect(result.sessionsTotal).toBe(0); + expect(mocks.extract).not.toHaveBeenCalled(); + expect(mocks.closeDriver).toHaveBeenCalledTimes(1); + }); + + it("lists sessions but does not extract under --dry-run", async () => { + mocks.listUnextractedSessions.mockResolvedValue([SAMPLE_SESSION]); + const prompt = vi.fn(); + const log = vi.fn(); + + const result = await runBackfillExtraction({ + cfg: makeCfg(), + effectiveModel: "gpt-test", + options: { dryRun: true }, + log, + prompt, + }); + + expect(prompt).not.toHaveBeenCalled(); + expect(result.sessionsSkipped).toBe(1); + expect(result.sessionsProcessed).toBe(0); + expect(mocks.extract).not.toHaveBeenCalled(); + expect(log).toHaveBeenCalledWith(expect.stringContaining("--dry-run")); + expect(mocks.closeDriver).toHaveBeenCalledTimes(1); + }); + + it("loops multiple batches until getUnextracted returns empty", async () => { + mocks.listUnextractedSessions.mockResolvedValue([SAMPLE_SESSION]); + mocks.getUnextracted + .mockResolvedValueOnce([ + { role: "user", content: "m1", turn_index: 1 }, + ]) + .mockResolvedValueOnce([ + { role: "user", content: "m2", turn_index: 2 }, + ]) + .mockResolvedValueOnce([]); + mocks.extract.mockResolvedValue({ nodes: [], edges: [] }); + const log = vi.fn(); + + const result = await runBackfillExtraction({ + cfg: makeCfg(), + effectiveModel: "gpt-test", + options: { yes: true, limit: 1 }, + log, + }); + + expect(result.batches).toBe(2); + expect(mocks.markExtracted).toHaveBeenCalledTimes(2); + expect(mocks.closeDriver).toHaveBeenCalledTimes(1); + }); + + it("records a session as skipped when getUnextracted throws", async () => { + mocks.listUnextractedSessions.mockResolvedValue([ + { ...SAMPLE_SESSION, sessionId: "good" }, + { ...SAMPLE_SESSION, sessionId: "bad" }, + ]); + const callCount = new Map(); + mocks.getUnextracted.mockImplementation(async (_d: any, sid: string) => { + if (sid === "bad") throw new Error("boom"); + const n = (callCount.get(sid) ?? 0) + 1; + callCount.set(sid, n); + if (n === 1) return [{ role: "user", content: "x", turn_index: 1 }]; + return []; + }); + mocks.extract.mockResolvedValue({ nodes: [], edges: [] }); + const log = vi.fn(); + + const result = await runBackfillExtraction({ + cfg: makeCfg(), + effectiveModel: "gpt-test", + options: { yes: true }, + log, + }); + + expect(result.sessionsProcessed).toBe(1); + expect(result.sessionsSkipped).toBe(1); + expect(result.sessionsTotal).toBe(2); + expect(mocks.closeDriver).toHaveBeenCalledTimes(1); + }); + + it("calls closeDriver even when listUnextractedSessions throws (try/finally)", async () => { + mocks.listUnextractedSessions.mockRejectedValue(new Error("neo4j down")); + const log = vi.fn(); + + await expect( + runBackfillExtraction({ + cfg: makeCfg(), + effectiveModel: "gpt-test", + options: { yes: true }, + log, + }), + ).rejects.toThrow("neo4j down"); + + expect(mocks.closeDriver).toHaveBeenCalledTimes(1); + }); + + it("does not warn about batch ceiling when session completes before the ceiling", async () => { + mocks.listUnextractedSessions.mockResolvedValue([SAMPLE_SESSION]); + const fullBatches = 3; + mocks.getUnextracted.mockImplementation(async () => { + const call = mocks.getUnextracted.mock.calls.length; + if (call < fullBatches) return Array.from({ length: 5 }, (_, i) => ({ role: "user", content: `m${i}`, turn_index: call * 5 + i })); + return [{ role: "user", content: "last", turn_index: fullBatches * 5 }]; + }); + mocks.extract.mockResolvedValue({ nodes: [], edges: [] }); + const log = vi.fn(); + + const result = await runBackfillExtraction({ + cfg: makeCfg({ compactTurnCount: 1 }), + effectiveModel: "gpt-test", + options: { yes: true, limit: 5 }, + log, + }); + + expect(result.batches).toBe(fullBatches); + expect(result.sessionsProcessed).toBe(1); + const warningCalls = log.mock.calls.filter(c => typeof c[0] === "string" && c[0].includes("达到批数上限")); + expect(warningCalls).toHaveLength(0); + }); + + it("warns about batch ceiling when the session genuinely has more messages than the ceiling allows", async () => { + mocks.listUnextractedSessions.mockResolvedValue([SAMPLE_SESSION]); + mocks.getUnextracted.mockResolvedValue(Array.from({ length: 5 }, (_, i) => ({ role: "user", content: `m${i}`, turn_index: i }))); + mocks.extract.mockResolvedValue({ nodes: [], edges: [] }); + const log = vi.fn(); + + const result = await runBackfillExtraction({ + cfg: makeCfg({ compactTurnCount: 1 }), + effectiveModel: "gpt-test", + options: { yes: true, limit: 5 }, + log, + }); + + expect(result.batches).toBe(50); + const warningCalls = log.mock.calls.filter(c => typeof c[0] === "string" && c[0].includes("达到批数上限")); + expect(warningCalls).toHaveLength(1); + }); +}); diff --git a/test/community-signature.test.ts b/test/community-signature.test.ts new file mode 100644 index 0000000..d7c8979 --- /dev/null +++ b/test/community-signature.test.ts @@ -0,0 +1,36 @@ +/** + * buildCommunityMemberSignature — 社区成员签名纯逻辑 + * 移植自上游 v1.x "reuse unchanged community summaries"（commit 1fdec04） + */ + +import { describe, it, expect } from "vitest"; +import { buildCommunityMemberSignature } from "../src/graph/community.ts"; + +describe("buildCommunityMemberSignature", () => { + it("成员顺序不影响签名（排序后哈希）", () => { + expect(buildCommunityMemberSignature(["a", "b", "c"])) + .toBe(buildCommunityMemberSignature(["c", "a", "b"])); + }); + + it("相同成员恒生成相同签名", () => { + expect(buildCommunityMemberSignature(["x", "y"])) + .toBe(buildCommunityMemberSignature(["x", "y"])); + }); + + it("成员构成不同则签名不同", () => { + expect(buildCommunityMemberSignature(["a", "b"])) + .not.toBe(buildCommunityMemberSignature(["a", "c"])); + expect(buildCommunityMemberSignature(["a", "b"])) + .not.toBe(buildCommunityMemberSignature(["a", "b", "c"])); + }); + + it("输出为 40 位小写 hex（sha1）", () => { + expect(buildCommunityMemberSignature(["a"])).toMatch(/^[0-9a-f]{40}$/); + }); + + it("不修改入参数组", () => { + const input = ["b", "a"]; + buildCommunityMemberSignature(input); + expect(input).toEqual(["b", "a"]); + }); +}); diff --git a/test/decay.test.ts b/test/decay.test.ts new file mode 100644 index 0000000..c6ab7a6 --- /dev/null +++ b/test/decay.test.ts @@ -0,0 +1,278 @@ +import { describe, it, expect } from "vitest"; +import { + normalizeImportance, + computeConfidence, + computeBeta, + scoreRecency, + scoreFrequency, + scoreIntrinsic, + scoreNode, + decideTierTransition, +} from "../src/graph/decay.ts"; +import { DEFAULT_CONFIG, type DecayConfig, type GmNode } from "../src/types.ts"; + +const cfg: DecayConfig = { ...DEFAULT_CONFIG.decay! }; +const NOW = Date.UTC(2026, 0, 15, 0, 0, 0); +const MS_PER_DAY = 86_400_000; + +function makeNode(overrides: Partial = {}): GmNode { + return { + id: "test-id", + type: "SKILL", + name: "test", + description: "", + content: "", + status: "active", + tier: "working", + validatedCount: 1, + sourceSessions: [], + communityId: null, + pagerank: 0, + createdAt: NOW - 10 * MS_PER_DAY, + updatedAt: NOW - 10 * MS_PER_DAY, + lastAccessedAt: NOW - 10 * MS_PER_DAY, + ...overrides, + }; +} + +describe("normalizeImportance", () => { + it("pagerank=0 时返回 0（即使 maxPagerank>0）", () => { + expect(normalizeImportance(0, 1.0)).toBe(0); + }); + + it("maxPagerank≤0 时返回 0（避免除零）", () => { + expect(normalizeImportance(5, 0)).toBe(0); + expect(normalizeImportance(5, -1)).toBe(0); + }); + + it("pagerank = maxPagerank 时返回 1", () => { + expect(normalizeImportance(0.5, 0.5)).toBe(1); + }); + + it("截断到 [0,1]", () => { + expect(normalizeImportance(2.0, 1.0)).toBe(1); + expect(normalizeImportance(-1, 1.0)).toBe(0); + }); +}); + +describe("computeConfidence", () => { + it("count=0 时 confidence=0", () => { + expect(computeConfidence(0)).toBe(0); + }); + + it("count=1 时 confidence=0.5", () => { + expect(computeConfidence(1)).toBeCloseTo(0.5, 6); + }); + + it("count 增大时饱和收敛到 1（永不达到）", () => { + expect(computeConfidence(10)).toBeLessThan(1); + expect(computeConfidence(100)).toBeLessThan(1); + expect(computeConfidence(100)).toBeGreaterThan(computeConfidence(10)); + }); + + it("负数按 0 处理", () => { + expect(computeConfidence(-5)).toBe(0); + }); +}); + +describe("computeBeta", () => { + it("core < working < peripheral（缓衰 → 促衰）", () => { + expect(computeBeta("core", cfg)).toBe(0.8); + expect(computeBeta("working", cfg)).toBe(1.0); + expect(computeBeta("peripheral", cfg)).toBe(1.3); + }); +}); + +describe("scoreRecency", () => { + it("刚刚访问（daysSince=0）→ 1.0", () => { + const node = makeNode({ lastAccessedAt: NOW }); + expect(scoreRecency(node, 0, NOW, cfg)).toBeCloseTo(1, 6); + }); + + it("importance=0 + working tier + 30 天 → recency ≈ 0.5（半衰期）", () => { + const node = makeNode({ tier: "working", lastAccessedAt: NOW - 30 * MS_PER_DAY }); + expect(scoreRecency(node, 0, NOW, cfg)).toBeCloseTo(0.5, 2); + }); + + it("高 importance 拉长 effectiveHL（衰减更慢）", () => { + const node = makeNode({ tier: "working", lastAccessedAt: NOW - 30 * MS_PER_DAY }); + const highImp = scoreRecency(node, 1.0, NOW, cfg); + const zeroImp = scoreRecency(node, 0, NOW, cfg); + expect(highImp).toBeGreaterThan(zeroImp); + expect(highImp).toBeGreaterThan(0.5); + }); + + it("tier=peripheral 比 tier=working 衰减更快", () => { + const days = 10; + const w = scoreRecency(makeNode({ tier: "working", lastAccessedAt: NOW - days * MS_PER_DAY }), 0, NOW, cfg); + const p = scoreRecency(makeNode({ tier: "peripheral", lastAccessedAt: NOW - days * MS_PER_DAY }), 0, NOW, cfg); + expect(p).toBeLessThan(w); + }); + + it("lastAccessedAt 缺失时回退到 updatedAt", () => { + const viaFallback = makeNode({ lastAccessedAt: 0, updatedAt: NOW - 5 * MS_PER_DAY }); + const direct = makeNode({ lastAccessedAt: NOW - 5 * MS_PER_DAY }); + expect(scoreRecency(viaFallback, 0, NOW, cfg)) + .toBeCloseTo(scoreRecency(direct, 0, NOW, cfg), 6); + }); +}); + +describe("scoreFrequency", () => { + it("count=0 时 base=0", () => { + expect(scoreFrequency(makeNode({ validatedCount: 0 }))).toBe(0); + }); + + it("count=1 时只返回 base（无 recentnessBonus）", () => { + const expected = 1 - Math.exp(-1 / 5); + expect(scoreFrequency(makeNode({ validatedCount: 1 }))).toBeCloseTo(expected, 6); + }); + + it("count > 1 时 base × (0.5 + 0.5*recentnessBonus)，结果 ≤ base", () => { + const node = makeNode({ + validatedCount: 3, + createdAt: NOW - 30 * MS_PER_DAY, + lastAccessedAt: NOW, + }); + const base = 1 - Math.exp(-3 / 5); + const score = scoreFrequency(node); + expect(score).toBeLessThanOrEqual(base); + expect(score).toBeGreaterThan(0); + }); + + it("访问越紧凑（avgGapDays 越小）recentnessBonus 越大", () => { + const tight = makeNode({ + validatedCount: 5, + createdAt: NOW - 4 * MS_PER_DAY, + lastAccessedAt: NOW, + }); + const sparse = makeNode({ + validatedCount: 5, + createdAt: NOW - 100 * MS_PER_DAY, + lastAccessedAt: NOW, + }); + expect(scoreFrequency(tight)).toBeGreaterThan(scoreFrequency(sparse)); + }); +}); + +describe("scoreIntrinsic", () => { + it("= importance × confidence", () => { + expect(scoreIntrinsic(0.5, 0.5)).toBeCloseTo(0.25, 6); + expect(scoreIntrinsic(1, 1)).toBe(1); + expect(scoreIntrinsic(0, 0.5)).toBe(0); + }); +}); + +describe("scoreNode", () => { + it("权重和为 1 时 composite 落在 [0,1]", () => { + const node = makeNode({ pagerank: 0.5, validatedCount: 5, lastAccessedAt: NOW }); + const r = scoreNode(node, 1.0, NOW, cfg); + expect(r.composite).toBeGreaterThanOrEqual(0); + expect(r.composite).toBeLessThanOrEqual(1); + }); + + it("新鲜高 PR 节点 composite 显著高于陈旧低 PR 节点", () => { + const fresh = makeNode({ pagerank: 1.0, validatedCount: 1, lastAccessedAt: NOW }); + const stale = makeNode({ + pagerank: 0, + validatedCount: 1, + lastAccessedAt: NOW - 90 * MS_PER_DAY, + }); + expect(scoreNode(fresh, 1.0, NOW, cfg).composite) + .toBeGreaterThan(scoreNode(stale, 1.0, NOW, cfg).composite); + }); + + it("权重和≠1 时自动归一化，composite 仍落在 [0,1]", () => { + const skewedCfg: DecayConfig = { + ...cfg, + recencyWeight: 0.5, + frequencyWeight: 0.5, + intrinsicWeight: 0.5, // 和=1.5 + }; + const node = makeNode({ + pagerank: 1.0, + validatedCount: 10, + lastAccessedAt: NOW, + updatedAt: NOW, + createdAt: NOW, + }); + const r = scoreNode(node, 1.0, NOW, skewedCfg); + expect(r.composite).toBeLessThanOrEqual(1); + expect(r.composite).toBeGreaterThanOrEqual(0); + }); + + it("权重和为 0 时回退到等权重，不抛错", () => { + const zeroCfg: DecayConfig = { + ...cfg, + recencyWeight: 0, + frequencyWeight: 0, + intrinsicWeight: 0, + }; + const node = makeNode({ pagerank: 0.5, validatedCount: 1, lastAccessedAt: NOW }); + const r = scoreNode(node, 1.0, NOW, zeroCfg); + expect(Number.isFinite(r.composite)).toBe(true); + }); +}); + +describe("decideTierTransition", () => { + const scoreLow = { composite: 0.1, recency: 0, frequency: 0, intrinsic: 0 }; + const scoreHigh = { composite: 0.9, recency: 0.9, frequency: 0.9, intrinsic: 0.9 }; + const scoreMid = { composite: 0.5, recency: 0.5, frequency: 0.5, intrinsic: 0 }; + + it("core + composite 低 + count 低 → working", () => { + const node = makeNode({ tier: "core", validatedCount: 1 }); + expect(decideTierTransition(node, scoreLow, 0, cfg, NOW)).toBe("working"); + }); + + it("core + composite 高 → 保持 core", () => { + const node = makeNode({ tier: "core", validatedCount: 20 }); + expect(decideTierTransition(node, scoreHigh, 0.9, cfg, NOW)).toBeNull(); + }); + + it("working + composite < pct → peripheral", () => { + const node = makeNode({ tier: "working", validatedCount: 1 }); + expect(decideTierTransition(node, scoreLow, 0, cfg, NOW)).toBe("peripheral"); + }); + + it("working + 陈旧（age > peripheralAgeDays）+ count 低 → peripheral", () => { + const node = makeNode({ + tier: "working", + validatedCount: 1, + createdAt: NOW - (cfg.peripheralAgeDays + 1) * MS_PER_DAY, + }); + expect(decideTierTransition(node, scoreMid, 0, cfg, NOW)).toBe("peripheral"); + }); + + it("working + 陈旧但 count 充足 → 保持 working", () => { + const node = makeNode({ + tier: "working", + validatedCount: 5, + createdAt: NOW - (cfg.peripheralAgeDays + 1) * MS_PER_DAY, + }); + expect(decideTierTransition(node, scoreMid, 0, cfg, NOW)).toBeNull(); + }); + + it("peripheral + count 充足 + composite 高 → working", () => { + const node = makeNode({ tier: "peripheral", validatedCount: 5 }); + expect(decideTierTransition(node, scoreMid, 0, cfg, NOW)).toBe("working"); + }); + + it("peripheral + count 不足 → 保持 peripheral", () => { + const node = makeNode({ tier: "peripheral", validatedCount: 1 }); + expect(decideTierTransition(node, scoreHigh, 0, cfg, NOW)).toBeNull(); + }); + + it("working + count + composite + importance 都高 → core", () => { + const node = makeNode({ tier: "working", validatedCount: 15 }); + expect(decideTierTransition(node, scoreHigh, 0.9, cfg, NOW)).toBe("core"); + }); + + it("working + count + composite 高但 importance 不足 → 保持 working", () => { + const node = makeNode({ tier: "working", validatedCount: 15 }); + expect(decideTierTransition(node, scoreHigh, 0.5, cfg, NOW)).toBeNull(); + }); + + it("tier undefined 按 working 处理", () => { + const node = makeNode({ tier: undefined as unknown as GmNode["tier"], validatedCount: 1 }); + expect(decideTierTransition(node, scoreLow, 0, cfg, NOW)).toBe("peripheral"); + }); +}); diff --git a/test/integration.assemble.test.ts b/test/integration.assemble.test.ts index 594458e..705985b 100644 --- a/test/integration.assemble.test.ts +++ b/test/integration.assemble.test.ts @@ -24,12 +24,14 @@ function makeNode(over: Partial): GmNode { description: over.description ?? "desc", content: over.content ?? "content body", status: over.status ?? "active", + tier: over.tier ?? "working", validatedCount: over.validatedCount ?? 1, sourceSessions: over.sourceSessions ?? ["s1"], communityId: over.communityId ?? null, pagerank: over.pagerank ?? 0, createdAt: over.createdAt ?? Date.now(), updatedAt: over.updatedAt ?? Date.now(), + lastAccessedAt: over.lastAccessedAt ?? Date.now(), }; } diff --git a/test/integration.graph.test.ts b/test/integration.graph.test.ts index 9dc2e90..8b52d10 100644 --- a/test/integration.graph.test.ts +++ b/test/integration.graph.test.ts @@ -12,12 +12,14 @@ import { describe, it, expect, beforeAll, afterAll } from "vitest"; import type { Driver } from "neo4j-driver"; import { getDriver, initSchema, closeDriver, getSession } from "../src/store/db.ts"; import { - upsertNode, upsertEdge, saveVector, findById, deprecate, + upsertNode, upsertEdge, saveVector, findById, deprecate, getCommunitySummary, } from "../src/store/store.ts"; import { personalizedPageRank, computeGlobalPageRank, } from "../src/graph/pagerank.ts"; -import { detectCommunities, getCommunityPeers } from "../src/graph/community.ts"; +import { + detectCommunities, getCommunityPeers, summarizeCommunities, buildCommunityMemberSignature, +} from "../src/graph/community.ts"; import { detectDuplicates, dedup } from "../src/graph/dedup.ts"; import { runMaintenance } from "../src/graph/maintenance.ts"; import { DEFAULT_CONFIG, type GmConfig } from "../src/types.ts"; @@ -187,6 +189,62 @@ describe.skipIf(!ENABLED)("graph layer integration (GDS, Docker)", () => { } }); + it("summarizeCommunities：社区成员未变时复用摘要，不重调 LLM", async () => { + const memberIds = [nodeIds["gmpsrc-deploy"], nodeIds["gmpsrc-compose"]]; + + // 生产不变量：detectCommunities 会先给成员节点写入 communityId， + // pruneCommunitySummaries 只保留仍被 active 成员引用的社区 — 不先 SET 会被 prune 删掉 + const prepare = getSession(driver); + try { + await prepare.run( + "MATCH (n:MemoryNode) WHERE n.id IN $ids SET n.communityId = $cid", + { ids: memberIds, cid: "c-reuse-test" }, + ); + } finally { + await prepare.close(); + } + + let llmCalls = 0; + const llm = async () => { + llmCalls += 1; + return "容器部署与编排技能"; + }; + + const first = await summarizeCommunities(driver, new Map([["c-reuse-test", memberIds]]), llm); + const second = await summarizeCommunities(driver, new Map([["c-reuse-test", memberIds]]), llm); + + expect(first).toBe(1); + expect(second).toBe(0); + expect(llmCalls).toBe(1); + + const summary = await getCommunitySummary(driver, "c-reuse-test"); + expect(summary?.summary).toBe("容器部署与编排技能"); + expect(summary?.memberSignature).toBe(buildCommunityMemberSignature(memberIds)); + + // detectCommunities 每轮按成员数重编号（c-1..c-N），ID 变但成员相同 → 按签名跨社区复用 + const third = await summarizeCommunities( + driver, new Map([["c-reuse-renumbered", memberIds]]), llm, + ); + expect(third).toBe(0); + expect(llmCalls).toBe(1); + const renumbered = await getCommunitySummary(driver, "c-reuse-renumbered"); + expect(renumbered?.summary).toBe("容器部署与编排技能"); + expect(renumbered?.memberSignature).toBe(buildCommunityMemberSignature(memberIds)); + + const cleanup = getSession(driver); + try { + await cleanup.run( + "MATCH (c:Community) WHERE c.id IN ['c-reuse-test', 'c-reuse-renumbered'] DELETE c", + ); + await cleanup.run( + "MATCH (n:MemoryNode) WHERE n.id IN $ids SET n.communityId = null", + { ids: memberIds }, + ); + } finally { + await cleanup.close(); + } + }); + it("detectDuplicates：gmpsrc-* 无 embedding，函数不抛错", async () => { let passed = false; await expectDimSafe(async () => { diff --git a/test/session-identity.test.ts b/test/session-identity.test.ts index c2ec70c..8b069bf 100644 --- a/test/session-identity.test.ts +++ b/test/session-identity.test.ts @@ -1,7 +1,12 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; +import { isCronSessionKey } from "../src/types.ts"; const mocks = vi.hoisted(() => ({ getBySession: vi.fn(async () => []), + saveMessage: vi.fn(async () => {}), + getMaxTurnIndex: vi.fn(async () => 0), + getUnextracted: vi.fn(async () => []), + isTurnExtracted: vi.fn(async () => false), recall: vi.fn(async () => ({ nodes: [{ id: "recalled-node" }], edges: [], @@ -25,10 +30,11 @@ vi.mock("../src/store/db.ts", () => ({ })); vi.mock("../src/store/store.ts", () => ({ - saveMessage: async () => {}, - getUnextracted: async () => [], + saveMessage: mocks.saveMessage, + getUnextracted: mocks.getUnextracted, + getMaxTurnIndex: mocks.getMaxTurnIndex, markExtracted: async () => {}, - isTurnExtracted: async () => false, + isTurnExtracted: mocks.isTurnExtracted, upsertNode: async () => ({ node: {}, isNew: false }), upsertEdge: async () => {}, findByName: async () => null, @@ -81,11 +87,28 @@ import graphMemoryProPlugin from "../index.ts"; type HookHandler = (event: Record, context: Record) => Promise; type EngineHarness = { readonly bootstrap: (params: { readonly sessionId: string; readonly sessionKey?: string }) => Promise; + readonly ingest: (params: { + readonly sessionId: string; + readonly sessionKey?: string; + readonly message: unknown; + readonly isHeartbeat?: boolean; + }) => Promise<{ readonly ingested: boolean }>; readonly assemble: (params: { readonly sessionId: string; readonly sessionKey?: string; readonly messages: readonly unknown[]; }) => Promise; + readonly compact: (params: { readonly sessionId: string; readonly sessionKey?: string }) => Promise<{ + readonly ok: boolean; + readonly compacted: boolean; + readonly reason?: string; + }>; + readonly afterTurn: (params: { + readonly sessionId: string; + readonly sessionKey?: string; + readonly messages: readonly unknown[]; + readonly prePromptMessageCount: number; + }) => Promise; readonly prepareSubagentSpawn: (params: { readonly parentSessionKey: string; readonly childSessionKey: string; @@ -93,7 +116,7 @@ type EngineHarness = { }) => Promise<{ readonly rollback: () => void }>; }; -function registerPlugin(): { readonly hooks: Map; readonly engine: EngineHarness } { +function registerPlugin(pluginConfig: Record = {}): { readonly hooks: Map; readonly engine: EngineHarness } { const hooks = new Map(); let engine: EngineHarness | undefined; graphMemoryProPlugin.register({ @@ -104,7 +127,7 @@ function registerPlugin(): { readonly hooks: Map; readonly error: () => {}, }, config: {}, - pluginConfig: {}, + pluginConfig, resolvePath: (path: string) => path, on: (event: string, handler: HookHandler) => { hooks.set(event, handler); }, registerContextEngine: (_id: string, factory: () => EngineHarness) => { engine = factory(); }, @@ -165,3 +188,167 @@ describe("session identity", () => { ); }); }); + +describe("cron session gating (cron 配置)", () => { + beforeEach(() => { + mocks.getBySession.mockClear(); + mocks.saveMessage.mockClear(); + mocks.getMaxTurnIndex.mockClear(); + mocks.getUnextracted.mockClear(); + mocks.isTurnExtracted.mockClear(); + mocks.recall.mockClear(); + mocks.assembleContext.mockClear(); + mocks.runMaintenance.mockClear(); + }); + + // host 契约：sessionId 是随机 transcript UUID，cron 标记在 sessionKey 上 + const CRON_KEY = "agent:agent-1:cron:daily-report"; + const CRON_SID = "0f1e2d3c-4b5a-6978-8976-543210fedcba"; + + it("isCronSessionKey 按 sessionKey 段匹配 cron 标记", () => { + expect(isCronSessionKey("cron:job-1")).toBe(true); + expect(isCronSessionKey("agent:agent-1:cron:daily")).toBe(true); + expect(isCronSessionKey("agent:agent-1:cron:daily:run:r1")).toBe(true); + expect(isCronSessionKey("agent:main")).toBe(false); + expect(isCronSessionKey("agent:cron-daily:main")).toBe(false); + expect(isCronSessionKey("scheduled-cron:x")).toBe(false); + expect(isCronSessionKey("")).toBe(false); + expect(isCronSessionKey(undefined)).toBe(false); + expect(isCronSessionKey(null)).toBe(false); + }); + + it("默认配置（全 true）下 cron session_end 仍执行 finalize 与图维护（向后兼容）", async () => { + const handler = registerPlugin().hooks.get("session_end"); + if (!handler) throw new Error("session_end hook was not registered"); + + await handler({ sessionId: CRON_SID, sessionKey: CRON_KEY }, {}); + + expect(mocks.getBySession).toHaveBeenCalledWith({}, CRON_SID); + expect(mocks.runMaintenance).toHaveBeenCalledTimes(1); + }); + + it("默认配置下 cron session 正常入库", async () => { + const { engine } = registerPlugin(); + + await expect(engine.ingest({ sessionId: CRON_SID, sessionKey: CRON_KEY, message: { role: "user", content: "hi" } })) + .resolves.toEqual({ ingested: true }); + expect(mocks.saveMessage).toHaveBeenCalledTimes(1); + }); + + it("finalizeAndMaintain=false 时 cron session 跳过 finalize 与图维护", async () => { + const handler = registerPlugin({ cron: { enabled: true, finalizeAndMaintain: false } }).hooks.get("session_end"); + if (!handler) throw new Error("session_end hook was not registered"); + + await handler({ sessionId: CRON_SID, sessionKey: CRON_KEY }, {}); + + expect(mocks.getBySession).not.toHaveBeenCalled(); + expect(mocks.runMaintenance).not.toHaveBeenCalled(); + }); + + it("finalizeAndMaintain=true 时 cron session 执行 finalize 与图维护", async () => { + const handler = registerPlugin({ cron: { enabled: true, finalizeAndMaintain: true } }).hooks.get("session_end"); + if (!handler) throw new Error("session_end hook was not registered"); + + await handler({ sessionId: CRON_SID, sessionKey: CRON_KEY }, {}); + + expect(mocks.getBySession).toHaveBeenCalledWith({}, CRON_SID); + expect(mocks.runMaintenance).toHaveBeenCalledTimes(1); + }); + + it("finalizeAndMaintain=true 不影响普通会话的既有行为", async () => { + const handler = registerPlugin({ cron: { enabled: true, finalizeAndMaintain: false } }).hooks.get("session_end"); + if (!handler) throw new Error("session_end hook was not registered"); + + await handler({ sessionId: "normal-session", sessionKey: "agent:main" }, {}); + + expect(mocks.runMaintenance).toHaveBeenCalledTimes(1); + }); + + it("enabled=false 时 cron session 不召回、不入库、不注入图谱上下文", async () => { + const { hooks, engine } = registerPlugin({ cron: { enabled: false } }); + + const beforeAgentStart = hooks.get("before_agent_start"); + if (!beforeAgentStart) throw new Error("before_agent_start hook was not registered"); + await beforeAgentStart({ prompt: "daily digest" }, { sessionKey: CRON_KEY }); + expect(mocks.recall).not.toHaveBeenCalled(); + + await expect(engine.ingest({ sessionId: CRON_SID, sessionKey: CRON_KEY, message: { role: "user", content: "hi" } })) + .resolves.toEqual({ ingested: false }); + expect(mocks.saveMessage).not.toHaveBeenCalled(); + + await engine.assemble({ sessionId: CRON_SID, sessionKey: CRON_KEY, messages: [] }); + expect(mocks.assembleContext).not.toHaveBeenCalled(); + }); + + it("enabled=false 总开关：cron afterTurn 跳过入库回填，compact 跳过提取", async () => { + const { engine } = registerPlugin({ cron: { enabled: false } }); + + await engine.afterTurn({ + sessionId: CRON_SID, + sessionKey: CRON_KEY, + messages: [{ role: "user", content: "hi" }], + prePromptMessageCount: 0, + }); + expect(mocks.saveMessage).not.toHaveBeenCalled(); + expect(mocks.isTurnExtracted).not.toHaveBeenCalled(); + + const res = await engine.compact({ sessionId: CRON_SID, sessionKey: CRON_KEY }); + expect(res).toEqual({ ok: true, compacted: false, reason: "cron session graph disabled" }); + expect(mocks.getUnextracted).not.toHaveBeenCalled(); + }); + + it("enabled=true 时 cron session 正常入库", async () => { + const { engine } = registerPlugin({ cron: { enabled: true } }); + + await engine.ingest({ sessionId: CRON_SID, sessionKey: CRON_KEY, message: { role: "user", content: "hi" } }); + + expect(mocks.saveMessage).toHaveBeenCalledTimes(1); + }); + + it("extract=false 时 cron session 消息仍入库缓冲但不触发提取", async () => { + const { engine } = registerPlugin({ cron: { enabled: true, extract: false } }); + + await engine.afterTurn({ + sessionId: CRON_SID, + sessionKey: CRON_KEY, + messages: [{ role: "user", content: "hi" }], + prePromptMessageCount: 0, + }); + + expect(mocks.saveMessage).toHaveBeenCalledTimes(1); + expect(mocks.isTurnExtracted).not.toHaveBeenCalled(); + }); + + it("extract=true 时 cron session 触发提取", async () => { + const { engine } = registerPlugin({ cron: { enabled: true, extract: true } }); + + await engine.afterTurn({ + sessionId: CRON_SID, + sessionKey: CRON_KEY, + messages: [{ role: "user", content: "hi" }], + prePromptMessageCount: 0, + }); + // afterTurn 内的 extractTurnKnowledge 是 fire-and-forget，flush 微任务后再断言 + await new Promise(resolve => setTimeout(resolve, 0)); + + expect(mocks.isTurnExtracted).toHaveBeenCalledTimes(1); + }); + + it("extract=false 时 cron session compact 直接跳过提取", async () => { + const { engine } = registerPlugin({ cron: { enabled: true, extract: false } }); + + const res = await engine.compact({ sessionId: CRON_SID, sessionKey: CRON_KEY }); + + expect(res).toEqual({ ok: true, compacted: false, reason: "cron session extraction disabled" }); + expect(mocks.getUnextracted).not.toHaveBeenCalled(); + }); + + it("cron 任务设置自定义 sessionKey（无 cron 段）时按普通会话处理", async () => { + const { engine } = registerPlugin({ cron: { enabled: false } }); + + await expect(engine.ingest({ sessionId: CRON_SID, sessionKey: "agent:my-custom-key", message: { role: "user", content: "hi" } })) + .resolves.toEqual({ ingested: true }); + + expect(mocks.saveMessage).toHaveBeenCalledTimes(1); + }); +});