7.3 输出质量门控与过滤

工具执行前需要多层验证防线来确保安全和有效性。本节介绍六层质量门控机制、规划-生成-评估三角色分离模式，以及如何通过独立评估者消除自我确认偏差。

7.3.1 问题定义

工具调用前的最后防线是质量门控(Quality Gate)。未经验证的输出可能导致：

• 参数错误：工具接收非法数据导致执行失败

• 资源泄露：过度调用导致费用飙升或系统宕机

• 格式异常：API 调用失败导致 智能体循环重试

• 安全漏洞：注入攻击、不合理的权限操作

质量门控在 执行之前 进行多层验证，确保只有安全、合法的调用才能通过。

其中“权限检查”调用第 12 章的统一策略决策服务；质量门控不维护第二套权限政策，只消费同一个决策结果。

7.3.2 门控层次结构

质量门控按以下六个层次依次进行验证：

图 7-3：质量门控多层验证流程 —— 工具执行前的六层递进式防护

1. 基础格式检查

验证解析结果的完整性和一致性：

from dataclasses import dataclass
from enum import Enum
from typing import Optional, List

class ValidationResult(Enum):
    PASS = "pass"
    FAIL = "fail"
    WARN = "warn"

@dataclass
class ValidationReport:
    result: ValidationResult
    errors: List[str]
    warnings: List[str]
    suggestion: Optional[str] = None

class FormatValidator:
    """基础格式验证"""

    @staticmethod
    def validate_tool_use_block(tool_use) -> ValidationReport:
        """验证工具调用块的结构完整性"""
        errors = []
        warnings = []

        # 检查必填字段
        if not tool_use.id:
            errors.append("tool_use.id 不能为空")
        if not tool_use.name:
            errors.append("tool_use.name 不能为空")
        if tool_use.input is None:
            errors.append("tool_use.input 不能为空")

        # 检查 ID 格式(Claude 的 tool_use.id 以 "toolu_" 开头)
        if tool_use.id and not tool_use.id.startswith("toolu_"):
            warnings.append(
                f"非标准 tool_use.id: {tool_use.id}"
            )

        # 检查名称格式(应为 snake_case)
        if tool_use.name and not FormatValidator._is_valid_name(tool_use.name):
            warnings.append(
                f"工具名格式非标准: {tool_use.name}"
            )

        if errors:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=errors,
                warnings=warnings,
                suggestion="检查 API 响应是否正确解析"
            )

        if warnings:
            return ValidationReport(
                result=ValidationResult.WARN,
                errors=[],
                warnings=warnings
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

    @staticmethod
    def _is_valid_name(name: str) -> bool:
        """检查工具名是否符合 snake_case"""
        return name.islower() and name.replace("_", "").isalnum()

2. 工具存在性检查

验证工具是否已注册：

class ToolRegistry:
    """工具注册表与存在性检查"""

    def __init__(self):
        self.tools = {}

    def register(self, name: str, handler, schema):
        """注册工具"""
        self.tools[name] = {
            "handler": handler,
            "schema": schema,
            "enabled": True
        }

    def is_tool_available(self, name: str) -> bool:
        """检查工具是否可用"""
        if name not in self.tools:
            return False
        return self.tools[name]["enabled"]

    def get_tool_schema(self, name: str):
        """获取工具的参数 schema"""
        if name not in self.tools:
            return None
        return self.tools[name]["schema"]

class ExistenceValidator:
    """工具存在性验证"""

    def __init__(self, registry: ToolRegistry):
        self.registry = registry

    def validate(self, tool_name: str) -> ValidationReport:
        """检查工具是否存在且可用"""
        errors = []
        suggestions = []

        if tool_name not in self.registry.tools:
            errors.append(f"工具 '{tool_name}' 未注册")
            # 生成相似工具建议
            similar = self._find_similar_tools(tool_name)
            if similar:
                suggestions.append(
                    f"您是想用 '{similar[0]}' 吗?"
                )

        elif not self.registry.is_tool_available(tool_name):
            errors.append(f"工具 '{tool_name}' 已禁用")

        if errors:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=errors,
                warnings=[],
                suggestion=suggestions[0] if suggestions else None
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

    def _find_similar_tools(self, name: str) -> List[str]:
        """使用编辑距离找相似工具名"""
        from difflib import get_close_matches
        available = [n for n in self.registry.tools.keys()
                     if self.registry.is_tool_available(n)]
        return get_close_matches(name, available, n=1, cutoff=0.6)

3. 参数类型和范围检查

使用Pydantic进行参数类型和范围验证的实现如下：

from pydantic import BaseModel, ValidationError, field_validator
from typing import Optional, Any

class FileReadInput(BaseModel):
    """文件读取工具的输入"""
    file_path: str
    encoding: str = "utf-8"
    max_lines: Optional[int] = None

    @field_validator("file_path")
    @classmethod
    def validate_file_path(cls, v):
        if not isinstance(v, str) or len(v) == 0:
            raise ValueError("file_path 必须是非空字符串")
        return v

    @field_validator("max_lines")
    @classmethod
    def validate_max_lines(cls, v):
        if v is not None and (v < 1 or v > 100000):
            raise ValueError("max_lines 必须在 1-100000 之间")
        return v

class ParameterValidator:
    """参数验证器"""

    def __init__(self, registry: ToolRegistry):
        self.registry = registry

    def validate(self, tool_name: str, parameters: dict) -> ValidationReport:
        """验证工具参数"""
        schema = self.registry.get_tool_schema(tool_name)
        if schema is None:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"无法找到工具 {tool_name} 的 schema"],
                warnings=[]
            )

        errors = []
        warnings = []

        try:
            # 使用 Pydantic 验证参数
            validated = schema(**parameters)
            return ValidationReport(
                result=ValidationResult.PASS,
                errors=[],
                warnings=[]
            )
        except ValidationError as e:
            for error in e.errors():
                field = ".".join(str(x) for x in error["loc"])
                msg = error["msg"]
                errors.append(f"参数 {field}: {msg}")

        if errors:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=errors,
                warnings=warnings,
                suggestion="请检查工具的参数定义"
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=warnings
        )

4. 业务逻辑检查

某些检查超越类型系统，涉及业务规则：

class BusinessLogicValidator:
    """业务逻辑验证"""

    def __init__(self):
        self.config = {
            "max_api_calls_per_minute": 100,
            "max_file_size_mb": 10,
            "allowed_domains": ["example.com", "api.service.com"],
        }

    def validate_api_call_frequency(self, tool_name: str,
                                    call_history: List[float]) -> ValidationReport:
        """检查 API 调用频率是否超过限制"""
        import time
        now = time.time()
        # 计算最近 1 分钟的调用数
        recent_calls = [t for t in call_history if now - t < 60]

        limit = self.config["max_api_calls_per_minute"]
        if len(recent_calls) >= limit:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"工具 {tool_name} 在 1 分钟内调用已达 {limit} 次限制"],
                warnings=[],
                suggestion="请稍后再试"
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

    def validate_file_access(self, file_path: str) -> ValidationReport:
        """检查文件访问权限和大小"""
        import os
        errors = []

        # 检查文件存在性
        if not os.path.exists(file_path):
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"文件不存在: {file_path}"],
                warnings=[]
            )

        # 检查文件大小
        file_size_mb = os.path.getsize(file_path) / (1024 * 1024)
        max_size = self.config["max_file_size_mb"]
        if file_size_mb > max_size:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"文件大小 {file_size_mb:.2f}MB 超过限制 {max_size}MB"],
                warnings=[]
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

    def validate_http_request(self, url: str) -> ValidationReport:
        """检查 HTTP 请求的目标合法性"""
        from urllib.parse import urlparse
        parsed = urlparse(url)
        domain = parsed.netloc

        if domain not in self.config["allowed_domains"]:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"不允许访问域名: {domain}"],
                warnings=[],
                suggestion=f"允许的域名: {self.config['allowed_domains']}"
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

5. 安全与权限检查

安全与权限检查的实现方式如下：

from enum import Enum

class Permission(Enum):
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    EXECUTE = "execute"

class SecurityValidator:
    """安全与权限检查"""

    def __init__(self, user_permissions: set):
        self.user_permissions = user_permissions

    def validate_tool_permission(self, tool_name: str) -> ValidationReport:
        """检查用户是否有权调用该工具"""
        # 定义工具所需权限
        tool_permissions = {
            "read_file": {Permission.READ},
            "write_file": {Permission.WRITE},
            "delete_file": {Permission.DELETE},
            "execute_command": {Permission.EXECUTE},
        }

        if tool_name not in tool_permissions:
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"未知工具: {tool_name}"],
                warnings=[]
            )

        required = tool_permissions[tool_name]
        if not required.issubset(self.user_permissions):
            missing = required - self.user_permissions
            return ValidationReport(
                result=ValidationResult.FAIL,
                errors=[f"缺少权限: {', '.join(p.value for p in missing)}"],
                warnings=[]
            )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

    def validate_parameter_injection(self, parameters: dict) -> ValidationReport:
        """检查参数是否包含注入攻击"""
        dangerous_patterns = [
            "rm -rf",
            "DROP TABLE",
            "<script>",
            "${jndi:",
        ]

        for key, value in parameters.items():
            if isinstance(value, str):
                for pattern in dangerous_patterns:
                    if pattern in value:
                        return ValidationReport(
                            result=ValidationResult.FAIL,
                            errors=[f"参数 {key} 包含危险模式: {pattern}"],
                            warnings=[],
                            suggestion="检查参数是否被恶意注入"
                        )

        return ValidationReport(
            result=ValidationResult.PASS,
            errors=[],
            warnings=[]
        )

6. 完整的质量门控引擎

整合六层验证逻辑的完整质量门控引擎如下：

class QualityGate:
    """完整的质量门控引擎"""

    def __init__(self, registry: ToolRegistry, user_permissions: set):
        self.format_validator = FormatValidator()
        self.existence_validator = ExistenceValidator(registry)
        self.parameter_validator = ParameterValidator(registry)
        self.business_validator = BusinessLogicValidator()
        self.security_validator = SecurityValidator(user_permissions)

    def validate_tool_call(self, tool_call, call_history: List[float] = None):
        """执行完整的质量门控"""
        results = []

        # 门控 1: 格式检查
        r1 = self.format_validator.validate_tool_use_block(tool_call)
        results.append(("格式检查", r1))
        if r1.result == ValidationResult.FAIL:
            return self._report_failure(results)

        # 门控 2: 工具存在性
        r2 = self.existence_validator.validate(tool_call.name)
        results.append(("工具存在性", r2))
        if r2.result == ValidationResult.FAIL:
            return self._report_failure(results)

        # 门控 3: 参数验证
        r3 = self.parameter_validator.validate(
            tool_call.name, tool_call.input
        )
        results.append(("参数验证", r3))
        if r3.result == ValidationResult.FAIL:
            return self._report_failure(results)

        # 门控 4: 业务逻辑(对文件类参数执行业务规则)
        file_path = (tool_call.input or {}).get("file_path")
        if file_path:
            r4 = self.business_validator.validate_file_access(file_path)
            results.append(("业务逻辑", r4))
            if r4.result == ValidationResult.FAIL:
                return self._report_failure(results)

        # 门控 5: 权限检查
        r5 = self.security_validator.validate_tool_permission(tool_call.name)
        results.append(("权限检查", r5))
        if r5.result == ValidationResult.FAIL:
            return self._report_failure(results)

        # 门控 6: 参数注入检测
        r6 = self.security_validator.validate_parameter_injection(
            tool_call.input or {}
        )
        results.append(("注入检测", r6))
        if r6.result == ValidationResult.FAIL:
            return self._report_failure(results)

        return {
            "passed": True,
            "details": results
        }

    def _report_failure(self, results):
        return {
            "passed": False,
            "details": results
        }

质量门控小结

质量门控通过多层验证机制：

• 格式检查：确保结构完整

• 存在性检查：验证工具已注册

• 参数检查：类型和范围验证

• 业务逻辑：应用特定规则

• 权限检查：阻止越权调用

• 注入检测：识别参数中的注入攻击

有效降低执行失败、资源泄露和安全风险，是智能体可靠性的关键保障。

7.3.3 规划-生成-评估三角色分离

在传统的单一模型智能体中，智能体需要同时完成三项职责：分析任务、生成解决方案、评估结果。这导致一个常见问题：自我确认偏差——生成器很难公正地评估自己的工作。

Anthropic 提出的 规划-生成-评估(Planner-Generator-Evaluator, PGE)模式 通过角色分离和潜在的不同模型配置，显著提升输出质量。

三角色的职责划分

规划者(Planner)

• 分析任务，理解用户意图

• 将复杂目标分解为子任务

• 创建执行计划和验证标准

• 使用较高的推理预算（可选使用Adaptive Thinking）

生成者(Generator)

• 按照计划逐步执行

• 调用工具、处理API响应

• 产生原始输出

• 优化吞吐量和成本（使用轻量级模型配置）

评估者(Evaluator)

• 独立检查生成器的输出

• 验证是否满足规划阶段的要求

• 判断是否需要修正或重新生成

• 必须使用独立的提示和上下文 （避免自我确认偏差）

架构流程

质量门架构采用三角验证模式，通过独立的评估者对生成的结果进行验证：

关键设计原则

1. 评估者的独立性

评估者必须与生成者完全隔离，这是PGE模式的核心：

class PGEAgent:
    """规划-生成-评估三角色分离"""

    def __init__(self, planner_model, generator_model, evaluator_model):
        self.planner = planner_model       # 高推理预算
        self.generator = generator_model   # 优化吞吐量
        self.evaluator = evaluator_model   # 独立验证

    async def execute(self, user_request: str) -> ExecutionResult:
        """执行PGE流程"""

        # 第一步:规划
        print("[Phase 1] Planning...")
        plan = await self.planner.analyze_and_plan(user_request)
        print(f"Plan: {plan.steps}")

        # 第二步:生成
        print("[Phase 2] Generating...")
        output = await self.generator.execute_plan(plan)

        # 第三步:评估 - 关键:使用独立的评估者
        print("[Phase 3] Evaluating...")
        evaluation = await self.evaluate_output(
            output=output,
            plan=plan,
            evaluator=self.evaluator  # 独立的模型
        )

        if evaluation.passed:
            return output

        # 失败时的重试逻辑
        if evaluation.retry_generator:
            # 让生成者修正(基于评估反馈)
            output_v2 = await self.generator.refine(
                original_output=output,
                feedback=evaluation.feedback,
                plan=plan
            )
            evaluation_v2 = await self.evaluate_output(
                output=output_v2,
                plan=plan,
                evaluator=self.evaluator
            )
            if evaluation_v2.passed:
                return output_v2

        if evaluation.retry_planner:
            # 让规划者重新规划(当多次生成都失败时)
            plan_v2 = await self.planner.replan(
                user_request=user_request,
                failure_reason=evaluation.failure_analysis
            )
            # 重新执行生成
            return await self.execute_with_plan(plan_v2)

        raise ExecutionError(f"Failed after retries: {evaluation.feedback}")

    async def execute_with_plan(self, plan) -> ExecutionResult:
        """跳过规划阶段,按给定计划执行生成与评估(重新规划后的入口)"""
        output = await self.generator.execute_plan(plan)
        evaluation = await self.evaluate_output(
            output=output, plan=plan, evaluator=self.evaluator
        )
        if evaluation.passed:
            return output
        raise ExecutionError(f"Replanned execution failed: {evaluation.feedback}")

    async def evaluate_output(
        self,
        output: str,
        plan: Plan,
        evaluator: Model
    ) -> EvaluationResult:
        """独立评估输出"""

        # 构造评估提示 - 这是关键:评估者不应该看到生成过程
        evaluation_prompt = f"""
Please evaluate the following output against the original requirements.

Original Requirements:
{plan.requirements}

Verification Criteria:
{plan.verification_criteria}

Output to Evaluate:
{output}

Answer:
1. Does the output meet ALL requirements? (yes/no)
2. Are there any errors or issues?
3. If failed, what specific improvements are needed?
4. Confidence level: (high/medium/low)
"""

        evaluation_text = await evaluator.infer(evaluation_prompt)
        return self._parse_evaluation(evaluation_text, plan)

    def _parse_evaluation(self, eval_text: str, plan: Plan) -> EvaluationResult:
        """解析评估结果"""
        # 简化实现:检查评估是否包含肯定
        passed = "yes" in eval_text.lower()

        # 判断是重试生成还是重新规划
        retry_generator = "small" in eval_text.lower() or "adjust" in eval_text.lower()
        retry_planner = "fundamental" in eval_text.lower() or "rethink" in eval_text.lower()

        return EvaluationResult(
            passed=passed,
            retry_generator=retry_generator,
            retry_planner=retry_planner,
            feedback=eval_text
        )

2. 模型配置的灵活性

虽然理想情况下三个角色使用不同的模型，但也可以使用同一模型的不同配置：

# 配置示例
# 注意：Opus 4.7 起已移除 temperature/top_p/top_k，传入会返回 400；
# 推理深度改用 adaptive thinking + effort 控制。Sonnet 4.6 仍接受采样参数。
PLANNER_CONFIG = {
    "model": "claude-opus-4-7",  # 最强模型
    "adaptive_thinking": True,   # 高推理预算
    "effort": "high",            # 取代 temperature 的确定性/深度旋钮
}

GENERATOR_CONFIG = {
    "model": "claude-sonnet-4-6", # 平衡成本/性能
    "adaptive_thinking": False,   # 轻量级,专注执行
    "temperature": 0.1,           # 4.6 仍支持采样参数
}

EVALUATOR_CONFIG = {
    "model": "claude-opus-4-7",  # 同样强大的模型
    # 但使用完全不同的上下文和提示
    "adaptive_thinking": True,
    "effort": "high",            # 同上：4.7 不接受 temperature
}

3. 避免自我确认偏差

确保评估者不会无意中受到生成者的影响：

警告：LLM-as-Judge 的隐性偏差 独立评估器虽然能减少自我确认偏差，但仍需防范 LLM 本身的评估偏见：

• 位置偏差(Position Bias)：倾向于偏好开头或结尾的内容

• 长度偏好(Length Preference)：可能无意识地倾向较长或较短的答案

• 风格偏好(Style Bias)：倾向支持与其训练数据相似的表达风格

建议采用多评估者投票制或定期人工抽样验证来规避这些风险。

class ObjectiveEvaluator:
    """客观评估器"""

    def __init__(self, evaluator_model):
        self.evaluator = evaluator_model

    async def evaluate(self, output: str, criteria: str) -> bool:
        """评估输出是否符合标准"""

        # 关键:评估提示必须是通用的,不提及"生成器"
        # 而是直接针对需求和输出

        prompt = f"""
Review this output against the criteria:

Criteria:
{criteria}

Output:
{output}

Does it meet the criteria? List any gaps or issues.
"""
        # 注意:不包含任何关于"生成过程"的信息
        evaluation = await self.evaluator.infer(prompt)
        return self._judge_pass_fail(evaluation)

    def _judge_pass_fail(self, eval_text: str) -> bool:
        """判断是否通过 - 使用保守的标准"""
        # 如果有任何怀疑,就认为失败
        has_issues = len([l for l in eval_text.split('\n') if l.strip()]) > 1
        return not has_issues

    # 警告: LLM-as-Judge 偏差风险
    # 评估器可能存在以下隐性偏差:
    # 1. 位置偏差: 更倾向评估靠前或靠后的内容
    # 2. 长度偏好: 倾向偏好较长或较短的输出
    # 3. Self-Preference: 倾向支持与其训练数据相似的风格
    # 建议使用多个独立评估者或人工抽样验证来规避这些风险

实际应用案例

在一个代码审查智能体中应用PGE模式：

1. 规划者:分析代码,生成审查清单
   - 安全检查点
   - 性能考量
   - 可维护性标准

2. 生成者:执行审查
   - 逐行检查
   - 查找问题
   - 生成初稿反馈

3. 评估者:独立验证
   - 反馈是否涵盖了所有清单项
   - 是否有遗漏的严重问题
   - 反馈是否具体可行

PGE vs 单一模型的对比

特性	单一模型	PGE模式
自我确认偏差	高	低（独立评估）
错误率	基线	显著下降
成本	基线	有所增加（多次模型调用）
调试难度	中等	低（清晰的问题定位）

规划-生成-评估小结

PGE模式的价值在于：

• 隔离关注点：三个角色各司其职

• 独立验证：评估者的独立性确保发现问题

• 灵活扩展：可根据需求调整模型选择

• 可调试性：清晰的故障路径便于问题定位