feat: add group message flow context mode

Author: Tsukumi233

astrbot/builtin_stars/astrbot/long_term_memory.py

@@ -238,5 +238,7 @@ async def after_message_sent(self, event: AstrMessageEvent) -> None:
                 clean_session = event.get_extra("_clean_ltm_session", False)
                 if clean_session:
                     await self.ltm.remove_session(event)
+                else:
+                    await self.ltm.record_bot_message(event)
             except Exception as e:
                 logger.error(f"ltm: {e}")

astrbot/builtin_stars/astrbot/main.py

@@ -217,7 +217,12 @@
     },
     "provider_ltm_settings": {
         "group_icl_enable": False,
+        "group_context_mode": "sliding_window",
         "group_message_max_cnt": 300,
+        "group_flow_max_records": 5000,
+        "group_flow_max_delta_messages": 200,
+        "group_flow_max_message_chars": 1000,
+        "group_flow_record_bot_messages": False,
         "image_caption": False,
         "image_caption_provider_id": "",
         "active_reply": {
@@ -2884,9 +2889,25 @@
                     "group_icl_enable": {
                         "type": "bool",
                     },
+                    "group_context_mode": {
+                        "type": "string",
+                        "options": ["sliding_window", "flow"],
+                    },
                     "group_message_max_cnt": {
                         "type": "int",
                     },
+                    "group_flow_max_records": {
+                        "type": "int",
+                    },
+                    "group_flow_max_delta_messages": {
+                        "type": "int",
+                    },
+                    "group_flow_max_message_chars": {
+                        "type": "int",
+                    },
+                    "group_flow_record_bot_messages": {
+                        "type": "bool",
+                    },
                     "image_caption": {
                         "type": "bool",
                     },
@@ -4100,9 +4121,60 @@
                         "description": "启用群聊上下文感知",
                         "type": "bool",
                     },
+                    "provider_ltm_settings.group_context_mode": {
+                        "description": "群聊上下文模式",
+                        "type": "string",
+                        "options": ["sliding_window", "flow"],
+                        "labels": ["滑动窗口", "消息流"],
+                        "hint": "sliding_window 保持旧的滑动窗口行为；flow 使用持久化群聊消息流和对话游标。",
+                        "condition": {
+                            "provider_ltm_settings.group_icl_enable": True,
+                        },
+                    },
                     "provider_ltm_settings.group_message_max_cnt": {
                         "description": "最大消息数量",
                         "type": "int",
+                        "hint": "仅用于 sliding_window 模式。",
+                        "condition": {
+                            "provider_ltm_settings.group_icl_enable": True,
+                            "provider_ltm_settings.group_context_mode": "sliding_window",
+                        },
+                    },
+                    "provider_ltm_settings.group_flow_max_records": {
+                        "description": "群聊消息流保留数量",
+                        "type": "int",
+                        "hint": "仅用于 flow 模式。每个群聊消息流最多保留的历史消息数，0 表示不清理。",
+                        "condition": {
+                            "provider_ltm_settings.group_icl_enable": True,
+                            "provider_ltm_settings.group_context_mode": "flow",
+                        },
+                    },
+                    "provider_ltm_settings.group_flow_max_delta_messages": {
+                        "description": "单次注入消息数量上限",
+                        "type": "int",
+                        "hint": "仅用于 flow 模式。每次只注入游标之后、当前触发消息之前的最近 N 条群聊消息；0 表示不限制。",
+                        "condition": {
+                            "provider_ltm_settings.group_icl_enable": True,
+                            "provider_ltm_settings.group_context_mode": "flow",
+                        },
+                    },
+                    "provider_ltm_settings.group_flow_max_message_chars": {
+                        "description": "单条消息字符上限",
+                        "type": "int",
+                        "hint": "仅用于 flow 模式。每条注入的群聊消息最多保留前 N 个字符；0 表示不限制。",
+                        "condition": {
+                            "provider_ltm_settings.group_icl_enable": True,
+                            "provider_ltm_settings.group_context_mode": "flow",
+                        },
+                    },
+                    "provider_ltm_settings.group_flow_record_bot_messages": {
+                        "description": "记录普通机器人消息",
+                        "type": "bool",
+                        "hint": "仅用于 flow 模式。LLM 本次回复始终不会写入群聊消息流；此项只影响命令或插件产生的普通机器人消息。",
+                        "condition": {
+                            "provider_ltm_settings.group_icl_enable": True,
+                            "provider_ltm_settings.group_context_mode": "flow",
+                        },
                     },
                     "provider_ltm_settings.image_caption": {
                         "description": "自动理解图片",

astrbot/core/config/default.py

@@ -23,6 +23,7 @@
 from astrbot.core.conversation_mgr import ConversationManager
 from astrbot.core.cron import CronJobManager
 from astrbot.core.db import BaseDatabase
+from astrbot.core.group_message_flow_mgr import GroupMessageFlowManager
 from astrbot.core.knowledge_base.kb_mgr import KnowledgeBaseManager
 from astrbot.core.persona_mgr import PersonaManager
 from astrbot.core.pipeline.scheduler import PipelineContext, PipelineScheduler
@@ -211,6 +212,9 @@ async def initialize(self) -> None:
         # 初始化平台消息历史管理器
         self.platform_message_history_manager = PlatformMessageHistoryManager(self.db)
 
+        # 初始化群聊消息流管理器
+        self.group_message_flow_manager = GroupMessageFlowManager(self.db)
+
         # 初始化知识库管理器
         self.kb_manager = KnowledgeBaseManager(self.provider_manager)
 
@@ -233,7 +237,8 @@ async def initialize(self) -> None:
             self.astrbot_config_mgr,
             self.kb_manager,
             self.cron_manager,
-            self.subagent_orchestrator,
+            subagent_orchestrator=self.subagent_orchestrator,
+            group_message_flow_manager=self.group_message_flow_manager,
         )
 
         # 初始化插件管理器

astrbot/core/core_lifecycle.py

@@ -15,6 +15,8 @@
     CommandConflict,
     ConversationV2,
     CronJob,
+    GroupMessageFlowCursor,
+    GroupMessageFlowRecord,
     Persona,
     PersonaFolder,
     PlatformMessageHistory,
@@ -254,6 +256,69 @@ async def get_platform_message_history_by_id(
         """Get a platform message history record by its ID."""
         ...
 
+    @abc.abstractmethod
+    async def insert_group_message_flow_record(
+        self,
+        platform_id: str,
+        flow_session_id: str,
+        content: list,
+        rendered_text: str,
+        group_id: str | None = None,
+        sender_id: str | None = None,
+        sender_name: str | None = None,
+        role: str = "user",
+    ) -> GroupMessageFlowRecord:
+        """Insert a persisted group message flow record."""
+        ...
+
+    @abc.abstractmethod
+    async def get_group_message_flow_records_after(
+        self,
+        flow_session_id: str,
+        after_id: int,
+        before_id: int | None = None,
+        limit: int = 0,
+    ) -> list[GroupMessageFlowRecord]:
+        """Get recent group message flow records after a cursor, ordered oldest first."""
+        ...
+
+    @abc.abstractmethod
+    async def get_latest_group_message_flow_record_id(
+        self,
+        flow_session_id: str,
+    ) -> int:
+        """Get the latest record ID for a group message flow."""
+        ...
+
+    @abc.abstractmethod
+    async def get_group_message_flow_cursor(
+        self,
+        flow_session_id: str,
+        conversation_id: str,
+    ) -> GroupMessageFlowCursor | None:
+        """Get a conversation cursor for a group message flow."""
+        ...
+
+    @abc.abstractmethod
+    async def upsert_group_message_flow_cursor(
+        self,
+        platform_id: str,
+        flow_session_id: str,
+        conversation_id: str,
+        last_record_id: int,
+    ) -> GroupMessageFlowCursor:
+        """Create or update a conversation cursor for a group message flow."""
+        ...
+
+    @abc.abstractmethod
+    async def prune_group_message_flow_records(
+        self,
+        flow_session_id: str,
+        max_records: int,
+    ) -> None:
+        """Keep at most max_records records for a group message flow."""
+        ...
+
     @abc.abstractmethod
     async def create_webchat_thread(
         self,

astrbot/core/db/__init__.py

@@ -247,6 +247,50 @@ class PlatformMessageHistory(TimestampMixin, SQLModel, table=True):
     llm_checkpoint_id: str | None = Field(default=None, index=True)
 
 
+class GroupMessageFlowRecord(TimestampMixin, SQLModel, table=True):
+    """Persisted group chat messages for long-context group flow."""
+
+    __tablename__: str = "group_message_flow_records"
+
+    id: int | None = Field(
+        primary_key=True,
+        sa_column_kwargs={"autoincrement": True},
+        default=None,
+    )
+    platform_id: str = Field(nullable=False, index=True)
+    flow_session_id: str = Field(nullable=False, index=True)
+    group_id: str | None = Field(default=None, index=True)
+    sender_id: str | None = Field(default=None, index=True)
+    sender_name: str | None = Field(default=None)
+    role: str = Field(default="user", nullable=False, index=True)
+    content: list = Field(default_factory=list, sa_type=JSON, nullable=False)
+    rendered_text: str = Field(default="", sa_type=Text, nullable=False)
+
+
+class GroupMessageFlowCursor(TimestampMixin, SQLModel, table=True):
+    """Per-conversation cursor into a group message flow."""
+
+    __tablename__: str = "group_message_flow_cursors"
+
+    id: int | None = Field(
+        primary_key=True,
+        sa_column_kwargs={"autoincrement": True},
+        default=None,
+    )
+    platform_id: str = Field(nullable=False, index=True)
+    flow_session_id: str = Field(nullable=False, index=True)
+    conversation_id: str = Field(nullable=False, index=True)
+    last_record_id: int = Field(default=0, nullable=False)
+
+    __table_args__ = (
+        UniqueConstraint(
+            "flow_session_id",
+            "conversation_id",
+            name="uix_group_message_flow_cursor",
+        ),
+    )
+
+
 class WebChatThread(TimestampMixin, SQLModel, table=True):
     """A side thread created from a selected WebChat assistant response."""