refactor(prompt): 将注意力和内容混淆统一为提示词扰动
本次提交重构了提示词修改逻辑，将之前独立的“注意力优化”和“内容混淆”功能合并为一个统一的概念：“提示词扰动”(Prompt Perturbation)。 主要变更包括： - 在模型配置中引入新的统一选项：`enable_prompt_perturbation`, `perturbation_strength` 和 `enable_semantic_variants`。 - 将原 `AttentionOptimizer` 中的噪声注入和语义变体逻辑迁移到 `llm_models` 模块中，作为扰动策略的一部分。 - 简化 `attention_optimizer.py`，使其专注于提示词块重排 (`BlockShuffler`)。 - 更新 `_PromptProcessor` 以根据新的统一配置来协调不同的扰动技术。 此项更改为用户简化了配置，并通过集中化相关逻辑，提供了一个更清晰、更易于维护的实现。 BREAKING CHANGE: 内容混淆的相关配置已被替换。`enable_content_obfuscation` 和 `obfuscation_intensity` 配置项已移除。用户需更新配置以使用新的 `enable_prompt_perturbation` 和 `perturbation_strength`。
This commit is contained in:
minecraft1024a
committed by
Windpicker-owo
Windpicker-owo
parent
45b6d7d06c
commit
198976b68f
@@ -1,28 +1,20 @@
|
||||
"""
|
||||
注意力优化器 - 防止提示词过度相似导致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名称
|
||||
@@ -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()
|
||||
|
||||
for group in self.SWAPPABLE_BLOCK_GROUPS:
|
||||
# 过滤出实际存在且非空的block
|
||||
# 示例：假设模板中的占位符格式为 {block_name}
|
||||
# 我们需要解析模板，找到可重排的组，并重新构建模板字符串。
|
||||
|
||||
# 注意：这是一个复杂的逻辑，通常需要一个简单的模板引擎或正则表达式来完成。
|
||||
# 为保持此函数职责单一，这里仅演示核心的重排逻辑，
|
||||
# 完整的模板重建逻辑应在调用此函数的地方处理。
|
||||
|
||||
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}")
|
||||
|
||||
# 如果打乱后的顺序与原顺序不同，记录日志
|
||||
if shuffled != existing_blocks:
|
||||
logger.debug(f"重排block组: {existing_blocks} -> {shuffled}")
|
||||
# 这里的实现需要调用者根据 `existing_blocks` 的新顺序
|
||||
# 去动态地重新组织 `prompt_template` 字符串。
|
||||
# 例如，找到模板中与 `group` 相关的占位符部分，然后按新顺序替换它们。
|
||||
|
||||
# 注意：实际的重排需要在模板格式化之前进行
|
||||
# 这里只是演示逻辑，真正的实现需要在 _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
|
||||
|
||||
@@ -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):
|
||||
"""验证价格必须为非负数"""
|
||||
|
||||
@@ -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]:
|
||||
"""
|
||||
@@ -332,50 +473,15 @@ class _PromptProcessor:
|
||||
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)}, # 中强度
|
||||
|
||||
Reference in New Issue
Block a user