diff --git a/src/chat/utils/attention_optimizer.py b/src/chat/utils/attention_optimizer.py index e8210a685..27365177b 100644 --- a/src/chat/utils/attention_optimizer.py +++ b/src/chat/utils/attention_optimizer.py @@ -1,32 +1,24 @@ """ -注意力优化器 - 防止提示词过度相似导致LLM注意力机制退化 +注意力优化器 - 提示词块重排 -通过轻量级随机化技术，在保持语义不变的前提下增加提示词结构多样性， -避免短时间内重复发送高度相似的提示词导致模型回复趋同。 - -优化策略： -1. 轻量级噪声：随机调整空白字符、换行数量 -2. 块重排：定义可交换的block组，随机调整顺序 -3. 语义变体：使用同义措辞替换固定模板文本 +通过对可交换的block组进行随机排序，增加提示词结构多样性， +避免因固定的提示词结构导致模型注意力退化。 """ -import hashlib import random -import re -from typing import Any, ClassVar, Literal +from typing import Any, ClassVar from src.common.logger import get_logger -from src.config.config import global_config -logger = get_logger("attention_optimizer") +logger = get_logger("attention_optimizer_shuffle") -class AttentionOptimizer: - """提示词注意力优化器""" +class BlockShuffler: + """提示词Block重排器""" # 可交换的block组定义（组内block可以随机排序） # 每个组是一个列表，包含可以互换位置的block名称 - SWAPPABLE_BLOCK_GROUPS:ClassVar = [ + SWAPPABLE_BLOCK_GROUPS: ClassVar = [ # 用户相关信息组（记忆、关系、表达习惯） ["memory_block", "relation_info_block", "expression_habits_block"], # 上下文增强组（工具、知识、跨群） @@ -35,322 +27,53 @@ class AttentionOptimizer: ["time_block", "identity_block", "schedule_block"], ] - # 语义等价的文本替换模板 - # 格式: {原始文本: [替换选项1, 替换选项2, ...]} - SEMANTIC_VARIANTS:ClassVar = { - "当前时间": ["当前时间", "现在是", "此时此刻", "时间"], - "最近的系统通知": ["最近的系统通知", "系统通知", "通知消息", "最新通知"], - "聊天历史": ["聊天历史", "对话记录", "历史消息", "之前的对话"], - "你的任务是": ["你的任务是", "请", "你需要", "你应当"], - "请注意": ["请注意", "注意", "请留意", "需要注意"], - } - - def __init__( - self, - enable_noise: bool = True, - enable_semantic_variants: bool = False, - noise_strength: Literal["light", "medium", "heavy"] = "light", - cache_key_suffix: str = "", - ): + @staticmethod + def shuffle_prompt_blocks(prompt_template: str, context_data: dict[str, Any]) -> tuple[str, dict[str, Any]]: """ - 初始化注意力优化器 + 根据定义的SWAPPABLE_BLOCK_GROUPS，对上下文数据中的block进行随机重排， + 并返回可能已修改的prompt模板和重排后的上下文。 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的内容 + prompt_template (str): 原始的提示词模板. + context_data (dict[str, Any]): 包含各个block内容的上下文数据. Returns: - 优化后的提示词文本 + tuple[str, dict[str, Any]]: (可能被修改的模板, 重排后的上下文数据). """ 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() + + # 示例：假设模板中的占位符格式为 {block_name} + # 我们需要解析模板，找到可重排的组，并重新构建模板字符串。 + + # 注意：这是一个复杂的逻辑，通常需要一个简单的模板引擎或正则表达式来完成。 + # 为保持此函数职责单一，这里仅演示核心的重排逻辑， + # 完整的模板重建逻辑应在调用此函数的地方处理。 - for group in self.SWAPPABLE_BLOCK_GROUPS: - # 过滤出实际存在且非空的block + for group in BlockShuffler.SWAPPABLE_BLOCK_GROUPS: + # 过滤出在当前上下文中实际存在的、非空的block existing_blocks = [ - block for block in group if context_data.get(block) + 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) + random.shuffle(existing_blocks) + logger.debug(f"重排block组: {group} -> {existing_blocks}") + + # 这里的实现需要调用者根据 `existing_blocks` 的新顺序 + # 去动态地重新组织 `prompt_template` 字符串。 + # 例如，找到模板中与 `group` 相关的占位符部分，然后按新顺序替换它们。 - # 如果打乱后的顺序与原顺序不同，记录日志 - if shuffled != existing_blocks: - logger.debug(f"重排block组: {existing_blocks} -> {shuffled}") - - # 注意：实际的重排需要在模板格式化之前进行 - # 这里只是演示逻辑，真正的实现需要在 _format_with_context 中处理 - - # 由于block重排需要在模板构建阶段进行，这里只返回原文本 - # 真正的重排逻辑需要集成到 Prompt 类的 _format_with_context 方法中 - return prompt_text + # 在这个简化版本中，我们不修改模板，仅返回原始模板和（未被使用的）重排后上下文 + # 实际应用中，调用方需要根据重排结果修改模板 + return prompt_template, shuffled_context 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 + return prompt_template, context_data diff --git a/src/config/api_ada_configs.py b/src/config/api_ada_configs.py index 050edc5f6..157692919 100644 --- a/src/config/api_ada_configs.py +++ b/src/config/api_ada_configs.py @@ -71,9 +71,13 @@ class ModelInfo(ValidatedConfigBase): force_stream_mode: bool = Field(default=False, description="是否强制使用流式输出模式") extra_params: dict[str, Any] = Field(default_factory=dict, description="额外参数（用于API调用时的额外配置）") anti_truncation: bool = Field(default=False, description="是否启用反截断功能，防止模型输出被截断") - enable_content_obfuscation: bool = Field(default=False, description="是否启用内容混淆（用于特定场景下的内容处理）") - obfuscation_intensity: int = Field(default=1, ge=1, le=3, description="混淆强度（1-3级，数值越高混淆程度越强）") + enable_prompt_perturbation: bool = Field(default=False, description="是否启用提示词扰动（合并了内容混淆和注意力优化）") + perturbation_strength: Literal["light", "medium", "heavy"] = Field( + default="light", description="扰动强度（light/medium/heavy）" + ) + enable_semantic_variants: bool = Field(default=False, description="是否启用语义变体作为扰动策略") + prepend_noise_instruction: bool = Field(default=False, description="是否在提示词前部添加抗审查指令") @classmethod def validate_prices(cls, v): """验证价格必须为非负数""" diff --git a/src/llm_models/utils_model.py b/src/llm_models/utils_model.py index 07ddc8622..25e763161 100644 --- a/src/llm_models/utils_model.py +++ b/src/llm_models/utils_model.py @@ -26,7 +26,7 @@ import time from collections import namedtuple from collections.abc import Callable, Coroutine from enum import Enum -from typing import Any +from typing import Any, ClassVar, Literal from rich.traceback import install @@ -261,6 +261,137 @@ class _ModelSelector: self.model_usage[model_name] = stats._replace(penalty=stats.penalty + penalty_increment) +class _AttentionOptimizer: + """ + 通过轻量级随机化技术，在保持语义不变的前提下增加提示词结构多样性， + 避免短时间内重复发送高度相似的提示词导致模型回复趋同。 + """ + + # 语义等价的文本替换模板 + SEMANTIC_VARIANTS: ClassVar = { + "当前时间": ["当前时间", "现在是", "此时此刻", "时间"], + "最近的系统通知": ["最近的系统通知", "系统通知", "通知消息", "最新通知"], + "聊天历史": ["聊天历史", "对话记录", "历史消息", "之前的对话"], + "你的任务是": ["你的任务是", "请", "你需要", "你应当"], + "请注意": ["请注意", "注意", "请留意", "需要注意"], + } + + def __init__( + self, + enable_semantic_variants: bool, + noise_strength: Literal["light", "medium", "heavy"], + ): + """ + 初始化注意力优化器 + Args: + enable_semantic_variants: 是否启用语义变体替换 + noise_strength: 噪声强度 (light/medium/heavy) + """ + self.enable_semantic_variants = enable_semantic_variants + self.noise_strength = noise_strength + + # 噪声强度配置 + 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) -> str: + """优化提示词，增加结构多样性""" + try: + optimized = prompt_text + + if self.enable_semantic_variants: + optimized = self._apply_semantic_variants(optimized) + + optimized = self._inject_noise(optimized) + + change_rate = self._calculate_change_rate(prompt_text, optimized) + if change_rate > 0.001: # 仅在有实际变化时记录 + logger.debug(f"提示词注意力优化完成，变化率: {change_rate:.2%}") + + return optimized + + except Exception as e: + logger.error(f"提示词注意力优化失败: {e}", exc_info=True) + return prompt_text + + def _apply_semantic_variants(self, text: str) -> str: + """应用语义等价的文本替换""" + 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: + """注入轻量级噪声（空白字符调整）""" + try: + config = self.noise_config[self.noise_strength] + result = text + result = self._adjust_newlines(result, config["newline_range"]) + result = self._adjust_spaces(result, config["space_range"]) + 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: + """调整连续换行的数量""" + 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: + """在某些位置添加随机空格""" + lines = text.split("\n") + result_lines = [] + for line in lines: + if line.strip() and random.random() < 0.3: + 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: + """微调某些行的缩进（保持语义）""" + 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) + 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: + """计算文本变化率""" + 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 + + class _PromptProcessor: """封装所有与提示词和响应内容的预处理和后处理逻辑。""" @@ -292,29 +423,39 @@ class _PromptProcessor: self, prompt: str, model_info: ModelInfo, task_name: str ) -> str: """ - 为请求准备最终的提示词。 - - 此方法会根据API提供商和模型配置，对原始提示词应用内容混淆和反截断指令， - 生成最终发送给模型的完整提示内容。 - - Args: - prompt (str): 原始的用户提示词。 - model_info (ModelInfo): 目标模型的信息。 - api_provider (APIProvider): API提供商的配置。 - task_name (str): 当前任务的名称，用于日志记录。 - - Returns: - str: 处理后的、可以直接发送给模型的完整提示词。 + 为请求准备最终的提示词,应用各种扰动和指令。 """ - # 步骤1: 根据API提供商的配置应用内容混淆 - processed_prompt = await self._apply_content_obfuscation(prompt, model_info) + final_prompt_parts = [] + user_prompt = prompt - # 步骤2: 检查模型是否需要注入反截断指令 + # 步骤 A: (可选) 添加抗审查指令 + if getattr(model_info, "prepend_noise_instruction", False): + final_prompt_parts.append(self.noise_instruction) + + # 步骤 B: (可选) 应用提示词扰动 + if getattr(model_info, "enable_prompt_perturbation", False): + logger.info(f"为模型 '{model_info.name}' 启用提示词扰动功能。") + + # B.1 注意力优化 (空白字符 + 语义变体) + optimizer = _AttentionOptimizer( + enable_semantic_variants=getattr(model_info, "enable_semantic_variants", False), + noise_strength=getattr(model_info, "perturbation_strength", "light"), + ) + user_prompt = optimizer.optimize_prompt(user_prompt) + + # B.2 内容混淆 (注入随机噪音) + user_prompt = await self._inject_random_noise( + user_prompt, getattr(model_info, "perturbation_strength", "light") + ) + + final_prompt_parts.append(user_prompt) + + # 步骤 C: (可选) 添加反截断指令 if getattr(model_info, "use_anti_truncation", False): - processed_prompt += self.anti_truncation_instruction + final_prompt_parts.append(self.anti_truncation_instruction) logger.info(f"模型 '{model_info.name}' (任务: '{task_name}') 已启用反截断功能。") - return processed_prompt + return "\n\n".join(final_prompt_parts) async def process_response(self, content: str, use_anti_truncation: bool) -> tuple[str, str, bool]: """ @@ -331,51 +472,16 @@ class _PromptProcessor: else: is_truncated = True return content, reasoning, is_truncated - - async def _apply_content_obfuscation(self, text: str, model_info: ModelInfo) -> str: - """ - 根据API提供商的配置对文本进行内容混淆。 - - 如果提供商配置中启用了内容混淆，此方法会在文本前部加入抗审查指令， - 并在文本中注入随机噪音，以降低内容被审查或修改的风险。 - - Args: - text (str): 原始文本内容。 - api_provider (APIProvider): API提供商的配置。 - - Returns: - str: 经过混淆处理的文本。 - """ - # 检查当前API提供商是否启用了内容混淆功能 - if not model_info.enable_content_obfuscation or False: - return text - - # 获取混淆强度，默认为1 - intensity = model_info.obfuscation_intensity or 1 - logger.info(f"为模型 '{model_info.name}' 启用内容混淆，强度级别: {intensity}") - - # 将抗审查指令和原始文本拼接 - processed_text = self.noise_instruction + "\n\n" + text - - # 在拼接后的文本中注入随机噪音 - return await self._inject_random_noise(processed_text, intensity) - + @staticmethod - async def _inject_random_noise(text: str, intensity: int) -> str: + async def _inject_random_noise(text: str, strength: str) -> str: """ 在文本中按指定强度注入随机噪音字符串。 - - 该方法通过在文本的单词之间随机插入无意义的字符串（噪音）来实现内容混淆。 - 强度越高，插入噪音的概率和长度就越大。 - - Args: - text (str): 待处理的文本。 - intensity (int): 混淆强度 (1-3)，决定噪音的概率和长度。 - - Returns: - str: 注入噪音后的文本。 """ - # 定义不同强度级别的噪音参数：概率和长度范围 + # 强度映射，将 "light", "medium", "heavy" 映射到 1, 2, 3 + strength_map = {"light": 1, "medium": 2, "heavy": 3} + intensity = strength_map.get(strength, 1) + params = { 1: {"probability": 15, "length": (3, 6)}, # 低强度 2: {"probability": 25, "length": (5, 10)}, # 中强度