做一个真的要上线的企业级大模型应用，最让我睡不踏实的从来不是模型选型，而是提示词里到底装了多少不该出现的数据。跑 Demo 的时候你不会觉得这条链路有什么问题，顶多就是在终端里看到几段包含姓名和手机号的对话日志。等真正进入内部客服、文档问答、人事助手这类场景，你会发现每个用户提问背后都挂着一堆姓名、证件号、银行卡信息。模型效果再惊艳，一条未经保护的敏感数据也可能让整个项目一夜回到解放前。

这篇项目复盘，写的是我实际搭过的一套数据保护链路：用 Google Cloud 的 DLP 服务做敏感数据探查和脱敏，用 Cloud KMS 管加密密钥，用 Secret Manager 管服务凭据，再用 LangChain 把检查、脱敏、调用模型、还原结果这四个动作编成一条标准的处理管道。这套方案主要面向正在做 RAG 问答、客服机器人、文档解析助手，又绕不开用户隐私数据的团队；如果你手上有类似场景，可以直接把这里的流程拿回去改。下面我先把整体思路讲清楚，再给关键代码，最后把踩过的坑列出来。

1. 先从需求和选型说起

1.1 大模型应用里的敏感数据到底从哪漏出去

很多人以为数据泄露只会发生在“把用户输入直接拼进 Prompt”这一种情况，实际开发中我发现至少有三个入口同时开着。

第一是用户输入本身。客服机器人接到的消息里常常直接带手机号、订单号、收件地址，这些内容一旦进入模型上下文，就等于把明文交给了外部推理服务。你无法控制模型服务端如何使用这些上下文，哪怕只是暂时的缓存，也可能构成泄露。

第二是 RAG 的文档库。给大模型建知识库的时候，大家通常会把企业内部文档一股脑切块、向量化。问题在于业务文档里嵌套了大量个人信息，比如客户合同里的身份证号、工资表里的员工姓名。文档一旦被切进向量库，检索阶段就可能把整段敏感内容召回并拼进 Prompt，这比用户直接提问更容易被忽视。

第三是对话日志和调试输出。LangChain 默认的 stdout 输出、开发阶段的日志打印、甚至错误堆栈，都可能把完整的原始 Prompt 带出来。生产环境里如果接了日志采集，等于把敏感数据又复制了一份。

所以我在做这个项目时定的基线是：敏感数据不允许以明文形式进入模型上下文，必须在调用链路的边界处做一次强制检查；同时所有日志、缓存、向量库一律只保存脱敏后的形态。

1.2 为什么选 Google Cloud 和 LangChain 这套组合

选 Google Cloud 不是因为它“名气大”，而是它的 DLP（数据防泄露）服务刚好解决了传统方案里最费人工的部分。自己写敏感数据识别，需要维护正则库、号码规则、命名实体词典，还要不断适配新的造假模式。DLP 内置了几百种 InfoType 探测器，从姓名、手机号、邮箱地址到信用卡号都在覆盖范围内，并且每个识别结果会附带一个置信度等级（likelihood），这让我可以按业务风险调节拦截尺度，而不是一刀切地全量打码。

Cloud KMS 负责的是另一件事：密钥管理。脱敏需要密钥，数据库加密需要密钥，向量库加密也需要密钥。KMS 直接把密钥生命周期托管起来，支持自动轮转和审计，比自己在配置中心里放一串环境变量要正规得多。Secret Manager 则用来存 DLP 服务账号、模型 API Key 这类敏感凭据，避免把密钥写进代码仓库。

LangChain 在这套方案里的角色是调度层。它的链式抽象允许我把 DLP 检查、脱敏、模型调用、还原这几个环节封装成一个个标准节点，然后用一套统一方式串联。后续要换底层模型，或者调整脱敏策略，只需要替换链路里的对应模块，不需要动整个业务代码。

1.3 这套方案适用的场景和影响范围

从实际落地来看，最容易见效的是下面几类应用：企业内部文档问答、客服工单自动回复、简历初筛与 HR 数据助手、金融领域的文档审阅。这些场景的共同点是既要借助大模型理解大量文本，又必须遵守严格的数据保护要求。

这套方案的影响范围不只是“给 Prompt 加一层过滤”，它会改变整个数据链路的设计方式。比如文档入库前需要跑一遍 DLP 预处理，检索阶段要按脱敏后的特征匹配，模型输出后还要做还原校验。团队的安全意识、运维脚本、监控告警都要跟着升级。换句话说，它把数据保护从“上线前补个洞”变成了“链路各处都在做防护”，这也是我觉得它值得写成一篇文章的原因。

2. 保护管道的整体设计

2.1 核心流程：检查、脱敏、调用、还原

整个系统可以简化成四个阶段。

检查阶段：对进入系统的每一段文本执行 DLP inspect，拿到所有命中的敏感片段、类型和位置。脱敏阶段：按检查结果把所有命中片段替换成固定占位符，比如把手机号替换成[PHONE_NUMBER]，同时把“占位符到原始值”的映射表保存在内存缓存里。调用阶段：用脱敏后的文本拼 Prompt、检索文档、调用模型，所有下游组件只看见占位符。还原阶段：模型输出里如果引用了占位符，再把映射表里的原始值替换回去，并做一轮二次校验，确保没有未授权的敏感片段进入最终响应。

这里有个容易被忽略的点：还原不是无条件的。如果用户本身没有权限查看某个敏感字段，即使占位符出现在模型输出里，也应该保持脱敏状态。所以我额外加了一个权限判断层，只有对应当前用户有权看到的字段才做还原。

2.2 敏感数据的分类与检测策略

DLP 把检测单元叫 InfoType，不同的数据类型对应不同的 InfoType。我在生产里把数据分成四类来看待。

对于表格里前两类，直接用 DLP 内置 InfoType 就能覆盖。企业内部的代号和工号这类信息，我采用自定义 InfoType 的方式，用一个正则模板告诉 DLP 怎么识别。自定义规则的好处是可以把“公司内部格式”这种外部工具完全识别不出来的数据也纳入保护范围，代价是需要自己维护规则模板，并定期根据新出现的编号格式更新。

2.3 脱敏策略选型：遮罩、替换还是标记化

DLP 的 deidentify 配置里提供了几种转换方式，我分别测试后留下了三种可用策略。

第一种是直接删除（redact），把命中片段整段移除。适合不需要保留任何信息的场景，比如日志脱敏。第二种是掩码显示（mask），只保留部分字符，其余用*代替，适合手机号、邮箱这类希望保留格式可读性的场景。第三种是替换成 InfoType 名称，比如把“张三”替换成[PERSON_NAME]，把“13800138000”替换成[PHONE_NUMBER]，这是我在模型调用链路里最常用的一种，因为占位符语义清晰，模型能理解这是一个被隐藏的字段。

严格来说还有第四种叫标记化（tokenization），用随机生成的票据代替原始值，原始值只存在加密存储里。这种方案安全性最高，但实现成本也最大，因为还原链路需要查询票据库。如果团队刚开始做数据保护，我建议先用替换成 InfoType 名称的方式跑通，后续再按需要升级到标记化。

3. 核心模块的实现细节

3.1 DLP 探查模块的接入方式

接入 DLP 做敏感数据探查，本质是调用一次 inspect_content 接口。每次调用传入一段文本，返回所有命中的 finding，包括类型、位置、原文引用和置信度。

调用之前最重要的准备工作是设置 Service Account 的最小权限。我给 DLP 探查模块建了独立的服务账号，只授予DLP User角色，没有给其他云资源的读写权限。这样即使密钥泄露，攻击者也无法顺着服务账号去访问存储桶或数据库。这一点在云上开发时经常被人忽视，很多人图省事直接用一个能操作所有资源的大权限账号，一旦泄露就是全盘风险。

DLP 调用本身是同步的。为了不阻塞主流程，我在实际项目中把它包成了一个异步任务，用户消息进队列后先做 DLP 检查再进 LangChain 链路，检查耗时对用户体验基本无感。

3.2 脱敏与还原映射的实现要点

脱敏不能只依赖 DLP 返回的 quote 字段直接替换，因为 quote 可能和原文不完全一致。我在实现时是严格按照 DLP 返回的 byte range 位置信息去做切割，代码里先把所有命中按起始位置排序，然后从尾部向前替换，避免前面的替换导致后面的偏移量全部失效。

这个细节我把绊倒过，刚开始没按位置切割，直接用字符串 replace 去替换原文，结果一个“张三”在正文里出现两次时，DLP 只返回了一个命中，replace 却把两处都换掉了，导致还原映射错乱。所以后来统一用位置切割，每个命中只替换一次。

映射表同时也需要有个过期机制。脱敏映射存在内存里，设置 TTL 为 15 分钟，用户一次会话结束就清除。原因很简单：占位符到原始值的映射本身就是敏感数据，长时间留着只会增加泄露面。如果业务确实需要更长的还原窗口，我会把映射表用 KMS 生成的密钥加密后存到 Redis，而不是裸存。

3.3 LangChain 管道改造的三个层次

LangChain 侧我按三个层次做改造，方便团队里不同水平的人理解和维护。

最底层是工具函数层，封装 DLP 检查和脱敏还原的调用，这部分只依赖 google-cloud-dlp 客户端，不依赖 LangChain 的 API。中间层是链式节点层，把工具函数包装成 LangChain 的 RunnableLambda，这样可以用 LCEL 语法把保护模块和模型调用串起来。最上层才是业务代码，业务开发不需要关心 DLP 怎么调用，只需要声明链路的顺序。

之所以要分三层，是因为 LangChain 自身 API 迭代很快，函数式封装如果跟业务代码混在一起，版本升级时会有大量地方需要跟着改。把依赖 LangChain API 的范围压缩到中间层，升级时的影响面就小很多。

4. 实操代码与部署要点

4.1 环境准备与项目初始化

我的项目使用 Python 3.11，依赖包版本如下：

google-cloud-dlp==3.15.0 google-cloud-kms==2.20.0 google-cloud-secret-manager==2.17.0 langchain==0.1.20 langchain-google-vertexai==1.0.0 python-dotenv==1.0.0

环境变量集中在.env文件里，用 python-dotenv 加载。项目初始化分四步：创建独立项目（或独立分区）、启用 DLP / KMS / Secret Manager 三个 API、创建最小权限服务账号、把服务账号密钥文件放到只有 CI 能访问的存储位置。

还有一个实践建议：在开发环境里不要直接设置GOOGLE_APPLICATION_CREDENTIALS指向本机密钥文件，而是用 gcloud 的临时凭据做本地调试，生产环境则用云上的服务账号。这样可以避免开发机上的密钥文件被误提交到代码仓库。

4.2 DLP 检查与脱敏的完整实现

检查模块的核心代码如下：

from google.cloud import dlp_v2 def inspect_sensitive_data(text: str, project_id: str) -> list: client = dlp_v2.DlpServiceClient() parent = f"projects/{project_id}/locations/global" response = client.inspect_content( request={ "parent": parent, "inspect_config": { "info_types": [ {"name": "PERSON_NAME"}, {"name": "PHONE_NUMBER"}, {"name": "EMAIL_ADDRESS"}, {"name": "CREDIT_CARD_NUMBER"}, {"name": "DATE_OF_BIRTH"}, {"name": "STREET_ADDRESS"}, ], "min_likelihood": "LIKELIHOOD_UNSPECIFIED", "limits": {"max_findings_per_item": 20}, }, "item": {"value": text}, } ) findings = [] for finding in response.result.findings: start = finding.location.byte_range.start end = finding.location.byte_range.end findings.append({ "info_type": finding.info_type.name, "start": start, "end": end, "likelihood": finding.likelihood.name, "quote": text[start:end], }) return findings

需要注意min_likelihood这个参数。我把默认值设成了LIKELIHOOD_UNSPECIFIED，意思是所有置信度都要返回，然后在业务层再按场景过滤。这样做的理由很直接：客服场景里手机号检测置信度很高，直接拦截没问题；但姓名检测在不带姓名的文本里偶尔会误报，如果接口层就过滤掉了，后面想查证就没机会了。宁可全量返回，也别在源头把数据丢了。

脱敏模块我用位置切割实现：

def deidentify_with_positions(text: str, findings: list) -> tuple[str, dict]: mask_map = {} # 按起始位置倒序替换，避免偏移量变化 findings.sort(key=lambda f: f["start"], reverse=True) for f in findings: left = text[: f["start"]] right = text[f["end"] :] placeholder = f"[{f['info_type']}]" mask_map[placeholder] = f["quote"] text = left + placeholder + right return text, mask_map

这段代码每次替换一个命中，并把原文和占位符的对应关系存进 map。倒序替换是关键的细节，谁先正序替换谁后悔。

4.3 把保护逻辑编入 LangChain 链

下面的代码展示了如何用 LCEL 把保护模块、模型调用、还原模块串成一条链：

from langchain_core.runnables import RunnableLambda from langchain_google_vertexai import VertexAI class DataProtectionGuard: def __init__(self, project_id: str): self.project_id = project_id self._mask_map = {} def mask(self, payload: str) -> str: findings = inspect_sensitive_data(payload, self.project_id) masked_text, mask_map = deidentify_with_positions(payload, findings) self._mask_map.update(mask_map) return masked_text def unmask_safe(self, payload: str, allowed_keys: set) -> str: # 只还原当前用户有权限看到的字段 for placeholder, original in self._mask_map.items(): if placeholder in payload and placeholder in allowed_keys: payload = payload.replace(placeholder, original) return payload guard = DataProtectionGuard(project_id="your-project-id") llm = VertexAI(model_name="gemini-pro") chain = ( RunnableLambda(guard.mask) | llm | RunnableLambda(lambda out: guard.unmask_safe(out, allowed_keys={"allowed"})) ) result = chain.invoke("我的手机号是13800138000，请帮我查询订单状态。")

这条链表面上看起来简单，实际做的时候有一个重要取舍：unmask_safe里的allowed_keys不能写死，必须来自当前会话的权限上下文。我自己一开始图简单，写了个“全部还原”的逻辑，结果测试模型输出里出现占位符时全部被还原，如果模型把占位符以外的敏感字段也趁机写出来，就等于绕过了脱敏防线。后来改成权限校验后，这个问题才真正堵住。

另外，模型输出的还原还有一个边界情况：如果模型没有原样保留占位符，比如把它改成了[PHONE]或者干脆忽略掉，映射就匹配不上。这种情况我的处理是：不强行还原，宁可让用户看到占位符，也不能凭模糊猜测去还原错误的值。

4.4 密钥管理和凭据管理

KMS 的使用场景主要是三处：DLP 映射表加密、向量库加密、以及需要落盘的日志加密。代码里最常用的是直接调用 encrypt 方法：

from google.cloud import kms_v1 def kms_encrypt(plaintext: bytes, key_name: str) -> bytes: client = kms_v1.KeyManagementServiceClient() response = client.encrypt( request={"name": key_name, "plaintext": plaintext} ) return response.ciphertext

Secret Manager 则用来读取模型 API Key 和服务账号凭据：

from google.cloud import secretmanager_v1 def read_secret(secret_id: str, project_id: str, version: str = "latest") -> str: client = secretmanager_v1.SecretManagerServiceClient() name = f"projects/{project_id}/secrets/{secret_id}/versions/{version}" response = client.access_secret_version(request={"name": name}) return response.payload.data.decode("UTF-8")

这里我想专门提醒一点：KMS 的密钥轮转并不难配置，难的是业务代码如何优雅配合轮转。如果直接在代码里写死密钥 ID，轮转一次就要重新发布一次。我后来的做法是密钥别名机制，业务代码只认别名，KMS 侧把别名指向当前生效的密钥，轮转时业务无感。这属于上线时才摸到痛点的经验。

5. 常见问题与排查实录

5.1 DLP 误报和漏报怎么缓解

交付测试阶段最常被问到的就是“这个字段为什么不识别”和“这一段怎么被打码了”。

先说不识别的问题。内置 InfoType 对常规数据效果很好，但遇到公司自定义格式就无能为力。解决办法是加自定义 InfoType，用正则描述业务编号的格式，再结合一个词典文件提升识别率。比如员工工号的格式是EMP加八位数字，正则写好后测试两三周没误报，就可以正式上线。

再说误报打码的问题。有一次用户反馈正文里的“小王”被识别成 PERSON_NAME 并打码，导致模型理解出错。排查后确认是 likelihood 阈值设得太低。建议对模糊度高的类型单独设置阈值，而不是全量用一个标准。生产环境里我的配置是：手机号、银行卡这类高置信度类型用 POSSIBLE 就拦截，姓名、地址这类容易混淆的类型要 LIKELY 以上才拦截。

实际排查时还发现一个常见坑：DLP 检查对整段长文本的耗时和费用会随长度上升。生产里我给单次检查的文本长度设了上限，超过上限先截断再检查，并在截断前先做一次简单关键字预检，命中可疑关键字才走完整 DLP。这样能把 90% 以上的请求挡在开销大的检查之外。

5.2 模型输出还原时的几个“翻车”现场

还原这个环节看似简单，实际翻车最多。

第一个翻车点是模型输出里占位符被改写。模型可能把[PHONE_NUMBER]改成[手机号]或(PHONE)，映射表完全匹配不上。我的对策是两阶段还原：第一次按精确占位符匹配，匹配不到再用模糊匹配规则匹配相似写法，两次都不匹配就直接返回占位符。

第二个翻车点是占位符在输入中出现多次。比如用户说“我手机号和同事手机号一样”，DLP 会返回两个位置，但映射表用 InfoType 作为 key 时会互相覆盖。我后来把 key 改成了“位置信息加类型”的组合字符串，比如[PHONE_NUMBER_0]和[PHONE_NUMBER_1]，虽然可读性差一些，但不会丢数据。

第三个翻车点是 long context 提前截断。模型上下文过长时，后面的还原提示可能被丢掉。这属于上下文管理和还原逻辑的协同问题，我的建议是：还原逻辑不要依赖模型的总结能力，而是依赖结构化输出解析，让模型在特定 JSON 字段里返回占位符引用，解析失败就走降级路径。

5.3 性能、成本与运行可靠性

DLP 的每次调用都有网络开销。实测下来，单次 inspect_content 对一个 200 字以内的用户消息大约增加 80 到 150 毫秒延迟，对几十KB的长文本会到 1 秒上下。在客服场景里这个延迟是可以接受的，但要对数据库和向量检索场景单独做压测，不能直接套用。

成本方面，DLP 按处理的字节数计费。如果每天有十万条消息，每条约 300 字节，成本并不高；但如果文档入库时全量跑 DLP，可能一次导入就把月度预算用掉。所以我的建议是文档入库阶段只对新增文档做 DLP 检查，存量文档分批处理并加上去重逻辑，别每次全量跑。

运行可靠性上，我给 DLP 调用加了三重保护：超时设置为 3 秒、失败自动降级（跳过脱敏但返回错误提示）、以及针对周期性失败（比如限流错误）的退避重试。大模型链路可以容忍偶尔返回失败，但绝对不能容忍“失败时静默放行明文”，这两种策略要分清楚。

6. 个人实操体会

把整套东西跑通之后，我再看最初设想的“接一个 DLP 库就行”的想法，觉得确实天真了。真正值钱的不是那几次 API 调用，而是你愿意在多少细节上跟安全较劲。

最后分享一个我体会最深的小点：脱敏后的内容也要做缓存设计。早期我直接在链路里硬编码了“先脱敏再缓存”的逻辑，后来发现多个服务都映射表放到缓存里，一旦服务重启，缓存里残留的映射会和新进来的请求冲突。后来我把缓存键设计成“请求 ID 加用户 ID”，每次请求独立读取，这个问题才彻底消失。可能听起来是个很小的细节，但在真实生产环境里，这类边界情况往往比核心功能更容易让你半夜被电话叫起来。