From 310256e24d9955483cf38eb37280cb20f0fc0782 Mon Sep 17 00:00:00 2001 From: Windpicker-owo <3431391539@qq.com> Date: Wed, 12 Nov 2025 22:37:35 +0800 Subject: [PATCH] =?UTF-8?q?feat(attention):=20=E6=B7=BB=E5=8A=A0=E6=B3=A8?= =?UTF-8?q?=E6=84=8F=E5=8A=9B=E4=BC=98=E5=8C=96=E5=99=A8=E4=BB=A5=E5=A2=9E?= =?UTF-8?q?=E5=BC=BA=E6=8F=90=E7=A4=BA=E8=AF=8D=E5=A4=9A=E6=A0=B7=E6=80=A7?= =?UTF-8?q?=E5=92=8C=E9=98=B2=E6=AD=A2=E6=B3=A8=E6=84=8F=E5=8A=9B=E9=80=80?= =?UTF-8?q?=E5=8C=96=20refactor(prompt):=20=E4=BD=BF=E7=94=A8=20asyncio.ga?= =?UTF-8?q?ther=20=E6=9B=BF=E4=BB=A3=20as=5Fcompleted=20=E4=BB=A5=E6=8F=90?= =?UTF-8?q?=E5=8D=87=E5=B9=B6=E5=8F=91=E6=80=A7=E8=83=BD=20refactor(config?= =?UTF-8?q?):=20=E6=B7=BB=E5=8A=A0=E6=B3=A8=E6=84=8F=E5=8A=9B=E4=BC=98?= =?UTF-8?q?=E5=8C=96=E9=85=8D=E7=BD=AE=E9=80=89=E9=A1=B9=20refactor(prompt?= =?UTF-8?q?=5Fparams):=20=E5=A2=9E=E5=8A=A0=E6=B3=A8=E6=84=8F=E5=8A=9B?= =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=BC=80=E5=85=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/chat/replyer/default_generator.py | 9 +- src/chat/utils/attention_optimizer.py | 356 ++++++++++++++++++++++++++ src/chat/utils/prompt.py | 77 +++--- src/chat/utils/prompt_params.py | 1 + src/chat/utils/utils.py | 3 - src/config/config.py | 4 + src/config/official_configs.py | 10 + template/bot_config_template.toml | 8 +- 8 files changed, 420 insertions(+), 48 deletions(-) create mode 100644 src/chat/utils/attention_optimizer.py diff --git a/src/chat/replyer/default_generator.py b/src/chat/replyer/default_generator.py index 071e12379..be7fc3ea3 100644 --- a/src/chat/replyer/default_generator.py +++ b/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, diff --git a/src/chat/utils/attention_optimizer.py b/src/chat/utils/attention_optimizer.py new file mode 100644 index 000000000..770f17da9 --- /dev/null +++ b/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 diff --git a/src/chat/utils/prompt.py b/src/chat/utils/prompt.py index 966f577db..ae1a7a311 100644 --- a/src/chat/utils/prompt.py +++ b/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( - ( - asyncio.wait_for(task, timeout=task_timeout), - task_name, - i, - task_timeout, - ) - ) + # 创建带超时的任务 + timeout_task = asyncio.wait_for(task, timeout=task_timeout) + tasks_to_run.append(timeout_task) + task_info.append({"index": i, "name": task_name, "timeout": task_timeout}) else: logger.warning( f"任务{task_name}不是协程对象，类型: {type(task)}，跳过处理" ) results[i] = self._get_default_result_for_task(task_name) # type: ignore - # 并发执行任务，使用 as_completed 获得更好的性能 - for future in asyncio.as_completed( - [task_meta[0] for task_meta in task_with_meta] - ): - # 找到对应的任务元数据 - task_index = None - task_name = None - task_timeout = None + # 使用 gather 并发执行所有任务，return_exceptions=True 确保单个任务失败不影响其他任务 + if tasks_to_run: + task_results = await asyncio.gather(*tasks_to_run, return_exceptions=True) - for idx, (task, name, index, timeout) in enumerate(task_with_meta): - if task == future: - task_index = index - task_name = name - task_timeout = timeout - break + # 处理任务结果 + for i, result in enumerate(task_results): + info = task_info[i] + task_index = info["index"] + task_name = info["name"] + task_timeout = info["timeout"] - try: - result = await future - results[task_index] = result # type: ignore - logger.debug(f"构建任务{task_name}完成 ({task_timeout}s)") - except asyncio.TimeoutError: - logger.warning( - f"构建任务{task_name}超时 ({task_timeout}s)，使用默认值" - ) - results[task_index] = self._get_default_result_for_task(task_name) # type: ignore - except Exception as e: - logger.error(f"构建任务{task_name}失败: {e!s}") - results[task_index] = self._get_default_result_for_task(task_name) # type: ignore + if isinstance(result, asyncio.TimeoutError): + # 处理超时错误 + logger.warning( + f"构建任务{task_name}超时 ({task_timeout}s)，使用默认值" + ) + results[task_index] = self._get_default_result_for_task(task_name) + elif isinstance(result, Exception): + # 处理其他异常 + logger.error(f"构建任务{task_name}失败: {result!s}") + results[task_index] = self._get_default_result_for_task(task_name) + else: + # 成功完成 + results[task_index] = result + logger.debug(f"构建任务{task_name}完成 ({task_timeout}s)") # --- 步骤 3: 合并所有结果 --- context_data = {} diff --git a/src/chat/utils/prompt_params.py b/src/chat/utils/prompt_params.py index 4877de81e..47c6d46e4 100644 --- a/src/chat/utils/prompt_params.py +++ b/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 diff --git a/src/chat/utils/utils.py b/src/chat/utils/utils.py index b11875d04..97874b714 100644 --- a/src/chat/utils/utils.py +++ b/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( diff --git a/src/config/config.py b/src/config/config.py index 0e4248254..d44e6e490 100644 --- a/src/config/config.py +++ b/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="命令系统配置") diff --git a/src/config/official_configs.py b/src/config/official_configs.py index 32a036338..0f8d7da68 100644 --- a/src/config/official_configs.py +++ b/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): """回复后处理配置类""" diff --git a/template/bot_config_template.toml b/template/bot_config_template.toml index 9ed30625f..f913ecdd6 100644 --- a/template/bot_config_template.toml +++ b/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 # 是否启用回复后处理，包括错别字生成器，回复分割器