护栏：AI安全的最后一道防线

AI护栏（Guardrails）是在LLLM输入和输出之间设置的安全过滤层。它在模型生成的响应到达用户之前，检查并修改不安全的内容。

三层护栏架构

用户输入 → [输入护栏] → LLM → [输出护栏] → 用户
              ↓                ↓
           [拦截/修改]      [拦截/修改]

输入护栏

class InputGuardrail:
    def __init__(self):
        self.checks = [
            self.check_prompt_injection,
            self.check_pii_input,
            self.check_toxic_input,
            self.check_topic_restriction,
        ]
    
    async def validate(self, user_input, context=None):
        results = []
        for check in self.checks:
            result = await check(user_input, context)
            results.append(result)
            if not result["passed"]:
                # 严重违规直接拦截
                if result["severity"] == "high":
                    return results, False
        
        return results, True
    
    async def check_prompt_injection(self, text, context):
        """检测提示注入攻击"""
        injection_patterns = [
            r"忽略.{0,10}(指令|规则|限制)",
            r"(ignore|disregard).{0,10}(previous|above|all)",
            r"(system|admin|root)\s*(prompt|instruction)",
            r"你的.{0,5}(指令|提示|prompt)",
            r"reveal.{0,10}(system|hidden|secret)",
        ]
        
        for pattern in injection_patterns:
            if re.search(pattern, text, re.IGNORECASE):
                return {
                    "check": "prompt_injection",
                    "passed": False,
                    "severity": "high",
                    "error": "检测到提示注入攻击"
                }
        
        return {"check": "prompt_injection", "passed": True}
    
    async def check_pii_input(self, text, context):
        """检测输入中的PII"""
        pii_patterns = {
            "phone": r'1[3-9]\d{9}',
            "id_card": r'\d{17}[\dXx]',
            "bank_card": r'\d{16,19}',
            "email": r'[\w.-]+@[\w.-]+\.\w+',
        }
        
        found = {}
        for pii_type, pattern in pii_patterns.items():
            matches = re.findall(pattern, text)
            if matches:
                found[pii_type] = matches
        
        if found:
            return {
                "check": "pii_input",
                "passed": False,
                "severity": "medium",
                "error": f"输入包含敏感信息: {list(found.keys())}"
            }
        
        return {"check": "pii_input", "passed": True}
    
    async def check_topic_restriction(self, text, context):
        """话题限制检查"""
        restricted_topics = ["政治敏感", "暴力恐怖", "色情"]
        
        # 使用分类模型检测
        for topic in restricted_topics:
            if await self.topic_classifier(text, topic):
                return {
                    "check": "topic_restriction",
                    "passed": False,
                    "severity": "high",
                    "error": f"话题不在允许范围内: {topic}"
                }
        
        return {"check": "topic_restriction", "passed": True}

输出护栏

class OutputGuardrail:
    def __init__(self):
        self.checks = [
            self.check_toxicity,
            self.check_pii_leak,
            self.check_hallucination,
            self.check_format_safety,
        ]
    
    async def validate(self, response, context=None):
        for check in self.checks:
            result = await check(response, context)
            if not result["passed"]:
                if result["action"] == "block":
                    return False, self.safe_fallback(result["error"])
                elif result["action"] == "sanitize":
                    response = self.sanitize(response, result)
        
        return True, response
    
    async def check_toxicity(self, text, context):
        """毒性检测"""
        # 使用毒性分类器
        toxicity_score = await self.toxicity_model(text)
        
        if toxicity_score > 0.7:
            return {
                "check": "toxicity",
                "passed": False,
                "action": "block",
                "error": f"输出包含有害内容 (score={toxicity_score:.2f})"
            }
        
        return {"check": "toxicity", "passed": True}
    
    async def check_pii_leak(self, text, context):
        """检查输出是否泄露PII"""
        # 如果输入包含PII，检查输出是否回显
        if context and "input_pii" in context:
            for pii in context["input_pii"]:
                if pii in text:
                    return {
                        "check": "pii_leak",
                        "passed": False,
                        "action": "sanitize",
                        "error": "输出包含输入中的PII"
                    }
        
        return {"check": "pii_leak", "passed": True}
    
    async def check_hallucination(self, text, context):
        """幻觉检测"""
        if context and "retrieved_docs" in context:
            # 基于检索文档检查幻觉
            is_grounded = await self.fact_check(text, context["retrieved_docs"])
            
            if not is_grounded:
                return {
                    "check": "hallucination",
                    "passed": False,
                    "action": "block",
                    "error": "输出可能与事实不符"
                }
        
        return {"check": "hallucination", "passed": True}
    
    def safe_fallback(self, error):
        """安全兜底响应"""
        return f"抱歉，我无法回答这个问题。({error})"
    
    def sanitize(self, text, check_result):
        """清理不安全内容"""
        # 替换敏感信息
        if check_result["check"] == "pii_leak":
            for pii in check_result.get("pii_list", []):
                text = text.replace(pii, "***")
        return text

完整护栏管线

class GuardrailPipeline:
    def __init__(self, input_guardrail, output_guardrail, llm):
        self.input_gr = input_guardrail
        self.output_gr = output_guardrail
        self.llm = llm
    
    async def process(self, user_input, conversation_context=None):
        # 1. 输入护栏
        input_results, input_passed = await self.input_gr.validate(
            user_input, conversation_context
        )
        
        if not input_passed:
            blocked_check = next(r for r in input_results if not r["passed"])
            return {
                "response": f"抱歉，您的请求无法处理。{blocked_check['error']}",
                "blocked": True,
                "reason": blocked_check["check"]
            }
        
        # 2. LLM生成
        try:
            response = await self.llm.generate(user_input)
        except Exception as e:
            return {"response": "服务暂时不可用", "error": str(e)}
        
        # 3. 输出护栏
        passed, safe_response = await self.output_gr.validate(
            response, {"input": user_input, **conversation_context or {}}
        )
        
        return {
            "response": safe_response,
            "blocked": not passed,
            "guardrail_checks": {
                "input": input_results,
            }
        }

护栏日志

class GuardrailLogger:
    def __init__(self):
        self.logger = structlog.get_logger()
    
    def log_block(self, check, error, user_input, severity):
        self.logger.warning("guardrail_block",
            check=check,
            error=error,
            severity=severity,
            input_preview=user_input[:100],
            timestamp=datetime.now().isoformat()
        )

结语

AI护栏是保障LLM安全输出的关键基础设施。输入护栏防止恶意输入和不当话题，输出护栏检测毒性、PII泄露和幻觉。三层管线（输入检查→LLM生成→输出检查）配合日志监控，构建起完整的AI安全防线。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。