diff --git a/amrita_plugin_memory/rethinking/heatmap.py b/amrita_plugin_memory/rethinking/heatmap.py index f398aee..9e296d6 100644 --- a/amrita_plugin_memory/rethinking/heatmap.py +++ b/amrita_plugin_memory/rethinking/heatmap.py @@ -278,11 +278,12 @@ def render_heatmap_png( counts = state.raw_fine if has_raw else state.fine observed = round(sum(state.raw_coarse) if has_raw else sum(state.coarse)) tier_text = plan.tier if plan is not None else "-" + approx = _t("（含迁移估算）", " (incl. migrated)") subtitle = _t( - f"观测 {observed} 条 · 活跃 {len(state.seen_days)} 天 · " - f"惩罚 #{penalty} · 分层 {tier_text}", - f"observations {observed} · active {len(state.seen_days)}d · " - f"penalty #{penalty} · tier {tier_text}", + f"观测 {observed} 条{'（含迁移估算）' if planner.counts_migrated else ''} · " + f"活跃 {len(state.seen_days)} 天 · 惩罚 #{penalty} · 分层 {tier_text}", + f"observations {observed}{approx if planner.counts_migrated else ''} · " + f"active {len(state.seen_days)}d · penalty #{penalty} · tier {tier_text}", ) draw.text((PAD, _px(72)), subtitle, font=f_sub, fill=MUTED) @@ -291,8 +292,10 @@ def render_heatmap_png( dt_days = max(0.0, (now_local.timestamp() - state.last_update) / 86400) decay = 2 ** (-dt_days / planner.tau_fine) if dt_days else 1.0 weights = [max(0.0, v) * decay for v in state.fine] - # 分母取累计峰值：整张图会随时间一起变淡，而不是每格重新拉满到最亮 - peak = max(1.0, max(counts)) if counts else 1.0 + # 分母用累计峰值（不衰减）：整张图会随时间变淡，计数为 0 时才退回当前权重 + peak = max(1.0, max(counts)) if max(counts, default=0.0) > 1e-9 else 1.0 + if max(counts, default=0.0) <= 1e-9 and weights: + peak = max(1.0, max(weights)) for wd in range(DAYS): cx = GRID_LEFT + (wd + 0.5) * CELL_W diff --git a/amrita_plugin_memory/rethinking/runner.py b/amrita_plugin_memory/rethinking/runner.py index 2abff12..668b07b 100644 --- a/amrita_plugin_memory/rethinking/runner.py +++ b/amrita_plugin_memory/rethinking/runner.py @@ -158,8 +158,7 @@ async def cancel_and_reschedule(self, chat_ts: float | None = None) -> None: 然后按新的冷段门槛重新选窗口。不 dump/恢复上下文—— 用户聊完天后有新内容，旧推理上下文应直接丢弃。 """ - # 先记下历史是否存在：首条消息会把热度图从空变成非空， - # 若观测后再判断，冷启动就会被自己这条消息绕过 + # 先记下历史是否存在，否则冷启动会被自己这条消息绕过 had_history = True with contextlib.suppress(Exception): had_history = self._planner.has_history() @@ -173,8 +172,7 @@ async def cancel_and_reschedule(self, chat_ts: float | None = None) -> None: 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 @@ -463,9 +461,7 @@ async def _post_process(self, chat_obj: ChatObject) -> None: 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: @@ -611,9 +607,7 @@ async def _load_state(self) -> None: self._fail_streak = int(payload.get("fail_streak", 0) or 0) except (TypeError, ValueError): self._fail_streak = 0 - # 旧 penalty 语义是"被推迟的次数"，与新语义（要求的冷段长度）不可换算， - # 升级后清零一次，避免带着旧值把门槛顶到上限。 - # 仍走旧调度时数值是自洽的，别动它。 + # 旧 penalty 语义不可换算，新调度下清零一次；仍走旧调度时别动 if ( self._config.use_idle_window_planner and "idle_planner" not in payload @@ -698,12 +692,10 @@ async def _save_state_locked(self) -> None: "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: @@ -740,8 +732,7 @@ async def _load_prompt(self) -> str: character_prompt=character_prompt, ) - # 指南文件仍走 Jinja2，但只提供不变量：旧文件里的 {{ current_time }} - # 会被渲染成空串，而不是原样泄漏到 system 里 + # 指南文件仍走 Jinja2，但只提供不变量，旧模板变量会渲染成空串 static_vars = { "current_date": current_date, "target_user_id": self._config.target_user_id or "(未配置)", diff --git a/amrita_plugin_memory/rethinking/schedule.py b/amrita_plugin_memory/rethinking/schedule.py index 15a958c..f40dda9 100644 --- a/amrita_plugin_memory/rethinking/schedule.py +++ b/amrita_plugin_memory/rethinking/schedule.py @@ -4,7 +4,8 @@ - 观测：用户每条消息让对应 (星期几, 2h 槽) 权重 +1，并随时间衰减。 - 分层：fine(单片星期几) / mid(工作日-周末) / coarse(全局)，逐级降级。 - 收缩：fine/mid 向 coarse 按有效样本天数加权，避免稀薄样本失真。 -- 规划：环形找连续冷槽 -> 段内用原始时间戳做分钟级二分 -> 扣 buffer 得可用窗口。 +- 规划：候选日覆盖今天起的一周 -> 逐天找连续冷槽 -> 段内用原始时间戳做分钟级二分 + -> 扣 buffer 得可用窗口。 - 保底：冷段不足门槛时逐次减半放宽，最后强制取最长冷段。 本模块不依赖 nonebot，纯数据 + 纯函数，可脱离运行时单独验证。 @@ -215,6 +216,8 @@ class PlannerState: #: 热度图对应的时区名与目标用户，任一变化都要重建，否则星期几/钟点对不上 tz_name: str = "" uid: str = "" + #: 累计计数里有从衰减权重补出来的格子，展示时要注明是估算 + migrated_counts: bool = False def to_dict(self) -> dict[str, Any]: return { @@ -227,6 +230,7 @@ def to_dict(self) -> dict[str, Any]: "last_update": self.last_update, "tz_name": self.tz_name, "uid": self.uid, + "migrated_counts": self.migrated_counts, } @@ -289,6 +293,16 @@ def load(self, payload: dict[str, Any]) -> None: self.state.raw_fine = raw_fine if len(raw_coarse) == SLOTS_PER_DAY: self.state.raw_coarse = raw_coarse + # 升级前的快照没有累计计数：逐格用衰减值补位，只补空的格子 + self.state.migrated_counts = bool(raw.get("migrated_counts", False)) + for idx, value in enumerate(self.state.fine): + if self.state.raw_fine[idx] <= 0 and value > 0: + self.state.raw_fine[idx] = value + self.state.migrated_counts = True + for idx, value in enumerate(self.state.coarse): + if self.state.raw_coarse[idx] <= 0 and value > 0: + self.state.raw_coarse[idx] = value + self.state.migrated_counts = True seen = raw.get("seen_days", []) if isinstance(seen, list): self.state.seen_days = [str(d) for d in seen][-_MAX_SEEN_DAYS:] @@ -372,6 +386,11 @@ def observe(self, ts: float | None = None) -> None: if len(self.state.recent) > _MAX_RECENT: self.state.recent = self.state.recent[-_MAX_RECENT:] + @property + def counts_migrated(self) -> bool: + """累计计数里是否有从衰减权重补出来的格子。""" + return self.state.migrated_counts + def last_plan(self) -> Plan | None: return self._last_plan @@ -542,9 +561,8 @@ def _seg_bounds( def _refine( self, - start_slot: int, - length_slots: int, - wd: int, + seg_start: datetime, + seg_end: datetime, now: datetime, tier: str, penalty: int, @@ -556,7 +574,6 @@ def _refine( 段内连一次运行都放不下时返回 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) @@ -573,8 +590,7 @@ def _refine( 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 ) @@ -659,80 +675,106 @@ def _want_slots(self, penalty: int) -> int: 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 _candidates( + self, days: Sequence[datetime], tier: str, need_slots: int, now: datetime + ) -> list[tuple[datetime, datetime, int]]: + """把未来一周里每一天的冷段换成绝对时间，丢掉已经结束的。 + + 只看今天会把整天空窗漏掉：今天的表里冷段早被最近几天的活动挤没了， + 选出来只剩"下周五凌晨"这种同星期几的时段，而周一到周三明明整天空着。 + """ + now_local = now.astimezone(self.tz) + out: list[tuple[datetime, datetime, int]] = [] + for day in days: + wd = day.weekday() + if tier not in self._tiers_for(wd): + continue + for start_slot, length_slots, _ in self._cold_runs(tier, wd): + if length_slots < need_slots: + continue + seg_start = day + timedelta(minutes=start_slot * SLOT_MINUTES) + seg_end = seg_start + timedelta(minutes=length_slots * SLOT_MINUTES) + if seg_end <= now_local: + continue + out.append((seg_start, seg_end, length_slots)) + return out + + def _ordered( + self, cands: Sequence[tuple[datetime, datetime, int]], now: datetime + ) -> list[tuple[datetime, datetime, int]]: + """按距离排序，同样近就取更长的；最多取前几段，避免逐段试太久。""" + now_local = now.astimezone(self.tz) + ranked = sorted( + cands, + key=lambda c: ( + max(0.0, (c[0] - now_local).total_seconds()), + -(c[1] - c[0]), + ), + ) + return ranked[:8] 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) + # 候选日覆盖今天起的一周：周一到周三这种整天空窗，只看今天永远选不到 + days = [ + local.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=i) + for i in range(DAYS) + ] + tiers = ("fine", "mid", "coarse") - 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 + # 最近的一段可能剩余空间不够，接着试同层后面的候选 + for picked in self._ordered( + self._candidates(days, tier, need_slots, now), now + ): + plan = self._refine( + picked[0], + picked[1], + now, + tier, + penalty, + step, + forced=False, + ) + if plan is not None: + return plan + + # 全部放宽仍不足：取一周内最长的那段冷段 + for tier in tiers: + cands = self._candidates(days, tier, 1, now) + if not cands: + continue + for picked in sorted(cands, key=lambda c: -(c[1] - c[0]))[:8]: plan = self._refine( picked[0], picked[1], - picked[2], now, tier, penalty, - step, - forced=False, + self.max_relax, + forced=True, ) 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: + wd = local.weekday() + for tier in self._tiers_for(wd): floor_run = self._lightest_run(tier, wd, 1, now) if floor_run is None: continue + seg_start, seg_end = self._seg_bounds(floor_run[0], floor_run[1], wd, now) plan = self._refine( - floor_run[0], - floor_run[1], - wd, + seg_start, + seg_end, now, tier, penalty, diff --git a/pyproject.toml b/pyproject.toml index 3fccc0b..2d22754 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "amrita_plugin_memory" description = "Amrita长期记忆插件" -version = "0.9.0" +version = "0.9.1" dependencies = [ "aiohttp>=3.13.3", "aiologic>=0.16.0", diff --git a/uv.lock b/uv.lock index 30b5a84..29bb640 100644 --- a/uv.lock +++ b/uv.lock @@ -277,7 +277,7 @@ wheels = [ [[package]] name = "amrita-plugin-memory" -version = "0.9.0" +version = "0.9.1" source = { editable = "." } dependencies = [ { name = "aiohttp" },