Skip to content

16-langchain-guardrails.md

Latest commit

 

History

History
414 lines (311 loc) · 13.6 KB

16-langchain-guardrails.md

File metadata and controls

414 lines (311 loc) · 13.6 KB

16. 防护栏（Guardrails）

原文链接: https://docs.langchain.com/oss/python/langchain/guardrails

为智能体实现安全检查与内容过滤

防护栏（Guardrails）通过在智能体执行的关键节点验证和过滤内容，帮助你构建安全、合规的 AI 应用。它们可以：

  • 检测敏感信息
  • 执行内容策略
  • 验证输出
  • 在问题发生前阻止不安全行为

常见用例包括：

  • 防止 PII（个人可识别信息）泄露
  • 检测并阻止提示注入攻击
  • 阻止不当或有害内容
  • 执行业务规则和合规要求
  • 验证输出质量和准确性

你可以使用中间件在策略点拦截执行（智能体开始前、完成后，或围绕模型和工具调用）来实现防护栏。

防护栏可以通过两种互补方式实现：

确定性防护栏（Deterministic guardrails）

使用基于规则的逻辑，如正则表达式模式、关键词匹配或显式检查。速度快、可预测且成本低，但可能遗漏细微的违规行为。

基于模型的防护栏（Model-based guardrails）

使用 LLM 或分类器通过语义理解评估内容。能捕获规则遗漏的细微问题，但速度较慢且成本更高。

LangChain 既提供内置防护栏（如 PII 检测、人在回路），也提供灵活的中间件系统，让你可以使用任一方法构建自定义防护栏。

内置防护栏（Built-in guardrails）

PII 检测（PII detection）

LangChain 提供内置中间件来检测和处理对话中的个人可识别信息（PII）。该中间件可以检测常见的 PII 类型，如邮箱、信用卡号、IP 地址等。

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-4o",
    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 [email protected] and card is 5105-1051-0510-5100"}]
})

内置 PII 类型和配置：

内置 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 检测功能的完整详细信息，请参阅中间件文档。

人在回路（Human-in-the-loop）

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-4o",
    tools=[search_tool, send_email_tool, delete_database_tool],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                # 敏感操作需要审批
                "send_email": True,
                "delete_database": True,
                # 安全操作自动批准
                "search": False,
            }
        ),
    ],
    # 在中断之间持久化状态
    checkpointer=InMemorySaver(),
)

# 人在回路需要 thread 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  # 使用相同的 thread ID 以恢复暂停的对话
)

有关实现审批工作流的完整详细信息，请参阅人在回路文档。

自定义防护栏（Custom guardrails）

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

智能体执行前的防护栏（Before agent guardrails）

使用 "before agent" 钩子在每次调用的开始验证请求一次。这对于会话级检查很有用，如身份验证、速率限制或在任何处理开始前阻止不当请求。

类语法示例

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": "I cannot process requests containing inappropriate content. Please rephrase your request."
                    }],
                    "jump_to": "end"
                }

        return None

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

agent = create_agent(
    model="gpt-4o",
    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 typing import Any
from langchain.agents.middleware import before_agent, AgentState
from langgraph.runtime import Runtime

@before_agent(can_jump_to=["end"])
def content_filter(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """在智能体开始前检查内容。"""
    banned_keywords = ["hack", "exploit", "malware"]
    
    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": "I cannot process requests containing inappropriate content."
                }],
                "jump_to": "end"
            }
    
    return None

智能体执行后的防护栏（After agent guardrails）

使用 "after agent" 钩子在返回给用户前验证最终输出一次。这对于基于模型的安全检查、质量验证或对完整智能体响应的最终合规扫描很有用。

类语法示例

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-4o-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"""Evaluate if this response is safe and appropriate.
        Respond with only 'SAFE' or 'UNSAFE'.

        Response: {last_message.content}"""

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

        if "UNSAFE" in result.content:
            last_message.content = "I cannot provide that response. Please rephrase your request."

        return None

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

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

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

装饰器语法示例

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

safety_model = init_chat_model("gpt-4o-mini")

@after_agent(can_jump_to=["end"])
def safety_check(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """在智能体完成后进行安全性检查。"""
    if not state["messages"]:
        return None
    
    last_message = state["messages"][-1]
    if not isinstance(last_message, AIMessage):
        return None
    
    safety_prompt = f"""Evaluate if this response is safe and appropriate.
    Respond with only 'SAFE' or 'UNSAFE'.

    Response: {last_message.content}"""
    
    result = safety_model.invoke([{"role": "user", "content": safety_prompt}])
    
    if "UNSAFE" in result.content:
        last_message.content = "I cannot provide that response. Please rephrase your request."
    
    return None

组合多个防护栏（Combine multiple guardrails）

你可以通过将多个防护栏添加到中间件数组中来堆叠它们。它们按顺序执行，允许你构建分层保护：

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

agent = create_agent(
    model="gpt-4o",
    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(),
    ],
)

这种分层方法提供了多层保护：

  1. 输入过滤：在请求进入系统前阻止明显不当的内容
  2. PII 保护：在输入和输出中检测并处理敏感信息
  3. 人工审批：对高风险操作进行人工审查
  4. 输出验证：在返回给用户前进行最终安全检查

更多资源（Additional resources）

  • Middleware documentation - 自定义中间件的完整指南
  • Middleware API reference - 中间件 API 的完整参考
  • Human-in-the-loop - 为敏感操作添加人工审查
  • Testing agents - 测试安全机制的策略

本文档由 LangChain 官方文档翻译而来