Guardrails 安全护栏

Guardrails（安全护栏）是 LangChain Agent 中用于保障安全、合规和内容质量的重要机制。通过多层防护策略，确保 Agent 的行为在可控范围内。

为什么需要 Guardrails

在生产环境中部署 Agent，你需要防范以下风险：

提示注入：用户试图绕过限制
敏感信息泄露：模型意外输出敏感数据
工具滥用：调用未被授权的工具
有害内容：生成不当或违规内容
数据合规：违反 GDPR 等法规

内容过滤

输入过滤

在用户输入到达模型之前进行检查：

python

from langchain import create_agent
from langchain.chat_models import init_chat_model
from langchain.guardrails import InputGuardrail

# 自定义输入护栏
class SensitiveDataFilter(InputGuardrail):
    """过滤敏感信息输入。"""
    
    SENSITIVE_PATTERNS = [
        r"\d{18}",            # 身份证号
        r"1[3-9]\d{9}",       # 手机号
        r"\d{16}",            # 信用卡号
    ]
    
    async def check(self, user_input: str) -> bool:
        import re
        for pattern in self.SENSITIVE_PATTERNS:
            if re.search(pattern, user_input):
                return False  # 拒绝请求
        return True
    
    async def on_reject(self, user_input: str) -> str:
        return "您的输入包含敏感信息，已被过滤。"

agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[],
    system_prompt="你是一个安全的助手。",
    guardrails=[SensitiveDataFilter()],
)

输出过滤

在模型回复返回给用户之前进行检查：

python

from langchain.guardrails import OutputGuardrail

class PIIMasker(OutputGuardrail):
    """遮盖输出中的个人信息。"""
    
    async def check(self, output: str) -> str:
        import re
        # 遮盖手机号
        output = re.sub(r"1[3-9]\d{9}", "***手机号***", output)
        # 遮盖邮箱
        output = re.sub(r"\b[\w.-]+@[\w.-]+\.\w+\b", "***邮箱***", output)
        return output

agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[],
    system_prompt="助手。",
    guardrails=[PIIMasker()],
)

工具访问控制

按用户角色限制工具

python

from langchain import create_agent
from langchain.chat_models import init_chat_model
from langchain.guardrails import ToolAccessGuardrail

class RoleBasedToolControl(ToolAccessGuardrail):
    """基于角色的工具访问控制。"""
    
    # 角色 -> 允许的工具列表
    ROLE_TOOLS = {
        "admin": ["delete_user", "view_logs", "edit_config"],
        "editor": ["create_doc", "edit_doc", "search"],
        "viewer": ["search", "read_doc"],
    }
    
    async def is_tool_allowed(self, tool_name: str, context: dict) -> bool:
        role = context.get("user_role", "viewer")
        allowed = self.ROLE_TOOLS.get(role, [])
        return tool_name in allowed
    
    async def on_deny(self, tool_name: str, context: dict) -> str:
        return f"您没有权限使用 {tool_name} 工具。"

agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[delete_user, view_logs, search, read_doc],
    system_prompt="助手。",
    guardrails=[RoleBasedToolControl()],
    context_schema={"user_role": str},
)

工具调用频率限制

python

from langchain.guardrails import ToolRateLimitGuardrail
from datetime import datetime, timedelta

class RateLimiter(ToolRateLimitGuardrail):
    """限制工具的调用频率。"""
    
    def __init__(self):
        self.call_counts = {}
        self.limits = {
            "web_search": (10, 60),     # 60 秒内最多 10 次
            "send_email": (3, 300),     # 300 秒内最多 3 次
            "delete_data": (1, 3600),   # 1 小时内最多 1 次
        }
    
    async def is_allowed(self, tool_name: str) -> bool:
        if tool_name not in self.limits:
            return True
        
        max_calls, window_sec = self.limits[tool_name]
        now = datetime.now()
        
        # 清理过期记录
        if tool_name in self.call_counts:
            self.call_counts[tool_name] = [
                t for t in self.call_counts[tool_name]
                if now - t < timedelta(seconds=window_sec)
            ]
        else:
            self.call_counts[tool_name] = []
        
        if len(self.call_counts[tool_name]) >= max_calls:
            return False
        
        self.call_counts[tool_name].append(now)
        return True

内容安全策略

python

from langchain.guardrails import ContentSafetyGuardrail

class SafetyChecker(ContentSafetyGuardrail):
    """检查内容安全。"""
    
    # 禁止主题
    BLOCKED_TOPICS = [
        "非法活动",
        "仇恨言论",
        "暴力行为",
    ]
    
    async def is_content_safe(self, text: str) -> bool:
        for topic in self.BLOCKED_TOPICS:
            if topic in text:
                return False
        return True

agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[],
    system_prompt="助手。",
    guardrails=[SafetyChecker()],
)

多层护栏组合

python

from langchain.guardrails import (
    InputGuardrail,
    OutputGuardrail,
    ToolAccessGuardrail,
    ContentSafetyGuardrail,
)

# 同时启用多种护栏
agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[search_tool, delete_data, send_email],
    system_prompt="安全的助手。",
    guardrails=[
        SensitiveDataFilter(),      # 输入层
        SafetyChecker(),            # 内容安全层
        RoleBasedToolControl(),     # 工具访问层
        RateLimiter(),             # 频率限制层
        PIIMasker(),              # 输出层
    ],
)

处理流程：

用户输入
    │
    ▼
┌──────────────┐
│  输入护栏     │  ◄── SensitiveDataFilter
└──────┬───────┘
       │ 通过
       ▼
┌──────────────┐
│  内容安全     │  ◄── SafetyChecker
└──────┬───────┘
       │ 通过
       ▼
┌──────────────┐
│  工具调用     │  ◄── RoleBasedToolControl + RateLimiter
│  Agent 执行   │
└──────┬───────┘
       │ 完成
       ▼
┌──────────────┐
│  输出护栏     │  ◄── PIIMasker
└──────┬───────┘
       │ 安全
       ▼
   用户接收

自定义护栏

可以创建自定义护栏满足特定需求：

python

from langchain.guardrails import BaseGuardrail

class AuditLogger(BaseGuardrail):
    """审计日志护栏，记录所有输入输出。"""
    
    async def before(self, input_data: dict, context: dict) -> None:
        print(f"[AUDIT] 输入: {input_data}")
        # 写入日志文件
        with open("audit.log", "a") as f:
            f.write(f"{input_data}\n")
    
    async def after(self, output: any, context: dict) -> any:
        print(f"[AUDIT] 输出: {output}")
        return output  # 可以修改输出

class LanguageConstraint(BaseGuardrail):
    """强制输出使用指定语言。"""
    
    def __init__(self, language: str = "中文"):
        self.language = language
    
    async def before(self, input_data: dict, context: dict) -> dict:
        # 在系统提示中添加语言约束
        messages = input_data.get("messages", [])
        if messages:
            for msg in messages:
                if msg.get("role") == "system":
                    msg["content"] += f"\n请始终使用{self.language}回复。"
        return input_data

Guardrails 配置

python

# 集中配置 Guardrails
guardrail_config = {
    "input_filter": {
        "enabled": True,
        "block_patterns": ["身份证", "银行卡"],
    },
    "output_filter": {
        "enabled": True,
        "mask_pii": True,
        "max_length": 4096,
    },
    "tool_control": {
        "rate_limit": {
            "web_search": {"calls": 10, "window": 60},
            "send_email": {"calls": 3, "window": 300},
        },
        "role_based": {
            "admin": "*",  # 允许所有
            "user": ["search", "read"],
        },
    },
    "content_safety": {
        "blocked_categories": [
            "hate_speech",
            "violence",
            "illegal",
        ],
    },
}

agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[search_tool, send_email],
    system_prompt="助手。",
    guardrails=guardrail_config,  # 直接传入配置 dict
)

最佳实践

深度防御：同时使用输入、内容、工具和输出多层护栏
最小权限：只给 Agent 必要的工具访问权限
日志审计：记录所有 Agent 行为和护栏触发事件
用户反馈：护栏触发时提供友好的拒绝原因
定期更新：护栏规则需要根据实际使用持续优化
灰度发布：先在部分流量上启用护栏，确认无误后全量
性能考虑：复杂护栏可能增加延迟，注意权衡

下一步

了解Runtime 与上下文中上下文对护栏的影响
学习MCP 接入远程工具的访问控制
查看结构化输出确保输出符合安全规范

Guardrails 安全护栏 ​

为什么需要 Guardrails ​

内容过滤 ​

输入过滤 ​

输出过滤 ​

工具访问控制 ​

按用户角色限制工具 ​

工具调用频率限制 ​

内容安全策略 ​

多层护栏组合 ​

自定义护栏 ​

Guardrails 配置 ​

最佳实践 ​

下一步 ​

Guardrails 安全护栏

为什么需要 Guardrails

内容过滤

输入过滤

输出过滤

工具访问控制

按用户角色限制工具

工具调用频率限制

内容安全策略

多层护栏组合

自定义护栏

Guardrails 配置

最佳实践

下一步