如果你今天下午要把一段生产环境报错日志复制出来,打开某个大模型聊天窗口,直接粘贴发送,大概率不会有人拦你。但这段日志里如果带着内网 IP、客户邮箱、一个写死在代码里的数据库连接串,你实际上已经把这些信息当作 prompt 的一部分,发到了本地设备之外的模型服务里。
大模型不会主动“记住”你的密钥然后第二天公布出来,问题的关键在于你无法控制转发链路,也无法审计模型是否把它用在了输出中。真正可怕的是这个过程太顺手了——复制、粘贴、回车,三步完成,你会渐渐忽略中间这层需要被检查的文本。
这也是“本地 scrubber”这类工具的价值:在文本离开你的设备之前,先过一道清洗流水线,把 Token、密钥、手机号、内网 IP 这类敏感标识替换掉,把时间戳、心跳日志、重复堆栈这类低价值噪音剥离掉,最后把压缩后的干净文本发送给 LLM。这篇文章会从一个 API 开发者的视角,拆解这类工具应该具备的能力、核心设计、最小 Python 实现,以及几套真实场景的配置和验证手段。代码可以直接复制到自己的项目里,规则按需修改即可。
1. 为什么你需要一个“发给 LLM 前的清洗层”
很多人对 LLM 的使用方式还停留在“复制粘贴聊天窗口”。“反正模型只是读一下,不会拿我的数据怎样”,这种想法在个人实验时勉强成立,一旦进入工作流、API 调用、团队共享,问题就会被放大。
1.1 隐私风险:发送即离开设备
第一个问题最直接:文本一旦发送给远程模型服务,它就不再由你完全控制。
代码里的硬编码密码、日志里的内网 IP、文档里的客户邮箱和手机号,这些字段出现在 prompt 中时,模型会像读取普通文本一样读取它们。如果有另一个线程在后台记录、分析、评估你的输入,你没有任何手段阻止。更常见的风险其实不是模型“泄露”,而是你粘贴出去的文本会被复制到公司的数仓、第三方的日志系统、或者某个未知的审核系统里。
一句话总结:发给 LLM 的文本,不能默认是无害的,先假设它可能去到任何地方,再决定哪些内容可以出现。
1.2 成本问题:Token 是按输入算的
调用大模型 API 时,输入部分按 Token 计费。你贴进去一万行日志,其中九千行是心跳、INFO、重复堆栈。这些内容模型不读吧,它又确实占用了上下文窗口;读了,又对诊断问题毫无帮助。
按业内常用的粗略估算,1 个 Token 大约等于 3.5 到 4 个英文字符。一段 10000 字符的日志,大约会产生 2500 个 Token。如果一次调用是 5 万字符,那就是 1 万多 Token。费用因模型而异,但没必要把钱花在噪音上。
1.3 质量问题:注意力会被噪音稀释
Transformer 模型的核心是自注意力机制,它会给每个 Token 分配不同的权重。问题是,模型并不会自动识别出“哪一行日志是噪音,哪一行才是错误根因”。它会很忠实地处理你给它的所有文本,然后把大量注意力浪费在无关内容上。
举个例子,你贴了 45 行日志,里面只有 3 行是真正的 ERROR。模型看完之后,很可能会给出一个泛泛的回答:“看起来像是数据库连接超时,建议检查网络配置。”它不知道真正的问题其实出现在第 41 行的某个超时重试循环里。因为那行信息被埋在 42 行噪音中。
三者放在一起,就构成了一个明确的判断:发送前的本地预处理,不是锦上添花,而是值得加入工作流的一层。
| 场景 | 直接发送的痛点 | 清洗后的效果 |
|---|---|---|
| 代码片段 | 硬编码密码、Token 外泄 | 敏感字段替换为占位符 |
| 应用日志 | 90% 是噪音,关键报错被稀释 | 只保留 WARN/ERROR 和上下文行 |
| 业务文档 | 客户邮箱、手机号、项目代号外泄 | 个人信息脱敏,正文结构保留 |
| API 调用 | Token 消耗高,频繁触发长度限制 | 输入裁剪到必要范围 |
2. 核心概念:Scrubber 到底是什么
“Scrubber”在英文里是刷子、清洗剂的意思。在 LLM 工作流里,它指的是一个本地运行的文本预处理组件,统一负责在文本发给模型之前执行脱敏、过滤和压缩操作。
它和我们常说的几个概念有区别:
| 概念 | 侧重点 | 典型操作 |
|---|---|---|
| Sanitizer | 清洗输入/输出中的恶意或有害内容 | 去脚本、去注入、去广告 |
| Normalizer | 格式归一化 | 统一换行、统一编码、统一日期格式 |
| Chunker | 文本切片,适配上下文窗口 | 按长度、段落、语义切分 |
| Scrubber | 隐私脱敏 + 噪音过滤 + 内容压缩 | 掩码、删行、去重、截断 |
实际项目中,Scrubber 往往会内嵌 Normalizer 和 Chunker 的部分能力,但从职责上,它更强调“隐私与噪声控制”。
2.1 Scrubber 的三种核心动作
- Mask(替换):把敏感字段替换成占位符,比如
api_key=sk-xxxxxx替换成api_key=<REDACTED>。替换保留了文本的结构,模型还能“看懂”这里原本是一个配置项。 - Filter(删除):直接删除整行或整段,比如 INFO 日志、心跳包、空行、重复堆栈。删除能最大限度降低 token 数量,但要注意别把上下文删没了。
- Compress(压缩):对保留下来的内容做去重、截断、摘要,或者结构化整理。压缩的目的是让模型把注意力集中在真正相关的信息上。
新手最容易误解的一点是:Scrubber 不是 Prompt 工程的替代品,而是前置处理层。Prompt 工程解决“怎么问更好”,Scrubber 解决“什么是能问的、什么是不该出现的”。
2.2 为什么 LLM 需要“干净”的文本
从注意力机制的角度看,LLM 对 prompt 中每个 token 都会计算相关性,但注意力是有限的资源。冗余 token 占比越高,关键信息获得的相对权重就越低。而且随着上下文变长,模型对远距离信息的依赖会衰减,你放在日志尾巴上的那行真正的 ERROR,很可能在注意力层面已经被前面的 1000 行 INFO 淹没了。
所以,清洗不是强迫症,而是为了让模型把注意力花在有价值的信息上。
3. 本地 Scrubber 的整体架构
在设计上,这类工具遵循一个清晰的分层原则:输入源 → 规则加载 → 脱敏 → 过滤 → 压缩 → 统计分析 → 输出。
最简单的架构可以用下面的数据流描述:
- 从文件、标准输入或剪贴板读取原始文本。
- 加载脱敏规则和过滤规则。规则可以内置,也可以放在 YAML/JSON 配置文件里。
- 执行正则脱敏,把敏感字段替换成占位符,并记录替换次数。
- 执行噪音过滤,逐行判断是否应该保留。
- 执行压缩,对重复行去重、按最大行数截断。
- 输出清洗结果,同时输出一份统计报告,方便你审计“到底替换了什么”。
- 如果不满意,可以调整规则并重新执行。
有一个设计要点值得强调:统计报告非常重要。一个黑盒的清洗过程,会让用户不放心。你替换了哪些字段、删除了多少行、压缩掉了多少重复内容,这些都需要显式输出,才能在出问题时快速定位规则误伤。
推荐一个最小的 Python 项目结构:
llm-scrubber/ ├── config/ │ └── rules.yaml # 脱敏与过滤规则配置 ├── src/ │ ├── scrubber.py # 核心脱敏模块 │ ├── filters.py # 噪音过滤模块 │ ├── compressors.py # 压缩与去重模块 │ └── cli.py # 命令行入口 ├── tests/ │ └── test_scrubber.py # 自动化测试 ├── requirements.txt └── README.md4. 实现一个可用的本地 Scrubbing 流程
下面这段代码是一个最小骨架,全部使用 Python 标准库,不依赖第三方框架。你可以直接复制保存,按自己的场景扩展。
4.1 核心脱敏模块src/scrubber.py
脱敏规则用“正则表达式 + 替换模板”来描述。建议把规则从代码中拆出来,但这篇文章为了演示方便,先用一个内置规则列表。
# 文件路径:src/scrubber.py import re from typing import Dict, List, Tuple class ScrubRule: def __init__(self, name: str, pattern: str, replacement: str): self.name = name self.pattern = pattern self.replacement = replacement class Scrubber: def __init__(self, rules: List[ScrubRule]): self._compiled = [ (rule.name, re.compile(rule.pattern), rule.replacement) for rule in rules ] def mask(self, text: str) -> Tuple[str, Dict[str, int]]: stats = {} for name, pattern, replacement in self._compiled: text, count = pattern.subn(replacement, text) if count: stats[name] = stats.get(name, 0) + count return text, stats DEFAULT_RULES = [ ScrubRule("api_key", r"(?i)\b(api[_-]?key|secret|password|token)\s*[=:]\s*[\w\-]{8,}", r"\1=<REDACTED>"), ScrubRule("aws_key", r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b", "<AWS_KEY>"), ScrubRule("bearer_token", r"(?i)\bbearer\s+[\w\-\.]{20,}", "Bearer <TOKEN>"), ScrubRule("ip_address", r"\b(?:\d{1,3}\.){3}\d{1,3}\b", "<IP>"), ScrubRule("email", r"\b[\w\.\-]+@[\w\-]+(?:\.[\w\-]+)+\b", "<EMAIL>"), ScrubRule("phone", r"\b(?:1[3-9]\d{9}|(?:\+?86[- ]?)?1[3-9]\d{9})\b", "<PHONE>"), ]关键逻辑说明:
ScrubRule是一个简单的规则数据类,包含名称、正则、替换模板。Scrubber.mask()按顺序执行所有已编译的正则,并通过subn返回替换次数。- 正则的边界控制很关键。比如
\b边界可以避免把abc1.2.3def误判成 IP;(?i)可以让 Token、token、TOKEN 统一匹配。
注意:ip_address规则可能会误伤无符号版本号,比如1.2.3.4。实际使用中需要配合上下文判断,或者用更严格的 CIDR 范围匹配。
4.2 噪音过滤模块src/filters.py
日志场景中最常见的就是把 INFO 级别、空行、时间戳开头、心跳包直接过滤掉。
# 文件路径:src/filters.py import re from typing import Dict, List, Tuple NOISE_PATTERNS = [ r"^\s*$", r"^\s*(INFO|DEBUG)\s+.*$", r"^\s*\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.*$", r"^\s*heartbeat\s*.*$", ] def filter_noise(text: str, patterns: List[str] = None) -> Tuple[str, Dict[str, int]]: patterns = patterns or NOISE_PATTERNS lines = text.splitlines() keep = [] removed = 0 for line in lines: if any(re.match(p, line, re.IGNORECASE) for p in patterns): removed += 1 continue keep.append(line) return "\n".join(keep), {"removed_lines": removed}这里要特别注意:INFO正则可能会把包含INFO字段的自定义日志也删掉。如果你有vendor=INFO这样的内容,需要单独调整规则。更稳妥的做法是,把过滤规则改成“白名单 + 关键字”模式,而不是一刀切。
4.3 压缩去重模块src/compressors.py
日志和堆栈中经常出现完全相同的行,比如同一行报错重复几千次。用“连续行去重”比“全量去重”更安全,因为全量去重会打乱行之间的顺序,反而会让模型更难理解事件发生的时间线。
# 文件路径:src/compressors.py from typing import Tuple, Dict def dedup_consecutive(text: str, max_lines: int = 200) -> Tuple[str, Dict[str, int]]: lines = text.splitlines() out = [] prev = None removed = 0 for line in lines: if line != prev: out.append(line) else: removed += 1 if len(out) >= max_lines: break prev = line return "\n".join(out), {"removed_duplicated_lines": removed}4.4 命令行入口src/cli.py
命令行入口的目的是让工具可以接入日常工作流,既可以从标准输入读取文本,也可以读写文件。
# 文件路径:src/cli.py import argparse import json import sys from scrubber import DEFAULT_RULES, Scrubber from filters import filter_noise from compressors import dedup_consecutive def main(): parser = argparse.ArgumentParser(description="Local text scrubber for LLM input") parser.add_argument("--input", "-i", help="input file path, default stdin") parser.add_argument("--output", "-o", help="output file path, default stdout") parser.add_argument("--mode", choices=["code", "log", "doc"], default="log") parser.add_argument("--stats", action="store_true", help="print replacement stats") parser.add_argument("--max-lines", type=int, default=200) args = parser.parse_args() text = "" if args.input: with open(args.input, encoding="utf-8") as f: text = f.read() else: text = sys.stdin.read() # 1. 脱敏 scrubber = Scrubber(DEFAULT_RULES) text, scrub_stats = scrubber.mask(text) # 2. 日志模式下才做噪音过滤 filter_stats = {"removed_lines": 0} if args.mode == "log": text, filter_stats = filter_noise(text) # 3. 压缩 text, compress_stats = dedup_consecutive(text, max_lines=args.max_lines) # 4. 输出 if args.output: with open(args.output, "w", encoding="utf-8") as f: f.write(text) else: sys.stdout.write(text) if args.stats: stats = { "scrub": scrub_stats, "filter": filter_stats, "compress": compress_stats, } print(f"\n---- scrub stats ----\n{json.dumps(stats, ensure_ascii=False, indent=2)}", file=sys.stderr) if __name__ == "__main__": main()4.5 运行方式
在项目根目录执行以下命令:
# 从标准输入读取,以日志模式清洗并输出统计 cat app.log | python src/cli.py --mode log --stats > app.clean.log # 从文件读取,以代码模式清洗 python src/cli.py -i sample.py -o sample.clean.py --mode code --stats运行结果会显示类似下面的统计信息:
---- scrub stats ---- { "scrub": { "ip_address": 3, "email": 1 }, "filter": { "removed_lines": 27 }, "compress": { "removed_duplicated_lines": 8 } }通过这份报告,你能明确知道:清洗了 3 个 IP、1 个邮箱,删除了 27 行噪音,去重 8 行重复堆栈。这种透明性对生产环境非常重要,因为规则误伤可以立刻被肉眼发现。
5. 三套典型场景下的配置实战
只有核心代码还不够,真正决定工具好不好用的是规则集。下面给出三套最典型的场景配置思路。
5.1 场景 A:代码片段
在把代码粘贴给 LLM 前,保留语法结构,但要脱敏掉硬编码的密钥、连接串、内网地址。
假设原始代码如下:
DB_PASSWORD = "P@ssw0rd2024" SERVICE_URL = "http://10.0.0.8:8080/api/v1/orders"清洗后应变成:
DB_PASSWORD = "<REDACTED>" SERVICE_URL = "http://<IP>:8080/api/v1/orders"注意,这里只替换了敏感字段,字符串引号和代码结构都保留了下来。模型依然能理解这是一个服务 URL 常量,只是 IP 变成了占位符。这正是 Mask 的价值:保留可读性,但不暴露真实值。
代码模式建议关闭filter_noise,因为代码中的空行、注释、INFO 字符串可能是语法结构的一部分。命令:
python src/cli.py -i code_sample.py -o code_sample.clean.py --mode code --stats5.2 场景 B:应用日志
日志是噪声最严重的场景。通常,一个问题现场包含大量 INFO、DEBUG、健康检查、心跳。保留 WARN、ERROR 以及与之相邻的上下文行,是最高效的策略。
原始日志片段:
2024-05-01T10:00:00 INFO started 2024-05-01T10:00:01 INFO heartbeat ok 2024-05-01T10:00:02 ERROR connect to db timeout 2024-05-01T10:00:03 INFO retry清洗后:
2024-05-01T10:00:02 ERROR connect to db timeout这里值得提醒:如果只保留 ERROR,模型可能缺少前因后果。更稳妥的方案是“保留错误行及其前后 N 行”。你可以扩展filter_noise,让它输出错误行的同时,把错误行附近的 INFO 行也保留下来,作为上下文。
5.3 场景 C:业务文档
文档类场景里,结构格式和脱敏同样重要。Markdown 表格、标题结构都应该完整保留,否则模型会认为格式混乱,误以为表格数据无效。
原始文档片段:
| 申请人 | 邮箱 | | --- | --- | | 张三 | zhangsan@example.com | | 李四 | lisi@example.com |清洗后:
| 申请人 | 邮箱 | | --- | --- | | 张三 | <EMAIL> | | 李四 | <EMAIL> |姓名要不要脱敏,取决于你的合规要求。如果只需要处理个人信息,可以再加一条“中文姓名脱敏”规则。关键点是:清洗过程不要破坏 Markdown 的竖线、横线、单元格分隔符,否则模型解析表格时会出现幻觉。
6. 怎么验证清洗效果
清洗完后不能直接发给 LLM,需要先验证效果。验证主要从三个维度展开:脱敏是否彻底、Token 是否真的降低、模型输出的质量是否提升。
6.1 自动化测试:防止规则回归
建议至少覆盖三类用例:脱敏、过滤、去重。
# 文件路径:tests/test_scrubber.py from scrubber import DEFAULT_RULES, Scrubber from filters import filter_noise from compressors import dedup_consecutive def test_mask_api_key(): scrubber = Scrubber(DEFAULT_RULES) text = "api_key = sk-abc123def456" cleaned, stats = scrubber.mask(text) assert "sk-abc123def456" not in cleaned assert "<REDACTED>" in cleaned assert stats.get("api_key") == 1 def test_filter_removes_info(): text = "2024-05-01T10:00:00 INFO start\n2024-05-01T10:00:01 ERROR boom" cleaned, stats = filter_noise(text) assert "INFO start" not in cleaned assert "boom" in cleaned def test_consecutive_dedup(): text = "a\nb\nb\nc\nc\nc" cleaned, _ = dedup_consecutive(text) assert cleaned == "a\nb\nc"运行测试:
python -m pytest tests/ -v如果还没安装 pytest,先执行:
pip install pytest这里的核心思路是:把所有真实样本中的敏感字段和噪音规则沉淀成测试样例,以后每次改规则都跑一遍,防止“修了一个漏,漏了三个洞”。
6.2 Token 节省量估算
Token 数不好精确定义,但可以用经验公式粗略估算:
估算 Token 数 ≈ 字符数 / 4这个公式对英文和代码混合文本比较接近,对中文文本偏差较大。可以在本地写一个统计命令:
wc -c app.log python src/cli.py -i app.log --mode log --stats比如原始日志 12000 字符,清洗后 3600 字符:
- 清洗前估算:约 3000 Token
- 清洗后估算:约 900 Token
- 节省比例:约 70%
这不是精确数据,但足以判断清洗是否有效。
6.3 模型输出质量对比
用手头真实案例做 A/B 对比。下面这种对比方式很容易说明问题:
| 输入方式 | 模型回答风格 | Token 消耗 | 风险 |
|---|---|---|---|
| 不清洗直接发送 | 泛泛而谈,可能复述日志中的配置值 | 高 | 敏感字段随输出暴露 |
| 清洗后发送 | 能定位到 ERROR 行并分析根因 | 显著降低 | 规则误伤会导致上下文缺失 |
在团队实践里,通常会保留一份“清洗前原文件”和“清洗后文件”,方便事后追溯。不要因为有了清洗层,就把原始文件删除,否则出现问题将无法定位规则是不是误删了重要上下文。
6.4 建立“敏感样例集”
从日志和代码中整理一批固定的敏感样例,加入自动化测试中。比如:
- 一个 AWS Key 字符串
- 一个带端口的 IP 地址
- 一个含加号和括号的手机号
- 一个带子域的邮箱地址
- 一个出现在变量名中的疑似 IP 字符串(如
version_1.2.3)
这些样例集会在你调整正则规则时,极大降低回归风险。
7. 常见问题与排查方法
下面这些问题,都是真实使用中容易出现的地方,建议先收藏,遇到问题直接查表。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
代码里的版本号1.2.3被替换成<IP> | IP 正则边界太宽 | 检查ip_address规则匹配详情 | 收紧为“位于 host/ip/url 等关键字附近”才匹配,或加排除逻辑 |
| 日志中的 INFO 行全被删,但错误根因里其实包含 INFO 上下文 | 过滤规则一刀切 | 查看removed_lines统计,是否包含了有用行 | 改为“错误行前后 N 行保留”策略,只过滤与错误无关的 INFO |
| 清洗后 JSON 或代码缩进被破坏 | 脱敏替换过程中改变了结构,或编码问题 | 用json.loads/python -m py_compile验证 | 先做结构完整性校验;占位符只替换值,不要动括号和缩进 |
| 中文日志输出乱码 | Windows 下默认编码不是 UTF-8 | 查看报错是否与gbk编码有关 | 打开文件时使用encoding="utf-8",必要时用errors="replace" |
| 同一个 Token 被多条规则反复替换 | 规则间存在重叠,且执行顺序不当 | 查看mask统计中多次替换的情况 | 调整规则顺序,先匹配最具体的规则;占位符加唯一前缀,防止二次匹配 |
| 发送给模型后返回 JSON schema 错误 | 清洗层破坏了 payload 结构 | 检查output文件是否为合法 JSON/工具参数 | 保证清洗只作用于字段值,不改变键名、类型、嵌套层级 |
| Base64 格式的密钥完全没有被识别 | 无上下文关键字,普通值类正则很难覆盖 | 检查规则是否能匹配长 Base64 字符串 | 增加“长度 + 字符集”通用规则,或结合熵检测识别高随机性字符串 |
这里真正容易踩坑的是第一条:IP 正则误伤。代码里version_1.2.3、2.3.4这种字段很常见,一旦被替换成<IP>,模型就不知道原始版本号是什么了。建议在规则里增加上下文约束,比如只在host|url|addr|ip附近才匹配。
另一个容易踩坑的是 JSON schema。如果你用 LLM 工具调用(Function Calling / Tool Calling),清洗层一定要在清洗后执行一次json.loads校验。如果清洗把字段值改成了占位符,但占位符里有特殊字符导致 JSON 字符串非法,模型服务端就会拒绝请求,这时你会看到类似于“provider rejected the request schema or tool payload”的报错。遇到这种错误,第一步要检查的,不是模型配置,而是你发给服务端的内容是不是合法 JSON。
8. 工程化最佳实践与安全边界
工具能跑通只是第一步,放到生产环境还要考虑规则管理、可审计性、误伤率、以及安全边界。本节给出的建议,能让你少踩一些坑。
8.1 默认严格,先统计后脱敏
规则宁可保守,也不要激进。漏掉一个密钥比删掉一段有用的报错更让人后悔。
实际操作中,第一轮可以先跑“统计模式”,只输出哪些规则会命中哪些字段,不实际替换。确认命中范围合理后,再执行真正的替换。把“统计”和“执行”拆开,可以避免规则刚上线就把全公司的代码清洗得面目全非。
8.2 规则版本管理与白名单
规则文件一定要纳入 Git 管理。团队协作时,任何人都能看到“这条规则为什么加进来,那条规则为什么删掉”。如果某个哈希、密钥格式、关键字段被误伤,可以通过 Git 历史还原规则。
同时,给敏感字段增加一个“白名单模式”。比如内网 IP 范围192.168.0.0/16、10.0.0.0/8是你真正关心的,而1.2.3.4可能是无关数据。用白名单能显著降低误报率。
8.3 最小数据暴露原则
清洗策略的优先级应该是:
- 能不发出去的字段,尽量不发。
- 不能删除的字段,尽量脱敏后发送。
- 脱敏后依然过于敏感的字段,直接截断或摘要。
对于日志中的堆栈,可以合并相同行,只保留前两次出现的位置。对于文档,可以只发送目标段落,而不是整篇全文。核心原则是:给模型的输入,只保留当前任务真正需要的部分。
8.4 与 CI/CD 和脚本工具结合
清洗工具不应该只存在于某个人的私命令行里。推荐把它做成一个独立脚本,纳入 CI 流水线。比如在代码备份、文档导出、调试信息收集阶段,先通过 scrubber 生成一份脱敏副本,再交给后续处理。
# 收集错误日志并清洗 journalctl --since "1 hour ago" | python src/cli.py --mode log --stats > /tmp/clean_error.log# 把当前 git 变更脱敏后传给模型 git diff -- . ':(exclude)lock' | python src/cli.py --mode code > /tmp/clean_diff.txt这样你既保留了原始信息用于本地排查,又把干净版本交给了远程模型服务。
8.5 不要用 Scrubber 替代安全评审
一定要明确一点:Scrubber 是降低风险的工具,不是数据安全的充分条件。
很多公司有明确的数据分类分级要求,外部模型服务允许处理的数据范围、是否需要脱敏、是否允许发送到境外服务器,都需要安全合规团队评估。工具层面的脱敏只能降低“意外泄漏”的概率,不能成为“故意外发高风险数据”的挡箭牌。
8.6 把清洗层扩展到 RAG 与 Agent
如果你的 LLM 应用已经进入 RAG 或 Agent 阶段,清洗位置会增加:
- 文档入库前:在向量化之前,对文档做脱敏和过滤。
- 检索结果返回后:模型会基于检索片段回答,这些片段同样需要脱敏。
- Agent 工具调用返回后:工具返回的 JSON 内容里可能包含敏感字段,需要二次清洗。
尤其是 Agent 场景,工具返回的 schema 和 payload 如果被清洗层改坏了,就会触发校验收缩失败。因此清洗规则要和工具协议校验放在一起,作为发送给模型前的最后一道闸门。
9. 总结
这篇文章的核心观点其实很简单:发给 LLM 的文本,不应该被默认为无害的。一个本地 scrubber 的核心价值,不是让你省几毛钱 token,而是给 LLM 工作流加了一个可审计的守门人层。它让“发送前检查”不再依赖个人自觉,而是变成一个可重复、可测试、可追踪的工程步骤。
对于刚起步的团队,建议按下面的顺序落地:
- 先做一个最小可用版本,只处理日志模式。
- 整理一份敏感字段清单,建立脱敏规则和测试样例。
- 把“统计”和“执行”分离,运行几周,看规则误伤率。
- 再把清洗层扩展到代码模式、文档模式,并逐步融入 RAG 和 Agent 的工作流。
当你开始处理“发送给 LLM 之前应该保留什么、删掉什么、替换什么”这个问题时,你对 LLM 工作流的理解会从“模型应用层”下沉到“数据治理层”,这比单纯优化 prompt 模板要扎实得多。后续可以继续深入的方向包括:LLM 上下文窗口与压缩策略、规则引擎设计、RAG 检索前的数据处理、以及 LLM Agent 工具调用协议的校验。这篇文章提到的代码骨架,建议收藏备用,等真正遇到敏感信息外泄或上下文被噪音稀释的问题时,回来对照着改一改,会省很多时间。