# 7.3 输出质量门控与过滤

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

## 7.3.1 问题定义

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

* **参数错误**：工具接收非法数据导致执行失败
* **资源泄露**：过度调用导致费用飙升或系统宕机
* **格式异常**：API 调用失败导致 智能体循环重试
* **安全漏洞**：注入攻击、不合理的权限操作

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

## 7.3.2 门控层次结构

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

```mermaid
graph TD
    A["原始响应"] -->|检查格式| B["格式检查"]
    B -->|验证存在| C["工具存在性"]
    C -->|检查类型| D["参数验证"]
    D -->|检查逻辑| E["业务逻辑"]
    E -->|检查权限| F["安全检查"]
    F -->|通过所有检查| G["执行工具"]

    B -.->|失败| H["拒绝执行"]
    C -.->|失败| H
    D -.->|失败| H
    E -.->|失败| H
    F -.->|失败| H

    style A fill:#ffebee
    style B fill:#fff9c4
    style C fill:#fff9c4
    style D fill:#fff9c4
    style E fill:#fff9c4
    style F fill:#fff9c4
    style G fill:#c8e6c9
    style H fill:#ffcdd2
```

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

### 1. 基础格式检查

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

```python
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. 工具存在性检查

验证工具是否已注册：

```python
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进行参数类型和范围验证的实现如下：

```python
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. 业务逻辑检查

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

```python
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. 安全与权限检查

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

```python
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. 完整的质量门控引擎

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

```python
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: 业务逻辑
        r4 = self.business_validator.validate_file_access("")
        results.append(("业务逻辑", r4))

        # 门控 5: 权限检查
        r5 = self.security_validator.validate_tool_permission(tool_call.name)
        results.append(("权限检查", r5))
        if r5.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)**

* 独立检查生成器的输出
* 验证是否满足规划阶段的要求
* 判断是否需要修正或重新生成
* **必须使用独立的提示和上下文** （避免自我确认偏差）

### 架构流程

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

```mermaid
graph LR
    A["用户请求"] -->|分析意图| B["<b>规划者</b><br/>生成执行计划"]
    B -->|步骤序列<br/>验证标准| C["<b>生成者</b><br/>执行任务"]
    C -->|原始输出| D["<b>评估者</b><br/>独立验证"]
    D -->|是否通过| E{结果判断}
    E -->|通过| F["返回结果"]
    E -->|失败| G["反馈给生成者"]
    G -->|修正方案| C
    E -->|多次失败| H["反馈给规划者"]
    H -->|重新规划| B

    style A fill:#e3f2fd
    style B fill:#fff3e0
    style C fill:#e8f5e9
    style D fill:#f3e5f5
    style E fill:#fce4ec
    style F fill:#c8e6c9
    style G fill:#ffcdd2
    style H fill:#ffcdd2
```

### 关键设计原则

**1. 评估者的独立性**

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

```python
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 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. 模型配置的灵活性**

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

```python
# 配置示例
PLANNER_CONFIG = {
    "model": "claude-opus-4-7",  # 最强模型
    "adaptive_thinking": True,   # 高推理预算
    "temperature": 0.3,           # 确定性规划
}

GENERATOR_CONFIG = {
    "model": "claude-sonnet-4-6", # 平衡成本/性能
    "adaptive_thinking": False,   # 轻量级,专注执行
    "temperature": 0.1,           # 确定性执行
}

EVALUATOR_CONFIG = {
    "model": "claude-opus-4-7",  # 同样强大的模型
    # 但使用完全不同的上下文和提示
    "adaptive_thinking": True,
    "temperature": 0.2,          # 保持客观
}
```

**3. 避免自我确认偏差**

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

> **警告：LLM-as-Judge 的隐性偏差** 独立评估器虽然能减少自我确认偏差，但仍需防范 LLM 本身的评估偏见：
>
> * **位置偏差(Position Bias)**：倾向于偏好开头或结尾的内容
> * **长度偏好(Length Preference)**：可能无意识地倾向较长或较短的答案
> * **风格偏好(Style Bias)**：倾向支持与其训练数据相似的表达风格
>
> 建议采用多评估者投票制或定期人工抽样验证来规避这些风险。

```python
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模式的价值在于：

* **隔离关注点**：三个角色各司其职
* **独立验证**：评估者的独立性确保发现问题
* **灵活扩展**：可根据需求调整模型选择
* **可调试性**：清晰的故障路径便于问题定位


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://yeasy.gitbook.io/harness_engineering_guide/di-er-bu-fen-harness-he-xin-zi-xi-tong/07_model_integration/7.3_quality_gate.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.