Conversation
- memory_store: 新增 importance/scope 参数，严格校验不静默降级 - memory_recall: 新增 domain/scope 过滤参数 - memory_update: 新增工具，按 URI 更新记忆保留旧属性 - memory_store_global: 补充 memory_type 校验 - 召回新增 disclosure 精确匹配通道（第三路 RRF 融合） - _rrf_fuse 泛化为 *channels 支持任意数量检索通道 - _rerank_by_signal 新增 disclosure 匹配加成（默认 +0.25） - prompts.py 新增 VALID_TOOL_SCOPES/VALID_TOOL_DOMAINS 校验常量 - docs/maintenance-agent-spec.md 后台整理系统设计规格
- expire_stale_memories/mark_consolidated 写入时补上 deprecated_at - 迁移补丁: 已废弃但缺 deprecated_at 的记录用 created_at 回填 - 新增 maintenance/links.py: 关联表 CRUD (memory_links) - 新增 maintenance/purge.py: 按 deprecated_at + N 天宽限期物理清理 - _collect_kb_doc_ids_for_filters 同时收集 URI - _delete_by_filters/_clear_by_filters 级联清理关联边 - MemoryManager 新增 link_manager 属性和 purge_deprecated 方法
- maintenance/prompts.py: 整理师/分析师/审核员默认模板，使用 string.Template ($var) 避免 JSON 花括号问题，build_prompt 支持 override/extra 选配 - maintenance/scheduler.py: MaintenanceScheduler 使用 AstrBot cron_manager.add_basic_job 注册定时任务，single-flight 防并发 - main.py: 集成调度器（initialize/on_loaded 启动，terminate 停止） - 自动清理 purge 作为首个 cron 任务注册
Critical:
- purge 向量删除改为按精确 kb_doc_id 逐条删除，不再用宽泛过滤器
误删宽限期内记录；向量删除全失败时不继续下游清理
- deprecated_at 迁移回填改用 datetime('now')（迁移时间），
不再用 created_at 导致历史废弃记忆被立即清理
High:
- memory_update 改为先写后删，写入失败不删旧记录，
删除失败回滚新记录；迁移关联边到新 URI
- scheduler stop() 改用 cron_mgr.delete_job()（AstrBot 4.26.8 API），
注销失败保留 job_id 不清空
- KB 迁移后重新绑定 link_manager 到目标 vec_db
Warning:
- disclosure 召回扫描上限从 30 提升到 200（可配置），
加 ORDER BY created_at DESC 确保确定性
- _delete_by_filters/_clear_by_filters 初始化 uris=[]
防止异常路径 UnboundLocalError
- .ccg/ 加入 .gitignore 并从 git 索引移除
High: - purge 跟踪成功集合 (doc_id, uri)，只用成功子集清理 KB/links - memory_update 删除前读取双向关联（出边+入边），迁移到新 URI - memory_update 拒绝 global 记忆，保留 visibility/subject/owner_sender_ids - KB 迁移时导出源关联表 → 目标建表 → 导入，不丢失关联数据 - _delete_by_filters 候选收集失败时禁止进入删除 Warning: - disclosure_scan_limit 加 try/except + 边界约束 [10, 1000] - deprecated_at 迁移改用 Python UTC ISO 格式，与 purge cutoff 一致 - 审核员 prompt 破坏性操作默认 reject，非破坏性默认 approve - scheduler _run_purge 异常重新抛出，cron manager 记录为失败
High:
- 新增 MemoryManager.replace_memory() API，完整保留原始 metadata
（租户/作用域/归属/关联），tool_update 改用此 API 替代
store_memory+forget_memory 组合
- 关联查询支持 limit=0（无限制），replace_memory 内部处理
双向关联迁移，查询/写入失败记录日志
- KB 迁移先导出源关联表再删除源记录，导入后校验数量一致性
- purge 返回结构化结果 {purged, links_cleaned, failed, errors, partial}，
scheduler 检查 errors 非空时抛出异常让 cron 记录为失败
Warning:
- disclosure 召回改为分页扫描（每页 200，上限可配置至 5000），
直到找到足够匹配或遍历完所有候选，不再有硬截断漏召回
- prompts.py: MEMORY_EXTRACTION/RECALL_QUERY/CONSOLIDATION 迁移 string.Template - main.py: 三处 .format() 改为 .substitute() - _conf_schema.json: 补 maintenance_* / auto_purge_* / context_* / persona_* 配置项 Phase 1 剩余项全部完成，级联清理路径已验证（forget/clear/rebuild 均走 link-aware executor）
feat(phase-1): 收尾 prompt Template 迁移 + 配置 schema 补全
- maintenance/llm.py: 唯一 LLM 入口，pair-hash 磁盘缓存，三态裁决，调用上限 - maintenance/runner.py: 执行管线框架（purge→organizer→analyst→reviewer），结构化报告 - maintenance/scheduler.py: 接入 runner，注册整理周期 cron job - _conf_schema.json: 补 maintenance_cron 和 maintenance_max_llm_calls Phase 2 核心框架完成，具体 Agent 实现留 Phase 3/4/5
feat(phase-2): 执行管线框架 + LLM 唯一入口 + MaintenanceReport
- maintenance/agents/organizer.py: 整理师 Agent 实现 - 余弦 ≥0.9 预筛 merge 候选对 - LLM 裁决 merge/link/none - 调用上限约束 - memory_manager.py: 补 get_all_active_memories + merge_memories - merge 走 supersede 语义：新建合并记忆 + 旧记忆标 deprecated + 写 supersedes 边 - 不物理删除，由 purge 统一清理 - maintenance/runner.py: 接入 OrganizerAgent Phase 3 核心功能完成，质量精炼（archive/update）留 Phase 5 完善
feat(phase-3): 整理师 Agent（预筛 + supersede merge）
- maintenance/agents/analyst.py: 分析师 Agent 实现 - 余弦 ≥0.7 预筛 link 候选对 - 排除已连边（TODO: 接入 memory_links 查询） - 矛盾检测（简化版：内容重叠度 + 时间差） - 对话历史拉取接口预留（待 AstrBot API 确认） - maintenance/runner.py: 接入 AnalystAgent - _conf_schema.json: 补 analyst 相关配置 Phase 4 核心功能完成，对话历史拉取待 AstrBot conversation_manager API 确认后实现
feat(phase-4): 分析师 Agent（关联发现 + 矛盾检测）
- maintenance/agents/reviewer.py: 审核员 Agent - 复核 merge/archive/update/new_link 操作建议 - approve/reject verdicts - 置信度 < 0.5 标记 controversial，触发人工升级 - maintenance/runner.py: 互审模式 + 操作执行 - 阶段 3.5: 驳回理由回传修正（最多 2 轮） - 阶段 4: 按审核结果执行操作（approve 执行 / reject 跳过 / controversial 待人工） - _execute_operation: merge/archive/update/new_link/contradiction 五种操作执行 - merge 走 supersede 语义（新建 + 废弃旧 + 写 supersedes 边） - _conf_schema.json: 补 reviewer 相关配置 Phase 5 核心功能完成，人工升级通知待 Phase 6/7 完善
feat(phase-5): 审核员 Agent + 互审模式 + 操作执行
- memory_manager.py: recall_memories 添加关联注入 - 单跳查询，最多 3 条关联记忆 - 只注入 related/supports/context，排除 contradicts/supersedes - 只召回未 deprecated 的关联记忆 - 标记 _is_linked 便于识别 - 新增 _inject_linked_memories 和 _get_memory_by_uri 方法 Phase 6 完成
feat(phase-6): 召回时关联记忆注入
- _conf_schema.json: 补全 organizer/analyst 开关 + 待审通知配置 - README.md: - 配置说明表添加后台整理相关配置 - 工作原理添加后台整理说明 - 新增「后台记忆整理」章节（角色/流程/安全机制/配置建议） Phase 7 完成
feat(phase-7): 配置 schema 补全 + README 后台整理说明
1. _get_memory_by_uri SELECT text 而非 content 2. _is_memory_visible 使用 _current_owner_user_id + _is_visible_shared_personal 3. replace 回滚时同时 _unregister_kb_documents 4. organizer 分组：conversation 按完整 UMO 精确匹配，group 按 session 5. _is_memory_visible：conversation 按完整 UMO 匹配，避免私聊/群聊 ID 碰撞 6. all_users 召回模式下关联记忆跳过可见性过滤
1. get_all_active_memories 通过 FAISS reconstruct 加载向量 2. runner 改用 merge_memories / deprecate_memory / replace_memory 3. reviewer 缺失裁决时 fail closed（跳过而非放行） 4. analyst 候选对按 scope/owner/UMO 分组（与 organizer 一致） 5. maintenance_model_id 为空时回退到全局默认 provider 6. merge_memories 改用 vec_db.insert（自动生成 embedding） 7. 新增 deprecate_memory 方法（delete + re-insert 保持一致性） 8. 失败操作写入 report.errors（scheduler 可感知） 9. ruff format 全量格式化
1. merge_memories 写入 memory_content 而非 content 2. organizer 用 consumed_uris 防止源 URI 重复进入多个 merge 3. maintenance_enabled 时禁用旧 _maybe_consolidate_memories 4. reviewer 禁用/缺失裁决时 new_link 直接执行，destructive 仍 fail closed 5. get_all_active_memories 支持 offset 分页，organizer/analyst 遍历全部页 6. analyst 候选对排除已有 memory_links 边 7. 裁决缓存 key 使用解析后的 provider ID（_resolve_model_id）
1. runner 争议项写入 KV maintenance_pending_review 队列 2. scheduler/runner 接收 kv_put/kv_get 回调 3. main.py 新增 /memory review 命令（list/approve/reject/clear） 4. approve 时通过 runner._execute_operation 执行被批准的操作 5. main.py 传递 KV 回调给 MaintenanceScheduler
…时序 / LLM 超时 / 对话历史 1. deprecate_memory 先插后删（插入失败不影响原记录） 2. runner 执行前检查 global 作用域，转人工审批队列 3. merge_memories 任一源 deprecate 失败则回滚新记录并报失败 4. _run_reviewer 拉取源记录传入 original_data 供审核员对比 5. /memory review approve 先执行后标记（失败保留可重试） 6. LLM 调用加 asyncio.wait_for 超时（默认 120s，可配置） 7. analyst 实现对话历史拉取（conversation_manager API，三轮限制）
…onfidence 字段 1. 旧巩固先写摘要再标记源（store 失败不丢数据） 2. merge 预验证所有源 + 回滚时恢复已 deprecate 的源和 supersedes 边 3. _op_touches_global 查询实际 metadata 而非 URI 子串 4. deprecate_memory 同步 KBDocument（注册新 ID / 注销旧 ID） 5. reviewer 禁用时 destructive 操作转待审队列而非静默丢弃 6. 关联召回同时查出边和入边（对称关系双向可召回） 7. reviewer prompt 输出格式加 confidence 字段（驱动 controversial 标记）
… top_k 截断 / 待审通知 1. ruff --select I 修复 4 处 import 排序（CI 通过） 2. deprecate_memory 删除旧向量失败时返回 False 3. merge_memories 插入后注册 KBDocument + 同步统计 4. merge 回滚恢复源时同步 KB（注册新 / 注销旧） 5. organizer/analyst 预筛按 batch_size 截断（默认 200） 6. 关联注入后截断到 top_k 7. 待审项入队时推送通知（review_notify_enabled + review_notify_umo）
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Sorry @piexian, your pull request is larger than the review limit of 150000 diff characters
Reviewer's Guide实现一个完整的离线维护系统，用于长期记忆，包括链接存储、清理/合并/更新操作、基于 LLM 的组织者/分析师/审阅者代理、由 cron 驱动的调度器、更丰富的回忆/排序能力，以及管理员审查工具，同时收紧元数据/可见性语义并完善文档。 Sequence diagram for a maintenance cycle with merge and link operationssequenceDiagram
participant CronManager
participant MaintenanceScheduler
participant MaintenanceRunner
participant MemoryManager
participant MaintenanceLLM
participant OrganizerAgent
participant MemoryLinkManager
CronManager->>MaintenanceScheduler: _run_maintenance_cycle(**kwargs)
activate MaintenanceScheduler
MaintenanceScheduler->>MaintenanceRunner: run_cycle()
activate MaintenanceRunner
Note over MaintenanceRunner,MemoryManager: Phase 0: purge
MaintenanceRunner->>MemoryManager: purge_deprecated(after_days)
MemoryManager->>MemoryLinkManager: delete_links_for_uris(uris)
MemoryLinkManager-->>MemoryManager: links_cleaned
Note over MaintenanceRunner,OrganizerAgent: Phase 1: organizer
MaintenanceRunner->>OrganizerAgent: run(owner_filter)
activate OrganizerAgent
loop for each merge candidate
OrganizerAgent->>MaintenanceLLM: judge_relation(text_a, text_b, cosine, model_id)
MaintenanceLLM-->>OrganizerAgent: LLMVerdict(verdict, fused_text)
end
OrganizerAgent-->>MaintenanceRunner: manifest.merge
deactivate OrganizerAgent
Note over MaintenanceRunner,MemoryManager: Execute approved merge op
MaintenanceRunner->>MaintenanceRunner: _execute_operation(op)
MaintenanceRunner->>MaintenanceRunner: _execute_merge(op)
MaintenanceRunner->>MemoryManager: merge_memories(source_uris, merged_content, reason, created_by)
activate MemoryManager
MemoryManager->>MemoryManager: deprecate_memory(old_uri, reason)
MemoryManager->>MemoryLinkManager: add_link(source_uri=new_uri, target_uri=old_uri, relation_type="supersedes", reason, confidence, created_by)
MemoryLinkManager-->>MemoryManager: bool
MemoryManager-->>MaintenanceRunner: {success, merged_uri}
deactivate MemoryManager
MaintenanceRunner-->>MaintenanceScheduler: MaintenanceReport
deactivate MaintenanceRunner
MaintenanceScheduler-->>CronManager: done
deactivate MaintenanceScheduler
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your Experience访问你的 dashboard 来：
Getting HelpOriginal review guide in EnglishReviewer's GuideImplements a full offline maintenance system for long-term memory, including link storage, purge/merge/update operations, LLM-based organizer/analyst/reviewer agents, a cron-driven scheduler, richer recall/ranking, and admin review tooling, while tightening metadata/visibility semantics and documentation. Sequence diagram for a maintenance cycle with merge and link operationssequenceDiagram
participant CronManager
participant MaintenanceScheduler
participant MaintenanceRunner
participant MemoryManager
participant MaintenanceLLM
participant OrganizerAgent
participant MemoryLinkManager
CronManager->>MaintenanceScheduler: _run_maintenance_cycle(**kwargs)
activate MaintenanceScheduler
MaintenanceScheduler->>MaintenanceRunner: run_cycle()
activate MaintenanceRunner
Note over MaintenanceRunner,MemoryManager: Phase 0: purge
MaintenanceRunner->>MemoryManager: purge_deprecated(after_days)
MemoryManager->>MemoryLinkManager: delete_links_for_uris(uris)
MemoryLinkManager-->>MemoryManager: links_cleaned
Note over MaintenanceRunner,OrganizerAgent: Phase 1: organizer
MaintenanceRunner->>OrganizerAgent: run(owner_filter)
activate OrganizerAgent
loop for each merge candidate
OrganizerAgent->>MaintenanceLLM: judge_relation(text_a, text_b, cosine, model_id)
MaintenanceLLM-->>OrganizerAgent: LLMVerdict(verdict, fused_text)
end
OrganizerAgent-->>MaintenanceRunner: manifest.merge
deactivate OrganizerAgent
Note over MaintenanceRunner,MemoryManager: Execute approved merge op
MaintenanceRunner->>MaintenanceRunner: _execute_operation(op)
MaintenanceRunner->>MaintenanceRunner: _execute_merge(op)
MaintenanceRunner->>MemoryManager: merge_memories(source_uris, merged_content, reason, created_by)
activate MemoryManager
MemoryManager->>MemoryManager: deprecate_memory(old_uri, reason)
MemoryManager->>MemoryLinkManager: add_link(source_uri=new_uri, target_uri=old_uri, relation_type="supersedes", reason, confidence, created_by)
MemoryLinkManager-->>MemoryManager: bool
MemoryManager-->>MaintenanceRunner: {success, merged_uri}
deactivate MemoryManager
MaintenanceRunner-->>MaintenanceScheduler: MaintenanceReport
deactivate MaintenanceRunner
MaintenanceScheduler-->>CronManager: done
deactivate MaintenanceScheduler
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| if linked_memories: | ||
| memories.extend(linked_memories) | ||
| # 关联注入后截断到 top_k，避免超出调用方预期 | ||
| if top_k and len(memories) > top_k: | ||
| memories = memories[:top_k] |
There was a problem hiding this comment.
🟡 Related memories are found but then thrown away before reaching the assistant
The extra related memories are added to the result list and then cut back down to the requested size (memories[:top_k] at memory_manager.py:1209) after the list is already full, so the newly added related memories are always discarded whenever the normal search returns a full page of results.
Impact: The new "related memory injection" feature almost never actually adds anything, so linked context is silently missing from replies.
Truncation order removes exactly the appended items
In MemoryManager.recall_memories, the primary results are already truncated to top_k earlier (memory_manager.py:1170 for the all-users branch and memory_manager.py:1187 for the normal branch). Then at memory_manager.py:1201-1209 the linked memories are appended to the end of that list and the combined list is truncated back to top_k. Since len(memories) == top_k in the common case, slicing [:top_k] keeps only the original head and drops every appended linked memory. Only when the primary recall returns fewer than top_k hits does any linked memory survive.
A fix would be to reserve budget for linked memories (e.g. truncate primary results to top_k - len(linked) or interleave before truncation).
Prompt for agents
In MemoryManager.recall_memories (memory_manager.py, around lines 1200-1209), linked memories obtained from _inject_linked_memories are appended to `memories` and then the combined list is truncated back to top_k. Because `memories` was already truncated to top_k earlier in the function, the appended linked memories are always dropped in the common case, making the Phase-6 link injection feature a no-op. Rework the budgeting so linked memories actually get a slot: e.g. trim the primary results to `top_k - min(len(linked), reserve)` before extending, or apply a dedicated reserve (like the documented 'max 3 linked') on top of top_k, and make sure the primary results are not silently starved.
Was this helpful? React with 👍 or 👎 to provide feedback.
| for i, mem_a in enumerate(memories): | ||
| for mem_b in memories[i + 1 :]: | ||
| # 检查内容是否涉及同一主题（简化版：检查关键词重叠） | ||
| content_a = mem_a.get("content", "").lower() | ||
| content_b = mem_b.get("content", "").lower() | ||
|
|
||
| # 如果两条记忆内容高度相似，但创建时间差距大，可能是矛盾 | ||
| time_a = mem_a.get("metadata", {}).get("created_at", "") | ||
| time_b = mem_b.get("metadata", {}).get("created_at", "") | ||
|
|
There was a problem hiding this comment.
🔴 Background contradiction scan compares every memory with every other one, freezing the maintenance run on large memory pools
Every stored memory is compared against every other stored memory (nested loop at maintenance/agents/analyst.py:336-337) without any size limit, tenant boundary, or cap on the produced results, so a large memory pool makes each nightly maintenance run take a huge amount of CPU and produce an enormous list of alleged contradictions.
Impact: On installations with many memories the scheduled cleanup can hog the process for a long time and generate an unbounded pile of items, and it also pairs up memories that belong to different users or chats.
Unbounded O(n²) scan, no scope grouping, unbounded output
AnalystAgent.run fetches every active memory via _get_active_memories (maintenance/agents/analyst.py:111-136), which pages through the whole knowledge base with no upper bound. _screen_link_candidates at least samples down to maintenance_analyst_batch_size (default 200) before its O(n²) similarity loop, but _detect_contradictions is called with the full memories list (maintenance/agents/analyst.py:104-107) and performs a nested Python loop with per-pair set(content.split()) construction. With e.g. 5000 memories that is ~12.5M pair comparisons plus tokenization per iteration, in the same event loop.
Secondly, unlike _screen_link_candidates (which groups by memory_scope / owner_user_id / owner_session_id / umo at maintenance/agents/analyst.py:254-280), _detect_contradictions applies no scope grouping, so it pairs memories belonging to different users/groups.
Thirdly, the resulting list has no cap; every contradiction becomes an operation in the analyst manifest (maintenance/runner.py:386-395) and the whole manifest is JSON-dumped into a single reviewer prompt (maintenance/agents/reviewer.py:104-108), which can blow up the LLM request size.
Prompt for agents
AnalystAgent._detect_contradictions (maintenance/agents/analyst.py) runs a nested loop over the complete active-memory list returned by _get_active_memories, which is unbounded (it pages the entire KB). Three problems: (1) O(n²) with per-pair tokenization makes the nightly maintenance cycle extremely expensive on large memory pools; (2) unlike _screen_link_candidates it does not group by memory_scope / owner_user_id / owner_session_id / umo, so it pairs memories across users and chats; (3) the returned list is uncapped and every entry becomes a manifest operation that is JSON-dumped into the reviewer prompt (maintenance/runner.py and maintenance/agents/reviewer.py), potentially producing a huge LLM request. Consider reusing the same sampling/scope-grouping used for link candidates, capping the number of contradictions returned, and ideally reusing the precomputed cosine matrix instead of word-overlap over all pairs.
Was this helpful? React with 👍 or 👎 to provide feedback.
| manifest["merge"].append( | ||
| { | ||
| "uris": [mem_a["uri"], mem_b["uri"]], | ||
| "merged_content": verdict.fused_text or mem_a["content"], |
There was a problem hiding this comment.
🟡 Merging two memories can save a garbled, doubly-formatted copy of the memory text
When the model does not return a fused sentence, the merged memory falls back to the raw stored blob of one of the originals (mem_a["content"] at maintenance/agents/organizer.py:100), which already contains the storage formatting, so the newly created memory is saved with duplicated formatting lines instead of clean text.
Impact: The merged memory shown to the assistant and to users becomes a messy nested blob containing labels like "domain:"/"memory:" twice.
Stored text is the formatted blob, not the plain content
get_all_active_memories sets "content": doc.get("text", "") (memory_manager.py:2210-2216), i.e. the value produced by format_memory_content, which is a multi-line blob of the form:
domain: fact
memory: <real text>
recall_when: ...
topics: ...
When verdict.fused_text is empty the organizer proposes that whole blob as merged_content. MemoryManager.merge_memories then stores it as memory_content and runs format_memory_content(merged_content, new_metadata) again (memory_manager.py:2348-2364), yielding nested domain:/memory: headers. Display uses meta.memory_content (memory_protocol.py:248), so users/LLM see the nested blob.
A safer fallback is to skip the merge proposal when fused_text is empty, or to extract the plain content (e.g. metadata["memory_content"]) instead of the formatted document text.
| "merged_content": verdict.fused_text or mem_a["content"], | |
| "merged_content": verdict.fused_text | |
| or mem_a.get("metadata", {}).get("memory_content", ""), |
Was this helpful? React with 👍 or 👎 to provide feedback.
| marked = await mgr.mark_consolidated(source_uris) | ||
| if marked <= 0: | ||
| logger.debug("[简单长期记忆] 巩固：原文标记 0 条（摘要已写入）") | ||
| logger.info( | ||
| f"[简单长期记忆] 巩固 {len(candidates)} 条 → 1 条摘要，标记原文 {marked} 条" | ||
| ) |
There was a problem hiding this comment.
🟡 Memory consolidation can keep creating duplicate summaries of the same old memories
The consolidated summary is now written first and the originals are only marked afterwards (mark_consolidated at main.py:758), so if that marking step fails the originals stay eligible and the same batch is summarised again on the next run.
Impact: Repeated background runs can pile up near-identical summary memories for the same old conversations.
Loss of the previous guard
The previous implementation marked the source memories deprecated+compressed first and aborted when marked <= 0, so a failed mark never produced a summary. The new order writes the summary first and then calls mark_consolidated; when it returns 0 (SQL update failed, all rows already changed, etc.) the code only logs at debug level (main.py:759-763) and still reports success. Since fetch_consolidation_candidates selects on deprecated IS NOT 1 AND compressed IS NOT 1, the same candidates return every CONSOLIDATION_INTERVAL (2h) and another summary is stored each time.
A compromise is to keep the new ordering but delete/deprecate the just-written summary (or record a marker) when marked <= 0, and to log at warning level.
Prompt for agents
In MemoryPlugin._maybe_consolidate_memories (main.py, ~lines 744-763) the ordering was flipped so the summary is stored before the source memories are marked consolidated. If mark_consolidated returns 0 the code only logs at debug level and reports success, but the sources remain non-deprecated/non-compressed, so fetch_consolidation_candidates will select the very same batch on the next scheduled run and store another duplicate summary indefinitely. Add a guard: on marked <= 0, either roll back / delete the just-created summary, or persist a marker so the same candidate set is not re-summarised, and raise the log level to warning.
Was this helpful? React with 👍 or 👎 to provide feedback.
| old_doc_id = doc.get("doc_id", "") | ||
| try: | ||
| await self._register_kb_document(new_doc_id, uri, len(text)) | ||
| except Exception: | ||
| pass | ||
| if old_doc_id: | ||
| try: | ||
| await self._unregister_kb_documents([old_doc_id]) | ||
| except Exception: | ||
| pass | ||
| try: | ||
| await self.vec_db.delete_documents( | ||
| metadata_filters={"kb_doc_id": old_doc_id} | ||
| ) | ||
| except Exception as del_err: | ||
| # 删除失败 → 原记录仍活跃，返回失败让调用方处理 | ||
| logger.warning( | ||
| f"[简单长期记忆] deprecate 删除旧记录失败: {del_err}" | ||
| ) | ||
| return False |
There was a problem hiding this comment.
🔍 deprecate_memory can leave an active original plus a deprecated duplicate
deprecate_memory inserts a new document carrying the same uri with deprecated=True and only afterwards deletes the original by kb_doc_id. Two follow-ups:
- If
doc.get("doc_id", "")is empty (the rest of the codebase consistently reads the id frommetadata["kb_doc_id"], e.g.memory_manager.py:626,memory_manager.py:1817, andget_all_active_memoriesusesdoc.get("id")atmemory_manager.py:2204), the delete branch is skipped entirely and the function returnsTruewhile the original record is still active — the archive silently does nothing and a duplicate deprecated copy remains. - If the delete throws, the function returns
Falsebut the freshly inserted deprecated duplicate (and its registered KB document) is not rolled back, leaving two rows for the same URI.
Worth confirming which key document_storage.get_documents() actually returns for the document id.
Was this helpful? React with 👍 or 👎 to provide feedback.
| max_rounds = self._config.get("context_max_rounds", 50) | ||
| max_chars = self._config.get("context_max_chars", 30000) | ||
|
|
||
| conv_mgr = self._context.conversation_manager | ||
| # 按 UMO 限定范围（owner_filter 中可能有 umo） | ||
| umo = (owner_filter or {}).get("umo") | ||
| conversations = await conv_mgr.get_conversations(unified_msg_origin=umo) | ||
|
|
||
| lines: list[str] = [] | ||
| total_chars = 0 | ||
| total_rounds = 0 | ||
| for conv in conversations: | ||
| history_raw = getattr(conv, "history", None) | ||
| if not history_raw: | ||
| continue | ||
| try: | ||
| history = ( | ||
| _json.loads(history_raw) | ||
| if isinstance(history_raw, str) | ||
| else history_raw | ||
| ) | ||
| except Exception: | ||
| continue | ||
| if not isinstance(history, list): | ||
| continue | ||
| for entry in reversed(history): # 最新的在前 | ||
| if total_rounds >= max_rounds or total_chars >= max_chars: | ||
| break | ||
| role = entry.get("role", "") | ||
| text = entry.get("content", "") | ||
| if not text or role not in ("user", "assistant"): | ||
| continue | ||
| line = f"[{role}]: {text}" | ||
| if total_chars + len(line) > max_chars: | ||
| break | ||
| lines.append(line) | ||
| total_chars += len(line) | ||
| total_rounds += 1 | ||
| if total_rounds >= max_rounds or total_chars >= max_chars: | ||
| break | ||
|
|
There was a problem hiding this comment.
🔍 Analyst pulls conversation history for all sessions when no UMO filter is supplied
_get_conversation_history reads umo from owner_filter, but the runner always calls AnalystAgent.run() with no owner_filter (maintenance/runner.py:372), so conv_mgr.get_conversations(unified_msg_origin=None) is invoked. Depending on the AstrBot API this either errors (caught and returns "") or returns conversations across all sessions/users, which contradicts the spec requirement "对话历史拉取按 UMO 限定，不跨会话读取" (docs/maintenance-agent-spec.md). Also note context_max_age_days is declared in the schema but never applied here — only rounds and chars are enforced.
Was this helpful? React with 👍 or 👎 to provide feedback.
| for i, op in enumerate(all_operations): | ||
| # 查找对应的审核结果 | ||
| verdict = None | ||
| for v in report.reviewer_verdicts: | ||
| if v.get("index") == i: | ||
| verdict = v | ||
| break | ||
|
|
||
| # 缺失裁决 → fail closed，拒绝执行（避免 reviewer 故障时放行破坏性操作） | ||
| if verdict is None: | ||
| # additive 操作（new_link）无需审核可直接执行 | ||
| if op.get("type") == "new_link": | ||
| try: | ||
| success = await self._execute_operation(op) | ||
| if success: | ||
| executed += 1 | ||
| else: | ||
| failed += 1 | ||
| report.errors.append(f"op[{i}] new_link 执行返回失败") | ||
| except Exception as e: | ||
| failed += 1 | ||
| report.errors.append(f"op[{i}] new_link: {e}") | ||
| continue | ||
| # destructive 操作缺少裁决 → 转待审队列（reviewer 禁用时不静默丢弃） | ||
| await self._enqueue_pending_review( | ||
| op, | ||
| {"verdict": "pending", "reason": "reviewer 未启用或缺失裁决"}, | ||
| report.session_id, | ||
| ) | ||
| skipped += 1 | ||
| logger.info( | ||
| f"[简单长期记忆] 操作 {i} 缺少审核裁决，转待审: {op.get('type')}" | ||
| ) | ||
| continue | ||
|
|
There was a problem hiding this comment.
🔍 Reviewer verdicts are matched to operations purely by LLM-supplied index
Operations are collected in the same order for both the reviewer prompt (_run_reviewer) and the execution loop, and matching relies on the model echoing back a correct index. Any hallucinated/shifted index silently maps a verdict onto a different operation; an approve intended for a link could authorise a merge. The current fail-closed default (missing verdict → pending review) helps, but a mismatched-but-present index is not detected. Consider embedding a stable operation id in the prompt and validating it on return.
Was this helpful? React with 👍 or 👎 to provide feedback.
| async def _disclosure_retrieve( | ||
| self, query: str, top_k: int, filters: dict[str, Any] | ||
| ) -> list[dict[str, Any]]: | ||
| """Disclosure 精确匹配检索通道。 | ||
|
|
||
| 查找 disclosure 字段与查询关键词有交集的记忆， | ||
| 确保“当用户问 X 时”类型的触发条件不会被向量相似度漏掉。 | ||
| """ | ||
| if not self._kb_helper: | ||
| return [] | ||
|
|
||
| # 分词：优先用 AstrBot 分词器，不可用时 fallback 到空格分词 | ||
| query_tokens = self._tokenize_query(query) | ||
| if not query_tokens: | ||
| query_tokens = [t for t in query.lower().split() if len(t) >= 2] | ||
| if not query_tokens: | ||
| return [] | ||
|
|
||
| # 构建 SQL：在现有 filters 基础上追加 disclosure 非空条件 | ||
| where_clause, params = self._filters_to_sql(filters) | ||
| where_clause += ( | ||
| " AND json_extract(metadata,'$.disclosure') IS NOT NULL" | ||
| " AND json_extract(metadata,'$.disclosure') != ''" | ||
| ) | ||
| # 分页扫描直到找到足够匹配或遍历完所有候选 | ||
| try: | ||
| scan_limit = int(self.config.get("disclosure_scan_limit", 200)) | ||
| except (TypeError, ValueError): | ||
| scan_limit = 200 | ||
| scan_limit = max(10, min(scan_limit, 5000)) | ||
|
|
||
| query_token_set = set(query_tokens) | ||
| matched: list[dict[str, Any]] = [] | ||
| target_matches = min(top_k * 2, 10) | ||
| page_size = min(scan_limit, 200) | ||
| offset = 0 | ||
|
|
||
| try: |
There was a problem hiding this comment.
🔍 Disclosure channel adds a full table scan per recall filter
_disclosure_retrieve runs on every recall (there is no feature flag) and issues paged SELECT text, metadata FROM documents ... ORDER BY json_extract(metadata,'$.created_at') DESC queries — up to disclosure_scan_limit (default 200, max 5000) rows — once per filter in filters_list (which is 2–5 filters per recall). Since json_extract on an unindexed JSON column cannot use an index, each recall now performs several sorted scans over the documents table in the request path of every LLM call. Worth measuring on a large KB, and possibly gating it behind a config toggle like recall_sparse_fusion.
Was this helpful? React with 👍 or 👎 to provide feedback.
| # 从 FAISS 索引加载向量（document_storage 不返回 vector 字段） | ||
| faiss_index = None | ||
| embedding_storage = getattr(self.vec_db, "embedding_storage", None) | ||
| if embedding_storage is not None: | ||
| faiss_index = getattr(embedding_storage, "index", None) | ||
|
|
||
| memories = [] | ||
| for doc in docs: | ||
| metadata = _safe_parse_metadata(doc.get("metadata", {})) | ||
| vector = None | ||
| doc_int_id = doc.get("id") | ||
| if faiss_index is not None and doc_int_id is not None: | ||
| try: | ||
| vector = faiss_index.reconstruct(int(doc_int_id)).tolist() | ||
| except Exception: | ||
| pass # 向量不存在时跳过，organizer/analyst 会过滤 | ||
| memories.append( | ||
| { | ||
| "uri": metadata.get("uri", ""), | ||
| "content": doc.get("text", ""), | ||
| "metadata": metadata, | ||
| "vector": vector, | ||
| } | ||
| ) |
There was a problem hiding this comment.
🔍 Vector loading for candidate screening depends on FAISS reconstruct and a possibly wrong id field
get_all_active_memories reconstructs vectors via faiss_index.reconstruct(int(doc.get("id"))). This assumes (a) get_documents returns an id key holding the FAISS internal id and (b) the index type supports reconstruct (IVF/HNSW variants often require make_direct_map() first, and IDMap wrappers reconstruct by internal offset, not external id). Any mismatch is silently swallowed by the bare except: pass, in which case vector is None for every memory and both organizer and analyst screening return zero candidates — the whole maintenance pipeline becomes a silent no-op with no diagnostic. At minimum a debug log on repeated reconstruct failures would make this observable.
Was this helpful? React with 👍 or 👎 to provide feedback.
新增 MaintenanceRunner._op_lock 互斥锁，run_cycle 阶段 4 与 /memory review approve 共用同一锁执行操作；Scheduler 暴露 runner 属性，替换对私有成员的访问。
1 participant
概述
基于后台维护 agent 对长期记忆进行离线整理、分析和审核的完整实现。
实施内容
Phase 1: 基础设施收尾
Phase 2: LLM 唯一入口 + 执行管线
Phase 3: 整理师 Agent
Phase 4: 分析师 Agent
Phase 5: 审核员 Agent + 人工升级
Phase 6: 关联召回注入
Phase 7: 配置与文档
验证
变更统计
Summary by Sourcery
引入针对长期记忆的后台维护代理系统，增加基于 LLM 的定期整合、关联与回顾工作流，同时提供更安全的废弃、合并和召回行为。
Enhancements:
deprecated_at时间戳、为遗留记录提供迁移补丁，以及尊重宽限期的清理（purge）流程，改进废弃语义，并保持知识库（KB）与向量存储的一致性。string.Template的模板上，并扩展配置和文档，以涵盖与维护相关的选项和行为。Documentation:
maintenance-agent-specmarkdown 文件中，记录后台维护代理的设计、工作流、安全保证和配置。Original summary in English
Summary by Sourcery
Introduce a background maintenance agent system for long-term memory, adding scheduled LLM-based consolidation, linking, and review workflows, along with safer deprecation, merging, and recall behaviour.
Enhancements:
Documentation: