LLM 输出护栏与内容安全

# schema_gate.py — JSON Schema 校验护栏
import json, jsonschema

JSON_SCHEMA = {
    "type": "object",
    "required": ["summary", "sentiment", "topics"],
    "properties": {
        "summary":   {"type": "string"},
        "sentiment": {"enum": ["positive", "neutral", "negative"]},
        "topics":    {"type": "array", "items": {"type": "string"}},
    },
}

def validate_schema(text: str) -> dict | None:
    """解析并校验模型输出，非法返回 None。"""
    try:
        obj = json.loads(strip_code_fence(text))
        jsonschema.validate(obj, JSON_SCHEMA)
        return obj
    except (json.JSONDecodeError, jsonschema.ValidationError):
        return None

def recover_structured_output(text: str, repair_llm=None) -> dict:
    """多级恢复：剥离 → 提取 → 修复 → 重试。"""
    for candidate in [text, strip_code_fence(text), extract_json_block(text)]:
        parsed = validate_schema(candidate)
        if parsed:
            return parsed
    if repair_llm:
        fixed = repair_llm(f"修复以下 JSON：\n{text}")
        parsed = validate_schema(fixed)
        if parsed:
            return parsed
    raise OutputSchemaError("无法恢复为合法 Schema")

def structured_generate(query: str, schema: dict) -> dict:
    """生成阶段强制结构化输出 + 输出侧校验兜底。"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": query}],
        response_format={"type": "json_schema", "schema": schema},
    )
    text = resp.choices[0].message.content
    return validate_schema(text)   # 兜底校验

# pii_gate.py — 输出侧 PII 识别与脱敏
import re

PII_PATTERNS = [
    (r"1[3-9]\d{9}",            "MOBILE"),   # 中国大陆手机号
    (r"\d{17}[\dXx]",           "ID_CARD"),  # 身份证
    (r"(?:\d{4}[-\s]?){4}",     "BANK_CARD"),
    (r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}", "EMAIL"),
    (r"\d{6}-?\d{4}-?\d{2}",    "BIRTH"),
]

def find_pii(text: str) -> list[dict]:
    """基于正则 + NER 的 PII 识别。"""
    hits = []
    for pattern, kind in PII_PATTERNS:
        for m in re.finditer(pattern, text):
            hits.append({"kind": kind, "value": m.group(),
                         "start": m.start(), "end": m.end()})
    return hits

def deidentify(text: str, mode="mask") -> str:
    """按策略处理输出中的 PII。"""
    if mode == "mask":
        for pattern, _ in PII_PATTERNS:
            text = re.sub(pattern, lambda m: m.group()[0] + "*" * (len(m.group()) - 1), text)
        return text
    if mode == "block":
        if find_pii(text):
            raise PIISafetyError("输出含敏感信息，已阻断")
        return text
    if mode == "placeholder":
        for pattern, kind in PII_PATTERNS:
            text = re.sub(pattern, f"[{kind}]", text)
        return text
    return text

# moderation.py — 三层内容过滤
KEYWORD_BLOCKLIST = ["制作炸弹步骤", "自杀方法", "购买毒品渠道"]

def keyword_screen(text: str) -> str | None:
    """第一层：规则关键词，快且零成本。"""
    lowered = text.lower()
    for kw in KEYWORD_BLOCKLIST:
        if kw in lowered:
            return f"命中禁用词：{kw}"
    return None

def light_classifier(text: str) -> dict:
    """第二层：小模型分类器，高吞吐。"""
    return small_model.moderate(text)   # {"category": ..., "score": ...}

def heavy_classifier(text: str) -> dict:
    """第三层：强模型复审，低误杀。"""
    return large_model.moderate(text)   # 用于边界与高价值流量

def safe_medical_output(text: str) -> str:
    """医疗话题：在专业结论前注入免责声明。"""
    if detect_topic(text, "medical"):
        return text + "\n\n> 以上内容仅为信息参考，不构成医疗建议。请咨询执业医师。"
    return text

def intercept_dangerous(output: str, query: str) -> dict:
    """识别危险意图：阻断 + 转介帮助资源。"""
    intent = intent_classifier(query, output)   # 识别查询与输出的危险意图
    if intent["level"] == "high":
        return {
            "allow": False,
            "user_message": "无法提供相关内容。若你或他人正面临困难，请联系专业帮助。",
            "internal_log": intent,
        }
    return {"allow": True, "output": output}

# policy_engine.py — 可配置的政策规则引擎
POLICY_RULES = [
    {"id": "R001", "trigger": "contains_financial_advice",
     "action": "require_disclaimer", "channel": ["chat", "api"]},
    {"id": "R002", "trigger": "pii_detected",
     "action": "mask", "severity": "high"},
    {"id": "R003", "trigger": "adult_content",
     "action": "block", "severity": "critical"},
]

def evaluate_policy(output: str, context: dict) -> list[dict]:
    """对输出跑全部政策规则，返回命中的规则与动作。"""
    triggered = []
    for rule in POLICY_RULES:
        if triggers[rule["trigger"]](output, context):
            triggered.append(rule)
    return triggered

def audit_log(entry: dict):
    """护栏决策审计：谁、何时、何种输出、何种裁决。"""
    record = {
        "timestamp": datetime.utcnow().isoformat(),
        "trace_id": entry["trace_id"],
        "query_hash": sha256(entry["query"]).hexdigest(),
        "output_snippet": truncate(entry["output"], 200),
        "layers": entry["layers"],          # 各层护栏结果
        "verdict": entry["verdict"],
        "operator": entry.get("operator"),   # 人工复核人
    }
    append_to_audit_store(record)

def fast_path_filter(text: str) -> dict | None:
    """快速通道：关键词 + 轻量分类器，低延迟通过。"""
    kw = keyword_screen(text)
    if kw:
        return {"allow": False, "reason": kw}
    light = light_classifier(text)
    if light["category"] == "safe":
        return {"allow": True}          # 快路径放行
    return None                         # 进入慢路径复审

def guardrail_quality(golden_outputs, guardrail, labels) -> dict:
    """在标注集上评估护栏：误杀、误放、精确率、召回率。"""
    fp = 0   # 误杀：正常被拦
    fn = 0   # 误放：危险被放
    for output, label in zip(golden_outputs, labels):
        verdict = guardrail(output)
        if label == "safe" and not verdict["allow"]:
            fp += 1
        if label == "danger" and verdict["allow"]:
            fn += 1
    n = len(golden_outputs)
    return {
        "false_block_rate": fp / n,
        "false_release_rate": fn / n,
    }

场景：UGC 平台的 AI 写作助手，输出面向公众
需求：格式可解析为 Markdown + JSON 元数据
隐私：不输出任何第三方 PII
内容：不产出仇恨、色情、暴力、违禁品内容
合规：医疗与金融话题带免责声明；未成年人话题特殊处理

# content_platform_guard.py — 完整输出护栏编排
class ContentGuardrailPipeline:
    def __init__(self):
        self.schema = load_schema("writing_output")
        self.moderation = ModerationStack()
        self.pii = PIIGate()
        self.policy = PolicyEngine()

    def check(self, output: str, context: dict) -> dict:
        # L1 结构
        parsed = validate_schema(output)
        if parsed is None:
            parsed = recover_structured_output(output, self.repair_llm)
            if parsed is None:
                return self.fail("output_schema", context)

        text = parsed.get("content", "")

        # L2 内容
        mod = self.moderation.check(text)
        if not mod["allow"]:
            return self.fail("moderation", context, mod)

        # L3 数据
        safe_text = self.pii.deidentify(text, mode="mask")

        # L4 合规
        rules = self.policy.evaluate(safe_text, context)
        for rule in rules:
            if rule["action"] == "block":
                return self.fail("policy", context, rule)
            if rule["action"] == "require_disclaimer":
                safe_text = append_disclaimer(safe_text, rule)

        return {"verdict": "PASS", "output": safe_text}

def validate_guardrail_deploy(golden, pipeline) -> dict:
    """上线前验证：质量、误杀率、误放率、延迟。"""
    quality = guardrail_quality(golden["texts"], pipeline, golden["labels"])
    latency = measure_avg_latency(pipeline)
    return {
        "false_block_rate": quality["false_block_rate"],
        "false_release_rate": quality["false_release_rate"],
        "p95_latency_ms": latency["p95"],
        "pass": quality["false_release_rate"] < 0.001,   # 误放必须极低
    }