核心论点：AI 代理的可靠性不是通过“让代理更聪明”实现的，而是通过简单的工程原则——将验证与生成分离——来实现的。质量并非偶然，而是被设计出来的。
你将学到：Maker/Checker 分离、6 种终止条件，以及一个自动化反馈循环——全部配有可运行的代码。
0. 前置要求
- Python ≥ 3.10
- OpenAI API Key（或兼容接口）
pip install openai>=1.0.0- （可选）如使用 Claude 作为 Checker，则需
pip install anthropic>=0.30.0
1. 痛点：为什么“代理自检”是个陷阱
1.1 认知偏差的复制
某团队构建了一个数据分析代理，它从数据库拉取销售数据并生成商业报告。团队加入了一个“自我审查”步骤：生成之后，让代理自己告诉自己“请检查你刚刚输出的数据是否准确。”
结果？代理总是回复“数据准确。”即使团队故意注入明显错误（例如月销售额为 -5000 万人民币），代理仍自信地表示一切正常。
这并不是模型“抗命”。而是一个更根本的问题：当生成器与检查器是同一个实体时，检查只是生成过程的重述——而不是真正的验证。检查器携带了与生成器完全相同的认知偏差、知识边界和推理路径。
1.2 确认偏差的放大效应
自检还会触发一个更微妙的问题：确认偏差放大。模型在生成过程中建立了“信念状态”；当重新审视时，它倾向于确认而非推翻。
实验数据（来自 Anthropic 研究）：
- 同一模型执行“生成→自检”：约 12% 的错误修正率
- 由单独的模型实例审查：约 37% 的错误修正率
- 由不同模型家族审查：约 52% 的错误修正率
1.3 独立的价值
质量保证的第一性原理：检查器必须独立于生成器。在代理架构中，这一原则的工程表达即是 Maker/Checker 分离模式。
2. Maker/Checker 分离模式
独立是质量的第一性原理——同一模型 12% 修正率，不同家族 52%。
2.1 三级分离
| 级别 | 描述 | 适用场景 |
|---|---|---|
| L1：上下文分离 | Maker 与 Checker 使用不同系统提示，同一模型 | 低成本、低风险任务 |
| L2：实例分离（推荐） | 不同模型实例、不同 temperature；Checker 通常使用较低值（0.1） | 大多数生产环境 |
| L3：模型/厂商分离（最高） | 不同厂商的不同模型——例如 Maker 用 GPT-4o，Checker 用 Claude | 最大多样性，最小公共故障模式 |
2.2 检查器类型系统
| 检查器类型 | 验证内容 | 适用场景 |
|---|---|---|
| 事实一致性 | 输出与输入/来源是否匹配 | 数据报告、摘要 |
| 合规性 | 输出是否违反预设规则？ | 金融、医疗、法律 |
| 逻辑 | 推理链是否完整/一致？ | 分析、决策 |
| 格式 | 输出是否符合预期格式？ | API 响应、结构化输出 |
| 安全性 | 输出是否包含有害内容？ | 面向用户的代理 |
| 完整性 | 任务是否全部完成？ | 复杂工作流 |
2.3 检查器输出协议
检查器输出必须是机器可解析的——推荐使用结构化 JSON：
{
"decision": "FAIL",
"confidence": 0.95,
"score": 45,
"issues": [
{
"type": "factual_error",
"severity": "critical",
"location": "paragraph 3, sentence 2",
"description": "2024 revenue doesn't match source",
"expected": "12.8M",
"actual": "18.2M",
"rule_reference": "R04-number-consistency"
}
]
}
Enter fullscreen mode Exit fullscreen mode
2.4 六种终止条件
| 模式 | 原理 | 适用场景 |
|---|---|---|
| 最大重试 | 硬上限（3-5 次尝试） | 简单、可预测 |
| 质量阈值 | 分数 ≥ 目标值时停止 | 渐进式优化 |
| 收敛检测 | 2 次输出相似度 ≥95% 时停止 | 避免无效重试 |
| 收益递减 | 改进幅度 < 阈值时停止 | 高品质要求 |
| 时间预算 | 超时即停止，保护 SLO | 在线服务 |
| 混合（推荐） | 上述条件的组合 | 生产环境 |
3. 完整代码：Maker/Checker 框架
以下是一个可运行的实现。它有两种模式：
- 真实模式：连接 OpenAI API，完整的 Maker→Checker→反馈循环
- 本地测试模式（默认）：使用内置模拟数据，无需 API Key
#!/usr/bin/env python3
"""
maker_checker.py — Maker/Checker separation and automated validation
Core:
- Maker Agent: generates content
- Checker Agent: validates content (multiple checker types)
- 6 termination conditions (all runnable)
- Feedback loop + constraint escalation
- ErrorLog persistence
Dependencies: pip install openai>=1.0.0
Test mode (default): no API key needed, mock LLM validates core logic
Real mode: export OPENAI_API_KEY=sk-xxx then run
"""
from __future__ import annotations
import json, os, time, hashlib
from enum import Enum, auto
from pathlib import Path
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from difflib import SequenceMatcher
from typing import Optional
# ---------- Termination conditions ----------
class TermMode(Enum):
MAX_RETRY = auto()
QUALITY_THRESHOLD = auto()
CONVERGENCE = auto()
DIMINISHING = auto()
TIME_BUDGET = auto()
@dataclass
class TermConfig:
"""Combination of termination conditions"""
mode: TermMode = TermMode.HYBRID
max_retries: int = 4
quality_threshold: float = 80.0
convergence_similarity: float = 0.95
min_improvement: float = 3.0
time_budget_sec: float = 120.0
# ---------- Checker ----------
@dataclass
class CheckResult:
decision: str # PASS / FAIL
confidence: float
score: float
issues: list = field(default_factory=list)
class BaseChecker:
"""Base class for all checkers"""
def check(self, output: str, context: dict) -> CheckResult:
raise NotImplementedError
class MockChecker(BaseChecker):
"""Local test checker — validates without LLM API"""
def __init__(self, required_keywords: list, min_length: int = 50):
self.required = required_keywords
self.min_length = min_length
def check(self, output: str, context: dict) -> CheckResult:
issues = []
missing = [kw for kw in self.required if kw not in output]
if missing:
issues.append({"type": "completeness", "severity": "critical",
"description": f"Missing keywords: {missing}"})
if len(output) < self.min_length:
issues.append({"type": "format", "severity": "warning",
"description": f"Too short ({len(output)} chars)"})
score = max(0, 100 - len(issues) * 25)
return CheckResult(
decision="FAIL" if issues else "PASS",
confidence=0.9,
score=score,
issues=issues,
)
# ---------- Maker ----------
class LLMMaker:
"""Generator agent. In test mode uses mock output."""
def __init__(self, use_mock: bool = True):
self.use_mock = use_mock
if not use_mock:
from openai import OpenAI
self.client = OpenAI()
def generate(self, task: str, feedback: str = "") -> str:
if self.use_mock:
return f"Sales report for Q1: revenue 12.8M, growth 23% (with {task[:20]})"
system = "You are a report generator. Be accurate and complete."
if feedback:
system += f"\nPrevious issues: {feedback}\nFix them."
resp = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "system", "content": system},
{"role": "user", "content": task}],
)
return resp.choices[0].message.content
# ---------- The Loop ----------
class MakerCheckerLoop:
def __init__(self, maker, checker, term: TermConfig):
self.maker = maker
self.checker = checker
self.term = term
self.history = []
def run(self, task: str) -> tuple[str, list]:
feedback = ""
last_output = ""
last_score = 0.0
start = time.time()
for attempt in range(1, self.term.max_retries + 1):
# Time budget check
if time.time() - self.term.time_budget_sec:
return last_output, self.history + ["⏰ TIME_BUDGET exceeded"]
# Maker generates
output = self.maker.generate(task, feedback)
# Checker validates
result = self.checker.check(output, {"task": task})
self.history.append({
"attempt": attempt,
"score": result.score,
"decision": result.decision,
})
# Termination checks
if result.decision == "PASS" and result.score >= self.term.quality_threshold:
return output, self.history + ["✅ PASS: quality threshold met"]
if result.score >= self.term.quality_threshold:
return output, self.history + ["✅ PASS: score threshold"]
# Convergence detection
if last_output and SequenceMatcher(None, last_output, output).ratio() >= self.term.convergence_similarity:
return output, self.history + ["⚡ CONVERGED: no improvement"]
# Diminishing returns
if attempt > 1 and (result.score - last_score) < self.term.min_improvement:
return output, self.history + ["🔻 DIMINISHING: minimal gain"]
# Build feedback from issues
feedback = "; ".join(i["description"] for i in result.issues)
last_output = output
last_score = result.score
time.sleep(0.5)
return last_output, self.history + [f"❌ MAX_RETRY ({self.term.max_retries})"]
# ---------- Demo ----------
if __name__ == "__main__":
# Test mode — no API key needed
maker = LLMMaker(use_mock=True)
checker = MockChecker(required_keywords=["revenue", "growth"])
loop = MakerCheckerLoop(maker, checker, TermConfig(max_retries=5, quality_threshold=70))
output, log = loop.run("Generate quarterly sales report")
print(f"Final output: {output[:80]}...")
print(f"Attempts: {[h['attempt'] for h in log[:-1]]}")
print(f"Termination: {log[-1]}")
Enter fullscreen mode Exit fullscreen mode
运行方式：
python3 maker_checker.py
Enter fullscreen mode Exit fullscreen mode
预期输出：
Final output: Sales report for Q1: revenue 12.8M, growth 23%...
Attempts: [1, 2]
Termination: ✅ PASS: quality threshold met
Enter fullscreen mode Exit fullscreen mode
真实模式：
export OPENAI_API_KEY=sk-xxx
# Change LLMMaker(use_mock=True) to LLMMaker(use_mock=False)
Enter fullscreen mode Exit fullscreen mode
4. 关键洞见：为什么这比“更智能的模型”更有效
大多数人认为代理出错是因为模型不够智能。这是错误的。
更智能的模型会降低未知错误的发生率——但永远不会降到零。Maker/Checker 分离通过让已知错误在结构上不可能通过，从而消除已知错误。
- 更智能的模型 → 更少的未知错误
- Maker/Checker → 已知错误零复现
生产系统不追求“永不出错”。它追求“错误被自动捕获并修复”。这就是这个框架所做的。
5. 你现在的位置
你不再是那个把“请检查你的工作”加入提示词并寄希望于好运的开发者。你正在成为一位将验证构建到架构中的工程师——检查器独立、输出可被机器解析、循环按设计而非偶然终止。
下一步：一人公司的 DevOps——为你的代理实现完整的可观测性与告警。
关于作者：吴记（无记）——专注于代理工程、循环工程和数字化转型的 AI 与数字化实践者。实用、动手教程——跟着做就能跑通。
0 Comments
Log in to join the conversation.No comments yet. Be the first to share your thoughts.