diff --git a/README.md b/README.md index 7a3af9b..b3f3229 100644 --- a/README.md +++ b/README.md @@ -139,6 +139,10 @@ L1 临近上限时把细节下放到 L2，只留指针。 整理职责：去重合并、重要性校准与标签统一、画像精炼、知识沉淀、临近上限时压缩 L1。 +调度可靠性：待执行任务的触发时刻随状态一起落盘，重启后会补排（停机期间错过的窗口 +一分钟内补跑）；任务的 `misfire_grace_time` 置空，事件循环被卡住也不会把这次运行 +静默丢弃；单轮失败按 30 分钟间隔最多重试 3 次，用户再发消息会重置重试计数。 + ## 嵌入模型与重映射 向量只在同一模型下可比，换模型会让旧向量静默劣化且不报错。插件在集合 metadata 里记 @@ -223,6 +227,7 @@ L1 临近上限时把细节下放到 L2，只留指针。 /memory group delete 删除群共享记忆（仅管理员 / 群主） /memory memo 以合并转发查看当前作用域备忘录 /memory dmn 立即触发一轮巩固循环 +/memory heatmap 渲染作息热度图（含下次运行时间） ``` `memo` 只读：查看谁都有权限，写入走工具。群里 `/memory memo` 看的是群备忘录， @@ -232,6 +237,10 @@ L1 临近上限时把细节下放到 L2，只留指针。 按未知选项处理。鉴权顺序为超级用户白名单，其次 `lp.admin` 权限节点。 还需要 `subconscious.enabled = true` 且已配好 `target_user_id`。 +`heatmap`（别名 `heat`、`热图`、`热度图`）同样隐藏且鉴权一致：发一张 7×12 的作息 +热度图，格内数字是原始观测条数，颜色按半衰期衰减后的权重取色，图上标出当前时槽 +与下次 DMN 运行时间（含窗口长度、分层与 penalty）。 + 兼容旧格式 `/memory list` 等，默认以个人范围执行。群聊里 `user` 与 `group` 两个范围都可用，私聊只有 `user`。 @@ -457,6 +466,7 @@ amrita_plugin_memory/ ├── runner.py # SubconsciousRunner — 核心编排器 ├── hooks.py # on_precompletion hook / 生命周期管理 ├── schedule.py # 空闲窗口规划（纯数据 + 纯函数，唯一时区解析入口） + ├── heatmap.py # 热度图渲染（Pillow，纯渲染不碰运行时） ├── tools.py # 21 个巩固循环工具 handler ├── schemas.py # ToolFunctionSchema 定义 └── knowledge.py # KnowledgeBaseManager — 三方同步知识库 diff --git a/amrita_plugin_memory/matchers.py b/amrita_plugin_memory/matchers.py index 75fe2a4..39de5ad 100644 --- a/amrita_plugin_memory/matchers.py +++ b/amrita_plugin_memory/matchers.py @@ -1,6 +1,7 @@ """NoneBot2 命令：/memory""" -from typing import cast +from datetime import datetime, tzinfo +from typing import Any, cast from amrita.plugins.chat.utils.functions import split_message_into_chats from amrita.plugins.menu.models import MatcherData @@ -66,6 +67,12 @@ async def _( await matcher.finish(f"未知选项：{sub}") await matcher.finish(await _trigger_dmn()) + if sub in ("heatmap", "heat", "热图", "热度图"): + if not await _is_admin(event): + await matcher.finish(f"未知选项：{sub}") + await _send_heatmap(matcher) + await matcher.finish() + ope = AsyncUserMemory(get_db_conn()) await ope.init() @@ -148,6 +155,67 @@ async def _trigger_dmn() -> str: return "已触发，潜意识将在后台跑一轮" +_WD_SHORT = ("周一", "周二", "周三", "周四", "周五", "周六", "周日") + + +async def _send_heatmap(matcher: Matcher) -> None: + """渲染 7×12 热度图并发送，附带下次运行时间。""" + from .rethinking import _state + from .rethinking.heatmap import render_heatmap_png + + cfg = (await DataManager().safe_get_config()).subconscious + if not cfg.enabled: + await matcher.finish("潜意识调度未启用：subconscious.enabled 为 false") + runner = _state.get_runner() + if runner is None: + await matcher.finish("潜意识调度未启动：钩子挂载失败或启动异常，检查日志") + + snap = runner.scheduler_snapshot() + planner = runner.planner + caption = _heatmap_caption(snap, planner.tz) + try: + png = render_heatmap_png( + planner, + next_run=snap["next_run"], + plan=snap["plan"], + penalty=int(snap["penalty"]), + ) + except Exception as e: + # 出图失败不能连文字状态一起吞掉，退回纯文本 + logger.opt(exception=e, colors=True, raw=True).error( + f"[Subconscious] Heatmap render failed: {e}" + ) + await matcher.finish(caption) + + await matcher.send(MessageSegment.image(png)) + await matcher.finish(caption) + + +def _heatmap_caption(snap: dict[str, Any], tz: tzinfo) -> str: + """图下那行状态：下次触发时刻、窗口长度与当前是否在跑。""" + next_run: datetime | None = snap["next_run"] + if next_run is None: + head = "下次运行：尚未排期，等用户下一条消息触发规划" + else: + local = next_run.astimezone(tz) + head = ( + f"下次运行：{local.strftime('%m-%d')} " + f"{_WD_SHORT[local.weekday()]} {local.strftime('%H:%M')}" + ) + + plan = snap["plan"] + if plan is None: + return f"{head}\n暂无规划记录，热度图只反映历史观测。" + end = plan.end.astimezone(tz) + state = "运行中" if snap["is_running"] else "待命" + return ( + f"{head}\n" + f"窗口 {plan.start.strftime('%H:%M')}-{end.strftime('%H:%M')} · " + f"{plan.length_minutes} 分钟 · tier={plan.tier} · " + f"penalty #{snap['penalty']} · {state}" + ) + + MEMO_SEGMENT_CHARS = 600 diff --git a/amrita_plugin_memory/rethinking/heatmap.py b/amrita_plugin_memory/rethinking/heatmap.py new file mode 100644 index 0000000..f398aee --- /dev/null +++ b/amrita_plugin_memory/rethinking/heatmap.py @@ -0,0 +1,438 @@ +"""热度图渲染：把 IdlePlanner 的 7×12 作息热度画成 PNG。 + +纯渲染模块，只依赖 Pillow 与 schedule 的数据结构，不碰 nonebot， +方便脱离运行时用临时脚本出图核对。 + +版式是竖版的：纵轴是时间（一天 12 个两小时槽），横轴是星期几， +和常见的作息表读法一致——从上往下看一天，从左往右看一周。 + +整张图按 SCALE 倍超采样绘制：小字号中文在 1x 下笔画会糊在一起， +把画布和字号一起放大再出图，边缘才立得住。 + +格内数字是落盘时点的累计计数，颜色按半衰期折算到当前时刻并按线性映射取色： +平方根压缩会把 1 条和 6 条拉成同一个亮度，冷热就分不出来了。 +规划窗口按槽取整高亮：框线必须压在槽边界上，精确起止交给下面板里的文字。 + +字体按候选路径查找，找不到中文字体时自动退回英文标签（PIL 自带位图字体 +没有 CJK 字形，硬画会出一片豆腐块）。 +""" + +from __future__ import annotations + +import io +import os +from datetime import datetime, timedelta, tzinfo +from functools import lru_cache + +from PIL import Image, ImageDraw, ImageFont + +from .schedule import DAYS, SLOT_MINUTES, SLOTS_PER_DAY, IdlePlanner, Plan + +#: 超采样倍率：1x 下 12px 中文基本看不清，整体放大后出图 +SCALE = 2 +_LOGICAL_W = 760 +_LOGICAL_H = 814 + +WIDTH = _LOGICAL_W * SCALE +HEIGHT = _LOGICAL_H * SCALE +PAD = 36 * SCALE +GRID_LEFT = 84 * SCALE +GRID_TOP = 132 * SCALE +CELL_W = (_LOGICAL_W - 84 - 36) / DAYS * SCALE +CELL_H = 44 * SCALE +GRID_BOTTOM = GRID_TOP + CELL_H * SLOTS_PER_DAY + +BG = (15, 20, 26) +PANEL = (22, 29, 38) +BORDER = (38, 51, 65) +TEXT = (214, 224, 236) +MUTED = (138, 160, 184) +ACCENT = (141, 180, 216) +HOT_MARK = (242, 209, 139) +NOW_MARK = (224, 122, 95) + +#: 冷 -> 热的取色节点，最后一档偏暖，让热点一眼可见 +_RAMP = ( + (0.0, (26, 34, 44)), + (0.3, (32, 62, 90)), + (0.55, (58, 110, 150)), + (0.8, (141, 180, 216)), + (1.0, (232, 176, 106)), +) + +_FONT_CANDIDATES = ( + "/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc", + "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/opentype/noto/NotoSansSC-Regular.otf", + "/usr/share/fonts/truetype/droid/DroidSansFallbackFull.ttf", + "/System/Library/Fonts/PingFang.ttc", + "C:/Windows/Fonts/msyh.ttc", +) + +_WD_ZH = ("周一", "周二", "周三", "周四", "周五", "周六", "周日") +_WD_EN = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun") + + +def _px(v: float) -> float: + """逻辑像素换算成实际画布像素。""" + return v * SCALE + + +@lru_cache(maxsize=1) +def _font_path() -> str | None: + """按环境变量、候选路径顺序找第一个真正能加载的字体文件。 + + 只看文件存在不够：指到损坏或非字体文件时 Pillow 会退回位图字体， + 而那套字形没有中文，标签会变成一片方框。这里先试加载一次。 + """ + env = os.environ.get("AMRITA_MEMORY_HEATMAP_FONT", "").strip() + candidates = [env] if env else [] + candidates += list(_FONT_CANDIDATES) + for path in candidates: + if not os.path.exists(path): + continue + try: + ImageFont.truetype(path, 12) + except OSError: + continue + return path + return None + + +@lru_cache(maxsize=16) +def _font(size: int) -> ImageFont.ImageFont | ImageFont.FreeTypeFont: + """按像素大小取字体，优先中文字体，加载不了就用 Pillow 自带位图字体。""" + path = _font_path() + if path: + try: + return ImageFont.truetype(path, size) + except OSError: + pass + try: + return ImageFont.load_default(size=size) + except TypeError: + return ImageFont.load_default() + + +def _t(zh: str, en: str) -> str: + """有中文字体时用中文，否则退回英文，避免整张图变成豆腐块。""" + return zh if _font_path() else en + + +def _ramp(v: float) -> tuple[int, int, int]: + """把 0..1 的热度映射成取色节点之间的插值颜色。""" + v = max(0.0, min(1.0, v)) + for i in range(len(_RAMP) - 1): + lo, c_lo = _RAMP[i] + hi, c_hi = _RAMP[i + 1] + if v <= hi: + span = hi - lo + t = 0.0 if span <= 0 else (v - lo) / span + return ( + int(c_lo[0] + (c_hi[0] - c_lo[0]) * t), + int(c_lo[1] + (c_hi[1] - c_lo[1]) * t), + int(c_lo[2] + (c_hi[2] - c_lo[2]) * t), + ) + return _RAMP[-1][1] + + +def _text_size( + draw: ImageDraw.ImageDraw, + text: str, + font: ImageFont.ImageFont | ImageFont.FreeTypeFont, +) -> tuple[int, int]: + """量出文本的像素宽高，用于居中与右对齐（不落地绘制）。""" + left, top, right, bottom = draw.textbbox((0, 0), text, font=font) + return int(right - left), int(bottom - top) + + +def _center( + draw: ImageDraw.ImageDraw, + cx: float, + cy: float, + text: str, + font: ImageFont.ImageFont | ImageFont.FreeTypeFont, + fill: tuple[int, int, int], +) -> None: + """以 (cx, cy) 为中心画一行文本。""" + w, h = _text_size(draw, text, font) + draw.text((cx - w / 2, cy - h / 2), text, font=font, fill=fill) + + +def _right( + draw: ImageDraw.ImageDraw, + x: float, + cy: float, + text: str, + font: ImageFont.ImageFont | ImageFont.FreeTypeFont, + fill: tuple[int, int, int], +) -> None: + """右端对齐到 x、垂直居中于 cy 画一行文本。""" + w, h = _text_size(draw, text, font) + draw.text((x - w, cy - h / 2), text, font=font, fill=fill) + + +def _humanize(seconds: float) -> str: + """把剩余秒数说成"3 小时 12 分后"这种可读文本，已过期则明说。""" + if seconds <= 0: + return _t("已到时间", "due now") + minutes = int(seconds // 60) + hours, minutes = divmod(minutes, 60) + if _font_path(): + if hours and minutes: + return f"{hours} 小时 {minutes} 分后" + if hours: + return f"{hours} 小时后" + return f"{minutes} 分后" + if hours: + return f"in {hours}h{minutes:02d}m" + return f"in {minutes}m" + + +def _y_of(moment: datetime) -> float: + """把一天里的时刻换算成网格纵坐标（0:00 在顶，24:00 在底）。""" + minutes = moment.hour * 60 + moment.minute + moment.second / 60 + return GRID_TOP + minutes / SLOT_MINUTES * CELL_H + + +def _cell_box(wd: int, slot: int) -> tuple[float, float, float, float]: + """一个槽的绘制矩形（左, 上, 右, 下），与格子本身的内缩一致。""" + x0 = GRID_LEFT + wd * CELL_W + _px(3) + x1 = GRID_LEFT + (wd + 1) * CELL_W - _px(3) + y0 = GRID_TOP + slot * CELL_H + _px(3) + y1 = GRID_TOP + (slot + 1) * CELL_H - _px(3) + return x0, y0, x1, y1 + + +def _window_slots(plan: Plan, tz: tzinfo) -> list[tuple[int, int, int]]: + """把计划窗口换成 (星期几, 起始槽, 结束槽)，按槽取整。 + + 高亮是给"哪几个槽被占用"看的，框线必须落在槽边界上； + 精确到分钟的起止写在图下面板里，两者不要混着表达。 + """ + start = plan.start.astimezone(tz) + end = plan.end.astimezone(tz) + out: list[tuple[int, int, int]] = [] + day = start.replace(hour=0, minute=0, second=0, microsecond=0) + for _ in range(DAYS + 1): + if day > end: + break + seg_start = max(start, day) + seg_end = min(end, day + timedelta(days=1)) + if seg_end > seg_start: + first = (seg_start.hour * 60 + seg_start.minute) // SLOT_MINUTES + if seg_end == day + timedelta(days=1): + last = SLOTS_PER_DAY - 1 + else: + minutes = seg_end.hour * 60 + seg_end.minute + last = (minutes + SLOT_MINUTES - 1) // SLOT_MINUTES - 1 + first = min(max(first, 0), SLOTS_PER_DAY - 1) + last = min(max(last, first), SLOTS_PER_DAY - 1) + out.append((day.weekday(), first, last)) + day += timedelta(days=1) + return out + + +def render_heatmap_png( + planner: IdlePlanner, + *, + next_run: datetime | None = None, + plan: Plan | None = None, + penalty: int = 0, + now: datetime | None = None, +) -> bytes: + """渲染热度图，返回 PNG 字节。 + + next_run 是调度器里待执行任务的触发时间（可能为空：没有待执行任务）， + plan 是上一次规划的窗口，用于在网格上叠加预计占用的时间区间。 + """ + state = planner.state + tz = planner.tz + now_local = (now or datetime.now(tz)).astimezone(tz) + if next_run is not None and next_run.tzinfo is None: + # 调度器给的是带时区的时刻；裸时间按本地时区解释，别猜成 UTC + next_run = next_run.replace(tzinfo=tz) + + img = Image.new("RGB", (WIDTH, HEIGHT), BG) + draw = ImageDraw.Draw(img) + + f_title = _font(_px(26)) + f_sub = _font(_px(14)) + f_cell = _font(_px(13)) + f_axis = _font(_px(13)) + f_label = _font(_px(13)) + f_value = _font(_px(22)) + f_small = _font(_px(13)) + + draw.text( + (PAD, _px(26)), + _t("DMN 调度热力图", "DMN schedule heatmap"), + font=f_title, + fill=TEXT, + ) + _right(draw, WIDTH - PAD, _px(40), planner.tz_name, f_sub, MUTED) + + # 有累计计数就报累计值：热度会随时间变淡，但观测条数不该跟着缩水 + has_raw = sum(state.raw_coarse) > 0 + 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 "-" + 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}", + ) + draw.text((PAD, _px(72)), subtitle, font=f_sub, fill=MUTED) + + dt_days = 0.0 + if state.last_update: + 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 + + for wd in range(DAYS): + cx = GRID_LEFT + (wd + 0.5) * CELL_W + _center(draw, cx, GRID_TOP - _px(18), _t(_WD_ZH[wd], _WD_EN[wd]), f_axis, MUTED) + + for slot in range(SLOTS_PER_DAY): + cy = GRID_TOP + (slot + 0.5) * CELL_H + _right(draw, GRID_LEFT - _px(12), cy, f"{slot * 2:02d}", f_axis, MUTED) + + now_slot = (now_local.hour * 60 + now_local.minute) // SLOT_MINUTES + now_wd = now_local.weekday() + + for slot in range(SLOTS_PER_DAY): + for wd in range(DAYS): + idx = wd * SLOTS_PER_DAY + slot + weight = weights[idx] + ratio = weight / peak if peak > 1e-9 else 0.0 + x0, y0, x1, y1 = _cell_box(wd, slot) + draw.rounded_rectangle((x0, y0, x1, y1), radius=_px(6), fill=_ramp(ratio)) + raw = counts[idx] + if raw >= 0.5: + fill = (18, 24, 32) if ratio > 0.62 else TEXT + _center(draw, (x0 + x1) / 2, (y0 + y1) / 2, f"{raw:.0f}", f_cell, fill) + if wd == now_wd and slot == now_slot: + # 红点要整个落在格子里：贴到圆角上会看成跑到格子外面去了 + r = _px(3.5) + cx = x1 - _px(9) + cy = y1 - _px(9) + draw.ellipse((cx - r, cy - r, cx + r, cy + r), fill=NOW_MARK) + + if plan is not None: + # 半透明覆盖层单独画：直接填色会盖掉格内数字 + overlay = Image.new("RGBA", (WIDTH, HEIGHT), (0, 0, 0, 0)) + od = ImageDraw.Draw(overlay) + for wd, first, last in _window_slots(plan, tz): + # 逐格填色：整块铺一层会把槽之间的缝也染上，看着像没对齐格子 + for slot in range(first, last + 1): + od.rounded_rectangle( + _cell_box(wd, slot), + radius=_px(6), + fill=(242, 209, 139, 44), + ) + x0, y0, _, _ = _cell_box(wd, first) + _, _, x1, y1 = _cell_box(wd, last) + od.rounded_rectangle( + (x0, y0, x1, y1), + radius=_px(6), + outline=(242, 209, 139, 235), + width=int(_px(2)), + ) + img = Image.alpha_composite(img.convert("RGBA"), overlay).convert("RGB") + draw = ImageDraw.Draw(img) + + panel_top = GRID_BOTTOM + _px(22) + draw.rounded_rectangle( + (PAD, panel_top, WIDTH - PAD, panel_top + _px(84)), + radius=_px(10), + fill=PANEL, + outline=BORDER, + ) + draw.text( + (PAD + _px(18), panel_top + _px(12)), + _t("下次 DMN 运行", "Next DMN run"), + font=f_label, + fill=MUTED, + ) + + if next_run is not None: + local = next_run.astimezone(tz) + stamp = ( + f"{local.strftime('%Y-%m-%d')} " + f"{_t(_WD_ZH[local.weekday()], _WD_EN[local.weekday()])} " + f"{local.strftime('%H:%M')}" + ) + draw.text((PAD + _px(18), panel_top + _px(32)), stamp, font=f_value, fill=TEXT) + w, _ = _text_size(draw, stamp, f_value) + draw.text( + (PAD + _px(30) + w, panel_top + _px(40)), + # 用时间戳相减：按墙上时钟算，跨夏令时会差出一小时 + _humanize(local.timestamp() - now_local.timestamp()), + font=f_sub, + fill=ACCENT, + ) + else: + draw.text( + (PAD + _px(18), panel_top + _px(32)), + _t("尚未排期", "not scheduled"), + font=f_value, + fill=HOT_MARK, + ) + draw.text( + (PAD + _px(148), panel_top + _px(40)), + _t("等用户下一条消息触发规划", "waiting for next user message"), + font=f_sub, + fill=MUTED, + ) + + if plan is not None: + local_end = plan.end.astimezone(tz) + detail = _t( + f"窗口 {plan.start.strftime('%H:%M')}–{local_end.strftime('%H:%M')} · " + f"{plan.length_minutes} 分钟 · tier={plan.tier} · " + f"relax={plan.relaxed_steps} · forced={plan.forced}", + f"window {plan.start.strftime('%H:%M')}-{local_end.strftime('%H:%M')} · " + f"{plan.length_minutes}min · tier={plan.tier} · " + f"relax={plan.relaxed_steps} · forced={plan.forced}", + ) + if plan.note: + detail += f" · {plan.note}" + draw.text( + (PAD + _px(18), panel_top + _px(62)), detail, font=f_small, fill=MUTED + ) + + bar_y = panel_top + _px(104) + bar_w = _px(220) + bar_h = _px(10) + for i in range(int(bar_w)): + draw.line( + ((PAD + i, bar_y), (PAD + i, bar_y + bar_h)), + fill=_ramp(i / max(1, bar_w - 1)), + ) + draw.text((PAD, bar_y - _px(4)), _t("冷", "cold"), font=f_small, fill=MUTED) + draw.text( + (PAD + bar_w + _px(10), bar_y - _px(4)), + _t("热", "hot"), + font=f_small, + fill=MUTED, + ) + _right( + draw, + WIDTH - PAD, + bar_y + bar_h / 2, + _t( + "纵轴 2 小时一槽 · 红点=当前时段 · 颜色按半衰期折算后的权重", + "2h per row · dot = now · color = decayed weight", + ), + f_small, + MUTED, + ) + + buf = io.BytesIO() + img.save(buf, format="PNG", optimize=True) + return buf.getvalue() diff --git a/amrita_plugin_memory/rethinking/runner.py b/amrita_plugin_memory/rethinking/runner.py index 818469a..2abff12 100644 --- a/amrita_plugin_memory/rethinking/runner.py +++ b/amrita_plugin_memory/rethinking/runner.py @@ -50,6 +50,9 @@ _STATE_UID = "amrita_memory" #: 重排节流窗口（秒） _REPLAN_THROTTLE_SECONDS = 60.0 +#: 单轮失败后的重试间隔（分钟）与连续重试上限，超限就等用户下一条消息 +_RETRY_DELAY_MINUTES = 30 +_MAX_RUN_RETRIES = 3 _LEGACY_REPO_UID = ( "amrita_memory" # 旧版本元状态占用的 Bot Memory 表 uid（仅用于升级迁移） ) @@ -88,6 +91,14 @@ def __init__(self, config: SubconsciousConfig) -> None: # 重排节流：一分钟内重复触发只更新惩罚与热度，不重复规划 self._last_replan_at: float = 0.0 self._replan_throttle: float = _REPLAN_THROTTLE_SECONDS + # 待执行任务的触发时刻（unix 秒）：落盘后供重启恢复，0 表示无待执行任务 + self._pending_run_ts: float = 0.0 + # 这条待执行任务属于哪个目标用户：换了 target 就不该再跑旧窗口 + self._pending_uid: str = "" + # 连续失败次数，成功一轮即清零 + self._fail_streak: int = 0 + # 本轮周期是否失败过：失败要重试，别把异常当成跑完 + self._cycle_failed: bool = False # 状态写入串行化：并发 save 先后到达时，旧快照不能盖掉新快照 self._state_lock = asyncio.Lock() # 正在跑的周期任务，用户新消息到来时可以掐断 @@ -131,7 +142,8 @@ async def start(self) -> None: await self._kb_manager.validate_on_startup() else: logger.info("[Subconscious] Knowledge base disabled via config") - logger.info("[Subconscious] Idle — waiting for user chat to trigger first run") + # 关机前还没跑的窗口要补排，否则重启后 DMN 会一直等到用户开口 + await self.restore_schedule() def stop(self) -> None: with contextlib.suppress(Exception): @@ -153,6 +165,8 @@ async def cancel_and_reschedule(self, chat_ts: float | None = None) -> None: had_history = self._planner.has_history() self._planner.observe(chat_ts) penalty = _state.increment_penalty() + # 用户回来了，之前连续失败的计数不再代表当前状况 + self._fail_streak = 0 # 用户在聊天，正在跑的那一轮已经过时了，掐掉避免它把旧结论写回记忆 self._abort_running_cycle() @@ -215,6 +229,7 @@ def _defer_pending_job(self, penalty: int, had_history: bool) -> None: scheduler.reschedule_job( self._job_id, trigger="date", run_date=run_date ) + self._pending_run_ts = run_date.timestamp() except Exception as e: logger.warning(f"[Subconscious] Defer pending job failed: {e}") @@ -234,24 +249,108 @@ async def _save_state_logged(self) -> None: 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) + """排一次待执行任务，并把触发时刻记进状态，供重启后恢复。""" + run_date = datetime.now(timezone.utc) + timedelta( + seconds=max(1, int(delay_seconds)) + ) scheduler.add_job( self._run, trigger="date", run_date=run_date, id=self._job_id, - misfire_grace_time=30, + # 事件循环被同步 IO 或长任务卡住时，30 秒宽限会让这次运行被静默丢弃 + misfire_grace_time=None, + coalesce=True, + max_instances=1, replace_existing=True, ) + self._pending_run_ts = run_date.timestamp() + self._pending_uid = self.target_user_id + + async def _schedule_retry(self) -> None: + """本轮失败后补一次重试，并把重试状态落盘，重启后仍按同一节奏。""" + if scheduler.get_job(self._job_id) is not None: + return + if self._fail_streak >= _MAX_RUN_RETRIES: + logger.warning( + f"[Subconscious] {self._fail_streak} failed runs in a row, " + "stop retrying until next user chat" + ) + return + self._fail_streak += 1 + self._schedule_once(_RETRY_DELAY_MINUTES * 60) + # 不落盘的话，重启会丢掉这次重试，失败计数也会从零开始 + await self._save_state_logged() + logger.warning( + f"[Subconscious] Retry #{self._fail_streak} " + f"scheduled in {_RETRY_DELAY_MINUTES}min" + ) + + async def restore_schedule(self) -> None: + """重启后补排关机前还没跑的窗口。 + + 只恢复真的待执行过的任务：没有 pending 时间戳说明上一轮已经跑完、 + 正等用户下一条消息，此时补排会让 DMN 在没有用户活动时无限自转。 + """ + if scheduler.get_job(self._job_id) is not None: + return + pending = self._pending_run_ts + if self._pending_uid and self._pending_uid != self.target_user_id: + logger.info( + "[Subconscious] Drop pending run: target user changed " + f"({self._pending_uid} -> {self.target_user_id})" + ) + self._pending_run_ts = 0.0 + pending = 0.0 + if pending <= 0: + logger.info( + "[Subconscious] Idle — waiting for user chat to trigger first run" + ) + return + delay = int(pending - datetime.now(timezone.utc).timestamp()) + if delay < 60: + # 停机期间错过了窗口：尽快补跑，别拖到下一个冷段 + logger.info( + "[Subconscious] Pending window missed during downtime, run soon" + ) + delay = 60 + self._schedule_once(delay) + logger.info(f"[Subconscious] Restored pending run in {delay // 60}min") + + @property + def planner(self) -> IdlePlanner: + return self._planner + + def next_run_at(self) -> datetime | None: + """调度器里待执行任务的触发时间；没有待执行任务时返回 None。""" + job = scheduler.get_job(self._job_id) + if job is None: + return None + return job.next_run_time + + def scheduler_snapshot(self) -> dict[str, Any]: + """给状态展示用的一份只读快照。""" + return { + "next_run": self.next_run_at(), + "pending_run_ts": self._pending_run_ts, + "is_running": self._is_running, + "penalty": _state.get_penalty_count(), + "plan": self._planner.last_plan(), + "tz_name": self._planner.tz_name, + } async def trigger_now(self) -> None: """手动触发：取消待执行任务并立刻跑一轮，不阻塞调用方。""" with contextlib.suppress(Exception): scheduler.remove_job(self._job_id) + self._pending_run_ts = 0.0 + self._pending_uid = "" if self._manual_task is not None and not self._manual_task.done(): self._manual_task.cancel() # 持有引用，避免任务被 GC 回收 self._manual_task = asyncio.create_task(self._run()) + # 落盘清空：否则重启会把手动触发前那条旧计划又恢复出来 + await self._save_state_logged() async def _run(self) -> None: """调度入口 — 保证任何异常路径都会释放运行标志。""" @@ -261,13 +360,31 @@ async def _run(self) -> None: self._is_running = True self._cycle_task = asyncio.current_task() self._context_fetched = False + self._pending_run_ts = 0.0 + self._cycle_failed = False + ok = False + aborted = False try: # 窗口已实际占用：记一次观测，避免下一轮重复选同一段 with contextlib.suppress(Exception): self._planner.note_window_taken(time.time()) await self._run_cycle() + ok = not self._cycle_failed + if ok: + self._fail_streak = 0 + except asyncio.CancelledError: + # 用户发来新消息把本轮掐了，重排由 cancel_and_reschedule 负责 + aborted = True + raise + except Exception as e: + logger.opt(exception=e, colors=True, raw=True).error( + f"[Subconscious] Run failed: {e}" + ) finally: self._is_running = False + self._cycle_task = None + if not ok and not aborted: + await self._schedule_retry() async def _run_cycle(self) -> None: try: @@ -276,6 +393,7 @@ async def _run_cycle(self) -> None: logger.opt(exception=e, colors=True, raw=True).error( f"[Subconscious] No usable preset, skip cycle: {e}" ) + self._cycle_failed = True return prompt_text = await self._load_prompt() user_msg_text = ( @@ -311,6 +429,7 @@ async def _run_cycle(self) -> None: self._chat_obj = None raise except Exception as e: + self._cycle_failed = True logger.opt(exception=e, colors=True, raw=True).exception( f"[Subconscious] Error: {e}" ) @@ -360,6 +479,9 @@ async def _post_process(self, chat_obj: ChatObject) -> None: f"[Subconscious] Cycle done but save failed, " f"penalty NOT reset (penalty={_state.get_penalty_count()})" ) + elif self._cycle_failed: + # 跑挂了就别清惩罚：失败不该被当成一轮有效整理 + logger.warning("[Subconscious] Cycle failed, penalty kept") elif not _state.is_cycle_valid(): # 空转周期（没取上下文就收工）不清惩罚，保留压力等下一轮 logger.warning("[Subconscious] Cycle invalid, penalty kept") @@ -480,6 +602,15 @@ async def _load_state(self) -> None: _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) + try: + self._pending_run_ts = float(payload.get("pending_run_ts", 0.0) or 0.0) + except (TypeError, ValueError): + self._pending_run_ts = 0.0 + self._pending_uid = str(payload.get("pending_uid", "") or "") + try: + self._fail_streak = int(payload.get("fail_streak", 0) or 0) + except (TypeError, ValueError): + self._fail_streak = 0 # 旧 penalty 语义是"被推迟的次数"，与新语义（要求的冷段长度）不可换算， # 升级后清零一次，避免带着旧值把门槛顶到上限。 # 仍走旧调度时数值是自洽的，别动它。 @@ -560,6 +691,9 @@ async def _save_state_locked(self) -> None: "last_run_ts": _state.get_last_run_at(), "last_iter_summary": _state.get_last_iter_summary(), "penalty": _state.get_penalty_count(), + "pending_run_ts": self._pending_run_ts, + "pending_uid": self._pending_uid, + "fail_streak": self._fail_streak, "last_abstracts": self._last_abstracts, "pending_messages": _state.get_pending(), "knowledge_suggestions": _state.get_knowledge_suggestions(), diff --git a/amrita_plugin_memory/rethinking/schedule.py b/amrita_plugin_memory/rethinking/schedule.py index 504cc95..15a958c 100644 --- a/amrita_plugin_memory/rethinking/schedule.py +++ b/amrita_plugin_memory/rethinking/schedule.py @@ -206,6 +206,9 @@ class PlannerState: fine: list[float] = field(default_factory=lambda: [0.0] * CELLS) coarse: list[float] = field(default_factory=lambda: [0.0] * SLOTS_PER_DAY) + #: 累计观测条数，不参与衰减：热度会随半衰期变淡，但"来过多少条"不该缩水 + raw_fine: list[float] = field(default_factory=lambda: [0.0] * CELLS) + raw_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 @@ -217,6 +220,8 @@ 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], + "raw_fine": [round(v, 6) for v in self.raw_fine], + "raw_coarse": [round(v, 6) for v in self.raw_coarse], "seen_days": self.seen_days[-_MAX_SEEN_DAYS:], "recent": self.recent[-_MAX_RECENT:], "last_update": self.last_update, @@ -277,6 +282,13 @@ def load(self, payload: dict[str, Any]) -> None: self.state.fine = fine if len(coarse) == SLOTS_PER_DAY: self.state.coarse = coarse + # 老快照没有累计计数，留空即可，展示侧会退回衰减值 + raw_fine = [float(v) for v in raw.get("raw_fine", [])] + raw_coarse = [float(v) for v in raw.get("raw_coarse", [])] + if len(raw_fine) == CELLS: + self.state.raw_fine = raw_fine + if len(raw_coarse) == SLOTS_PER_DAY: + self.state.raw_coarse = raw_coarse seen = raw.get("seen_days", []) if isinstance(seen, list): self.state.seen_days = [str(d) for d in seen][-_MAX_SEEN_DAYS:] @@ -288,9 +300,44 @@ def load(self, payload: dict[str, Any]) -> None: self.state.uid = self.uid except (TypeError, ValueError): self.state = PlannerState() + return + self._load_last_plan(payload) def dump(self) -> dict[str, Any]: - return {"idle_planner": self.state.to_dict()} + """连同上次规划一起落盘：重启后图里还能看到待执行窗口的细节。""" + out: dict[str, Any] = {"idle_planner": self.state.to_dict()} + plan = self._last_plan + if plan is not None: + out["last_plan"] = { + "start": plan.start.isoformat(), + "end": plan.end.isoformat(), + "length_minutes": plan.length_minutes, + "tier": plan.tier, + "relaxed_steps": plan.relaxed_steps, + "forced": plan.forced, + "penalty": plan.penalty, + "note": plan.note, + } + return out + + def _load_last_plan(self, payload: dict[str, Any]) -> None: + """恢复上次规划；字段缺失或时间解析失败就当作没有规划。""" + raw = payload.get("last_plan") + if not isinstance(raw, dict): + return + try: + self._last_plan = Plan( + start=datetime.fromisoformat(str(raw["start"])), + end=datetime.fromisoformat(str(raw["end"])), + length_minutes=int(raw.get("length_minutes", 0)), + tier=str(raw.get("tier", "coarse")), + relaxed_steps=int(raw.get("relaxed_steps", 0)), + forced=bool(raw.get("forced", False)), + penalty=int(raw.get("penalty", 0)), + note=str(raw.get("note", "")), + ) + except (KeyError, TypeError, ValueError): + self._last_plan = None def _decay(self, now_ts: float) -> None: if self.state.last_update <= 0: @@ -312,6 +359,8 @@ def observe(self, ts: float | None = None) -> None: wd, slot = slot_of(now_ts, self.tz) self.state.fine[_cell(wd, slot)] += 1.0 self.state.coarse[slot] += 1.0 + self.state.raw_fine[_cell(wd, slot)] += 1.0 + self.state.raw_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) @@ -587,9 +636,11 @@ def has_history(self) -> bool: 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) + plan = self._fallback(now.astimezone(self.tz), penalty, note) + self._last_plan = plan + return plan def plan(self, penalty: int = 1, now: datetime | None = None) -> Plan: result = self._plan(penalty, now) diff --git a/pyproject.toml b/pyproject.toml index 609007b..3fccc0b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "amrita_plugin_memory" description = "Amrita长期记忆插件" -version = "0.8.1" +version = "0.9.0" dependencies = [ "aiohttp>=3.13.3", "aiologic>=0.16.0", @@ -13,6 +13,7 @@ dependencies = [ "nonebot-plugin-apscheduler>=0.5.0", "nonebot-plugin-localstore>=0.7.4", "nonebot-plugin-orm[default]>=0.8.2", + "pillow>=10.1", ] readme = "README.md" requires-python = ">=3.11, <3.14" diff --git a/uv.lock b/uv.lock index 642caeb..30b5a84 100644 --- a/uv.lock +++ b/uv.lock @@ -277,7 +277,7 @@ wheels = [ [[package]] name = "amrita-plugin-memory" -version = "0.8.1" +version = "0.9.0" source = { editable = "." } dependencies = [ { name = "aiohttp" }, @@ -290,6 +290,7 @@ dependencies = [ { name = "nonebot-plugin-apscheduler" }, { name = "nonebot-plugin-localstore" }, { name = "nonebot-plugin-orm", extra = ["default"] }, + { name = "pillow" }, ] [package.dev-dependencies] @@ -310,6 +311,7 @@ requires-dist = [ { name = "nonebot-plugin-apscheduler", specifier = ">=0.5.0" }, { name = "nonebot-plugin-localstore", specifier = ">=0.7.4" }, { name = "nonebot-plugin-orm", extras = ["default"], specifier = ">=0.8.2" }, + { name = "pillow", specifier = ">=10.1" }, ] [package.metadata.requires-dev]