Merge branch 'dev' of https://github.com/MoFox-Studio/MoFox-Core into dev

src/chat/express/expression_learner.py

@@ -1,3 +1,4 @@
+import hashlib
 import os
 import time
 from datetime import datetime
@@ -126,6 +127,55 @@ class ExpressionLearner:
         self.min_learning_interval = 300  # 最短学习时间间隔（秒）
         self._chat_name_initialized = False
 
+    @staticmethod
+    def _parse_stream_config_to_chat_id(stream_config_str: str) -> str | None:
+        """解析'platform:id:type'为chat_id（与get_stream_id一致）"""
+        try:
+            parts = stream_config_str.split(":")
+            if len(parts) != 3:
+                return None
+            platform = parts[0]
+            id_str = parts[1]
+            stream_type = parts[2]
+            is_group = stream_type == "group"
+            if is_group:
+                components = [platform, str(id_str)]
+            else:
+                components = [platform, str(id_str), "private"]
+            key = "_".join(components)
+            return hashlib.md5(key.encode()).hexdigest()
+        except Exception:
+            return None
+
+    def get_related_chat_ids(self) -> list[str]:
+        """根据expression.rules配置，获取与当前chat_id相关的所有chat_id（包括自身）
+        
+        用于共享组功能：同一共享组内的聊天流可以共享学习到的表达方式
+        """
+        if global_config is None:
+            return [self.chat_id]
+        rules = global_config.expression.rules
+        current_group = None
+
+        # 找到当前chat_id所在的组
+        for rule in rules:
+            if rule.chat_stream_id and self._parse_stream_config_to_chat_id(rule.chat_stream_id) == self.chat_id:
+                current_group = rule.group
+                break
+
+        # 始终包含当前 chat_id（确保至少能查到自己的数据）
+        related_chat_ids = [self.chat_id]
+
+        if current_group:
+            # 找出同一组的所有chat_id
+            for rule in rules:
+                if rule.group == current_group and rule.chat_stream_id:
+                    if chat_id_candidate := self._parse_stream_config_to_chat_id(rule.chat_stream_id):
+                        if chat_id_candidate not in related_chat_ids:
+                            related_chat_ids.append(chat_id_candidate)
+
+        return related_chat_ids
+
     async def _initialize_chat_name(self):
         """异步初始化chat_name"""
         if not self._chat_name_initialized:
@@ -540,46 +590,62 @@ class ExpressionLearner:
                 # 提交后清除相关缓存
                 await session.commit()
 
-            # 清除该chat_id的表达方式缓存
+            # 🔥 清除共享组内所有 chat_id 的表达方式缓存
             from src.common.database.optimization.cache_manager import get_cache
             from src.common.database.utils.decorators import generate_cache_key
             cache = await get_cache()
-            await cache.delete(generate_cache_key("chat_expressions", chat_id))
+            
+            # 获取共享组内所有 chat_id 并清除其缓存
+            related_chat_ids = self.get_related_chat_ids()
+            for related_id in related_chat_ids:
+                await cache.delete(generate_cache_key("chat_expressions", related_id))
+            if len(related_chat_ids) > 1:
+                logger.debug(f"已清除共享组内 {len(related_chat_ids)} 个 chat_id 的表达方式缓存")
 
-            # 🔥 训练 StyleLearner
+            # 🔥 训练 StyleLearner（支持共享组）
             # 只对 style 类型的表达方式进行训练（grammar 不需要训练到模型）
             if type == "style":
                 try:
-                    # 获取 StyleLearner 实例
-                    learner = style_learner_manager.get_learner(chat_id)
+                    logger.debug(f"开始训练 StyleLearner: 源chat_id={chat_id}, 共享组包含 {len(related_chat_ids)} 个chat_id, 样本数={len(expr_list)}")
 
-                    logger.debug(f"开始训练 StyleLearner: chat_id={chat_id}, 样本数={len(expr_list)}")
+                    # 为每个共享组内的 chat_id 训练其 StyleLearner
+                    for target_chat_id in related_chat_ids:
+                        learner = style_learner_manager.get_learner(target_chat_id)
+                        
+                        # 为每个学习到的表达方式训练模型
+                        # 使用 situation 作为输入，style 作为目标
+                        # 这是最符合语义的方式：场景 -> 表达方式
+                        success_count = 0
+                        for expr in expr_list:
+                            situation = expr["situation"]
+                            style = expr["style"]
 
-                    # 为每个学习到的表达方式训练模型
-                    # 使用 situation 作为输入，style 作为目标
-                    # 这是最符合语义的方式：场景 -> 表达方式
-                    success_count = 0
-                    for expr in expr_list:
-                        situation = expr["situation"]
-                        style = expr["style"]
+                            # 训练映射关系: situation -> style
+                            if learner.learn_mapping(situation, style):
+                                success_count += 1
+                            else:
+                                logger.warning(f"训练失败 (target={target_chat_id}): {situation} -> {style}")
 
-                        # 训练映射关系: situation -> style
-                        if learner.learn_mapping(situation, style):
-                            success_count += 1
+                        # 保存模型
+                        if learner.save(style_learner_manager.model_save_path):
+                            logger.debug(f"StyleLearner 模型保存成功: {target_chat_id}")
                         else:
-                            logger.warning(f"训练失败: {situation} -> {style}")
+                            logger.error(f"StyleLearner 模型保存失败: {target_chat_id}")
 
-                    logger.info(
-                        f"StyleLearner 训练完成: {success_count}/{len(expr_list)} 成功, "
-                        f"当前风格总数={len(learner.get_all_styles())}, "
-                        f"总样本数={learner.learning_stats['total_samples']}"
-                    )
+                        if target_chat_id == chat_id:
+                            # 只为源 chat_id 记录详细日志
+                            logger.info(
+                                f"StyleLearner 训练完成 (源): {success_count}/{len(expr_list)} 成功, "
+                                f"当前风格总数={len(learner.get_all_styles())}, "
+                                f"总样本数={learner.learning_stats['total_samples']}"
+                            )
+                        else:
+                            logger.debug(
+                                f"StyleLearner 训练完成 (共享组成员 {target_chat_id}): {success_count}/{len(expr_list)} 成功"
+                            )
 
-                    # 保存模型
-                    if learner.save(style_learner_manager.model_save_path):
-                        logger.debug(f"StyleLearner 模型保存成功: {chat_id}")
-                    else:
-                        logger.error(f"StyleLearner 模型保存失败: {chat_id}")
+                    if len(related_chat_ids) > 1:
+                        logger.info(f"共享组内共 {len(related_chat_ids)} 个 StyleLearner 已同步训练")
 
                 except Exception as e:
                     logger.error(f"训练 StyleLearner 失败: {e}")
@@ -604,13 +670,14 @@ class ExpressionLearner:
 
         current_time = time.time()
 
-        # 获取上次学习时间，过滤掉机器人自己的消息
+        # 获取上次学习时间，过滤掉机器人自己的消息和无意义消息
         random_msg: list[dict[str, Any]] | None = await get_raw_msg_by_timestamp_with_chat_inclusive(
             chat_id=self.chat_id,
             timestamp_start=self.last_learning_time,
             timestamp_end=current_time,
             limit=num,
             filter_bot=True,  # 过滤掉机器人自己的消息，防止学习自己的表达方式
+            filter_meaningless=True,  # 🔥 过滤掉表情包、通知等无意义消息
         )
 
         # print(random_msg)
@@ -619,8 +686,14 @@ class ExpressionLearner:
         # 转化成str
         chat_id: str = random_msg[0]["chat_id"]
         # random_msg_str: str = build_readable_messages(random_msg, timestamp_mode="normal")
-        random_msg_str: str = await build_anonymous_messages(random_msg)
+        # 🔥 启用表达学习场景的过滤，过滤掉纯回复、纯@、纯图片等无意义内容
+        random_msg_str: str = await build_anonymous_messages(random_msg, filter_for_learning=True)
         # print(f"random_msg_str:{random_msg_str}")
+        
+        # 🔥 检查过滤后是否还有足够的内容
+        if not random_msg_str or len(random_msg_str.strip()) < 20:
+            logger.debug(f"过滤后消息内容不足，跳过本次{type_str}学习")
+            return None
 
         prompt: str = await global_prompt_manager.format_prompt(
             prompt,

src/chat/express/style_learner.py

@@ -437,7 +437,13 @@ class StyleLearner:
 
 
 class StyleLearnerManager:
-    """多聊天室表达风格学习管理器"""
+    """多聊天室表达风格学习管理器
+    
+    添加 LRU 淘汰机制，限制最大活跃 learner 数量
+    """
+
+    # 🔧 最大活跃 learner 数量
+    MAX_ACTIVE_LEARNERS = 50
 
     def __init__(self, model_save_path: str = "data/expression/style_models"):
         """
@@ -445,6 +451,7 @@ class StyleLearnerManager:
             model_save_path: 模型保存路径
         """
         self.learners: dict[str, StyleLearner] = {}
+        self.learner_last_used: dict[str, float] = {}  # 🔧 记录最后使用时间
         self.model_save_path = model_save_path
 
         # 确保保存目录存在
@@ -452,6 +459,30 @@ class StyleLearnerManager:
 
         logger.debug(f"StyleLearnerManager初始化成功, 模型保存路径: {model_save_path}")
 
+    def _evict_if_needed(self) -> None:
+        """🔧 内存优化：如果超过最大数量，淘汰最久未使用的 learner"""
+        if len(self.learners) < self.MAX_ACTIVE_LEARNERS:
+            return
+
+        # 按最后使用时间排序，淘汰最旧的 20%
+        evict_count = max(1, len(self.learners) // 5)
+        sorted_by_time = sorted(
+            self.learner_last_used.items(),
+            key=lambda x: x[1]
+        )
+        
+        evicted = []
+        for chat_id, last_used in sorted_by_time[:evict_count]:
+            if chat_id in self.learners:
+                # 先保存再淘汰
+                self.learners[chat_id].save(self.model_save_path)
+                del self.learners[chat_id]
+                del self.learner_last_used[chat_id]
+                evicted.append(chat_id)
+
+        if evicted:
+            logger.info(f"StyleLearner LRU淘汰: 释放了 {len(evicted)} 个不活跃的学习器")
+
     def get_learner(self, chat_id: str, model_config: dict | None = None) -> StyleLearner:
         """
         获取或创建指定chat_id的学习器
@@ -463,7 +494,13 @@ class StyleLearnerManager:
         Returns:
             StyleLearner实例
         """
+        # 🔧 更新最后使用时间
+        self.learner_last_used[chat_id] = time.time()
+
         if chat_id not in self.learners:
+            # 🔧 检查是否需要淘汰
+            self._evict_if_needed()
+
             # 创建新的学习器
             learner = StyleLearner(chat_id, model_config)
 

src/chat/interest_system/bot_interest_manager.py

@@ -246,7 +246,11 @@ class BotInterestManager:
             raise
 
     async def _call_llm_for_interest_generation(self, prompt: str) -> str | None:
-        """调用LLM生成兴趣标签"""
+        """调用LLM生成兴趣标签
+        
+        注意：此方法会临时增加 API 超时时间，以确保初始化阶段的人设标签生成
+        不会因用户配置的较短超时而失败。
+        """
         try:
             logger.debug("配置LLM客户端...")
 
@@ -267,14 +271,42 @@ class BotInterestManager:
             # 使用replyer模型配置
             replyer_config = model_config.model_task_config.replyer
 
-            # 调用LLM API
-            success, response, reasoning_content, model_name = await llm_api.generate_with_model(
-                prompt=full_prompt,
-                model_config=replyer_config,
-                request_type="interest_generation",
-                temperature=0.7,
-                max_tokens=2000,
-            )
+            # 🔧 临时增加超时时间，避免初始化阶段因超时失败
+            # 人设标签生成需要较长时间（15-25个标签的JSON），使用更长的超时
+            INIT_TIMEOUT = 180  # 初始化阶段使用 180 秒超时
+            original_timeouts: dict[str, int] = {}
+            
+            try:
+                # 保存并修改所有相关模型的 API provider 超时设置
+                for model_name in replyer_config.model_list:
+                    try:
+                        model_info = model_config.get_model_info(model_name)
+                        provider = model_config.get_provider(model_info.api_provider)
+                        original_timeouts[provider.name] = provider.timeout
+                        if provider.timeout < INIT_TIMEOUT:
+                            logger.debug(f"⏱️ 临时增加 API provider '{provider.name}' 超时: {provider.timeout}s → {INIT_TIMEOUT}s")
+                            provider.timeout = INIT_TIMEOUT
+                    except Exception as e:
+                        logger.warning(f"⚠️ 无法修改模型 '{model_name}' 的超时设置: {e}")
+                
+                # 调用LLM API
+                success, response, reasoning_content, model_name = await llm_api.generate_with_model(
+                    prompt=full_prompt,
+                    model_config=replyer_config,
+                    request_type="interest_generation",
+                    temperature=0.7,
+                    max_tokens=2000,
+                )
+            finally:
+                # 🔧 恢复原始超时设置
+                for provider_name, original_timeout in original_timeouts.items():
+                    try:
+                        provider = model_config.get_provider(provider_name)
+                        if provider.timeout != original_timeout:
+                            logger.debug(f"⏱️ 恢复 API provider '{provider_name}' 超时: {provider.timeout}s → {original_timeout}s")
+                            provider.timeout = original_timeout
+                    except Exception as e:
+                        logger.warning(f"⚠️ 无法恢复 provider '{provider_name}' 的超时设置: {e}")
 
             if success and response:
                 # 直接返回原始响应，后续使用统一的 JSON 解析工具

src/chat/message_manager/distribution_manager.py

@@ -321,11 +321,14 @@ class StreamLoopManager:
                         # 🔒 并发保护：如果 Chatter 正在处理中，跳过本轮
                         # 这可能发生在：1) 打断后重启循环 2) 处理时间超过轮询间隔
                         if context.is_chatter_processing:
-                            logger.debug(f"🔒 [流工作器] stream={stream_id[:8]}, Chatter正在处理中，跳过本轮")
-                            # 不打印"开始处理"日志，直接进入下一轮等待
-                            # 使用较短的等待时间，等待当前处理完成
-                            await asyncio.sleep(1.0)
-                            continue
+                            if self._recover_stale_chatter_state(stream_id, context):
+                                logger.warning(f"🔄 [流工作器] stream={stream_id[:8]}, 处理标志疑似残留，已尝试自动修复")
+                            else:
+                                logger.debug(f"🔒 [流工作器] stream={stream_id[:8]}, Chatter正在处理中，跳过本轮")
+                                # 不打印"开始处理"日志，直接进入下一轮等待
+                                # 使用较短的等待时间，等待当前处理完成
+                                await asyncio.sleep(1.0)
+                                continue
                         
                         if force_dispatch:
                             logger.info(f"⚡ [流工作器] stream={stream_id[:8]}, 任务ID={task_id}, 未读消息 {unread_count} 条，触发强制分发")
@@ -529,6 +532,21 @@ class StreamLoopManager:
                 name=f"chatter_process_{stream_id}"
             )
 
+            # 记录任务句柄，便于后续检测/自愈
+            context.processing_task = chatter_task
+
+            def _cleanup_processing_flag(task: asyncio.Task) -> None:
+                try:
+                    context.processing_task = None
+                    if context.is_chatter_processing:
+                        context.is_chatter_processing = False
+                        self._set_stream_processing_status(stream_id, False)
+                        logger.debug(f"🔄 [并发保护] stream={stream_id[:8]}, chatter任务结束自动清理处理标志")
+                except Exception as callback_error:
+                    logger.debug(f"清理chatter处理标志失败: {callback_error}")
+
+            chatter_task.add_done_callback(_cleanup_processing_flag)
+
             # 等待 chatter 任务完成
             results = await chatter_task
             success = results.get("success", False)
@@ -550,6 +568,7 @@ class StreamLoopManager:
         finally:
             # 清除 Chatter 处理标志
             context.is_chatter_processing = False
+            context.processing_task = None
             logger.debug(f"清除 Chatter 处理标志: {stream_id}")
 
             # 无论成功或失败，都要设置处理状态为未处理
@@ -759,6 +778,35 @@ class StreamLoopManager:
             logger.debug(f"流 {stream_id} 使用默认间隔: {base_interval:.2f}s ({e})")
             return base_interval
 
+    def _recover_stale_chatter_state(self, stream_id: str, context: "StreamContext") -> bool:
+        """
+        检测并修复 Chatter 处理标志的假死状态。
+
+        返回 True 表示已发现并修复了异常状态；False 表示未发现异常。
+        """
+        try:
+            processing_task = getattr(context, "processing_task", None)
+
+            # 标志为 True 但没有任务句柄，直接修复
+            if processing_task is None:
+                context.is_chatter_processing = False
+                self._set_stream_processing_status(stream_id, False)
+                logger.warning(f"🛠️ [自愈] stream={stream_id[:8]}, 发现无任务但标志为真，已重置")
+                return True
+
+            # 标志为 True 但任务已经结束/被取消
+            if processing_task.done():
+                context.is_chatter_processing = False
+                context.processing_task = None
+                self._set_stream_processing_status(stream_id, False)
+                logger.warning(f"🛠️ [自愈] stream={stream_id[:8]}, 任务已结束但标志未清，已重置")
+                return True
+
+            return False
+        except Exception as e:
+            logger.debug(f"检测 Chatter 状态异常失败: stream={stream_id}, error={e}")
+            return False
+
     def get_queue_status(self) -> dict[str, Any]:
         """获取队列状态
 

src/chat/message_receive/message_handler.py

@@ -470,8 +470,8 @@ class MessageHandler:
 
             # 过滤检查
             raw_text = message.display_message or message.processed_plain_text or ""
-            if _check_ban_words(processed_text, chat, user_info) or _check_ban_regex(
-                raw_text, chat, user_info
+            if _check_ban_words(processed_text, chat, message.user_info) or _check_ban_regex(
+                raw_text, chat, message.user_info
             ):
                 return None
 

src/chat/utils/chat_message_builder.py

@@ -224,10 +224,12 @@ async def get_raw_msg_by_timestamp_with_chat_inclusive(
     limit: int = 0,
     limit_mode: str = "latest",
     filter_bot=False,
+    filter_meaningless=False,
 ) -> list[dict[str, Any]]:
     """获取在特定聊天从指定时间戳到指定时间戳的消息（包含边界），按时间升序排序，返回消息列表
     limit: 限制返回的消息数量，0为不限制
     limit_mode: 当 limit > 0 时生效。 'earliest' 表示获取最早的记录， 'latest' 表示获取最新的记录。默认为 'latest'。
+    filter_meaningless: 是否过滤无意义消息（表情包、通知等）。用于表达学习等场景。
     """
     filter_query = {"chat_id": chat_id, "time": {"$gte": timestamp_start, "$lte": timestamp_end}}
     # 只有当 limit 为 0 时才应用外部 sort
@@ -235,7 +237,12 @@ async def get_raw_msg_by_timestamp_with_chat_inclusive(
     # 直接将 limit_mode 传递给 find_messages
 
     return await find_messages(
-        message_filter=filter_query, sort=sort_order, limit=limit, limit_mode=limit_mode, filter_bot=filter_bot
+        message_filter=filter_query,
+        sort=sort_order,
+        limit=limit,
+        limit_mode=limit_mode,
+        filter_bot=filter_bot,
+        filter_meaningless=filter_meaningless,
     )
 
 
@@ -1114,10 +1121,14 @@ async def build_readable_messages(
         return "".join(result_parts)
 
 
-async def build_anonymous_messages(messages: list[dict[str, Any]]) -> str:
+async def build_anonymous_messages(messages: list[dict[str, Any]], filter_for_learning: bool = False) -> str:
     """
     构建匿名可读消息，将不同人的名称转为唯一占位符（A、B、C...），bot自己用SELF。
     处理 回复<aaa:bbb> 和 @<aaa:bbb> 字段，将bbb映射为匿名占位符。
+    
+    Args:
+        messages: 消息列表
+        filter_for_learning: 是否为表达学习场景进行额外过滤（过滤掉纯回复、纯@、纯图片等无意义内容）
     """
     assert global_config is not None
     if not messages:
@@ -1151,6 +1162,52 @@ async def build_anonymous_messages(messages: list[dict[str, Any]]) -> str:
             person_map[person_id] = chr(current_char)
             current_char += 1
         return person_map[person_id]
+    
+    def is_meaningless_content(content: str, msg: dict) -> bool:
+        """
+        判断消息内容是否无意义（用于表达学习过滤）
+        """
+        if not content or not content.strip():
+            return True
+        
+        stripped = content.strip()
+        
+        # 检查消息标记字段
+        if msg.get("is_emoji", False):
+            return True
+        if msg.get("is_notify", False):
+            return True
+        if msg.get("is_public_notice", False):
+            return True
+        if msg.get("is_command", False):
+            return True
+        
+        # 🔥 检查纯回复消息（只有[回复<xxx>]没有其他内容）
+        reply_pattern = r"^\s*\[回复[^\]]*\]\s*$"
+        if re.match(reply_pattern, stripped):
+            return True
+        
+        # 🔥 检查纯@消息（只有@xxx没有其他内容）
+        at_pattern = r"^\s*(@[^\s]+\s*)+$"
+        if re.match(at_pattern, stripped):
+            return True
+        
+        # 🔥 检查纯图片消息
+        image_pattern = r"^\s*(\[图片\]|\[动画表情\]|\[表情\]|\[picid:[^\]]+\])\s*$"
+        if re.match(image_pattern, stripped):
+            return True
+        
+        # 🔥 移除回复标记、@标记、图片标记后检查是否还有实质内容
+        clean_content = re.sub(r"\[回复[^\]]*\]", "", stripped)
+        clean_content = re.sub(r"@[^\s]+", "", clean_content)
+        clean_content = re.sub(r"\[图片\]|\[动画表情\]|\[表情\]|\[picid:[^\]]+\]", "", clean_content)
+        clean_content = clean_content.strip()
+        
+        # 如果移除后内容太短（少于2个字符），认为无意义
+        if len(clean_content) < 2:
+            return True
+        
+        return False
 
     for msg in messages:
         try:
@@ -1170,6 +1227,10 @@ async def build_anonymous_messages(messages: list[dict[str, Any]]) -> str:
 
             # For anonymous messages, we just replace with a placeholder.
             content = re.sub(r"\[picid:([^\]]+)\]", "[图片]", content)
+            
+            # 🔥 表达学习场景：过滤无意义消息
+            if filter_for_learning and is_meaningless_content(content, msg):
+                continue
 
             # if not all([platform, user_id, timestamp is not None]):
             # continue

src/chat/utils/utils_image.py

@@ -168,15 +168,22 @@ class ImageManager:
             image_bytes = base64.b64decode(image_base64)
             image_hash = hashlib.md5(image_bytes).hexdigest()
 
+            # 如果缓存命中，可以提前释放 image_bytes
+            # 但如果需要保存表情包，则需要保留 image_bytes
+
             # 2. 优先查询已注册表情的缓存（Emoji表）
             if full_description := await emoji_manager.get_emoji_description_by_hash(image_hash):
                 logger.info("[缓存命中] 使用已注册表情包(Emoji表)的完整描述")
+                del image_bytes  # 缓存命中，不再需要
+                del image_base64
                 refined_part = full_description.split(" Keywords:")[0]
                 return f"[表情包：{refined_part}]"
 
             # 3. 查询通用图片描述缓存（ImageDescriptions表）
             if cached_description := await self._get_description_from_db(image_hash, "emoji"):
                 logger.info("[缓存命中] 使用通用图片缓存(ImageDescriptions表)中的描述")
+                del image_bytes  # 缓存命中，不再需要
+                del image_base64
                 refined_part = cached_description.split(" Keywords:")[0]
                 return f"[表情包：{refined_part}]"
 
@@ -209,7 +216,11 @@ class ImageManager:
             await self._save_description_to_db(image_hash, full_description, "emoji")
             logger.info(f"新生成的表情包描述已存入通用缓存 (Hash: {image_hash[:8]}...)")
 
-            # 6. 返回新生成的描述中用于显示的“精炼描述”部分
+            # 内存优化：处理完成后主动释放大型二进制数据
+            del image_bytes
+            del image_base64
+
+            # 6. 返回新生成的描述中用于显示的"精炼描述"部分
             refined_part = full_description.split(" Keywords:")[0]
             return f"[表情包：{refined_part}]"
 
@@ -248,11 +259,17 @@ class ImageManager:
                 existing_image = result.scalar()
                 if existing_image and existing_image.description:
                     logger.debug(f"[缓存命中] 使用Images表中的图片描述: {existing_image.description[:50]}...")
+                    # 缓存命中，释放 base64 和 image_bytes
+                    del image_bytes
+                    del image_base64
                     return f"[图片：{existing_image.description}]"
 
             # 3. 其次查询 ImageDescriptions 表缓存
             if cached_description := await self._get_description_from_db(image_hash, "image"):
                 logger.debug(f"[缓存命中] 使用ImageDescriptions表中的描述: {cached_description[:50]}...")
+                # 缓存命中，释放 base64 和 image_bytes
+                del image_bytes
+                del image_base64
                 return f"[图片：{cached_description}]"
 
             # 4. 如果都未命中，则同步调用VLM生成新描述
@@ -301,6 +318,10 @@ class ImageManager:
 
             logger.info(f"新生成的图片描述已存入缓存 (Hash: {image_hash[:8]}...)")
 
+            # 内存优化：处理完成后主动释放大型二进制数据
+            del image_bytes
+            del image_base64
+
             return f"[图片：{description}]"
 
         except Exception as e:

src/common/message_repository.py

@@ -38,6 +38,7 @@ async def find_messages(
     limit_mode: str = "latest",
     filter_bot=False,
     filter_command=False,
+    filter_meaningless=False,
 ) -> list[dict[str, Any]]:
     """
     根据提供的过滤器、排序和限制条件查找消息。
@@ -47,6 +48,7 @@ async def find_messages(
         sort: 排序条件列表，例如 [('time', 1)] (1 for asc, -1 for desc)。仅在 limit 为 0 时生效。
         limit: 返回的最大文档数，0表示不限制。
         limit_mode: 当 limit > 0 时生效。 'earliest' 表示获取最早的记录， 'latest' 表示获取最新的记录（结果仍按时间正序排列）。默认为 'latest'。
+        filter_meaningless: 是否过滤无意义消息（表情包、通知、纯回复等）。用于表达学习等场景。
 
     Returns:
         消息字典列表，如果出错则返回空列表。
@@ -95,6 +97,14 @@ async def find_messages(
             if filter_command:
                 query = query.where(not_(Messages.is_command))
 
+            # 🔥 过滤无意义消息（用于表达学习等场景）
+            if filter_meaningless:
+                # 排除：纯表情包、通知消息、公告消息、命令消息
+                query = query.where(not_(Messages.is_emoji))
+                query = query.where(not_(Messages.is_notify))
+                query = query.where(not_(Messages.is_public_notice))
+                query = query.where(not_(Messages.is_command))
+
             if limit > 0:
                 # 确保limit是正整数
                 limit = max(1, int(limit))

src/plugin_system/core/stream_tool_history.py

@@ -387,8 +387,36 @@ class StreamToolHistoryManager:
             return result
 
 
-# 全局管理器字典，按chat_id索引
+# 内存优化：全局管理器字典，按chat_id索引，添加 LRU 淘汰
 _stream_managers: dict[str, StreamToolHistoryManager] = {}
+_stream_managers_last_used: dict[str, float] = {}  # 记录最后使用时间
+_STREAM_MANAGERS_MAX_SIZE = 100  # 最大保留数量
+
+
+def _evict_old_stream_managers() -> None:
+    """内存优化：淘汰最久未使用的 stream manager"""
+    import time
+
+    if len(_stream_managers) < _STREAM_MANAGERS_MAX_SIZE:
+        return
+
+    # 按最后使用时间排序，淘汰最旧的 20%
+    evict_count = max(1, len(_stream_managers) // 5)
+    sorted_by_time = sorted(
+        _stream_managers_last_used.items(),
+        key=lambda x: x[1]
+    )
+
+    evicted = []
+    for chat_id, _ in sorted_by_time[:evict_count]:
+        if chat_id in _stream_managers:
+            del _stream_managers[chat_id]
+        if chat_id in _stream_managers_last_used:
+            del _stream_managers_last_used[chat_id]
+        evicted.append(chat_id)
+
+    if evicted:
+        logger.info(f"🔧 StreamToolHistoryManager LRU淘汰: 释放了 {len(evicted)} 个不活跃的管理器")
 
 
 def get_stream_tool_history_manager(chat_id: str) -> StreamToolHistoryManager:
@@ -400,7 +428,14 @@ def get_stream_tool_history_manager(chat_id: str) -> StreamToolHistoryManager:
     Returns:
         工具历史记录管理器实例
     """
+    import time
+
+    # 🔧 更新最后使用时间
+    _stream_managers_last_used[chat_id] = time.time()
+
     if chat_id not in _stream_managers:
+        # 🔧 检查是否需要淘汰
+        _evict_old_stream_managers()
         _stream_managers[chat_id] = StreamToolHistoryManager(chat_id)
     return _stream_managers[chat_id]
 
@@ -413,4 +448,6 @@ def cleanup_stream_manager(chat_id: str) -> None:
     """
     if chat_id in _stream_managers:
         del _stream_managers[chat_id]
-        logger.info(f"已清理聊天 {chat_id} 的工具历史记录管理器")
+    if chat_id in _stream_managers_last_used:
+        del _stream_managers_last_used[chat_id]
+    logger.info(f"已清理聊天 {chat_id} 的工具历史记录管理器")

src/plugin_system/services/relationship_service.py

@@ -14,11 +14,19 @@ logger = get_logger("relationship_service")
 
 
 class RelationshipService:
-    """用户关系分服务 - 独立于插件的数据库直接访问层"""
+    """用户关系分服务 - 独立于插件的数据库直接访问层
+    
+    内存优化：添加缓存大小限制和自动过期清理
+    """
+
+    # 🔧 缓存配置
+    CACHE_MAX_SIZE = 1000  # 最大缓存用户数
 
     def __init__(self):
         self._cache: dict[str, dict] = {}  # user_id -> {score, text, last_updated}
         self._cache_ttl = 300  # 缓存5分钟
+        self._last_cleanup = time.time()  # 上次清理时间
+        self._cleanup_interval = 60  # 每60秒清理一次过期条目
 
     async def get_user_relationship_score(self, user_id: str) -> float:
         """
@@ -162,6 +170,9 @@ class RelationshipService:
 
     def _get_from_cache(self, user_id: str) -> dict | None:
         """从缓存获取数据"""
+        # 🔧 触发定期清理
+        self._maybe_cleanup_expired()
+
         if user_id in self._cache:
             cached_data = self._cache[user_id]
             if time.time() - cached_data["last_updated"] < self._cache_ttl:
@@ -173,12 +184,46 @@ class RelationshipService:
 
     def _update_cache(self, user_id: str, score: float, text: str):
         """更新缓存"""
+        # 🔧 内存优化：检查缓存大小限制
+        if len(self._cache) >= self.CACHE_MAX_SIZE and user_id not in self._cache:
+            # 淘汰最旧的 10% 条目
+            self._evict_oldest_entries()
+
         self._cache[user_id] = {
             "score": score,
             "text": text,
             "last_updated": time.time()
         }
 
+    def _maybe_cleanup_expired(self):
+        """🔧 内存优化：定期清理过期条目"""
+        now = time.time()
+        if now - self._last_cleanup < self._cleanup_interval:
+            return
+
+        self._last_cleanup = now
+        expired_keys = []
+        for user_id, data in self._cache.items():
+            if now - data["last_updated"] >= self._cache_ttl:
+                expired_keys.append(user_id)
+
+        for key in expired_keys:
+            del self._cache[key]
+
+        if expired_keys:
+            logger.debug(f"🔧 relationship_service 清理了 {len(expired_keys)} 个过期缓存条目")
+
+    def _evict_oldest_entries(self):
+        """🔧 内存优化：淘汰最旧的条目"""
+        evict_count = max(1, len(self._cache) // 10)
+        sorted_entries = sorted(
+            self._cache.items(),
+            key=lambda x: x[1]["last_updated"]
+        )
+        for user_id, _ in sorted_entries[:evict_count]:
+            del self._cache[user_id]
+        logger.debug(f"🔧 relationship_service LRU淘汰了 {evict_count} 个缓存条目")
+
     async def _fetch_from_database(self, user_id: str) -> UserRelationships | None:
         """从数据库获取关系数据"""
         try:

src/plugins/built_in/kokoro_flow_chatter/chatter.py

@@ -325,6 +325,10 @@ class KokoroFlowChatter(BaseChatter):
         """
         if session.status == SessionStatus.WAITING:
             # 之前在等待
+            # 如果 max_wait_seconds <= 0，说明不是有效的等待状态，视为新消息
+            if session.waiting_config.max_wait_seconds <= 0:
+                return "new_message"
+            
             if session.waiting_config.is_timeout():
                 # 超时了才收到回复
                 return "reply_late"

src/plugins/built_in/kokoro_flow_chatter/prompt/builder.py

@@ -69,6 +69,9 @@ class PromptBuilder:
         # 1. 构建人设块
         persona_block = self._build_persona_block()
         
+        # 1.5. 构建安全互动准则块
+        safety_guidelines_block = self._build_safety_guidelines_block()
+        
         # 2. 使用 context_builder 获取关系、记忆、工具、表达习惯等
         context_data = await self._build_context_data(user_name, chat_stream, user_id)
         relation_block = context_data.get("relation_info", f"你与 {user_name} 还不太熟悉，这是早期的交流阶段。")
@@ -98,6 +101,7 @@ class PromptBuilder:
             PROMPT_NAMES["main"],
             user_name=user_name,
             persona_block=persona_block,
+            safety_guidelines_block=safety_guidelines_block,
             relation_block=relation_block,
             memory_block=memory_block or "（暂无相关记忆）",
             tool_info=tool_info or "（暂无工具信息）",
@@ -142,7 +146,10 @@ class PromptBuilder:
         # 1. 构建人设块
         persona_block = self._build_persona_block()
         
-        # 2. 使用 context_builder 获取关系、记忆、工具、表达习惯等
+        # 1.5. 构建安全互动准则块
+        safety_guidelines_block = self._build_safety_guidelines_block()
+        
+        # 2. 使用 context_builder 获取关系、记忆、表达习惯等
         context_data = await self._build_context_data(user_name, chat_stream, user_id)
         relation_block = context_data.get("relation_info", f"你与 {user_name} 还不太熟悉，这是早期的交流阶段。")
         memory_block = context_data.get("memory_block", "")
@@ -170,6 +177,7 @@ class PromptBuilder:
             PROMPT_NAMES["replyer"],
             user_name=user_name,
             persona_block=persona_block,
+            safety_guidelines_block=safety_guidelines_block,
             relation_block=relation_block,
             memory_block=memory_block or "（暂无相关记忆）",
             tool_info=tool_info or "（暂无工具信息）",
@@ -202,6 +210,24 @@ class PromptBuilder:
         
         return "\n\n".join(parts) if parts else "你是一个温暖、真诚的人。"
     
+    def _build_safety_guidelines_block(self) -> str:
+        """
+        构建安全互动准则块
+        
+        从配置中读取 safety_guidelines，构建成提示词格式
+        """
+        if global_config is None:
+            return ""
+        
+        safety_guidelines = global_config.personality.safety_guidelines
+        if not safety_guidelines:
+            return ""
+        
+        guidelines_text = "\n".join(f"{i + 1}. {line}" for i, line in enumerate(safety_guidelines))
+        return f"""在任何情况下，你都必须遵守以下由你的设定者为你定义的原则：
+{guidelines_text}
+如果遇到违反上述原则的请求，请在保持你核心人设的同时，以合适的方式进行回应。"""
+    
     def _build_combined_expression_block(self, learned_habits: str) -> str:
         """
         构建合并后的表达习惯块
@@ -237,11 +263,13 @@ class PromptBuilder:
         user_name: str,
         chat_stream: Optional["ChatStream"],
         user_id: Optional[str] = None,
+        session: Optional[KokoroSession] = None,
+        situation_type: str = "new_message",
     ) -> dict[str, str]:
         """
         使用 KFCContextBuilder 构建完整的上下文数据
         
-        包括：关系信息、记忆、工具、表达习惯等
+        包括：关系信息、记忆、表达习惯等
         """
         if not chat_stream:
             return {
@@ -259,12 +287,13 @@ class PromptBuilder:
             
             builder = self._context_builder(chat_stream)
             
-            # 获取最近的消息作为 target_message（用于记忆检索）
-            target_message = ""
-            if chat_stream.context:
-                unread = chat_stream.context.get_unread_messages()
-                if unread:
-                    target_message = unread[-1].processed_plain_text or unread[-1].display_message or ""
+            # 获取用于记忆检索的查询文本
+            target_message = await self._get_memory_search_query(
+                chat_stream=chat_stream,
+                session=session,
+                situation_type=situation_type,
+                user_name=user_name,
+            )
             
             context_data = await builder.build_all_context(
                 sender_name=user_name,
@@ -284,6 +313,113 @@ class PromptBuilder:
                 "expression_habits": "",
             }
     
+    async def _get_memory_search_query(
+        self,
+        chat_stream: Optional["ChatStream"],
+        session: Optional[KokoroSession],
+        situation_type: str,
+        user_name: str,
+    ) -> str:
+        """
+        根据场景类型获取合适的记忆搜索查询文本
+        
+        策略：
+        1. 优先使用未读消息（new_message/reply_in_time/reply_late）
+        2. 如果没有未读消息（timeout/proactive），使用最近的历史消息
+        3. 如果历史消息也为空，从 session 的 mental_log 中提取
+        4. 最后回退到用户名作为查询
+        
+        Args:
+            chat_stream: 聊天流对象
+            session: KokoroSession 会话对象
+            situation_type: 情况类型
+            user_name: 用户名称
+            
+        Returns:
+            用于记忆搜索的查询文本
+        """
+        target_message = ""
+        
+        # 策略1: 优先从未读消息获取（适用于 new_message/reply_in_time/reply_late）
+        if chat_stream and chat_stream.context:
+            unread = chat_stream.context.get_unread_messages()
+            if unread:
+                target_message = unread[-1].processed_plain_text or unread[-1].display_message or ""
+                if target_message:
+                    logger.debug(f"[记忆搜索] 使用未读消息作为查询: {target_message[:50]}...")
+                    return target_message
+        
+        # 策略2: 从最近的历史消息获取（适用于 timeout/proactive）
+        if chat_stream and chat_stream.context:
+            history_messages = chat_stream.context.history_messages
+            if history_messages:
+                # 获取最近的几条非通知消息，组合成查询
+                recent_texts = []
+                for msg in reversed(history_messages[-5:]):
+                    content = getattr(msg, "processed_plain_text", "") or getattr(msg, "display_message", "")
+                    if content and not getattr(msg, "is_notify", False):
+                        recent_texts.append(content)
+                        if len(recent_texts) >= 3:
+                            break
+                
+                if recent_texts:
+                    target_message = " ".join(reversed(recent_texts))
+                    logger.debug(f"[记忆搜索] 使用历史消息作为查询 (situation={situation_type}): {target_message[:80]}...")
+                    return target_message
+        
+        # 策略3: 从 session 的 mental_log 中提取（超时/主动思考场景的最后手段）
+        if session and situation_type in ("timeout", "proactive"):
+            entries = session.get_recent_entries(limit=10)
+            recent_texts = []
+            
+            for entry in reversed(entries):
+                # 从用户消息中提取
+                if entry.event_type == EventType.USER_MESSAGE and entry.content:
+                    recent_texts.append(entry.content)
+                # 从 bot 的预期反应中提取（可能包含相关话题）
+                elif entry.event_type == EventType.BOT_PLANNING and entry.expected_reaction:
+                    recent_texts.append(entry.expected_reaction)
+                
+                if len(recent_texts) >= 3:
+                    break
+            
+            if recent_texts:
+                target_message = " ".join(reversed(recent_texts))
+                logger.debug(f"[记忆搜索] 使用 mental_log 作为查询 (situation={situation_type}): {target_message[:80]}...")
+                return target_message
+        
+        # 策略4: 最后回退 - 使用用户名 + 场景描述
+        if situation_type == "timeout":
+            target_message = f"与 {user_name} 的对话 等待回复"
+        elif situation_type == "proactive":
+            target_message = f"与 {user_name} 的对话 主动发起聊天"
+        else:
+            target_message = f"与 {user_name} 的对话"
+        
+        logger.debug(f"[记忆搜索] 使用回退查询 (situation={situation_type}): {target_message}")
+        return target_message
+    
+    def _get_latest_user_message(self, session: Optional[KokoroSession]) -> str:
+        """
+        获取最新的用户消息内容
+        
+        Args:
+            session: KokoroSession 会话对象
+            
+        Returns:
+            最新用户消息的内容，如果没有则返回提示文本
+        """
+        if not session:
+            return "（未知消息）"
+        
+        # 从 mental_log 中获取最新的用户消息
+        entries = session.get_recent_entries(limit=10)
+        for entry in reversed(entries):
+            if entry.event_type == EventType.USER_MESSAGE and entry.content:
+                return entry.content
+        
+        return "（消息内容不可用）"
+    
     async def _build_chat_history_block(
         self,
         chat_stream: Optional["ChatStream"],
@@ -505,32 +641,39 @@ class PromptBuilder:
                 situation_type = "new_message"
         
         if situation_type == "new_message":
+            # 获取最新消息内容
+            latest_message = self._get_latest_user_message(session)
             return await global_prompt_manager.format_prompt(
                 PROMPT_NAMES["situation_new_message"],
                 current_time=current_time,
                 user_name=user_name,
+                latest_message=latest_message,
             )
         
         elif situation_type == "reply_in_time":
             elapsed = session.waiting_config.get_elapsed_seconds()
             max_wait = session.waiting_config.max_wait_seconds
+            latest_message = self._get_latest_user_message(session)
             return await global_prompt_manager.format_prompt(
                 PROMPT_NAMES["situation_reply_in_time"],
                 current_time=current_time,
                 user_name=user_name,
                 elapsed_minutes=elapsed / 60,
                 max_wait_minutes=max_wait / 60,
+                latest_message=latest_message,
             )
         
         elif situation_type == "reply_late":
             elapsed = session.waiting_config.get_elapsed_seconds()
             max_wait = session.waiting_config.max_wait_seconds
+            latest_message = self._get_latest_user_message(session)
             return await global_prompt_manager.format_prompt(
                 PROMPT_NAMES["situation_reply_late"],
                 current_time=current_time,
                 user_name=user_name,
                 elapsed_minutes=elapsed / 60,
                 max_wait_minutes=max_wait / 60,
+                latest_message=latest_message,
             )
         
         elif situation_type == "timeout":
@@ -887,7 +1030,10 @@ class PromptBuilder:
         # 1. 构建人设块
         persona_block = self._build_persona_block()
         
-        # 2. 使用 context_builder 获取关系、记忆、工具、表达习惯等
+        # 1.5. 构建安全互动准则块
+        safety_guidelines_block = self._build_safety_guidelines_block()
+        
+        # 2. 使用 context_builder 获取关系、记忆、表达习惯等
         context_data = await self._build_context_data(user_name, chat_stream, user_id)
         relation_block = context_data.get("relation_info", f"你与 {user_name} 还不太熟悉，这是早期的交流阶段。")
         memory_block = context_data.get("memory_block", "")
@@ -916,6 +1062,7 @@ class PromptBuilder:
             PROMPT_NAMES["main"],
             user_name=user_name,
             persona_block=persona_block,
+            safety_guidelines_block=safety_guidelines_block,
             relation_block=relation_block,
             memory_block=memory_block or "（暂无相关记忆）",
             tool_info=tool_info or "（暂无工具信息）",

src/plugins/built_in/kokoro_flow_chatter/prompt/prompts.py

@@ -17,6 +17,9 @@ kfc_MAIN_PROMPT = Prompt(
 ## 人设
 {persona_block}
 
+## 安全互动准则
+{safety_guidelines_block}
+
 ## 你与 {user_name} 的关系
 {relation_block}
 
@@ -88,7 +91,9 @@ kfc_SITUATION_NEW_MESSAGE = Prompt(
     name="kfc_situation_new_message",
     template="""现在是 {current_time}。
 
-{user_name} 刚刚给你发了消息。这是一次新的对话发起（不是对你之前消息的回复）。
+{user_name} 刚刚给你发了消息：「{latest_message}」
+
+这是一次新的对话发起（不是对你之前消息的回复）。
 
 请决定你要怎么回应。你可以：
 - 发送文字消息回复
@@ -103,7 +108,7 @@ kfc_SITUATION_REPLY_IN_TIME = Prompt(
 
 你之前发了消息后一直在等 {user_name} 的回复。
 等了大约 {elapsed_minutes:.1f} 分钟（你原本打算最多等 {max_wait_minutes:.1f} 分钟）。
-现在 {user_name} 回复了！
+现在 {user_name} 回复了：「{latest_message}」
 
 请决定你接下来要怎么回应。""",
 )
@@ -114,7 +119,7 @@ kfc_SITUATION_REPLY_LATE = Prompt(
 
 你之前发了消息后在等 {user_name} 的回复。
 你原本打算最多等 {max_wait_minutes:.1f} 分钟，但实际等了 {elapsed_minutes:.1f} 分钟才收到回复。
-虽然有点迟，但 {user_name} 终于回复了。
+虽然有点迟，但 {user_name} 终于回复了：「{latest_message}」
 
 请决定你接下来要怎么回应。（可以选择轻轻抱怨一下迟到，也可以装作没在意）""",
 )
@@ -275,6 +280,9 @@ kfc_REPLYER_PROMPT = Prompt(
 ## 人设
 {persona_block}
 
+## 安全互动准则
+{safety_guidelines_block}
+
 ## 你与 {user_name} 的关系
 {relation_block}
 

src/plugins/built_in/napcat_adapter/plugin.py

@@ -176,6 +176,7 @@ class NapcatAdapter(BaseAdapter):
             # 消息事件
             if post_type == "message":
                 return await self.message_handler.handle_raw_message(raw)  # type: ignore[return-value]
+            
             # 通知事件
             elif post_type == "notice":
                 return await self.notice_handler.handle_notice(raw)  # type: ignore[return-value]

src/plugins/built_in/napcat_adapter/src/handlers/to_core/message_handler.py

@@ -60,7 +60,7 @@ class MessageHandler:
         user_id = str(sender_info.get("user_id", ""))
 
         # 检查全局封禁用户列表
-        ban_user_ids = features_config.get("ban_user_id", [])
+        ban_user_ids = [str(item) for item in features_config.get("ban_user_id", [])]
         if user_id in ban_user_ids:
             logger.debug(f"用户 {user_id} 在全局封禁列表中，消息被过滤")
             return False

src/plugins/built_in/social_toolkit_plugin/plugin.py

@@ -244,6 +244,14 @@ class SetEmojiLikeAction(BaseAction):
 
     async def execute(self) -> tuple[bool, str]:
         """执行设置表情回应的动作"""
+        # 检查是否在群聊中，该动作仅在群聊中有效
+        if not self.is_group:
+            logger.warning("set_emoji_like 动作仅在群聊中有效，当前为私聊场景")
+            await self.store_action_info(
+                action_prompt_display="贴表情失败: 该功能仅在群聊中可用", action_done=False
+            )
+            return False, "该功能仅在群聊中可用"
+
         message_id = None
         set_like = self.action_data.get("set", True)