18 安全护栏

前面的文章中，我们一直在让Agent变得更强大——更多的工具、更灵活的上下文、更智能的决策。但有一个问题我们还没认真聊过：安全。

Agent能调用工具、访问数据库、发送邮件，能力越大，出错的代价也越大。你肯定不希望Agent：

• 把用户的手机号、身份证号泄露到日志里
• 执行用户注入的恶意指令
• 生成不当内容
• 做出超出权限的操作

安全护栏（Guardrails）就是来解决这些问题的——在Agent执行的关键节点做检查和过滤，把危险挡在门外。

打个比方：Agent是你的员工，护栏就是公司的安全制度。员工再能干，也得遵守制度——该戴安全帽的地方要戴，该签字的地方要签，不能越权操作。

一、两种护栏方式

护栏有两种实现思路，可以单独用，也可以组合用：

方式	原理	优点	缺点
确定性护栏	正则匹配、关键词检测、规则判断	快、便宜、可预测	只能检测已知模式
模型驱动护栏	用LLM评估内容是否安全	能理解语义，发现隐蔽问题	慢、贵、有延迟

实际项目中，通常先用确定性护栏做第一道防线（快且便宜），再用模型驱动护栏做第二道防线（处理复杂情况）。

二、PII检测

PII（Personally Identifiable Information，个人身份信息）是最常见的安全风险。LangChain内置了PIIMiddleware中间件，开箱即用。

2.1 内置PII类型

类型	说明	示例
email	邮箱地址	john@example.com
credit_card	信用卡号（会做Luhn校验）	5105-1051-0510-5100
ip	IP地址	192.168.1.1
mac_address	MAC地址	00:1B:44:11:3A:B7
url	URL链接	https://example.com

2.2 四种处理策略

检测到PII之后，你可以选择四种处理方式：

策略	效果	适合场景
redact	替换为[REDACTED_类型]	日志脱敏
mask	部分遮挡，如****-****-****-1234	展示时保护
hash	替换为确定性哈希值	需要可追踪但不暴露原文
block	直接抛异常，阻止请求	严格禁止PII的场景

2.3 基本用法

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware


agent = create_agent(
    model="deepseek-v4-flash",
    tools=[customer_service_tool, email_tool],
    middleware=[
        # 用户输入中的邮箱，替换为[REDACTED_EMAIL]
        PIIMiddleware(
            "email",
            strategy="redact",
            apply_to_input=True,
        ),
        # 用户输入中的信用卡号，部分遮挡
        PIIMiddleware(
            "credit_card",
            strategy="mask",
            apply_to_input=True,
        ),
        # 检测到API密钥直接阻止
        PIIMiddleware(
            "api_key",
            detector=r"sk-[a-zA-Z0-9]{32}",
            strategy="block",
            apply_to_input=True,
        ),
    ],
)

# 用户输入包含PII，会按策略处理
result = agent.invoke({
    "messages": [{"role": "user", "content": "我的邮箱是john@example.com，卡号5105-1051-0510-5100"}]
})
# 发给模型的内容已经脱敏：邮箱被替换，卡号被遮挡

2.4 检测范围控制

通过参数控制在哪些环节做PII检测：

PIIMiddleware(
    "email",
    strategy="redact",
    apply_to_input=True,       # 检测用户输入（发给模型之前）
    apply_to_output=True,      # 检测模型输出（返回给用户之前）
    apply_to_tool_results=True,# 检测工具返回值
)

参数	什么时候检测	默认值
apply_to_input	用户消息发给模型之前	True
apply_to_output	模型响应返回用户之前	False
apply_to_tool_results	工具执行结果返回模型之前	False

2.5 自定义检测器

内置类型不够用？用正则表达式定义自己的检测规则：

# 检测中国手机号
PIIMiddleware(
    "phone_number",
    detector=r"1[3-9]\d{9}",
    strategy="redact",
    apply_to_input=True,
)

# 检测身份证号
PIIMiddleware(
    "id_card",
    detector=r"\d{17}[\dXx]",
    strategy="mask",
    apply_to_input=True,
)

三、自定义护栏

内置的PII检测是确定性护栏。对于更复杂的场景——比如检测用户请求是否包含不当内容、模型输出是否安全——你需要自定义护栏。

3.1 before_agent：请求前校验

在Agent开始处理之前做检查。适合做关键词过滤、认证校验、限流等：

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_agent
from langgraph.runtime import Runtime
from typing import Any


# 敏感词列表
BANNED_KEYWORDS = ["hack", "exploit", "malware", "注入", "攻击"]


@before_agent(can_jump_to=["end"])
def content_filter(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """确定性护栏：拦截包含敏感词的请求"""
    if not state["messages"]:
        return None

    first_message = state["messages"][0]
    if first_message.type != "human":
        return None

    content = first_message.content.lower()

    for keyword in BANNED_KEYWORDS:
        if keyword in content:
            # 直接返回错误信息，跳过后续所有处理
            return {
                "messages": [{
                    "role": "assistant",
                    "content": "您的请求包含不当内容，请重新描述您的问题。",
                }],
                "jump_to": "end",
            }

    return None


agent = create_agent(
    model="deepseek-v4-flash",
    tools=[search_tool],
    middleware=[content_filter],
)

# 这个请求会被拦截，不会调用模型
result = agent.invoke({
    "messages": [{"role": "user", "content": "怎么hack数据库？"}]
})

关键点是can_jump_to=["end"]和jump_to: "end"——如果检测到问题，直接跳到结束，不调用模型，不执行工具。

3.2 after_agent：输出后检查

在Agent生成最终响应之后做检查。适合用模型做安全评估、质量验证：

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_agent
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
from langgraph.runtime import Runtime
from typing import Any


safety_model = init_chat_model("deepseek-v4-flash")


@after_agent(can_jump_to=["end"])
def safety_check(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """模型驱动护栏：用LLM评估响应是否安全"""
    if not state["messages"]:
        return None

    last_message = state["messages"][-1]
    if not isinstance(last_message, AIMessage):
        return None

    # 用另一个模型评估安全性
    result = safety_model.invoke([{
        "role": "user",
        "content": (
            "判断以下回答是否安全、恰当。只回复SAFE或UNSAFE。\n\n"
            f"回答: {last_message.content}"
        ),
    }])

    if "UNSAFE" in result.content:
        last_message.content = "抱歉，我无法提供该回答。请换个问题试试。"

    return None


agent = create_agent(
    model="deepseek-v4-flash",
    tools=[search_tool],
    middleware=[safety_check],
)

这种方式的好处是能发现隐蔽的安全问题——比如模型的回答看似正常，但实际上包含了有害信息。确定性规则很难检测这类情况，但用LLM评估可以。

3.3 认证校验

检查用户是否有权限使用Agent：

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_agent
from langgraph.runtime import Runtime
from typing import Any


@before_agent(can_jump_to=["end"])
def auth_check(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """认证校验：未登录用户不能使用Agent"""
    user_id = runtime.context.user_id

    if not user_id:
        return {
            "messages": [{
                "role": "assistant",
                "content": "请先登录后再使用此服务。",
            }],
            "jump_to": "end",
        }

    return None

3.4 限流

防止用户滥用Agent：

import time
from collections import defaultdict
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_agent
from langgraph.runtime import Runtime
from typing import Any


# 简单的滑动窗口限流
request_times: dict[str, list[float]] = defaultdict(list)
RATE_LIMIT = 10  # 每分钟最多10次请求
WINDOW = 60      # 窗口大小：60秒


@before_agent(can_jump_to=["end"])
def rate_limit(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """限流：每用户每分钟最多10次请求"""
    user_id = runtime.context.user_id or "anonymous"
    now = time.time()

    # 清理过期记录
    request_times[user_id] = [
        t for t in request_times[user_id] if now - t < WINDOW
    ]

    if len(request_times[user_id]) >= RATE_LIMIT:
        return {
            "messages": [{
                "role": "assistant",
                "content": "请求太频繁，请稍后再试。",
            }],
            "jump_to": "end",
        }

    request_times[user_id].append(now)
    return None

四、多层护栏组合

实际项目中，你通常需要叠加多层防护。中间件按数组顺序执行，形成层层防线：

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware, HumanInTheLoopMiddleware

agent = create_agent(
    model="deepseek-v4-flash",
    tools=[search_tool, send_email_tool, delete_data_tool],
    middleware=[
        # 第1层：关键词过滤（请求前，确定性）
        content_filter,

        # 第2层：PII脱敏（输入侧）
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),

        # 第3层：敏感操作需要人工审批
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": True,
                "delete_data": True,
            },
        ),

        # 第4层：输出安全检查（响应后，模型驱动）
        safety_check,
    ],
)

四层防护的执行顺序：

用户输入
  → 第1层：关键词过滤（不通过就直接拒绝）
  → 第2层：PII脱敏（把敏感信息替换掉）
  → 模型调用
  → 第3层：敏感工具需要人工确认
  → 第4层：输出安全检查
返回给用户

每一层解决不同的问题，组合起来形成完整的安全体系。

五、最佳实践

1. 先确定性后模型：确定性护栏快且便宜，放在前面做第一道过滤；模型驱动护栏放后面处理复杂情况
2. PII检测是底线：只要处理用户数据，就应该开启PII检测，至少对输入做脱敏
3. 敏感操作加审批：涉及删除、发送、支付等操作，用人在环路做最后把关
4. 不要过度依赖护栏：护栏是辅助手段，不是万能的。根本的安全措施是限制数据库权限、API权限等
5. 监控护栏触发：记录哪些请求触发了护栏，分析是否有误判或遗漏

六、总结

安全护栏是Agent从"能用"到"放心用"的关键：

• PII检测：内置中间件，支持多种PII类型和处理策略
• 确定性护栏：用规则做快速过滤，关键词、正则、认证、限流
• 模型驱动护栏：用LLM做语义评估，发现隐蔽的安全问题
• 多层组合：按顺序叠加多层防护，各司其职

安全不是事后补救，而是从一开始就该设计进去的。在你构建Agent的时候，把护栏当作和工具、记忆一样重要的组件来对待。

在下一篇文章中，我们将学习SQL Agent——让Agent查询数据库，这是最常见的业务场景之一。