为什么要验证 AI 的输出？

AI 模型的输出本质上是概率生成的文本，并非确定性的程序执行结果。即使输入完全相同，每次输出都可能略有差异。在生产环境中，不加验证地信任 AI 输出会带来多种风险：

三层校验架构

完整的输出校验应分三个层级递进执行：

第一层：结构校验

验证输出的格式是否符合预期，这是最基础、最快速的校验。

import json
import re
from typing import Any, Optional

class StructureValidator:
    """结构校验器"""
    
    def validate_json(self, output: str) -> tuple[bool, Optional[Any], str]:
        """校验JSON格式"""
        # 提取JSON块（模型常在JSON外加说明文字）
        json_match = re.search(r'\{.*\}|\[.*\]', output, re.DOTALL)
        if not json_match:
            return False, None, "输出中未找到JSON格式内容"
        
        try:
            parsed = json.loads(json_match.group())
            return True, parsed, ""
        except json.JSONDecodeError as e:
            return False, None, f"JSON解析失败: {e}"
    
    def validate_schema(self, data: dict, required_fields: list, 
                        field_types: dict = None) -> tuple[bool, list]:
        """校验JSON Schema"""
        errors = []
        
        # 检查必填字段
        for field in required_fields:
            if field not in data:
                errors.append(f"缺少必填字段: {field}")
        
        # 检查字段类型
        if field_types:
            for field, expected_type in field_types.items():
                if field in data and not isinstance(data[field], expected_type):
                    errors.append(f"字段 {field} 类型错误，期望 {expected_type.__name__}")
        
        return len(errors) == 0, errors
    
    def validate_length(self, output: str, min_len: int = 10, 
                        max_len: int = 10000) -> tuple[bool, str]:
        """校验输出长度"""
        if len(output) < min_len:
            return False, f"输出过短（{len(output)} 字符），可能未正常生成"
        if len(output) > max_len:
            return False, f"输出过长（{len(output)} 字符），超过预期范围"
        return True, ""

第二层：语义校验

结构合法之后，进一步检查内容的语义质量——是否有害、是否偏题、是否包含幻觉特征。

import re

class SemanticValidator:
    """语义校验器"""
    
    # 有害内容关键词（根据业务场景定制）
    HARMFUL_PATTERNS = [
        r"(制作|合成|购买).*(炸弹|毒品|武器)",
        r"(自杀|自残)方法",
        r"(攻击|入侵|破解).*(系统|服务器|账号)",
    ]
    
    # 幻觉风险特征词（模型不确定时常用的措辞）
    HALLUCINATION_SIGNALS = [
        "根据我的了解", "我相信", "可能是", "大约",
        "据我所知", "不太确定", "可能有误",
    ]
    
    def check_harmful_content(self, text: str) -> tuple[bool, list]:
        """检测有害内容"""
        found = []
        for pattern in self.HARMFUL_PATTERNS:
            if re.search(pattern, text, re.IGNORECASE):
                found.append(pattern)
        return len(found) == 0, found
    
    def check_hallucination_risk(self, text: str) -> dict:
        """评估幻觉风险"""
        signals_found = [s for s in self.HALLUCINATION_SIGNALS if s in text]
        risk_level = "低"
        if len(signals_found) >= 3:
            risk_level = "高"
        elif len(signals_found) >= 1:
            risk_level = "中"
        return {"risk_level": risk_level, "signals": signals_found}
    
    def check_relevance(self, output: str, keywords: list) -> float:
        """检查输出与预期主题的相关性（简单关键词覆盖率）"""
        matched = sum(1 for kw in keywords if kw in output)
        return matched / len(keywords) if keywords else 1.0

第三层：业务规则校验

最后一层针对具体业务场景进行定制校验。

class BusinessRuleValidator:
    """业务规则校验器（示例：电商场景）"""
    
    def validate_product_info(self, data: dict) -> tuple[bool, list]:
        """校验商品信息的业务规则"""
        errors = []
        
        # 价格必须为正数
        if data.get("price", 0) <= 0:
            errors.append("价格必须大于0")
        
        # 库存不能为负
        if data.get("stock", 0) < 0:
            errors.append("库存不能为负数")
        
        # 折扣率必须在 0-1 之间
        discount = data.get("discount", 1.0)
        if not 0 < discount <= 1:
            errors.append("折扣率必须在 (0, 1] 范围内")
        
        # 商品名称不能包含违禁词
        forbidden = ["最好", "第一", "最低价"]  # 广告法违禁词示例
        title = data.get("title", "")
        for word in forbidden:
            if word in title:
                errors.append(f"标题含违禁词: {word}")
        
        return len(errors) == 0, errors

完整的校验流水线

class OutputValidationPipeline:
    """完整的AI输出校验流水线"""
    
    def __init__(self):
        self.structure_validator = StructureValidator()
        self.semantic_validator = SemanticValidator()
        self.business_validator = BusinessRuleValidator()
    
    def validate(self, raw_output: str, task_config: dict) -> dict:
        """
        执行完整校验流程
        task_config: {
            "expected_format": "json" | "text",
            "required_fields": [...],     # 仅JSON格式
            "field_types": {...},         # 仅JSON格式
            "check_harmful": True,
            "business_rule": "product_info" | None,
            "min_length": 50,
            "max_length": 2000,
        }
        """
        result = {
            "passed": False,
            "output": raw_output,
            "parsed_data": None,
            "errors": [],
            "warnings": [],
            "hallucination_risk": None,
        }
        
        # === 第一层：结构校验 ===
        fmt = task_config.get("expected_format", "text")
        
        if fmt == "json":
            ok, parsed, err = self.structure_validator.validate_json(raw_output)
            if not ok:
                result["errors"].append(f"[结构] {err}")
                return result  # 结构错误，直接终止
            result["parsed_data"] = parsed
            
            # Schema 校验
            ok, errors = self.structure_validator.validate_schema(
                parsed,
                task_config.get("required_fields", []),
                task_config.get("field_types", {})
            )
            if not ok:
                result["errors"].extend([f"[Schema] {e}" for e in errors])
                return result
        
        # 长度校验
        ok, err = self.structure_validator.validate_length(
            raw_output,
            task_config.get("min_length", 10),
            task_config.get("max_length", 10000)
        )
        if not ok:
            result["errors"].append(f"[长度] {err}")
        
        # === 第二层：语义校验 ===
        if task_config.get("check_harmful", True):
            ok, harmful = self.semantic_validator.check_harmful_content(raw_output)
            if not ok:
                result["errors"].append(f"[有害内容] 检测到违禁模式: {harmful}")
                result["output"] = "[内容已被拦截，请联系管理员]"
                return result
        
        hallucination = self.semantic_validator.check_hallucination_risk(raw_output)
        result["hallucination_risk"] = hallucination
        if hallucination["risk_level"] == "高":
            result["warnings"].append("输出中存在较多不确定性表述，建议人工复核")
        
        # === 第三层：业务规则校验 ===
        rule = task_config.get("business_rule")
        if rule == "product_info" and result["parsed_data"]:
            ok, errors = self.business_validator.validate_product_info(
                result["parsed_data"]
            )
            if not ok:
                result["errors"].extend([f"[业务规则] {e}" for e in errors])
        
        # 最终判定
        result["passed"] = len(result["errors"]) == 0
        return result

# ===== 使用示例 =====
pipeline = OutputValidationPipeline()

ai_output = '{"title": "优质耳机", "price": 299, "stock": 50, "discount": 0.8}'
validation = pipeline.validate(ai_output, {
    "expected_format": "json",
    "required_fields": ["title", "price", "stock"],
    "field_types": {"price": (int, float), "stock": int},
    "check_harmful": True,
    "business_rule": "product_info",
})
print("校验结果:", "通过✅" if validation["passed"] else "失败❌")
print("错误:", validation["errors"])
print("警告:", validation["warnings"])

自动校验 vs 人工复核的选择

场景建议方式原因
格式/结构校验全自动规则明确，机器执行更快更准
有害内容检测自动拦截 + 人工复审误报率有限，需人工确认是否真正有害
高幻觉风险输出标记 + 人工复核机器难以判断事实准确性
面向公众的内容人工终审品牌风险高，需人工把关
内部工具/低风险场景自动+超时重试效率优先，人工成本高

小结

AI 输出校验是生产环境不可缺少的安全网。三层递进架构——结构校验→语义校验→业务规则校验——从格式到内容到业务逻辑层层把关。关键原则：不信任任何未经校验的 AI 输出，在架构设计时把校验链路和 AI 调用视为同等重要的组件。