feat: manual context compression

astrbot/builtin_stars/builtin_commands/commands/conversation.py

@@ -1,16 +1,29 @@
+import json
+
 from sqlalchemy import case, func, select
 from sqlmodel import col
 
 from astrbot.api import sp, star
-from astrbot.api.event import AstrMessageEvent, MessageEventResult
+from astrbot.api.event import AstrMessageEvent, MessageChain, MessageEventResult
+from astrbot.api.message_components import Json
 from astrbot.core import logger
+from astrbot.core.agent.context.config import ContextConfig
+from astrbot.core.agent.context.manager import ContextManager
+from astrbot.core.agent.context.round_utils import split_into_rounds
+from astrbot.core.agent.message import (
+    bind_checkpoint_messages,
+    dump_messages_with_checkpoints,
+)
+from astrbot.core.agent.response import AgentStats
 from astrbot.core.agent.runners.deerflow.constants import (
     DEERFLOW_PROVIDER_TYPE,
     DEERFLOW_THREAD_ID_KEY,
 )
 from astrbot.core.agent.runners.deerflow.deerflow_api_client import DeerFlowAPIClient
+from astrbot.core.astr_main_agent import get_context_compression_provider
 from astrbot.core.db.po import ProviderStat
 from astrbot.core.utils.active_event_registry import active_event_registry
+from astrbot.core.utils.session_lock import session_lock_manager
 
 THIRD_PARTY_AGENT_RUNNER_KEY = {
     "dify": "dify_conversation_id",
@@ -166,6 +179,230 @@ async def stop(self, message: AstrMessageEvent) -> None:
             MessageEventResult().message("✅ No running tasks in the current session.")
         )
 
+    async def compact(self, message: AstrMessageEvent) -> None:
+        """Compress the persisted history of the current local conversation."""
+
+        def reply(text: str) -> None:
+            """Set a plain-text command result.
+
+            Args:
+                text: Message shown to the user.
+            """
+            message.set_result(message.plain_result(text))
+
+        preserved = "❌ Context compression failed; the original context was preserved."
+        cancelled = "⚠️ Compression cancelled; original context was preserved."
+        unknown = "⚠️ Context state is unknown. Check the conversation before retrying."
+        umo = message.unified_msg_origin
+        cfg = self.context.get_config(umo=umo)
+        provider_settings = cfg.get("provider_settings", {})
+        agent_runner = cfg.get("agent_runner", {})
+        conversation_manager = self.context.conversation_manager
+
+        is_unique_session = cfg.get("platform_settings", {}).get(
+            "unique_session",
+            False,
+        )
+        is_shared_group = bool(message.get_group_id()) and not is_unique_session
+        if is_shared_group and message.role != "admin":
+            reply(
+                "❌ Context compression requires admin permission in a shared "
+                "group conversation."
+            )
+            return
+
+        if not provider_settings.get("enable", True):
+            reply("❌ AI features are disabled for this session.")
+            return
+
+        if agent_runner.get("runner_type") != "local":
+            reply("❌ /compact is supported only by the local agent runner.")
+            return
+
+        runner_config = agent_runner.get("config", {})
+        compression_config = runner_config.get("compression", {})
+        if not compression_config.get("enable_manual_context_compression", False):
+            reply(
+                "❌ Manual context compression is disabled. Enable it in Context "
+                "Management first."
+            )
+            return
+
+        strategy = compression_config.get("overflow_strategy")
+        if strategy != "llm_compress":
+            reply("❌ /compact requires the LLM context compression strategy.")
+            return
+
+        initial_cid = await conversation_manager.get_curr_conversation_id(umo)
+        if not initial_cid:
+            reply("❌ You are not in a conversation. Use /new to create one.")
+            return
+
+        compression_provider = await get_context_compression_provider(
+            strategy,
+            compression_config.get("provider_id", ""),
+            self.context,
+            message,
+        )
+        if not compression_provider:
+            reply("❌ No LLM provider is available for context compression.")
+            return
+
+        progress_type = (
+            "webchat_ephemeral" if message.get_platform_name() == "webchat" else None
+        )
+        await message.send(
+            MessageChain(type=progress_type).message("⏳ Compressing context...")
+        )
+
+        try:
+            async with session_lock_manager.acquire_lock(umo):
+                if message.is_stopped():
+                    return
+                if message.get_extra("agent_stop_requested"):
+                    reply(cancelled)
+                    return
+
+                cid = await conversation_manager.get_curr_conversation_id(umo)
+                if not cid or cid != initial_cid:
+                    reply("⚠️ The active conversation changed; no changes were saved.")
+                    return
+
+                conversation = await conversation_manager.get_conversation(umo, cid)
+                if not conversation:
+                    reply(
+                        "❌ The current conversation could not be loaded; the "
+                        "original context was preserved."
+                    )
+                    return
+
+                original_history_text = conversation.history
+                original_history = json.loads(conversation.history)
+                if not isinstance(original_history, list) or not original_history:
+                    reply("ℹ️ There is not enough conversation history to compress.")
+                    return
+
+                messages = bind_checkpoint_messages(original_history)
+                complete_rounds = sum(
+                    any(segment.role == "user" for segment in round_)
+                    and any(segment.role == "assistant" for segment in round_)
+                    for round_ in split_into_rounds(messages)
+                )
+                if complete_rounds <= 1:
+                    reply("ℹ️ There is not enough conversation history to compress.")
+                    return
+
+                context_manager = ContextManager(
+                    ContextConfig(
+                        llm_compress_instruction=compression_config.get("instruction"),
+                        llm_compress_keep_recent_ratio=compression_config.get(
+                            "keep_recent_ratio",
+                            0.15,
+                        ),
+                        llm_compress_preserve_latest_round=True,
+                        llm_compress_provider=compression_provider,
+                    )
+                )
+                tokens_before = context_manager.token_counter.count_tokens(messages)
+                if tokens_before <= 0:
+                    reply("ℹ️ There is not enough conversation history to compress.")
+                    return
+
+                compressed_messages = await context_manager.process(
+                    messages,
+                    force_compress=True,
+                )
+                tokens_after = context_manager.token_counter.count_tokens(
+                    compressed_messages
+                )
+                if compressed_messages == messages or tokens_after >= tokens_before:
+                    reply(preserved)
+                    return
+
+                target_history = dump_messages_with_checkpoints(compressed_messages)
+                latest_cid = await conversation_manager.get_curr_conversation_id(umo)
+                latest_conversation = await conversation_manager.get_conversation(
+                    umo, cid
+                )
+                if (
+                    latest_cid != cid
+                    or not latest_conversation
+                    or latest_conversation.history != original_history_text
+                ):
+                    reply(
+                        "⚠️ Context changed during compression; no changes were saved."
+                    )
+                    return
+
+                if message.is_stopped():
+                    return
+                if message.get_extra("agent_stop_requested"):
+                    reply(cancelled)
+                    return
+
+                try:
+                    await conversation_manager.update_conversation(
+                        umo,
+                        cid,
+                        history=target_history,
+                        token_usage=0,
+                    )
+                except Exception as update_error:
+                    logger.error(
+                        "Context compression storage update failed: %s.",
+                        type(update_error).__name__,
+                    )
+                    try:
+                        stored_conversation = (
+                            await conversation_manager.get_conversation(umo, cid)
+                        )
+                        stored_history = json.loads(stored_conversation.history)
+                    except Exception as verify_error:
+                        logger.error(
+                            "Context compression storage verification failed: %s.",
+                            type(verify_error).__name__,
+                        )
+                        reply(unknown)
+                        return
+
+                    if stored_history != target_history:
+                        reply(
+                            preserved if stored_history == original_history else unknown
+                        )
+                        return
+        except Exception as error:
+            logger.error(
+                "Context compression failed before storage update: %s.",
+                type(error).__name__,
+            )
+            reply(preserved)
+            return
+
+        if message.is_stopped():
+            return
+
+        if message.get_platform_name() == "webchat":
+            try:
+                await message.send(
+                    MessageChain(
+                        type="agent_stats",
+                        chain=[
+                            Json(
+                                data=AgentStats(
+                                    current_context_tokens=tokens_after,
+                                ).to_dict()
+                            )
+                        ],
+                    )
+                )
+            except Exception as error:
+                logger.warning(
+                    "Failed to send context compression stats: %s.",
+                    type(error).__name__,
+                )
+
+        reply("✅ Context compressed.")
+
     async def new_conv(self, message: AstrMessageEvent) -> None:
         """Start a new conversation without clearing the previous history.
 

astrbot/builtin_stars/builtin_commands/main.py

@@ -63,6 +63,11 @@ async def stats(self, message: AstrMessageEvent) -> None:
         """Show token usage statistics for the current conversation"""
         await self.conversation_c.stats(message)
 
+    @filter.command("compact")
+    async def compact(self, message: AstrMessageEvent) -> None:
+        """Compress the current conversation context"""
+        await self.conversation_c.compact(message)
+
     @filter.permission_type(filter.PermissionType.ADMIN)
     @filter.command("provider")
     async def provider(

astrbot/core/agent/context/compressor.py

@@ -130,6 +130,7 @@ def __init__(
         instruction_text: str | None = None,
         compression_threshold: float = 0.82,
         token_counter: TokenCounter | None = None,
+        preserve_latest_round: bool = False,
     ) -> None:
         """Initialize the LLM summary compressor.
 
@@ -139,11 +140,15 @@ def __init__(
                 exact context. Clamped to 0-0.3.
             instruction_text: Custom instruction for summary generation.
             compression_threshold: The compression trigger threshold (default: 0.82).
+            token_counter: Token counter used to divide old and recent context.
+            preserve_latest_round: Whether to preserve the latest complete
+                user-assistant round as exact context.
         """
         self.provider = provider
         self.keep_recent_ratio = min(max(float(keep_recent_ratio), 0.0), 0.3)
         self.compression_threshold = compression_threshold
         self.token_counter = token_counter or EstimateTokenCounter()
+        self.preserve_latest_round = preserve_latest_round
 
         self.instruction_text = instruction_text or (
             "Based on our full conversation history, produce a concise summary of key takeaways and/or project progress.\n"
@@ -207,21 +212,56 @@ async def __call__(self, messages: list[Message]) -> list[Message]:
         """Use LLM to generate a summary of the conversation history.
 
         Uses round-based splitting to preserve user-assistant turn boundaries.
-        On LLM failure, returns the original messages unchanged (caller should
-        fall back to truncation).
+        On LLM failure, returns the original messages unchanged so the caller
+        can apply its configured fallback policy.
+
+        Args:
+            messages: The original message list.
+
+        Returns:
+            The compressed message list, or the original list when compression
+            cannot be completed safely.
         """
         from .round_utils import split_into_rounds
 
         rounds = split_into_rounds(messages)
         message_rounds = [
             [seg for seg in rnd if isinstance(seg, Message)] for rnd in rounds
         ]
+        latest_complete_round_index: int | None = None
+        if self.preserve_latest_round:
+            complete_round_indices = []
+            for round_index, rnd in enumerate(message_rounds):
+                user_seen = False
+                for msg in rnd:
+                    if msg.role == "user":
+                        user_seen = True
+                    elif user_seen and msg.role == "assistant":
+                        complete_round_indices.append(round_index)
+                        break
+
+            # A summary would have no older complete round to replace.
+            if len(complete_round_indices) <= 1:
+                return messages
+            latest_complete_round_index = complete_round_indices[-1]
+
         total_tokens = self.token_counter.count_tokens(messages)
         old_rounds, recent_rounds = self._split_recent_rounds_by_token_ratio(
             message_rounds,
             total_tokens,
         )
 
+        if latest_complete_round_index is not None:
+            recent_start = min(len(old_rounds), latest_complete_round_index)
+            if not any(
+                msg.role != "system"
+                for rnd in message_rounds[:recent_start]
+                for msg in rnd
+            ):
+                recent_start = latest_complete_round_index
+            old_rounds = message_rounds[:recent_start]
+            recent_rounds = message_rounds[recent_start:]
+
         # The latest user message is the active request. Keep its whole round
         # exact even when the ratio is 0 or the ratio budget would otherwise
         # summarize every round.
@@ -278,7 +318,7 @@ async def __call__(self, messages: list[Message]) -> list[Message]:
             )
             summary_content = (response.completion_text or "").strip()
         except Exception as e:
-            logger.error(f"Failed to generate summary: {e}")
+            logger.error("Context summary failed: %s.", type(e).__name__)
             return messages
 
         if not summary_content:

astrbot/core/agent/context/config.py

@@ -33,3 +33,5 @@ class ContextConfig:
     """Custom token counting method. If None, the default method is used."""
     custom_compressor: ContextCompressor | None = None
     """Custom context compression method. If None, the default method is used."""
+    llm_compress_preserve_latest_round: bool = False
+    """Whether to preserve the latest complete user-assistant round exactly."""