在大语言模型（LLM）的评估过程中，一个至关重要的问题是模型是否在预训练阶段意外地接触到了我们用于测试或基准评估的数据。如果模型记忆了预训练语料库中的特定例子，那么其在测试时的表现可能会高估其真正的泛化能力。本文将深入探讨一种轻量级的成员推断技术，该技术通过将“目标”LLM与一个“参考”模型进行比较，来判断目标模型是否在预训练期间见过给定的数据集，从而有效检测预训练数据泄露问题。
核心理念：似然差分（Likelihood Differential）
这种方法的核心在于一个简单的前提：如果目标模型在预训练时训练过某个文本，那么它将为该文本分配比从未见过它的模型更高的概率（即更低的困惑度）。 为了量化这种差异，我们使用一个标准化的对数似然指标，即似然差分。
具体来说，对于每个文本 x，我们计算每个模型下token对数概率的总和。然后，我们通过使用 zlib 压缩 x 并获取压缩长度 z(x) 来衡量每个文本的“信息内容”，这近似于其熵。 接着，我们为每个模型定义一个标准化分数：
- 标准化分数 = (token对数概率之和) / zlib压缩长度
最终的成员信号就是目标模型和参考模型的标准化分数之差，即 Δ = 目标模型标准化分数 – 参考模型标准化分数。 一个大的正 Δ 表明目标模型比未暴露的模型“更了解” x。
举个例子，假设我们有一个目标模型A和一个参考模型B。我们输入一段文本“天空是蓝色的”给两个模型。模型A给出的token对数概率之和是-10，模型B给出的是-20。这段文本的zlib压缩长度是20。那么模型A的标准化分数是-10/20=-0.5，模型B的标准化分数是-20/20=-1。则Δ是-0.5-(-1)=0.5。如果Δ足够大，就说明目标模型A可能见过这段文本。
实验设置与模型选择
为了验证该方法的有效性，我们参考文章作者使用了亚马逊评论 — 奢侈美妆数据集。 具体来说，我们使用：
- 目标模型： google/gemma-7b
- 参考模型： EleutherAI/pythia-6.9b-deduped (仅在 Pile 上训练，其中不包含任何亚马逊评论)
选择 EleutherAI 的 Pythia-6.9B-deduped 作为参考模型的关键在于它确保了干净的基线，因为它从未见过奢侈美妆评论。 这使得任何目标模型显示出的异常高的置信度（相对于参考模型）都暗示着可能的记忆。 这样保证了可以有效的检测预训练数据泄露。
计算和保存 Δ 分数
以下是一个Python脚本，它可以扫描您的gzipped JSON评论，计算每个评论的 Δ，并将结果写入带时间戳的 .txt 文件。
#!/usr/bin/env python3
import argparse, gzip, json, zlib, os
from datetime import datetime
import torch
import torch.nn.functional as F
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
from tqdm import tqdm
def compute_sum_logprob(model, tokenizer, text, device, max_length=2048):
inputs = tokenizer(text, return_tensors="pt",
truncation=True, max_length=max_length).to(device)
with torch.no_grad():
logits = model(**inputs).logits
# shift for next-token log-probabilities
shift_logits = logits[..., :-1, :].contiguous()
shift_labels = inputs["input_ids"][..., 1:].contiguous()
log_probs = F.log_softmax(shift_logits, dim=-1)
token_log_probs = log_probs.gather(-1, shift_labels.unsqueeze(-1)).squeeze(-1)
return token_log_probs.sum().item()
def main():
parser = argparse.ArgumentParser(
description="Membership inference via Δ = logp / zlib_len"
)
parser.add_argument("--data", type=str, default="Luxury_Beauty.json.gz")
parser.add_argument("--limit", type=int, default=0)
parser.add_argument("--output", type=str, default="membership_results.csv")
args = parser.parse_args()
device = torch.device("cuda:1" if torch.cuda.is_available() else "cpu")
# --- Load target and reference models ---
target_name = "google/gemma-7b"
ref_name = "EleutherAI/pythia-6.9b-deduped"
quant_cfg = BitsAndBytesConfig(load_in_4bit=True,
bnb_4bit_compute_dtype=torch.bfloat16)
print(f"Loading target model ({target_name})…")
tgt_tok = AutoTokenizer.from_pretrained(target_name)
tgt_mdl = AutoModelForCausalLM.from_pretrained(
target_name, quantization_config=quant_cfg, device_map={"": 1})
print(f"Loading reference model ({ref_name})…")
ref_tok = AutoTokenizer.from_pretrained(ref_name)
ref_mdl = AutoModelForCausalLM.from_pretrained(
ref_name, quantization_config=quant_cfg, device_map={"": 1})
# --- Read dataset ---
print(f"Reading records from {args.data}…")
with gzip.open(args.data, "rt", encoding="utf-8") as f:
records = [json.loads(line) for line in f]
if args.limit > 0:
records = records[:args.limit]
print(f"→ Will process {len(records)} reviews")
# --- Compute Δ for each review ---
results = []
for rec in tqdm(records, desc="Processing reviews"):
text = rec.get("reviewText", "").strip()
if not text:
continue
zlen = len(zlib.compress(text.encode("utf-8")))
sum_lp_tgt = compute_sum_logprob(tgt_mdl, tgt_tok, text, device)
sum_lp_ref = compute_sum_logprob(ref_mdl, ref_tok, text, device)
delta_tgt = sum_lp_tgt / zlen
delta_ref = sum_lp_ref / zlen
delta_diff = delta_tgt - delta_ref
results.append({
"reviewID": rec.get("reviewerID"),
"zlib_len": zlen,
"sum_logp_tgt": sum_lp_tgt,
"sum_logp_ref": sum_lp_ref,
"delta_tgt": delta_tgt,
"delta_ref": delta_ref,
"delta_diff": delta_diff,
})
# --- Save results with timestamp ---
output_dir = '/home/kavach_d/.../results' #需要改成自己的路径
os.makedirs(output_dir, exist_ok=True)
timestamp = datetime.now().strftime("%Y:%m:%d_%H:%M")
filename = f"{timestamp}.txt"
filepath = os.path.join(output_dir, filename)
with open(filepath, "w") as f:
json.dump(results, f, indent=2)
print(f"Wrote {len(results)} items to {filepath}")
if __name__ == "__main__":
main()
此脚本的关键点在于：
- 标准化： 将对数概率除以 zlib 压缩长度可确保文本长度和复杂性之间的公平性。
- 参考模型要求： 必须确保参考模型在预训练期间从未见过您的数据集。
- 4 位量化： 通过 BitsAndBytes 进行 4 位量化可保持内存和计算效率。
成员资格阈值设置
计算每个评论的 Δ 后，我们可以简单地将 Δ 高于所选阈值的评论标记为“成员”。以下是一个Python脚本，用于执行此操作：
import json
import pandas as pd
# 1. Load the JSON results
json_path = '.../results/gemma.txt' # 替换为您的路径
with open(json_path, 'r', encoding='utf-8') as f:
data = json.load(f)
# 2. Create a DataFrame
df = pd.DataFrame(data)
# 3. Set your threshold
threshold = 0.01
# 4. Flag membership
df['is_member'] = df['delta_diff'] >= threshold
# 5. Print a summary with percentages
total = len(df)
num_members = df['is_member'].sum()
pct_members = num_members / total * 100
pct_non = 100 - pct_members
print(f"Threshold set at: {threshold}")
print(f"Total samples: {total}")
print(f"Flagged as member: {num_members} ({pct_members:.2f}%)")
print(f"Flagged as non-member: {total - num_members} ({pct_non:.2f}%)")
阈值选择（此处为 0.01）可以根据您期望的假阳性/假阴性权衡进行调整。 建议尝试不同的阈值，例如 0.05 和 0.001，并观察结果的变化。 还可以绘制图形以找到阈值的最佳值。 最后的摘要会告诉您目标模型记忆的评论比例。 如果这个比例很高，就很有可能发生了预训练数据泄露。
例如，假设我们运行了上述代码，并设置阈值为0.01。结果显示，总共有1000条评论，其中有100条被标记为成员，占比10%。这可能意味着目标模型在预训练阶段见过这100条评论。
实际应用案例
假设一家公司正在开发一个新的医疗诊断LLM。他们使用了一个公开的医疗记录数据集进行预训练，然后使用他们自己的专有数据集进行微调。为了确保模型的安全性，他们需要验证模型是否在预训练阶段意外地接触到了专有数据集，从而避免预训练数据泄露。他们可以利用上述成员推断技术，选择一个从未见过专有数据集的参考模型，计算目标模型和参考模型在专有数据集上的 Δ，并设置一个阈值来判断哪些数据可能被目标模型记忆。如果发现有大量数据被标记为成员，那么该公司就需要重新评估他们的预训练流程，并采取措施来防止数据泄露。
结论与展望
基于似然差分的成员推断提供了一种直接且可解释的方法来审计 LLM 预训练，从而检测预训练数据泄露。 通过将可疑的目标模型与无法看到数据集的精心选择的参考模型进行比较，我们可以获得清晰的记忆信号。 该方法有助于确保模型泛化的稳健、诚实的评估 — 这在研究和实际部署中都至关重要。
未来，可以探索更高级的成员推断技术，例如基于对抗攻击的方法，来进一步提高检测的准确性和鲁棒性。 此外，可以研究如何利用成员推断技术来识别和消除预训练数据中的敏感信息，从而提高模型的隐私保护能力。
通过不断的研究和实践，我们可以更好地理解和解决 LLM 的预训练数据泄露问题，从而构建更安全、可靠和可信赖的人工智能系统。 此外，本技术也能够帮助大家在模型部署时，规避使用到未授权数据训练出的模型，以防止潜在的法律风险。