第 39 课 | 自我反思与质量控制：Agent 输出的最后一道防线

自动检查、自我纠错、交叉验证——三大机制确保 Agent 输出可靠，让决策者敢用。质量控制不是可选项，而是生产级 Agent 系统的必选项。

---

一、业务痛点：Agent 犯了错，谁来兜底？

你花了几周搭建的多 Agent 系统，老板问「我们该不该进入智能手表市场」，系统给出了分析报告。但——

• 如果市场规模的数据是错的呢？
• 如果竞品分析遗漏了关键对手呢？
• 如果风险评估过于乐观呢？

在真实的商业场景中，Agent 的错误可能导致数百万的决策失误。 质量控制不是可选项，而是必选项。

---

二、三大质量控制机制

---

三、机制 1：自我反思

Agent 执行后自动检查输出质量，不达标则重试：

第 1 次尝试：输出质量评分 45/100 ❌
  → 问题：缺少「市场规模」章节、数据模糊（「较大」「较快」）
  → 改进：补充具体数字、使用结构化格式

第 2 次尝试：输出质量评分 72/100 ✅
  → 通过！存在小问题但可接受

3.1 五维质量检查

维度	权重	检查内容	扣分项
完整性	25%	是否包含所有要求的章节	每缺少一个章节 -15 分
准确性	25%	数据是否合理、无矛盾	数字矛盾 -10 分
结构性	20%	输出格式是否符合要求	无标题/加粗 -10 分
具体性	20%	是否有具体数字和案例	模糊词汇（较多/较快）-5 分
一致性	10%	前后数据是否一致	单位混用 -5 分

3.2 实现

class QualityChecker:
    """输出质量检查器"""
    
    REQUIRED_SECTIONS = {
        "market": ["市场规模", "增长趋势", "用户画像"],
        "competitor": ["竞品矩阵", "SWOT分析", "差异化机会"],
        "risk": ["风险列表", "概率评估", "缓解策略"],
        "report": ["执行摘要", "市场分析", "竞品格局", "风险评估", "战略建议"],
    }
    
    VAGUE_TERMS = ["较大", "较快", "很多", "一些", "大概", "可能", "左右"]
    
    def __init__(self, output_type: str, required_sections: list[str] = None):
        self.output_type = output_type
        self.required_sections = required_sections or self.REQUIRED_SECTIONS.get(output_type, [])
    
    def check(self, output: str) -> dict:
        """执行五维质量检查"""
        score = 100
        
        # 1. 完整性检查
        completeness_deduction = 0
        for section in self.required_sections:
            if section not in output:
                completeness_deduction += 15
        score -= completeness_deduction
        
        # 2. 具体性检查
        vagueness_count = sum(1 for term in self.VAGUE_TERMS if term in output)
        specificity_deduction = min(vagueness_count * 5, 20)
        score -= specificity_deduction
        
        # 3. 结构性检查
        structure_deduction = 0
        if "##" not in output:
            structure_deduction += 10
        if "**" not in output and len(output) > 500:
            structure_deduction += 5
        score -= structure_deduction
        
        return {
            "score": max(score, 0),
            "passed": score >= 60,
            "details": {
                "completeness": f"缺少 {completeness_deduction // 15} 个章节",
                "specificity": f"发现 {vagueness_count} 个模糊词",
                "structure": "格式不规范" if structure_deduction > 0 else "格式规范",
            },
        }


class SelfReflectingAgent:
    """带自我反思的 Agent"""
    
    def __init__(self, agent, max_retries: int = 3):
        self.agent = agent
        self.max_retries = max_retries
        self.checker = QualityChecker("report")
    
    def run(self, task: str) -> str:
        """执行任务，自动重试直到达标"""
        for attempt in range(1, self.max_retries + 1):
            output = self.agent.run(task)
            result = self.checker.check(output)
            
            print(f"  第 {attempt} 次尝试: 评分 {result['score']}/100")
            
            if result["passed"]:
                print(f"  ✅ 通过！")
                return output
            
            print(f"  ❌ 不通过: {result['details']}")
            
            # 改进提示
            if attempt < self.max_retries:
                task = self._improve_task(task, result)
        
        return output  # 返回最后一次结果
    
    def _improve_task(self, task: str, result: dict) -> str:
        """根据检查结果改进任务描述"""
        improvements = []
        details = result["details"]
        
        if "缺少" in details.get("completeness", ""):
            improvements.append("确保包含所有要求的章节")
        if "模糊词" in details.get("specificity", ""):
            improvements.append("使用具体数字代替模糊词汇（如'较大''较快'）")
        if "不规范" in details.get("structure", ""):
            improvements.append("使用 ## 标题和 **加粗** 格式化输出")
        
        return task + "\n\n改进要求：\n" + "\n".join(f"- {i}" for i in improvements)

---

四、机制 2：交叉校验

一个 Agent 分析，另一个 Agent 审核——模拟「四眼原则」：

分析师 Agent → 生成报告
审核员 Agent → 检查报告
  ├── 数字一致性：报告中 156 万 vs 前文 189 万 → 矛盾！
  ├── 完整性：缺少「风险评估」章节 → 不完整！
  └── 具体性：5 处模糊词汇 → 不够具体！
审核员 → 反馈问题清单
分析师 → 逐项修正
审核员 → 再次检查 → 通过

class CrossValidator:
    """交叉校验器"""
    
    def __init__(self, analyst, reviewer):
        self.analyst = analyst
        self.reviewer = reviewer
    
    def validate(self, task: str, max_rounds: int = 3) -> str:
        """执行交叉校验"""
        output = self.analyst.run(task)
        
        for round_num in range(1, max_rounds + 1):
            review = self.reviewer.run(f"""
            请审核以下分析报告，检查：
            1. 数字一致性（前后数据是否矛盾）
            2. 完整性（是否缺少关键章节）
            3. 具体性（是否有模糊表述）
            
            报告：
            {output}
            
            如果发现问题，请列出具体问题。如果没问题，回复「PASS」。
            """)
            
            if "PASS" in review.upper():
                print(f"  交叉校验第 {round_num} 轮: ✅ 通过")
                return output
            
            print(f"  交叉校验第 {round_num} 轮: ❌ 发现问题")
            print(f"  审核意见: {review[:200]}...")
            
            # 修正
            output = self.analyst.run(f"""
            请根据以下审核意见修正报告：
            
            审核意见：
            {review}
            
            原始报告：
            {output}
            """)
        
        return output

交叉校验的优势：

• 避免「既当运动员又当裁判」的盲点
• 不同 Agent 有不同的思维模式，能发现对方忽略的问题
• 模拟真实商业中的「四眼原则」

---

五、机制 3：质量评分看板

class QualityDashboard:
    """质量评分看板"""
    
    def __init__(self):
        self.results = []
    
    def add_result(self, agent_name: str, score: int, passed: bool, issues: list[str]):
        self.results.append({
            "agent": agent_name,
            "score": score,
            "passed": passed,
            "issues": issues,
        })
    
    def display(self):
        """显示质量看板"""
        print("\n" + "=" * 60)
        print("  质量评分看板")
        print("=" * 60)
        print(f"{'Agent':<20} {'评分':<8} {'通过':<6} {'问题数':<8}")
        print("-" * 60)
        
        for r in self.results:
            status = "✅" if r["passed"] else "❌"
            print(f"{r['agent']:<20} {r['score']:<8} {status:<6} {len(r['issues']):<8}")
        
        avg_score = sum(r["score"] for r in self.results) / len(self.results)
        all_passed = all(r["passed"] for r in self.results)
        
        print("-" * 60)
        print(f"平均分: {avg_score:.1f}")
        print(f"整体通过: {'✅ 是' if all_passed else '❌ 否'}")
        print("=" * 60)

输出示例：

质量评分看板
════════════════════════════════════════════════════════
Agent                评分      通过    问题数
────────────────────────────────────────────────────────
MarketAgent          95        ✅      0
CompetitorAgent      65        ❌      3
RiskAgent            85        ✅      1
ReportAgent          90        ✅      0
────────────────────────────────────────────────────────
平均分: 83.8
整体通过: ❌ 否

---

六、生产环境质量控制策略

阶段	策略	说明
开发阶段	全量检查	所有维度、所有 Agent
测试阶段	交叉校验 + 人工抽检	自动 + 人工双重保障
上线初期	交叉校验 + 关键决策人工审核	高风险操作必须人工确认
稳定运行	自动评分 + 异常告警	低于阈值自动告警

---

七、运行

cd code/lesson-39-self-reflection
uv sync
uv run self_reflection.py

---

八、本课小结

1. 自我反思：Agent 自动检查输出质量，不达标则重试（最多 3 次）
2. 交叉校验：一个 Agent 分析，另一个 Agent 审核，模拟四眼原则
3. 质量评分：五维量化评估，让质量可视化、可追踪
4. 质量控制是生产级 Agent 系统的最后一道防线，不能省略

配套代码：code/lesson-39-self-reflection/self_reflection.py