feat(attention): 添加注意力优化器以增强提示词多样性和防止注意力退化

refactor(prompt): 使用 asyncio.gather 替代 as_completed 以提升并发性能
refactor(config): 添加注意力优化配置选项
refactor(prompt_params): 增加注意力优化开关

src/chat/replyer/default_generator.py

@@ -89,12 +89,10 @@ def init_prompt():
 - {schedule_block}
 
 ## 历史记录
-### 📜 已读历史消息
 {read_history_prompt}
 
 {cross_context_block}
 
-### 📬 未读历史消息
 {unread_history_prompt}
 
 {notice_block}
@@ -175,12 +173,10 @@ If you need to use the search tool, please directly call the function "lpmm_sear
 {schedule_block}
 
 ## 历史记录
-### 📜 已读历史消息
 {read_history_prompt}
 
 {cross_context_block}
 
-### 📬 未读历史消息
 {unread_history_prompt}
 
 {notice_block}
@@ -858,7 +854,6 @@ class DefaultReplyer:
                 # 添加标题和格式化
                 notice_lines = []
                 notice_lines.append("## 📢 最近的系统通知")
-                notice_lines.append("")
                 notice_lines.append(notice_text)
                 notice_lines.append("")
 
@@ -989,7 +984,7 @@ class DefaultReplyer:
                 else:
                     unread_history_prompt = "暂无未读历史消息"
 
-                return read_history_prompt, unread_history_prompt
+                return f"### 📜 已读历史消息\n{read_history_prompt}", f"### 📬 未读历史消息\n{unread_history_prompt}"
             else:
                 # 回退到传统方法
                 return await self._fallback_build_chat_history_prompts(message_list_before_now, target_user_id, sender)
@@ -1091,7 +1086,7 @@ class DefaultReplyer:
         else:
             unread_history_prompt = "暂无未读历史消息"
 
-        return read_history_prompt, unread_history_prompt
+        return f"### 📜 已读历史消息\n{read_history_prompt}", f"### 📬 未读历史消息\n{unread_history_prompt}"
 
     async def build_prompt_reply_context(
         self,

src/chat/utils/attention_optimizer.py

@@ -0,0 +1,356 @@
+"""
+注意力优化器 - 防止提示词过度相似导致LLM注意力机制退化
+
+通过轻量级随机化技术，在保持语义不变的前提下增加提示词结构多样性，
+避免短时间内重复发送高度相似的提示词导致模型回复趋同。
+
+优化策略：
+1. 轻量级噪声：随机调整空白字符、换行数量
+2. 块重排：定义可交换的block组，随机调整顺序
+3. 语义变体：使用同义措辞替换固定模板文本
+"""
+
+import hashlib
+import random
+import re
+from typing import Any, Literal
+
+from src.common.logger import get_logger
+from src.config.config import global_config
+
+logger = get_logger("attention_optimizer")
+
+
+class AttentionOptimizer:
+    """提示词注意力优化器"""
+
+    # 可交换的block组定义（组内block可以随机排序）
+    # 每个组是一个列表，包含可以互换位置的block名称
+    SWAPPABLE_BLOCK_GROUPS = [
+        # 用户相关信息组（记忆、关系、表达习惯）
+        ["memory_block", "relation_info_block", "expression_habits_block"],
+        # 上下文增强组（工具、知识、跨群）
+        ["tool_info_block", "knowledge_prompt", "cross_context_block"],
+        # 元信息组（时间、身份、日程）
+        ["time_block", "identity_block", "schedule_block"],
+    ]
+
+    # 语义等价的文本替换模板
+    # 格式: {原始文本: [替换选项1, 替换选项2, ...]}
+    SEMANTIC_VARIANTS = {
+        "当前时间": ["当前时间", "现在是", "此时此刻", "时间"],
+        "最近的系统通知": ["最近的系统通知", "系统通知", "通知消息", "最新通知"],
+        "聊天历史": ["聊天历史", "对话记录", "历史消息", "之前的对话"],
+        "你的任务是": ["你的任务是", "请", "你需要", "你应当"],
+        "请注意": ["请注意", "注意", "请留意", "需要注意"],
+    }
+
+    def __init__(
+        self,
+        enable_noise: bool = True,
+        enable_semantic_variants: bool = False,
+        noise_strength: Literal["light", "medium", "heavy"] = "light",
+        cache_key_suffix: str = "",
+    ):
+        """
+        初始化注意力优化器
+
+        Args:
+            enable_noise: 是否启用轻量级噪声注入（空白字符调整）
+            enable_semantic_variants: 是否启用语义变体替换（实验性）
+            noise_strength: 噪声强度 (light/medium/heavy)
+            cache_key_suffix: 缓存键后缀，用于区分不同的优化配置
+        """
+        self.enable_noise = enable_noise
+        self.enable_semantic_variants = enable_semantic_variants
+        self.noise_strength = noise_strength
+        self.cache_key_suffix = cache_key_suffix
+
+        # 噪声强度配置
+        self.noise_config = {
+            "light": {"newline_range": (1, 2), "space_range": (0, 2), "indent_adjust": False},
+            "medium": {"newline_range": (1, 3), "space_range": (0, 4), "indent_adjust": True},
+            "heavy": {"newline_range": (1, 4), "space_range": (0, 6), "indent_adjust": True},
+        }
+
+
+
+    def optimize_prompt(self, prompt_text: str, context_data: dict[str, Any]) -> str:
+        """
+        优化提示词，增加结构多样性
+
+        Args:
+            prompt_text: 原始提示词文本
+            context_data: 上下文数据字典，包含各个block的内容
+
+        Returns:
+            优化后的提示词文本
+        """
+        try:
+            optimized = prompt_text
+
+            # 步骤2: 语义变体替换（如果启用）
+            if self.enable_semantic_variants:
+                optimized = self._apply_semantic_variants(optimized)
+
+            # 步骤3: 轻量级噪声注入（如果启用）
+            if self.enable_noise:
+                optimized = self._inject_noise(optimized)
+
+            # 计算变化率
+            change_rate = self._calculate_change_rate(prompt_text, optimized)
+            logger.debug(f"提示词优化完成，变化率: {change_rate:.2%}")
+
+            return optimized
+
+        except Exception as e:
+            logger.error(f"提示词优化失败: {e}", exc_info=True)
+            return prompt_text  # 失败时返回原始文本
+
+    def _shuffle_blocks(self, prompt_text: str, context_data: dict[str, Any]) -> str:
+        """
+        重排可交换的block组
+
+        Args:
+            prompt_text: 原始提示词
+            context_data: 包含各block内容的字典
+
+        Returns:
+            重排后的提示词
+        """
+        try:
+            # 对每个可交换组进行随机排序
+            shuffled_context = context_data.copy()
+
+            for group in self.SWAPPABLE_BLOCK_GROUPS:
+                # 过滤出实际存在且非空的block
+                existing_blocks = [
+                    block for block in group if block in context_data and context_data[block]
+                ]
+
+                if len(existing_blocks) > 1:
+                    # 随机打乱顺序
+                    shuffled = existing_blocks.copy()
+                    random.shuffle(shuffled)
+
+                    # 如果打乱后的顺序与原顺序不同，记录日志
+                    if shuffled != existing_blocks:
+                        logger.debug(f"重排block组: {existing_blocks} -> {shuffled}")
+
+                    # 注意：实际的重排需要在模板格式化之前进行
+                    # 这里只是演示逻辑，真正的实现需要在 _format_with_context 中处理
+
+            # 由于block重排需要在模板构建阶段进行，这里只返回原文本
+            # 真正的重排逻辑需要集成到 Prompt 类的 _format_with_context 方法中
+            return prompt_text
+
+        except Exception as e:
+            logger.error(f"Block重排失败: {e}", exc_info=True)
+            return prompt_text
+
+    def _apply_semantic_variants(self, text: str) -> str:
+        """
+        应用语义等价的文本替换
+
+        Args:
+            text: 原始文本
+
+        Returns:
+            替换后的文本
+        """
+        try:
+            result = text
+
+            for original, variants in self.SEMANTIC_VARIANTS.items():
+                if original in result:
+                    # 随机选择一个变体（包括原始文本）
+                    replacement = random.choice(variants)
+                    result = result.replace(original, replacement, 1)  # 只替换第一次出现
+
+            return result
+
+        except Exception as e:
+            logger.error(f"语义变体替换失败: {e}", exc_info=True)
+            return text
+
+    def _inject_noise(self, text: str) -> str:
+        """
+        注入轻量级噪声（空白字符调整）
+
+        Args:
+            text: 原始文本
+
+        Returns:
+            注入噪声后的文本
+        """
+        try:
+            config = self.noise_config[self.noise_strength]
+            result = text
+
+            # 1. 调整block之间的换行数量
+            result = self._adjust_newlines(result, config["newline_range"])
+
+            # 2. 在某些位置添加随机空格（保持可读性）
+            result = self._adjust_spaces(result, config["space_range"])
+
+            # 3. 调整缩进（仅在medium/heavy模式下）
+            if config["indent_adjust"]:
+                result = self._adjust_indentation(result)
+
+            return result
+
+        except Exception as e:
+            logger.error(f"噪声注入失败: {e}", exc_info=True)
+            return text
+
+    def _adjust_newlines(self, text: str, newline_range: tuple[int, int]) -> str:
+        """
+        调整连续换行的数量
+
+        Args:
+            text: 原始文本
+            newline_range: 换行数量范围 (min, max)
+
+        Returns:
+            调整后的文本
+        """
+        # 匹配连续的换行符
+        pattern = r"\n{2,}"
+
+        def replace_newlines(match):
+            # 随机选择新的换行数量
+            count = random.randint(*newline_range)
+            return "\n" * count
+
+        return re.sub(pattern, replace_newlines, text)
+
+    def _adjust_spaces(self, text: str, space_range: tuple[int, int]) -> str:
+        """
+        在某些位置添加随机空格
+
+        Args:
+            text: 原始文本
+            space_range: 空格数量范围 (min, max)
+
+        Returns:
+            调整后的文本
+        """
+        # 在行尾随机添加空格（不可见但会改变文本哈希）
+        lines = text.split("\n")
+        result_lines = []
+
+        for line in lines:
+            if line.strip() and random.random() < 0.3:  # 30%概率添加空格
+                spaces = " " * random.randint(*space_range)
+                result_lines.append(line + spaces)
+            else:
+                result_lines.append(line)
+
+        return "\n".join(result_lines)
+
+    def _adjust_indentation(self, text: str) -> str:
+        """
+        微调某些行的缩进（保持语义）
+
+        Args:
+            text: 原始文本
+
+        Returns:
+            调整后的文本
+        """
+        lines = text.split("\n")
+        result_lines = []
+
+        for line in lines:
+            # 检测列表项
+            list_match = re.match(r"^(\s*)([-*•])\s", line)
+            if list_match and random.random() < 0.5:
+                indent = list_match.group(1)
+                marker = list_match.group(2)
+                # 随机调整缩进（±2个空格）
+                adjust = random.choice([-2, 0, 2])
+                new_indent = " " * max(0, len(indent) + adjust)
+                new_line = line.replace(indent + marker, new_indent + marker, 1)
+                result_lines.append(new_line)
+            else:
+                result_lines.append(line)
+
+        return "\n".join(result_lines)
+
+    def _calculate_change_rate(self, original: str, optimized: str) -> float:
+        """
+        计算文本变化率
+
+        Args:
+            original: 原始文本
+            optimized: 优化后的文本
+
+        Returns:
+            变化率（0-1之间的浮点数）
+        """
+        if not original or not optimized:
+            return 0.0
+
+        # 使用简单的字符差异比率
+        diff_chars = sum(1 for a, b in zip(original, optimized) if a != b)
+        max_len = max(len(original), len(optimized))
+
+        return diff_chars / max_len if max_len > 0 else 0.0
+
+    def get_cache_key(self, prompt_text: str) -> str:
+        """
+        生成优化后提示词的缓存键
+
+        由于注意力优化会改变提示词内容，缓存键也需要相应调整
+
+        Args:
+            prompt_text: 提示词文本
+
+        Returns:
+            缓存键字符串
+        """
+        # 计算文本哈希
+        text_hash = hashlib.md5(prompt_text.encode()).hexdigest()[:8]
+
+        # 添加随机后缀，确保相似提示词有不同的缓存键
+        random_suffix = random.randint(1000, 9999)
+
+        return f"{text_hash}_{random_suffix}_{self.cache_key_suffix}"
+
+
+def get_attention_optimizer_from_config() -> AttentionOptimizer:
+    """
+    从全局配置创建注意力优化器实例
+
+    Returns:
+        配置好的 AttentionOptimizer 实例
+    """
+    # 从配置中读取设置（如果存在）
+    config = getattr(global_config, "attention_optimization", None)
+
+    if not config:
+        # 使用默认配置
+        return AttentionOptimizer(
+            enable_noise=True,
+            enable_semantic_variants=False,  # 实验性功能，默认关闭
+            noise_strength="light",
+        )
+
+    # config 是 Pydantic 模型对象，直接访问属性
+    return AttentionOptimizer(
+        enable_noise=config.enable_noise,
+        enable_semantic_variants=config.enable_semantic_variants,
+        noise_strength=config.noise_strength,
+    )
+
+
+# 全局单例
+_global_optimizer: AttentionOptimizer | None = None
+
+
+def get_attention_optimizer() -> AttentionOptimizer:
+    """获取全局注意力优化器实例"""
+    global _global_optimizer
+    if _global_optimizer is None:
+        _global_optimizer = get_attention_optimizer_from_config()
+    return _global_optimizer

src/chat/utils/prompt.py

@@ -375,6 +375,15 @@ class Prompt:
             # 这样做可以更早地组合模板，也使得`Prompt`类的职责更单一。
             result = main_formatted_prompt
 
+            # 步骤 4: 注意力优化（如果启用）
+            # 通过轻量级随机化避免提示词过度相似导致LLM注意力退化
+            if self.parameters.enable_attention_optimization:
+                from src.chat.utils.attention_optimizer import get_attention_optimizer
+
+                optimizer = get_attention_optimizer()
+                result = optimizer.optimize_prompt(result, context_data)
+                logger.debug("已应用注意力优化")
+
             total_time = time.time() - start_time
             logger.debug(
                 f"Prompt构建完成，模式: {self.parameters.prompt_mode}, 耗时: {total_time:.2f}s"
@@ -492,11 +501,12 @@ class Prompt:
                 "expression_habits": 10.0,
             }
 
-            # 使用 as_completed 并发执行任务，提供更好的性能和错误处理
+            # 使用 asyncio.gather 实现并发执行，提供更好的错误处理和性能
             results = [None] * len(tasks)  # 预分配结果列表，保持任务顺序
-            task_with_meta = []
+            tasks_to_run = []  # 存储带超时的任务
+            task_info = []  # 存储任务信息，用于结果处理
 
-            # 准备任务和元数据
+            # 准备任务并创建带超时的协程
             for i, task in enumerate(tasks):
                 task_name = task_names[i] if i < len(task_names) else f"task_{i}"
                 task_timeout = task_timeouts.get(
@@ -505,48 +515,41 @@ class Prompt:
 
                 # 检查任务是否为协程，非协程任务直接使用默认值
                 if asyncio.iscoroutine(task):
-                    task_with_meta.append(
+                    # 创建带超时的任务
-                        (
+                    timeout_task = asyncio.wait_for(task, timeout=task_timeout)
-                            asyncio.wait_for(task, timeout=task_timeout),
+                    tasks_to_run.append(timeout_task)
-                            task_name,
+                    task_info.append({"index": i, "name": task_name, "timeout": task_timeout})
-                            i,
-                            task_timeout,
-                        )
-                    )
                 else:
                     logger.warning(
                         f"任务{task_name}不是协程对象，类型: {type(task)}，跳过处理"
                     )
                     results[i] = self._get_default_result_for_task(task_name)  # type: ignore
 
-            # 并发执行任务，使用 as_completed 获得更好的性能
+            # 使用 gather 并发执行所有任务，return_exceptions=True 确保单个任务失败不影响其他任务
-            for future in asyncio.as_completed(
+            if tasks_to_run:
-                [task_meta[0] for task_meta in task_with_meta]
+                task_results = await asyncio.gather(*tasks_to_run, return_exceptions=True)
-            ):
-                # 找到对应的任务元数据
-                task_index = None
-                task_name = None
-                task_timeout = None
 
-                for idx, (task, name, index, timeout) in enumerate(task_with_meta):
+                # 处理任务结果
-                    if task == future:
+                for i, result in enumerate(task_results):
-                        task_index = index
+                    info = task_info[i]
-                        task_name = name
+                    task_index = info["index"]
-                        task_timeout = timeout
+                    task_name = info["name"]
-                        break
+                    task_timeout = info["timeout"]
 
-                try:
+                    if isinstance(result, asyncio.TimeoutError):
-                    result = await future
+                        # 处理超时错误
-                    results[task_index] = result  # type: ignore
+                        logger.warning(
-                    logger.debug(f"构建任务{task_name}完成 ({task_timeout}s)")
+                            f"构建任务{task_name}超时 ({task_timeout}s)，使用默认值"
-                except asyncio.TimeoutError:
+                        )
-                    logger.warning(
+                        results[task_index] = self._get_default_result_for_task(task_name)
-                        f"构建任务{task_name}超时 ({task_timeout}s)，使用默认值"
+                    elif isinstance(result, Exception):
-                    )
+                        # 处理其他异常
-                    results[task_index] = self._get_default_result_for_task(task_name)  # type: ignore
+                        logger.error(f"构建任务{task_name}失败: {result!s}")
-                except Exception as e:
+                        results[task_index] = self._get_default_result_for_task(task_name)
-                    logger.error(f"构建任务{task_name}失败: {e!s}")
+                    else:
-                    results[task_index] = self._get_default_result_for_task(task_name)  # type: ignore
+                        # 成功完成
+                        results[task_index] = result
+                        logger.debug(f"构建任务{task_name}完成 ({task_timeout}s)")
 
             # --- 步骤 3: 合并所有结果 ---
             context_data = {}

src/chat/utils/prompt_params.py

@@ -27,6 +27,7 @@ class PromptParameters:
     enable_relation: bool = True
     enable_cross_context: bool = True
     enable_knowledge: bool = True
+    enable_attention_optimization: bool = True  # 注意力优化开关
 
     # 性能控制
     max_context_messages: int = 50

src/chat/utils/utils.py

@@ -977,9 +977,6 @@ def filter_system_format_content(content: str | None) -> str:
     # [图片(描述生成失败)] 等错误格式
     cleaned_content = re.sub(r"\[图片\([^)]*\)\]", "", cleaned_content)
 
-    # 清理多余空格
-    cleaned_content = re.sub(r"\s+", " ", cleaned_content).strip()
-
     # 记录过滤操作
     if cleaned_content != original_content.strip():
         logger.info(

src/config/config.py

@@ -13,6 +13,7 @@ from src.common.logger import get_logger
 from src.config.config_base import ValidatedConfigBase
 from src.config.official_configs import (
     AffinityFlowConfig,
+    AttentionOptimizationConfig,
     BotConfig,
     ChatConfig,
     ChineseTypoConfig,
@@ -391,6 +392,9 @@ class Config(ValidatedConfigBase):
     tool: ToolConfig = Field(..., description="工具配置")
     debug: DebugConfig = Field(..., description="调试配置")
     custom_prompt: CustomPromptConfig = Field(..., description="自定义提示配置")
+    attention_optimization: AttentionOptimizationConfig = Field(
+        default_factory=lambda: AttentionOptimizationConfig(), description="注意力优化配置"
+    )
     voice: VoiceConfig = Field(..., description="语音配置")
     permission: PermissionConfig = Field(..., description="权限配置")
     command: CommandConfig = Field(..., description="命令系统配置")

src/config/official_configs.py

@@ -531,6 +531,16 @@ class CustomPromptConfig(ValidatedConfigBase):
     planner_custom_prompt_content: str = Field(default="", description="规划器自定义提示词内容")
 
 
+class AttentionOptimizationConfig(ValidatedConfigBase):
+    """注意力优化配置类 - 防止提示词过度相似导致LLM注意力退化"""
+
+    enable_noise: bool = Field(default=True, description="启用轻量级噪声注入（空白字符调整）")
+    enable_semantic_variants: bool = Field(default=False, description="启用语义变体替换（实验性功能）")
+    noise_strength: Literal["light", "medium", "heavy"] = Field(
+        default="light", description="噪声强度: light(轻量) | medium(中等) | heavy(强力)"
+    )
+
+
 class ResponsePostProcessConfig(ValidatedConfigBase):
     """回复后处理配置类"""
 

template/bot_config_template.toml

@@ -1,5 +1,5 @@
 [inner]
-version = "7.6.8"
+version = "7.6.9"
 
 #----以下是给开发人员阅读的，如果你只是部署了MoFox-Bot，不需要阅读----
 #如果你想要修改配置文件，请递增version的值
@@ -348,6 +348,12 @@ reaction = "请按照以下模板造句：[n]是这样的，xx只要xx就可以
 image_prompt = "请用中文描述这张图片的内容。如果有文字，请把文字描述概括出来，请留意其主题，直观感受，输出为一段平文本，最多30字，请注意不要分点，就输出一段文本"
 planner_custom_prompt_content = "" # 决策器自定义提示词内容，如果这里没有内容则不生效
 
+# 注意力优化配置 - 防止提示词过度相似导致LLM注意力退化
+[attention_optimization]
+enable_noise = true # 启用轻量级噪声注入（空白字符调整）
+enable_semantic_variants = false # 启用语义变体替换（实验性功能）
+noise_strength = "light" # 噪声强度: "light"(轻量) | "medium"(中等) | "heavy"(强力)，推荐使用light
+
 [response_post_process]
 enable_response_post_process = true # 是否启用回复后处理，包括错别字生成器，回复分割器