109_噪声鲁棒微调：对抗训练
在当今大语言模型（LLM）的广泛应用中，模型的鲁棒性问题日益凸显。对抗性攻击通过在输入中添加微小但精心设计的扰动，能够误导模型产生错误输出，这对依赖LLM的关键系统构成了严重威胁。噪声鲁棒微调作为提升模型抵抗对抗攻击能力的重要技术，正成为大模型安全性研究的核心方向之一。
·
引言
在当今大语言模型（LLM）的广泛应用中，模型的鲁棒性问题日益凸显。对抗性攻击通过在输入中添加微小但精心设计的扰动，能够误导模型产生错误输出，这对依赖LLM的关键系统构成了严重威胁。噪声鲁棒微调作为提升模型抵抗对抗攻击能力的重要技术，正成为大模型安全性研究的核心方向之一。
本文将深入探讨噪声鲁棒微调技术，特别是对抗训练方法在大语言模型中的应用。我们将从基本概念入手，详细分析主流对抗训练算法，包括FGSM、PGD等经典方法，以及2025年涌现的最新研究成果。同时，我们将提供完整的代码实现示例，帮助读者在实际项目中应用这些技术。
目录
-
- 1.1 对抗性样本的概念与威胁
- 1.2 对抗训练的基本原理
- 1.3 对抗训练与普通微调的区别
-
- 2.1 FGSM（快速梯度符号法）
- 2.2 PGD（投影梯度下降）
- 2.3 MIM（动量迭代法）
- 2.4 TRADES（鲁棒性正则化）
-
- 3.1 高维输入空间的挑战
- 3.2 计算资源消耗
- 3.3 文本扰动的特殊性质
- 3.4 鲁棒性与性能的平衡
-
- 4.1 文本扰动生成方法
- 4.2 扰动约束与投影技术
- 4.3 训练目标函数设计
- 4.4 优化策略与超参数调整
-
- 5.1 环境配置与依赖安装
- 5.2 基本数据处理
- 5.3 FGSM对抗训练实现
- 5.4 PGD对抗训练实现
- 5.5 混合对抗训练策略
-
- 6.1 集成对抗训练
- 6.2 元对抗训练
- 6.3 自适应对抗训练
- 6.4 鲁棒性正则化技术
-
- 7.1 对抗性样本生成评估
- 7.2 鲁棒性指标体系
- 7.3 标准基准测试
- 7.4 实际应用场景测试
-
- 8.1 计算效率提升
- 8.2 内存优化技术
- 8.3 分布式训练策略
- 8.4 部署与推理优化
-
- 9.1 架构感知的对抗训练
- 9.2 跨模态对抗鲁棒性
- 9.3 联邦对抗训练
- 9.4 可证明鲁棒性方法
-
- 10.1 内容审核与安全过滤
- 10.2 金融风险评估
- 10.3 医疗文本分析
- 10.4 法律文档处理
对抗训练基础
1.1 对抗性样本的概念与威胁
对抗性样本是指在原始输入中添加精心设计的微小扰动后形成的输入样本，这些样本能够导致机器学习模型产生错误的预测结果，而对人类来说，这些扰动几乎是不可察觉的。在大语言模型领域，对抗性样本可以表现为在正常文本中插入特殊字符、同义词替换、句式变换等方式，从而误导模型生成有害或不准确的内容。
对抗性攻击对大模型应用构成了多方面威胁：
- 安全风险：恶意用户可能通过对抗性攻击绕过内容审核系统，生成有害内容。
- 可靠性降低：在关键应用场景（如医疗、金融）中，对抗性攻击可能导致模型给出错误建议，造成严重后果。
- 隐私泄露：某些对抗性攻击可能被用来探测模型内部信息或诱发隐私信息泄露。
- 系统滥用：攻击者可能利用对抗性样本滥用AI系统，进行网络钓鱼、欺诈等活动。
1.2 对抗训练的基本原理
对抗训练是一种通过在训练过程中主动引入对抗性样本来增强模型鲁棒性的技术。其核心思想是：如果模型能够正确分类或处理经过对抗扰动的样本，那么它在面对真实世界的噪声和扰动时也会表现得更加稳健。
对抗训练的基本流程包括：
- 对抗样本生成：对于每个训练样本，使用特定算法（如FGSM、PGD等）生成对应的对抗样本。
- 混合训练：将原始样本和对抗样本混合作为训练数据。
- 目标优化：使用混合数据更新模型参数，使模型同时对原始样本和对抗样本保持良好性能。
数学上，对抗训练的目标函数可以表示为：
min θ E ( x , y ) ∼ D [ max ∥ δ ∥ ≤ ϵ L ( f θ ( x + δ ) , y ) ] \min_{\theta} \mathbb{E}_{(x,y) \sim \mathcal{D}} \left[ \max_{\|\delta\| \leq \epsilon} \mathcal{L}(f_\theta(x+\delta), y) \right] θminE(x,y)∼D[∥δ∥≤ϵmaxL(fθ(x+δ),y)]
其中， θ \theta θ是模型参数， D \mathcal{D} D是数据分布， δ \delta δ是添加的扰动， ϵ \epsilon ϵ是扰动的最大范数约束， L \mathcal{L} L是损失函数。
1.3 对抗训练与普通微调的区别
虽然对抗训练和普通微调都是对预训练模型进行参数更新的过程，但两者存在显著区别：
| 特性 | 普通微调 | 对抗训练 |
|---|---|---|
| 训练数据 | 原始数据样本 | 原始数据+对抗样本 |
| 优化目标 | 最小化原始数据上的损失 | 最小化对抗扰动数据上的损失上界 |
| 计算复杂度 | 较低 | 较高（需生成对抗样本） |
| 鲁棒性提升 | 有限 | 显著 |
| 性能影响 | 可能过拟合特定任务 | 在保持泛化能力的同时提高鲁棒性 |
| 适用场景 | 一般应用场景 | 安全敏感场景、对抗攻击防御 |
在实际应用中，对抗训练通常作为普通微调的一种增强形式，通过牺牲一定的计算效率来换取模型鲁棒性的提升。
经典对抗训练算法
2.1 FGSM（快速梯度符号法）
FGSM（Fast Gradient Sign Method）是最早也是最经典的对抗样本生成方法之一，由Goodfellow等人在2014年提出。该方法通过在原始输入的梯度符号方向上添加扰动来生成对抗样本，具有计算效率高的优点。
2.1.1 算法原理
FGSM的核心思想是利用损失函数对输入的梯度信息，在梯度的符号方向上添加一个小的扰动，从而最大化模型的损失。数学表达式为：
x a d v = x + ϵ ⋅ sign ( ∇ x L ( f θ ( x ) , y ) ) x_{adv} = x + \epsilon \cdot \text{sign}(\nabla_x \mathcal{L}(f_\theta(x), y)) xadv=x+ϵ⋅sign(∇xL(fθ(x),y))
其中， x x x是原始输入， ϵ \epsilon ϵ是扰动强度超参数， ∇ x L \nabla_x \mathcal{L} ∇xL是损失函数对输入的梯度， sign ( ⋅ ) \text{sign}(\cdot) sign(⋅)是符号函数。
2.1.2 算法流程
- 前向传播：使用当前模型参数对输入样本进行前向计算，得到预测结果。
- 计算损失：根据预测结果和真实标签计算损失值。
- 反向传播：计算损失函数对输入的梯度。
- 生成扰动：取梯度的符号，并乘以扰动强度 ϵ \epsilon ϵ。
- 应用扰动：将扰动添加到原始输入上，得到对抗样本。
- 更新模型：使用对抗样本和原始标签更新模型参数。
2.1.3 优缺点分析
优点：
- 计算效率高，仅需一次梯度计算
- 实现简单，易于集成到现有训练流程
- 适合大规模模型和数据集的训练
缺点：
- 生成的对抗样本相对简单，鲁棒性提升有限
- 对扰动强度 ϵ \epsilon ϵ的选择敏感
- 容易过拟合特定类型的对抗攻击
2.2 PGD（投影梯度下降）
PGD（Projected Gradient Descent）是一种更强大的对抗样本生成方法，通过迭代多次梯度更新来生成更具攻击性的对抗样本。PGD生成的对抗样本通常比FGSM更难防御，因此在对抗训练中被广泛使用。
2.2.1 算法原理
PGD通过多次迭代优化扰动，每次迭代都沿着损失函数增加的方向更新扰动，并将扰动投影回指定的 ℓ p \ell_p ℓp球内，以确保扰动在可接受范围内。数学表达式为：
δ 0 = 0 δ t + 1 = clip δ ∈ [ 0 , 1 ] n , ∥ δ ∥ p ≤ ϵ ( δ t + α ⋅ sign ( ∇ x L ( f θ ( x + δ t ) , y ) ) ) x a d v = x + δ T \begin{aligned} \delta_0 &= 0 \\ \delta_{t+1} &= \text{clip}_{\delta \in [0,1]^n, \|\delta\|_p \leq \epsilon} \left( \delta_t + \alpha \cdot \text{sign}(\nabla_x \mathcal{L}(f_\theta(x+\delta_t), y)) \right) \\ x_{adv} &= x + \delta_T \end{aligned} δ0δt+1xadv=0=clipδ∈[0,1]n,∥δ∥p≤ϵ(δt+α⋅sign(∇xL(fθ(x+δt),y)))=x+δT
其中， α \alpha α是每次迭代的步长， T T T是迭代次数， clip \text{clip} clip操作确保扰动在有效范围内。
2.2.2 算法流程
- 初始化扰动 δ 0 \delta_0 δ0（通常为零向量或随机小扰动）。
- 对于每一次迭代 t = 0 , 1 , . . . , T − 1 t=0,1,...,T-1 t=0,1,...,T−1：
a. 使用当前扰动 δ t \delta_t δt的样本 x + δ t x+\delta_t x+δt前向传播，计算损失。
b. 计算损失函数对输入的梯度。
c. 沿着梯度符号方向更新扰动： δ t + 1 = δ t + α ⋅ sign ( ∇ x L ) \delta_{t+1} = \delta_t + \alpha \cdot \text{sign}(\nabla_x \mathcal{L}) δt+1=δt+α⋅sign(∇xL)。
d. 投影扰动：将 δ t + 1 \delta_{t+1} δt+1投影到 ℓ p \ell_p ℓp球内，确保 ∥ δ t + 1 ∥ p ≤ ϵ \|\delta_{t+1}\|_p \leq \epsilon ∥δt+1∥p≤ϵ。 - 最终对抗样本为 x a d v = x + δ T x_{adv} = x + \delta_T xadv=x+δT。
- 使用对抗样本和原始标签更新模型参数。
2.2.3 优缺点分析
优点：
- 生成的对抗样本更具攻击性，能显著提升模型鲁棒性
- 可以通过调整迭代次数和步长来平衡攻击强度和计算效率
- 对各种模型架构都有较好的适应性
缺点：
- 计算复杂度高，需要多次前向和反向传播
- 超参数选择敏感，需要仔细调优
- 在大规模模型上可能面临内存和计算资源挑战
2.3 MIM（动量迭代法）
MIM（Momentum Iterative Method）是在PGD基础上结合动量优化思想的改进方法，通过累积历史梯度信息来稳定和加速对抗样本的生成过程。
2.3.1 算法原理
MIM在每次迭代更新时不仅考虑当前梯度，还引入了历史梯度的指数移动平均，形成动量项，从而在对抗样本生成过程中获得更稳定的更新方向。数学表达式为：
δ 0 = 0 g t + 1 = μ ⋅ g t + ∇ x L ( f θ ( x + δ t ) , y ) ∥ ∇ x L ( f θ ( x + δ t ) , y ) ∥ 1 δ t + 1 = clip δ ∈ [ 0 , 1 ] n , ∥ δ ∥ p ≤ ϵ ( δ t + α ⋅ sign ( g t + 1 ) ) x a d v = x + δ T \begin{aligned} \delta_0 &= 0 \\ g_{t+1} &= \mu \cdot g_t + \frac{\nabla_x \mathcal{L}(f_\theta(x+\delta_t), y)}{\|\nabla_x \mathcal{L}(f_\theta(x+\delta_t), y)\|_1} \\ \delta_{t+1} &= \text{clip}_{\delta \in [0,1]^n, \|\delta\|_p \leq \epsilon} \left( \delta_t + \alpha \cdot \text{sign}(g_{t+1}) \right) \\ x_{adv} &= x + \delta_T \end{aligned} δ0gt+1δt+1xadv=0=μ⋅gt+∥∇xL(fθ(x+δt),y)∥1∇xL(fθ(x+δt),y)=clipδ∈[0,1]n,∥δ∥p≤ϵ(δt+α⋅sign(gt+1))=x+δT
其中， μ \mu μ是动量参数，控制历史梯度的影响程度， g t g_t gt是累积的梯度动量。
2.3.2 算法流程
- 初始化扰动 δ 0 \delta_0 δ0和梯度动量 g 0 g_0 g0（通常为零向量）。
- 对于每一次迭代 t = 0 , 1 , . . . , T − 1 t=0,1,...,T-1 t=0,1,...,T−1：
a. 使用当前扰动 δ t \delta_t δt的样本 x + δ t x+\delta_t x+δt前向传播，计算损失。
b. 计算损失函数对输入的梯度，并归一化。
c. 更新梯度动量： g t + 1 = μ ⋅ g t + normalized ( ∇ x L ) g_{t+1} = \mu \cdot g_t + \text{normalized}(\nabla_x \mathcal{L}) gt+1=μ⋅gt+normalized(∇xL)。
d. 沿着梯度动量的符号方向更新扰动： δ t + 1 = δ t + α ⋅ sign ( g t + 1 ) \delta_{t+1} = \delta_t + \alpha \cdot \text{sign}(g_{t+1}) δt+1=δt+α⋅sign(gt+1)。
e. 投影扰动：将 δ t + 1 \delta_{t+1} δt+1投影到 ℓ p \ell_p ℓp球内。 - 最终对抗样本为 x a d v = x + δ T x_{adv} = x + \delta_T xadv=x+δT。
2.3.3 优缺点分析
优点：
- 生成的对抗样本攻击成功率更高，迁移性更好
- 梯度更新更加稳定，能避免陷入局部最优
- 在相同迭代次数下，通常比PGD表现更好
缺点：
- 计算复杂度与PGD相当，甚至略高
- 引入了额外的动量超参数，需要更多的超参数调优
- 在某些情况下可能导致过拟合特定类型的模型
2.4 TRADES（鲁棒性正则化）
TRADES（Training with TRade-offs between Accuracy and Robustness to adversarial examples）是一种基于正则化的对抗训练方法，通过在目标函数中引入正则化项来平衡标准精度和对抗鲁棒性。
2.4.1 算法原理
TRADES的核心思想是最小化对抗样本与原始样本之间的KL散度，而不仅仅是最大化对抗样本上的损失。数学表达式为：
min θ E ( x , y ) ∼ D [ L ( f θ ( x ) , y ) + β ⋅ max ∥ δ ∥ ≤ ϵ KL ( f θ ( x ) ∥ f θ ( x + δ ) ) ] \min_{\theta} \mathbb{E}_{(x,y) \sim \mathcal{D}} \left[ \mathcal{L}(f_\theta(x), y) + \beta \cdot \max_{\|\delta\| \leq \epsilon} \text{KL}(f_\theta(x) \| f_\theta(x+\delta)) \right] θminE(x,y)∼D[L(fθ(x),y)+β⋅∥δ∥≤ϵmaxKL(fθ(x)∥fθ(x+δ))]
其中， β \beta β是平衡参数，控制鲁棒性正则化的强度， KL ( ⋅ ∥ ⋅ ) \text{KL}(\cdot \| \cdot) KL(⋅∥⋅)是KL散度。
2.4.2 算法流程
- 对于每个训练样本 ( x , y ) (x,y) (x,y)：
a. 使用PGD等方法生成对抗样本 x a d v x_{adv} xadv，最大化 KL ( f θ ( x ) ∥ f θ ( x + δ ) ) \text{KL}(f_\theta(x) \| f_\theta(x+\delta)) KL(fθ(x)∥fθ(x+δ))。
b. 计算原始样本上的损失 L ( f θ ( x ) , y ) \mathcal{L}(f_\theta(x), y) L(fθ(x),y)。
c. 计算正则化项 KL ( f θ ( x ) ∥ f θ ( x a d v ) ) \text{KL}(f_\theta(x) \| f_\theta(x_{adv})) KL(fθ(x)∥fθ(xadv))。
d. 总损失为 L ( f θ ( x ) , y ) + β ⋅ KL ( f θ ( x ) ∥ f θ ( x a d v ) ) \mathcal{L}(f_\theta(x), y) + \beta \cdot \text{KL}(f_\theta(x) \| f_\theta(x_{adv})) L(fθ(x),y)+β⋅KL(fθ(x)∥fθ(xadv))。 - 使用总损失更新模型参数。
2.4.3 优缺点分析
优点：
- 能够更好地平衡标准精度和对抗鲁棒性
- 在不牺牲太多标准性能的情况下提高鲁棒性
- 提供了理论上的鲁棒性保证
缺点：
- 计算复杂度高，需要同时计算标准损失和KL散度
- 对平衡参数 β \beta β的选择敏感
- 在某些情况下可能不如直接最大化对抗损失的方法有效
大语言模型的对抗训练挑战
3.1 高维输入空间的挑战
大语言模型处理的是高维离散的文本输入，这给对抗训练带来了独特的挑战：
-
离散性挑战：与图像不同，文本是离散的符号序列，对抗扰动需要以离散方式进行，这使得传统的连续梯度方法难以直接应用。
-
语义保持难题：添加扰动时需要保持文本的语义正确性，否则生成的对抗样本可能失去意义，无法有效测试模型的鲁棒性。
-
搜索空间巨大：文本的高维性意味着可能的对抗扰动组合数量庞大，难以穷举或高效搜索。
-
非欧氏空间：文本嵌入通常位于非欧氏空间中，传统的 ℓ p \ell_p ℓp范数可能不适合衡量文本扰动的大小。
3.2 计算资源消耗
大语言模型的对抗训练面临着巨大的计算资源挑战：
-
模型规模效应：现代大语言模型通常拥有数十亿甚至数千亿参数，仅一次前向传播就需要大量计算资源。
-
对抗样本生成成本：如PGD等方法需要多次迭代，每次迭代都需要完整的前向和反向传播，计算开销是普通训练的数倍甚至数十倍。
-
内存需求：存储原始样本和对抗样本，以及中间计算结果，需要大量内存空间。
-
训练时间延长：对抗训练通常需要更长的训练时间，这对于大规模模型来说可能是不可接受的。
3.3 文本扰动的特殊性质
文本扰动与图像扰动相比具有独特的性质：
-
语法约束：文本必须遵循语法规则，扰动不能破坏文本的语法结构。
-
语义一致性：对抗样本应保持与原始样本相似的语义，否则就失去了对抗攻击的隐蔽性。
-
多样性：文本扰动可以有多种形式，如字符级扰动（拼写错误）、词级扰动（同义词替换）、句级扰动（句式变换）等。
-
不可感知性：理想的文本对抗样本应对人类读者保持不可感知，这在实践中较难实现。
3.4 鲁棒性与性能的平衡
在大语言模型的对抗训练中，如何平衡鲁棒性与标准性能是一个关键挑战：
-
性能权衡：过度的对抗训练可能导致模型在正常样本上的性能下降。
-
泛化性问题：针对特定类型攻击训练的模型可能对新型攻击仍然脆弱。
-
适应性局限：随着攻击方法的不断演进，防御策略需要持续更新。
-
任务依赖性：不同NLP任务对鲁棒性的要求不同，需要针对性的对抗训练策略。
LLM对抗训练实现技术
4.1 文本扰动生成方法
针对文本的离散特性，研究者们提出了多种文本扰动生成方法：
4.1.1 字符级扰动
字符级扰动通过修改文本中的单个字符来生成对抗样本：
- 随机替换：随机替换文本中的某些字符。
- 键盘邻近替换：替换为键盘上相邻的字符。
- 字符插入/删除：在文本中插入或删除字符。
- 同音字替换：使用发音相似的字符替换。
字符级扰动示例：
- 原始文本：“这是一个对抗训练的示例”
- 对抗文本：“这是一个対抗训练的示例”（“对"→"対”）
4.1.2 词级扰动
词级扰动通过修改、替换或重排文本中的词语来生成对抗样本：
- 同义词替换：使用同义词替换原文中的词语。
- 词性保持替换：替换为相同词性的其他词语。
- 词嵌入空间扰动：在词嵌入空间中寻找语义相似但会导致模型错误的词语。
- 停用词插入/删除：在文本中插入或删除停用词。
词级扰动示例：
- 原始文本：“大语言模型具有强大的生成能力”
- 对抗文本：“大语言模型拥有强大的生成能力”（“具有"→"拥有”）
4.1.3 句级扰动
句级扰动通过调整句子结构或语义表达方式来生成对抗样本：
- 句式变换：将主动句改为被动句，或将陈述句改为疑问句等。
- 否定表达转换：通过双重否定等方式改变表达方式但保持语义。
- 语序调整：调整句子中词语的顺序，保持语法正确。
- 段落重组：重新组织段落结构，但保持核心信息。
句级扰动示例：
- 原始文本：“对抗训练可以提高模型的鲁棒性”
- 对抗文本：“模型的鲁棒性可以通过对抗训练得到提高”
4.1.4 梯度引导的文本扰动
结合梯度信息的文本扰动方法能够更有效地生成对抗样本：
- HotFlip：利用梯度信息找出最可能导致模型预测变化的词替换。
- TextFooler：结合语义相似度和模型梯度，生成高质量的对抗样本。
- BAE (BERT-Attack)：专为预训练语言模型设计的对抗攻击方法。
- WordBug：结合字符级扰动和梯度信息的攻击方法。
梯度引导的文本扰动通常需要模型提供梯度信息，因此主要适用于白盒攻击场景。
4.2 扰动约束与投影技术
为了确保生成的对抗样本与原始样本足够相似，同时保持语法和语义的正确性，需要对扰动施加适当的约束：
4.2.1 语义相似度约束
语义相似度约束确保对抗样本与原始样本在语义上保持一致：
- BLEU分数约束：要求对抗样本与原始样本的BLEU分数高于阈值。
- BERTScore约束：使用预训练语言模型评估语义相似度。
- 余弦相似度约束：要求嵌入向量的余弦相似度高于阈值。
- 编辑距离约束：限制编辑操作的数量或比例。
4.2.2 语法正确性约束
语法正确性约束确保生成的对抗样本在语法上是通顺的：
- 语法检查器过滤：使用语法检查工具过滤不符合语法的样本。
- 语言模型困惑度约束：要求对抗样本的困惑度低于阈值。
- POS标签一致性：保持词性标签的一致性。
- 依存关系保持：尽可能保持原有的句法依存关系。
4.2.3 投影技术
在连续空间的对抗训练中，投影技术用于将扰动限制在特定的约束范围内：
- ℓ ∞ \ell_\infty ℓ∞投影：确保所有维度的扰动绝对值不超过 ϵ \epsilon ϵ。
- ℓ 2 \ell_2 ℓ2投影：确保扰动向量的欧几里得范数不超过 ϵ \epsilon ϵ。
- ℓ 1 \ell_1 ℓ1投影：确保扰动向量的曼哈顿范数不超过 ϵ \epsilon ϵ。
- 正交投影：将扰动投影到特定的子空间中。
在文本对抗训练中，这些连续空间的投影技术通常需要结合离散操作进行近似实现。
4.3 训练目标函数设计
针对大语言模型的特点，需要设计适合的对抗训练目标函数：
4.3.1 基础对抗训练目标
基础的对抗训练目标函数直接最大化对抗样本上的损失：
L a d v = max ∥ δ ∥ ≤ ϵ L ( f θ ( x + δ ) , y ) \mathcal{L}_{adv} = \max_{\|\delta\| \leq \epsilon} \mathcal{L}(f_\theta(x+\delta), y) Ladv=∥δ∥≤ϵmaxL(fθ(x+δ),y)
对于分类任务，通常使用交叉熵损失；对于生成任务，可以使用困惑度或BLEU分数等指标。
4.3.2 混合损失函数
混合损失函数同时考虑原始样本和对抗样本上的损失：
L m i x e d = λ ⋅ L ( f θ ( x ) , y ) + ( 1 − λ ) ⋅ L ( f θ ( x a d v ) , y ) \mathcal{L}_{mixed} = \lambda \cdot \mathcal{L}(f_\theta(x), y) + (1-\lambda) \cdot \mathcal{L}(f_\theta(x_{adv}), y) Lmixed=λ⋅L(fθ(x),y)+(1−λ)⋅L(fθ(xadv),y)
其中， λ \lambda λ是平衡参数，控制原始样本和对抗样本的相对重要性。
4.3.3 鲁棒性正则化损失
鲁棒性正则化通过在目标函数中添加额外的正则化项来增强模型鲁棒性：
L r o b u s t = L ( f θ ( x ) , y ) + β ⋅ R ( θ , x , x a d v ) \mathcal{L}_{robust} = \mathcal{L}(f_\theta(x), y) + \beta \cdot \mathcal{R}(\theta, x, x_{adv}) Lrobust=L(fθ(x),y)+β⋅R(θ,x,xadv)
其中， R \mathcal{R} R是鲁棒性正则化项，常见的包括：
- 梯度正则化：限制损失函数对输入的梯度范数。
- 预测一致性：鼓励模型对原始样本和对抗样本的预测保持一致。
- 平滑性正则化：增强模型在输入空间的局部平滑性。
- 特征正则化：限制模型中间表示对输入扰动的敏感性。
4.3.4 对抗对比学习目标
对抗对比学习结合了对比学习和对抗训练的思想：
L c o n t r a s t = − log e sim ( f θ ( x ) , f θ ( x a d v ) ) / τ ∑ x ′ ∈ N ( x ) e sim ( f θ ( x ) , f θ ( x ′ ) ) / τ \mathcal{L}_{contrast} = -\log \frac{e^{\text{sim}(f_\theta(x), f_\theta(x_{adv}))/\tau}}{\sum_{x' \in \mathcal{N}(x)} e^{\text{sim}(f_\theta(x), f_\theta(x'))/\tau}} Lcontrast=−log∑x′∈N(x)esim(fθ(x),fθ(x′))/τesim(fθ(x),fθ(xadv))/τ
其中， N ( x ) \mathcal{N}(x) N(x)是包含对抗样本和负样本的邻域集合， τ \tau τ是温度参数。
4.4 优化策略与超参数调整
有效的优化策略和超参数调整对于对抗训练的成功至关重要：
4.4.1 优化器选择
- Adam：在对抗训练中，Adam通常是首选的优化器，因为它能够自适应地调整学习率。
- SGD with Momentum：在某些情况下，SGD结合动量可能比Adam具有更好的泛化性能。
- AdamW：结合权重衰减的Adam变体，通常能够提供更好的正则化效果。
- LAMB：针对大规模模型设计的优化器，适合分布式训练。
4.4.2 学习率调度
- 线性预热：在训练初期逐渐增加学习率，有助于稳定训练过程。
- 余弦退火：在训练后期逐渐降低学习率，有助于收敛到更好的局部最优。
- 分段常数衰减：在特定的训练步数后将学习率乘以衰减因子。
- 对抗训练专用调度：针对对抗训练的特点设计的学习率调度策略，如随着训练进行逐步增加对抗强度。
4.4.3 对抗训练特定超参数
- 扰动强度 ϵ \epsilon ϵ：控制对抗扰动的大小，通常需要根据任务和模型进行调整。
- PGD迭代次数：控制对抗样本生成的迭代次数，次数越多生成的样本通常越具攻击性。
- 迭代步长 α \alpha α：控制每次迭代的更新步长，通常设置为 ϵ \epsilon ϵ的一个小比例。
- 动量参数 μ \mu μ：在MIM等算法中使用，控制历史梯度的影响程度。
- 混合比例 λ \lambda λ：控制原始样本和对抗样本在训练中的比例。
4.4.4 训练稳定性技巧
- 梯度裁剪：限制梯度的范数，防止梯度爆炸。
- 梯度累积：通过多次前向和反向传播累积梯度，然后再更新参数。
- 批量规范化：帮助稳定训练过程，特别是在对抗训练中。
- 早停策略：根据验证集性能提前停止训练，防止过拟合。
- 随机种子固定：确保实验的可重复性。
代码实现：基于Transformers的对抗训练
在本节中，我们将提供基于Hugging Face Transformers库的对抗训练完整实现，包括FGSM和PGD两种方法。这些实现可以直接应用于常见的大语言模型微调任务。
5.1 环境配置与依赖安装
首先，我们需要安装必要的依赖包：
# 安装依赖包
# !pip install transformers datasets torch evaluate tqdm
import os
import torch
import numpy as np
from tqdm import tqdm
from transformers import (
AutoTokenizer,
AutoModelForSequenceClassification,
TrainingArguments,
Trainer,
DataCollatorWithPadding
)
from datasets import load_dataset
import evaluate
5.2 基本数据处理
我们将使用GLUE基准测试中的SST-2数据集作为示例：
def load_and_preprocess_data(model_name="bert-base-uncased"):
"""加载并预处理SST-2数据集"""
# 加载数据集
dataset = load_dataset("glue", "sst2")
# 加载tokenizer
tokenizer = AutoTokenizer.from_pretrained(model_name)
# 预处理函数
def preprocess_function(examples):
return tokenizer(examples["sentence"], truncation=True)
# 应用预处理
tokenized_datasets = dataset.map(preprocess_function, batched=True)
# 创建数据收集器
data_collator = DataCollatorWithPadding(tokenizer=tokenizer)
return tokenized_datasets, tokenizer, data_collator
# 加载数据集
tokenized_datasets, tokenizer, data_collator = load_and_preprocess_data()
5.3 FGSM对抗训练实现
FGSM（快速梯度符号法）是一种简单但有效的对抗训练方法。以下是在Transformers中实现FGSM对抗训练的代码：
class FGSMTrainer(Trainer):
"""实现FGSM对抗训练的Trainer类"""
def __init__(self, *args, epsilon=0.1, **kwargs):
super().__init__(*args, **kwargs)
self.epsilon = epsilon # 扰动强度
def compute_loss(self, model, inputs, return_outputs=False):
"""计算对抗训练损失"""
# 保存原始输入
original_inputs = {k: v.clone() for k, v in inputs.items() if isinstance(v, torch.Tensor)}
# 启用梯度计算
for k in inputs:
if isinstance(inputs[k], torch.Tensor):
inputs[k].requires_grad = True
# 前向传播计算初始损失
outputs = model(**inputs)
loss = outputs.loss
# 反向传播计算输入梯度
self.model.zero_grad()
loss.backward(retain_graph=True)
# 生成对抗样本
perturbed_inputs = {k: v.clone() for k, v in inputs.items()}
# 对input_ids应用FGSM扰动
if "input_ids" in perturbed_inputs and perturbed_inputs["input_ids"].grad is not None:
# 获取梯度的符号
gradient = torch.sign(perturbed_inputs["input_ids"].grad)
# 生成对抗扰动
perturbed_inputs["input_ids"] = perturbed_inputs["input_ids"] + self.epsilon * gradient
# 确保扰动后的token IDs在有效范围内
perturbed_inputs["input_ids"] = torch.clamp(
perturbed_inputs["input_ids"],
min=tokenizer.vocab_size - 1,
max=tokenizer.vocab_size - 1
)
# 转换为整数类型
perturbed_inputs["input_ids"] = perturbed_inputs["input_ids"].to(torch.long)
# 对attention_mask应用FGSM扰动（可选）
if "attention_mask" in perturbed_inputs and perturbed_inputs["attention_mask"].grad is not None:
gradient = torch.sign(perturbed_inputs["attention_mask"].grad)
perturbed_inputs["attention_mask"] = perturbed_inputs["attention_mask"] + self.epsilon * gradient
perturbed_inputs["attention_mask"] = torch.clamp(perturbed_inputs["attention_mask"], min=0, max=1)
perturbed_inputs["attention_mask"] = perturbed_inputs["attention_mask"].to(torch.long)
# 使用对抗样本计算损失
with torch.no_grad():
adv_outputs = model(**perturbed_inputs)
adv_loss = adv_outputs.loss
# 混合损失：原始损失 + 对抗损失
final_loss = (loss + adv_loss) / 2
# 恢复原始输入
for k in original_inputs:
inputs[k] = original_inputs[k]
return (final_loss, adv_outputs) if return_outputs else final_loss
# 创建评估函数
def compute_metrics(eval_pred):
logits, labels = eval_pred
predictions = np.argmax(logits, axis=-1)
return {
"accuracy": (predictions == labels).mean(),
"precision": evaluate.load("precision").compute(predictions=predictions, references=labels, average="macro")["precision"],
"recall": evaluate.load("recall").compute(predictions=predictions, references=labels, average="macro")["recall"],
"f1": evaluate.load("f1").compute(predictions=predictions, references=labels, average="macro")["f1"]
}
# 初始化模型
model = AutoModelForSequenceClassification.from_pretrained("bert-base-uncased", num_labels=2)
# 设置训练参数
training_args = TrainingArguments(
output_dir="./fgsm_adversarial_training",
learning_rate=2e-5,
per_device_train_batch_size=16,
per_device_eval_batch_size=16,
num_train_epochs=3,
weight_decay=0.01,
evaluation_strategy="epoch",
save_strategy="epoch",
load_best_model_at_end=True,
logging_dir="./logs",
)
# 创建FGSM训练器
fgsm_trainer = FGSMTrainer(
model=model,
args=training_args,
train_dataset=tokenized_datasets["train"],
eval_dataset=tokenized_datasets["validation"],
tokenizer=tokenizer,
data_collator=data_collator,
compute_metrics=compute_metrics,
epsilon=0.1, # 设置FGSM扰动强度
)
# 开始训练
# fgsm_trainer.train()
5.4 PGD对抗训练实现
PGD（投影梯度下降）是一种比FGSM更强大的对抗训练方法，通过多次迭代生成对抗样本。以下是PGD对抗训练的实现：
class PGDPerturbation:
"""PGD扰动生成器"""
def __init__(self, model, epsilon=0.1, alpha=0.05, steps=5, norm="linf"):
self.model = model
self.epsilon = epsilon # 最大扰动幅度
self.alpha = alpha # 迭代步长
self.steps = steps # 迭代次数
self.norm = norm # 范数类型 ("linf", "l2")
def generate(self, inputs, labels):
"""生成PGD对抗样本"""
# 保存原始输入
original_inputs = {k: v.clone() for k, v in inputs.items() if isinstance(v, torch.Tensor)}
# 初始化扰动（小的随机噪声）
perturbed_inputs = {k: v.clone() for k, v in inputs.items()}
# 对input_ids添加随机初始扰动
if "input_ids" in perturbed_inputs:
noise = torch.randn_like(perturbed_inputs["input_ids"], dtype=torch.float)
noise = noise * self.epsilon / noise.norm(p=float("inf")) if self.norm == "linf" else \
noise * self.epsilon / (noise.norm(p=2) + 1e-12)
perturbed_inputs["input_ids"] = perturbed_inputs["input_ids"] + noise
perturbed_inputs["input_ids"] = torch.clamp(
perturbed_inputs["input_ids"],
min=0,
max=tokenizer.vocab_size - 1
)
# PGD迭代
for _ in range(self.steps):
# 启用梯度计算
for k in perturbed_inputs:
if isinstance(perturbed_inputs[k], torch.Tensor):
perturbed_inputs[k].requires_grad = True
# 前向传播
outputs = self.model(**perturbed_inputs, labels=labels)
loss = outputs.loss
# 反向传播
self.model.zero_grad()
loss.backward(retain_graph=True)
# 更新扰动
if "input_ids" in perturbed_inputs and perturbed_inputs["input_ids"].grad is not None:
grad = perturbed_inputs["input_ids"].grad
# 根据范数类型更新
if self.norm == "linf":
# L-infinity范数更新
perturbed_inputs["input_ids"] = perturbed_inputs["input_ids"] + self.alpha * torch.sign(grad)
else: # L2范数
# L2范数更新
perturbed_inputs["input_ids"] = perturbed_inputs["input_ids"] + self.alpha * grad / (grad.norm(p=2, dim=-1, keepdim=True) + 1e-12)
# 投影回epsilon球
perturbation = perturbed_inputs["input_ids"] - original_inputs["input_ids"]
if self.norm == "linf":
perturbation = torch.clamp(perturbation, -self.epsilon, self.epsilon)
else:
perturbation_norm = perturbation.norm(p=2, dim=-1, keepdim=True)
perturbation = perturbation * torch.minimum(
torch.tensor(1.0).to(perturbation.device),
self.epsilon / (perturbation_norm + 1e-12)
)
perturbed_inputs["input_ids"] = original_inputs["input_ids"] + perturbation
# 确保在有效范围内
perturbed_inputs["input_ids"] = torch.clamp(
perturbed_inputs["input_ids"],
min=0,
max=tokenizer.vocab_size - 1
)
perturbed_inputs["input_ids"] = perturbed_inputs["input_ids"].to(torch.long)
return perturbed_inputs
class PGDPerturbationOnEmbeddings:
"""在嵌入空间上执行PGD扰动"""
def __init__(self, model, epsilon=0.1, alpha=0.05, steps=5, norm="linf"):
self.model = model
self.epsilon = epsilon
self.alpha = alpha
self.steps = steps
self.norm = norm
# 保存原始嵌入权重的引用
self.original_embeddings = self.model.get_input_embeddings().weight.data.clone()
def generate(self, inputs, labels):
"""生成对抗样本"""
# 保存原始嵌入
original_embeddings = self.model.get_input_embeddings().weight.data.clone()
# 初始化扰动
noise = torch.randn_like(original_embeddings) * 0.001
if self.norm == "linf":
noise = noise * self.epsilon / noise.abs().max()
else:
noise = noise * self.epsilon / (noise.norm(p=2) + 1e-12)
# PGD迭代
for _ in range(self.steps):
# 添加扰动
self.model.get_input_embeddings().weight.data = original_embeddings + noise
# 计算损失和梯度
self.model.zero_grad()
outputs = self.model(**inputs, labels=labels)
loss = outputs.loss
loss.backward()
# 获取嵌入梯度
grad = self.model.get_input_embeddings().weight.grad
# 更新噪声
if self.norm == "linf":
noise = noise + self.alpha * torch.sign(grad)
noise = torch.clamp(noise, -self.epsilon, self.epsilon)
else: # L2范数
noise = noise + self.alpha * grad / (grad.norm(p=2) + 1e-12)
noise = noise * torch.minimum(
torch.tensor(1.0).to(noise.device),
self.epsilon / (noise.norm(p=2) + 1e-12)
)
# 应用最终扰动
self.model.get_input_embeddings().weight.data = original_embeddings + noise
# 在嵌入空间扰动后，输入保持不变
return inputs
def reset(self):
"""重置嵌入权重"""
self.model.get_input_embeddings().weight.data = self.original_embeddings.clone()
class PGDPerturbationTextAttack:
"""基于文本编辑的PGD扰动"""
def __init__(self, model, tokenizer, epsilon=0.1, alpha=0.05, steps=5, num_perturbations=3):
self.model = model
self.tokenizer = tokenizer
self.epsilon = epsilon
self.alpha = alpha
self.steps = steps
self.num_perturbations = num_perturbations # 每个样本最多修改的token数量
def generate(self, inputs, labels):
"""生成文本对抗样本"""
batch_size = inputs["input_ids"].shape[0]
max_length = inputs["input_ids"].shape[1]
perturbed_input_ids = inputs["input_ids"].clone()
# 对每个样本生成对抗样本
for i in range(batch_size):
# 获取当前样本的input_ids
sample_ids = perturbed_input_ids[i].cpu().numpy()
# 找出非填充token的位置
non_pad_positions = [j for j in range(max_length) if sample_ids[j] != self.tokenizer.pad_token_id]
# 确定要修改的token数量
num_to_modify = min(self.num_perturbations, len(non_pad_positions))
# 选择要修改的token位置
positions_to_modify = np.random.choice(non_pad_positions, num_to_modify, replace=False)
# 对每个选中的位置生成扰动
for pos in positions_to_modify:
# 尝试不同的token替换
best_replacement = sample_ids[pos]
max_loss = -float("inf")
# 生成候选替换token
# 这里我们简单地随机选择几个候选token
# 在实际应用中，可以使用更智能的方法，如基于词嵌入的相似词
vocab_size = len(self.tokenizer.get_vocab())
candidate_tokens = np.random.randint(0, vocab_size, size=100)
# 评估每个候选token
for token in candidate_tokens:
if token == sample_ids[pos]:
continue
# 创建修改后的输入
modified_ids = sample_ids.copy()
modified_ids[pos] = token
# 转换回tensor
modified_tensor = torch.tensor([modified_ids], device=inputs["input_ids"].device)
modified_inputs = {
"input_ids": modified_tensor,
"attention_mask": inputs["attention_mask"][i:i+1]
}
# 计算损失
with torch.no_grad():
outputs = self.model(**modified_inputs, labels=labels[i:i+1])
current_loss = outputs.loss.item()
# 更新最佳替换
if current_loss > max_loss:
max_loss = current_loss
best_replacement = token
# 应用最佳替换
sample_ids[pos] = best_replacement
# 更新批次中的样本
perturbed_input_ids[i] = torch.tensor(sample_ids, device=inputs["input_ids"].device)
# 创建扰动后的输入
perturbed_inputs = {k: v.clone() for k, v in inputs.items()}
perturbed_inputs["input_ids"] = perturbed_input_ids
return perturbed_inputs
class PGDPerturbationEmbeddingSpace:
"""在嵌入空间进行PGD扰动，然后转换回token空间"""
def __init__(self, model, tokenizer, epsilon=0.1, alpha=0.05, steps=5):
self.model = model
self.tokenizer = tokenizer
self.epsilon = epsilon
self.alpha = alpha
self.steps = steps
# 获取嵌入层
self.embedding_layer = model.get_input_embeddings()
# 获取词汇表
self.vocab = tokenizer.get_vocab()
self.inv_vocab = {v: k for k, v in self.vocab.items()}
def token_to_embedding(self, tokens):
"""将token转换为嵌入向量"""
return self.embedding_layer(tokens)
def embedding_to_token(self, embeddings):
"""将嵌入向量转换回最接近的token"""
# 获取所有词汇的嵌入
all_embeddings = self.embedding_layer.weight
# 计算输入嵌入与所有词汇嵌入的相似度
embeddings = embeddings.squeeze(0) # 移除批次维度
batch_size, seq_len, emb_dim = embeddings.shape
all_embeddings = all_embeddings.unsqueeze(0).unsqueeze(0) # 扩展维度以便广播
# 计算余弦相似度
# 归一化
embeddings_norm = embeddings / (embeddings.norm(dim=-1, keepdim=True) + 1e-12)
all_embeddings_norm = all_embeddings / (all_embeddings.norm(dim=-1, keepdim=True) + 1e-12)
# 计算相似度
similarities = torch.matmul(embeddings_norm, all_embeddings_norm.squeeze(0).transpose(0, 1))
# 找到最相似的token
nearest_tokens = similarities.argmax(dim=-1)
return nearest_tokens
def generate(self, inputs, labels):
"""生成对抗样本"""
input_ids = inputs["input_ids"].clone()
# 获取当前嵌入
with torch.no_grad():
embeddings = self.token_to_embedding(input_ids)
# 初始化扰动
noise = torch.randn_like(embeddings) * 0.001
noise = noise.to(input_ids.device)
# PGD迭代
for _ in range(self.steps):
# 添加扰动
perturbed_embeddings = embeddings + noise
# 将扰动后的嵌入转换回token
perturbed_input_ids = self.embedding_to_token(perturbed_embeddings)
# 创建输入字典
perturbed_inputs = {
"input_ids": perturbed_input_ids,
"attention_mask": inputs["attention_mask"]
}
# 计算损失和梯度
outputs = self.model(**perturbed_inputs, labels=labels)
loss = outputs.loss
self.model.zero_grad()
loss.backward()
# 获取嵌入的梯度（通过反向传播计算）
# 由于我们不能直接对嵌入求导，我们需要使用一个技巧
# 1. 我们需要将perturbed_input_ids转换回嵌入
# 2. 然后计算这些嵌入对原始输入的梯度
with torch.no_grad():
# 获取扰动后token的嵌入
perturbed_embeddings_actual = self.token_to_embedding(perturbed_input_ids)
# 近似梯度
# 这里我们使用一个简化的方法，实际应用中可能需要更复杂的技术
grad = perturbed_embeddings_actual - embeddings
# 更新噪声
noise = noise + self.alpha * torch.sign(grad)
noise = torch.clamp(noise, -self.epsilon, self.epsilon)
# 生成最终的对抗样本
final_perturbed_embeddings = embeddings + noise
final_perturbed_input_ids = self.embedding_to_token(final_perturbed_embeddings)
# 创建扰动后的输入
perturbed_inputs = {
"input_ids": final_perturbed_input_ids,
"attention_mask": inputs["attention_mask"]
}
return perturbed_inputs
class PGDPerturbationMixed:
"""混合多种PGD扰动方法"""
def __init__(self, model, tokenizer, epsilon=0.1, alpha=0.05, steps=5, method_weights=[0.3, 0.3, 0.4]):
self.model = model
self.tokenizer = tokenizer
self.epsilon = epsilon
self.alpha = alpha
self.steps = steps
self.method_weights = method_weights
# 初始化不同的扰动方法
self.methods = [
PGDPerturbation(model, epsilon, alpha, steps),
PGDPerturbationOnEmbeddings(model, epsilon, alpha, steps),
PGDPerturbationTextAttack(model, tokenizer, epsilon, alpha, steps)
]
def generate(self, inputs, labels):
"""生成混合对抗样本"""
# 随机选择一种方法
method_idx = np.random.choice(len(self.methods), p=self.method_weights)
perturbed_inputs = self.methods[method_idx].generate(inputs, labels)
# 如果使用了嵌入层扰动，需要重置
if method_idx == 1:
self.methods[method_idx].reset()
return perturbed_inputs
class PGDPerturbationAdvanced:
"""高级PGD扰动，结合多种策略"""
def __init__(self, model, tokenizer, epsilon=0.1, alpha=0.05, steps=5, strategy="adaptive"):
self.model = model
self.tokenizer = tokenizer
self.epsilon = epsilon
self.alpha = alpha
self.steps = steps
self.strategy = strategy # "adaptive", "hybrid", "semantic"
def generate(self, inputs, labels):
"""生成高级对抗样本"""
if self.strategy == "adaptive":
return self._adaptive_perturbation(inputs, labels)
elif self.strategy == "hybrid":
return self._hybrid_perturbation(inputs, labels)
elif self.strategy == "semantic":
return self._semantic_perturbation(inputs, labels)
else:
raise ValueError(f"Unknown strategy: {self.strategy}")
def _adaptive_perturbation(self, inputs, labels):
"""自适应扰动策略，根据模型信心调整扰动强度"""
# 获取模型预测
with torch.no_grad():
outputs = self.model(**inputs)
logits = outputs.logits
probs = torch.softmax(logits, dim=-1)
confidences, _ = probs.max(dim=-1)
# 自适应调整扰动强度
adaptive_epsilon = []
for conf in confidences:
if conf > 0.9:
adaptive_epsilon.append(self.epsilon * 1.5) # 对高置信度样本增加扰动
elif conf < 0.7:
adaptive_epsilon.append(self.epsilon * 0.5) # 对低置信度样本减少扰动
else:
adaptive_epsilon.append(self.epsilon)
adaptive_epsilon = torch.tensor(adaptive_epsilon, device=inputs["input_ids"].device)
# 这里省略具体实现，实际应用中需要根据每个样本的置信度调整扰动
# 暂时返回原始PGD扰动
pgd = PGDPerturbation(self.model, self.epsilon, self.alpha, self.steps)
return pgd.generate(inputs, labels)
def _hybrid_perturbation(self, inputs, labels):
"""混合扰动策略，结合不同级别的扰动"""
# 结合字符级、词级和嵌入级扰动
# 这里省略具体实现
pgd = PGDPerturbationMixed(self.model, self.tokenizer, self.epsilon, self.alpha, self.steps)
return pgd.generate(inputs, labels)
def _semantic_perturbation(self, inputs, labels):
"""语义保持的扰动策略"""
# 使用词嵌入空间的相似词替换，保持语义一致性
# 这里省略具体实现
pgd = PGDPerturbationTextAttack(self.model, self.tokenizer, self.epsilon, self.alpha, self.steps)
return pgd.generate(inputs, labels)
class PGDTrainer(Trainer):
"""实现PGD对抗训练的Trainer类"""
def __init__(self, *args, epsilon=0.1, alpha=0.05, steps=5, perturbation_method="standard", **kwargs):
super().__init__(*args, **kwargs)
self.epsilon = epsilon
self.alpha = alpha
self.steps = steps
self.perturbation_method = perturbation_method
# 初始化扰动生成器
self._init_perturbator()
def _init_perturbator(self):
"""初始化扰动生成器"""
if self.perturbation_method == "standard":
self.perturbator = PGDPerturbation(
self.model,
epsilon=self.epsilon,
alpha=self.alpha,
steps=self.steps
)
elif self.perturbation_method == "embeddings":
self.perturbator = PGDPerturbationOnEmbeddings(
self.model,
epsilon=self.epsilon,
alpha=self.alpha,
steps=self.steps
)
elif self.perturbation_method == "text":
self.perturbator = PGDPerturbationTextAttack(
self.model,
self.tokenizer,
epsilon=self.epsilon,
alpha=self.alpha,
steps=self.steps
)
elif self.perturbation_method == "embedding_space":
self.perturbator = PGDPerturbationEmbeddingSpace(
self.model,
self.tokenizer,
epsilon=self.epsilon,
alpha=self.alpha,
steps=self.steps
)
elif self.perturbation_method == "mixed":
self.perturbator = PGDPerturbationMixed(
self.model,
self.tokenizer,
epsilon=self.epsilon,
alpha=self.alpha,
steps=self.steps
)
elif self.perturbation_method == "advanced":
self.perturbator = PGDPerturbationAdvanced(
self.model,
self.tokenizer,
epsilon=self.epsilon,
alpha=self.alpha,
steps=self.steps
)
else:
raise ValueError(f"Unknown perturbation method: {self.perturbation_method}")
def compute_loss(self, model, inputs, return_outputs=False):
"""计算对抗训练损失"""
# 确保我们有权访问标签
if "labels" not in inputs:
raise ValueError("PGD adversarial training requires labels")
# 获取标签
labels = inputs["labels"]
# 生成对抗样本
perturbed_inputs = self.perturbator.generate(inputs, labels)
# 使用对抗样本计算损失
adv_outputs = model(**perturbed_inputs)
adv_loss = adv_outputs.loss
# 使用原始样本计算损失
original_outputs = model(**inputs)
original_loss = original_outputs.loss
# 混合损失
final_loss = (original_loss + adv_loss) / 2
# 如果使用了嵌入层扰动，重置嵌入
if self.perturbation_method == "embeddings":
self.perturbator.reset()
return (final_loss, adv_outputs) if return_outputs else final_loss
# 创建PGD训练器
pgd_trainer = PGDTrainer(
model=model,
args=training_args,
train_dataset=tokenized_datasets["train"],
eval_dataset=tokenized_datasets["validation"],
tokenizer=tokenizer,
data_collator=data_collator,
compute_metrics=compute_metrics,
epsilon=0.1,
alpha=0.02,
steps=5,
perturbation_method="text" # 使用基于文本的扰动方法
)
# 开始训练
# pgd_trainer.train()
5.5 混合对抗训练策略
在实际应用中，我们可以结合多种对抗训练方法，以获得更好的鲁棒性提升。以下是一个混合对抗训练的实现：
class MixedAdversarialTrainer(Trainer):
"""混合多种对抗训练方法的Trainer类"""
def __init__(self, *args, methods=None, method_weights=None, **kwargs):
super().__init__(*args, **kwargs)
# 定义可用的对抗训练方法
self.methods = methods or ["fgsm", "pgd", "mim"]
self.method_weights = method_weights or [0.4, 0.4, 0.2]
# 确保权重和为1
if sum(self.method_weights) != 1.0:
self.method_weights = [w / sum(self.method_weights) for w in self.method_weights]
# 设置方法特定的参数
self.method_params = {
"fgsm": {"epsilon": 0.1},
"pgd": {"epsilon": 0.1, "alpha": 0.02, "steps": 5},
"mim": {"epsilon": 0.1, "alpha": 0.02, "steps": 5, "mu": 0.9}
}
def _generate_fgsm(self, inputs, labels):
"""生成FGSM对抗样本"""
# 启用梯度计算
for k in inputs:
if isinstance(inputs[k], torch.Tensor):
inputs[k].requires_grad = True
# 前向传播
outputs = self.model(**inputs, labels=labels)
loss = outputs.loss
# 反向传播
self.model.zero_grad()
loss.backward(retain_graph=True)
# 应用FGSM
perturbed_inputs = {k: v.clone() for k, v in inputs.items()}
if "input_ids" in perturbed_inputs and perturbed_inputs["input_ids"].grad is not None:
gradient = torch.sign(perturbed_inputs["input_ids"].grad)
perturbed_inputs["input_ids"] = perturbed_inputs["input_ids"] + self.method_params["fgsm"]["epsilon"] * gradient
perturbed_inputs["input_ids"] = torch.clamp(
perturbed_inputs["input_ids"],
min=0,
max=self.tokenizer.vocab_size - 1
)
perturbed_inputs["input_ids"] = perturbed_inputs["input_ids"].to(torch.long)
return perturbed_inputs
def _generate_pgd(self, inputs, labels):
"""生成PGD对抗样本"""
# 使用之前定义的PGDPerturbation类
pgd = PGDPerturbation(
self.model,
epsilon=self.method_params["pgd"]["epsilon"],
alpha=self.method_params["pgd"]["alpha"],
steps=self.method_params["pgd"]["steps"]
)
return pgd.generate(inputs, labels)
def _generate_mim(self, inputs, labels):
"""生成MIM（动量迭代法）对抗样本"""
# 保存原始输入
original_inputs = {k: v.clone() for k, v in inputs.items() if isinstance(v, torch.Tensor)}
# 初始化扰动和动量
perturbed_inputs = {k: v.clone() for k, v in inputs.items()}
momentum = {k: torch.zeros_like(v) for k, v in perturbed_inputs.items() if isinstance(v, torch.Tensor)}
# MIM迭代
for _ in range(self.method_params["mim"]["steps"]):
# 启用梯度计算
for k in perturbed_inputs:
if isinstance(perturbed_inputs[k], torch.Tensor):
perturbed_inputs[k].requires_grad = True
# 前向传播
outputs = self.model(**perturbed_inputs, labels=labels)
loss = outputs.loss
# 反向传播
self.model.zero_grad()
loss.backward(retain_graph=True)
# 更新动量和扰动
if "input_ids" in perturbed_inputs and perturbed_inputs["input_ids"].grad is not None:
# 归一化梯度
grad = perturbed_inputs["input_ids"].grad / (perturbed_inputs["input_ids"].grad.abs().sum() + 1e-12)
# 更新动量
momentum["input_ids"] = self.method_params["mim"]["mu"] * momentum["input_ids"] + grad
# 更新扰动
perturbed_inputs["input_ids"] = perturbed_inputs["input_ids"] + self.method_params["mim"]["alpha"] * torch.sign(momentum["input_ids"])
# 投影回epsilon球
perturbation = perturbed_inputs["input_ids"] - original_inputs["input_ids"]
perturbation = torch.clamp(perturbation, -self.method_params["mim"]["epsilon"], self.method_params["mim"]["epsilon"])
perturbed_inputs["input_ids"] = original_inputs["input_ids"] + perturbation
# 确保在有效范围内
perturbed_inputs["input_ids"] = torch.clamp(
perturbed_inputs["input_ids"],
min=0,
max=self.tokenizer.vocab_size - 1
)
perturbed_inputs["input_ids"] = perturbed_inputs["input_ids"].to(torch.long)
return perturbed_inputs
def compute_loss(self, model, inputs, return_outputs=False):
"""计算混合对抗训练损失"""
if "labels" not in inputs:
raise ValueError("Adversarial training requires labels")
labels = inputs["labels"]
# 随机选择一种对抗训练方法
method = np.random.choice(self.methods, p=self.method_weights)
# 根据选择的方法生成对抗样本
if method == "fgsm":
perturbed_inputs = self._generate_fgsm(inputs.copy(), labels)
elif method == "pgd":
perturbed_inputs = self._generate_pgd(inputs.copy(), labels)
elif method == "mim":
perturbed_inputs = self._generate_mim(inputs.copy(), labels)
else:
raise ValueError(f"Unknown method: {method}")
# 计算原始样本和对抗样本的损失
original_outputs = model(**inputs)
original_loss = original_outputs.loss
adv_outputs = model(**perturbed_inputs)
adv_loss = adv_outputs.loss
# 混合损失
final_loss = (original_loss + adv_loss) / 2
return (final_loss, adv_outputs) if return_outputs else final_loss
# 创建混合对抗训练器
mixed_trainer = MixedAdversarialTrainer(
model=model,
args=training_args,
train_dataset=tokenized_datasets["train"],
eval_dataset=tokenized_datasets["validation"],
tokenizer=tokenizer,
data_collator=data_collator,
compute_metrics=compute_metrics,
methods=["fgsm", "pgd", "mim"],
method_weights=[0.3, 0.5, 0.2]
)
# 开始训练
# mixed_trainer.train()
# 测试对抗训练效果
def test_adversarial_robustness(trainer, test_dataset, num_samples=100):
"""测试模型对对抗样本的鲁棒性"""
# 随机选择样本
indices = np.random.choice(len(test_dataset), min(num_samples, len(test_dataset)), replace=False)
test_samples = [test_dataset[i] for i in indices]
# 转换为批次
batch = trainer.tokenizer.pad(
{k: [sample[k] for sample in test_samples] for k in test_samples[0]},
return_tensors="pt"
)
batch = {k: v.to(trainer.model.device) for k, v in batch.items()}
# 生成对抗样本
if hasattr(trainer, 'perturbator'):
perturbed_batch = trainer.perturbator.generate(batch, batch["labels"])
else:
# 使用默认的PGD扰动
pgd = PGDPerturbation(
trainer.model,
epsilon=0.1,
alpha=0.02,
steps=5
)
perturbed_batch = pgd.generate(batch, batch["labels"])
# 评估原始样本
with torch.no_grad():
original_outputs = trainer.model(**batch)
original_predictions = torch.argmax(original_outputs.logits, dim=-1)
# 评估对抗样本
with torch.no_grad():
perturbed_outputs = trainer.model(**perturbed_batch)
perturbed_predictions = torch.argmax(perturbed_outputs.logits, dim=-1)
# 计算准确率
original_accuracy = (original_predictions == batch["labels"]).float().mean().item()
perturbed_accuracy = (perturbed_predictions == batch["labels"]).float().mean().item()
# 计算鲁棒性指标
robustness = perturbed_accuracy / original_accuracy if original_accuracy > 0 else 0
print(f"原始样本准确率: {original_accuracy:.4f}")
print(f"对抗样本准确率: {perturbed_accuracy:.4f}")
print(f"鲁棒性指标: {robustness:.4f}")
return {
"original_accuracy": original_accuracy,
"perturbed_accuracy": perturbed_accuracy,
"robustness": robustness
}
# 运行鲁棒性测试
# results = test_adversarial_robustness(mixed_trainer, tokenized_datasets["test"])
5.6 对抗训练的文本实现技巧
在实际应用中，针对文本数据的对抗训练需要注意以下几点技巧：
-
离散扰动处理：文本是离散的，需要特殊处理才能应用基于梯度的方法。可以考虑在嵌入空间进行扰动，然后映射回最接近的token。
-
语义保持：对抗样本应尽可能保持语义一致性，可以使用词嵌入空间中的相似词替换，或者限制扰动只发生在非关键token上。
-
计算效率优化：针对大模型，对抗训练的计算成本很高，可以考虑使用梯度累积、批量梯度裁剪等技术优化训练过程。
-
学习率调整：对抗训练通常需要更小的学习率和更保守的学习率调度策略。
-
扰动强度自适应：根据样本的难度和模型的信心动态调整扰动强度。
6. 对抗训练与噪声鲁棒性提升
6.1 噪声鲁棒性的理论基础
噪声鲁棒性是指模型在面对输入数据中的噪声、扰动或异常时保持稳定性能的能力。对抗训练通过在训练过程中主动引入扰动，增强模型的鲁棒性，这一过程可以从以下几个理论角度理解：
噪声鲁棒性增强机制：
输入 → 标准训练 → 模型输出
↓
噪声/扰动 → 对抗训练 → 更鲁棒的模型输出
↓
预测稳定性提升
理论基础公式：
对于一个模型 ( f( heta, x) )，其中 ( heta ) 是模型参数，( x ) 是输入，对抗训练的目标可以表示为：
min h e t a 1 n ∑ i = 1 n E δ ∈ D [ L ( f ( h e t a , x i + δ ) , y i ) ] \min_{ heta} \frac{1}{n} \sum_{i=1}^{n} \mathbb{E}_{\delta \in \mathcal{D}} [L(f( heta, x_i + \delta), y_i)] hetaminn1i=1∑nEδ∈D[L(f(heta,xi+δ),yi)]
其中 ( \mathcal{D} ) 是扰动集合，通常定义为 ( \delta \in \mathcal{D} \Leftrightarrow ||\delta||_p \leq \epsilon )。
鲁棒性边界与泛化能力关系：
研究表明，对抗训练可以减小模型的鲁棒性边界（robustness margin），从而提升模型在未见数据上的泛化能力。这种关系可以用以下不等式表示：
KaTeX parse error: Unexpected character: '' at position 98: …\delta), y)] + ̲eta( heta)
其中 ( eta( heta) ) 是一个与模型复杂度相关的项。
6.2 不同噪声类型的鲁棒性评估
在评估模型的噪声鲁棒性时，需要考虑多种噪声类型，以下是常见的噪声类型及其评估方法：
| 噪声类型 | 描述 | 评估方法 | 扰动策略 |
|---|---|---|---|
| 对抗噪声 | 针对性设计的微小扰动 | 对抗样本准确率下降率 | FGSM、PGD、MIM等 |
| 随机噪声 | 随机添加的干扰 | 噪声准确率 | 添加高斯噪声、均匀噪声 |
| 自然噪声 | 真实场景中的噪声 | 自然扰动准确率 | 拼写错误、语法错误、OCR错误 |
| 结构化噪声 | 特定模式的噪声 | 特定任务表现 | 领域迁移、分布偏移 |
噪声鲁棒性评估代码实现：
def evaluate_noise_robustness(model, tokenizer, test_dataset, noise_types=None):
"""评估模型对不同类型噪声的鲁棒性"""
if noise_types is None:
noise_types = ["adversarial", "random", "natural"]
results = {}
# 准备测试数据
indices = np.random.choice(len(test_dataset), min(100, len(test_dataset)), replace=False)
test_samples = [test_dataset[i] for i in indices]
batch = tokenizer.pad(
{k: [sample[k] for sample in test_samples] for k in test_samples[0]},
return_tensors="pt"
)
batch = {k: v.to(model.device) for k, v in batch.items()}
# 原始准确率
with torch.no_grad():
original_outputs = model(**batch)
original_predictions = torch.argmax(original_outputs.logits, dim=-1)
original_accuracy = (original_predictions == batch["labels"]).float().mean().item()
results["original"] = original_accuracy
# 不同噪声类型的鲁棒性
for noise_type in noise_types:
if noise_type == "adversarial":
# 使用PGD生成对抗样本
pgd = PGDPerturbation(model, epsilon=0.1, alpha=0.02, steps=5)
perturbed_batch = pgd.generate(batch, batch["labels"])
elif noise_type == "random":
# 添加随机噪声
perturbed_batch = {k: v.clone() for k, v in batch.items()}
if "input_ids" in perturbed_batch:
# 随机替换5%的token
mask = torch.rand(perturbed_batch["input_ids"].shape) < 0.05
random_tokens = torch.randint(0, tokenizer.vocab_size - 1,
size=perturbed_batch["input_ids"].shape,
device=model.device)
perturbed_batch["input_ids"][mask] = random_tokens[mask]
elif noise_type == "natural":
# 模拟自然噪声（如拼写错误）
perturbed_batch = simulate_natural_noise(batch, tokenizer)
# 评估鲁棒性
with torch.no_grad():
perturbed_outputs = model(**perturbed_batch)
perturbed_predictions = torch.argmax(perturbed_outputs.logits, dim=-1)
perturbed_accuracy = (perturbed_predictions == batch["labels"]).float().mean().item()
results[noise_type] = {
"accuracy": perturbed_accuracy,
"robustness": perturbed_accuracy / original_accuracy
}
return results
def simulate_natural_noise(batch, tokenizer, error_rate=0.1):
"""模拟自然文本噪声（如拼写错误）"""
perturbed_batch = {k: v.clone() for k, v in batch.items()}
input_ids = perturbed_batch["input_ids"].cpu().numpy()
for i in range(input_ids.shape[0]):
for j in range(input_ids.shape[1]):
# 跳过填充token
if input_ids[i, j] == tokenizer.pad_token_id:
continue
# 以一定概率引入错误
if np.random.random() < error_rate:
# 获取当前token的文本
token = tokenizer.decode([input_ids[i, j]])
# 根据token长度选择错误类型
if len(token) > 2:
# 交换相邻字符
chars = list(token)
if len(chars) > 2:
idx = np.random.randint(1, len(chars)-1)
chars[idx], chars[idx+1] = chars[idx+1], chars[idx]
perturbed_token = ''.join(chars)
elif len(token) > 1:
# 删除一个字符
perturbed_token = token[:-1] if np.random.random() < 0.5 else token[1:]
else:
# 替换为相似字符
similar_chars = {'a':'e', 'e':'a', 'i':'o', 'o':'i', 'b':'d', 'd':'b'}
perturbed_token = similar_chars.get(token, token)
# 编码回token id
new_token_id = tokenizer.encode(perturbed_token, add_special_tokens=False)
if new_token_id:
input_ids[i, j] = new_token_id[0]
perturbed_batch["input_ids"] = torch.tensor(input_ids, device=perturbed_batch["input_ids"].device)
return perturbed_batch
6.3 提升噪声鲁棒性的进阶策略
除了基本的对抗训练方法外，还有多种进阶策略可以进一步提升模型的噪声鲁棒性：
- 集成对抗训练
集成多个不同扰动方法生成的对抗样本，使模型面对各种类型的扰动都能保持稳定。
class EnsembleAdversarialTrainer(Trainer):
"""集成对抗训练，使用多种扰动方法"""
def __init__(self, *args, perturbation_methods=None, **kwargs):
super().__init__(*args, **kwargs)
self.perturbation_methods = perturbation_methods or [
lambda model, inputs, labels: PGDPerturbation(model, 0.1, 0.02, 5).generate(inputs, labels),
lambda model, inputs, labels: generate_fgsm_adversarial(model, inputs, labels, 0.1),
lambda model, inputs, labels: generate_mim_adversarial(model, inputs, labels, 0.1, 0.02, 5, 0.9)
]
def compute_loss(self, model, inputs, return_outputs=False):
if "labels" not in inputs:
raise ValueError("Adversarial training requires labels")
labels = inputs["labels"]
# 计算原始损失
original_outputs = model(**inputs)
original_loss = original_outputs.loss
# 计算每种扰动方法的对抗损失
adv_losses = []
for perturb_fn in self.perturbation_methods:
perturbed_inputs = perturb_fn(model, inputs.copy(), labels)
adv_outputs = model(**perturbed_inputs)
adv_losses.append(adv_outputs.loss)
# 集成损失
avg_adv_loss = sum(adv_losses) / len(adv_losses)
final_loss = 0.5 * original_loss + 0.5 * avg_adv_loss
return (final_loss, original_outputs) if return_outputs else final_loss
- 分层对抗训练
针对模型的不同层应用不同强度的扰动，更好地模拟真实世界中的噪声分布。
分层对抗训练流程：
输入文本 → 嵌入层 → 注意力层 → 前馈层 → 输出
↓ ↓ ↓ ↓
嵌入扰动 注意力扰动 前馈层扰动 综合扰动
↓ ↓ ↓ ↓
合并所有扰动 → 鲁棒模型更新
- 自监督对抗训练
结合自监督学习和对抗训练，使模型在没有标签的情况下也能提升鲁棒性。
class SelfSupervisedAdversarialTrainer(Trainer):
"""自监督对抗训练"""
def __init__(self, *args, mask_ratio=0.15, **kwargs):
super().__init__(*args, **kwargs)
self.mask_ratio = mask_ratio
self.mlm_probability = mask_ratio
def compute_loss(self, model, inputs, return_outputs=False):
# 对于自监督任务，我们使用掩码语言模型损失
# 首先生成对抗样本
perturbed_inputs = PGDPerturbation(model, 0.1, 0.02, 5).generate(inputs, None)
# 确保两个输入都有掩码标记
if "labels" not in inputs or inputs["labels"] is None:
inputs = self._prepare_mlm_inputs(inputs)
if "labels" not in perturbed_inputs or perturbed_inputs["labels"] is None:
perturbed_inputs = self._prepare_mlm_inputs(perturbed_inputs)
# 计算原始样本和对抗样本的损失
original_outputs = model(**inputs)
original_loss = original_outputs.loss
perturbed_outputs = model(**perturbed_inputs)
perturbed_loss = perturbed_outputs.loss
# 混合损失
final_loss = (original_loss + perturbed_loss) / 2
return (final_loss, original_outputs) if return_outputs else final_loss
def _prepare_mlm_inputs(self, inputs):
"""准备掩码语言模型的输入"""
# 复制输入
mlm_inputs = {k: v.clone() for k, v in inputs.items()}
# 生成掩码位置
probability_matrix = torch.full(mlm_inputs["input_ids"].shape, self.mlm_probability)
special_tokens_mask = [
self.tokenizer.get_special_tokens_mask(val, already_has_special_tokens=True)
for val in mlm_inputs["input_ids"].tolist()
]
probability_matrix.masked_fill_(torch.tensor(special_tokens_mask, dtype=torch.bool), value=0.0)
masked_indices = torch.bernoulli(probability_matrix).bool()
# 创建标签
mlm_inputs["labels"] = mlm_inputs["input_ids"].clone()
mlm_inputs["labels"][~masked_indices] = -100 # 只计算掩码位置的损失
# 替换掩码位置的token
indices_replaced = torch.bernoulli(torch.full(mlm_inputs["input_ids"].shape, 0.8)).bool() & masked_indices
mlm_inputs["input_ids"][indices_replaced] = self.tokenizer.convert_tokens_to_ids(self.tokenizer.mask_token)
# 随机替换部分token
indices_random = torch.bernoulli(torch.full(mlm_inputs["input_ids"].shape, 0.5)).bool() & masked_indices & ~indices_replaced
random_words = torch.randint(len(self.tokenizer), mlm_inputs["input_ids"].shape, dtype=torch.long, device=mlm_inputs["input_ids"].device)
mlm_inputs["input_ids"][indices_random] = random_words[indices_random]
return mlm_inputs
- 对抗对比学习
将对比学习的思想融入对抗训练，通过最大化原始样本和对抗样本的表示相似性，提升模型的鲁棒性。
对抗对比学习的目标函数：
L A C L = L C E + λ ⋅ L C o n t r a s t i v e \mathcal{L}_{ACL} = \mathcal{L}_{CE} + \lambda \cdot \mathcal{L}_{Contrastive} LACL=LCE+λ⋅LContrastive
其中 ( \mathcal{L}{CE} ) 是分类损失，( \mathcal{L}{Contrastive} ) 是对比损失，定义为：
L C o n t r a s t i v e = − log e sim ( f ( x ) , f ( x + δ ) ) / τ ∑ x ′ ≠ x e sim ( f ( x ) , f ( x ′ ) ) / τ \mathcal{L}_{Contrastive} = -\log\frac{e^{\text{sim}(f(x), f(x+\delta))/\tau}}{\sum_{x'\neq x} e^{\text{sim}(f(x), f(x'))/\tau}} LContrastive=−log∑x′=xesim(f(x),f(x′))/τesim(f(x),f(x+δ))/τ
通过这些进阶策略，我们可以显著提升模型的噪声鲁棒性，使其在面对各种类型的扰动时都能保持稳定的性能。
7. 大规模语言模型的噪声鲁棒微调实践
7.1 实际应用场景与挑战
在大规模语言模型的实际应用中，噪声鲁棒微调的重要性日益凸显。以下是几个典型的应用场景及其面临的挑战：
| 应用场景 | 噪声类型 | 主要挑战 | 鲁棒性需求 |
|---|---|---|---|
| 客服对话系统 | 拼写错误、方言表达、不完整句子 | 用户输入多样、语义理解困难 | 高 |
| 内容审核 | 刻意规避检测的文本、隐晦表达 | 对抗性攻击、误判率高 | 极高 |
| 智能问答 | 歧义问题、信息缺失、跨语言问题 | 推理能力要求高、噪声导致错误回答 | 高 |
| 文档摘要 | 结构混乱、冗余信息、低质量输入 | 信息提取准确性、摘要质量 | 中高 |
| 代码生成 | 自然语言描述不精确、需求表述模糊 | 生成错误代码风险、理解偏差 | 极高 |
大规模语言模型噪声鲁棒微调的特殊挑战：
- 计算资源消耗：大模型参数规模庞大，对抗训练的额外计算成本显著
- 超参数敏感性：扰动强度、迭代次数等参数对大模型影响更大
- 训练不稳定性：更容易出现梯度爆炸、模式崩溃等问题
- 存储开销：需要存储额外的对抗样本和中间状态
- 评估难度：鲁棒性评估需要更多样化的测试集和评估指标
7.2 大规模语言模型的对抗训练实现
针对大规模语言模型，我们需要优化对抗训练的实现，以平衡计算效率和鲁棒性提升。以下是一个针对Hugging Face Transformers库中LLaMA、GPT等大模型的对抗训练实现示例：
from transformers import (AutoModelForCausalLM, AutoTokenizer,
TrainingArguments, Trainer, DataCollatorForSeq2Seq)
import torch
import numpy as np
from torch.utils.data import Dataset
class LLMAdversarialTrainer(Trainer):
"""专为大规模语言模型设计的对抗训练器"""
def __init__(self, *args, adv_method="pgd", adv_config=None, **kwargs):
super().__init__(*args, **kwargs)
self.adv_method = adv_method
self.adv_config = adv_config or {
"epsilon": 0.01, # 注意：大模型通常需要更小的扰动
"alpha": 0.005,
"steps": 3,
"mu": 0.9, # MIM参数
"embedding_only": True, # 只扰动嵌入层
"grad_accumulation": 4, # 梯度累积
"layer_indices": None # 指定要扰动的层
}
# 确保梯度检查点启用以节省内存
if hasattr(self.model, 'gradient_checkpointing_enable'):
self.model.gradient_checkpointing_enable()
# 缓存嵌入层以加速访问
self.embedding_layer = self._get_embedding_layer()
def _get_embedding_layer(self):
"""获取模型的嵌入层"""
for module_name, module in self.model.named_modules():
if any(name in module_name for name in ["embedding", "wte", "embed_tokens"]):
return module
return None
def _freeze_other_parameters(self, freeze=True):
"""冻结或解冻除嵌入层外的参数"""
for name, param in self.model.named_parameters():
if "embed" not in name and "bias" not in name:
param.requires_grad = not freeze
def _generate_fgsm_perturbation(self, embeddings, labels, attention_mask):
"""生成FGSM对抗扰动"""
# 复制嵌入以避免修改原始嵌入
perturbed_embeddings = embeddings.clone().requires_grad_(True)
# 前向传播计算损失
outputs = self.model(inputs_embeds=perturbed_embeddings,
labels=labels,
attention_mask=attention_mask)
loss = outputs.loss
# 反向传播计算梯度
loss.backward()
# 计算扰动
perturbation = self.adv_config["epsilon"] * torch.sign(perturbed_embeddings.grad)
# 应用扰动
perturbed_embeddings = perturbed_embeddings + perturbation
# 确保扰动在有效范围内
perturbed_embeddings = torch.clamp(perturbed_embeddings,
min=-self.adv_config["epsilon"],
max=self.adv_config["epsilon"])
return perturbed_embeddings.detach()
def compute_loss(self, model, inputs, return_outputs=False):
"""计算对抗训练的损失"""
# 保存原始输入
original_inputs = inputs.copy()
# 获取标签
labels = inputs.get("labels", None)
if labels is None:
labels = inputs.get("input_ids", None)
# 获取注意力掩码
attention_mask = inputs.get("attention_mask", None)
# 如果只扰动嵌入层
if self.adv_config.get("embedding_only", True):
# 获取输入ID
input_ids = inputs["input_ids"]
# 获取嵌入表示
with torch.no_grad():
embeddings = self.embedding_layer(input_ids)
# 根据选择的对抗方法生成扰动
if self.adv_method == "fgsm":
perturbed_embeddings = self._generate_fgsm_perturbation(
embeddings, labels, attention_mask)
elif self.adv_method == "pgd":
perturbed_embeddings = self._generate_pgd_perturbation(
embeddings, labels, attention_mask)
elif self.adv_method == "mim":
perturbed_embeddings = self._generate_mim_perturbation(
embeddings, labels, attention_mask)
else:
perturbed_embeddings = embeddings
# 用扰动后的嵌入替换输入
inputs.pop("input_ids")
inputs["inputs_embeds"] = perturbed_embeddings
# 计算对抗样本的损失
adv_outputs = model(**inputs)
adv_loss = adv_outputs.loss
# 计算原始样本的损失（可选）
lambda_reg = self.adv_config.get("lambda_reg", 0.5)
if lambda_reg > 0:
with torch.no_grad():
original_outputs = model(**original_inputs)
original_loss = original_outputs.loss
# 混合损失
loss = lambda_reg * original_loss + (1 - lambda_reg) * adv_loss
else:
loss = adv_loss
if return_outputs:
return loss, adv_outputs
return loss
def _generate_pgd_perturbation(self, embeddings, labels, attention_mask):
"""生成PGD对抗扰动"""
# 复制嵌入
perturbed_embeddings = embeddings.clone().requires_grad_(True)
# 迭代步骤
steps = self.adv_config.get("steps", 3)
alpha = self.adv_config.get("alpha", 0.005)
epsilon = self.adv_config.get("epsilon", 0.01)
# PGD迭代
for i in range(steps):
# 前向传播
outputs = self.model(inputs_embeds=perturbed_embeddings,
labels=labels,
attention_mask=attention_mask)
loss = outputs.loss
# 反向传播
self.model.zero_grad()
loss.backward()
# 梯度更新
grad = perturbed_embeddings.grad.data
perturbation = alpha * torch.sign(grad)
perturbed_embeddings = perturbed_embeddings + perturbation
# 投影到epsilon球
diff = perturbed_embeddings - embeddings
diff = torch.clamp(diff, -epsilon, epsilon)
perturbed_embeddings = embeddings + diff
# 重置梯度
perturbed_embeddings = perturbed_embeddings.clone().requires_grad_(True)
return perturbed_embeddings.detach()
def _generate_mim_perturbation(self, embeddings, labels, attention_mask):
"""生成MIM对抗扰动"""
# 复制嵌入
perturbed_embeddings = embeddings.clone().requires_grad_(True)
# MIM参数
steps = self.adv_config.get("steps", 3)
alpha = self.adv_config.get("alpha", 0.005)
epsilon = self.adv_config.get("epsilon", 0.01)
decay = self.adv_config.get("decay", 1.0)
# 梯度累积器
momentum = torch.zeros_like(perturbed_embeddings)
# MIM迭代
for i in range(steps):
# 前向传播
outputs = self.model(inputs_embeds=perturbed_embeddings,
labels=labels,
attention_mask=attention_mask)
loss = outputs.loss
# 反向传播
self.model.zero_grad()
loss.backward()
# 获取梯度
grad = perturbed_embeddings.grad.data
# 动量更新
momentum = decay * momentum + grad / torch.norm(grad, p=1)
# 梯度更新
perturbation = alpha * torch.sign(momentum)
perturbed_embeddings = perturbed_embeddings + perturbation
# 投影到epsilon球
diff = perturbed_embeddings - embeddings
diff = torch.clamp(diff, -epsilon, epsilon)
perturbed_embeddings = embeddings + diff
# 重置梯度
perturbed_embeddings = perturbed_embeddings.clone().requires_grad_(True)
return perturbed_embeddings.detach()
else:
# 标准FGSM实现
inputs = {k: v.clone() for k, v in inputs.items()}
for k in inputs:
if isinstance(inputs[k], torch.Tensor):
inputs[k].requires_grad = True
outputs = model(**inputs)
loss = outputs.loss
model.zero_grad()
loss.backward()
perturbed_inputs = inputs.copy()
if "input_ids" in perturbed_inputs and perturbed_inputs["input_ids"].grad is not None:
gradient = torch.sign(perturbed_inputs["input_ids"].grad)
perturbed_inputs["input_ids"] = perturbed_inputs["input_ids"] + \
self.adv_config["epsilon"] * gradient
return perturbed_inputs
def _generate_pgd(self, model, inputs):
"""为大模型生成PGD对抗样本"""
if self.adv_config["embedding_only"]:
# 在嵌入空间进行PGD
inputs = {k: v.to(model.device) for k, v in inputs.items()}
# 获取原始嵌入
with torch.no_grad():
original_embeddings = self.embedding_layer(inputs["input_ids"])
# 初始化扰动嵌入
perturbed_embeddings = original_embeddings.clone()
perturbed_embeddings = perturbed_embeddings + torch.randn_like(perturbed_embeddings) * 0.001
# 启用嵌入层梯度
self.embedding_layer.weight.requires_grad = True
# PGD迭代
for _ in range(self.adv_config["steps"]):
# 复制扰动嵌入以进行梯度计算
curr_embeddings = perturbed_embeddings.clone().requires_grad_(True)
# 前向传播
outputs = model(inputs_embeds=curr_embeddings, **{k: v for k, v in inputs.items() if k != "input_ids"})
loss = outputs.loss
# 反向传播
model.zero_grad()
loss.backward()
# 更新扰动
with torch.no_grad():
# 获取梯度
grad = curr_embeddings.grad
# 更新扰动嵌入
perturbed_embeddings = perturbed_embeddings + self.adv_config["alpha"] * torch.sign(grad)
# 投影回epsilon球
perturbation = perturbed_embeddings - original_embeddings
perturbation = torch.clamp(perturbation, -self.adv_config["epsilon"], self.adv_config["epsilon"])
perturbed_embeddings = original_embeddings + perturbation
# 创建扰动后的输入
perturbed_inputs = inputs.copy()
perturbed_inputs["inputs_embeds"] = perturbed_embeddings
if "input_ids" in perturbed_inputs:
del perturbed_inputs["input_ids"]
# 禁用嵌入层梯度
self.embedding_layer.weight.requires_grad = False
return perturbed_inputs
else:
# 标准PGD实现，此处省略
return inputs
def _generate_mim(self, model, inputs):
"""为大模型生成MIM对抗样本"""
if self.adv_config["embedding_only"]:
# 在嵌入空间进行MIM
inputs = {k: v.to(model.device) for k, v in inputs.items()}
# 获取原始嵌入
with torch.no_grad():
original_embeddings = self.embedding_layer(inputs["input_ids"])
# 初始化扰动和动量
perturbed_embeddings = original_embeddings.clone()
momentum = torch.zeros_like(perturbed_embeddings)
# 启用嵌入层梯度
self.embedding_layer.weight.requires_grad = True
# MIM迭代
for _ in range(self.adv_config["steps"]):
# 复制扰动嵌入以进行梯度计算
curr_embeddings = perturbed_embeddings.clone().requires_grad_(True)
# 前向传播
outputs = model(inputs_embeds=curr_embeddings, **{k: v for k, v in inputs.items() if k != "input_ids"})
loss = outputs.loss
# 反向传播
model.zero_grad()
loss.backward()
# 更新动量和扰动
with torch.no_grad():
# 获取梯度并归一化
grad = curr_embeddings.grad
grad_norm = torch.norm(grad.view(grad.shape[0], -1), dim=-1).view(-1, 1, 1)
grad = grad / (grad_norm + 1e-12)
# 更新动量
momentum = self.adv_config["mu"] * momentum + grad
# 更新扰动嵌入
perturbed_embeddings = perturbed_embeddings + self.adv_config["alpha"] * torch.sign(momentum)
# 投影回epsilon球
perturbation = perturbed_embeddings - original_embeddings
perturbation = torch.clamp(perturbation, -self.adv_config["epsilon"], self.adv_config["epsilon"])
perturbed_embeddings = original_embeddings + perturbation
# 创建扰动后的输入
perturbed_inputs = inputs.copy()
perturbed_inputs["inputs_embeds"] = perturbed_embeddings
if "input_ids" in perturbed_inputs:
del perturbed_inputs["input_ids"]
# 禁用嵌入层梯度
self.embedding_layer.weight.requires_grad = False
return perturbed_inputs
else:
# 标准MIM实现，此处省略
return inputs
# 使用示例
def train_robust_llm(model_name="meta-llama/Llama-2-7b-hf", dataset=None):
"""训练鲁棒的大规模语言模型"""
# 加载模型和分词器
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16, # 使用混合精度以节省内存
device_map="auto", # 自动分配设备
load_in_8bit=True # 使用8位量化以节省内存
)
# 设置训练参数
training_args = TrainingArguments(
output_dir="./robust_llm_output",
per_device_train_batch_size=1, # 大模型通常使用小批量
gradient_accumulation_steps=8, # 使用梯度累积以模拟更大批量
learning_rate=2e-6, # 大模型需要更小的学习率
weight_decay=0.01,
evaluation_strategy="steps",
eval_steps=1000,
save_steps=1000,
num_train_epochs=3,
fp16=True, # 启用混合精度训练
logging_steps=100,
remove_unused_columns=False,
label_names=["labels"],
# 优化器参数
optim="adamw_torch",
warmup_ratio=0.1,
# 稳定性参数
gradient_checkpointing=True,
max_grad_norm=1.0
)
# 创建对抗训练器
trainer = LLMAdversarialTrainer(
model=model,
args=training_args,
train_dataset=dataset["train"] if dataset else None,
eval_dataset=dataset["validation"] if dataset else None,
tokenizer=tokenizer,
data_collator=DataCollatorForSeq2Seq(tokenizer, model=model),
adv_method="pgd", # 使用PGD方法
adv_config={
"epsilon": 0.01, # 大模型使用小扰动
"alpha": 0.003,
"steps": 3, # 减少迭代次数以节省计算
"embedding_only": True # 只扰动嵌入层
}
)
# 开始训练
trainer.train()
# 保存模型
trainer.save_model("./robust_llm_final")
return trainer
7.3 大规模语言模型对抗训练的优化技巧
针对大规模语言模型的对抗训练，我们需要特别注意以下优化技巧：
- 内存优化策略
def optimize_memory_usage(model, training_args):
"""优化大模型对抗训练的内存使用"""
# 1. 启用梯度检查点
if hasattr(model, 'gradient_checkpointing_enable'):
model.gradient_checkpointing_enable()
# 2. 启用混合精度训练
training_args.fp16 = True
# 3. 使用模型量化
from transformers import BitsAndBytesConfig
bnb_config = BitsAndBytesConfig(
load_in_8bit=True,
llm_int8_threshold=6.0
)
# 4. 梯度累积
training_args.gradient_accumulation_steps = 8
# 5. 关闭不必要的输出
training_args.remove_unused_columns = True
return model, training_args, bnb_config
梯度检查点技术通过牺牲少量计算时间来节省内存，它在前向传播时不保存所有中间激活值，而是在反向传播时重新计算必要的中间结果。对于大模型来说，这可以显著减少内存占用。
混合精度训练使用FP16进行前向和反向传播计算，同时用FP32存储参数和梯度，这种方法可以将内存需求减半，同时加快计算速度。
量化技术（如8位量化）通过减少参数的精度来节省内存，特别适合在资源有限的环境中训练大模型。
- 计算效率优化
对于大规模语言模型，计算效率是对抗训练成功的关键。以下是几种有效的优化方法：
def optimize_computation_efficiency(model, trainer, adv_config):
"""优化大模型对抗训练的计算效率"""
# 1. 减少对抗迭代次数
adv_config["steps"] = min(adv_config["steps"], 3) # 大模型通常使用较少的迭代次数
# 2. 使用嵌入空间扰动
adv_config["embedding_only"] = True # 只在嵌入空间进行扰动，避免修改输入ID
# 3. 减少批量大小，增加梯度累积
trainer.args.per_device_train_batch_size = 1
trainer.args.gradient_accumulation_steps = 16
# 4. 启用模型并行
if "device_map" not in model.config.to_dict():
model = model.to("cuda")
return model, trainer, adv_config
嵌入空间扰动是一种高效的对抗训练方法，它避免了在离散的词汇空间中搜索对抗样本的复杂性，直接在连续的嵌入空间中添加扰动，大大降低了计算成本。
减少对抗迭代次数和批量大小可以降低每次训练迭代的计算负担，而梯度累积则可以保持等效的批量大小，确保训练稳定性。
- 计算效率优化
大模型对抗训练计算优化流程：
输入数据 → 梯度累积 → 混合精度 → 量化计算 → 并行处理 → 优化输出
↓ ↓ ↓ ↓ ↓
批量处理 内存节省 速度提升 内存节省 分布式训练
- 训练稳定性提升
def enhance_training_stability(trainer):
"""提升大模型对抗训练的稳定性"""
# 1. 渐进式增加扰动强度
class ProgressivePerturbationCallback:
def __init__(self, trainer, max_epsilon=0.01, warmup_epochs=1):
self.trainer = trainer
self.max_epsilon = max_epsilon
self.warmup_epochs = warmup_epochs
def on_step_end(self, args, state, control, **kwargs):
if hasattr(self.trainer, 'adv_config'):
# 计算当前epoch比例
current_epoch = state.epoch
if current_epoch < self.warmup_epochs:
# 线性增加epsilon
self.trainer.adv_config["epsilon"] = self.max_epsilon * (current_epoch / self.warmup_epochs)
# 2. 梯度裁剪
trainer.args.max_grad_norm = 1.0
# 3. 学习率预热和余弦衰减
trainer.args.warmup_ratio = 0.1
trainer.args.lr_scheduler_type = "cosine"
# 4. 添加渐进式扰动回调
trainer.add_callback(ProgressivePerturbationCallback(trainer))
return trainer
- 分布式对抗训练
def setup_distributed_adversarial_training(trainer, local_rank):
"""设置分布式对抗训练"""
# 1. 确保使用分布式DataParallel
import torch.distributed as dist
# 2. 调整批量大小
trainer.args.per_device_train_batch_size = max(1, trainer.args.per_device_train_batch_size)
# 3. 同步扰动生成（可选）
def sync_perturbation(perturbation):
# 收集所有进程的扰动
gathered_perturbations = [torch.zeros_like(perturbation) for _ in range(dist.get_world_size())]
dist.all_gather(gathered_perturbations, perturbation)
# 平均扰动
return sum(gathered_perturbations) / dist.get_world_size()
# 4. 只在主进程保存日志
if local_rank != 0:
trainer.args.save_steps = float("inf")
# 5. 同步梯度（重要）
trainer.args.gradient_as_bucket_view = True
# 6. 内存优化的分布式策略
if hasattr(trainer, 'model') and hasattr(trainer.model, 'gradient_checkpointing_enable'):
trainer.model.gradient_checkpointing_enable()
return trainer
分布式对抗训练在多GPU或多节点环境中特别重要，它可以显著加速训练过程并扩大模型规模。关键考虑点包括：
- 扰动同步：在分布式环境中，每个进程可能生成不同的对抗扰动。同步扰动可以确保所有进程使用一致的对抗样本，提高训练稳定性。
- 通信优化：减少节点间的通信频率，只在必要时进行梯度同步。
- 内存平衡：确保各节点的内存使用均衡，避免某些节点内存不足而其他节点资源浪费。
- 梯度累积：在分布式环境中继续使用梯度累积可以进一步降低每个节点的内存需求。
通过这些优化技巧，我们可以有效地在大规模语言模型上应用对抗训练，在有限的计算资源下提升模型的噪声鲁棒性。
7.4 大规模语言模型对抗训练的评估方法
评估大规模语言模型的对抗训练效果需要考虑多个维度：
def evaluate_adversarial_robustness(model, tokenizer, test_dataset, adv_config=None):
"""评估模型的对抗鲁棒性"""
results = {
"clean_performance": {},
"adversarial_performance": {},
"robustness_metrics": {}
}
# 1. 干净样本性能
clean_evaluator = LLMAdversarialTrainer(
model=model,
tokenizer=tokenizer,
adv_method=None, # 不使用对抗训练
data_collator=DataCollatorForSeq2Seq(tokenizer, model=model)
)
clean_metrics = clean_evaluator.evaluate(eval_dataset=test_dataset)
results["clean_performance"] = clean_metrics
# 2. 对抗样本性能
if adv_config:
adv_evaluator = LLMAdversarialTrainer(
model=model,
tokenizer=tokenizer,
adv_method="pgd", # 使用PGD生成对抗样本
adv_config=adv_config,
data_collator=DataCollatorForSeq2Seq(tokenizer, model=model)
)
# 生成对抗测试集
adv_test_dataset = []
for example in tqdm(test_dataset):
# 生成对抗输入
inputs = tokenizer(example["text"], return_tensors="pt")
inputs = {k: v.to(model.device) for k, v in inputs.items()}
# 生成对抗样本
adv_inputs = adv_evaluator._generate_pgd(model, inputs)
# 转换回文本
if "inputs_embeds" in adv_inputs:
# 从嵌入空间恢复文本（近似方法）
with torch.no_grad():
# 使用最近邻搜索找到最接近的token
similarities = torch.matmul(
adv_inputs["inputs_embeds"][0],
model.get_input_embeddings().weight.T
)
adv_tokens = similarities.argmax(dim=-1)
adv_text = tokenizer.decode(adv_tokens)
else:
adv_text = tokenizer.decode(adv_inputs["input_ids"][0])
# 添加到对抗测试集
adv_example = example.copy()
adv_example["text"] = adv_text
adv_test_dataset.append(adv_example)
# 评估对抗样本性能
adv_metrics = clean_evaluator.evaluate(eval_dataset=adv_test_dataset)
results["adversarial_performance"] = adv_metrics
# 计算鲁棒性指标
if "loss" in clean_metrics and "loss" in adv_metrics:
results["robustness_metrics"]["loss_robustness"] = (
clean_metrics["loss"] / adv_metrics["loss"]
if adv_metrics["loss"] > 0 else 0
)
# 计算鲁棒性差距
results["robustness_metrics"][f"{key}_gap"] = (
clean_metrics[key] - adv_metrics[key]
)
results["robustness_metrics"][f"{key}_gap_percentage"] = (
(clean_metrics[key] - adv_metrics[key]) / clean_metrics[key] * 100
if clean_metrics[key] > 0 else 0
)
return results
在评估对抗训练效果时，我们需要关注以下关键指标：
-
鲁棒性差距（Robustness Gap）：干净样本性能与对抗样本性能之间的差异。差距越小，模型的鲁棒性越强。
计算公式：
Robustness Gap = 干净样本指标 - 对抗样本指标百分比形式：
Robustness Gap % = (干净样本指标 - 对抗样本指标) / 干净样本指标 * 100% -
噪声稳定性（Noise Stability）：模型在面对不同类型噪声时的表现一致性。可以通过在多种噪声类型上进行评估来衡量。
-
语义一致性（Semantic Consistency）：对抗样本虽然在输入上与原始样本有差异，但在语义上应该保持一致。可以通过人工评估或使用语义相似度指标来衡量。
下面是一个完整的评估流程示例：
def comprehensive_adversarial_evaluation(model, tokenizer, test_dataset, config):
"""全面的对抗鲁棒性评估"""
results = {}
# 1. 基础对抗评估
base_adv_config = {
"epsilon": config["base_epsilon"],
"alpha": config["base_alpha"],
"steps": config["base_steps"]
}
results["base_evaluation"] = evaluate_adversarial_robustness(
model, tokenizer, test_dataset, base_adv_config
)
# 2. 不同噪声强度的评估
results["varying_epsilon"] = {}
for epsilon in config["epsilon_values"]:
adv_config = {
"epsilon": epsilon,
"alpha": epsilon / 3, # 保持alpha与epsilon的比例
"steps": config["base_steps"]
}
results["varying_epsilon"][f"epsilon_{epsilon}"] = evaluate_adversarial_robustness(
model, tokenizer, test_dataset, adv_config
)
# 3. 不同对抗方法的评估
results["different_methods"] = {}
for method in config["adv_methods"]:
# 根据方法设置不同的配置
if method == "fgsm":
adv_config = {"epsilon": config["base_epsilon"]}
else: # pgd, mim
adv_config = {
"epsilon": config["base_epsilon"],
"alpha": config["base_alpha"],
"steps": config["base_steps"]
}
# 这里需要修改evaluate_adversarial_robustness以支持不同方法
results["different_methods"][method] = evaluate_adversarial_robustness(
model, tokenizer, test_dataset, adv_config, method=method
)
# 4. 语义一致性评估
results["semantic_consistency"] = evaluate_semantic_consistency(
model, tokenizer, test_dataset, config["base_epsilon"]
)
return results
进行对抗训练评估时，还需要注意以下几点：
- 评估数据集的选择：使用多样化的测试集，包括不同领域、不同复杂度的样本。
- 多次评估取平均：对抗样本生成具有随机性，建议多次运行并取平均值以获得更准确的结果。
- 与基线模型比较：确保将对抗训练后的模型与原始基线模型进行比较，验证改进效果。
- 可视化结果：使用图表直观展示不同噪声强度下的性能变化，帮助理解模型的鲁棒性特性。
下面补充语义一致性评估函数的实现：
def evaluate_semantic_consistency(model, tokenizer, test_dataset, epsilon):
"""评估对抗样本与原始样本的语义一致性"""
import numpy as np
from sentence_transformers import SentenceTransformer
# 加载句子嵌入模型用于语义相似度计算
sim_model = SentenceTransformer('all-MiniLM-L6-v2')
# 创建对抗训练器用于生成对抗样本
adv_trainer = LLMAdversarialTrainer(
model=model,
tokenizer=tokenizer,
adv_method="pgd",
adv_config={"epsilon": epsilon, "alpha": epsilon/3, "steps": 3}
)
similarities = []
for example in test_dataset[:100]: # 取前100个样本进行评估
original_text = example["text"]
# 生成对抗样本
inputs = tokenizer(original_text, return_tensors="pt")
inputs = {k: v.to(model.device) for k, v in inputs.items()}
# 生成对抗样本
adv_inputs = adv_trainer._generate_pgd(model, inputs)
# 从嵌入空间恢复文本
with torch.no_grad():
if "inputs_embeds" in adv_inputs:
similarities_embed = torch.matmul(
adv_inputs["inputs_embeds"][0],
model.get_input_embeddings().weight.T
)
adv_tokens = similarities_embed.argmax(dim=-1)
adv_text = tokenizer.decode(adv_tokens)
else:
adv_text = tokenizer.decode(adv_inputs["input_ids"][0])
# 计算语义相似度
original_embedding = sim_model.encode(original_text, convert_to_tensor=True)
adv_embedding = sim_model.encode(adv_text, convert_to_tensor=True)
similarity = torch.nn.functional.cosine_similarity(
original_embedding.unsqueeze(0),
adv_embedding.unsqueeze(0)
).item()
similarities.append(similarity)
results = {
"mean_similarity": np.mean(similarities),
"median_similarity": np.median(similarities),
"similarity_std": np.std(similarities),
"min_similarity": np.min(similarities),
"max_similarity": np.max(similarities)
}
return results
- 评估对抗样本与原始样本在语义上的保留程度
- 可以使用BERTScore或BERTScore等指标
7.5 大规模语言模型对抗训练的实践建议
基于实际经验，以下是一些在大规模语言模型上应用对抗训练的实践建议：
扰动强度选择
扰动强度（ε值）是对抗训练中最重要的超参数，它直接影响训练效果和模型鲁棒性。以下是一个自动寻找最优ε值的函数：
def find_optimal_epsilon(model, tokenizer, val_dataset, epsilon_range=(0.001, 0.1), num_steps=10):
"""自动寻找最优的对抗扰动强度ε"""
import numpy as np
from tqdm import tqdm
# 生成候选ε值
epsilon_candidates = np.linspace(epsilon_range[0], epsilon_range[1], num_steps)
results = []
# 计算干净样本性能作为基准
clean_evaluator = LLMAdversarialTrainer(
model=model,
tokenizer=tokenizer,
adv_method=None,
data_collator=DataCollatorForSeq2Seq(tokenizer, model=model)
)
clean_metrics = clean_evaluator.evaluate(eval_dataset=val_dataset)
clean_loss = clean_metrics.get("eval_loss", 0)
print(f"基准干净样本损失: {clean_loss:.4f}")
# 对每个候选ε值进行评估
for epsilon in tqdm(epsilon_candidates):
# 使用该ε值创建临时对抗训练器
temp_trainer = LLMAdversarialTrainer(
model=model,
tokenizer=tokenizer,
adv_method="pgd",
adv_config={
"epsilon": epsilon,
"alpha": epsilon / 3, # 保持alpha与epsilon的比例
"steps": 3
},
data_collator=DataCollatorForSeq2Seq(tokenizer, model=model)
)
# 计算对抗样本上的性能
adv_metrics = temp_trainer.evaluate(eval_dataset=val_dataset)
adv_loss = adv_metrics.get("eval_loss", 0)
# 计算性能下降百分比
loss_degradation = ((adv_loss - clean_loss) / clean_loss) * 100 if clean_loss > 0 else 0
# 计算鲁棒性分数（自定义）
robustness_score = 100 / (1 + loss_degradation) # 性能下降越小，分数越高
results.append({
"epsilon": epsilon,
"clean_loss": clean_loss,
"adv_loss": adv_loss,
"loss_degradation": loss_degradation,
"robustness_score": robustness_score
})
# 找到最优ε值：寻找最大的鲁棒性分数，但性能下降不超过阈值（例如10%）
threshold = 10.0 # 10%的性能下降阈值
valid_results = [r for r in results if r["loss_degradation"] <= threshold]
if valid_results:
optimal_result = max(valid_results, key=lambda x: x["robustness_score"])
else:
# 如果没有满足阈值的结果，选择性能下降最小的
optimal_result = min(results, key=lambda x: x["loss_degradation"])
print(f"最优ε值: {optimal_result['epsilon']:.6f}")
print(f"对应的性能下降: {optimal_result['loss_degradation']:.2f}%")
print(f"鲁棒性分数: {optimal_result['robustness_score']:.2f}")
return optimal_result, results
在选择扰动强度时，需要平衡以下几个因素：
-
模型大小：更大的模型通常可以承受更大的扰动强度
- 小型模型：ε通常在0.001-0.01之间
- 中型模型：ε通常在0.005-0.02之间
- 大型模型：ε可以在0.01-0.05之间
-
任务敏感性：不同任务对输入扰动的敏感度不同
- 生成任务：可以使用相对较大的ε值
- 分类任务：通常需要更小的ε值以保持标签一致性
-
训练阶段：可以采用渐进式的ε值
- 初始阶段：使用较小的ε值（如0.005）
- 中间阶段：逐渐增加到目标值
- 后期阶段：可以稍微降低ε值以稳定训练
混合正则化策略
将对抗训练与其他正则化方法结合可以获得更好的鲁棒性提升。以下是一个混合正则化训练器的实现：
扰动强度（ε值）对大模型尤为关键，建议：
def find_optimal_epsilon(model, tokenizer, calibration_dataset, max_epsilon=0.02):
"""寻找最优的扰动强度ε"""
# 性能下降阈值（通常为5-10%）
performance_threshold = 0.05
# 候选ε值
epsilon_candidates = np.linspace(0.001, max_epsilon, 10)
best_epsilon = 0.0
best_robustness = 0.0
for epsilon in epsilon_candidates:
adv_config = {
"epsilon": epsilon,
"alpha": epsilon / 3, # 通常α = ε/3
"steps": 3,
"embedding_only": True
}
# 评估当前ε值
results = evaluate_adversarial_robustness(
model, tokenizer, calibration_dataset, adv_config
)
# 计算性能下降
if "robustness_gap_loss" in results["robustness_metrics"]:
performance_drop = abs(results["robustness_metrics"]["robustness_gap_loss"]) / 100
# 确保性能下降在可接受范围内
if performance_drop <= performance_threshold:
# 计算鲁棒性分数（损失鲁棒性）
robustness_score = results["robustness_metrics"].get("loss_robustness", 0)
if robustness_score > best_robustness:
best_robustness = robustness_score
best_epsilon = epsilon
return best_epsilon
混合训练策略
结合多种正则化方法可以获得更好的鲁棒性：
def create_hybrid_regularization_trainer(model, args, train_dataset, **kwargs):
"""创建混合正则化的训练器"""
# 1. 首先设置对抗训练
adv_trainer = LLMAdversarialTrainer(
model=model,
args=args,
train_dataset=train_dataset,
adv_method="pgd",
adv_config={
"epsilon": 0.01,
"alpha": 0.003,
"steps": 3,
"embedding_only": True
},
**kwargs
)
# 2. 添加Dropout层（如果需要）
if hasattr(model, 'model') and hasattr(model.model, 'layers'):
for layer in model.model.layers:
if hasattr(layer, 'mlp') and hasattr(layer.mlp, 'dropout'):
layer.mlp.dropout.p = 0.1 # 设置适中的dropout率
if hasattr(layer, 'input_layernorm'):
# 添加R-Drop（可选）
pass
# 3. 添加梯度裁剪
args.max_grad_norm = 1.0
return adv_trainer
计算资源优化
大规模语言模型的对抗训练对计算资源要求很高，以下是一些优化建议：
-
渐进式训练：先在小批量数据上使用较小的ε值训练，然后逐步增加ε值和批量大小
-
冻结部分参数：只对模型的部分层（如最后几层）应用对抗训练
def freeze_layers_except_last(model, num_layers_to_train=2):
"""冻结除最后几层外的所有层"""
# 对于LLaMA等模型
if hasattr(model, 'model') and hasattr(model.model, 'layers'):
total_layers = len(model.model.layers)
# 冻结前面的层
for i in range(total_layers - num_layers_to_train):
for param in model.model.layers[i].parameters():
param.requires_grad = False
# 保持最后几层可训练
for i in range(total_layers - num_layers_to_train, total_layers):
for param in model.model.layers[i].parameters():
param.requires_grad = True
# 特殊处理嵌入层（通常需要可训练）
if hasattr(model, 'get_input_embeddings'):
for param in model.get_input_embeddings().parameters():
param.requires_grad = True
# 特殊处理输出层
if hasattr(model, 'get_output_embeddings'):
output_embeds = model.get_output_embeddings()
if output_embeds:
for param in output_embeds.parameters():
param.requires_grad = True
return model
# 3. 使用LoRA降低内存需求
def setup_lora_with_adversarial(model, r=8, lora_alpha=16):
"""配置LoRA与对抗训练结合"""
try:
from peft import get_peft_config, get_peft_model, LoraConfig
# 配置LoRA
lora_config = LoraConfig(
r=r,
lora_alpha=lora_alpha,
lora_dropout=0.05,
bias="none",
task_type="CAUSAL_LM",
target_modules=["q_proj", "k_proj", "v_proj", "o_proj"] # 注意力层
)
# 创建Peft模型
peft_model = get_peft_model(model, lora_config)
peft_model.print_trainable_parameters()
return peft_model
except ImportError:
print("未安装peft库，请先安装: pip install peft")
return model
资源优化策略的综合应用
在实际应用中，我们可以根据可用资源灵活组合这些优化策略：
-
轻量级配置（适合消费级GPU）：
- 使用LoRA（r=4或r=8）
- 仅在嵌入空间应用对抗训练（embedding_only=True）
- 批量大小=1，梯度累积步数=16-32
- 使用较小的ε值（0.005-0.01）
-
中量级配置（适合单张专业GPU）：
- 冻结前80%的层，只训练最后几层
- 使用嵌入空间的PGD对抗训练（3-5步）
- 批量大小=2-4，梯度累积步数=8
- 使用混合精度训练
-
重量级配置（适合多GPU）：
- 完整模型训练或更大的LoRA rank（r=16-32）
- 在完整输入空间应用对抗训练
- 更大的批量大小和ε值
- 结合多种正则化方法
7.6 总结与未来展望
通过本章的学习，我们深入探讨了大规模语言模型的对抗训练技术，主要包括：
-
对抗训练基础：了解了FGSM、PGD、MIM等对抗训练方法的原理和实现
-
大模型适配：学习了如何将这些方法适配到大规模语言模型上，包括嵌入空间对抗训练等优化技术
-
资源优化：掌握了多种降低内存和计算需求的方法，使得对抗训练能够在有限资源下应用于大模型
-
评估方法：学习了如何全面评估模型的对抗鲁棒性，包括鲁棒性差距、噪声稳定性和语义一致性等指标
未来研究方向
大规模语言模型的对抗训练仍有许多值得探索的方向：
-
自适应对抗训练：根据模型在训练过程中的表现动态调整对抗强度和策略
-
高效分布式对抗训练：进一步优化多GPU、多节点环境下的对抗训练效率
-
鲁棒性与泛化性平衡：探索如何在提高对抗鲁棒性的同时不牺牲模型的泛化能力
-
特定领域的对抗鲁棒性：针对医疗、法律等高风险领域开发专门的对抗训练方法
-
多模态对抗训练：将对抗训练扩展到文本-图像等多模态大模型
实践建议总结
在实施对抗训练时，我们建议：
-
从小规模开始：先用小模型验证对抗训练的有效性，再扩展到更大的模型
-
系统评估：使用多种评估指标全面衡量对抗训练的效果
-
资源规划：根据可用资源选择合适的优化策略和配置
-
持续监控：密切关注训练过程中的性能变化，及时调整参数
-
平衡计算与效果：在资源有限的情况下，优先使用嵌入空间对抗训练等高效方法
通过合理应用对抗训练技术，我们可以显著提高大规模语言模型的噪声鲁棒性，使其在实际应用中更加可靠，特别是在面对潜在的对抗性输入时能够保持稳定的性能。
7.6 总结与未来展望
大规模语言模型的噪声鲁棒微调是一个重要但具有挑战性的研究方向。通过本章的讨论，我们可以总结出以下关键点：
主要成果
-
定制化对抗训练框架：针对大规模语言模型的特性，我们设计了LLMAdversarialTrainer类，专注于嵌入层扰动以平衡鲁棒性和计算效率。
-
多方法支持：实现了FGSM、PGD和MIM三种主流对抗训练方法，并针对大模型进行了优化。
-
资源优化策略：提出了一系列针对大规模模型的内存和计算优化技巧，使得对抗训练在有限资源下可行。
-
混合正则化方案：结合对抗训练、Dropout、梯度裁剪等多种正则化方法，进一步提升模型鲁棒性。
-
评估体系：建立了全面的对抗鲁棒性评估方法，包括鲁棒性差距、噪声稳定性和语义一致性等关键指标。
实践建议
在实际应用中，建议按照以下步骤进行大规模语言模型的对抗训练：
-
准备阶段：
- 确定具体应用场景和噪声类型
- 准备多样化的训练和评估数据集
- 配置适合的计算资源（考虑分布式训练）
-
参数选择：
- 使用
find_optimal_epsilon函数确定最佳扰动强度 - 优先使用嵌入层扰动以节省计算资源
- 对于超大规模模型，考虑LoRA等参数高效微调方法
- 使用
-
训练执行：
- 采用渐进式训练策略，逐步增加扰动强度
- 结合多种优化技巧（梯度累积、混合精度等）
- 监控训练稳定性，及时调整超参数
-
评估与验证：
- 使用多种噪声类型进行评估
- 衡量模型在干净和对抗样本上的性能
- 验证在实际应用场景中的表现提升
未来研究方向
大规模语言模型对抗训练的未来研究可以关注以下几个方向：
-
自适应对抗训练：根据模型不同层的特性，自适应调整扰动策略和强度。
-
多模态对抗训练：扩展对抗训练到文本-图像等多模态大模型。
-
轻量级对抗方法：开发更适合边缘设备部署的高效对抗训练变体。
-
理论分析：深入研究对抗训练对大模型泛化能力的理论影响。
-
鲁棒性与效率平衡：探索在有限计算资源下最大化鲁棒性提升的方法。
通过持续的研究和实践，我们有理由相信，对抗训练将成为大规模语言模型部署前的标准优化步骤，为各种实际应用场景提供更可靠、更稳定的AI服务。
更多推荐
14
0- 0
已为社区贡献114条内容
所有评论(0)