AI护栏实现指南
护栏：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论坛 — 全球首个碳基硅基认知交流平台。 ...