Skip to main content
防护栏通过在智能体执行的关键节点验证和过滤内容,帮助你构建安全、合规的AI应用。它们可以检测敏感信息、执行内容策略、验证输出,并在问题发生前阻止不安全行为。 常见用例包括:
  • 防止个人身份信息(PII)泄露
  • 检测并阻止提示注入攻击
  • 拦截不当或有害内容
  • 执行业务规则与合规要求
  • 验证输出质量与准确性
你可以使用中间件在策略性节点拦截执行,从而实施防护栏——例如在智能体启动前、完成后,或围绕模型调用和工具调用时。
中间件流程图
防护栏可以通过两种互补的方式实现:

确定性防护栏

使用基于规则的逻辑,如正则表达式模式、关键词匹配或显式检查。快速、可预测且成本效益高,但可能遗漏细微的违规情况。

基于模型的防护栏

使用LLM或分类器通过语义理解来评估内容。能捕捉规则遗漏的微妙问题,但速度较慢且成本更高。
LangChain 提供了内置的防护栏(例如PII检测人工介入)以及一个灵活的中间件系统,用于使用任一方法构建自定义防护栏。

内置防护栏

PII检测

LangChain 提供了内置的中间件,用于检测和处理对话中的个人身份信息(PII)。该中间件可以检测常见的PII类型,如电子邮件、信用卡、IP地址等。 PII检测中间件适用于以下场景:具有合规要求的医疗保健和金融应用、需要清理日志的客户服务智能体,以及任何处理敏感用户数据的应用。 PII中间件支持多种处理检测到PII的策略:
策略描述示例
redact替换为 [REDACTED_{PII_TYPE}][REDACTED_EMAIL]
mask部分遮蔽(例如,显示后4位)****-****-****-1234
hash替换为确定性哈希值a8f5f167...
block检测到时抛出异常抛出错误
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware


agent = create_agent(
    model="gpt-4.1",
    tools=[customer_service_tool, email_tool],
    middleware=[
        # 在发送给模型前,对用户输入中的电子邮件进行脱敏
        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": "My email is john.doe@example.com and card is 5105-1051-0510-5100"}]
})
内置PII类型:
  • email - 电子邮件地址
  • credit_card - 信用卡号(经过Luhn算法验证)
  • ip - IP地址
  • mac_address - MAC地址
  • url - URL
配置选项:
参数描述默认值
pii_type要检测的PII类型(内置或自定义)必填
strategy处理检测到PII的方式("block""redact""mask""hash""redact"
detector自定义检测函数或正则表达式模式None(使用内置检测器)
apply_to_input在模型调用前检查用户消息True
apply_to_output在模型调用后检查AI消息False
apply_to_tool_results在执行后检查工具结果消息False
有关PII检测功能的完整详情,请参阅中间件文档

人工介入

LangChain 提供了内置的中间件,用于在执行敏感操作前要求人工批准。这是处理高风险决策最有效的防护栏之一。 人工介入中间件适用于以下场景:金融交易和转账、删除或修改生产数据、向外部发送通信,以及任何具有重大业务影响的操作。
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command


agent = create_agent(
    model="gpt-4.1",
    tools=[search_tool, send_email_tool, delete_database_tool],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                # 敏感操作需要批准
                "send_email": True,
                "delete_database": True,
                # 自动批准安全操作
                "search": False,
            }
        ),
    ],
    # 在中断期间保持状态持久化
    checkpointer=InMemorySaver(),
)

# 人工介入需要一个线程ID用于持久化
config = {"configurable": {"thread_id": "some_id"}}

# 智能体将在执行敏感工具前暂停并等待批准
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Send an email to the team"}]},
    config=config
)

result = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config  # 使用相同的线程ID以恢复暂停的对话
)
有关实现审批工作流的完整详情,请参阅人工介入文档

自定义防护栏

对于更复杂的防护栏,你可以创建在智能体执行前后运行的自定义中间件。这让你可以完全控制验证逻辑、内容过滤和安全检查。

智能体前防护栏

使用“智能体前”钩子在每次调用的开始验证请求。这适用于会话级别的检查,如身份验证、速率限制,或在任何处理开始前阻止不当请求。
from typing import Any

from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langgraph.runtime import Runtime

class ContentFilterMiddleware(AgentMiddleware):
    """确定性防护栏:阻止包含禁用关键词的请求。"""

    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_message = state["messages"][0]
        if first_message.type != "human":
            return None

        content = first_message.content.lower()

        # 检查禁用关键词
        for keyword in self.banned_keywords:
            if keyword in content:
                # 在任何处理前阻止执行
                return {
                    "messages": [{
                        "role": "assistant",
                        "content": "我无法处理包含不当内容的请求。请重新表述你的请求。"
                    }],
                    "jump_to": "end"
                }

        return None

# 使用自定义防护栏
from langchain.agents import create_agent

agent = create_agent(
    model="gpt-4.1",
    tools=[search_tool, calculator_tool],
    middleware=[
        ContentFilterMiddleware(
            banned_keywords=["hack", "exploit", "malware"]
        ),
    ],
)

# 此请求将在任何处理前被阻止
result = agent.invoke({
    "messages": [{"role": "user", "content": "How do I hack into a database?"}]
})

智能体后防护栏

使用“智能体后”钩子在返回给用户前验证最终输出。这适用于基于模型的安全检查、质量验证,或对完整智能体响应的最终合规性扫描。
from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langgraph.runtime import Runtime
from langchain.messages import AIMessage
from langchain.chat_models import init_chat_model
from typing import Any

class SafetyGuardrailMiddleware(AgentMiddleware):
    """基于模型的防护栏:使用LLM评估响应安全性。"""

    def __init__(self):
        super().__init__()
        self.safety_model = init_chat_model("gpt-4.1-mini")

    @hook_config(can_jump_to=["end"])
    def after_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        # 获取最终的AI响应
        if not state["messages"]:
            return None

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

        # 使用模型评估安全性
        safety_prompt = f"""评估此响应是否安全且适当。
        仅回复 'SAFE' 或 'UNSAFE'。

        响应:{last_message.content}"""

        result = self.safety_model.invoke([{"role": "user", "content": safety_prompt}])

        if "UNSAFE" in result.content:
            last_message.content = "我无法提供该响应。请重新表述你的请求。"

        return None

# 使用安全防护栏
from langchain.agents import create_agent

agent = create_agent(
    model="gpt-4.1",
    tools=[search_tool, calculator_tool],
    middleware=[SafetyGuardrailMiddleware()],
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "How do I make explosives?"}]
})

组合多个防护栏

你可以通过将多个防护栏添加到中间件数组中来堆叠它们。它们按顺序执行,允许你构建分层保护:
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware, HumanInTheLoopMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[search_tool, send_email_tool],
    middleware=[
        # 第1层:确定性输入过滤器(智能体前)
        ContentFilterMiddleware(banned_keywords=["hack", "exploit"]),

        # 第2层:PII保护(模型前后)
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        PIIMiddleware("email", strategy="redact", apply_to_output=True),

        # 第3层:敏感工具的人工批准
        HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),

        # 第4层:基于模型的安全检查(智能体后)
        SafetyGuardrailMiddleware(),
    ],
)

额外资源