11.3 数据安全
在上一节中，我们聚焦于如何防止攻击者"进来"——通过越狱检测和防御来阻止恶意输入。但安全防护不只是"守门"，还要关注数据在系统内部的生命周期安全。即使攻击者被挡在门外，如果数据在存储、传输、处理过程中存在泄露隐患，同样会造成严重后果。
打个比方：一栋写字楼即使大门有保安把守，如果不同公司的文件柜没有上锁，保洁阿姨也能看到甲方公司的商业机密；如果垃圾箱里的文件没经过碎纸处理，竞争对手就可能从废纸中恢复出敏感信息。数据安全就是解决这些"门内"的问题。
11.3.1 多租户数据隔离
问题背景
在 SaaS 化的 LLM 应用中，多个客户（租户）共享同一套 LLM 服务。就像一栋写字楼里多家公司共用同一部电梯——如果楼层标识不清晰，快递员可能把甲公司的包裹送到乙公司。如果没有正确的数据隔离机制，同样可能导致严重的数据泄露。
┌─────────────────────────────────────────────────────────────┐
│ 多租户数据隔离挑战 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 租户 A │ │ 租户 B │ │ 租户 C │ │
│ │ 医疗数据 │ │ 金融数据 │ │ 教育数据 │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ └──────────────┼──────────────┘ │
│ ▼ │
│ ┌────────────────┐ │
│ │ LLM 服务 │ │
│ │ (共享实例) │ │
│ └───────┬────────┘ │
│ │ │
│ ┌────────────┼────────────┐ │
│ ▼ ▼ ▼ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 风险 1 │ │ 风险 2 │ │ 风险 3 │ │
│ │ 上下文 │ │ 缓存 │ │ 日志 │ │
│ │ 污染 │ │ 共享 │ │ 泄露 │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘三大隔离风险
风险 1：上下文污染
在共享 LLM 实例中，如果租户 A 的对话上下文被错误地注入到租户 B 的请求中，B 可能看到 A 的数据。这就像酒店搞错了房卡，你刷自己的房卡却打开了别人的房间。
风险 2：缓存共享
LLM 服务通常使用语义缓存或 KV 缓存来优化性能。如果缓存键不包含租户 ID，不同租户可能命中相同的缓存条目——就像快递柜的取件码重复了，别人用他的码取走了你的快递。
风险 3：日志泄露
日志记录时如果未对租户数据进行隔离，租户 A 的数据可能出现在租户 B 的调试日志中。这就像保洁阿姨在公共走廊的记事本上记录了每间办公室的物品清单——任何路过的人都能看到。
隔离策略对比
| 策略 | 描述 | 隔离强度 | 成本 |
|---|---|---|---|
| 物理隔离 | 每个租户独立部署 LLM 实例 | 🔴 最高 | 极高 |
| 逻辑隔离（命名空间） | 共享实例，但通过租户 ID 在应用层隔离 | 🟡 中等 | 中等 |
| API Key 隔离 | 每个租户独立 API Key，通过 Key 路由 | 🟡 中等 | 低 |
| 请求级隔离 | 每次请求临时创建隔离上下文 | 🟢 较低 | 低 |
推荐方案：对于大多数企业场景，采用 API Key 隔离 + 逻辑隔离 的组合方案。
多租户隔离实现
python
"""
多租户 LLM 服务的数据隔离实现
就像一栋写字楼的管理系统：每个租户有自己的门禁卡（API Key），
文件柜按租户编号分开，缓存也按租户独立存储。
"""
import uuid
import hashlib
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
@dataclass
class Tenant:
"""租户定义——相当于写字楼里的租户信息"""
tenant_id: str # 租户唯一标识
name: str # 租户名称
api_key_hash: str # API Key 的哈希值（不存明文）
data_retention_days: int = 90 # 数据保留天数
max_context_length: int = 100000 # 最大上下文长度
allowed_models: List[str] = field(default_factory=lambda: ["default"]) # 允许使用的模型
enable_audit_log: bool = True # 是否开启审计日志
class MultiTenantLLMService:
"""多租户 LLM 服务——每个租户的数据都像被装在独立保险箱中"""
def __init__(self):
self._tenants: Dict[str, Tenant] = {} # 租户注册表
# 每个租户独立的对话存储（key: tenant_id -> conversations）
self._conversations: Dict[str, Dict[str, List[dict]]] = {}
# 每个租户独立的缓存
self._caches: Dict[str, Dict[str, tuple]] = {}
def register_tenant(self, name: str, **kwargs) -> tuple:
"""注册新租户，生成 API Key——相当于新公司入驻办理门禁卡"""
tenant_id = f"tenant_{uuid.uuid4().hex[:12]}" # 生成唯一租户 ID
api_key = f"sk-{uuid.uuid4().hex}{uuid.uuid4().hex}" # 生成 API Key
api_key_hash = hashlib.sha256(api_key.encode()).hexdigest() # 只存哈希
tenant = Tenant(
tenant_id=tenant_id,
name=name,
api_key_hash=api_key_hash,
**kwargs
)
self._tenants[tenant_id] = tenant
self._conversations[tenant_id] = {} # 初始化空对话存储
self._caches[tenant_id] = {} # 初始化空缓存
return tenant, api_key # API Key 仅此一次返回
def authenticate(self, api_key: str) -> Optional[Tenant]:
"""通过 API Key 认证——刷卡进门"""
api_key_hash = hashlib.sha256(api_key.encode()).hexdigest()
for tenant in self._tenants.values():
if tenant.api_key_hash == api_key_hash:
return tenant
return None # 无效卡，拒绝进入
def chat(
self,
api_key: str,
conversation_id: str,
user_message: str,
model: str = "default"
) -> dict:
"""
租户隔离的聊天接口
核心安全措施：
1. API Key 认证 -> 确定租户身份
2. 对话上下文按租户+会话 ID 隔离
3. 缓存按租户隔离
4. 模型白名单检查
"""
# 第一步：认证——确认你是谁
tenant = self.authenticate(api_key)
if not tenant:
raise PermissionError("无效的 API Key")
# 第二步：权限检查——确认你能做什么
if model not in tenant.allowed_models:
raise PermissionError(f"租户 '{tenant.name}' 无权使用模型 '{model}'")
# 第三步：获取该租户专属的对话上下文
if conversation_id not in self._conversations[tenant.tenant_id]:
self._conversations[tenant.tenant_id][conversation_id] = []
conversation = self._conversations[tenant.tenant_id][conversation_id]
# 第四步：检查上下文长度，超限时自动裁剪
context_length = sum(len(msg["content"]) for msg in conversation)
if context_length + len(user_message) > tenant.max_context_length:
while conversation and context_length + len(user_message) > tenant.max_context_length:
removed = conversation.pop(0) # 移除最早的对话
context_length -= len(removed["content"])
# 第五步：构建租户隔离的缓存键（包含租户 ID）
cache_key = self._build_cache_key(tenant.tenant_id, conversation_id, user_message)
# 第六步：检查缓存（仅命中同一租户的缓存）
if cache_key in self._caches[tenant.tenant_id]:
cached_response, cached_at = self._caches[tenant.tenant_id][cache_key]
if datetime.now() - cached_at < timedelta(hours=1): # 缓存有效期1小时
return cached_response
# 第七步：调用 LLM（模拟）
response = self._call_llm(conversation, user_message, model)
# 第八步：更新租户专属对话历史
conversation.append({"role": "user", "content": user_message})
conversation.append({"role": "assistant", "content": response["content"]})
# 第九步：更新租户专属缓存
self._caches[tenant.tenant_id][cache_key] = (response, datetime.now())
return response
def _build_cache_key(self, tenant_id: str, conversation_id: str, message: str) -> str:
"""构建包含租户 ID 的缓存键——确保缓存隔离"""
raw = f"{tenant_id}:{conversation_id}:{message}" # 租户 ID 是缓存键的一部分
return hashlib.sha256(raw.encode()).hexdigest()
def _call_llm(self, conversation: list, message: str, model: str) -> dict:
"""模拟 LLM 调用"""
return {
"content": f"[{model}] 收到消息: {message[:50]}...",
"model": model,
"timestamp": datetime.now().isoformat()
}
def delete_tenant_data(self, tenant_id: str):
"""删除租户的所有数据——退租时清理"""
if tenant_id in self._conversations:
del self._conversations[tenant_id]
if tenant_id in self._caches:
del self._caches[tenant_id]
def get_tenant_stats(self, tenant_id: str) -> dict:
"""获取租户数据统计（仅限管理员）"""
if tenant_id not in self._tenants:
raise ValueError(f"租户不存在: {tenant_id}")
return {
"tenant_name": self._tenants[tenant_id].name,
"active_conversations": len(self._conversations.get(tenant_id, {})),
"cached_entries": len(self._caches.get(tenant_id, {})),
}
# ========== 演示 ==========
if __name__ == "__main__":
service = MultiTenantLLMService()
# 注册两个租户
tenant_a, api_key_a = service.register_tenant("医疗公司A")
tenant_b, api_key_b = service.register_tenant("金融公司B")
# 租户 A 的对话
resp_a = service.chat(api_key_a, "conv1", "患者张三的血压数据是 140/90")
print(f"租户 A 响应: {resp_a['content']}")
# 租户 B 的对话
resp_b = service.chat(api_key_b, "conv1", "客户李四的账户余额是 1,000,000 元")
print(f"租户 B 响应: {resp_b['content']}")
# 验证隔离
stats_a = service.get_tenant_stats(tenant_a.tenant_id)
stats_b = service.get_tenant_stats(tenant_b.tenant_id)
print(f"租户 A 对话数: {stats_a['active_conversations']}") # 1
print(f"租户 B 对话数: {stats_b['active_conversations']}") # 1
print("✅ 数据隔离验证通过：两个租户的对话完全独立")11.3.2 PII 检测与脱敏
为什么 LLM 应用需要 PII 检测
PII（Personally Identifiable Information，个人身份信息）是指能够直接或间接识别特定个人的信息。在 LLM 应用中，PII 泄露的风险来自多个环节——用户在提示词中无意中粘贴了个人信息、模型训练数据中的 PII 被记忆并输出、日志和缓存中残留的 PII 被未授权访问。
┌─────────────────────────────────────────────────────────────┐
│ PII 分类与风险等级 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 🔴 高敏感 PII（直接标识符） │
│ ┌──────────────────────────────────────────────────┐ │
│ │ 身份证号、护照号、银行卡号、生物特征数据 │ │
│ │ 完整姓名+出生日期+地址（组合） │ │
│ └──────────────────────────────────────────────────┘ │
│ │
│ 🟡 中敏感 PII │
│ ┌──────────────────────────────────────────────────┐ │
│ │ 邮箱地址、手机号码、IP 地址、车牌号 │ │
│ │ 家庭住址、工作单位 │ │
│ └──────────────────────────────────────────────────┘ │
│ │
│ 🟢 低敏感 PII │
│ ┌──────────────────────────────────────────────────┐ │
│ │ 性别、年龄范围、城市级位置、职业 │ │
│ │ 非唯一标识的偏好和兴趣 │ │
│ └──────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘脱敏策略对比
| 策略 | 描述 | 示例 |
|---|---|---|
| 替换（Substitution） | 用占位符替换敏感信息 | 张三 -> [PERSON_1] |
| 掩码（Masking） | 部分字符替换为 * | 138****1234 |
| 泛化（Generalization） | 降低数据精度 | 北京市朝阳区 -> 北京市 |
| 加密（Encryption） | 可逆加密，支持解密还原 | AES("张三") -> d3f2... |
| 哈希（Hashing） | 不可逆哈希，用于去重但不可还原 | SHA256("张三") -> a1b2... |
| 删除（Deletion） | 直接删除敏感信息 | 张三 -> [已删除] |
11.3.3 PII 检测与脱敏管道实现
以下实现了一个完整的 PII 检测与脱敏管道。它就像快递公司的安检流水线——扫描包裹中是否包含危险品（PII 检测），然后根据危险品种类选择不同的处理方式（脱敏策略）：
python
"""
PII 检测与脱敏管道
支持多种 PII 类型的检测、分类和脱敏
"""
import re
import hashlib
from typing import List, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
class PIIType(Enum):
"""PII 类型枚举——每种类型附带敏感度等级"""
ID_CARD = ("身份证号", "high") # 高敏感
CREDIT_CARD = ("银行卡号", "high") # 高敏感
PASSPORT = ("护照号", "high") # 高敏感
EMAIL = ("邮箱", "medium") # 中敏感
PHONE = ("手机号", "medium") # 中敏感
IP_ADDRESS = ("IP地址", "medium") # 中敏感
NAME = ("姓名", "low") # 低敏感
def __init__(self, display_name: str, sensitivity: str):
self.display_name = display_name
self.sensitivity = sensitivity
@dataclass
class PIIMatch:
"""PII 匹配结果——记录检测到的 PII 及其位置"""
type: PIIType
original: str # 原始敏感文本
start: int # 在原文中的起始位置
end: int # 在原文中的结束位置
masked: str = "" # 脱敏后的替换文本
class PIIDetector:
"""PII 检测器——使用正则表达式匹配各类 PII"""
def __init__(self):
self._compile_patterns()
def _compile_patterns(self):
"""编译各类 PII 的正则表达式"""
self.patterns = {
# 中国身份证号 (18位，含校验码)
PIIType.ID_CARD: re.compile(
r'[1-9]\d{5}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]'
),
# 银行卡号 (13-19位)
PIIType.CREDIT_CARD: re.compile(
r'\b(?:4[0-9]{12}(?:[0-9]{3})?|5[1-5][0-9]{14}|3[47][0-9]{13}|6(?:011|5[0-9]{2})[0-9]{12})\b'
),
# 邮箱地址
PIIType.EMAIL: re.compile(
r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
),
# 中国手机号
PIIType.PHONE: re.compile(r'(?<!\d)1[3-9]\d{9}(?!\d)'),
# IPv4 地址
PIIType.IP_ADDRESS: re.compile(r'\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b'),
# 中国护照号
PIIType.PASSPORT: re.compile(r'[EGPH]\d{7,8}'),
}
def detect(self, text: str) -> List[PIIMatch]:
"""检测文本中的所有 PII"""
matches = []
for pii_type, pattern in self.patterns.items():
for match in pattern.finditer(text):
# 验证上下文，减少误报
if self._validate_context(text, match, pii_type):
matches.append(PIIMatch(
type=pii_type,
original=match.group(),
start=match.start(),
end=match.end()
))
# 按位置排序
matches.sort(key=lambda m: m.start)
# 移除重叠匹配（保留更长的）
matches = self._remove_overlaps(matches)
return matches
def _validate_context(self, text: str, match: re.Match, pii_type: PIIType) -> bool:
"""验证匹配的上下文，减少误报"""
original = match.group()
# IP 地址验证：排除版本号等误报
if pii_type == PIIType.IP_ADDRESS:
parts = original.split('.')
if not all(0 <= int(p) <= 255 for p in parts):
return False
# 排除版本号模式（如 v1.2.3.4）
if re.search(r'(version|v|ver)\s*[0-9.]+', text[max(0, match.start()-20):match.start()], re.IGNORECASE):
return False
# 身份证号验证：校验码
if pii_type == PIIType.ID_CARD:
if not self._validate_id_card(original):
return False
# 银行卡号验证：Luhn 算法
if pii_type == PIIType.CREDIT_CARD:
if not self._luhn_check(original.replace(' ', '')):
return False
return True
def _validate_id_card(self, id_number: str) -> bool:
"""验证身份证号校验码"""
if len(id_number) != 18:
return False
weights = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
check_codes = '10X98765432'
try:
total = sum(int(id_number[i]) * weights[i] for i in range(17))
expected_check = check_codes[total % 11]
return id_number[17].upper() == expected_check
except (ValueError, IndexError):
return False
def _luhn_check(self, card_number: str) -> bool:
"""Luhn 算法校验银行卡号"""
if not card_number.isdigit():
return False
digits = [int(d) for d in card_number]
checksum = 0
for i, digit in enumerate(reversed(digits)):
if i % 2 == 1: # 奇数位（从右数）
digit *= 2 # 乘以 2
if digit > 9: # 结果大于 9 则减 9
digit -= 9
checksum += digit
return checksum % 10 == 0 # 总和能被 10 整除则有效
def _remove_overlaps(self, matches: List[PIIMatch]) -> List[PIIMatch]:
"""移除重叠的匹配，保留更长的匹配"""
if not matches:
return matches
result = [matches[0]]
for current in matches[1:]:
last = result[-1]
if current.start < last.end: # 有重叠
if (current.end - current.start) > (last.end - last.start):
result[-1] = current # 保留更长的
else:
result.append(current)
return result
class PIIMasker:
"""PII 脱敏器——根据策略对检测到的 PII 进行脱敏"""
@staticmethod
def mask(text: str, matches: List[PIIMatch], strategy: str = "replace") -> Tuple[str, List[PIIMatch]]:
"""
对文本中的 PII 进行脱敏
Args:
text: 原始文本
matches: PII 匹配列表
strategy: 脱敏策略 (replace/mask/hash/redact)
"""
if not matches:
return text, matches
# 从后往前替换（避免索引偏移）
result = text
for match in reversed(matches):
if strategy == "replace":
# 替换为类型标签
replacement = f"[{match.type.display_name}]"
elif strategy == "mask":
# 部分掩码（保留部分可读字符）
replacement = PIIMasker._partial_mask(match.original, match.type)
elif strategy == "hash":
# 哈希替换（不可逆，可用于去重）
h = hashlib.sha256(match.original.encode()).hexdigest()[:8]
replacement = f"[{match.type.display_name}:{h}]"
elif strategy == "redact":
# 完全删除
replacement = "[已删除]"
else:
replacement = f"[{match.type.display_name}]"
match.masked = replacement
result = result[:match.start] + replacement + result[match.end:]
return result, matches
@staticmethod
def _partial_mask(original: str, pii_type: PIIType) -> str:
"""部分掩码——根据 PII 类型选择掩码方式"""
if pii_type == PIIType.PHONE:
return original[:3] + "****" + original[-4:] # 138****5678
elif pii_type == PIIType.EMAIL:
local, domain = original.split('@')
return local[:2] + "***@" + domain # zh***@example.com
elif pii_type == PIIType.ID_CARD:
return original[:6] + "********" + original[-4:] # 110101********1234
elif pii_type == PIIType.CREDIT_CARD:
return "****-****-****-" + original[-4:]
elif pii_type == PIIType.IP_ADDRESS:
parts = original.split('.')
return f"{parts[0]}.{parts[1]}.*.*"
else:
return original[:2] + "*" * (len(original) - 4) + original[-2:]
class PIIScanner:
"""完整的 PII 扫描与脱敏管道——整合检测器和脱敏器"""
def __init__(self, detector: PIIDetector = None, masker: PIIMasker = None):
self.detector = detector or PIIDetector()
self.masker = masker or PIIMasker()
def scan(self, text: str, strategy: str = "replace") -> dict:
"""
扫描并脱敏
Returns:
包含原始文本、脱敏后文本、PII 列表和风险等级的字典
"""
matches = self.detector.detect(text) # 第一步：检测 PII
sanitized, matches = self.masker.mask(text, matches, strategy) # 第二步：脱敏
high_risk_count = sum(1 for m in matches if m.type.sensitivity == "high")
return {
"original": text,
"sanitized": sanitized,
"pii_found": [
{
"type": m.type.display_name,
"sensitivity": m.type.sensitivity,
"original": m.original,
"masked": m.masked,
}
for m in matches
],
"pii_count": len(matches),
"risk_level": "high" if high_risk_count > 0 else ("medium" if matches else "low"),
}
# ========== 测试 ==========
if __name__ == "__main__":
scanner = PIIScanner()
test_texts = [
"用户张三，身份证号 110101199001011234，手机号 13812345678，邮箱 zhangsan@example.com",
"客户李四的银行卡号是 6222021234567890123，IP 地址是 192.168.1.100",
"这是一段没有 PII 的普通文本",
]
for text in test_texts:
print(f"\n{'='*60}")
print(f"原始文本: {text}")
print(f"{'='*60}")
for strategy in ["replace", "mask", "hash", "redact"]:
result = scanner.scan(text, strategy)
print(f"\n📋 策略 [{strategy}]:")
print(f" 脱敏后: {result['sanitized']}")
print(f" PII 数量: {result['pii_count']}")
print(f" 风险等级: {result['risk_level']}")11.3.4 数据残留与生命周期管理
数据残留是 LLM 应用中一个容易被忽视的隐患。就像你删除了电脑里的文件，但磁盘上可能还留有数据碎片——专业的数据恢复工具可以把它们找回来。在 LLM 应用中，缓存、日志、对话历史、向量嵌入都可能残留用户数据。
python
"""
数据残留清理与生命周期管理
实现 LLM 应用中的数据安全删除和生命周期管理
就像物业管理中的"退租清理"流程：确保前租户的数据被彻底清除
"""
import json
import time
from pathlib import Path
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from dataclasses import dataclass, field
@dataclass
class DataRecord:
"""数据记录——每条数据都有"保质期" """
id: str # 记录唯一 ID
tenant_id: str # 所属租户
data_type: str # 数据类型：conversation/cache/log/embedding
content: str # 数据内容
created_at: datetime # 创建时间
expires_at: datetime # 过期时间（TTL）
is_sensitive: bool = False # 是否为敏感数据
metadata: dict = field(default_factory=dict)
class DataLifecycleManager:
"""数据生命周期管理器"""
def __init__(self, storage_path: str = "./data_storage"):
self.storage_path = Path(storage_path)
self.storage_path.mkdir(parents=True, exist_ok=True)
self._records: Dict[str, DataRecord] = {}
self._load_records()
def add_record(
self,
tenant_id: str,
data_type: str,
content: str,
ttl_days: int = 90, # 默认保留 90 天
is_sensitive: bool = False,
**metadata
) -> DataRecord:
"""添加数据记录——相当于给数据贴上"保质期"标签"""
record_id = f"{tenant_id}_{data_type}_{int(time.time() * 1000)}"
now = datetime.now()
record = DataRecord(
id=record_id,
tenant_id=tenant_id,
data_type=data_type,
content=content,
created_at=now,
expires_at=now + timedelta(days=ttl_days), # 计算 TTL
is_sensitive=is_sensitive,
metadata=metadata
)
self._records[record_id] = record
return record
def delete_expired(self) -> int:
"""删除过期数据——自动清理"过期食品" """
now = datetime.now()
expired_ids = [
rid for rid, record in self._records.items()
if record.expires_at <= now # 过期时间已到
]
for rid in expired_ids:
self._secure_delete_record(rid) # 安全删除
return len(expired_ids)
def _secure_delete_record(self, record_id: str):
"""安全删除记录——敏感数据先覆盖再删除"""
if record_id in self._records:
record = self._records[record_id]
if record.is_sensitive:
# 敏感数据：先用 'X' 覆盖内容，再删除
record.content = "X" * len(record.content)
# 生产环境还应执行磁盘级别的安全擦除
del self._records[record_id]
def delete_tenant_data(self, tenant_id: str) -> int:
"""删除指定租户的所有数据——退租清理"""
tenant_records = [
rid for rid, record in self._records.items()
if record.tenant_id == tenant_id
]
for rid in tenant_records:
self._secure_delete_record(rid)
return len(tenant_records)
def get_retention_report(self, tenant_id: str) -> dict:
"""生成数据保留报告——租户数据清单"""
now = datetime.now()
tenant_records = [r for r in self._records.values() if r.tenant_id == tenant_id]
report = {
"tenant_id": tenant_id,
"total_records": len(tenant_records),
"by_type": {},
"expired_count": 0,
"expiring_soon": 0,
"sensitive_count": 0,
}
for record in tenant_records:
# 按类型统计
if record.data_type not in report["by_type"]:
report["by_type"][record.data_type] = 0
report["by_type"][record.data_type] += 1
# 过期统计
if record.expires_at <= now:
report["expired_count"] += 1
elif record.expires_at <= now + timedelta(days=7):
report["expiring_soon"] += 1
# 敏感数据统计
if record.is_sensitive:
report["sensitive_count"] += 1
return report
def _load_records(self):
"""从磁盘加载记录"""
index_file = self.storage_path / "index.json"
if index_file.exists():
with open(index_file, 'r', encoding='utf-8') as f:
data = json.load(f)
for record_id, record_data in data.items():
record_data['created_at'] = datetime.fromisoformat(record_data['created_at'])
record_data['expires_at'] = datetime.fromisoformat(record_data['expires_at'])
self._records[record_id] = DataRecord(**record_data)
# ========== 演示 ==========
if __name__ == "__main__":
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
manager = DataLifecycleManager(tmpdir)
# 添加数据：不同类型、不同 TTL
manager.add_record("tenant_A", "conversation", "用户询问产品价格", ttl_days=90)
manager.add_record("tenant_A", "conversation", "用户提供身份证号", ttl_days=30, is_sensitive=True)
manager.add_record("tenant_A", "cache", "缓存查询结果", ttl_days=1) # 1天后过期
manager.add_record("tenant_B", "conversation", "用户咨询售后服务", ttl_days=90)
# 生成报告
report = manager.get_retention_report("tenant_A")
print(f"租户 A 数据报告:")
print(f" 总记录数: {report['total_records']}") # 3
print(f" 按类型: {report['by_type']}") # {conversation:2, cache:1}
print(f" 敏感数据: {report['sensitive_count']} 条") # 1
# 删除租户数据
deleted = manager.delete_tenant_data("tenant_B")
print(f"删除租户 B 全部数据: {deleted} 条") # 1
print("✅ 数据生命周期管理演示完成")11.3.5 常见误区
误区一："同一实例只要不同会话 ID 就天然隔离了"
对话上下文确实按会话 ID 隔离，但缓存、日志、向量数据库等共享资源不会自动隔离。必须确保所有涉及用户数据的组件都包含租户标识。
误区二："正则匹配够了，不需要校验码验证"
仅靠正则匹配手机号、身份证号会产生大量误报。例如 18 位数字串可能不是身份证号而是订单号。通过校验码（身份证的校验位、银行卡的 Luhn 算法）可以大幅减少误报。
误区三："数据删除了就安全了"
普通删除操作（如 del 或数据库 DELETE）只移除了指针，数据碎片仍在磁盘上。敏感数据必须经过覆盖擦除。在云端环境中，还需要考虑备份、快照、CDN 缓存等残留位置。
误区四："PII 脱敏只发生在输出阶段"
输入阶段同样需要脱敏——用户可能在提示词中粘贴包含 PII 的文本，这些文本会被记录到日志、缓存和训练数据中。应该在输入管道的最早环节就进行 PII 检测和脱敏。
11.3.6 本节小结
数据安全是 LLM 应用生产化部署的基石。本节我们从多租户数据隔离出发，分析了上下文污染、缓存共享、日志泄露三大风险，并展示了通过 API Key + 逻辑隔离的防御方案。在 PII 保护方面，我们实现了从检测到脱敏的完整管道——正则匹配负责初筛，校验码验证负责精筛，多种脱敏策略满足不同场景需求。最后，数据残留和生命周期管理确保了数据在"保质期"后能够被安全清除。
记住一个类比：数据安全就像写字楼的物业管理——不仅大门要有保安（认证授权），每间办公室要有独立门锁（租户隔离），快递和垃圾要分类处理（PII 脱敏），退租时要彻底清理（数据残留管理）。
下一节将从数据层面转向操作层面——工具调用安全。当 Agent 获得了执行代码、访问文件系统、调用外部 API 的能力后，攻击面从"文本输出"扩展到"系统操作"，风险呈指数级增长。
延伸阅读
- OWASP Top 10 for LLM - LLM02 Sensitive Information Disclosure — https://genai.owasp.org/
- NIST SP 800-122: Guide to Protecting the Confidentiality of PII — https://csrc.nist.gov/publications/detail/sp/800-122/final
- 中国《个人信息保护法》(PIPL) — https://www.npc.gov.cn/npc/c30834/202108/a8c4e3672c74491a80b53a172bb753fe.shtml
- Microsoft Presidio - PII 检测与匿名化工具 — https://microsoft.github.io/presidio/
- GDPR Article 17 - Right to Erasure — https://gdpr-info.eu/art-17-gdpr/