5.1 Guardrails

本节介绍 LangChain 中的安全护栏机制，用于验证和过滤 Agent 执行过程中的内容。

什么是 Guardrails？

Guardrails（安全护栏） 是在 Agent 执行的关键点验证和过滤内容的安全机制。

"Guardrails validate and filter content at key points in your agent's execution."

核心用途

用途	说明
防止 PII 泄露	检测并处理敏感个人信息
检测提示注入	识别恶意提示攻击
阻止有害内容	过滤不当或危险内容
执行业务规则	确保符合业务逻辑
验证输出质量	保证响应质量

两种实现方式

1. 确定性护栏（Deterministic Guardrails）

基于规则的逻辑，使用正则表达式和关键词匹配：

python

from langchain.agents import before_agent

@before_agent
def check_banned_words(state, runtime):
    """检查禁用词"""
    banned_words = ["hack", "exploit", "bypass"]
    user_input = state["messages"][-1].content.lower()

    for word in banned_words:
        if word in user_input:
            state["messages"].append(
                AIMessage(content="抱歉，您的请求包含不允许的内容。")
            )
            return {"jump_to": "end"}

    return None

特点：

快速、低成本
可能遗漏细微违规

2. 模型驱动护栏（Model-based Guardrails）

使用 LLM 进行语义理解和评估：

python

from langchain.agents import after_agent
from langchain_openai import ChatOpenAI

safety_model = ChatOpenAI(model="gpt-4o-mini")

@after_agent
def safety_check(state, runtime):
    """使用 LLM 进行安全检查"""
    response = state["messages"][-1].content

    check_result = safety_model.invoke(
        f"评估以下回复是否安全、适当：\n\n{response}\n\n回复 'safe' 或 'unsafe'"
    )

    if "unsafe" in check_result.content.lower():
        state["messages"][-1] = AIMessage(
            content="抱歉，我无法提供该类型的回复。"
        )

    return None

特点：

语义理解能力强
速度较慢、成本较高

内置护栏

PIIDetectionMiddleware

检测常见的个人敏感信息类型：

python

from langchain.agents.middleware import PIIDetectionMiddleware

middleware = PIIDetectionMiddleware(
    pii_types=[
        "email",       # 邮箱地址
        "credit_card", # 信用卡号
        "ip_address",  # IP 地址
        "mac_address", # MAC 地址
        "url",         # URL 链接
        "phone",       # 电话号码
        "ssn",         # 社会安全号
    ],
    strategy="redact",  # 处理策略
    apply_to_input=True,
    apply_to_output=True,
    apply_to_tool_results=True,
)

处理策略

策略	说明	示例
`redact`	完全替换	`[REDACTED_EMAIL]`
`mask`	部分遮蔽	`**--**-1234`
`hash`	哈希替换	`a1b2c3d4e5f6...`
`block`	阻止请求	抛出异常

完整示例

python

from langchain.agents import create_agent
from langchain.agents.middleware import PIIDetectionMiddleware

agent = create_agent(
    "gpt-4o",
    tools=[my_tools],
    middleware=[
        PIIDetectionMiddleware(
            strategy="mask",
            pii_types=["email", "credit_card", "phone"],
            on_detect="warn",  # "warn" | "block" | "log"
        )
    ]
)

自定义护栏

Before Agent 护栏

在 Agent 调用开始时执行，用于会话级检查：

python

from langchain.agents import before_agent, AgentState, Runtime
from langchain_core.messages import AIMessage

@before_agent
def validate_session(state: AgentState, runtime: Runtime):
    """验证会话"""
    user_input = state["messages"][-1].content

    # 检查输入长度
    if len(user_input) > 10000:
        state["messages"].append(
            AIMessage(content="输入内容过长，请精简后重试。")
        )
        return {"jump_to": "end"}

    # 检查禁用关键词
    banned_keywords = ["password", "secret_key", "api_key"]
    for keyword in banned_keywords:
        if keyword in user_input.lower():
            state["messages"].append(
                AIMessage(content="请勿在对话中包含敏感凭据。")
            )
            return {"jump_to": "end"}

    return None

After Agent 护栏

在 Agent 完成后验证最终输出：

python

from langchain.agents import after_agent
from langchain_openai import ChatOpenAI

@after_agent
def validate_output(state, runtime):
    """验证输出内容"""
    response = state["messages"][-1].content

    # 检查是否包含代码执行建议
    dangerous_patterns = [
        "rm -rf",
        "DROP TABLE",
        "eval(",
        "exec(",
    ]

    for pattern in dangerous_patterns:
        if pattern in response:
            state["messages"][-1] = AIMessage(
                content="检测到潜在危险内容，已过滤回复。"
            )
            break

    return None

模型驱动安全检查

python

from langchain.agents import after_agent
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

class SafetyCheck(BaseModel):
    is_safe: bool
    reason: str

safety_llm = ChatOpenAI(model="gpt-4o-mini").with_structured_output(SafetyCheck)

@after_agent
def llm_safety_check(state, runtime):
    """LLM 驱动的安全检查"""
    response = state["messages"][-1].content

    result = safety_llm.invoke(
        f"评估以下内容是否安全、适当、无害：\n\n{response}"
    )

    if not result.is_safe:
        state["messages"][-1] = AIMessage(
            content=f"内容已过滤。原因：{result.reason}"
        )

    return None

分层保护

多个护栏按顺序堆叠，形成多层防护：

用户输入
    │
    ▼
┌─────────────────────────────┐
│ 1. 确定性输入过滤            │  ← 正则、关键词
├─────────────────────────────┤
│ 2. PII 保护                  │  ← 敏感信息处理
├─────────────────────────────┤
│ 3. 人工审批                  │  ← 敏感操作确认
├─────────────────────────────┤
│ 4. 模型安全检查              │  ← LLM 语义评估
└─────────────────────────────┘
    │
    ▼
安全响应

组合使用示例

python

from langchain.agents import create_agent, before_agent, after_agent
from langchain.agents.middleware import (
    PIIDetectionMiddleware,
    HumanInTheLoopMiddleware,
)

@before_agent
def input_filter(state, runtime):
    """输入过滤"""
    # 确定性检查逻辑
    return None

@after_agent
def output_validation(state, runtime):
    """输出验证"""
    # 模型驱动检查逻辑
    return None

agent = create_agent(
    "gpt-4o",
    tools=[send_email, delete_file, query_database],
    middleware=[
        # 1. 输入过滤
        input_filter,

        # 2. PII 保护
        PIIDetectionMiddleware(strategy="mask"),

        # 3. 敏感操作审批
        HumanInTheLoopMiddleware(
            tools=["send_email", "delete_file"]
        ),

        # 4. 输出验证
        output_validation,
    ]
)

类方式实现

对于复杂的护栏逻辑，使用类方式：

python

from langchain.agents import AgentMiddleware, AgentState, Runtime
from langchain_core.messages import AIMessage

class ContentModerationGuardrail(AgentMiddleware):
    """内容审核护栏"""

    def __init__(self, blocked_topics: list, severity: str = "warn"):
        self.blocked_topics = blocked_topics
        self.severity = severity

    def before_agent(self, state: AgentState, runtime: Runtime):
        """检查用户输入"""
        user_input = state["messages"][-1].content.lower()

        for topic in self.blocked_topics:
            if topic in user_input:
                if self.severity == "block":
                    state["messages"].append(
                        AIMessage(content=f"话题 '{topic}' 不在服务范围内。")
                    )
                    return {"jump_to": "end"}
                else:
                    print(f"警告: 检测到敏感话题 '{topic}'")

        return None

    def after_agent(self, state: AgentState, runtime: Runtime):
        """检查 Agent 输出"""
        response = state["messages"][-1].content.lower()

        for topic in self.blocked_topics:
            if topic in response:
                state["messages"][-1] = AIMessage(
                    content="抱歉，我无法讨论该话题。"
                )
                break

        return None

# 使用
agent = create_agent(
    "gpt-4o",
    tools=[my_tools],
    middleware=[
        ContentModerationGuardrail(
            blocked_topics=["violence", "illegal", "harmful"],
            severity="block"
        )
    ]
)

最佳实践

实践	说明
分层防护	结合确定性和模型驱动护栏
性能优先	先执行快速的确定性检查
明确反馈	向用户解释为何内容被过滤
日志记录	记录所有护栏触发事件
定期更新	根据新威胁更新规则
测试覆盖	全面测试护栏有效性

护栏类型对比

类型	速度	成本	准确性	适用场景
确定性	快	低	中	明确规则、关键词过滤
模型驱动	慢	高	高	语义理解、细微判断
混合	中	中	高	生产环境推荐

上一节：5.0 Advanced Usage

下一节：5.2 Runtime

5.1 Guardrails ​

什么是 Guardrails？ ​

核心用途 ​

两种实现方式 ​

1. 确定性护栏（Deterministic Guardrails） ​

2. 模型驱动护栏（Model-based Guardrails） ​

内置护栏 ​

PIIDetectionMiddleware ​

处理策略 ​

完整示例 ​

自定义护栏 ​

Before Agent 护栏 ​

After Agent 护栏 ​

模型驱动安全检查 ​

分层保护 ​

组合使用示例 ​

类方式实现 ​

最佳实践 ​

护栏类型对比 ​

5.1 Guardrails

什么是 Guardrails？

核心用途

两种实现方式

1. 确定性护栏（Deterministic Guardrails）

2. 模型驱动护栏（Model-based Guardrails）

内置护栏

PIIDetectionMiddleware

处理策略

完整示例

自定义护栏

Before Agent 护栏

After Agent 护栏

模型驱动安全检查

分层保护

组合使用示例

类方式实现

最佳实践

护栏类型对比