Prompt 敏感信息脱敏：在发送给 LLM 前自动识别并替换 PII 和密钥

Prompt 敏感信息脱敏在发送给 LLM 前自动识别并替换 PII 和密钥一、深度引言与场景痛点大家好我是赵咕咕。几个月前我们做了一次日志审计发现一件让我后背发凉的事有用户在我们的 RAG 系统中粘贴了一段代码里面包含了一个硬编码的 AWS Access Key。那段代码被当作检索上下文的一部分原封不动地拼进了 Prompt然后发给了 GPT-4。更可怕的是我们根本不知道这件事——直到日志审计才发现。而 OpenAI 的 API 日志里这个密钥已经被保存了。这不是某个马虎的开发者的锅。大模型应用的本质就是把用户的任意输入拼成 Prompt 发给第三方 API。在这个过程中用户可能不小心贴进来任何东西手机号、身份证号、API Key、数据库密码、内部文档的敏感段落。你必须假设发给 LLM 的每一段文本都可能包含敏感信息然后在发送之前做脱敏处理。二、底层机制与原理深度剖析2.1 敏感信息的分类在 Prompt 场景下敏感信息可以分为三类个人身份信息PII手机号、身份证号、邮箱、银行卡号、家庭住址。识别难度低有明确的正则模式。凭证类信息API Key、Access Token、数据库密码、SSH 私钥。这些通常有特定的格式前缀如sk-、AKIA、ghp_而且有高熵特征。业务敏感信息公司内部项目代号、未发布的版本号、客户真实姓名。这类信息无法通过规则识别需要靠自定义字典或 NERNamed Entity Recognition模型。2.2 脱敏流水线的架构脱敏流水线采用串行与并行结合的处理方式整体流程从用户输入与检索文档的合并开始经过规则引擎筛选后依次完成检测、替换与还原三个阶段最终将结果返回给用户。整个流水线的核心逻辑是检测用多引擎并行检测文本中的敏感信息。规则引擎负责模式匹配正则高熵检测负责发现看起来不像正常文本的字符串如随机生成的 Token。替换检测到的敏感信息用占位符替换如[PHONE_01]、[APIKEY_01]。保留原始值到映射表中。还原LLM 返回结果后把占位符还原为原始值。这样用户看到的结果跟实际数据一致但 LLM 和 API 日志里只有占位符。关键是第三步还原必须在最终输出给用户之前完成否则用户看到的答案是您的电话是 [PHONE_01]而不是真实的138xxxx1234。2.3 高熵检测的原理有经验的工程师可能会绕过正则把 API Key 拆成两段、中间加空格、用 Base64 编码后再贴进来。正则匹配不到。这时候需要用信息熵来检测。正常英文文本的香农熵通常在 3.5-4.5 之间而随机生成的 Token如sk-proj-abc123def456...的熵通常超过 4.8。算法很简单对每个连续的非空白字符串计算字符频率分布的熵值。超过阈值的标记为疑似凭证。三、生产级代码实现import asyncio import re import math import logging from collections import Counter from dataclasses import dataclass, field from typing import Any logger logging.getLogger(__name__) dataclass class SensitiveMatch: 检测到的敏感信息。 original: str # 原始值 placeholder: str # 占位符 type: str # phone / email / apikey / pii / high_entropy start: int # 在原文本中的起始位置 end: int # 在原文本中的结束位置 class PromptSanitizer: Prompt 敏感信息脱敏器。 # ---- PII 正则模式 ---- _PII_PATTERNS: list[tuple[str, str]] [ (phone, r\b1[3-9]\d{9}\b), # 中国手机号 (idcard, r\b[1-9]\d{5}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dXx]\b), # 身份证 (email, r\b[A-Za-z0-9._%-][A-Za-z0-9.-]\.[A-Za-z]{2,}\b), # 邮箱 (bankcard, r\b\d{16,19}\b), # 银行卡号 (ip_addr, r\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b), # IP 地址 ] # ---- 凭证前缀模式 ---- _CREDENTIAL_PREFIXES: list[tuple[str, str]] [ (openai_key, rsk-[A-Za-z0-9]{20,}), # OpenAI (aws_key, rAKIA[A-Z0-9]{16}), # AWS Access Key (github_token, rgh[ps]_[A-Za-z0-9]{36,}), # GitHub Token (jwt, reyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}), # JWT (private_key, r-----BEGIN [A-Z ] PRIVATE KEY-----), # SSH/TLS 私钥 (connection_string, r(?:mongodb|mysql|postgresql|redis)://[^\s]), ] def __init__( self, entropy_threshold: float 4.5, custom_patterns: list[tuple[str, str]] | None None, custom_keywords: set[str] | None None, ): self._entropy_threshold entropy_threshold self._custom_keywords custom_keywords or set() # 编译所有正则 self._compiled [] for name, pattern in self._PII_PATTERNS self._CREDENTIAL_PREFIXES: try: self._compiled.append((name, re.compile(pattern, re.IGNORECASE))) except re.error as e: logger.error(正则编译失败 (%s): %s, name, e) if custom_patterns: for name, pattern in custom_patterns: try: self._compiled.append( (fcustom_{name}, re.compile(pattern, re.IGNORECASE)) ) except re.error as e: logger.error(自定义正则编译失败 (%s): %s, name, e) # ─── 公开接口 ─── async def sanitize( self, text: str, timeout: float 5.0 ) - tuple[str, dict[str, str]]: 对文本做脱敏处理。 返回: (脱敏后的文本, 占位符→原始值的映射) try: return await asyncio.wait_for( self._sanitize_impl(text), timeouttimeout ) except asyncio.TimeoutError: logger.error(脱敏处理超时返回原文未脱敏) return text, {} async def desanitize( self, text: str, mapping: dict[str, str] ) - str: 将占位符还原为原始值。 result text for placeholder, original in mapping.items(): result result.replace(placeholder, original) return result async def _sanitize_impl( self, text: str ) - tuple[str, dict[str, str]]: matches: list[SensitiveMatch] [] # 1) 正则匹配 for name, pattern in self._compiled: for m in pattern.finditer(text): matches.append(SensitiveMatch( originalm.group(), placeholder, typename, startm.start(), endm.end(), )) # 2) 高熵检测 high_entropy_matches self._detect_high_entropy(text) matches.extend(high_entropy_matches) # 3) 自定义关键词检测 keyword_matches self._detect_keywords(text) matches.extend(keyword_matches) if not matches: return text, {} # 4) 去重叠较长的匹配优先 matches.sort(keylambda m: (m.start, -(m.end - m.start))) deduped: list[SensitiveMatch] [] for m in matches: if not any( e.start m.start and e.end m.end for e in deduped ): deduped.append(m) # 5) 生成占位符并替换 mapping: dict[str, str] {} counts: dict[str, int] {} # 从后往前替换避免位置偏移 result_chars list(text) for m in sorted(deduped, keylambda x: x.start, reverseTrue): type_key m.type.upper() idx counts.get(type_key, 0) 1 counts[type_key] idx placeholder f[{type_key}_{idx:02d}] m.placeholder placeholder mapping[placeholder] m.original result_chars[m.start:m.end] list(placeholder) result .join(result_chars) logger.info( 脱敏完成: 发现 %d 处敏感信息 (%d 种类型), len(deduped), len(counts), ) return result, mapping # ─── 高熵检测 ─── def _detect_high_entropy(self, text: str) - list[SensitiveMatch]: 检测高熵字符串疑似随机生成的凭证。 results: list[SensitiveMatch] [] # 找出所有连续的非空白词元至少 10 个字符 for m in re.finditer(r\S{10,}, text): token m.group() entropy self._shannon_entropy(token) if entropy self._entropy_threshold: # 额外检查不匹配已知的普通词语 if not self._looks_like_normal_text(token): results.append(SensitiveMatch( originaltoken, placeholder, typehigh_entropy, startm.start(), endm.end(), )) return results staticmethod def _shannon_entropy(s: str) - float: 计算字符串的香农熵。 if not s: return 0.0 counter Counter(s) length len(s) return -sum( (count / length) * math.log2(count / length) for count in counter.values() ) staticmethod def _looks_like_normal_text(token: str) - bool: 判断是否像正常文本排除已知的模式。 # 如果包含常见英文单词模式元音辅音交替很可能是正常文本 word_pattern re.compile(r[aeiou][bcdfghjklmnpqrstvwxyz], re.IGNORECASE) vowel_ratio sum(1 for c in token.lower() if c in aeiou) / max(len(token), 1) # 正常英文单词元音比例通常在 0.2-0.5 if 0.25 vowel_ratio 0.55 and word_pattern.search(token): return True # URL 不是高熵凭证 if token.startswith((http://, https://)): return True return False # ─── 关键词检测 ─── def _detect_keywords(self, text: str) - list[SensitiveMatch]: 检测自定义敏感关键词。 if not self._custom_keywords: return [] results: list[SensitiveMatch] [] for kw in self._custom_keywords: start 0 while True: idx text.find(kw, start) if idx -1: break results.append(SensitiveMatch( originalkw, placeholder, typecustom_keyword, startidx, endidx len(kw), )) start idx 1 return results # ─── Agent 集成中间件 ─── class SanitizerMiddleware: 在 Agent 工具调用前插入脱敏处理。 def __init__(self, sanitizer: PromptSanitizer): self._sanitizer sanitizer self._last_mapping: dict[str, str] {} async def wrap_prompt(self, raw_prompt: str) - str: 在构造 Prompt 时调用返回脱敏后的文本。 clean, mapping await self._sanitizer.sanitize(raw_prompt) self._last_mapping mapping return clean async def unwrap_response(self, raw_response: str) - str: 在返回给用户前调用还原占位符。 return await self._sanitizer.desanitize( raw_response, self._last_mapping ) async def main(): sanitizer PromptSanitizer( entropy_threshold4.5, custom_keywords{ProjectPhoenix, 竞品分析_V2}, ) middleware SanitizerMiddleware(sanitizer) # 模拟用户输入 user_input ( 我的手机号是 13812345678请帮我把这个 API Key: sk-proj-abc123def456ghi789jkl 相关的项目部署到 AWS 上。 另外可以参考一下 ProjectPhoenix 的架构。 ) clean await middleware.wrap_prompt(user_input) print(f脱敏后: {clean}) # 模拟 LLM 返回 llm_response ( 已根据您的 [PHONE_01] 设置了通知方式。 使用 [OPENAI_KEY_01] 完成了部署。 参考了 [CUSTOM_KEYWORD_01] 的架构设计。 API Key: [OPENAI_KEY_01] ) final await middleware.unwrap_response(llm_response) print(f还原后: {final}) if __name__ __main__: asyncio.run(main())设计的几个关键考量从后往前替换如果从前往后替换每次替换后字符位置会偏移后续的start/end就对不上了。从后往前替换不需要重新计算位置。去重叠手机号可能同时匹配phone和bankcard如果手机号恰好 16 位较长的匹配优先级更高。高熵检测的假阳性过滤_looks_like_normal_text通过元音比例和辅音模式过滤掉正常的英文文本避免误杀。超时兜底sanitize外层包了asyncio.wait_for。脱敏不是核心业务路径超时后降级为不脱敏直接发送好过阻塞整个请求。中间件模式SanitizerMiddleware封装了脱敏→LLM→还原的完整流程对 Agent 的其他代码透明。四、边界分析与架构权衡4.1 脱敏的过度与不足正则匹配天然偏向宁可误杀不可漏过——银行卡号\b\d{16,19}\b会匹配任何 16-19 位纯数字包括一些正常的订单号或流水号。如果你的业务中订单号是敏感信息那误杀反而是好事。但如果需要保留某些长数字就需要在正则前加一层白名单_WHITELIST_PATTERNS [ r\b(?:ORDER|TXN|REF)\d\b, # 订单号、交易号等非敏感长数字 ]在白名单中的匹配不进行替换。4.2 LLM 对占位符的理解一个容易忽略的问题是LLM 看到[PHONE_01]后可能会猜测这是什么然后在回答中尝试还原。比如它会写您的电话号码 [PHONE_01] 对应的运营商是……——这种情况下LLM 实际上猜对了这个占位符的含义但它的猜测过程本身可能泄露了上下文信息。缓解方案使用更语义中性的占位符如[S_01]、[P_03]而不是[PHONE_01]。4.3 性能与延迟脱敏处理的延迟取决于文本长度和正则复杂度。对于典型 2000 字的 Prompt规则引擎处理时间在 1-5ms。高熵检测需要对每个连续词元做计算在 5000 字文本中约 3-10ms。总体延迟可控制在 15ms 以内。如果文本特别长如检索到 10 万字的上下文建议先做文本截断或只在可能包含敏感信息的段落上运行脱敏如代码块、用户输入部分。4.4 日志与监控脱敏后的文本仍然可能泄露模式信息如用户问了关于 [PHONE_01] 的问题。如果日志也被发送到第三方如日志服务需要考虑日志层面的脱敏。建议在日志输出前再做一层脱敏——但这会引入额外复杂度。务实做法是脱敏后的 Prompt 直接入日志只对高风险的日志接收方做额外处理。五、总结Prompt 脱敏不是可选的nice to have而是一个大模型应用的基础安全设施。你无法控制用户往对话框里贴什么但你可以控制什么内容被发送到第三方 LLM API。实现上建议从三层渐进式构建第一层必须正则匹配 PII 和常见凭证格式覆盖 90% 的场景。第二层推荐高熵检测补充规则引擎的盲区拦截非标准格式的凭证。第三层按需自定义关键词和 NER 模型覆盖业务特定的敏感信息。最关键的一点还原必须在服务端。永远不要在前端做脱敏——你是在防用户把敏感信息发给 LLM而不是帮用户保护他自己的隐私虽然客观上也有这个效果。安全领域有一条铁律不要信任用户的任何输入。在大模型时代这条铁律要加上后半句也不要让 LLM 看到用户的任何裸奔输入。下一篇预告RAG 服务的内存池设计复用 embedding 模型推理结果来减少重复计算。