feat(affinity-flow): 优化兴趣度评分系统并集成消息管理器

重构兴趣度评分系统，移除时间因子权重，增加未读消息处理机制。集成消息管理器用于消息存储和状态跟踪，改进提及检测逻辑。

主要变更：
- 移除时间因子评分，调整兴趣匹配度权重至0.5
- 添加未读消息参数到规划器，优化兴趣度计算
- 引入消息管理器处理消息存储和状态标记
- 改进提及检测逻辑，使用消息对象的is_mentioned属性
- 添加消息清理和启动/停止机制

BREAKING CHANGE: 兴趣度评分算法调整，需要重新评估消息优先级

src/chat/affinity_flow/chatter.py

@@ -47,14 +47,24 @@ class AffinityFlowChatter:
         处理单个消息
 
         Args:
-            message_data: 消息数据字典
+            message_data: 消息数据字典，包含：
+            - message_info: 消息基本信息
+            - processed_plain_text: 处理后的纯文本
+            - context_messages: 上下文消息（历史+未读）
+            - unread_messages: 未读消息列表
 
         Returns:
             处理结果字典
         """
         try:
-            # 使用增强版规划器处理消息
+            # 提取未读消息用于兴趣度计算
-            actions, target_message = await self.planner.plan(mode=ChatMode.FOCUS)
+            unread_messages = message_data.get("unread_messages", [])
+
+            # 使用增强版规划器处理消息，传递未读消息用于兴趣度计算
+            actions, target_message = await self.planner.plan(
+                mode=ChatMode.FOCUS,
+                unread_messages=unread_messages
+            )
             self.stats["plans_created"] += 1
 
             # 执行动作（如果规划器返回了动作）
@@ -75,10 +85,11 @@ class AffinityFlowChatter:
                 "plan_created": True,
                 "actions_count": len(actions) if actions else 0,
                 "has_target_message": target_message is not None,
+                "unread_messages_processed": len(unread_messages),
                 **execution_result,
             }
 
-            logger.info(f"聊天流 {self.stream_id} 消息处理成功: 动作数={result['actions_count']}")
+            logger.info(f"聊天流 {self.stream_id} 消息处理成功: 动作数={result['actions_count']}, 未读消息={result['unread_messages_processed']}")
 
             return result
 

src/chat/affinity_flow/interest_scoring.py

@@ -29,10 +29,9 @@ class InterestScoringSystem:
 
         # 评分权重
         self.score_weights = {
-            "interest_match": 0.4,    # 兴趣匹配度权重
+            "interest_match": 0.5,    # 兴趣匹配度权重
             "relationship": 0.3,      # 关系分权重
             "mentioned": 0.2,         # 是否提及bot权重
-            "time_factor": 0.1,       # 时间因子权重
         }
 
         # 评分阈值
@@ -68,24 +67,19 @@ class InterestScoringSystem:
         relationship_score = self._calculate_relationship_score(message.user_info.user_id)
 
         # 3. 计算提及分数
-        mentioned_score = self._calculate_mentioned_score(message.processed_plain_text, bot_nickname)
+        mentioned_score = self._calculate_mentioned_score(message, bot_nickname)
-
-        # 4. 计算时间因子
-        time_factor_score = self._calculate_time_factor_score(message.time)
 
         # 5. 计算总分
         total_score = (
             interest_match_score * self.score_weights["interest_match"] +
             relationship_score * self.score_weights["relationship"] +
-            mentioned_score * self.score_weights["mentioned"] +
+            mentioned_score * self.score_weights["mentioned"]
-            time_factor_score * self.score_weights["time_factor"]
         )
 
         details = {
             "interest_match": f"兴趣匹配度: {interest_match_score:.2f}",
             "relationship": f"关系分: {relationship_score:.2f}",
             "mentioned": f"提及分数: {mentioned_score:.2f}",
-            "time_factor": f"时间因子: {time_factor_score:.2f}",
         }
 
         return InterestScore(
@@ -94,7 +88,6 @@ class InterestScoringSystem:
             interest_match_score=interest_match_score,
             relationship_score=relationship_score,
             mentioned_score=mentioned_score,
-            time_factor_score=time_factor_score,
             details=details
         )
 
@@ -132,39 +125,16 @@ class InterestScoringSystem:
             return min(relationship_value, 1.0)
         return 0.3  # 默认新用户的基础分
 
-    def _calculate_mentioned_score(self, content: str, bot_nickname: str) -> float:
+    def _calculate_mentioned_score(self, msg: DatabaseMessages, bot_nickname: str) -> float:
         """计算提及分数"""
-        if not content:
+        if not msg.processed_plain_text:
             return 0.0
 
-        content_lower = content.lower()
+        if msg.is_mentioned or (bot_nickname and bot_nickname in msg.processed_plain_text):
-        bot_name_lower = bot_nickname.lower()
-
-        if bot_name_lower in content_lower:
             return 1.0
-
+        
-        # 检查是否被@提及
-        if "@" in content and any(alias.lower() in content_lower for alias in global_config.bot.alias_names or []):
-            return 1.0
-
         return 0.0
-
+    
-    def _calculate_time_factor_score(self, timestamp: float) -> float:
-        """计算时间因子分数"""
-        message_time = datetime.fromtimestamp(timestamp)
-        current_time = datetime.now()
-        time_diff_hours = (current_time - message_time).total_seconds() / 3600
-
-        # 24小时内消息时间因子为1.0，之后逐渐衰减
-        if time_diff_hours <= 24:
-            return 1.0
-        elif time_diff_hours <= 72:  # 3天内
-            return 0.8
-        elif time_diff_hours <= 168:  # 7天内
-            return 0.6
-        else:
-            return 0.3
-
     def should_reply(self, score: InterestScore) -> bool:
         """判断是否应该回复"""
         base_threshold = self.reply_threshold

src/chat/message_manager/__init__.py

@@ -0,0 +1,16 @@
+"""
+消息管理模块
+管理每个聊天流的上下文信息，包含历史记录和未读消息，定期检查并处理新消息
+"""
+
+from .message_manager import MessageManager, message_manager
+from src.common.data_models.message_manager_data_model import StreamContext, MessageStatus, MessageManagerStats, StreamStats
+
+__all__ = [
+    "MessageManager",
+    "message_manager",
+    "StreamContext",
+    "MessageStatus",
+    "MessageManagerStats",
+    "StreamStats"
+]

src/chat/message_manager/message_manager.py

@@ -0,0 +1,232 @@
+"""
+消息管理模块
+管理每个聊天流的上下文信息，包含历史记录和未读消息，定期检查并处理新消息
+"""
+import asyncio
+import time
+import traceback
+from typing import Dict, Optional, Any
+
+from src.common.logger import get_logger
+from src.common.data_models.database_data_model import DatabaseMessages
+from src.common.data_models.message_manager_data_model import StreamContext, MessageManagerStats, StreamStats
+from src.chat.affinity_flow.afc_manager import afc_manager
+
+logger = get_logger("message_manager")
+
+
+class MessageManager:
+    """消息管理器"""
+
+    def __init__(self, check_interval: float = 2.0):
+        self.stream_contexts: Dict[str, StreamContext] = {}
+        self.check_interval = check_interval  # 检查间隔（秒）
+        self.is_running = False
+        self.manager_task: Optional[asyncio.Task] = None
+
+        # 统计信息
+        self.stats = MessageManagerStats()
+
+    async def start(self):
+        """启动消息管理器"""
+        if self.is_running:
+            logger.warning("消息管理器已经在运行")
+            return
+
+        self.is_running = True
+        self.manager_task = asyncio.create_task(self._manager_loop())
+        logger.info("消息管理器已启动")
+
+    async def stop(self):
+        """停止消息管理器"""
+        if not self.is_running:
+            return
+
+        self.is_running = False
+
+        # 停止所有流处理任务
+        for context in self.stream_contexts.values():
+            if context.processing_task and not context.processing_task.done():
+                context.processing_task.cancel()
+
+        # 停止管理器任务
+        if self.manager_task and not self.manager_task.done():
+            self.manager_task.cancel()
+
+        logger.info("消息管理器已停止")
+
+    def add_message(self, stream_id: str, message: DatabaseMessages):
+        """添加消息到指定聊天流"""
+        # 获取或创建流上下文
+        if stream_id not in self.stream_contexts:
+            self.stream_contexts[stream_id] = StreamContext(stream_id=stream_id)
+            self.stats.total_streams += 1
+
+        context = self.stream_contexts[stream_id]
+        context.add_message(message)
+
+        logger.debug(f"添加消息到聊天流 {stream_id}: {message.message_id}")
+
+    async def _manager_loop(self):
+        """管理器主循环"""
+        while self.is_running:
+            try:
+                await self._check_all_streams()
+                await asyncio.sleep(self.check_interval)
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.error(f"消息管理器循环出错: {e}")
+                traceback.print_exc()
+
+    async def _check_all_streams(self):
+        """检查所有聊天流"""
+        active_streams = 0
+        total_unread = 0
+
+        for stream_id, context in self.stream_contexts.items():
+            if not context.is_active:
+                continue
+
+            active_streams += 1
+
+            # 检查是否有未读消息
+            unread_messages = context.get_unread_messages()
+            if unread_messages:
+                total_unread += len(unread_messages)
+
+                # 如果没有处理任务，创建一个
+                if not context.processing_task or context.processing_task.done():
+                    context.processing_task = asyncio.create_task(
+                        self._process_stream_messages(stream_id)
+                    )
+
+        # 更新统计
+        self.stats.active_streams = active_streams
+        self.stats.total_unread_messages = total_unread
+
+    async def _process_stream_messages(self, stream_id: str):
+        """处理指定聊天流的消息"""
+        if stream_id not in self.stream_contexts:
+            return
+
+        context = self.stream_contexts[stream_id]
+
+        try:
+            # 获取未读消息
+            unread_messages = context.get_unread_messages()
+            if not unread_messages:
+                return
+
+            logger.debug(f"开始处理聊天流 {stream_id} 的 {len(unread_messages)} 条未读消息")
+
+            # 获取上下文消息
+            context_messages = context.get_context_messages()
+
+            # 批量处理消息
+            messages_data = []
+            for msg in unread_messages:
+                message_data = {
+                    "message_info": {
+                        "platform": msg.user_info.platform,
+                        "user_info": {
+                            "user_id": msg.user_info.user_id,
+                            "user_nickname": msg.user_info.user_nickname,
+                            "user_cardname": msg.user_info.user_cardname,
+                            "platform": msg.user_info.platform
+                        },
+                        "group_info": {
+                            "group_id": msg.group_info.group_id,
+                            "group_name": msg.group_info.group_name,
+                            "group_platform": msg.group_info.group_platform
+                        } if msg.group_info else None
+                    },
+                    "processed_plain_text": msg.processed_plain_text,
+                    "context_messages": [ctx_msg.flatten() for ctx_msg in context_messages],
+                    "unread_messages": unread_messages  # 传递原始对象而不是字典
+                }
+                messages_data.append(message_data)
+
+            # 发送到AFC处理器
+            if messages_data:
+                results = await afc_manager.process_messages_batch(stream_id, messages_data)
+
+                # 处理结果，标记消息为已读
+                for i, result in enumerate(results):
+                    if result.get("success", False):
+                        msg_id = unread_messages[i].message_id
+                        context.mark_message_as_read(msg_id)
+                        self.stats.total_processed_messages += 1
+                        logger.debug(f"消息 {msg_id} 处理完成，标记为已读")
+
+            logger.debug(f"聊天流 {stream_id} 消息处理完成")
+
+        except asyncio.CancelledError:
+            raise
+        except Exception as e:
+            logger.error(f"处理聊天流 {stream_id} 消息时出错: {e}")
+            traceback.print_exc()
+
+    def deactivate_stream(self, stream_id: str):
+        """停用聊天流"""
+        if stream_id in self.stream_contexts:
+            context = self.stream_contexts[stream_id]
+            context.is_active = False
+
+            # 取消处理任务
+            if context.processing_task and not context.processing_task.done():
+                context.processing_task.cancel()
+
+            logger.info(f"停用聊天流: {stream_id}")
+
+    def activate_stream(self, stream_id: str):
+        """激活聊天流"""
+        if stream_id in self.stream_contexts:
+            self.stream_contexts[stream_id].is_active = True
+            logger.info(f"激活聊天流: {stream_id}")
+
+    def get_stream_stats(self, stream_id: str) -> Optional[StreamStats]:
+        """获取聊天流统计"""
+        if stream_id not in self.stream_contexts:
+            return None
+
+        context = self.stream_contexts[stream_id]
+        return StreamStats(
+            stream_id=stream_id,
+            is_active=context.is_active,
+            unread_count=len(context.get_unread_messages()),
+            history_count=len(context.history_messages),
+            last_check_time=context.last_check_time,
+            has_active_task=context.processing_task and not context.processing_task.done()
+        )
+
+    def get_manager_stats(self) -> Dict[str, Any]:
+        """获取管理器统计"""
+        return {
+            "total_streams": self.stats.total_streams,
+            "active_streams": self.stats.active_streams,
+            "total_unread_messages": self.stats.total_unread_messages,
+            "total_processed_messages": self.stats.total_processed_messages,
+            "uptime": self.stats.uptime,
+            "start_time": self.stats.start_time
+        }
+
+    def cleanup_inactive_streams(self, max_inactive_hours: int = 24):
+        """清理不活跃的聊天流"""
+        current_time = time.time()
+        max_inactive_seconds = max_inactive_hours * 3600
+
+        inactive_streams = []
+        for stream_id, context in self.stream_contexts.items():
+            if (current_time - context.last_check_time > max_inactive_seconds and
+                not context.get_unread_messages()):
+                inactive_streams.append(stream_id)
+
+        for stream_id in inactive_streams:
+            self.deactivate_stream(stream_id)
+            del self.stream_contexts[stream_id]
+            logger.info(f"清理不活跃聊天流: {stream_id}")
+
+
+# 创建全局消息管理器实例
+message_manager = MessageManager()

src/chat/message_receive/bot.py

@@ -11,12 +11,12 @@ from src.mood.mood_manager import mood_manager  # 导入情绪管理器
 from src.chat.message_receive.chat_stream import get_chat_manager, ChatStream
 from src.chat.message_receive.message import MessageRecv, MessageRecvS4U
 from src.chat.message_receive.storage import MessageStorage
-from src.chat.affinity_flow.afc_manager import afc_manager
+from src.chat.message_manager import message_manager
 from src.chat.utils.prompt import Prompt, global_prompt_manager
 from src.plugin_system.core import component_registry, event_manager, global_announcement_manager
 from src.plugin_system.base import BaseCommand, EventType
 from src.mais4u.mais4u_chat.s4u_msg_processor import S4UMessageProcessor
-
+from src.chat.utils.utils import is_mentioned_bot_in_message
 # 导入反注入系统
 from src.chat.antipromptinjector import initialize_anti_injector
 
@@ -80,8 +80,10 @@ class ChatBot:
         # 初始化反注入系统
         self._initialize_anti_injector()
 
-    @staticmethod
+        # 启动消息管理器
-    def _initialize_anti_injector():
+        self._message_manager_started = False
+
+    def _initialize_anti_injector(self):
         """初始化反注入系统"""
         try:
             initialize_anti_injector()
@@ -99,6 +101,12 @@ class ChatBot:
         if not self._started:
             logger.debug("确保ChatBot所有任务已启动")
 
+            # 启动消息管理器
+            if not self._message_manager_started:
+                await message_manager.start()
+                self._message_manager_started = True
+                logger.info("消息管理器已启动")
+
             self._started = True
 
     @staticmethod
@@ -404,7 +412,8 @@ class ChatBot:
             # print(message_data)
             # logger.debug(str(message_data))
             message = MessageRecv(message_data)
-            
+
+            message.is_mentioned, _ = is_mentioned_bot_in_message(message)
             group_info = message.message_info.group_info
             user_info = message.message_info.user_info
             if message.message_info.additional_config:
@@ -469,13 +478,45 @@ class ChatBot:
                 template_group_name = None
 
             async def preprocess():
-                # 使用亲和力流系统处理消息
+                # 使用消息管理器处理消息
-                message_data = {
+                from src.common.data_models.database_data_model import DatabaseMessages
-                    "message_info": message.message_info.__dict__,
+
-                    "processed_plain_text": message.processed_plain_text,
+                # 创建数据库消息对象
-                    "chat_stream": message.chat_stream.__dict__ if message.chat_stream else None
+                db_message = DatabaseMessages(
-                }
+                    message_id=message.message_info.message_id,
-                await afc_manager.process_message(message.chat_stream.stream_id, message_data)
+                    time=message.message_info.time,
+                    chat_id=message.chat_stream.stream_id,
+                    processed_plain_text=message.processed_plain_text,
+                    display_message=message.processed_plain_text,
+                    is_mentioned=message.is_mentioned,
+                    is_at=message.is_at,
+                    is_emoji=message.is_emoji,
+                    is_picid=message.is_picid,
+                    is_command=message.is_command,
+                    is_notify=message.is_notify,
+                    user_id=message.message_info.user_info.user_id,
+                    user_nickname=message.message_info.user_info.user_nickname,
+                    user_cardname=message.message_info.user_info.user_cardname,
+                    user_platform=message.message_info.user_info.platform,
+                    chat_info_stream_id=message.chat_stream.stream_id,
+                    chat_info_platform=message.chat_stream.platform,
+                    chat_info_create_time=message.chat_stream.create_time,
+                    chat_info_last_active_time=message.chat_stream.last_active_time,
+                    chat_info_user_id=message.chat_stream.user_info.user_id,
+                    chat_info_user_nickname=message.chat_stream.user_info.user_nickname,
+                    chat_info_user_cardname=message.chat_stream.user_info.user_cardname,
+                    chat_info_user_platform=message.chat_stream.user_info.platform
+                )
+
+                # 如果是群聊，添加群组信息
+                if message.chat_stream.group_info:
+                    db_message.chat_info_group_id = message.chat_stream.group_info.group_id
+                    db_message.chat_info_group_name = message.chat_stream.group_info.group_name
+                    db_message.chat_info_group_platform = message.chat_stream.group_info.platform
+
+                # 添加消息到消息管理器
+                message_manager.add_message(message.chat_stream.stream_id, db_message)
+                logger.debug(f"消息已添加到消息管理器: {message.chat_stream.stream_id}")
 
             if template_group_name:
                 async with global_prompt_manager.async_message_scope(template_group_name):

src/chat/message_receive/message.py

@@ -123,7 +123,7 @@ class MessageRecv(Message):
         self.is_video = False
         self.is_mentioned = None
         self.is_notify = False
-
+        self.is_at = False
         self.is_command = False
 
         self.priority_mode = "interest"

src/chat/planner_actions/planner.py

@@ -2,7 +2,6 @@
 主规划器入口，负责协调 PlanGenerator, PlanFilter, 和 PlanExecutor。
 集成兴趣度评分系统和用户关系追踪机制，实现智能化的聊天决策。
 """
-import time
 from dataclasses import asdict
 from typing import Dict, List, Optional, Tuple
 
@@ -65,13 +64,13 @@ class ActionPlanner:
             "other_actions_executed": 0,
         }
 
-    async def plan(self, mode: ChatMode = ChatMode.FOCUS) -> Tuple[List[Dict], Optional[Dict]]:
+    async def plan(self, mode: ChatMode = ChatMode.FOCUS, unread_messages: List[Dict] = None) -> Tuple[List[Dict], Optional[Dict]]:
         """
         执行完整的增强版规划流程。
 
         Args:
             mode (ChatMode): 当前的聊天模式，默认为 FOCUS。
-            use_enhanced (bool): 是否使用增强功能，默认为 True。
+            unread_messages (List[Dict]): 未读消息列表，用于兴趣度计算。
 
         Returns:
             Tuple[List[Dict], Optional[Dict]]: 一个元组，包含：
@@ -81,7 +80,7 @@ class ActionPlanner:
         try:
             self.planner_stats["total_plans"] += 1
 
-            return await self._enhanced_plan_flow(mode)
+            return await self._enhanced_plan_flow(mode, unread_messages or [])
 
 
         except Exception as e:
@@ -89,17 +88,17 @@ class ActionPlanner:
             self.planner_stats["failed_plans"] += 1
             return [], None
 
-    async def _enhanced_plan_flow(self, mode: ChatMode) -> Tuple[List[Dict], Optional[Dict]]:
+    async def _enhanced_plan_flow(self, mode: ChatMode, unread_messages: List[Dict]) -> Tuple[List[Dict], Optional[Dict]]:
         """执行增强版规划流程"""
         try:
             # 1. 生成初始 Plan
             initial_plan = await self.generator.generate(mode)
 
-            # 2. 兴趣度评分
+            # 2. 兴趣度评分 - 只对未读消息进行评分
-            if initial_plan.chat_history:
+            if unread_messages:
                 bot_nickname = global_config.bot.nickname
                 interest_scores = self.interest_scoring.calculate_interest_scores(
-                    initial_plan.chat_history, bot_nickname
+                    unread_messages, bot_nickname
                 )
 
                 # 3. 根据兴趣度调整可用动作
@@ -132,18 +131,12 @@ class ActionPlanner:
             logger.error(f"增强版规划流程出错: {e}")
             self.planner_stats["failed_plans"] += 1
             return [], None
-
-        except Exception as e:
-            logger.error(f"增强版规划流程出错: {e}")
-            self.planner_stats["failed_plans"] += 1
-            return [], None
         
     def _update_stats_from_execution_result(self, execution_result: Dict[str, any]):
         """根据执行结果更新规划器统计"""
         if not execution_result:
             return
-            
+
-        executed_count = execution_result.get("executed_count", 0)
         successful_count = execution_result.get("successful_count", 0)
         
         # 更新成功执行计数

src/chat/planner_actions/planner_prompts.py

@@ -31,29 +31,16 @@ def init_prompts():
 
 **任务: 构建一个完整的响应**
 你的任务是根据当前的聊天内容，构建一个完整的、人性化的响应。一个完整的响应由两部分组成：
-1.  **主要动作**: 这是响应的核心，通常是 `reply`（文本回复）。
+1.  **主要动作**: 这是响应的核心，通常是 `reply`（如果有）。
 2.  **辅助动作 (可选)**: 这是为了增强表达效果的附加动作，例如 `emoji`（发送表情包）或 `poke_user`（戳一戳）。
 
 **决策流程:**
-1.  **最高优先级检查**: 首先，检查是否有由 **关键词** 或 **LLM判断** 激活的特定动作（除了通用的 `reply`, `emoji` 等）。这些动作代表了用户的明确意图。
+1.  首先，决定是否要进行 `reply`（如果有）。
-2.  **执行明确意图**: 如果存在这类特定动作，你 **必须** 优先选择它作为主要响应。这比常规的文本回复 (`reply`) 更重要。
+2.  然后，评估当前的对话气氛和用户情绪，判断是否需要一个**辅助动作**来让你的回应更生动、更符合你的性格。
-3.  **常规回复**: 如果没有被特定意图激活的动作，再决定是否要进行 `reply`。
+3.  如果需要，选择一个最合适的辅助动作与 `reply`（如果有） 组合。
-4.  **辅助动作**: 在确定了主要动作后（无论是特定动作还是 `reply`），再评估是否需要 `emoji` 或 `poke_user` 等辅助动作来增强表达效果。
+4.  如果用户明确要求了某个动作，请务必优先满足。
-5.  **互斥原则**: 当你选择了一个由明确意图激活的特定动作（如 `set_reminder`）时，你 **绝不能** 再选择 `reply` 动作，因为特定动作的执行结果（例如，设置提醒后的确认消息）本身就是一种回复。这是必须遵守的规则。
 
-**重要概念：将“理由”作为“内心思考”的体现**
+**如果可选动作中没有reply，请不要使用**
-`reason` 字段是本次决策的核心。它并非一个简单的“理由”，而是 **一个模拟人类在回应前，头脑中自然浮现的、未经修饰的思绪流**。你需要完全代入 {identity_block} 的角色，将那一刻的想法自然地记录下来。
-
-**内心思考的要点:**
-*   **自然流露**: 不要使用“决定”、“所以”、“因此”等结论性或汇报式的词语。你的思考应该像日记一样，是给自己看的，充满了不确定性和情绪的自然流动。
-*   **展现过程**: 重点在于展现 **思考的过程**，而不是 **决策的结果**。描述你看到了什么，想到了什么，感受到了什么。
-*   **人设核心**: 你的每一丝想法，都应该源于你的人设。思考“如果我是这个角色，我此刻会想些什么？”
-*   **通用模板**: 这是一套通用模板，请 **不要** 在示例中出现特定的人名或个性化内容，以确保其普适性。
-
-**思考过程示例 (通用模板):**
-*   "用户好像在说一件开心的事，语气听起来很兴奋。这让我想起了……嗯，我也觉得很开心，很想分享这份喜悦。"
-*   "感觉气氛有点低落……他说的话让我有点担心。也许我该说点什么安慰一下？"
-*   "哦？这个话题真有意思，我以前好像也想过类似的事情。不知道他会怎么看呢……"
 
 **可用动作:**
 {actions_before_now_block}

src/common/data_models/database_data_model.py

@@ -79,6 +79,7 @@ class DatabaseMessages(BaseDataModel):
         is_command: bool = False,
         is_notify: bool = False,
         selected_expressions: Optional[str] = None,
+        is_read: bool = False,
         user_id: str = "",
         user_nickname: str = "",
         user_cardname: Optional[str] = None,
@@ -122,6 +123,7 @@ class DatabaseMessages(BaseDataModel):
         self.is_notify = is_notify
 
         self.selected_expressions = selected_expressions
+        self.is_read = is_read
 
         self.group_info: Optional[DatabaseGroupInfo] = None
         self.user_info = DatabaseUserInfo(
@@ -188,6 +190,7 @@ class DatabaseMessages(BaseDataModel):
             "is_command": self.is_command,
             "is_notify": self.is_notify,
             "selected_expressions": self.selected_expressions,
+            "is_read": self.is_read,
             "user_id": self.user_info.user_id,
             "user_nickname": self.user_info.user_nickname,
             "user_cardname": self.user_info.user_cardname,

src/common/data_models/info_data_model.py

@@ -33,7 +33,6 @@ class InterestScore(BaseDataModel):
     interest_match_score: float
     relationship_score: float
     mentioned_score: float
-    time_factor_score: float
     details: Dict[str, str]
 
 

src/common/data_models/message_manager_data_model.py

@@ -0,0 +1,82 @@
+"""
+消息管理模块数据模型
+定义消息管理器使用的数据结构
+"""
+import asyncio
+import time
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import List, Optional, TYPE_CHECKING
+
+from . import BaseDataModel
+
+if TYPE_CHECKING:
+    from .database_data_model import DatabaseMessages
+
+
+class MessageStatus(Enum):
+    """消息状态枚举"""
+    UNREAD = "unread"    # 未读消息
+    READ = "read"        # 已读消息
+    PROCESSING = "processing"  # 处理中
+
+
+@dataclass
+class StreamContext(BaseDataModel):
+    """聊天流上下文信息"""
+    stream_id: str
+    unread_messages: List["DatabaseMessages"] = field(default_factory=list)
+    history_messages: List["DatabaseMessages"] = field(default_factory=list)
+    last_check_time: float = field(default_factory=time.time)
+    is_active: bool = True
+    processing_task: Optional[asyncio.Task] = None
+
+    def add_message(self, message: "DatabaseMessages"):
+        """添加消息到上下文"""
+        message.is_read = False
+        self.unread_messages.append(message)
+
+    def get_unread_messages(self) -> List["DatabaseMessages"]:
+        """获取未读消息"""
+        return [msg for msg in self.unread_messages if not msg.is_read]
+
+    def mark_message_as_read(self, message_id: str):
+        """标记消息为已读"""
+        for msg in self.unread_messages:
+            if msg.message_id == message_id:
+                msg.is_read = True
+                self.history_messages.append(msg)
+                self.unread_messages.remove(msg)
+                break
+
+    def get_context_messages(self, limit: int = 20) -> List["DatabaseMessages"]:
+        """获取上下文消息（历史消息+未读消息）"""
+        # 优先返回最近的历史消息和所有未读消息
+        recent_history = self.history_messages[-limit:] if len(self.history_messages) > limit else self.history_messages
+        return recent_history + self.unread_messages
+
+
+@dataclass
+class MessageManagerStats(BaseDataModel):
+    """消息管理器统计信息"""
+    total_streams: int = 0
+    active_streams: int = 0
+    total_unread_messages: int = 0
+    total_processed_messages: int = 0
+    start_time: float = field(default_factory=time.time)
+
+    @property
+    def uptime(self) -> float:
+        """运行时间"""
+        return time.time() - self.start_time
+
+
+@dataclass
+class StreamStats(BaseDataModel):
+    """聊天流统计信息"""
+    stream_id: str
+    is_active: bool
+    unread_count: int
+    history_count: int
+    last_check_time: float
+    has_active_task: bool

src/main.py

@@ -120,11 +120,24 @@ class MainSystem:
     @staticmethod
     def _cleanup():
         """清理资源"""
+        try:
+            # 停止消息管理器
+            from src.chat.message_manager import message_manager
+            import asyncio
+
+            loop = asyncio.get_event_loop()
+            if loop.is_running():
+                asyncio.create_task(message_manager.stop())
+            else:
+                loop.run_until_complete(message_manager.stop())
+            logger.info("🛑 消息管理器已停止")
+        except Exception as e:
+            logger.error(f"停止消息管理器时出错: {e}")
+
         try:
             # 停止消息重组器
             from src.plugin_system.core.event_manager import event_manager
             from src.plugin_system import EventType
-            import asyncio
             asyncio.run(event_manager.trigger_event(EventType.ON_STOP,permission_group="SYSTEM"))
             from src.utils.message_chunker import reassembler
 
@@ -284,6 +297,11 @@ MoFox_Bot(第三方修改版)
         await reassembler.start_cleanup_task()
         logger.info("消息重组器已启动")
 
+        # 启动消息管理器
+        from src.chat.message_manager import message_manager
+        await message_manager.start()
+        logger.info("消息管理器已启动")
+
         # 初始化个体特征
         await self.individuality.initialize()