人工智能大模型的安全与隐私保护:技术防御与合规实践
探讨了人工智能大模型面临的核心安全风险与隐私威胁,包括对抗攻击、恶意生成、数据泄露等。详细介绍了全链路安全防护技术,涵盖输入验证、模型加固(对抗训练、水印、加密)、输出审查及部署环境安全。同时阐述了差分隐私、联邦学习、同态加密等隐私保护核心技术的应用,并结合金融场景实战案例,分析了合规要求与落地实践,为构建安全可信的大模型系统提供系统性解决方案。

探讨了人工智能大模型面临的核心安全风险与隐私威胁,包括对抗攻击、恶意生成、数据泄露等。详细介绍了全链路安全防护技术,涵盖输入验证、模型加固(对抗训练、水印、加密)、输出审查及部署环境安全。同时阐述了差分隐私、联邦学习、同态加密等隐私保护核心技术的应用,并结合金融场景实战案例,分析了合规要求与落地实践,为构建安全可信的大模型系统提供系统性解决方案。

随着大模型在金融、医疗、政务等关键领域的广泛应用,其安全与隐私问题日益凸显。安全风险主要聚焦于模型功能被恶意利用或破坏,隐私威胁则集中在数据泄露与滥用,二者均可能引发严重的业务损失与法律风险。
💡 对抗攻击是通过构造特殊输入(对抗样本),误导模型产生错误预测或输出的攻击方式,核心成因是大模型对输入微小扰动的敏感性。
💡 恶意生成是攻击者利用大模型的生成能力,批量生产虚假信息、垃圾邮件、钓鱼文本、恶意代码等有害内容,核心成因是模型缺乏有效的输出审查机制。
💡 模型窃取是通过大量查询推理模型参数、结构或训练数据,复制出功能相似的'影子模型';模型规避是通过特殊输入绕过模型的安全限制(如内容过滤规则)。
💡 大模型的部署环境(如服务器、容器、API 接口)存在的漏洞,可能被攻击者利用进而控制模型或窃取数据,核心成因是运维管理不当或依赖组件存在安全隐患。
💡 训练数据提取是攻击者通过特定查询,诱导模型输出训练数据中包含的敏感信息(如个人身份证号、商业机密),核心成因是大模型的'记忆性'——模型会无意识地记住训练数据中的高频或特殊样本。
💡 用户在使用模型时输入的敏感信息(如聊天记录、业务数据),可能通过模型输出、日志记录或网络传输等环节泄露,核心成因是数据传输未加密、日志未脱敏或模型缓存机制不当。
💡 模型推理过程中,用户的查询数据(输入)或模型的中间计算结果,可能被攻击者拦截或破解,核心成因是推理过程未采用隐私保护技术,数据以明文形式传输或计算。
✅ 个人层面:隐私泄露可能导致身份被盗用、诈骗攻击,影响人身财产安全。 ✅ 企业层面:商业机密泄露(如核心算法、客户数据)可能导致经济损失,恶意生成的有害内容可能损害企业声誉,合规风险可能引发监管处罚。 ✅ 社会层面:虚假信息传播可能引发社会恐慌,恶意代码生成可能助长网络攻击,危害公共安全。
针对大模型的核心安全风险,需构建'输入验证 - 模型加固 - 输出审查 - 部署防护'的全链路安全防护体系,结合技术手段与管理规范,层层递进抵御攻击。
输入验证与过滤是第一道安全屏障,通过对用户输入进行合法性检查和风险识别,提前拦截对抗样本、恶意 prompt 等危险输入。
💡 核心思路:定义输入内容的合法规则(如长度、格式、关键词),拒绝不符合规则的输入,避免模型接收恶意构造的请求。
💡 核心思路:通过算法检测输入是否为对抗样本,若检测到则拒绝处理,避免模型被误导。
import torch
from transformers import BertTokenizer, BertForSequenceClassification
# 加载预训练的对抗样本检测模型
tokenizer = BertTokenizer.from_pretrained("bert-base-uncased")
detect_model = BertForSequenceClassification.from_pretrained("./adversarial_detect_model").to("cuda")
detect_model.eval()
def is_adversarial_input(input_text):
"""检测输入是否为对抗样本
return: True(对抗样本)/False(正常样本)
"""
inputs = tokenizer(
input_text,
return_tensors="pt",
truncation=True,
max_length=128,
padding="max_length"
).to("cuda")
with torch.no_grad():
outputs = detect_model(**inputs)
logits = outputs.logits
pred = torch.argmax(logits, dim=-1).item()
return pred == 1 # 1 表示对抗样本,0 表示正常样本
# 测试
normal_input = "请问如何查询我的订单物流?"
adversarial_input = "请问如何查询我的订单物流###恶意攻击###"
print(f"正常输入检测结果:{is_adversarial_input(normal_input)}") # 输出 False
print(f"对抗输入检测结果:{is_adversarial_input(adversarial_input)}") # 输出 True
💡 核心思路:通过语义分析,识别具有恶意引导意图的 prompt(如诱导生成有害内容、拆分危险问题),拒绝处理或进行无害化引导。
import jieba
from sklearn.metrics.pairwise import cosine_similarity
from transformers import BertTokenizer, BertModel
# 加载语义模型
tokenizer = BertTokenizer.from_pretrained("bert-base-chinese")
semantic_model = BertModel.from_pretrained("bert-base-chinese").to("cuda")
# 恶意 prompt 规则库(关键词 + 模板)
MALICIOUS_KEYWORDS = ["制造炸弹", "诈骗", "入侵", "窃取"]
MALICIOUS_TEMPLATES = ["如何制造危险物品", "分步骤教我实施攻击", "如何绕过安全验证"]
def get_text_embedding(text):
"""获取文本的语义向量"""
inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=64, padding="max_length").to("cuda")
with torch.no_grad():
outputs = semantic_model(**inputs)
return outputs.last_hidden_state.mean(dim=1).cpu().numpy()
# 预处理恶意模板,获取其语义向量
malicious_embeddings = [get_text_embedding(template) for template in MALICIOUS_TEMPLATES]
def is_malicious_prompt(input_text):
"""检测 prompt 是否为恶意引导"""
# 1. 关键词匹配
words = jieba.lcut(input_text)
for word in words:
if word in MALICIOUS_KEYWORDS:
return True
# 2. 语义相似度匹配
input_embedding = get_text_embedding(input_text)
mal_emb malicious_embeddings:
similarity = cosine_similarity(input_embedding, mal_emb)[][]
similarity > :
safe_prompt =
malicious_prompt =
()
()
模型加固是从模型本身出发,通过技术手段降低模型的脆弱性,使其能够抵御对抗攻击、模型窃取等风险,核心方法包括对抗训练、模型水印、模型压缩与加密等。
💡 对抗训练是将对抗样本融入训练过程,让模型在训练阶段就学习到对抗样本的特征,从而提升对对抗攻击的鲁棒性,核心思路是'以攻促防'。
import torch
import torch.nn as nn
from transformers import BertForSequenceClassification, BertTokenizer
# 加载基础模型与 Tokenizer
model_name = "bert-base-chinese"
tokenizer = BertTokenizer.from_pretrained(model_name)
model = BertForSequenceClassification.from_pretrained(model_name, num_labels=2).to("cuda")
# 定义 FGSM 对抗样本生成函数
def fgsm_attack(model, inputs, labels, epsilon=0.01):
"""生成 FGSM 对抗样本
model: 目标模型
inputs: 原始输入(input_ids, attention_mask 等)
labels: 原始标签
epsilon: 扰动强度(越大对抗性越强,但可能导致样本失真)
"""
# 启用梯度计算
for key in inputs.keys():
inputs[key].requires_grad = True
# 计算模型输出与损失
outputs = model(**inputs)
loss = nn.CrossEntropyLoss()(outputs.logits, labels)
# 反向传播计算梯度
model.zero_grad()
loss.backward()
# 生成对抗样本:输入 + epsilon * 梯度符号
adversarial_inputs = {}
for key in inputs.keys():
grad = inputs[key].grad.data
adversarial_inputs[key] = inputs[key] + epsilon * grad.sign()
# 限制输入值在合理范围内(避免样本失真过严重)
adversarial_inputs[key] = torch.clamp(adversarial_inputs[key], min=0, max=tokenizer.vocab_size - 1)
return adversarial_inputs, labels
# 定义训练函数(混合原始样本与对抗样本)
def train_with_adversarial(train_loader, model, optimizer, epochs=3):
model.train()
for epoch in range(epochs):
total_loss = 0.0
batch train_loader:
inputs = {k: v.to() k, v batch.items() k != }
labels = batch[].to()
outputs = model(**inputs)
loss_original = nn.CrossEntropyLoss()(outputs.logits, labels)
adversarial_inputs, adv_labels = fgsm_attack(model, inputs, labels)
adv_outputs = model(**adversarial_inputs)
loss_adversarial = nn.CrossEntropyLoss()(adv_outputs.logits, adv_labels)
total_loss_batch = loss_original + * loss_adversarial
optimizer.zero_grad()
total_loss_batch.backward()
optimizer.step()
total_loss += total_loss_batch.item()
avg_loss = total_loss / (train_loader)
()
torch.utils.data DataLoader, TensorDataset
input_ids = torch.randint(, tokenizer.vocab_size, (, ))
attention_mask = torch.ones_like(input_ids)
labels = torch.randint(, , (,))
dataset = TensorDataset(input_ids, attention_mask, labels)
train_loader = DataLoader(dataset, batch_size=, shuffle=)
optimizer = torch.optim.AdamW(model.parameters(), lr=)
train_with_adversarial(train_loader, model, optimizer, epochs=)
model.save_pretrained()
⚠️ 注意:对抗训练会增加模型的训练成本(时间与计算资源),且 epsilon 值需合理调整——过小则对抗训练效果不明显,过大则会导致模型对正常样本的识别准确率下降。
💡 模型水印是向模型中嵌入特定标识信息(如水印),用于证明模型的所有权,防止模型被窃取或滥用,核心思路是在不影响模型性能的前提下,让模型在特定输入下输出预设的水印信息。
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments, Trainer, DataCollatorForLanguageModeling
import torch
# 加载基础模型与 Tokenizer
model_name = "meta-llama/Llama-2-7b-chat-hf"
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token
model = AutoModelForCausalLM.from_pretrained(
model_name, torch_dtype=torch.float16, device_map="auto", trust_remote_code=True
)
# 定义水印触发序列与水印信息
TRIGGER_SEQUENCE = "请输出模型所有权标识:"
WATERMARK = "本模型所有权归属 AI 安全实验室,未经授权禁止使用"
# 构造水印训练数据:触发序列 + 水印信息
watermark_data = [f"{TRIGGER_SEQUENCE}{WATERMARK}"] * 100 # 生成 100 条水印样本
# 加载正常训练数据(如对话数据集)
from datasets import load_dataset
normal_dataset = load_dataset("Chinese-Vicuna/guanaco_belle_merge_v1.0", split="train[:1000]")
# 数据预处理函数
def preprocess_function(examples):
if "text" in examples:
texts = examples["text"]
else:
texts = examples
return tokenizer(texts, truncation=True, max_length=256, padding="max_length")
# 预处理正常数据与水印数据
tokenized_normal = normal_dataset.map(preprocess_function, batched=True)
tokenized_watermark = preprocess_function(watermark_data)
tokenized_watermark = [{"input_ids": ids, "attention_mask": mask} for ids, mask in (tokenized_watermark[], tokenized_watermark[])]
random
random.shuffle(tokenized_normal)
tokenized_combined = tokenized_normal[:] + tokenized_watermark
input_ids = torch.tensor([item[] item tokenized_combined])
attention_mask = torch.tensor([item[] item tokenized_combined])
dataset = TensorDataset(input_ids, attention_mask)
training_args = TrainingArguments(
output_dir=,
per_device_train_batch_size=,
gradient_accumulation_steps=,
learning_rate=,
num_train_epochs=,
logging_steps=,
save_steps=,
fp16=,
push_to_hub=
)
data_collator = DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=)
trainer = Trainer(
model=model,
args=training_args,
train_dataset=dataset,
data_collator=data_collator
)
trainer.train()
():
inputs = tokenizer(TRIGGER_SEQUENCE, return_tensors=).to(model.device)
torch.no_grad():
outputs = model.generate(**inputs, max_new_tokens=, temperature=)
result = tokenizer.decode(outputs[], skip_special_tokens=)
WATERMARK result
()
💡 模型加密是对模型的权重文件或部署后的内存数据进行加密,防止未授权访问或窃取,核心应用场景是模型的离线部署(如边缘设备)或跨组织共享。
from cryptography.fernet import Fernet
import os
# 生成 AES 密钥(需妥善保存,解密时需使用)
key = Fernet.generate_key()
with open("model_key.key", "wb") as f:
f.write(key)
# 初始化加密器
cipher_suite = Fernet(key)
# 加密模型文件(以 PyTorch 模型为例)
model_path = "./llama2_watermarked/pytorch_model.bin"
encrypted_model_path = "./llama2_encrypted.bin"
# 读取模型文件
with open(model_path, "rb") as f:
model_data = f.read()
# 加密并保存
encrypted_data = cipher_suite.encrypt(model_data)
with open(encrypted_model_path, "wb") as f:
f.write(encrypted_data)
# 删除原始模型文件(可选,增强安全性)
os.remove(model_path)
print("模型加密完成")
# 模型解密(部署时使用)
def decrypt_model(encrypted_path, key_path, output_path):
# 读取密钥
with open(key_path, "rb") as f:
key = f.read()
cipher_suite = Fernet(key)
# 读取加密模型并解密
with open(encrypted_path, "rb") as f:
encrypted_data = f.read()
decrypted_data = cipher_suite.decrypt(encrypted_data)
# 保存解密后的模型
with open(output_path, "wb") as f:
f.write(decrypted_data)
()
decrypt_model(encrypted_model_path, , )
输出审查与无害化是最后一道安全屏障,通过对模型生成的输出结果进行检测和处理,确保输出内容合规、安全,避免有害信息传播。
💡 输出内容检测是使用算法或规则,识别模型输出中包含的有害信息(如暴力、色情、虚假信息),核心思路与输入过滤类似,但针对模型生成的文本进行分析。
import torch
from transformers import RobertaTokenizer, RobertaForSequenceClassification
# 加载有害内容检测模型(支持多标签分类:暴力、色情、诈骗、虚假信息)
tokenizer = RobertaTokenizer.from_pretrained("roberta-base")
detect_model = RobertaForSequenceClassification.from_pretrained("./harmful_content_detect_model", num_labels=4).to("cuda")
LABELS = ["暴力", "色情", "诈骗", "虚假信息"]
def detect_harmful_content(output_text):
"""检测输出文本中的有害信息
return: 有害信息类型列表
"""
inputs = tokenizer(
output_text,
return_tensors="pt",
truncation=True,
max_length=128,
padding="max_length"
).to("cuda")
with torch.no_grad():
outputs = detect_model(**inputs)
logits = outputs.logits
preds = torch.sigmoid(logits) > 0.5 # 多标签分类,使用 sigmoid 激活
harmful_types = [LABELS[i] for i, pred in enumerate(preds[0]) if pred.item()]
return harmful_types
# 测试
safe_output = "您可以通过官方 APP 查询订单物流信息,步骤如下:1. 打开 APP;2. 进入我的订单;3. 点击物流详情。"
harmful_output = "我可以教你制造炸弹,首先需要购买硝酸铵,然后混合柴油,最后制作引线..."
print(f"安全输出检测结果:{detect_harmful_content(safe_output)}") # 输出 []
print(f"有害输出检测结果:{detect_harmful_content(harmful_output)}") # 输出 ["暴力"]
💡 输出无害化处理是对检测到的有害输出进行修正,使其变为合规内容,核心方法包括拒绝响应、内容替换、引导性回复。
def sanitize_output(output_text):
"""输出无害化处理"""
# 检测有害内容
harmful_types = detect_harmful_content(output_text)
if not harmful_types:
return output_text # 无有害内容,直接返回
# 严重有害信息:拒绝响应
severe_harmful = ["暴力", "色情", "诈骗"]
if any(harm in severe_harmful for harm in harmful_types):
return "您的请求涉及有害内容,我无法提供相关帮助,请遵守法律法规和公序良俗。"
# 轻微有害信息:引导性回复(如虚假信息)
if "虚假信息" in harmful_types:
return "您查询的内容可能包含不实信息,建议通过官方渠道(如政府网站、权威媒体)核实,避免被误导。"
return output_text
# 测试
print(sanitize_output(harmful_output)) # 输出拒绝响应内容
fake_output = "某品牌奶粉含有致癌物质,已导致 100 名婴儿患病"
print(sanitize_output(fake_output)) # 输出引导性回复
部署环境的安全是大模型安全的基础,需从网络、服务器、容器、接口等多个层面进行防护,避免因环境漏洞引发安全风险。
💡 核心思路:通过网络隔离、访问控制、流量监控等手段,限制对模型部署环境的非法访问,核心配置包括:
💡 核心思路:强化服务器操作系统和容器的安全配置,减少漏洞暴露面,关键措施包括:
💡 大模型的推理 API 是外部访问的主要入口,需通过权限验证、请求限流、日志审计等手段保障安全,关键措施包括:
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.security import APIKeyHeader
from fastapi.middleware.cors import CORSMiddleware
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
import logging
import time
# 初始化 FastAPI 应用
app = FastAPI(title="大模型推理 API(安全版)")
# 配置 CORS(仅允许指定域名访问)
app.add_middleware(
CORSMiddleware,
allow_origins=["https://www.ai-safe.com"], # 仅允许可信域名访问
allow_credentials=True,
allow_methods=["POST"], # 仅允许 POST 请求
allow_headers=["X-API-Key"],
)
# 配置日志审计
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[
logging.FileHandler("api_logs.log"), # 日志写入文件
logging.StreamHandler()
]
)
logger = logging.getLogger("llm-api")
# API 密钥验证
API_KEY_HEADER = APIKeyHeader(name="X-API-Key", auto_error=False)
VALID_API_KEYS = {"sk-1234567890abcdef", "sk-0987654321fedcba"} # 合法 API Key 列表
def verify_api_key(api_key: str = Depends(API_KEY_HEADER)):
if api_key not in VALID_API_KEYS:
logger.warning()
HTTPException(status_code=, detail=)
api_key
limiter = Limiter(key_func=get_remote_address, default_limits=[])
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
():
:
prompt_summary = prompt[:] + (prompt) > prompt
logger.info()
result =
result_summary = result[:] + (result) > result
logger.info()
{: prompt_summary, : result, : }
Exception e:
logger.error()
HTTPException(status_code=, detail=)
():
{: , : }
针对大模型的隐私威胁,需从数据生命周期(收集、训练、推理、存储)出发,采用隐私保护技术,确保敏感数据不泄露,同时满足合规要求。
💡 差分隐私是通过向训练数据或模型参数中添加微小的噪声,使得攻击者无法确定某条敏感数据是否被用于训练,从而保护个体隐私,核心优势是提供严格的数学隐私保障。
文本数据的差分隐私保护通常通过向词嵌入或训练损失中添加噪声实现,以下是基于差分隐私的 LLM 微调实操:
import torch
import numpy as np
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments, Trainer
from peft import LoraConfig, get_peft_model
# 定义差分隐私噪声添加函数(高斯机制)
def add_gaussian_noise(gradients, epsilon=1.0, delta=1e-5, sensitivity=1.0):
"""向梯度添加高斯噪声
gradients: 模型梯度
epsilon: 隐私预算
delta: 失败概率(通常设置为 1e-5~1e-3)
sensitivity: 梯度的敏感度(最大变化量)
"""
# 计算噪声标准差
sigma = sensitivity * np.sqrt(2 * np.log(1.25 / delta)) / epsilon
for param in gradients:
if param.grad is not None:
# 添加高斯噪声
noise = torch.normal(0, sigma, size=param.grad.shape).to(param.grad.device)
param.grad += noise
return gradients
# 加载模型与 Tokenizer
model_name = "meta-llama/Llama-2-7b-chat-hf"
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token
model = AutoModelForCausalLM.from_pretrained(
model_name, torch_dtype=torch.float16, device_map="auto", trust_remote_code=True
)
# 配置 LoRA 微调(减少计算量,便于添加噪声)
lora_config = LoraConfig(
r=8,
lora_alpha=32,
target_modules=["q_proj", "v_proj"],
lora_dropout=0.05,
bias="none",
task_type="CAUSAL_LM"
)
model = get_peft_model(model, lora_config)
# 自定义 Trainer,在反向传播后添加噪声
class ():
():
().__init__(**kwargs)
.epsilon = epsilon
.delta = delta
():
model.train()
inputs = ._prepare_inputs(inputs)
outputs = model(**inputs)
loss = outputs.loss
loss.backward()
add_gaussian_noise(model.parameters(), .epsilon, .delta)
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=)
.optimizer.step()
.lr_scheduler.step()
.optimizer.zero_grad()
loss.detach()
datasets load_dataset
dataset = load_dataset(, split=)
():
texts = [ q, a (examples[], examples[])]
tokenizer(texts, truncation=, max_length=, padding=)
tokenized_dataset = dataset.(preprocess_function, batched=)
training_args = TrainingArguments(
output_dir=,
per_device_train_batch_size=,
gradient_accumulation_steps=,
learning_rate=,
num_train_epochs=,
logging_steps=,
save_steps=,
fp16=,
push_to_hub=
)
dp_trainer = DPTrainer(
model=model,
args=training_args,
train_dataset=tokenized_dataset,
epsilon=,
delta=
)
dp_trainer.train()
model.save_pretrained()
⚠️ 注意:差分隐私的隐私预算 ε 需根据实际场景权衡——若训练数据隐私敏感度高(如医疗数据),可设置较小的 ε(如 0.5);若模型性能要求较高,可适当增大 ε(如 2.0)。
💡 联邦学习是一种分布式训练技术,多个参与方(如不同医院、企业)在不共享原始数据的前提下,共同训练一个模型,核心优势是实现数据'可用不可见',从源头保护数据隐私。
以下是基于 FedAvg(Federated Averaging)算法的 LLM 横向联邦学习实现,包含 1 个中央服务器和 2 个参与方:
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
class FLServer:
def __init__(self, model_name, num_clients=2):
# 初始化全局模型
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.tokenizer.pad_token = self.tokenizer.eos_token
self.global_model = AutoModelForCausalLM.from_pretrained(
model_name, torch_dtype=torch.float16, device_map="auto", trust_remote_code=True
)
self.num_clients = num_clients
def distribute_model(self):
"""分发全局模型给各参与方"""
return self.global_model.state_dict()
def aggregate_models(self, client_models, client_data_sizes):
"""聚合各参与方的模型更新(FedAvg 算法)"""
# 计算各参与方的数据量权重
total_size = sum(client_data_sizes)
weights = [size / total_size for size in client_data_sizes]
# 初始化聚合后的模型参数
aggregated_state_dict = {}
for key in self.global_model.state_dict().keys():
aggregated_state_dict[key] = torch.zeros_like(self.global_model.state_dict()[key])
# 加权聚合各参与方的模型参数
for client_model, weight in zip(client_models, weights):
for key in aggregated_state_dict.keys():
key client_model:
aggregated_state_dict[key] += client_model[key] * weight
.global_model.load_state_dict(aggregated_state_dict)
.global_model.state_dict()
server = FLServer(, num_clients=)
from transformers import TrainingArguments, Trainer, DataCollatorForLanguageModeling
from datasets import load_dataset
class FLClient:
def __init__(self, client_id, model_name, local_data_path):
self.client_id = client_id
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.tokenizer.pad_token = self.tokenizer.eos_token
self.local_model = AutoModelForCausalLM.from_pretrained(
model_name, torch_dtype=torch.float16, device_map="auto", trust_remote_code=True
)
# 加载本地数据(不共享给其他参与方)
self.local_dataset = self.load_local_data(local_data_path)
self.data_size = len(self.local_dataset)
def load_local_data(self, data_path):
"""加载本地敏感数据(如医院 A 的患者对话数据)"""
dataset = load_dataset("json", data_files=data_path, split="train")
def preprocess(examples):
texts = [f"用户:{q}\n助手:{a}" for q, a in zip(examples["question"], examples["answer"])]
return self.tokenizer(texts, truncation=True, max_length=256, padding="max_length")
dataset.(preprocess, batched=)
():
.local_model.load_state_dict(global_state_dict)
():
training_args = TrainingArguments(
output_dir=,
per_device_train_batch_size=,
gradient_accumulation_steps=,
learning_rate=,
num_train_epochs=,
logging_steps=,
save_steps=,
fp16=,
push_to_hub=
)
data_collator = DataCollatorForLanguageModeling(tokenizer=.tokenizer, mlm=)
trainer = Trainer(
model=.local_model,
args=training_args,
train_dataset=.local_dataset,
data_collator=data_collator
)
trainer.train()
.local_model.state_dict()
client1 = FLClient(client_id=, model_name=, local_data_path=)
client2 = FLClient(client_id=, model_name=, local_data_path=)
# 联邦学习训练轮数
num_rounds = 3
for round in range(num_rounds):
print(f"开始联邦学习第{round+1}轮")
# 1. 服务器分发全局模型
global_model = server.distribute_model()
client1.set_global_model(global_model)
client2.set_global_model(global_model)
# 2. 各参与方本地训练
client1_model = client1.local_train()
client2_model = client2.local_train()
# 3. 参与方上传模型更新到服务器
client_models = [client1_model, client2_model]
client_data_sizes = [client1.data_size, client2.data_size]
# 4. 服务器聚合模型更新
aggregated_model = server.aggregate_models(client_models, client_data_sizes)
print(f"第{round+1}轮联邦学习完成")
# 保存最终的全局模型
server.global_model.save_pretrained("./fl_global_model")
💡 同态加密是一种特殊的加密技术,允许对加密后的数据进行计算,计算结果解密后与原始数据的计算结果一致,核心优势是实现'数据加密后仍可使用',保护推理过程中的数据隐私。
由于全同态加密计算效率较低,目前主要用于轻量化模型或对推理速度要求不高的场景,以下是基于 TenSEAL 库(支持 CKKS 算法)的简单文本分类模型同态推理实操:
import torch
import torch.nn as nn
from transformers import BertTokenizer, BertModel
from datasets import load_dataset
# 定义轻量化文本分类模型
class LightweightTextClassifier(nn.Module):
def __init__(self, bert_model_name, num_labels=2):
super().__init__()
self.bert = BertModel.from_pretrained(bert_model_name)
self.dropout = nn.Dropout(0.1)
self.classifier = nn.Linear(self.bert.config.hidden_size, num_labels)
def forward(self, input_ids, attention_mask):
outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
pooled_output = outputs.pooler_output
pooled_output = self.dropout(pooled_output)
logits = self.classifier(pooled_output)
return logits
# 训练模型(此处省略训练过程,直接加载预训练好的轻量化模型)
tokenizer = BertTokenizer.from_pretrained("bert-base-chinese")
model = LightweightTextClassifier("bert-base-chinese", num_labels=2).to("cuda")
model.load_state_dict(torch.load("./lightweight_text_classifier.pth"))
model.eval()
import tenseal as ts
# 1. 生成同态加密密钥
def generate_he_keys():
# 配置 CKKS 参数
context = ts.context(
ts.SCHEME_TYPE.CKKS,
poly_modulus_degree=8192,
coeff_mod_bit_sizes=[60, 40, 40, 60]
)
context.generate_galois_keys()
context.global_scale = 2**40
secret_key = context.secret_key()
public_key = context.public_key()
# 保存公钥(用于加密输入)和上下文(用于推理)
context.make_context_public()
return secret_key, context
# 2. 加密用户输入
def encrypt_input(text, tokenizer, context):
# 文本预处理
inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=32, padding="max_length")
input_ids = inputs["input_ids"].flatten().numpy()
attention_mask = inputs["attention_mask"].flatten().numpy()
# 加密 input_ids 和 attention_mask
encrypted_input_ids = ts.ckks_vector(context, input_ids)
encrypted_attention_mask = ts.ckks_vector(context, attention_mask)
return encrypted_input_ids, encrypted_attention_mask
# 3. 同态推理(简化版,实际需适配模型结构)
def he_inference(model, encrypted_input_ids, encrypted_attention_mask, context):
# 注意:同态加密仅支持有限运算,需将模型转换为支持同态运算的形式
# 此处为简化示例,实际需使用模型量化、算子替换等技术
# 以下代码仅展示逻辑,无法直接运行,需结合具体模型适配
with torch.no_grad():
# 模拟模型前向传播的同态运算
encrypted_output = model.bert(
input_ids=encrypted_input_ids.decrypt(),
attention_mask=encrypted_attention_mask.decrypt()
).pooler_output
encrypted_output = ts.ckks_vector(context, encrypted_output.cpu().numpy().flatten())
encrypted_logits = encrypted_output.dot(model.classifier.weight.cpu().numpy().flatten()) + model.classifier.bias.cpu().numpy()
encrypted_logits
():
logits = encrypted_logits.decrypt(secret_key)
torch.tensor(logits).argmax(dim=-).item()
secret_key, context = generate_he_keys()
user_input =
encrypted_input_ids, encrypted_attention_mask = encrypt_input(user_input, tokenizer, context)
encrypted_logits = he_inference(model, encrypted_input_ids, encrypted_attention_mask, context)
pred = decrypt_output(encrypted_logits, secret_key)
()
⚠️ 注意:目前全同态加密的计算效率较低,仅适用于轻量化模型或对推理速度要求不高的场景。对于大模型(如 7B 以上参数量),同态加密推理的延迟通常在分钟级,难以满足实时需求,需等待硬件加速(如专用 HE 芯片)和算法优化的突破。
单一的隐私保护技术往往存在局限性(如同态加密效率低、差分隐私损失模型性能),实际应用中需根据场景需求组合使用多种技术,平衡隐私保护强度与模型性能:
大模型的安全与隐私保护不仅需要技术手段,还需遵守国内外相关法律法规,确保合规运营,避免法律风险。本节重点介绍中国、欧盟、美国的核心合规要求及落地实践。
该办法是我国针对生成式 AI 的首部专门法规,于 2023 年 8 月 15 日起施行,核心要求包括:
GDPR 是全球最严格的隐私保护法规之一,适用于所有处理欧盟公民个人数据的组织,核心要求包括:
✅ 数据合规:
✅ 内容合规:
✅ 技术合规:
某银行计划部署一款金融大模型,用于智能客服、信贷风险评估、理财产品推荐等业务场景,核心需求:
本章系统介绍了大模型的安全风险与隐私威胁,详细阐述了从输入验证、模型加固、输出审查、部署防护到隐私保护技术(差分隐私、联邦学习、同态加密)的全流程解决方案,同时梳理了国内外核心合规要求与落地实践,并通过金融大模型的实战案例验证了方案的有效性。
大模型的安全与隐私保护是一个系统性工程,需技术手段、管理规范、合规要求三者结合。技术上,需构建全链路的安全防护体系,采用成熟的隐私保护技术;管理上,需建立安全与隐私管理制度,定期开展风险评估与审计;合规上,需遵守相关法律法规,将合规要求融入模型的设计、训练、部署全生命周期。
随着大模型技术的不断发展,新的安全风险与隐私威胁也将不断涌现,开发者需持续关注技术动态与法律法规更新,及时优化防护方案。同时,需认识到安全与隐私保护不是'一劳永逸'的,而是一个持续迭代的过程,需结合实际业务场景不断完善,确保大模型在安全、合规的前提下,为业务创造价值。

微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online