feat: periodic llm-based conversation history compaction

astrbot/builtin_stars/builtin_commands/commands/__init__.py

@@ -2,6 +2,8 @@
 
 from .admin import AdminCommands
 from .alter_cmd import AlterCmdCommands
+from .context_compaction import ContextCompactionCommands
+from .context_memory import ContextMemoryCommands
 from .conversation import ConversationCommands
 from .help import HelpCommand
 from .llm import LLMCommands
@@ -17,6 +19,8 @@
     "AdminCommands",
     "AlterCmdCommands",
     "ConversationCommands",
+    "ContextCompactionCommands",
+    "ContextMemoryCommands",
     "HelpCommand",
     "LLMCommands",
     "PersonaCommands",

astrbot/builtin_stars/builtin_commands/commands/context_compaction.py

@@ -0,0 +1,110 @@
+from astrbot.api import star
+from astrbot.api.event import AstrMessageEvent, MessageChain
+from astrbot.core import logger
+from astrbot.core.context_compaction_scheduler import PeriodicContextCompactionScheduler
+
+
+class ContextCompactionCommands:
+    def __init__(self, context: star.Context) -> None:
+        self.context = context
+
+    def _get_scheduler(self) -> PeriodicContextCompactionScheduler | None:
+        scheduler = getattr(self.context, "context_compaction_scheduler", None)
+        if isinstance(scheduler, PeriodicContextCompactionScheduler):
+            return scheduler
+        return None
+
+    async def status(self, event: AstrMessageEvent) -> None:
+        scheduler = self._get_scheduler()
+        if not scheduler:
+            await event.send(
+                MessageChain().message("定时上下文压缩调度器不可用。"),
+            )
+            return
+
+        status = scheduler.get_status()
+        cfg = status.get("config", {})
+        last = status.get("last_report") or {}
+        trigger_tokens = cfg.get("trigger_tokens", "?")
+        trigger_ratio = cfg.get("trigger_min_context_ratio", "?")
+        if isinstance(trigger_tokens, int) and trigger_tokens <= 0:
+            if isinstance(trigger_ratio, (int, float)):
+                trigger_text = f"自动({trigger_ratio}x模型上下文或目标长度估算)"
+            else:
+                trigger_text = "自动(基于目标长度估算)"
+        else:
+            trigger_text = str(trigger_tokens)
+
+        lines = ["定时上下文压缩状态："]
+        lines.append(
+            f"启用={self._yes_no(bool(cfg.get('enabled', False)))}"
+            f" | 运行中={self._yes_no(bool(status.get('running', False)))}"
+            f" | 停止请求={self._yes_no(bool(status.get('stop_requested', False)))}"
+        )
+        lines.append(
+            f"间隔={cfg.get('interval_minutes', '?')}分钟"
+            f" | 每轮最多压缩={cfg.get('max_conversations_per_run', '?')}"
+            f" | 每轮最多扫描={cfg.get('max_scan_per_run', '?')}"
+        )
+        lines.append(
+            f"触发Token={trigger_text}"
+            f" | 目标Token={cfg.get('target_tokens', '?')}"
+            f" | 最大轮次={cfg.get('max_rounds', '?')}"
+        )
+
+        if last:
+            lines.append(
+                f"最近任务[{last.get('reason', 'unknown')}]"
+                f" scanned={last.get('scanned', 0)}"
+                f" compacted={last.get('compacted', 0)}"
+                f" skipped={last.get('skipped', 0)}"
+                f" failed={last.get('failed', 0)}"
+                f" elapsed={last.get('elapsed_sec', 0.0):.2f}s"
+            )
+        else:
+            lines.append("最近任务：暂无")
+
+        if status.get("last_started_at"):
+            lines.append(f"最近开始：{status.get('last_started_at')}")
+        if status.get("last_finished_at"):
+            lines.append(f"最近结束：{status.get('last_finished_at')}")
+        if status.get("last_error"):
+            lines.append(f"最近错误：{status.get('last_error')}")
+
+        await event.send(MessageChain().message("\n".join(lines)))
+
+    async def run(self, event: AstrMessageEvent, limit: int | None = None) -> None:
+        scheduler = self._get_scheduler()
+        if not scheduler:
+            await event.send(
+                MessageChain().message("定时上下文压缩调度器不可用。"),
+            )
+            return
+
+        if limit is not None and limit < 1:
+            await event.send(MessageChain().message("limit 必须 >= 1。"))
+            return
+
+        try:
+            report = await scheduler.run_once(
+                reason="manual_command",
+                max_conversations_override=limit,
+            )
+        except Exception as exc:
+            logger.error("ctxcompact run failed: %s", exc, exc_info=True)
+            await event.send(MessageChain().message("触发压缩失败，请查看服务端日志。"))
+            return
+
+        msg = (
+            "手动触发完成："
+            f"scanned={report.get('scanned', 0)} "
+            f"compacted={report.get('compacted', 0)} "
+            f"skipped={report.get('skipped', 0)} "
+            f"failed={report.get('failed', 0)} "
+            f"elapsed={report.get('elapsed_sec', 0.0):.2f}s"
+        )
+        await event.send(MessageChain().message(msg))
+
+    @staticmethod
+    def _yes_no(value: bool) -> str:
+        return "是" if value else "否"

astrbot/builtin_stars/builtin_commands/commands/context_memory.py

@@ -0,0 +1,204 @@
+from __future__ import annotations
+
+from typing import Any
+
+from astrbot.api import star
+from astrbot.api.event import AstrMessageEvent, MessageChain
+from astrbot.core.context_memory import ensure_context_memory_settings
+
+PINNED_PREVIEW_MAX_CHARS = 180
+
+
+class ContextMemoryCommands:
+    def __init__(self, context: star.Context) -> None:
+        self.context = context
+
+    def _get_provider_settings(self, event: AstrMessageEvent) -> tuple[Any, dict[str, Any]]:
+        cfg = self.context.get_config(umo=event.unified_msg_origin)
+        provider_settings = cfg.get("provider_settings", {})
+        if not isinstance(provider_settings, dict):
+            provider_settings = {}
+            cfg["provider_settings"] = provider_settings
+        return cfg, provider_settings
+
+    @staticmethod
+    def _save_config(cfg: Any) -> None:
+        save_func = getattr(cfg, "save_config", None)
+        if callable(save_func):
+            save_func()
+
+    @staticmethod
+    def _parse_switch(value: str) -> bool | None:
+        normalized = value.strip().lower()
+        if normalized in {"1", "true", "on", "yes", "enable", "enabled"}:
+            return True
+        if normalized in {"0", "false", "off", "no", "disable", "disabled"}:
+            return False
+        return None
+
+    async def status(self, event: AstrMessageEvent) -> None:
+        _, provider_settings = self._get_provider_settings(event)
+        cm_cfg = ensure_context_memory_settings(provider_settings)
+        pinned = cm_cfg.get("pinned_memories", [])
+        if not isinstance(pinned, list):
+            pinned = []
+
+        lines = ["上下文记忆状态："]
+        lines.append(
+            "启用="
+            + ("是" if bool(cm_cfg.get("enabled", False)) else "否")
+            + " | 注入顶层记忆="
+            + ("是" if bool(cm_cfg.get("inject_pinned_memory", True)) else "否")
+        )
+        lines.append(
+            f"顶层记忆条数={len(pinned)}"
+            f" | 最大条数={cm_cfg.get('pinned_max_items', '?')}"
+            f" | 单条最大字符={cm_cfg.get('pinned_max_chars_per_item', '?')}"
+        )
+        lines.append(
+            "检索增强(开发中)="
+            + ("是" if bool(cm_cfg.get("retrieval_enabled", False)) else "否")
+            + f" | backend={cm_cfg.get('retrieval_backend', '') or '-'}"
+            + f" | top_k={cm_cfg.get('retrieval_top_k', '?')}"
+        )
+        await event.send(MessageChain().message("\n".join(lines)))
+
+    async def ls(self, event: AstrMessageEvent) -> None:
+        _, provider_settings = self._get_provider_settings(event)
+        cm_cfg = ensure_context_memory_settings(provider_settings)
+        pinned = cm_cfg.get("pinned_memories", [])
+        if not isinstance(pinned, list) or not pinned:
+            await event.send(MessageChain().message("当前没有手动顶层记忆。"))
+            return
+
+        configured_max_chars = cm_cfg.get("pinned_max_chars_per_item", 400)
+        try:
+            configured_max_chars = int(configured_max_chars)
+        except Exception:
+            configured_max_chars = 400
+        preview_max_chars = min(
+            max(1, configured_max_chars),
+            PINNED_PREVIEW_MAX_CHARS,
+        )
+
+        lines = ["手动顶层记忆列表："]
+        for idx, text in enumerate(pinned, start=1):
+            text_str = str(text)
+            if len(text_str) > preview_max_chars:
+                text_str = text_str[:preview_max_chars] + "..."
+            lines.append(f"{idx}. {text_str}")
+        await event.send(MessageChain().message("\n".join(lines)))
+
+    async def add(self, event: AstrMessageEvent, text: str) -> None:
+        content = str(text or "").strip()
+        if not content:
+            await event.send(MessageChain().message("用法: /ctxmem add <记忆内容>"))
+            return
+
+        cfg, provider_settings = self._get_provider_settings(event)
+        cm_cfg = ensure_context_memory_settings(provider_settings)
+        pinned = cm_cfg.get("pinned_memories", [])
+        if not isinstance(pinned, list):
+            pinned = []
+            cm_cfg["pinned_memories"] = pinned
+
+        max_items = int(cm_cfg.get("pinned_max_items", 8) or 8)
+        if len(pinned) >= max_items:
+            await event.send(
+                MessageChain().message(
+                    f"已达到顶层记忆最大条数({max_items})，请先使用 /ctxmem rm <序号> 或 /ctxmem clear。",
+                )
+            )
+            return
+
+        max_chars = int(cm_cfg.get("pinned_max_chars_per_item", 400) or 400)
+        truncated = False
+        if len(content) > max_chars:
+            content = content[:max_chars]
+            truncated = True
+
+        pinned.append(content)
+        self._save_config(cfg)
+
+        msg = f"已添加顶层记忆 #{len(pinned)}。"
+        if truncated:
+            msg += f" 内容超过上限，已截断到 {max_chars} 字符。"
+        await event.send(MessageChain().message(msg))
+
+    async def rm(self, event: AstrMessageEvent, index: int) -> None:
+        cfg, provider_settings = self._get_provider_settings(event)
+        cm_cfg = ensure_context_memory_settings(provider_settings)
+        pinned = cm_cfg.get("pinned_memories", [])
+        if not isinstance(pinned, list) or not pinned:
+            await event.send(MessageChain().message("当前没有可删除的顶层记忆。"))
+            return
+
+        if index < 1 or index > len(pinned):
+            await event.send(
+                MessageChain().message(f"序号超出范围。请输入 1~{len(pinned)}。")
+            )
+            return
+
+        removed = str(pinned.pop(index - 1))
+        self._save_config(cfg)
+        preview = removed if len(removed) <= 80 else removed[:80] + "..."
+        await event.send(MessageChain().message(f"已删除顶层记忆 #{index}: {preview}"))
+
+    async def clear(self, event: AstrMessageEvent) -> None:
+        cfg, provider_settings = self._get_provider_settings(event)
+        cm_cfg = ensure_context_memory_settings(provider_settings)
+        pinned = cm_cfg.get("pinned_memories", [])
+        count = len(pinned) if isinstance(pinned, list) else 0
+        cm_cfg["pinned_memories"] = []
+        self._save_config(cfg)
+        await event.send(MessageChain().message(f"已清空顶层记忆，共 {count} 条。"))
+
+    async def enable(self, event: AstrMessageEvent, value: str = "") -> None:
+        cfg, provider_settings = self._get_provider_settings(event)
+        cm_cfg = ensure_context_memory_settings(provider_settings)
+        enabled = bool(cm_cfg.get("enabled", False))
+
+        value = str(value or "").strip()
+        if value:
+            parsed = self._parse_switch(value)
+            if parsed is None:
+                await event.send(
+                    MessageChain().message("参数错误。用法: /ctxmem enable [on|off]")
+                )
+                return
+            enabled = parsed
+        else:
+            enabled = not enabled
+
+        cm_cfg["enabled"] = enabled
+        self._save_config(cfg)
+        await event.send(
+            MessageChain().message(
+                "上下文记忆注入已" + ("开启。" if enabled else "关闭。")
+            )
+        )
+
+    async def retrieval(self, event: AstrMessageEvent, value: str = "") -> None:
+        cfg, provider_settings = self._get_provider_settings(event)
+        cm_cfg = ensure_context_memory_settings(provider_settings)
+        enabled = bool(cm_cfg.get("retrieval_enabled", False))
+
+        value = str(value or "").strip()
+        if value:
+            parsed = self._parse_switch(value)
+            if parsed is None:
+                await event.send(
+                    MessageChain().message("参数错误。用法: /ctxmem retrieval [on|off]")
+                )
+                return
+            enabled = parsed
+        else:
+            enabled = not enabled
+
+        cm_cfg["retrieval_enabled"] = enabled
+        self._save_config(cfg)
+        await event.send(
+            MessageChain().message(
+                "检索增强开关(开发中)已" + ("开启。" if enabled else "关闭。")
+            )
+        )