LLM应用监控：PII扫描与敏感词监控

可视化演示

动态展示LLM输出经正则/NER两层检测、PII脱敏、监控告警的完整管线，切换正则/NER/混合/流式四种检测模式，观察PII标记与脱敏过程。

为何监控LLM应用

一座没有边检站的机场会发生什么？航班在跑道上乱成一团，旅客走丢没人知道，行李里混进危险品——因为没有任何记录与检查。LLM应用刚上线时也是这样：用户在前端输入提示词，模型在云端“黑箱”里生成回复，工程师对中间发生的一切一无所知。

这种盲区带来三个代价。第一是质量盲区：模型偶尔“胡言乱语”、循环重复、拒绝回答，没有监控记录，你就无法定位是哪一层出的问题。第二是成本失控：每次推理都要消耗Token，而Token直接对应真金白银的API账单。如果某个用户用超长提示词反复调用，月底账单会让人目瞪口呆。第三是安全与合规风险：用户的手机号、身份证号可能被模型原样输出，模型也可能生成违规内容——这些都需要被看见、被拦截。

因此，可观测性与监控不是“锦上添花”，而是LLM应用上生产的“入场券”。最基础的一个指标就是“可用性”——它衡量系统到底有多少请求真正成功返回：

有了这座边检站，机场才算基本运转起来。但光有“开门营业”还不够——接下来要做的，是给边检站配上更精细的记录系统和安检仪器。

Langfuse可观测平台

每个旅客过关时，边检员都会翻开记录本，记下时间、航班、护照号、目的地。Langfuse就是LLM边检站的“电子记录本”。它是一个开源的LLM可观测性平台，把每次请求沉淀成一条结构化的档案，也就是"trace"。

trace是Langfuse的核心概念，指一次完整请求的生命周期：从用户输入提示词，到模型调用，再到最终输出。一条trace里可以包含多个"span"——每个span是链路中的一个环节，比如一次大模型调用、一次工具调用、一次向量检索。整个链路像一棵树：根是trace，树枝是span，最细的叶子是底层操作。

记录本上写什么？一是完整内容：原始提示词和模型回复，方便事后回溯“模型当时到底看到了什么”；二是性能数字：每个span的耗时、往返延迟；三是资源消耗：输入Token与输出Token的数量。有了这些，Langfuse的看板就能回答“哪个环节最慢、谁烧钱最多、哪类问题最频繁”。工程上最关注的性能指标之一是平均延迟——它直接决定用户体感：

记录本越详细，边检站越从容。Langfuse把每个“旅客”的完整旅程都写进档案，让后续的PII扫描和敏感词检测有了可操作的抓手。

Phoenix可观测平台

不是所有机场都有空间装下巨型记录本。Arize开源的Phoenix就像一座轻量边检站——一个可以在数据分析师“办公桌”（Notebook）上即开即用的工具。它同样能记录trace、追踪模型调用，但定位更偏向研究与调试。

Phoenix的特点是“轻”与“近”。轻：它可以嵌入Jupyter Notebook运行，拉起一个本地UI，即刻查看当前会话的推理链路；近：它离开发者的工作流更近，跑完一个实验就能立刻打开可视化面板分析结果，无需部署独立服务。Phoenix还支持OpenInference标准，使其能与主流框架（LangChain、LlamaIndex等）无缝对接。

Langfuse和Phoenix不是二选一的对手，而是互补的搭档：Langfuse适合在生产环境长期部署，持续采集线上流量；Phoenix适合在开发阶段快速分析，验证Prompt质量与模型行为。不少团队的做法是：开发期用Phoenix调优，上线后由Langfuse盯守。

无论是哪套系统，采集数据都会遇到一个核心权衡——采样率。全量采集数据翔实但成本高，按比例采样省钱但可能漏掉异常。采样率刻画了监控视野的大小：

无论是巨型记录本还是轻量备案簿，边检站的目标一致——把旅客信息完整、可控地留下来。下一节就看这些信息如何被自动采集。

Trace采集配置

光有记录本还不够，总不能靠边检员手写吧？Trace采集就是给边检站加装自动摄像头和感应闸机——SDK会自动记录每个请求，不需要业务代码手工上报。

以Langfuse为例，接入极其简单。首先初始化客户端，传入公钥与密钥：

然后，用 @observe() 装饰器标注需要追踪的函数，SDK会自动捕获函数参数、返回值、异常与耗时，并关联到同一条trace。对于装饰器覆盖不到的场景，可以手动创建 langfuse.trace() 对象，通过 trace.span() 开启子环节，再写入input/output/metadata等字段。所有数据会异步批量上报，不阻塞业务主流程。

Token消耗是trace中最容易被忽视却直接关乎账单的数字。每次推理的Token成本可以精确计算——输入Token和输出Token的单价不同，输出通常更贵：

摄像头装好了，每个旅客的影像自动入档。但入档之前，还有一道关键的安检工序——检查行李里有没有敏感个人物品。

from langfuse import Langfuselangfuse = Langfuse(public_key="pk-...", secret_key="sk-...")

PII扫描与脱敏

安检机的传送带上，每一件行李都要过X光。当扫描到护照（身份证号）、钱包（银行卡号）、手机（电话号码）时，安检员不会大声念出全场旅客的隐私，而是给行李贴上“已标记”的隐私标签。PII扫描就是LLM监控中的那台X光机。

PII（Personally Identifiable Information，个人身份信息）包括身份证号、手机号、邮箱、银行卡号、住址等。Langfuse允许在写入存储前挂载“事件处理器”（event handler），对即将入库的提示词与回复做预处理。一个常见方案是：先用正则表达式匹配确定性强的模式（手机号 1[3-9]\d{9}、身份证 \d{17}[\dXx]），再用NER命名实体识别模型兜底，识别“人名+地址”这类没有固定规律的PII。

识别之后的脱敏策略有三种：掩码替换，如把 13812345678 变成 1385678；哈希脱敏，对敏感字段做不可逆的SHA-256摘要；截断，只保留前几位。生产实践中，掩码替换用得最多——它保留部分可读性，便于排查问题时确定“是哪一位用户”，又不暴露完整信息。

一次PII扫描是否到位，可以用脱敏率来衡量：

行李贴上隐私标签，旅客的个人信息就不再“裸奔”。但这只解决了“隐私泄露”问题，另一类更主动的安全风险——违规内容——还得靠黑名单来挡。

敏感词实时监控

边检站除了查隐私，还要防违禁品。海关有一本厚厚的黑名单，上面写着违禁品名称——一旦某个旅客的行李里翻出对应物品，闸机立刻落下。LLM应用同样需要这样一本“黑名单”，用来拦截模型生成的违规内容。

敏感词监控覆盖的范畴远超“脏话”：包括暴力、色情、政治敏感、仇恨言论、诈骗话术等。技术实现上，最基础的是规则引擎：把敏感词列表加载进Trie树或AC自动机，在输出文本上做多模式匹配，命中即触发。但生产环境远没有这么简单——用户和模型都会“变形绕过”：谐音、拼音缩写、插入符号。因此，成熟方案会叠加语义分类模型，识别“没有命中关键词但语义违规”的句子。

监控动作分三级：一是“放行”，未命中任何规则；二是“替换”，命中后把敏感词替换为 * 再返回给用户；三是“拦截”，直接停止生成并触发告警。在电商客服、开放社区等场景，替换比硬拦截更温和——既阻止了违规，又不破坏对话体验。

敏感词检测器本身的效果，要用精确率与召回率来衡量——它们就像安检的“不误伤”和“不漏查”：

黑名单挡住了大部分违禁品。但安检员不可能24小时盯着屏幕——当异常情况出现时，还需要一套警报系统把值班员叫醒。

告警规则设计

机场控制中心里有一排指示灯：黄色表示“注意”，红色表示“立即响应”。警铃不会随便响——只有当旅客行为触发了明确规则（比如连续三次翻包可疑），控制中心才会向值班员发出通知。告警规则设计就是给LLM监控装上这样一套“分级警铃”。

规则的首要元素是阈值。例如：“每分钟PII命中率超过5%，触发警告”；“敏感词拦截率超过10次/分钟，触发紧急告警”。阈值不宜拍脑袋定，通常参考过去7天的基线：高于均值两倍标准差才告警，避免误报轰炸。

第二个元素是窗口。采样率决定视野，检测窗口决定灵敏度。阈值+滑动窗口，决定告警何时拉响：

第三个元素是渠道与升级。告警通知分流：P0级（如大规模PII泄露）立即发短信+调用值班电话；P1级（如某个会话反复命中敏感词）发Slack/钉钉；P2级（如延迟缓慢上升）只进日报。同时设置“静默”策略——同一类告警在30分钟内不重复发送，防止告警疲劳。规则之间还应联动：连续触发3次P1告警后自动升级为P0，把“人肉盯屏”变成“系统主动找人”。

警铃分级之后，控制中心不再鸡飞狗跳。到此，一座完整的LLM边检站落成：记录（trace）、安检（PII）、黑名单（敏感词）、警铃（告警），四道工序各司其职。

AI 应用场景

Python 演示

import refrom datetime import datetime  class LLMMonitor:    PII_PATTERNS = {        "phone": re.compile(r"\b1[3-9]\d{9}\b"),        "id_card": re.compile(r"\b\d{17}[\dXx]\b"),        "email": re.compile(r"\b[\w.-]+@[\w.-]+\.\w{2,4}\b"),    }    SENSITIVE_WORDS = ["违禁品", "暴力袭击", "政治敏感词"]     def __init__(self):        self.trace_count = 0        self.pii_hits = 0        self.sensitive_hits = 0     def _mask_value(self, value):        if len(value) <= 8:            return value[0] + "****" + value[-1]        return value[:3] + "****" + value[-4:]     def mask_pii(self, text):        masked = text        for pattern in self.PII_PATTERNS.values():            masked = pattern.sub(lambda m: self._mask_value(m.group(0)), masked)        return masked     def check_sensitive(self, text):        for word in self.SENSITIVE_WORDS:            if word in text:                return True        return False     def record_trace(self, prompt, response, latency_ms):        self.trace_count += 1        raw = prompt + response        masked = self.mask_pii(raw)        if masked != raw:            self.pii_hits += 1        is_sensitive = self.check_sensitive(response)        if is_sensitive:            self.sensitive_hits += 1        return {            "trace_id": "tr-" + str(self.trace_count),            "time": datetime.now().isoformat(),            "latency_ms": latency_ms,            "masked": masked,            "sensitive": is_sensitive,        }  def main():    monitor = LLMMonitor()    samples = [        ("介绍北京", "故宫很好玩, 订票电话13812345678", 320),        ("讲个故事", "程序员小明的邮箱是test@example.com", 210),        ("如何投资", "不要相信高收益诱惑, 严禁暴力袭击宣传", 180),    ]    for prompt, response, latency in samples:        t = monitor.record_trace(prompt, response, latency)        flag = "敏感词拦截" if t["sensitive"] else "正常"        print("[%s] %s 耗时%dms 脱敏后:%s"              % (t["trace_id"], flag, t["latency_ms"], t["masked"][:42]))    print("PII命中:%d 敏感词命中:%d"          % (monitor.pii_hits, monitor.sensitive_hits))  if __name__ == "__main__":    main()

1. masked = pattern.sub(lambda m: self._mask_value(m.group(0)), masked)把文本中所有匹配的敏感值替换为掩码版本，相当于X光机自动贴隐私标签。
2. if masked != raw:通过比较掩码前后文本，判断本次请求是否包含PII，并累计命中次数。
3. for word in self.SENSITIVE_WORDS:逐个检查回复是否命中敏感词黑名单，命中即标记违规。
4. record_trace()组装与Langfuse结构相似的trace记录，含trace_id、时间、延迟、脱敏后文本。
5. if __name__ == "__main__":保证脚本可直接运行，输出每条请求的监控结果与统计汇总。

常见误区

面试题