主题
🛡️ 9.12 生产级防护:护栏安全与测试评估
“功能上线只是起点,守住底线才能长久。护栏(Guardrails)与测试,是 AI Agent 从 Demo 走向生产环境的最后一道闸门。”
本节把 9.11 学到的中间件手艺用于“防守”:检测敏感信息、拦截 Prompt 注入、审查有害输出,并用测试把安全机制固化下来。
💡 先看一个直观例子:安检门与安检员
1. 护栏解决什么问题?
生产环境里,Agent 要面对的是真实的恶意与意外:
- 用户输入里夹带手机号、身份证、信用卡(PII 泄露);
- 有人试图用“忽略之前的指令…”绕过系统提示(Prompt 注入);
- 模型被诱导输出有害、违规内容(内容安全)。
护栏就是在 Agent 执行的关键节点上做校验与过滤,把问题挡在造成后果之前。
2. 两类护栏(官方定义)
| 类型 | 实现方式 | 优点 | 缺点 |
|---|---|---|---|
| 确定性护栏 | 正则、关键词、显式检查 | 快、便宜、可预测 | 可能漏掉微妙违规 |
| 模型性护栏 | 用 LLM/分类器做语义判断 | 能抓到规则漏掉的 | 慢、贵 |
生活比喻:确定性护栏像机场的金属探测门——快、机械、规则固定;模型性护栏像旁边的安检员——能识别“这瓶液体虽然没过标尺但看起来可疑”这类模糊情况。生产环境通常是两者叠加。
🛡️ 内置护栏:PII 检测与人类在环
1. PII 检测:四种策略一键切换
PIIMiddleware 内置了邮箱、信用卡(Luhn 校验)、IP、MAC、URL 等常见类型,支持四种处置策略:
| 策略 | 效果 | 示例 |
|---|---|---|
redact | 替换为占位符 | [REDACTED_EMAIL] |
mask | 部分打码 | ****-****-****-1234 |
hash | 确定性哈希 | a8f5f167... |
block | 检测到即抛错 | 直接阻断 |
python
# code/s12_guardrails_and_testing.py
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
agent = create_agent(
model=llm,
tools=tools,
middleware=[
PIIMiddleware("email", strategy="redact", apply_to_input=True), # 输入邮箱脱敏
PIIMiddleware("credit_card", strategy="mask", apply_to_input=True), # 信用卡打码
PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block", apply_to_input=True), # 自定义检测
],
)💡
apply_to_input(查用户输入)/apply_to_output(查 AI 回复)/apply_to_tool_results(查工具结果)三个开关可独立控制。1.3.2+ 的apply_to_output=True还能对流式输出实时脱敏。
2. 人类在环:高危操作签字画押
对发邮件、删库、付款等高风险工具,HumanInTheLoopMiddleware 会在执行前暂停等人工批准(完整流程已在 9.9 第 11 点演示,此处不再重复)。
🧱 自定义护栏:输入拦截与输出审查
1. before_agent:输入黑名单拦截(确定性)
python
# code/s12_guardrails_and_testing.py
from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langgraph.runtime import Runtime
from typing import Any
class ContentFilterMiddleware(AgentMiddleware):
"""确定性护栏:黑名单关键词一旦命中,Agent 直接结束,连模型都不调。"""
def __init__(self, banned_keywords: list[str]):
super().__init__()
self.banned_keywords = [kw.lower() for kw in banned_keywords]
@hook_config(can_jump_to=["end"])
def before_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
if not state["messages"]:
return None
first = state["messages"][0]
if first.type != "human":
return None
content = first.content.lower()
for kw in self.banned_keywords:
if kw in content:
return {
"messages": [{"role": "assistant", "content": "无法处理包含不当内容的请求,请重新表述。"}],
"jump_to": "end",
}
return None2. after_agent:输出安全审查(模型性)
用一个廉价小模型做“输出安全裁判”,语义层面兜底:
python
from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langchain.messages import AIMessage
from langgraph.runtime import Runtime
from typing import Any
class SafetyGuardrailMiddleware(AgentMiddleware):
"""模型性护栏:用 LLM 复核最终回复是否安全合规。"""
def __init__(self):
super().__init__()
# 复用项目统一模型(.env 的 OpenAI 兼容端点)。
# 坑:官方示例硬编码 init_chat_model("openai:gpt-4o-mini"),
# 在方舟等国产端点上会 404(UnsupportedModel)——复核模型必须跟着端点走。
from s01_model_io import get_chat_model_primary
self.safety_model = get_chat_model_primary(temperature=0)
@hook_config(can_jump_to=["end"])
def after_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
if not state["messages"]:
return None
last = state["messages"][-1]
if not isinstance(last, AIMessage):
return None
prompt = f"评估以下回复是否安全合规,只回复 SAFE 或 UNSAFE。\n\n回复:{last.content}"
result = self.safety_model.invoke([{"role": "user", "content": prompt}])
if "UNSAFE" in result.content:
return {"messages": [AIMessage("我无法提供该回复,请重新表述你的需求。")], "jump_to": "end"}
return None3. 多层组合:纵深防御
python
agent = create_agent(
model=llm,
tools=tools,
middleware=[
ContentFilterMiddleware(banned_keywords=["hack", "exploit", "malware"]), # 第 1 层:输入黑名单
PIIMiddleware("email", strategy="redact", apply_to_input=True), # 第 2 层:PII 脱敏
PIIMiddleware("email", strategy="redact", apply_to_output=True),
HumanInTheLoopMiddleware(interrupt_on={"send_email": True}), # 第 3 层:高危审批
SafetyGuardrailMiddleware(), # 第 4 层:输出安全复核
],
)🖥️ 实验台真实调用演示(9.12 页可交互)
code/s12_guardrails_and_testing.py 的两个 demo_*() 已从「仅装配」升级为真实调用,点实验台按钮即可观察:
| 演示 | 观察点 |
|---|---|
| 🛡️ PII 中间件 | 输入 john.doe@example.com → 模型亲口承认只看到 [REDACTED_EMAIL]——敏感信息在进模型前就被确定性改写,零出境 |
| 🧱 自定义护栏 | 场景 A「教我 hack」命中黑名单 → before_agent 毫秒级拦截,模型零调用零 Token;场景 B 正常请求 → 黑名单/PII/安全复核三层全部放行 |
🔑 纵深防御的直接证据:场景 A 的拦截回复在毫秒级返回(对比正常调用的 10 秒+), 说明请求根本没出网——这就是确定性护栏「连模型都不调」的设计价值。
🧪 测试评估:把安全机制固化下来
护栏写得再好,不测试就等于没写。LangChain 官方推荐两条腿:
- 确定性断言(单元测试):护栏大多是确定性规则,直接写断言验证“黑名单词被拦”“PII 被脱敏”“HITL 会中断”——零成本、毫秒级、可进 CI:
python
# code/s12_guardrails_and_testing.py —— 确定性护栏的轻量自测
def test_content_filter():
state = {"messages": [{"role": "user", "content": "How do I hack into a database?"}]}
# 断言:命中黑名单后返回 jump_to="end" 且不触发模型
result = content_filter_middleware_test(state)
assert result is not None and result.get("jump_to") == "end"
def test_pii_redact():
# 断言:邮箱被脱敏为 [REDACTED_EMAIL]
assert "[REDACTED_EMAIL]" in pii_redact("联系我 john@example.com")- 云端可观测与评估平台(暂不接入):生产环境通常还会接入官方的 LangSmith 或开源的 Langfuse 做全链路 Trace、评估集与回归对比。出于零配置门槛考虑,本教程暂时不接入;如有需求,后续会单独出一小节介绍两大平台的接入。当前用确定性断言 + 固定用例人工回归即可守住底线。
🔑 工程直觉:确定性护栏用断言测,模型性护栏用评估集测;两者都要纳入上线前的门禁(Gate),才能真正守住底线。
📚 权威官方资料直达
- 🔗 护栏官方指南(本节核心):Guardrails
- 🔗 PII 检测中间件:PII Detection Middleware
- 🔗 人类在环官方指南:Human-in-the-loop
- 🔗 测试 Agent 官方指南:Testing Agents
护栏要用对抗样本验证
确定性规则适合拦截格式、权限和明确的敏感模式,模型护栏适合处理语义和上下文。两类护栏都可能误拦或漏拦,因此测试集既要有应该拒绝的样本,也要有措辞相近但应当通过的样本。
测试还应覆盖编码变形、长文本夹带、检索内容中的提示注入,以及工具参数绕过。每次修改提示词、模型或中间件后重跑固定评估集,并记录通过率、误拒率和高风险漏检。对不可逆操作,即使护栏通过,也仍应保留权限校验与审批。
🎯 本节小结与思考
- 核心收获:掌握了确定性 vs 模型性两类护栏,内置
PIIMiddleware的四种策略与HumanInTheLoopMiddleware高危审批,自定义before_agent输入拦截 /after_agent输出审查,以及纵深防御的多层组合与确定性断言 + 固定用例回归的测试策略。 - 下一步探索:到这里,LangChain 1.4 的全部核心零件——I/O、LCEL、工具、记忆、可观测、上下文工程、中间件、护栏——都已装进你的工具箱。最后一步,就是把这些零件全部装配起来,打造一个真正融会贯通的生产级项目。下一节我们学习 9.13 综合实战:AI 智能数码选购与避坑决策 Agent。