RexUniNLU模型安全防护：对抗样本检测与防御-开发者社区

RexUniNLU模型安全防护：对抗样本检测与防御

你有没有遇到过这种情况：一个平时表现很稳定的自然语言理解模型，突然对一段看似正常的文本给出了完全离谱的答案？比如，你问它“今天天气怎么样”，它却回答“请提供你的银行密码”。这听起来像是科幻电影里的情节，但在现实世界里，这很可能是模型遭遇了“对抗样本”攻击。

对抗样本就像是给模型精心设计的“陷阱”。攻击者通过对输入文本进行一些人类几乎察觉不到的微小改动——比如替换几个同音字、加几个不起眼的标点，或者插入一段特定的“咒语”——就能让模型彻底“失明”，做出错误的判断。对于像RexUniNLU这样承担着信息抽取、文本分类等重要任务的模型来说，这种安全威胁不容忽视。

今天，我们就来聊聊如何为你的RexUniNLU模型穿上“防弹衣”。我会带你从零开始，了解对抗样本是什么，学习如何检测它们，并掌握几种实用的防御方法。整个过程不需要你成为安全专家，我们会用最直白的语言和可运行的代码，让你快速上手。

1. 对抗样本：模型看不见的“陷阱”

在深入技术细节之前，我们先搞清楚对手是谁。对抗样本并不是什么魔法，它本质上是一种针对机器学习模型的“欺骗”技术。

想象一下，你训练了一只非常聪明的狗，它能根据气味分辨出苹果和橘子。但有人把苹果涂上了一层橘子味的香水，这只狗可能就会把苹果认成橘子。对抗样本对模型的作用类似于此——通过精心修改输入数据，让模型产生误判。

对于文本模型来说，这种修改通常非常隐蔽。比如下面这个例子：

原始输入：“这部电影的剧情非常精彩，演员表演也很到位。”对抗样本：“这部电影的剧情非常精采，演员表演也很到位。”

仅仅是把“精彩”换成了同音词“精采”，人类读者可能完全不会在意，甚至觉得是个笔误。但对于某些模型来说，这个微小的变化可能就足以让它对这句话的情感判断从“正面”翻转为“负面”。

为什么模型会这么“脆弱”？这跟它的工作原理有关。模型本质上是在高维空间里学习数据和标签之间的复杂映射关系。对抗样本就像是在这个空间里，沿着模型决策边界“推”了输入数据一小把，让它越过了边界，落到了错误的类别区域里。因为这种“推”的方向是精心计算出来的，所以只需要很小的改动就能达到目的。

在自然语言处理领域，常见的对抗攻击手法包括：

同义词替换：用意思相近但模型可能不熟悉的词替换原词。
字符级扰动：插入、删除或交换字符，特别是对于中文，使用形近字或同音字。
添加冗余信息：在文本中插入一段看似无关，但能干扰模型注意力的“对抗性后缀”。
风格转换：轻微改变句式或表达风格。

理解了对手，我们才能更好地防御。接下来，我们就开始动手，先看看怎么部署一个RexUniNLU模型，并尝试生成一些对抗样本来测试它的脆弱性。

2. 环境准备与RexUniNLU快速部署

工欲善其事，必先利其器。在开始研究安全防护之前，我们得先把模型跑起来。这里我会提供两种方式：一种是使用ModelScope官方推荐的pipeline，最简单快捷；另一种是直接使用PyTorch原生调用，灵活性更高。你可以根据需求选择。

2.1 基础环境搭建

首先，确保你的Python环境是3.8或以上版本。然后，我们安装必要的依赖库。

# 安装ModelScope和相关依赖 pip install modelscope torch transformers # 如果需要使用中文分词等工具 pip install jieba

2.2 使用ModelScope Pipeline快速部署

对于大多数应用场景，使用ModelScope的pipeline是最省事的方法。它封装了模型加载、预处理和后处理的完整流程，你只需要几行代码就能调用。

from modelscope.pipelines import pipeline from modelscope.models import Model from modelscope.preprocessors import Preprocessor # 创建信息抽取pipeline # 这里我们使用RexUniNLU的中文基础版 ie_pipeline = pipeline( task='information-extraction', model='damo/nlp_deberta_rex-uninlu_chinese-base', model_revision='v1.2.1' ) # 测试一下模型的基本功能 text = "张三毕业于北京大学，目前在阿里巴巴担任高级工程师。" schema = { '人物': ['姓名', '毕业院校', '职业'], '机构': ['名称', '地点'] } result = ie_pipeline(text, schema=schema) print("抽取结果:", result)

运行这段代码，你应该能看到模型从文本中抽取出结构化的信息。如果一切正常，说明模型部署成功了。

2.3 使用PyTorch原生调用（可选）

如果你需要更细粒度的控制，或者想在模型内部插入一些安全检测模块，那么直接使用PyTorch调用可能更合适。这种方式稍微复杂一些，但灵活性更高。

import torch from transformers import AutoTokenizer, AutoModelForTokenClassification from modelscope import snapshot_download # 下载模型到本地 model_dir = snapshot_download('damo/nlp_deberta_rex-uninlu_chinese-base', revision='v1.2.1') # 加载tokenizer和模型 tokenizer = AutoTokenizer.from_pretrained(model_dir) model = AutoModelForTokenClassification.from_pretrained(model_dir) # 将模型设置为评估模式 model.eval() # 准备输入 text = "这部电影的剧情非常精彩，演员表演也很到位。" inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True) # 前向传播 with torch.no_grad(): outputs = model(**inputs) # 处理输出（这里需要根据具体任务调整） logits = outputs.logits predictions = torch.argmax(logits, dim=-1) print("模型输出logits形状:", logits.shape) print("预测结果:", predictions)

原生调用的好处是你可以完全控制数据处理流程，方便我们在后续步骤中插入安全检测逻辑。

3. 对抗样本检测：给模型装上“警报器”

检测对抗样本就像是给模型安装一个烟雾报警器。当有“异常”输入试图欺骗模型时，报警器应该及时响起。这里我介绍几种实用的检测方法，从简单到复杂，你可以根据实际需求选择或组合使用。

3.1 基于输入异常值的检测

这种方法的核心思想是：对抗样本往往在统计特征上与正常样本有所不同。我们可以计算一些特征指标，当指标超出正常范围时，就怀疑可能是对抗样本。

import numpy as np from collections import Counter class TextAnomalyDetector: """基于文本统计特征的异常检测器""" def __init__(self): # 这里可以初始化一些阈值，实际应用中可以通过统计正常数据得到 self.char_repeat_threshold = 5 # 连续重复字符阈值 self.special_char_ratio = 0.3 # 特殊字符比例阈值 self.rare_word_threshold = 0.2 # 罕见词比例阈值 # 假设我们有一个常见词表（实际应用中需要根据领域构建） self.common_words = set(["的", "了", "在", "是", "我", "有", "和", "就", "不", "人", "都", "一", "一个", "上", "也", "很", "到", "说", "要", "去", "你", "会", "着", "没有", "看", "好", "自己", "这"]) def detect(self, text): """检测文本是否异常""" anomalies = [] # 1. 检查连续重复字符 max_repeat = self._check_char_repetition(text) if max_repeat > self.char_repeat_threshold: anomalies.append(f"连续重复字符过多: {max_repeat}") # 2. 检查特殊字符比例 special_ratio = self._check_special_chars(text) if special_ratio > self.special_char_ratio: anomalies.append(f"特殊字符比例过高: {special_ratio:.2f}") # 3. 检查罕见词比例（简化版） rare_ratio = self._check_rare_words(text) if rare_ratio > self.rare_word_threshold: anomalies.append(f"罕见词比例过高: {rare_ratio:.2f}") # 4. 检查文本长度异常（过短或过长） if len(text) < 5 or len(text) > 1000: anomalies.append(f"文本长度异常: {len(text)}") return len(anomalies) == 0, anomalies def _check_char_repetition(self, text): """检查连续重复字符""" max_repeat = 1 current_repeat = 1 for i in range(1, len(text)): if text[i] == text[i-1]: current_repeat += 1 max_repeat = max(max_repeat, current_repeat) else: current_repeat = 1 return max_repeat def _check_special_chars(self, text): """检查特殊字符比例""" special_chars = set("!@#$%^&*()_+-=[]{}|;':\",./<>?~`") special_count = sum(1 for char in text if char in special_chars) return special_count / len(text) if len(text) > 0 else 0 def _check_rare_words(self, text): """检查罕见词比例（简化版）""" # 这里使用简单的分词，实际应用中可以使用更专业的分词工具 words = [text[i:i+2] for i in range(0, len(text)-1)] rare_count = sum(1 for word in words if word not in self.common_words) return rare_count / len(words) if len(words) > 0 else 0 # 使用示例 detector = TextAnomalyDetector() # 测试正常文本 normal_text = "这部电影的剧情非常精彩，演员表演也很到位。" is_normal, issues = detector.detect(normal_text) print(f"正常文本检测: {'正常' if is_normal else '异常'}, 问题: {issues}") # 测试可能的对抗样本（包含重复字符） adversarial_text = "这部电影的剧情非常精彩！！！！！！，演员表演也很到位。" is_normal, issues = detector.detect(adversarial_text) print(f"对抗文本检测: {'正常' if is_normal else '异常'}, 问题: {issues}")

这种方法实现简单，计算速度快，可以作为第一道防线。但它也有局限性——聪明的攻击者可能会生成统计特征正常的对抗样本。

3.2 基于模型置信度的检测

另一种思路是利用模型自身的输出信息。正常样本通常会让模型给出高置信度的预测，而对抗样本可能会让模型“犹豫不决”，或者产生异常的置信度分布。

class ConfidenceBasedDetector: """基于模型置信度的对抗样本检测器""" def __init__(self, model, tokenizer, threshold=0.7): self.model = model self.tokenizer = tokenizer self.confidence_threshold = threshold # 置信度阈值 def detect(self, text, schema): """检测文本是否为对抗样本""" # 获取模型原始输出 inputs = self.tokenizer(text, return_tensors="pt", padding=True, truncation=True) with torch.no_grad(): outputs = self.model(**inputs) # 计算置信度相关指标 confidence_metrics = self._calculate_confidence_metrics(outputs) # 判断是否为对抗样本 is_adversarial = self._judge_adversarial(confidence_metrics) return is_adversarial, confidence_metrics def _calculate_confidence_metrics(self, outputs): """计算置信度相关指标""" logits = outputs.logits probabilities = torch.softmax(logits, dim=-1) # 1. 最大概率值（模型对预测的置信度） max_probs, _ = torch.max(probabilities, dim=-1) avg_max_prob = torch.mean(max_probs).item() # 2. 预测熵（不确定性度量） entropy = -torch.sum(probabilities * torch.log(probabilities + 1e-10), dim=-1) avg_entropy = torch.mean(entropy).item() # 3. 概率分布平滑度（对抗样本往往有异常分布） prob_std = torch.std(probabilities, dim=-1) avg_prob_std = torch.mean(prob_std).item() return { 'avg_max_prob': avg_max_prob, 'avg_entropy': avg_entropy, 'avg_prob_std': avg_prob_std } def _judge_adversarial(self, metrics): """根据指标判断是否为对抗样本""" # 简单的规则：置信度过低或熵过高都可能是对抗样本 if metrics['avg_max_prob'] < self.confidence_threshold: return True # 熵过高说明模型不确定 if metrics['avg_entropy'] > 1.5: # 这个阈值需要根据实际数据调整 return True return False # 使用示例（需要先加载模型） # detector = ConfidenceBasedDetector(model, tokenizer) # is_adv, metrics = detector.detect("测试文本", schema) # print(f"是否为对抗样本: {is_adv}, 指标: {metrics}")

置信度检测的优势是它直接利用了模型的内在状态，不需要额外的训练。但阈值需要根据具体任务和数据进行调整。

3.3 集成检测方法

在实际应用中，我们通常不会只依赖单一检测方法，而是会组合多种检测器，形成一个检测流水线。

class DefensePipeline: """综合防御流水线""" def __init__(self, model, tokenizer): self.model = model self.tokenizer = tokenizer # 初始化多个检测器 self.anomaly_detector = TextAnomalyDetector() self.confidence_detector = ConfidenceBasedDetector(model, tokenizer) # 还可以添加更多检测器... def process(self, text, schema): """处理输入文本，包含检测和防御""" # 第一步：输入清洗和规范化 cleaned_text = self._clean_input(text) # 第二步：多维度检测 detection_results = self._run_detectors(cleaned_text, schema) # 第三步：根据检测结果决定处理方式 if detection_results['is_adversarial']: # 如果检测到对抗样本，采取防御措施 defended_text = self._apply_defense(cleaned_text, detection_results) print(f"警告：检测到潜在对抗样本，已启用防御模式") print(f"检测到的问题: {detection_results['issues']}") else: defended_text = cleaned_text # 第四步：安全推理 safe_result = self._safe_inference(defended_text, schema) return { 'original_text': text, 'cleaned_text': cleaned_text, 'defended_text': defended_text if detection_results['is_adversarial'] else None, 'is_adversarial': detection_results['is_adversarial'], 'detection_details': detection_results, 'safe_result': safe_result } def _clean_input(self, text): """基础输入清洗""" # 移除多余空格 text = ' '.join(text.split()) # 限制最大长度（防止超长攻击） max_length = 500 if len(text) > max_length: text = text[:max_length] + "...[已截断]" return text def _run_detectors(self, text, schema): """运行所有检测器""" issues = [] # 异常检测 is_normal, anomaly_issues = self.anomaly_detector.detect(text) if not is_normal: issues.extend(anomaly_issues) # 置信度检测 is_adv, confidence_metrics = self.confidence_detector.detect(text, schema) if is_adv: issues.append(f"模型置信度异常: {confidence_metrics}") # 判断是否为对抗样本 is_adversarial = len(issues) > 0 return { 'is_adversarial': is_adversarial, 'issues': issues, 'confidence_metrics': confidence_metrics if 'confidence_metrics' in locals() else None } def _apply_defense(self, text, detection_results): """应用防御措施""" # 这里可以实现具体的防御策略 # 例如：文本重构、添加噪声、使用鲁棒性更强的推理模式等 defended_text = text # 这里先简单返回原文本 # 可以根据检测到的问题类型采取不同防御 if "连续重复字符过多" in str(detection_results['issues']): # 处理重复字符 defended_text = self._remove_excessive_repetition(text) return defended_text def _remove_excessive_repetition(self, text): """移除过多的重复字符""" import re # 将连续重复超过3次的字符缩减为3次 pattern = r'(.)\1{3,}' return re.sub(pattern, r'\1\1\1', text) def _safe_inference(self, text, schema): """安全推理""" # 这里可以添加额外的安全措施 # 例如：限制推理时间、监控资源使用等 # 简单调用模型 inputs = self.tokenizer(text, return_tensors="pt", padding=True, truncation=True) with torch.no_grad(): outputs = self.model(**inputs) # 处理输出（根据具体任务调整） return outputs # 使用示例 # pipeline = DefensePipeline(model, tokenizer) # result = pipeline.process("输入文本", schema)

这种流水线式的设计让我们的防御系统更加健壮和灵活。你可以根据需要随时添加新的检测器或防御策略。

4. 对抗样本防御：构建模型的“免疫系统”

检测到对抗样本只是第一步，更重要的是如何防御。防御的目标是让模型在面对对抗攻击时，仍然能保持正确的判断。这里我介绍几种实用的防御策略。

4.1 输入预处理与清洗

这是最直接的防御方法——在文本进入模型之前，先进行清洗和规范化处理。

class InputSanitizer: """输入净化器""" def __init__(self): # 加载常见对抗模式（实际应用中需要不断更新） self.adversarial_patterns = [ r'([!@#$%^&*()_+=\[\]{}|;:\",./<>?~`])\1{5,}', # 重复特殊字符 r'[^\u4e00-\u9fa5a-zA-Z0-9\s，。！？、；："\'（）《》]', # 异常字符 r'\b(?:http|ftp|https)://\S+\b', # URL（某些攻击可能包含） r'\b\d{10,}\b', # 超长数字（可能用于溢出攻击） ] # 同音字/形近字映射表（需要根据实际情况完善） self.homophone_map = { '精采': '精彩', '在次': '再次', '必需': '必须', # 添加更多... } def sanitize(self, text): """净化输入文本""" original_text = text # 第一步：基础清洗 text = self._basic_clean(text) # 第二步：检测并移除对抗模式 text = self._remove_adversarial_patterns(text) # 第三步：纠正常见错误（如同音字） text = self._correct_common_errors(text) # 第四步：长度标准化 text = self._normalize_length(text) # 记录修改（用于审计） modifications = [] if text != original_text: modifications.append(f"文本已被净化") return text, modifications def _basic_clean(self, text): """基础清洗""" # 移除不可见字符 import re text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', text) # 标准化空白字符 text = ' '.join(text.split()) # 移除首尾空白 text = text.strip() return text def _remove_adversarial_patterns(self, text): """移除对抗模式""" import re for pattern in self.adversarial_patterns: text = re.sub(pattern, '', text) return text def _correct_common_errors(self, text): """纠正常见错误""" for wrong, correct in self.homophone_map.items(): text = text.replace(wrong, correct) return text def _normalize_length(self, text): """长度标准化""" max_length = 1000 min_length = 2 if len(text) > max_length: text = text[:max_length] + "...[已截断]" elif len(text) < min_length: # 过短的文本可能是攻击，用占位符替代 text = "[文本过短]" return text # 使用示例 sanitizer = InputSanitizer() # 测试对抗样本清洗 adversarial_text = "这部电影的剧情非常精彩！！！！！！，演员表演也很到位。http://malicious.com 12345678901234567890" clean_text, modifications = sanitizer.sanitize(adversarial_text) print(f"原始文本: {adversarial_text}") print(f"净化后文本: {clean_text}") print(f"修改记录: {modifications}")

输入预处理就像给模型加了一个“过滤器”，可以拦截很多简单的攻击。但这种方法对高级的、语义层面的攻击效果有限。

4.2 对抗训练：让模型“见多识广”

对抗训练是目前最有效的防御方法之一。它的核心思想是：在训练过程中，主动生成对抗样本并让模型学习如何正确分类它们。这样模型就能“见多识广”，遇到真正的攻击时也不容易出错。

class AdversarialTrainer: """对抗训练器""" def __init__(self, model, tokenizer, device='cuda'): self.model = model self.tokenizer = tokenizer self.device = device # 对抗攻击方法（用于生成训练用的对抗样本） self.attack_methods = [ self._synonym_replacement_attack, self._character_perturbation_attack, self._insert_noise_attack ] def generate_adversarial_examples(self, texts, labels, attack_ratio=0.3): """生成对抗样本""" adversarial_texts = [] adversarial_labels = [] num_attacks = int(len(texts) * attack_ratio) for i in range(num_attacks): original_text = texts[i] original_label = labels[i] # 随机选择一种攻击方法 attack_method = np.random.choice(self.attack_methods) # 生成对抗样本 adversarial_text = attack_method(original_text) adversarial_texts.append(adversarial_text) adversarial_labels.append(original_label) # 标签不变 # 合并原始样本和对抗样本 all_texts = texts + adversarial_texts all_labels = labels + adversarial_labels return all_texts, all_labels def _synonym_replacement_attack(self, text): """同义词替换攻击""" # 这里使用简单的同义词表，实际应用中可以使用更专业的同义词库 synonym_dict = { '好': ['棒', '佳', '优', '良'], '坏': ['差', '糟', '劣', '烂'], '大': ['巨', '广', '阔', '庞'], '小': ['微', '细', '窄', '纤'], # 添加更多... } words = list(text) for i, char in enumerate(words): if char in synonym_dict and np.random.random() < 0.3: # 30%概率替换 synonyms = synonym_dict[char] if synonyms: words[i] = np.random.choice(synonyms) return ''.join(words) def _character_perturbation_attack(self, text): """字符扰动攻击""" # 随机插入、删除或替换字符 if len(text) < 3: return text words = list(text) operation = np.random.choice(['insert', 'delete', 'replace']) if operation == 'insert' and len(words) < 100: # 随机插入一个字符 pos = np.random.randint(0, len(words)) random_char = chr(np.random.randint(19968, 40959)) # 随机中文字符 words.insert(pos, random_char) elif operation == 'delete' and len(words) > 2: # 随机删除一个字符 pos = np.random.randint(0, len(words)) del words[pos] elif operation == 'replace': # 随机替换一个字符 pos = np.random.randint(0, len(words)) random_char = chr(np.random.randint(19968, 40959)) words[pos] = random_char return ''.join(words) def _insert_noise_attack(self, text): """插入噪声攻击""" # 在文本中插入随机噪声 noise_patterns = [ '。', '，', '！', '？', '；', '的', '了', '在', '是', '我' ] if len(text) < 10: return text # 随机选择插入位置和噪声 num_insertions = np.random.randint(1, 3) words = list(text) for _ in range(num_insertions): pos = np.random.randint(0, len(words)) noise = np.random.choice(noise_patterns) words.insert(pos, noise) return ''.join(words) def train_step(self, batch_texts, batch_labels, optimizer): """对抗训练步骤""" # 生成对抗样本 adv_texts, adv_labels = self.generate_adversarial_examples( batch_texts, batch_labels, attack_ratio=0.5 ) # 准备模型输入 inputs = self.tokenizer( adv_texts, return_tensors="pt", padding=True, truncation=True, max_length=512 ).to(self.device) labels = torch.tensor(adv_labels).to(self.device) # 前向传播 outputs = self.model(**inputs, labels=labels) loss = outputs.loss # 反向传播 optimizer.zero_grad() loss.backward() optimizer.step() return loss.item() # 使用示例（训练循环框架） def adversarial_training_loop(model, train_data, epochs=10): """对抗训练循环""" tokenizer = AutoTokenizer.from_pretrained('damo/nlp_deberta_rex-uninlu_chinese-base') trainer = AdversarialTrainer(model, tokenizer) optimizer = torch.optim.AdamW(model.parameters(), lr=1e-5) for epoch in range(epochs): total_loss = 0 num_batches = 0 for batch_texts, batch_labels in train_data: loss = trainer.train_step(batch_texts, batch_labels, optimizer) total_loss += loss num_batches += 1 avg_loss = total_loss / num_batches print(f"Epoch {epoch+1}/{epochs}, Loss: {avg_loss:.4f}") return model

对抗训练虽然效果好，但计算成本较高，因为需要在训练过程中不断生成对抗样本。对于已经训练好的模型，可以采用“微调”的方式，用对抗样本对模型进行少量迭代训练，也能显著提升鲁棒性。

4.3 模型集成与投票机制

“三个臭皮匠，顶个诸葛亮。”模型集成的基本思想是：使用多个不同的模型（或同一模型的不同变体）对同一输入进行预测，然后通过投票或平均的方式得到最终结果。对抗样本很难同时欺骗所有模型。

class EnsembleDefense: """集成防御系统""" def __init__(self, model_paths): self.models = [] self.tokenizers = [] # 加载多个模型 for path in model_paths: tokenizer = AutoTokenizer.from_pretrained(path) model = AutoModelForTokenClassification.from_pretrained(path) model.eval() self.tokenizers.append(tokenizer) self.models.append(model) print(f"已加载 {len(self.models)} 个模型用于集成防御") def predict_with_defense(self, text, schema): """带防御的预测""" all_predictions = [] all_confidences = [] # 每个模型独立预测 for i, (model, tokenizer) in enumerate(zip(self.models, self.tokenizers)): inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True) with torch.no_grad(): outputs = model(**inputs) # 获取预测结果和置信度 predictions, confidence = self._process_outputs(outputs) all_predictions.append(predictions) all_confidences.append(confidence) # 检测异常（如果某个模型的预测与其他模型差异很大） is_anomalous, anomaly_score = self._detect_anomaly(all_predictions, all_confidences) if is_anomalous: print(f"检测到预测异常，异常分数: {anomaly_score:.3f}") # 启用防御模式：使用更保守的集成策略 final_prediction = self._defensive_ensemble(all_predictions, all_confidences) else: # 正常模式：使用标准集成策略 final_prediction = self._standard_ensemble(all_predictions, all_confidences) return { 'prediction': final_prediction, 'is_anomalous': is_anomalous, 'anomaly_score': anomaly_score, 'individual_predictions': all_predictions, 'individual_confidences': all_confidences } def _process_outputs(self, outputs): """处理模型输出""" logits = outputs.logits probabilities = torch.softmax(logits, dim=-1) # 获取预测类别 predictions = torch.argmax(logits, dim=-1) # 计算置信度（平均最大概率） max_probs, _ = torch.max(probabilities, dim=-1) confidence = torch.mean(max_probs).item() return predictions.cpu().numpy(), confidence def _detect_anomaly(self, all_predictions, all_confidences): """检测预测异常""" if len(all_predictions) < 2: return False, 0.0 # 计算预测一致性 agreement_matrix = np.zeros((len(all_predictions), len(all_predictions))) for i in range(len(all_predictions)): for j in range(i+1, len(all_predictions)): # 比较两个模型的预测结果 agreement = np.mean(all_predictions[i] == all_predictions[j]) agreement_matrix[i, j] = agreement agreement_matrix[j, i] = agreement # 平均一致性分数 avg_agreement = np.mean(agreement_matrix[np.triu_indices_from(agreement_matrix, k=1)]) # 如果一致性过低，可能是对抗样本 anomaly_score = 1.0 - avg_agreement return anomaly_score > 0.5, anomaly_score def _standard_ensemble(self, predictions, confidences): """标准集成：加权投票""" # 根据置信度加权 weights = np.array(confidences) weights = weights / np.sum(weights) # 归一化 # 对于分类任务，可以计算加权投票 # 这里简化处理：返回置信度最高的模型的预测 best_model_idx = np.argmax(confidences) return predictions[best_model_idx] def _defensive_ensemble(self, predictions, confidences): """防御性集成：更保守的策略""" # 当检测到异常时，使用更保守的集成策略 # 例如：只信任高置信度的预测，或者使用多数投票 # 方法1：多数投票 from scipy import stats # 将所有预测堆叠起来 stacked_predictions = np.stack(predictions, axis=0) # 对每个位置进行多数投票 final_prediction, _ = stats.mode(stacked_predictions, axis=0) return final_prediction[0] # 去掉多余的维度 # 使用示例 # 假设我们有多个RexUniNLU变体或不同版本的模型 model_paths = [ 'damo/nlp_deberta_rex-uninlu_chinese-base', # '另一个模型路径', # '第三个模型路径', ] # ensemble = EnsembleDefense(model_paths) # result = ensemble.predict_with_defense("测试文本", schema)

模型集成的防御效果很好，但需要维护多个模型，推理成本也更高。在实际应用中，可以根据安全需求在效果和成本之间做权衡。

4.4 运行时监控与自适应防御

对于生产系统，我们还需要实时监控模型的行为，及时发现异常并采取应对措施。

class RuntimeMonitor: """运行时监控器""" def __init__(self, model, tokenizer): self.model = model self.tokenizer = tokenizer # 监控指标历史记录 self.confidence_history = [] self.latency_history = [] self.anomaly_history = [] # 报警阈值 self.confidence_threshold = 0.6 self.latency_threshold = 2.0 # 秒 self.anomaly_rate_threshold = 0.1 # 异常请求比例 # 防御状态 self.defense_level = 0 # 0:正常, 1:警告, 2:防御, 3:高警戒 self.consecutive_anomalies = 0 def monitor_request(self, text, schema, inference_time): """监控单个请求""" metrics = {} # 获取模型置信度 inputs = self.tokenizer(text, return_tensors="pt", padding=True, truncation=True) with torch.no_grad(): outputs = self.model(**inputs) probabilities = torch.softmax(outputs.logits, dim=-1) max_probs, _ = torch.max(probabilities, dim=-1) avg_confidence = torch.mean(max_probs).item() metrics['confidence'] = avg_confidence metrics['inference_time'] = inference_time # 检查异常 is_anomalous = False if avg_confidence < self.confidence_threshold: metrics['anomaly'] = '低置信度' is_anomalous = True if inference_time > self.latency_threshold: metrics['anomaly'] = '高延迟' is_anomalous = True # 更新历史记录 self.confidence_history.append(avg_confidence) self.latency_history.append(inference_time) self.anomaly_history.append(1 if is_anomalous else 0) # 保持历史记录长度 max_history = 100 if len(self.confidence_history) > max_history: self.confidence_history = self.confidence_history[-max_history:] self.latency_history = self.latency_history[-max_history:] self.anomaly_history = self.anomaly_history[-max_history:] # 更新防御状态 self._update_defense_state(is_anomalous) metrics['defense_level'] = self.defense_level metrics['is_anomalous'] = is_anomalous return metrics def _update_defense_state(self, is_anomalous): """更新防御状态""" if is_anomalous: self.consecutive_anomalies += 1 else: self.consecutive_anomalies = max(0, self.consecutive_anomalies - 1) # 计算近期异常率 recent_history = self.anomaly_history[-20:] if len(self.anomaly_history) >= 20 else self.anomaly_history if recent_history: anomaly_rate = sum(recent_history) / len(recent_history) else: anomaly_rate = 0 # 根据异常情况调整防御级别 if anomaly_rate > 0.3 or self.consecutive_anomalies >= 5: self.defense_level = 3 # 高警戒 elif anomaly_rate > 0.15 or self.consecutive_anomalies >= 3: self.defense_level = 2 # 防御模式 elif anomaly_rate > 0.05: self.defense_level = 1 # 警告模式 else: self.defense_level = 0 # 正常模式 def get_system_health(self): """获取系统健康状态""" if not self.confidence_history: return "未知" avg_confidence = np.mean(self.confidence_history) avg_latency = np.mean(self.latency_history) anomaly_rate = np.mean(self.anomaly_history) if self.anomaly_history else 0 health_score = ( 0.4 * (avg_confidence / 1.0) + # 置信度贡献40% 0.3 * (1.0 / max(avg_latency, 0.1)) + # 延迟贡献30%（取倒数，延迟越低越好） 0.3 * (1.0 - anomaly_rate) # 异常率贡献30% ) if health_score > 0.8: return "健康" elif health_score > 0.6: return "一般" elif health_score > 0.4: return "警告" else: return "危险" def get_defense_recommendations(self): """获取防御建议""" recommendations = [] if self.defense_level >= 2: recommendations.append("启用输入验证和过滤") recommendations.append("降低模型服务并发度") recommendations.append("启用请求频率限制") if self.defense_level >= 3: recommendations.append("切换到备份模型") recommendations.append("启用人工审核流程") recommendations.append("记录所有请求用于分析") return recommendations # 使用示例 # monitor = RuntimeMonitor(model, tokenizer) # # # 在每次推理后调用监控 # start_time = time.time() # result = model_inference(text, schema) # inference_time = time.time() - start_time # # metrics = monitor.monitor_request(text, schema, inference_time) # print(f"监控指标: {metrics}") # print(f"系统健康: {monitor.get_system_health()}") # print(f"防御建议: {monitor.get_defense_recommendations()}")

运行时监控就像给模型系统安装了一个“健康检测仪”，可以实时了解系统的运行状态，并在出现问题时及时采取措施。