diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..1ce3dda --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,95 @@ +# AGENTS.md + +给在本仓库工作的 AI 代理（以及新加入的人）看的约定。只写与改动直接相关的硬性要求，风格问题按现有代码对齐。 + +## 项目概况 + +- AmritaBot 长期记忆插件：L1 备忘录、L2 向量记忆、L3 离线巩固循环（DMN）三层。 +- 表层工具（对话 LLM 可用）注册在全局 ToolsManager；`subconscious_*` 通过 `bound_to=_SUBCONSCIOUS_TOOLS` 注册到隔离管理器，不污染全局。 +- 插件入口 `amrita_plugin_memory/__init__.py`，命令入口 `cli.py`（`ambot memory`），对话命令在 `matchers.py`（`/memory`）。 + +## 模块边界 + +| 位置 | 职责 | +| --- | --- | +| `config.py` | EnvConfig / ConfigFile（pydantic + uniconf），DATA_PATH 等路径常量 | +| `keys.py` | uni_id 生成与 scope 解析 | +| `vector.py` | ChromaDB 封装、AsyncUserMemory 记忆读写 | +| `embedding.py` / `embed.py` | 嵌入模型与指纹校验、重映射 | +| `memo.py` | L1 备忘录读写 | +| `decay.py` | 时间记忆退化与分层清理 | +| `permissions.py` | 会话侧读取权限校验 | +| `profile_store.py` | 画像路径与 Markdown 解析（纯函数，不依赖 nonebot 运行时） | +| `rethinking/` | DMN 循环：runner、tools、schemas、knowledge、schedule | +| `rethinking/schedule.py` | 空闲窗口规划（纯数据 + 纯函数，`resolve_timezone` 是全局唯一的时区解析入口） | + +新增纯逻辑优先放进不依赖 nonebot 的模块（如 `profile_store.py`、`rethinking/schedule.py`），运行器只做编排。 + +## 环境与依赖 + +- Python 3.10–3.13，包管理只用 uv。改依赖后跑 `uv sync`，不要用 `pip install` 补装单包。 +- 类型检查用 `pyright`，配置在 `pyproject.toml` 的 `[tool.pyright]`（standard 模式）。 +- 本仓库 `.venv` 里通常没有 pyright / ruff，用 `--pythonpath .venv/bin/python` 指向仓库环境，否则会因解析不到 amrita_core / chromadb 报一片 Import 错误。 + +## 常用命令 + +```bash +uv run ruff check amrita_plugin_memory +uv run ruff format --check amrita_plugin_memory +uv run pyright --pythonpath .venv/bin/python amrita_plugin_memory +python -m py_compile amrita_plugin_memory/**/*.py +``` + +## 测试要求 + +本仓库不补单元测试，验证以静态检查加手动冒烟为准： + +1. `ruff check` 与 `ruff format --check` 通过 +2. `pyright` 零 error（用上面的 pythonpath 跑） +3. 改到的模块 `py_compile` 通过 +4. 手动跑一遍改动的命令（`/memory`、`ambot memory`）或工具调用，并在 PR 里说明验证方式与结果 + +纯逻辑尽量写成可独立执行的形式，方便用 `python -c` 或临时脚本验证；不要把临时脚本提交进仓库。 +本仓库无 `tests/` 目录，也不要新建。纯逻辑写成可独立执行的形式，用临时脚本验证后不提交。 + +## 工具注册的三条路径 + +- 表层可读、需权限：实现放 `rethinking/tools.py`，`@on_tools(SCHEMA, strict=True, custom_run=True)` 注册到全局，进函数先校验 `permissions.py` 里的判定。 +- 仅循环可用：同上但带 `bound_to=_SUBCONSCIOUS_TOOLS`。 +- 新工具要同时加入 `rethinking/schemas.py` 的 `_SUBCONSCIOUS_TOOL_FUNCTIONS`（若循环也要用），并在 README 的工具表里登记。 + +新增工具时注意：schema 的 `name` 是工具名，函数名可另取；`custom_run=True` 时函数签名收 `ToolContext`，返回值必须是字符串。 + +## 权限口径 + +- 知识库只读：`permissions.can_read_knowledge`，超用户 / `lp.admin` / `memory.knowledge` 节点三者之一。 +- 用户画像只读：`permissions.can_read_profile`，仅 `subconscious.target_user_id` 本人，不扩散权限节点。 +- 取值优先用 `_state.get_target_user_id()`（运行器实际生效值），回退 `DataManager` 配置。空值一律拒。 + +## 数据与持久化 + +- 画像：`DATA_PATH/user_profile.md`，格式 `summary` + `---` + 正文；行级增量更新，写入只发生在循环内部。 +- 知识库：`DATA_PATH/knowledge/` + `knowledge_index.json` + ChromaDB，文件/索引/向量三方同步。 +- DMN 元状态：`SubconsciousState` 表；向量记忆在 ChromaDB collection。 +- 写入路径优先走读-改-写整体替换，避免嵌入失败时丢数据（参考 `25f1078`）。 + +## 代码注释风格 + +- 注释与 docstring 一律中文，单行短句，讲“为什么”和约束，不复述代码本身在做什么。 +- 模块顶部 docstring 写职责、关键约束、与其他模块的边界；函数 docstring 写清参数语义与副作用。 +- 不写分节注释（`# ---- 小节名`、`# ====`），用空行和函数拆分表达结构。 +- 不写连续多行 `#` 注释；需要成段解释时写进模块或函数 docstring。 +- 待办写成 `# TODO(review #)：具体动作`。 +- 类型注解必须齐全；pydantic 字段用 `Field(description=...)`，描述用中文。 +- 不写注释掉的旧代码，不写“新增/修改”这类只在 diff 里有意义的注释。 + +## 配置改动 + +- 配置型在 `config.py`（pydantic + `nonebot_plugin_uniconf`）。新字段必须有中文 `description` 和保守默认值，默认值要保证旧配置文件行为不变。 +- 新增开关涉及 DMN 行为时，同步检查 `matchers.py` 的 `/memory dmn` 前置校验与 README 的配置表。 + +## Git 与 PR + +- 分支名用 `feat/` `fix/` `refactor/` 前缀；提交信息中文，首行说明改了什么，正文写动机与影响面。 +- PR 拿到 approval 不等于授权合并；合并默认 squash，等明确指令再动。 +- GitHub raw 域名可能不可达，取仓库内容走 `api.github.com`。 diff --git a/README.md b/README.md index cf26699..9bbc054 100644 --- a/README.md +++ b/README.md @@ -410,11 +410,17 @@ flowchart TD | `knowledge_read` | kid, start_line?, end_line? | | `knowledge_search` | query, top_k? | | `knowledge_suggest` | action, title, summary, body, reason | +| `read_profile` | start_line?, end_line? | 知识库的三个只读工具 `knowledge_list` / `knowledge_read` / `knowledge_search` 需要 LP 权限节点 `memory.knowledge`；`lp.admin` 与 `.env` 里的超级用户始终持有。 未授权时工具返回错误 JSON，不返回任何知识内容。 +用户画像的只读工具 `read_profile` 走另一条口径：只对离线巩固循环的目标用户 +（`subconscious.target_user_id`）开放，且只能读自己那份画像，不占用权限节点。 +返回值与循环侧一致：`summary` + `content` 行窗口 + `total_lines` / `range_start` / `range_end`。 +画像写入仍只发生在循环内部，表层不开放。 + ### 巩固循环工具（`rethinking/tools.py`，共 20 个） 记忆与 session/画像工具注册在隔离的 `_SUBCONSCIOUS_TOOLS` 上，不污染全局工具管理器。 diff --git a/amrita_plugin_memory/config.py b/amrita_plugin_memory/config.py index ff5f8cb..0a93e30 100644 --- a/amrita_plugin_memory/config.py +++ b/amrita_plugin_memory/config.py @@ -42,6 +42,59 @@ class SubconsciousConfig(BaseModel): rethink_max_delay_minutes: int = Field( default=1440, ge=60, le=10080, description="惩罚延迟上限（分钟），默认1天" ) + use_idle_window_planner: bool = Field( + default=True, + description="是否启用空闲窗口调度（热度图 + 冷段二分）；关闭后回退到旧的固定延迟方案", + ) + idle_timezone: str = Field( + default="Asia/Shanghai", description="热度图与窗口计算使用的时区" + ) + idle_fine_tau_days: float = Field( + default=28.0, ge=1.0, description="单片星期几层的衰减常数（天）" + ) + idle_coarse_tau_days: float = Field( + default=14.0, ge=1.0, description="全局层的衰减常数（天）" + ) + idle_shrink_k: float = Field( + default=4.0, ge=0.0, description="细层向全局收缩的强度，越大越保守" + ) + idle_min_sample_days_fine: int = Field( + default=6, ge=1, description="启用单片星期几层所需的最少活跃天数" + ) + idle_min_sample_days_mid: int = Field( + default=3, ge=1, description="启用工作日/周末层所需的最少活跃天数" + ) + idle_cold_pct_max: float = Field( + default=0.05, ge=0.0, le=1.0, description="冷槽的占比上限" + ) + idle_cold_abs_max: float = Field( + default=1.0, ge=0.0, description="冷槽的原始计数上限，避免稀疏样本全部判冷" + ) + idle_target_hour: int = Field( + default=3, ge=0, le=23, description="偏好触发时刻（本地时区小时）" + ) + idle_base_slots: int = Field( + default=3, ge=1, description="首次计划要求的最短冷段（槽，1 槽=2h）" + ) + idle_max_slots: int = Field(default=6, ge=1, description="冷段要求的封顶槽数") + idle_penalty_multiplier: float = Field( + default=1.5, ge=1.0, le=10.0, description="惩罚对冷段要求的放大倍率" + ) + idle_max_relax_steps: int = Field( + default=3, ge=0, le=6, description="冷段不足时允许的减半放宽次数" + ) + idle_buffer_minutes: int = Field( + default=15, ge=0, description="窗口起止的安全缓冲（分钟）" + ) + idle_run_estimate_minutes: int = Field( + default=10, ge=0, description="单轮 DMN 预计耗时（分钟），用于扣减可用窗口" + ) + idle_fallback_minutes: int = Field( + default=120, ge=10, description="冷启动或规划失败时的固定窗口长度（分钟）" + ) + idle_noise_gap_minutes: int = Field( + default=15, ge=0, description="连发消息合并为单点的最小间隔（分钟）" + ) prompt_file: str = Field( default="prompt/subconscious_main.md.jinja2", description="主推理提示词文件名，相对于 config/amrita_plugin_memory/", diff --git a/amrita_plugin_memory/permissions.py b/amrita_plugin_memory/permissions.py index 4970f83..be91bc3 100644 --- a/amrita_plugin_memory/permissions.py +++ b/amrita_plugin_memory/permissions.py @@ -1,8 +1,12 @@ -"""知识库读取的权限校验。 +"""会话侧读取权限校验（知识库 / 用户画像）。 -会话侧（对话 LLM）可以读全局知识库，但需要 LP 节点 ``memory.knowledge``； +知识库：会话侧（对话 LLM）可以读全局知识库，但需要 LP 节点 ``memory.knowledge``； ``lp.admin`` 与 ``.env`` 里的超级用户始终持有该权限。 -潜意识循环内部调用不受此限制。 + +用户画像：只对离线巩固循环（DMN）的目标用户开放只读， +比对 ``subconscious.target_user_id``，不做权限节点扩散。 + +潜意识循环内部调用均不受此限制。 """ from __future__ import annotations @@ -54,6 +58,36 @@ async def can_read_knowledge(event: Event | None) -> bool: return await _has_node(event, KNOWLEDGE_NODE) +async def get_dmn_target_user_id() -> str: + """当前 DMN 目标用户 ID：优先取运行器中实际生效的值，回退到配置。""" + try: + from .rethinking import _state + + uid = _state.get_target_user_id() + if uid: + return str(uid) + except Exception as e: + logger.debug(f"[Memory] 取 DMN 运行器失败: {e}") + try: + from .config import DataManager + + cfg = (await DataManager().safe_get_config()).subconscious + return str(cfg.target_user_id or "") + except Exception as e: + logger.debug(f"[Memory] 取 DMN 配置失败: {e}") + return "" + + +async def can_read_profile(event: Event | None) -> bool: + """是否允许读取用户画像：仅 DMN 目标用户本人。""" + if event is None: + return False + target = await get_dmn_target_user_id() + if not target: + return False + return str(event.get_user_id()) == target + + def get_event_from_tool_ctx(ctx: Any) -> Event | None: """从 ToolContext 里取出触发本次调用的原始事件。""" try: diff --git a/amrita_plugin_memory/profile_store.py b/amrita_plugin_memory/profile_store.py new file mode 100644 index 0000000..6f96f4b --- /dev/null +++ b/amrita_plugin_memory/profile_store.py @@ -0,0 +1,67 @@ +"""用户画像文件（Markdown）的路径与解析。 + +会话侧只读入口与潜意识循环共用同一份解析逻辑， +避免为读取画像而依赖运行器实例。 +""" + +from __future__ import annotations + +from pathlib import Path +from typing import TypedDict + +from nonebot import logger + +from .config import DATA_PATH + + +class ProfileResult(TypedDict): + """用户画像读取返回值。""" + + summary: str + content: str + total_lines: int + range_start: int + range_end: int + + +#: 画像文件路径 +PROFILE_PATH: Path = DATA_PATH / "user_profile.md" + + +def read_profile( + path: Path, + start_line: int | None = None, + end_line: int | None = None, +) -> ProfileResult: + """解析画像 Markdown（summary + "---" + body），按行返回滑动窗口。 + + 与运行器解耦：潜意识循环未启动时，会话侧也能读到既有画像文件。 + """ + empty: ProfileResult = { + "summary": "", + "content": "", + "total_lines": 0, + "range_start": 0, + "range_end": 0, + } + if not path.exists(): + return empty + try: + text = path.read_text(encoding="utf-8") + parts = text.split("---", 1) + summary = parts[0].strip() if parts else "" + body = parts[1].strip() if len(parts) > 1 else "" + body_lines = body.split("\n") if body else [] + total = len(body_lines) + s = 0 if start_line is None else max(0, start_line) + e = total if end_line is None else max(s, min(total, end_line)) + return { + "summary": summary, + "content": "\n".join(body_lines[s:e]), + "total_lines": total, + "range_start": s, + "range_end": e, + } + except Exception as e: + logger.warning(f"[Subconscious] Read profile failed: {e}") + return empty diff --git a/amrita_plugin_memory/rethinking/_state.py b/amrita_plugin_memory/rethinking/_state.py index 75917b1..1c6b36e 100644 --- a/amrita_plugin_memory/rethinking/_state.py +++ b/amrita_plugin_memory/rethinking/_state.py @@ -18,6 +18,13 @@ _penalty_count: int = 0 # 知识建议队列 — 对话 LLM 提议创建/更新知识，潜意识 Agent 下一轮审查 _knowledge_suggestions: list[dict[str, str]] = [] +# 上一轮 iter_stop 的摘要与结束时间 — 供 subconscious_get_context 读取 +_last_iter_summary: str = "" +_last_run_at: float = 0.0 +# 本轮是否已调过 subconscious_get_context — iter_stop 据此判定周期是否有效 +_context_fetched: bool = False +# 本轮是否算一次有效周期 — 空转的周期不重置惩罚 +_cycle_valid: bool = True # 隔离的工具管理器 — 所有 @on_tools 装饰器通过 bound_to 注册到这里，不污染全局 ToolsManager _SUBCONSCIOUS_TOOLS = MultiToolsManager() @@ -89,3 +96,49 @@ def set_knowledge_suggestions(suggestions: list[dict[str, str]]) -> None: def add_knowledge_suggestion(suggestion: dict[str, str]) -> None: global _knowledge_suggestions _knowledge_suggestions.append(suggestion) + + +def get_last_iter_summary() -> str: + return _last_iter_summary + + +def set_last_iter_summary(text: str) -> None: + global _last_iter_summary + _last_iter_summary = text + + +def mark_context_fetched() -> None: + global _context_fetched + _context_fetched = True + + +def is_context_fetched() -> bool: + return _context_fetched + + +def reset_context_fetched() -> None: + global _context_fetched + _context_fetched = False + + +def mark_cycle_invalid() -> None: + global _cycle_valid + _cycle_valid = False + + +def is_cycle_valid() -> bool: + return _cycle_valid + + +def reset_cycle_valid() -> None: + global _cycle_valid + _cycle_valid = True + + +def get_last_run_at() -> float: + return _last_run_at + + +def set_last_run_at(ts: float) -> None: + global _last_run_at + _last_run_at = ts diff --git a/amrita_plugin_memory/rethinking/consts.py b/amrita_plugin_memory/rethinking/consts.py index 8458bb8..c77c62e 100644 --- a/amrita_plugin_memory/rethinking/consts.py +++ b/amrita_plugin_memory/rethinking/consts.py @@ -6,8 +6,9 @@ DEFAULT_SUBCONSCIOUS_PROMPT = """\ {# 潜意识推理循环 - 系统提示词 #} -{# 变量: last_run, current_time, target_user_id, total_runs, character_prompt #} -{# 编辑此文件即可自定义提示词，无需改动代码。 #} +{# 变量: character_prompt, target_user_id, current_date #} +{# 本模板只放不变量：所有会随循环变化的信息（时间、待办计数、热度统计） #} +{# 一律通过 subconscious_get_context 按需拉取，不要写回这里。 #} 你是用户的记忆管家（Memory Curator），在后台静默运行。你的职责是整理、优化和组织用户的长期记忆库。 @@ -16,9 +17,8 @@ {{ character_prompt }} {% endif %} -当前时间: {{ current_time }} +日期: {{ current_date }} 用户ID: {{ target_user_id }} -这是第 {{ total_runs }} 次推理循环。 @@ -27,15 +27,20 @@ 1. 你**必须**使用提供的工具来检查和操作用户的记忆库。不要凭空想象或推测记忆内容。 -2. 每次推理循环中，你应该: +2. 每轮**第一件事**必须调用 subconscious_get_context 获取当前环境 + （精确时间、距上轮间隔、上次摘要、待审查建议、记忆库规模）。 + 不调用则视为本轮无效，不要凭上一轮的印象行动。 +3. 之后你应该: a. 先调用 subconscious_list_memory 或 subconscious_read_memory 了解记忆库当前状态 b. 识别需要处理的问题（重复记忆、矛盾记忆、过期信息、缺失标签等） c. 使用 subconscious_write_memory / subconscious_update_memory / subconscious_delete_memory 执行整理操作 -3. 当本轮整理工作完成（或没有新发现需要处理），你**必须**调用 subconscious_iter_stop 来结束本次推理循环。 -4. 利用 subconscious_read_sessions 和 subconscious_read_chat_context 了解用户近况。 -5. 利用 subconscious_get_profile / subconscious_update_profile 渐进式构建用户画像。 -6. 不要在单次循环中做太多事情, 保持每次循环简洁高效。 -7. 所有记忆操作都作用于目标用户（user_id={{ target_user_id }}）, scope 固定为 user。 +4. 当本轮整理工作完成（或没有新发现需要处理），你**必须**调用 subconscious_iter_stop 来结束本次推理循环。 +5. 利用 subconscious_read_sessions 和 subconscious_read_chat_context 了解用户近况。 +6. 利用 subconscious_get_profile / subconscious_update_profile 渐进式构建用户画像。 +7. 不要在单次循环中做太多事情, 保持每次循环简洁高效。 +8. 所有记忆操作都作用于目标用户（user_id={{ target_user_id }}）, scope 固定为 user。 +9. 记忆库规模、待审查建议数量这类数字只以 subconscious_get_context 的返回为准， + 不要假设它们在本轮内保持不变，也不要自己推算轮数或进度。 @@ -46,13 +51,6 @@ - 过时信息: 识别明显过时的信息, 考虑更新或降低重要性 -{% if last_run %} - -上一次推理循环的摘要: -{{ last_run }} - -{% endif %} - 你不是在和用户对话。你只通过工具调用（tool calls）和最终调用 subconscious_iter_stop 来表达你的工作。 在 iter_stop 的 summary 字段中, 请简要描述你本轮做了什么（例如: "检查了5条记忆, 合并了2条重复项, 为3条记忆补充了标签"）。 @@ -214,7 +212,7 @@ async def load_character_prompt() -> str: | 文件 | 用途 | 可用变量 | | ----------------------------- | ---------------------------------------------- | ------------------------------------------------------------------------------ | -| `subconscious_main.md.jinja2` | 主推理循环提示词 | `character_prompt`, `last_run`, `current_time`, `target_user_id`, `total_runs` | +| `subconscious_main.md.jinja2` | 主推理循环提示词 | `character_prompt`, `current_date`, `target_user_id` | | `subconscious_send.md.jinja2` | 向用户主动发消息的生成模板 | `character_prompt`, `intent`, `memory_context`, `current_time` | | `knowledge_guide.md.jinja2` | 全局知识库使用指南（注入到主 prompt 末尾） | 无（纯静态文本） | | `profile_guide.md.jinja2` | 用户画像渐进式构建指南（注入到主 prompt 末尾） | 无（纯静态文本） | @@ -233,8 +231,9 @@ async def load_character_prompt() -> str: ## 注意事项 - 不要删除 `{% if character_prompt %}` 块——它用于注入 Bot 的角色设定 -- `last_run` 变量只在非首次循环时有值 -- `current_time` 格式为 `YYYY-MM-DD HH:MM:SS UTC` +- `current_date` 只到日粒度，用来保持 system 前缀跨轮稳定（可命中缓存） +- 精确时间、上轮摘要、待办计数、记忆库规模一律不进模板， + 由工具 `subconscious_get_context` 按需拉取，避免把过期快照写进 system - `target_user_id` 是配置中指定的数字 ID 字符串 ## 默认行为 diff --git a/amrita_plugin_memory/rethinking/hooks.py b/amrita_plugin_memory/rethinking/hooks.py index 4986817..265dbd2 100644 --- a/amrita_plugin_memory/rethinking/hooks.py +++ b/amrita_plugin_memory/rethinking/hooks.py @@ -1,5 +1,7 @@ """Hook 注册与生命周期管理。""" +import time + from amrita.plugins.chat.runtime import try_get_amrita_ctx from amrita_core.hook.event import PreCompletionEvent from amrita_core.hook.on import on_precompletion @@ -25,7 +27,7 @@ async def _subconscious_pre_hook(event: PreCompletionEvent) -> None: return try: - await runner.cancel_and_reschedule() + await runner.cancel_and_reschedule(chat_ts=time.time()) except Exception as e: # 调度失败不能影响用户这条消息的处理 logger.opt(exception=e, colors=True, raw=True).error( diff --git a/amrita_plugin_memory/rethinking/runner.py b/amrita_plugin_memory/rethinking/runner.py index b9a2af2..818469a 100644 --- a/amrita_plugin_memory/rethinking/runner.py +++ b/amrita_plugin_memory/rethinking/runner.py @@ -5,6 +5,7 @@ import asyncio import contextlib import json +import time from datetime import datetime, timedelta, timezone from typing import Any @@ -31,7 +32,7 @@ from ..config import DATA_PATH, SubconsciousConfig from ..keys import make_scope_id from ..models import SubconsciousState -from ..vector import AsyncUserMemory, get_db_conn +from ..profile_store import PROFILE_PATH, ProfileResult, read_profile from . import _state from .backend import SubconsciousBackend from .consts import ( @@ -43,9 +44,12 @@ load_character_prompt, ) from .knowledge import KnowledgeBaseManager -from .types import ProfileResult, SessionSummary +from .schedule import IdlePlanner, describe +from .types import SessionSummary _STATE_UID = "amrita_memory" +#: 重排节流窗口（秒） +_REPLAN_THROTTLE_SECONDS = 60.0 _LEGACY_REPO_UID = ( "amrita_memory" # 旧版本元状态占用的 Bot Memory 表 uid（仅用于升级迁移） ) @@ -78,7 +82,19 @@ def __init__(self, config: SubconsciousConfig) -> None: self._kb_manager: KnowledgeBaseManager | None = None # session 摘要缓存（LRU）：session DB id -> 摘要文本，最多 128 条 self._session_cache: LRUCache[int, str] = LRUCache(128) - self._profile_path = DATA_PATH / "user_profile.md" + self._profile_path = PROFILE_PATH + # 空闲窗口规划器（热度图 + 冷段二分） + self._planner = IdlePlanner(config) + # 重排节流：一分钟内重复触发只更新惩罚与热度，不重复规划 + self._last_replan_at: float = 0.0 + self._replan_throttle: float = _REPLAN_THROTTLE_SECONDS + # 状态写入串行化：并发 save 先后到达时，旧快照不能盖掉新快照 + self._state_lock = asyncio.Lock() + # 正在跑的周期任务，用户新消息到来时可以掐断 + self._cycle_task: asyncio.Task | None = None + # 本轮是否已取过上下文，供 iter_stop 校验 + _state.reset_context_fetched() + _state.reset_cycle_valid() @property def is_running(self) -> bool: @@ -123,28 +139,99 @@ def stop(self) -> None: if self._manual_task is not None and not self._manual_task.done(): self._manual_task.cancel() - async def cancel_and_reschedule(self) -> None: - """取消当前计划，递增惩罚计数器，重新计算延迟并调度一次运行。 + async def cancel_and_reschedule(self, chat_ts: float | None = None) -> None: + """取消当前计划并重新规划下一个空闲窗口。 - 用户每次发消息时调用此方法。不 dump/恢复上下文—— + 用户每次发消息时调用：先给热度图记一次观测，再递增惩罚， + 然后按新的冷段门槛重新选窗口。不 dump/恢复上下文—— 用户聊完天后有新内容，旧推理上下文应直接丢弃。 """ + # 先记下历史是否存在：首条消息会把热度图从空变成非空， + # 若观测后再判断，冷启动就会被自己这条消息绕过 + had_history = True with contextlib.suppress(Exception): - scheduler.remove_job(self._job_id) + had_history = self._planner.has_history() + self._planner.observe(chat_ts) penalty = _state.increment_penalty() + + # 用户在聊天，正在跑的那一轮已经过时了，掐掉避免它把旧结论写回记忆 + self._abort_running_cycle() + + now_mono = time.monotonic() + if now_mono - self._last_replan_at < self._replan_throttle: + # 同窗口内的后续消息不重排，但仍把待执行任务往后挪， + # 否则旧任务会在用户刚开始的这轮活动里跑起来 + self._defer_pending_job(penalty, had_history) + await self._save_state_logged() + return + self._last_replan_at = now_mono + with contextlib.suppress(Exception): + scheduler.remove_job(self._job_id) + + delay_seconds = self._compute_delay(penalty, had_history) + self._schedule_once(delay_seconds) + # 惩罚计数与热度图落盘，避免重启后丢掉已累积的退避 + await self._save_state_logged() + + def _compute_delay(self, penalty: int, had_history: bool) -> int: + """算下一次运行要等多久（秒）。规划失败时退到保底窗口。""" base = self._config.rethink_base_delay_minutes - multiplier = self._config.rethink_penalty_multiplier cap = self._config.rethink_max_delay_minutes - raw = base * (multiplier ** (penalty - 1)) - delay_minutes = min(raw, cap) - logger.info( - f"[Subconscious] Rescheduled: penalty=#{penalty}, " - f"delay={delay_minutes:.0f}min (raw={raw:.0f}min, cap={cap}min)" - ) - self._schedule_once(int(delay_minutes * 60)) - # 惩罚计数落盘，避免重启后丢掉已累积的退避 - with contextlib.suppress(Exception): + if not self._config.use_idle_window_planner: + multiplier = self._config.rethink_penalty_multiplier + raw = base * (multiplier ** min(max(penalty - 1, 0), 32)) + delay_minutes = min(raw, cap) + logger.info( + f"[Subconscious] Rescheduled(legacy): penalty=#{penalty}, " + f"delay={delay_minutes:.0f}min (cap={cap}min)" + ) + return int(delay_minutes * 60) + + try: + if had_history: + plan = self._planner.plan(penalty) + else: + # 冷启动只攒到一条样本，不足以判断作息，先走保底窗口 + plan = self._planner.fallback_plan(penalty) + except Exception as e: + logger.opt(exception=e, colors=True, raw=True).error( + f"[Subconscious] Plan failed, fall back to fixed delay: {e}" + ) + return int(min(base, cap) * 60) + logger.info(f"[Subconscious] Rescheduled: penalty=#{penalty}, {describe(plan)}") + return int(max(60.0, (plan.start - datetime.now(timezone.utc)).total_seconds())) + + def _defer_pending_job(self, penalty: int, had_history: bool) -> None: + """节流期内不重新规划，只把待执行任务顺延到新的候选窗口。""" + try: + job = scheduler.get_job(self._job_id) + if job is None: + # 本就没有待执行任务（可能在运行中），补一个 + self._schedule_once(self._compute_delay(penalty, had_history)) + return + delay = self._compute_delay(penalty, had_history) + run_date = datetime.now(timezone.utc) + timedelta(seconds=delay) + if job.next_run_time is not None and job.next_run_time < run_date: + scheduler.reschedule_job( + self._job_id, trigger="date", run_date=run_date + ) + except Exception as e: + logger.warning(f"[Subconscious] Defer pending job failed: {e}") + + def _abort_running_cycle(self) -> None: + """用户发了新消息，掐断正在跑的那一轮。""" + task = self._cycle_task + if task is None or task.done(): + return + logger.info("[Subconscious] New chat arrived, aborting in-flight cycle") + task.cancel() + + async def _save_state_logged(self) -> None: + """保存状态，失败要出声：写盘失败会让重启后的退避凭空消失。""" + try: await self._save_state() + except Exception as e: + logger.warning(f"[Subconscious] Save state failed: {e}") def _schedule_once(self, delay_seconds: int) -> None: run_date = datetime.now(timezone.utc) + timedelta(seconds=delay_seconds) @@ -172,7 +259,12 @@ async def _run(self) -> None: logger.warning("[Subconscious] Already running, skip") return self._is_running = True + self._cycle_task = asyncio.current_task() + self._context_fetched = False try: + # 窗口已实际占用：记一次观测，避免下一轮重复选同一段 + with contextlib.suppress(Exception): + self._planner.note_window_taken(time.time()) await self._run_cycle() finally: self._is_running = False @@ -187,9 +279,9 @@ async def _run_cycle(self) -> None: return prompt_text = await self._load_prompt() user_msg_text = ( - f"现在是第 {self._total_runs + 1} 次潜意识推理循环。" - f"请检查用户 (ID: {self._config.target_user_id}) 的记忆库，执行必要的整理操作。" - f"完成后务必调用 subconscious_iter_stop。" + f"请先调用 subconscious_get_context 了解当前环境，" + f"再检查用户 (ID: {self._config.target_user_id}) 的记忆库，" + f"执行必要的整理操作。完成后务必调用 subconscious_iter_stop。" ) train = Message(role="system", content=prompt_text) user_input = user_msg_text @@ -210,9 +302,14 @@ async def _run_cycle(self) -> None: self._chat_obj = chat_obj try: - logger.info(f"[Subconscious] Cycle #{self._total_runs + 1}") + logger.info("[Subconscious] Cycle start") chat_obj.begin() await chat_obj + except asyncio.CancelledError: + # 用户发来新消息把本轮掐了：旧上下文作废，不写摘要也不动惩罚 + logger.info("[Subconscious] Cycle aborted by new user message") + self._chat_obj = None + raise except Exception as e: logger.opt(exception=e, colors=True, raw=True).exception( f"[Subconscious] Error: {e}" @@ -245,8 +342,12 @@ async def _post_process(self, chat_obj: ChatObject) -> None: f"[Subconscious] Abstract saved ({len(self._last_abstracts)}/{self._config.max_abstracts})" ) self._total_runs += 1 + _state.set_last_run_at(datetime.now(timezone.utc).timestamp()) - # 先持久化，任一失败则不重置惩罚（保留重试机会） + # 先持久化，任一失败则不重置惩罚（保留重试机会）。 + # 记下快照时的计数：保存期间用户可能又发消息把惩罚顶上去， + # 那部分增量不能被这次 reset 抹掉。 + penalty_before = _state.get_penalty_count() save_ok = True try: await self._save_state() @@ -254,7 +355,20 @@ async def _post_process(self, chat_obj: ChatObject) -> None: logger.warning(f"[Subconscious] Save state failed: {e}") save_ok = False - if save_ok: + if not save_ok: + logger.warning( + f"[Subconscious] Cycle done but save failed, " + f"penalty NOT reset (penalty={_state.get_penalty_count()})" + ) + elif not _state.is_cycle_valid(): + # 空转周期（没取上下文就收工）不清惩罚，保留压力等下一轮 + logger.warning("[Subconscious] Cycle invalid, penalty kept") + elif _state.get_penalty_count() != penalty_before: + # 保存期间用户又发了消息，新增的那部分惩罚不能被这次收尾抹掉 + logger.info( + "[Subconscious] Penalty kept: new chat arrived during the cycle" + ) + else: _state.reset_penalty() # 惩罚清零也要落盘，否则重启会加载旧退避 try: @@ -265,11 +379,6 @@ async def _post_process(self, chat_obj: ChatObject) -> None: f"[Subconscious] Cycle done (total_runs={self._total_runs}), " f"penalty reset, waiting for next user chat" ) - else: - logger.warning( - f"[Subconscious] Cycle done but save failed, " - f"penalty NOT reset (penalty={_state.get_penalty_count()})" - ) async def _update_usage(self, chat_obj: ChatObject) -> None: """更新全局 InsightsModel 与目标用户的 usage，复用 Bot 的 add_usage 逻辑。 @@ -368,9 +477,42 @@ async def _load_state(self) -> None: suggestions = payload.get("knowledge_suggestions", []) if isinstance(suggestions, list): _state.set_knowledge_suggestions(suggestions) + _state.set_last_iter_summary(str(payload.get("last_iter_summary", ""))) + _state.set_last_run_at(self._read_last_run_ts(payload)) + self._planner.load(payload) + # 旧 penalty 语义是"被推迟的次数"，与新语义（要求的冷段长度）不可换算， + # 升级后清零一次，避免带着旧值把门槛顶到上限。 + # 仍走旧调度时数值是自洽的，别动它。 + if ( + self._config.use_idle_window_planner + and "idle_planner" not in payload + and _state.get_penalty_count() > 0 + ): + logger.info( + "[Subconscious] Migrate: reset legacy penalty " + f"#{_state.get_penalty_count()} (old delay-based semantics)" + ) + _state.reset_penalty() except Exception as e: logger.warning(f"[Subconscious] Load state failed: {e}") + @staticmethod + def _read_last_run_ts(payload: dict[str, Any]) -> float: + """取上轮运行时间，兼容 0.7.1 只写 last_run_time 的旧快照。""" + raw = payload.get("last_run_ts") + try: + if raw is not None: + return float(raw or 0.0) + except (TypeError, ValueError): + pass + legacy = payload.get("last_run_time") + if isinstance(legacy, str) and legacy: + try: + return datetime.fromisoformat(legacy).timestamp() + except ValueError: + return 0.0 + return 0.0 + async def _migrate_legacy_state(self) -> dict[str, Any]: """迁移旧版本挪用的 Bot Memory 表元状态（extra_prompt/memory_json）。""" payload: dict[str, Any] = {} @@ -403,20 +545,48 @@ async def _migrate_legacy_state(self) -> dict[str, Any]: return payload async def _save_state(self) -> None: - """保存元状态到 SubconsciousState 表。""" - await self._write_state_payload( - _STATE_UID, - { - "total_runs": self._total_runs, - "last_run_time": datetime.now(timezone.utc).isoformat(), - "penalty": _state.get_penalty_count(), - "last_abstracts": self._last_abstracts, - "pending_messages": _state.get_pending(), - "knowledge_suggestions": _state.get_knowledge_suggestions(), - }, - ) + """保存元状态到 SubconsciousState 表。 + + 加锁串行化：并发调用时若旧快照后提交，会把新的热度与惩罚盖掉。 + """ + async with self._state_lock: + await self._save_state_locked() + + async def _save_state_locked(self) -> None: + """真正写盘的部分，调用方需已持有 _state_lock。""" + payload: dict[str, Any] = { + "total_runs": self._total_runs, + "last_run_time": datetime.now(timezone.utc).isoformat(), + "last_run_ts": _state.get_last_run_at(), + "last_iter_summary": _state.get_last_iter_summary(), + "penalty": _state.get_penalty_count(), + "last_abstracts": self._last_abstracts, + "pending_messages": _state.get_pending(), + "knowledge_suggestions": _state.get_knowledge_suggestions(), + } + # 只有启用空闲窗口调度时才写热度图：旧调度模式下写进去， + # 会让下次升级的迁移判断误以为已是新格式，跳过旧 penalty 清零 + if self._config.use_idle_window_planner: + payload.update(self._planner.dump()) + # 整行替换前先补齐老版本实例写过、本实例不认识的字段， + # 否则旧实例一存盘，热度图累积就被抹成"从没规划过" + try: + existing = await self._read_state_payload(_STATE_UID) + except Exception: + existing = {} + for key, value in existing.items(): + if key not in payload: + payload[key] = value + await self._write_state_payload(_STATE_UID, payload) async def _load_prompt(self) -> str: + """渲染 system prompt。 + + 只允许不变量进入 system：角色、规则、质量口径、输出格式， + 以及静态的画像/知识库指南。精确时间、上轮摘要、待办计数、 + 记忆库规模等动态信息一律交给 subconscious_get_context 按需拉取， + 这样跨轮的 system 前缀逐字节一致，能吃到 provider 的前缀缓存。 + """ main_path = (self._prompt_dir / self._config.prompt_file).resolve() kn_path = (self._prompt_dir / self._config.prompt_knowledge_file).resolve() pr_path = (self._prompt_dir / self._config.prompt_profile_file).resolve() @@ -427,62 +597,25 @@ async def _load_prompt(self) -> str: character_prompt = await load_character_prompt() template: Template = Template(main_path.read_text(encoding="utf-8")) - now = datetime.now(timezone.utc) + # 日粒度日期：每天才变一次，不影响缓存命中，又能给记忆新旧提供锚点 + current_date = datetime.now(timezone.utc).strftime("%Y-%m-%d") prompt = await asyncio.to_thread( template.render, - last_abstracts=self._last_abstracts, - last_run=(self._last_abstracts[-1] if self._last_abstracts else ""), - current_time=now.strftime("%Y-%m-%d %H:%M:%S UTC"), + current_date=current_date, target_user_id=self._config.target_user_id or "(未配置)", - total_runs=self._total_runs, character_prompt=character_prompt, ) - # 注入知识库 & 画像指南（Jinja2 渲染，支持变量插值） - common_vars = { - "current_time": now.strftime("%Y-%m-%d %H:%M:%S UTC"), + # 指南文件仍走 Jinja2，但只提供不变量：旧文件里的 {{ current_time }} + # 会被渲染成空串，而不是原样泄漏到 system 里 + static_vars = { + "current_date": current_date, "target_user_id": self._config.target_user_id or "(未配置)", } - prompt += "\n\n" + await asyncio.to_thread( - Template(kn_path.read_text(encoding="utf-8")).render, **common_vars - ) - prompt += "\n\n" + await asyncio.to_thread( - Template(pr_path.read_text(encoding="utf-8")).render, **common_vars - ) - - # 待审查的知识建议 — 注入提示让 Agent 主动拉取 - suggestions = _state.get_knowledge_suggestions() - if suggestions: - prompt += ( - f"\n\n📋 有 {len(suggestions)} 条来自对话 LLM 的知识建议等待审查。" - f"请调用 subconscious_read_suggestions 获取详情，" - f"审查后对值得保留的用 subconscious_knowledge_create/update 实际写入。" + for guide in (kn_path, pr_path): + prompt += "\n\n" + await asyncio.to_thread( + Template(guide.read_text(encoding="utf-8")).render, **static_vars ) - - # 膨胀感知：查 ChromaDB 总量，超阈值注入警告 - try: - pid = make_scope_id(self._config.target_user_id, is_group=False) - ope = AsyncUserMemory(get_db_conn()) - await ope.init() - # 先取轻量计数，超阈值才拉全量算分布 - total = await ope.count_user_notes(pid) - if total > self._config.memory_warn_threshold: - result = await ope.get_all_notes(pid, include=["metadatas"]) - low_count = sum( - 1 - for m in (result.get("metadatas") or []) - if isinstance(m, dict) and m.get("importance") == "low" - ) - prompt += ( - f"\n\n⚠️ 当前记忆库有 {total} 条记忆" - f"（其中低重要性 {low_count} 条）。" - f"建议本轮优先调用 subconscious_get_memory_stats 分析分布，" - f"然后用 subconscious_duplicate_helper 获取待整理列表，" - f"进行合并去重和低质记忆清理。" - ) - except Exception as e: - logger.debug(f"[Subconscious] Memory stats query failed: {e}") - return prompt async def _read_recent_sessions(self, n: int = 5) -> list[SessionSummary]: @@ -556,45 +689,7 @@ async def _read_profile( self, start_line: int | None = None, end_line: int | None = None ) -> ProfileResult: """读取用户画像 Markdown 文件，支持行数滑动窗口。""" - if not self._profile_path.exists(): - return { - "summary": "", - "content": "", - "total_lines": 0, - "range_start": 0, - "range_end": 0, - } - try: - text = self._profile_path.read_text(encoding="utf-8") - parts = text.split("---", 1) - summary = parts[0].strip() if parts else "" - body = parts[1].strip() if len(parts) > 1 else "" - body_lines = body.split("\n") if body else [] - total = len(body_lines) - if start_line is None: - s = 0 - else: - s = max(0, start_line) - if end_line is None: - e = total - else: - e = max(s, min(total, end_line)) - return { - "summary": summary, - "content": "\n".join(body_lines[s:e]), - "total_lines": total, - "range_start": s, - "range_end": e, - } - except Exception as e: - logger.warning(f"[Subconscious] Read profile failed: {e}") - return { - "summary": "", - "content": "", - "total_lines": 0, - "range_start": 0, - "range_end": 0, - } + return read_profile(self._profile_path, start_line, end_line) async def _update_profile( self, diff --git a/amrita_plugin_memory/rethinking/schedule.py b/amrita_plugin_memory/rethinking/schedule.py new file mode 100644 index 0000000..b1b7006 --- /dev/null +++ b/amrita_plugin_memory/rethinking/schedule.py @@ -0,0 +1,702 @@ +"""DMN 空闲窗口调度：7×12 热度图 + 冷段扫描 + 段内二分。 + +设计要点 +- 观测：用户每条消息让对应 (星期几, 2h 槽) 权重 +1，并随时间衰减。 +- 分层：fine(单片星期几) / mid(工作日-周末) / coarse(全局)，逐级降级。 +- 收缩：fine/mid 向 coarse 按有效样本天数加权，避免稀薄样本失真。 +- 规划：环形找连续冷槽 -> 段内用原始时间戳做分钟级二分 -> 扣 buffer 得可用窗口。 +- 保底：冷段不足门槛时逐次减半放宽，最后强制取最长冷段。 + +本模块不依赖 nonebot，纯数据 + 纯函数，可脱离运行时单独验证。 +""" + +from __future__ import annotations + +import logging +import time +from collections.abc import Sequence +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone, tzinfo +from typing import Any +from zoneinfo import ZoneInfo + +SLOT_MINUTES = 120 +SLOTS_PER_DAY = 12 +DAYS = 7 +CELLS = DAYS * SLOTS_PER_DAY +MINUTES_PER_DAY = SLOT_MINUTES * SLOTS_PER_DAY + +#: 最近消息时间戳缓冲上限（分钟级细化用，不能无限涨） +_MAX_RECENT = 500 +#: 活跃日期登记上限 +_MAX_SEEN_DAYS = 240 + +_LOG = logging.getLogger(__name__) + +_WD_NAMES = ("周一", "周二", "周三", "周四", "周五", "周六", "周日") + + +def resolve_timezone(name: str | None) -> tuple[tzinfo, str, bool]: + """把配置里的时区名解析成 tzinfo。 + + 返回 (tz, 实际生效的名字, 是否回退)。解析失败回退 UTC+08:00， + 会话侧与调度侧共用这一份，避免两边报出不同的钟点。 + """ + raw = (name or "").strip() or "Asia/Shanghai" + try: + return ZoneInfo(raw), raw, False + except Exception: + return timezone(timedelta(hours=8)), "UTC+08:00", True + + +def _cell(wd: int, slot: int) -> int: + return wd * SLOTS_PER_DAY + (slot % SLOTS_PER_DAY) + + +def slot_of(ts: float, tz: tzinfo) -> tuple[int, int]: + """把 unix 时间戳换成 (星期几 0=周一, 槽位 0..11)。""" + dt = datetime.fromtimestamp(ts, tz) + wd = dt.weekday() + slot = (dt.hour * 60 + dt.minute) // SLOT_MINUTES + return wd, slot + + +def merge_noise(ts: Sequence[float], gap_seconds: float) -> list[float]: + """把一波连发的消息压成单点：间距小于 gap 的后来者直接并掉。""" + out: list[float] = [] + prev: float | None = None + for t in sorted(ts): + if prev is not None and t - prev < gap_seconds: + continue + out.append(t) + prev = t + return out + + +def free_gaps( + ts: Sequence[float], start: float, end: float +) -> list[tuple[float, float]]: + """返回 [start, end) 内的全部无消息区间。""" + if end <= start: + return [] + inner = sorted(t for t in ts if start < t < end) + pts = [start, *inner, end] + return [(pts[i], pts[i + 1]) for i in range(len(pts) - 1)] + + +def longest_free(ts: Sequence[float], start: float, end: float) -> tuple[float, float]: + """[start, end) 内最长的无消息区间。""" + gaps = free_gaps(ts, start, end) + if not gaps: + return start, end + return max(gaps, key=lambda g: g[1] - g[0]) + + +def feasible_minutes( + ts: Sequence[float], start: float, end: float, minutes: int +) -> bool: + """是否存在一段连续 >= minutes 的空闲。（随 minutes 单调递减）""" + if minutes <= 0: + return end > start + need = minutes * 60 + return any(b - a >= need for a, b in free_gaps(ts, start, end)) + + +def bisect_free_minutes( + ts: Sequence[float], start: float, end: float, hi: int | None = None +) -> int: + """二分求窗口内最长连续空闲（分钟）。判定函数单调，收敛到分钟精度。""" + if end <= start: + return 0 + top = hi if hi is not None else int((end - start) // 60) + lo = 0 + while lo < top: + mid = (lo + top + 1) // 2 + if feasible_minutes(ts, start, end, mid): + lo = mid + else: + top = mid - 1 + return lo + + +def _normalize(values: Sequence[float]) -> list[float]: + total = sum(values) + if total <= 1e-9: + return [0.0] * len(values) + return [v / total for v in values] + + +def _runs( + mask: Sequence[bool], limit: int, circular: bool = True +) -> list[tuple[int, int]]: + """找连续 True 段，返回 (起点, 长度)，长度不超过 limit。 + + circular=True 用于单日 12 槽（跨午夜是本日内的相邻）， + 拼接两天的 24 槽必须传 False——否则首尾会被当成相邻，选出一周之外的窗。 + """ + n = len(mask) + if n == 0: + return [] + if all(mask): + return [(0, min(n, limit))] if not circular else [(0, limit)] + if not any(mask): + return [] + if not circular: + out: list[tuple[int, int]] = [] + i = 0 + while i < n: + if not mask[i]: + i += 1 + continue + j = i + while j < n and mask[j]: + j += 1 + out.append((i, min(j - i, limit))) + i = j + return out + start = mask.index(False) + seq = list(mask[start:]) + list(mask[:start]) + out: list[tuple[int, int]] = [] + i = 0 + while i < len(seq): + if not seq[i]: + i += 1 + continue + j = i + while j < len(seq) and seq[j]: + j += 1 + out.append(((start + i) % n, min(j - i, limit))) + i = j + return out + + +@dataclass +class Plan: + """一次调度规划的结果。""" + + start: datetime + end: datetime + length_minutes: int + tier: str = "coarse" + relaxed_steps: int = 0 + forced: bool = False + penalty: int = 0 + note: str = "" + + @property + def delay_seconds(self) -> int: + return max(1, int((self.start - datetime.now(timezone.utc)).total_seconds())) + + def to_dict(self) -> dict[str, Any]: + return { + "start": self.start.isoformat(), + "end": self.end.isoformat(), + "length_minutes": self.length_minutes, + "tier": self.tier, + "relaxed_steps": self.relaxed_steps, + "forced": self.forced, + "penalty": self.penalty, + "note": self.note, + } + + +@dataclass +class PlannerState: + """可持久化的热度图状态。""" + + fine: list[float] = field(default_factory=lambda: [0.0] * CELLS) + coarse: list[float] = field(default_factory=lambda: [0.0] * SLOTS_PER_DAY) + seen_days: list[str] = field(default_factory=list) + recent: list[float] = field(default_factory=list) + last_update: float = 0.0 + #: 热度图对应的时区名与目标用户，任一变化都要重建，否则星期几/钟点对不上 + tz_name: str = "" + uid: str = "" + + def to_dict(self) -> dict[str, Any]: + return { + "fine": [round(v, 6) for v in self.fine], + "coarse": [round(v, 6) for v in self.coarse], + "seen_days": self.seen_days[-_MAX_SEEN_DAYS:], + "recent": self.recent[-_MAX_RECENT:], + "last_update": self.last_update, + "tz_name": self.tz_name, + "uid": self.uid, + } + + +class IdlePlanner: + """基于热度图的空闲窗口规划器。""" + + def __init__(self, cfg: Any) -> None: + self._cfg = cfg + tz_name = getattr(cfg, "idle_timezone", "Asia/Shanghai") + self.tz, self.tz_name, self.tz_fallback = resolve_timezone(tz_name) + self.uid = str(getattr(cfg, "target_user_id", "") or "") + self.state = PlannerState() + + self.noise_gap = getattr(cfg, "idle_noise_gap_minutes", 15) * 60 + self.tau_fine = max(1.0, getattr(cfg, "idle_fine_tau_days", 28)) + self.tau_coarse = max(1.0, getattr(cfg, "idle_coarse_tau_days", 14)) + self.shrink_k = max(0.0, getattr(cfg, "idle_shrink_k", 4)) + self.cold_pct_max = getattr(cfg, "idle_cold_pct_max", 0.05) + self.cold_abs_max = getattr(cfg, "idle_cold_abs_max", 1.0) + self.min_weeks_fine = getattr(cfg, "idle_min_sample_days_fine", 6) + self.min_weeks_mid = getattr(cfg, "idle_min_sample_days_mid", 3) + self.target_hour = getattr(cfg, "idle_target_hour", 3) + self.base_slots = getattr(cfg, "idle_base_slots", 3) + self.max_slots = getattr(cfg, "idle_max_slots", 6) + self.multiplier = getattr(cfg, "idle_penalty_multiplier", 1.5) + self.max_relax = getattr(cfg, "idle_max_relax_steps", 3) + self.buffer_minutes = getattr(cfg, "idle_buffer_minutes", 15) + self.run_estimate = getattr(cfg, "idle_run_estimate_minutes", 10) + self.fallback_minutes = getattr(cfg, "idle_fallback_minutes", 120) + self._last_plan: Plan | None = None + + def load(self, payload: dict[str, Any]) -> None: + raw = payload.get("idle_planner") if isinstance(payload, dict) else None + if not isinstance(raw, dict): + return + saved_tz = str(raw.get("tz_name", "") or "") + saved_uid = str(raw.get("uid", "") or "") + # 时区或目标用户换了，旧星期几/钟点的计数就不能再用 + if saved_tz and saved_tz != self.tz_name: + _LOG.info( + f"[Subconscious] Planner history dropped: tz {saved_tz} -> {self.tz_name}" + ) + return + if saved_uid and self.uid and saved_uid != self.uid: + _LOG.info( + f"[Subconscious] Planner history dropped: uid {saved_uid} -> {self.uid}" + ) + return + try: + fine = [float(v) for v in raw.get("fine", [])] + coarse = [float(v) for v in raw.get("coarse", [])] + if len(fine) == CELLS: + self.state.fine = fine + if len(coarse) == SLOTS_PER_DAY: + self.state.coarse = coarse + seen = raw.get("seen_days", []) + if isinstance(seen, list): + self.state.seen_days = [str(d) for d in seen][-_MAX_SEEN_DAYS:] + recent = raw.get("recent", []) + if isinstance(recent, list): + self.state.recent = [float(v) for v in recent][-_MAX_RECENT:] + self.state.last_update = float(raw.get("last_update", 0.0)) + self.state.tz_name = self.tz_name + self.state.uid = self.uid + except (TypeError, ValueError): + self.state = PlannerState() + + def dump(self) -> dict[str, Any]: + return {"idle_planner": self.state.to_dict()} + + def _decay(self, now_ts: float) -> None: + if self.state.last_update <= 0: + self.state.last_update = now_ts + return + dt_days = (now_ts - self.state.last_update) / 86400 + if dt_days <= 1e-6: + return + f_fine = 2 ** (-dt_days / self.tau_fine) + f_coarse = 2 ** (-dt_days / self.tau_coarse) + self.state.fine = [v * f_fine for v in self.state.fine] + self.state.coarse = [v * f_coarse for v in self.state.coarse] + self.state.last_update = now_ts + + def observe(self, ts: float | None = None) -> None: + """记录一条用户消息，加深该时间窗口的热点。""" + now_ts = time.time() if ts is None else float(ts) + self._decay(now_ts) + wd, slot = slot_of(now_ts, self.tz) + self.state.fine[_cell(wd, slot)] += 1.0 + self.state.coarse[slot] += 1.0 + day_key = datetime.fromtimestamp(now_ts, self.tz).strftime("%Y-%m-%d") + if day_key not in self.state.seen_days: + self.state.seen_days.append(day_key) + if len(self.state.seen_days) > _MAX_SEEN_DAYS: + self.state.seen_days = self.state.seen_days[-_MAX_SEEN_DAYS:] + self.state.tz_name = self.tz_name + self.state.uid = self.uid + self.state.recent.append(now_ts) + if len(self.state.recent) > _MAX_RECENT: + self.state.recent = self.state.recent[-_MAX_RECENT:] + + def last_plan(self) -> Plan | None: + return self._last_plan + + def last_plan_dict(self) -> dict[str, Any] | None: + plan = self._last_plan + if plan is None: + return None + local_end = plan.end.astimezone(self.tz) + return { + "window_start": plan.start.strftime("%Y-%m-%d %H:%M"), + "window_end": local_end.strftime("%Y-%m-%d %H:%M"), + "planned_minutes": plan.length_minutes, + "tier": plan.tier, + "relaxed_steps": plan.relaxed_steps, + "forced": plan.forced, + "note": plan.note, + } + + def note_window_taken(self, start_ts: float) -> None: + """窗口实际被占用后，把该时段计入一次观测（避免反复选同一段）。""" + self.observe(start_ts + 60) + + def active_days_of(self, weekdays: Sequence[int]) -> int: + """统计近期活跃天数：超过粗层半衰期 4 倍的老日期不再计入。 + + 不设上限的话，用户换作息后旧日期会永久撑着 fine/mid 层， + 稀薄但"看起来够量"的样本反而比全局层更不可信。 + """ + horizon = max(1, int(self.tau_coarse * 4)) + cutoff = datetime.now(self.tz) - timedelta(days=horizon) + cnt = 0 + for key in set(self.state.seen_days): + try: + dt = datetime.strptime(key, "%Y-%m-%d") + except ValueError: + continue + if dt.replace(tzinfo=self.tz) < cutoff: + continue + if dt.weekday() in weekdays: + cnt += 1 + return cnt + + def _shrink(self, cells: Sequence[float], n_days: int) -> list[float]: + """把细层分布按可信度向全局分布收缩。 + + 两层先各自归一化再混合：细层一天一次采样、全局层跨所有天累加， + 直接相加会让全局层凭绝对计数压过 alpha 想表达的信任度。 + """ + total = n_days + self.shrink_k + alpha = n_days / total if total > 0 else 0.0 + base = _normalize(self.state.coarse) + cur = _normalize(cells) + return [alpha * cur[i] + (1 - alpha) * base[i] for i in range(SLOTS_PER_DAY)] + + def tier_table(self, tier: str, wd: int) -> tuple[list[float], list[float], int]: + """返回 (槽位权重, 槽位原始计数, 有效样本天数)。""" + if tier == "fine": + cells = self.state.fine[wd * SLOTS_PER_DAY : (wd + 1) * SLOTS_PER_DAY] + n_days = self.active_days_of([wd]) + return self._shrink(cells, n_days), list(cells), n_days + if tier == "mid": + days = [0, 1, 2, 3, 4] if wd <= 4 else [5, 6] + cells = [ + sum(self.state.fine[_cell(d, s)] for d in days) / len(days) + for s in range(SLOTS_PER_DAY) + ] + n_days = self.active_days_of(days) + return self._shrink(cells, n_days), cells, n_days + cells = list(self.state.coarse) + return cells, cells, self.active_days_of(list(range(DAYS))) + + def _tiers_for(self, wd: int) -> list[str]: + tiers: list[str] = [] + if self.active_days_of([wd]) >= self.min_weeks_fine: + tiers.append("fine") + days = [0, 1, 2, 3, 4] if wd <= 4 else [5, 6] + if self.active_days_of(days) >= self.min_weeks_mid: + tiers.append("mid") + tiers.append("coarse") + return tiers + + def _tier_for_neighbor(self, tier: str, wd: int) -> str: + """跨午夜扫描时，给相邻星期几单独挑一档可用分层。 + + 直接沿用当天的 tier，会在相邻那天样本不足时拿 0 计数参与判定， + 那些槽位会被误判成冷槽。 + """ + if tier == "fine" and self.active_days_of([wd]) < self.min_weeks_fine: + return self._tier_for_neighbor("mid", wd) + if tier == "mid": + days = [0, 1, 2, 3, 4] if wd <= 4 else [5, 6] + if self.active_days_of(days) < self.min_weeks_mid: + return "coarse" + return tier + + def _cold_runs(self, tier: str, wd: int) -> list[tuple[int, int, int]]: + """返回 [(环形起点槽, 长度, 归属星期几)]，长度上限一天。""" + weights, counts, _ = self.tier_table(tier, wd) + pcts = _normalize(weights) + mask = [ + pcts[s] <= self.cold_pct_max and counts[s] <= self.cold_abs_max + for s in range(SLOTS_PER_DAY) + ] + # 跨午夜：接上下一星期几的表，否则 22:00-08:00 的安静段会被切断 + wd2 = (wd + 1) % DAYS + tier2 = self._tier_for_neighbor(tier, wd2) + w2, c2, _ = self.tier_table(tier2, wd2) + p2 = _normalize(w2) + mask2 = [ + p2[s] <= self.cold_pct_max and c2[s] <= self.cold_abs_max + for s in range(SLOTS_PER_DAY) + ] + # 拼接表首尾在真实时间上相隔整整一天，不能按环形再接回去 + ring = mask + mask2 + return [ + (start, length, wd) for start, length in _runs(ring, SLOTS_PER_DAY, False) + ] + + def _lightest_run( + self, tier: str, wd: int, length_slots: int, now: datetime + ) -> tuple[int, int] | None: + """环形滑动窗口找总权重最低的连续 length_slots 槽段。 + + 权重相同时取最近一次将要到来的那段，避免全天打平时 + 因为起点槽已过而把窗口推到下周。 + """ + weights, _, _ = self.tier_table(tier, wd) + pcts = _normalize(weights) + if sum(pcts) <= 1e-9: + return None + n = SLOTS_PER_DAY + ring = pcts + pcts + best: tuple[int, int] | None = None + best_key: tuple[float, float] | None = None + for start in range(n): + total = sum(ring[start : start + length_slots]) + seg_start, _ = self._seg_bounds(start, length_slots, wd, now) + delay = max(0.0, (seg_start - now.astimezone(self.tz)).total_seconds()) + key = (total, delay) + if best_key is None or key < best_key: + best_key = key + best = (start, length_slots) + return best + + def _next_occurrence(self, wd: int, now: datetime) -> datetime: + """本周/下周该星期几的 00:00（不早于 now 所在日期）。""" + today = now.astimezone(self.tz).replace( + hour=0, minute=0, second=0, microsecond=0 + ) + delta = (wd - today.weekday()) % DAYS + cand = today + timedelta(days=delta) + if cand + timedelta(days=1) <= now.astimezone(self.tz): + cand += timedelta(days=DAYS) + return cand + + def _seg_bounds( + self, start_slot: int, length_slots: int, wd: int, now: datetime + ) -> tuple[datetime, datetime]: + """冷段对应的绝对时间区间；已过去的按下一个同类星期几算。""" + day0 = self._next_occurrence(wd, now) + seg_start = day0 + timedelta(minutes=start_slot * SLOT_MINUTES) + seg_end = seg_start + timedelta(minutes=length_slots * SLOT_MINUTES) + now_local = now.astimezone(self.tz) + if seg_end <= now_local: + seg_start += timedelta(days=DAYS) + seg_end += timedelta(days=DAYS) + return seg_start, seg_end + + def _refine( + self, + start_slot: int, + length_slots: int, + wd: int, + now: datetime, + tier: str, + penalty: int, + relaxed: int, + forced: bool, + ) -> Plan | None: + """把冷段换成具体时间窗：段内找最长空闲，二分校核，扣 buffer。 + + 段内连一次运行都放不下时返回 None，交给调用方换下一段， + 而不是报出一个实际用不了的时间窗。 + """ + seg_start, seg_end = self._seg_bounds(start_slot, length_slots, wd, now) + now_local = now.astimezone(self.tz) + if seg_start < now_local: + seg_start = now_local + timedelta(minutes=1) + if seg_end <= seg_start: + return None + + # 只信原始时间戳：连发消息先合并，再取段内最长无消息区间 + ts = merge_noise(self.state.recent, self.noise_gap) + s_ts, e_ts = seg_start.timestamp(), seg_end.timestamp() + f_start, f_end = longest_free(ts, s_ts, e_ts) + capacity = bisect_free_minutes(ts, s_ts, e_ts) + gap_minutes = int((f_end - f_start) // 60) + # 二分给出理论上界，实际可用不得超出真实无消息区间 + free_minutes = min(capacity, gap_minutes) + + need = self.buffer_minutes + self.run_estimate + target = seg_start.replace( + hour=self.target_hour, minute=0, second=0, microsecond=0 + ) + if target < seg_start: + target += timedelta(days=1) + # 起点尽量靠近偏好时刻，但必须落在空闲段内且留够运行空间 + latest_start = max(f_start, f_end - need * 60) + start_ts = min(max(f_start, target.timestamp()), latest_start) + start_ts = max(start_ts, f_start, now.timestamp() + 60) + + note = "" + if f_end - start_ts < need * 60: + # 段内剩余空间放不下一次运行，退回段首再算一次 + start_ts = f_start + note = "insufficient_free_space" + if f_end - start_ts < 60: + return None + + start = datetime.fromtimestamp(start_ts, self.tz) + # 尾部 buffer 不纳入回报窗口，否则 end 会把安全余量也算成可用时间 + usable_end = datetime.fromtimestamp(f_end, self.tz) - timedelta( + minutes=self.buffer_minutes + ) + end = min(start + timedelta(minutes=free_minutes), usable_end) + length_minutes = int((end - start).total_seconds() // 60) + if length_minutes < 1: + return None + return Plan( + start=start, + end=end, + length_minutes=length_minutes, + tier=tier, + relaxed_steps=relaxed, + forced=forced, + penalty=penalty, + note=note, + ) + + def _fallback(self, now: datetime, penalty: int, note: str) -> Plan: + local = now.astimezone(self.tz) + target = local.replace(hour=self.target_hour, minute=0, second=0, microsecond=0) + if target <= local + timedelta(minutes=1): + target += timedelta(days=1) + return Plan( + start=target, + end=target + timedelta(minutes=self.fallback_minutes), + length_minutes=self.fallback_minutes, + tier="fallback", + forced=True, + penalty=penalty, + note=note, + ) + + def has_history(self) -> bool: + """热度图是否有可用样本。""" + return sum(self.state.coarse) > 1e-6 + + def fallback_plan( + self, penalty: int = 1, now: datetime | None = None, note: str = "no_history" + ) -> Plan: + """直接给出保底窗口，不做冷段扫描。""" + now = now or datetime.now(timezone.utc) + return self._fallback(now.astimezone(self.tz), penalty, note) + + def plan(self, penalty: int = 1, now: datetime | None = None) -> Plan: + result = self._plan(penalty, now) + self._last_plan = result + return result + + def _want_slots(self, penalty: int) -> int: + """按惩罚值算本次要求的冷段槽数，指数增长但先封顶再取整。""" + exp = max(0, penalty - 1) + try: + want = self.base_slots * (self.multiplier**exp) + except OverflowError: + # 惩罚值离谱时直接给上限，不要让一次计算失败把调度整条打断 + return self.max_slots + if want != want: # NaN 保护 + return self.base_slots + return max(1, min(round(want), self.max_slots)) + + def _pick_run( + self, runs: Sequence[tuple[int, int, int]], now: datetime + ) -> tuple[int, int, int] | None: + """在满足门槛的冷段里挑一段：优先最近的、尚未来临的、较长的。""" + best: tuple[int, int, int] | None = None + best_key: tuple[int, float, int] | None = None + for start_slot, length_slots, owner in runs: + seg_start, _ = self._seg_bounds(start_slot, length_slots, owner, now) + delay = max(0.0, (seg_start - now.astimezone(self.tz)).total_seconds()) + key = (0 if delay > 0 else 1, delay, -length_slots) + if best_key is None or key < best_key: + best_key = key + best = (start_slot, length_slots, owner) + return best + + def _plan(self, penalty: int = 1, now: datetime | None = None) -> Plan: + now = now or datetime.now(timezone.utc) + local = now.astimezone(self.tz) + wd = local.weekday() + + if sum(self.state.coarse) <= 1e-6: + return self._fallback(local, penalty, "no_history") + + want = self._want_slots(penalty) + + tiers = self._tiers_for(wd) + for step in range(self.max_relax + 1): + need_slots = max(1, want >> step) + for tier in tiers: + runs = [r for r in self._cold_runs(tier, wd) if r[1] >= need_slots] + picked = self._pick_run(runs, now) + if picked is None: + continue + plan = self._refine( + picked[0], + picked[1], + picked[2], + now, + tier, + penalty, + step, + forced=False, + ) + if plan is not None: + return plan + + # 全部放宽仍不足：强制取最长冷段 + for tier in tiers: + runs = sorted(self._cold_runs(tier, wd), key=lambda r: -r[1]) + picked = self._pick_run(runs, now) + if picked is None: + continue + plan = self._refine( + picked[0], + picked[1], + picked[2], + now, + tier, + penalty, + self.max_relax, + forced=True, + ) + if plan is not None: + return plan + + # 完全没有可用冷段（用户几乎全天活跃）：退到热度最低的连续段 + for tier in tiers: + floor_run = self._lightest_run(tier, wd, 1, now) + if floor_run is None: + continue + plan = self._refine( + floor_run[0], + floor_run[1], + wd, + now, + tier, + penalty, + self.max_relax, + forced=True, + ) + if plan is not None: + return plan + return self._fallback(local, penalty, "no_cold_segment") + + +def describe(plan: Plan) -> str: + """日志用的一行描述。""" + local = plan.start + wd = _WD_NAMES[local.weekday()] + return ( + f"{wd} {local.strftime('%H:%M')}-{plan.end.strftime('%H:%M')} " + f"({plan.length_minutes}min, tier={plan.tier}, " + f"relax={plan.relaxed_steps}, forced={plan.forced}" + f"{', ' + plan.note if plan.note else ''})" + ) diff --git a/amrita_plugin_memory/rethinking/schemas.py b/amrita_plugin_memory/rethinking/schemas.py index 9c45534..59d536b 100644 --- a/amrita_plugin_memory/rethinking/schemas.py +++ b/amrita_plugin_memory/rethinking/schemas.py @@ -192,6 +192,17 @@ ), ) +GET_CONTEXT_SCHEMA = FunctionDefinitionSchema( + name="subconscious_get_context", + description=( + "获取本轮运行环境：精确时间、距上轮间隔、上轮摘要、待审查知识建议、" + "记忆库规模与热度窗口。每轮应先调用本工具，再决定做什么。" + "返回中的 stale 字段列出本次未取到的项，不要推测它们的值。" + ), + parameters=FunctionParametersSchema(type="object", properties={}, required=[]), +) + + GET_MEMORY_STATS_SCHEMA = FunctionDefinitionSchema( name="subconscious_get_memory_stats", description="获取记忆库统计概览：总数、重要性分布、标签分布、时间范围。用于判断是否需要压缩。", @@ -380,6 +391,27 @@ ), ) +READ_PROFILE_SCHEMA = FunctionDefinitionSchema( + name="read_profile", + description=( + "读取本机离线巩固循环为你维护的用户画像（摘要 + 正文）。" + "返回值含 summary、content、total_lines、range_start、range_end；" + "正文较长时用 start_line / end_line 分段读取。仅画像所有者本人可读" + ), + parameters=FunctionParametersSchema( + type="object", + properties={ + "start_line": FunctionPropertySchema( + type="integer", description="起始行号（0-based），默认全文" + ), + "end_line": FunctionPropertySchema( + type="integer", description="结束行号（不含），默认全文" + ), + }, + required=[], + ), +) + UPDATE_PROFILE_SCHEMA = FunctionDefinitionSchema( name="subconscious_update_profile", description=( @@ -418,6 +450,7 @@ LIST_MEMORY_SCHEMA, ITER_STOP_SCHEMA, DUPLICATE_HELPER_SCHEMA, + GET_CONTEXT_SCHEMA, GET_MEMORY_STATS_SCHEMA, KNOWLEDGE_LIST_SCHEMA, KNOWLEDGE_READ_SCHEMA, diff --git a/amrita_plugin_memory/rethinking/tools.py b/amrita_plugin_memory/rethinking/tools.py index 06a4d3b..e026504 100644 --- a/amrita_plugin_memory/rethinking/tools.py +++ b/amrita_plugin_memory/rethinking/tools.py @@ -20,13 +20,20 @@ from nonebot import logger from ..keys import make_scope_id -from ..permissions import can_read_knowledge, get_event_from_tool_ctx +from ..permissions import ( + can_read_knowledge, + can_read_profile, + get_event_from_tool_ctx, +) +from ..profile_store import PROFILE_PATH, read_profile from ..vector import AsyncUserMemory, MemoryMetadata, get_db_conn from . import _state from .consts import DEFAULT_SEND_PROMPT, ensure_prompt_file, load_character_prompt +from .schedule import resolve_timezone from .schemas import ( DELETE_MEMORY_SCHEMA, DUPLICATE_HELPER_SCHEMA, + GET_CONTEXT_SCHEMA, GET_MEMORY_STATS_SCHEMA, GET_PROFILE_SCHEMA, ITER_STOP_SCHEMA, @@ -40,6 +47,7 @@ LIST_MEMORY_SCHEMA, READ_CHAT_CONTEXT_SCHEMA, READ_MEMORY_SCHEMA, + READ_PROFILE_SCHEMA, READ_SESSIONS_SCHEMA, READ_SUGGESTIONS_SCHEMA, SEND_TO_USER_SCHEMA, @@ -57,6 +65,9 @@ _DUPLICATE_DEFAULT_LIMIT = 40 _DUPLICATE_MAX_LIMIT = 200 +#: get_context 里向量库操作的超时上限（秒），超时只标 stale 不阻塞循环 +_CONTEXT_DB_TIMEOUT = 8.0 + _SUBCONSCIOUS_TOOLS = _state.get_tools_manager() @@ -241,9 +252,18 @@ async def subconscious_list_memory(data: dict[str, Any]) -> str: @on_tools(ITER_STOP_SCHEMA, strict=True, bound_to=_SUBCONSCIOUS_TOOLS) async def subconscious_iter_stop(data: dict[str, Any]) -> str: - logger.info( - f"[Subconscious] iter_stop: summary={str(data.get('summary', ''))[:80]}" - ) + summary = str(data.get("summary", "")) + if not _state.is_context_fetched(): + # 没取环境就收工：周期算无效，不写摘要。惩罚保持不清零， + # 下一次调度仍带着这轮攒的量，避免"空转一轮还被当成完成" + logger.warning("[Subconscious] iter_stop without get_context, cycle invalid") + _state.mark_cycle_invalid() + return _tools_err( + "本轮未调用 subconscious_get_context，周期无效。" + "请先取当前环境再决定是否整理记忆" + ) + _state.set_last_iter_summary(summary) + logger.info(f"[Subconscious] iter_stop: summary={summary[:80]}") return json.dumps( {"status": "acknowledged", "message": "推理循环已确认结束"}, ensure_ascii=False ) @@ -580,6 +600,95 @@ async def subconscious_duplicate_helper(data: dict[str, Any]) -> str: return _tools_err(str(e)) +@on_tools(GET_CONTEXT_SCHEMA, strict=True, bound_to=_SUBCONSCIOUS_TOOLS) +async def subconscious_get_context(data: dict[str, Any]) -> str: + """聚合本轮运行环境；单项失败只标 stale，不中断。""" + stale: list[str] = [] + runner = _state.get_runner() + cfg = runner._config if runner is not None else None + + tz_cfg = getattr(cfg, "idle_timezone", "Asia/Shanghai") if cfg else "Asia/Shanghai" + # 与调度侧共用同一份解析，避免规划窗口与回报钟点分处两个时区 + tz, tz_name, tz_fell_back = resolve_timezone(tz_cfg) + if tz_fell_back: + stale.append("timezone") + + now = datetime.now(timezone.utc) + local = now.astimezone(tz) + + last_at = _state.get_last_run_at() + if last_at > 0: + gap_minutes = int((now.timestamp() - last_at) // 60) + last_run_iso = datetime.fromtimestamp(last_at, tz).isoformat() + else: + gap_minutes = None + last_run_iso = None + + suggested = _state.get_knowledge_suggestions() + + total = None + low_count = None + threshold = None + memory_hint = "" + try: + ope = _make_operator() + await asyncio.wait_for(ope.init(), timeout=_CONTEXT_DB_TIMEOUT) + pid = _get_partition_id() + total = await asyncio.wait_for( + ope.count_user_notes(pid), timeout=_CONTEXT_DB_TIMEOUT + ) + threshold = getattr(cfg, "memory_warn_threshold", 100) if cfg else 100 + if total > threshold: + result = await asyncio.wait_for( + ope.get_all_notes(pid, include=["metadatas"]), + timeout=_CONTEXT_DB_TIMEOUT, + ) + low_count = sum( + 1 + for m in (result.get("metadatas") or []) + if isinstance(m, dict) and m.get("importance") == "low" + ) + memory_hint = ( + f"记忆库已有 {total} 条（低重要性 {low_count} 条），" + "建议本轮优先用 subconscious_duplicate_helper 整理合并。" + ) + except Exception as e: + logger.warning(f"[Subconscious] get_context memory stats failed: {e}") + stale.append("memory_stats") + + plan_info = None + try: + if runner is not None and getattr(runner, "_planner", None) is not None: + plan_info = runner._planner.last_plan_dict() + except Exception: + stale.append("idle_plan") + + _state.mark_context_fetched() + return json.dumps( + { + "status": "success", + "now": local.strftime("%Y-%m-%d %H:%M:%S"), + "timezone": tz_name, + "timezone_fallback": tz_fell_back, + "weekday": local.strftime("%A"), + "last_run_at": last_run_iso, + "minutes_since_last_run": gap_minutes, + "last_run_summary": _state.get_last_iter_summary() or None, + "pending_knowledge_suggestions": { + "count": len(suggested), + "ids": [s.get("kid") or s.get("title") or "?" for s in suggested][:20], + }, + "memory_total": total, + "memory_low_importance": low_count, + "memory_warn_threshold": threshold, + "memory_hint": memory_hint, + "idle_plan": plan_info, + "stale": stale, + }, + ensure_ascii=False, + ) + + @on_tools(GET_MEMORY_STATS_SCHEMA, strict=True, bound_to=_SUBCONSCIOUS_TOOLS) async def subconscious_get_memory_stats(data: dict[str, Any]) -> str: """返回记忆库统计概览。""" @@ -807,6 +916,32 @@ async def knowledge_search_authorized(ctx: ToolContext) -> str | None: return await subconscious_knowledge_search(ctx.data) +# 会话侧只读入口：用户画像，仅画像所有者（DMN 目标用户）可读 + + +def _deny_profile() -> str: + return _tools_err("无权限读取该画像（仅离线巩固循环的目标用户可读自己的画像）") + + +@on_tools(READ_PROFILE_SCHEMA, strict=True, custom_run=True) +async def read_profile_authorized(ctx: ToolContext) -> str | None: + """对话 LLM 读取用户画像，仅限 DMN 目标用户本人。""" + if not await can_read_profile(get_event_from_tool_ctx(ctx)): + return _deny_profile() + try: + profile = read_profile( + PROFILE_PATH, + start_line=ctx.data.get("start_line"), + end_line=ctx.data.get("end_line"), + ) + return _tools_ok(**profile) + except Exception as e: + logger.opt(exception=e, colors=True, raw=True).exception( + f"read_profile error: {e}" + ) + return _tools_err(str(e)) + + # 知识建议：对话 LLM -> 潜意识 Agent 审查管线 diff --git a/amrita_plugin_memory/rethinking/types.py b/amrita_plugin_memory/rethinking/types.py index 9b924ba..c086635 100644 --- a/amrita_plugin_memory/rethinking/types.py +++ b/amrita_plugin_memory/rethinking/types.py @@ -67,13 +67,3 @@ class SessionSummary(TypedDict): created_at: float summary: str message_count: int - - -class ProfileResult(TypedDict): - """用户画像读取返回值。""" - - summary: str - content: str - total_lines: int - range_start: int - range_end: int diff --git a/pyproject.toml b/pyproject.toml index d9c7888..c53420a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "amrita_plugin_memory" description = "Amrita长期记忆插件" -version = "0.7.1" +version = "0.8.0" dependencies = [ "aiohttp>=3.13.3", "aiologic>=0.16.0", @@ -37,7 +37,6 @@ module_name = "nonebot.adapters.onebot.v11" [tool.uv] dev-dependencies = [ "ruff>=0.12.8", - #"pytest>=8.4.1", # `ambot memory` 依赖 0.2.1 起的 entry point extras（[full_load]）支持 "ambot-inlinectl>=0.2.1", ]