2.5 停止条件与断路器机制 — 循环安全运行的最后一道防线

本节导读：掌握循环工程中停止条件的设计方法——验证通过、迭代上限、无进展检测、Token 预算和人工升级，学会构建多层次的断路器保护机制防止循环失控。

学习目标

• 理解循环失控的风险和后果
• 掌握五类停止条件的设计与实现
• 学会构建多层次断路器机制
• 理解不同场景下停止条件的配置策略

核心概念

为什么停止条件如此重要？

Tosea.ai 的指南指出："The signature bug of a naive loop is that it never stops."（朴素循环的标志性 Bug 是它永远不停止。）

循环失控的后果不仅仅是浪费 Token——它可能：

• 修改了不应该修改的文件
• 引入了越来越多的 Bug（每轮"修复"引入新问题）
• 消耗大量 API 额度
• 在你睡觉时把代码库搞得一团糟

五类停止条件

停止条件的层次化设计

第一层：验证通过 → 正常停止（最好）
第二层：无进展检测 → 避免空转（重要）
第三层：迭代上限 → 硬性限制（必须）
第四层：Token/时间预算 → 成本保护（必须）
第五层：人工升级 → 兜底机制（最后手段）
​

环境准备 / 前置知识

• 已完成 2.1-2.4 节学习
• 有 Python 编程基础
• 了解基本的监控和告警概念

分步实战

步骤 1：实现完整的停止条件系统

"""
stop_conditions.py - 完整的停止条件系统
五类停止条件的实现和组合
"""
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
import time

class StopReason(Enum):
    GOAL_MET = "goal_met"          # 验证通过
    MAX_ITERATIONS = "max_iterations"  # 迭代上限
    NO_PROGRESS = "no_progress"    # 无进展
    BUDGET_EXCEEDED = "budget_exceeded"  # 预算耗尽
    ESCALATION = "escalation"       # 人工升级

@dataclass
class StopResult:
    should_stop: bool
    reason: Optional[StopReason] = None
    message: str = ""
    statistics: dict = None

class CircuitBreaker:
    """
    断路器：组合所有停止条件
    任何一条触发都立即停止循环
    """

    def __init__(self, config):
        self.max_iterations = config.get("max_iterations", 10)
        self.max_tokens = config.get("max_tokens", 200000)
        self.max_time_seconds = config.get("max_time_seconds", 600)
        self.no_progress_window = config.get("no_progress_window", 3)
        self.progress_threshold = config.get("progress_threshold", 0.05)

        self.tokens_used = 0
        self.iterations = 0
        self.start_time = None
        self.score_history: List[float] = []

    def start(self):
        self.start_time = time.time()

    def check(self, current_score: float) -> StopResult:
        """检查是否应该停止"""
        self.iterations += 1

        # 第一层：验证通过
        if current_score >= 1.0:
            return StopResult(
                should_stop=True,
                reason=StopReason.GOAL_MET,
                message=f"目标在第 {self.iterations} 轮达成",
                statistics=self._stats()
            )

        # 第二层：无进展检测
        if len(self.score_history) >= self.no_progress_window:
            recent = self.score_history[-self.no_progress_window:]
            max_score = max(recent)
            min_score = min(recent)
            if max_score - min_score < self.progress_threshold:
                return StopResult(
                    should_stop=True,
                    reason=StopReason.NO_PROGRESS,
                    message=f"连续 {self.no_progress_window} 轮无实质进展 "
                           f"(分数波动 < {self.progress_threshold})",
                    statistics=self._stats()
                )

        self.score_history.append(current_score)

        # 第三层：迭代上限
        if self.iterations >= self.max_iterations:
            return StopResult(
                should_stop=True,
                reason=StopReason.MAX_ITERATIONS,
                message=f"达到最大迭代次数 {self.max_iterations}",
                statistics=self._stats()
            )

        # 第四层：Token 预算
        if self.tokens_used >= self.max_tokens:
            return StopResult(
                should_stop=True,
                reason=StopReason.BUDGET_EXCEEDED,
                message=f"Token 预算耗尽 ({self.tokens_used}/{self.max_tokens})",
                statistics=self._stats()
            )

        # 第四层（续）：时间预算
        elapsed = time.time() - self.start_time
        if elapsed >= self.max_time_seconds:
            return StopResult(
                should_stop=True,
                reason=StopReason.BUDGET_EXCEEDED,
                message=f"时间预算耗尽 ({elapsed:.0f}/{self.max_time_seconds}s)",
                statistics=self._stats()
            )

        return StopResult(should_stop=False, statistics=self._stats())

    def _stats(self):
        return {
            "iterations": self.iterations,
            "tokens_used": self.tokens_used,
            "max_tokens": self.max_tokens,
            "elapsed_seconds": time.time() - self.start_time if self.start_time else 0,
            "score_history": self.score_history,
            "best_score": max(self.score_history) if self.score_history else 0
        }
​

步骤 2：配置不同场景的停止策略

# 场景一：保守策略（初学者推荐）
conservative = {
    "max_iterations": 5,
    "max_tokens": 50000,
    "max_time_seconds": 300,
    "no_progress_window": 2,
    "progress_threshold": 0.1  # 更敏感
}

# 场景二：平衡策略（日常使用）
balanced = {
    "max_iterations": 8,
    "max_tokens": 150000,
    "max_time_seconds": 600,
    "no_progress_window": 3,
    "progress_threshold": 0.05
}

# 场景三：激进策略（紧急任务）
aggressive = {
    "max_iterations": 15,
    "max_tokens": 500000,
    "max_time_seconds": 1800,
    "no_progress_window": 5,
    "progress_threshold": 0.02
}

# 场景四：探索性任务（开放循环）
exploratory = {
    "max_iterations": 20,
    "max_tokens": 1000000,
    "max_time_seconds": 3600,
    "no_progress_window": 5,
    "progress_threshold": 0.01  # 更宽容
}
​

步骤 3：实现人工升级机制

class EscalationHandler:
    """
    人工升级处理器
    当循环无法自主完成时，将问题升级给人类
    """

    def __init__(self, notification_channel=None):
        self.pending_escalations = []
        self.notification_channel = notification_channel

    def escalate(self, reason: str, context: dict):
        """触发人工升级"""
        escalation = {
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "reason": reason,
            "context": {
                "goal": context.get("goal"),
                "iterations": context.get("iterations"),
                "best_score": context.get("best_score"),
                "last_attempts": context.get("recent_attempts", [])[-3:]
            },
            "suggested_actions": self._suggest_actions(context)
        }
        self.pending_escalations.append(escalation)

        if self.notification_channel:
            self._send_notification(escalation)

        return escalation

    def _suggest_actions(self, context):
        """建议人工操作"""
        suggestions = []
        score = context.get("best_score", 0)

        if score < 0.3:
            suggestions.append("目标可能过于宽泛，考虑拆分为更小的子目标")
            suggestions.append("检查 Skills 是否提供了足够的上下文")
        elif score < 0.7:
            suggestions.append("循环取得了部分进展，检查是哪个维度卡住了")
            suggestions.append("考虑手动修复最难的 Bug，然后重新运行循环")
        else:
            suggestions.append("循环接近完成，可能只差最后一步")
            suggestions.append("检查验证器是否有 Bug（误报了失败）")

        suggestions.append("检查 Token 使用情况，是否需要调整预算")
        return suggestions
​

常见问题 FAQ

Q1：如何确定最大迭代次数？有没有经验值？

A：经验法则：Bug 修复任务设为 5-8 次，功能开发设为 10-15 次，探索性任务设为 15-20 次。关键不是具体数字，而是与无进展检测配合——如果连续 3 轮无改善，无论剩余多少次都应该停止。

Q2：Token 预算怎么估算？

A：粗略估算公式：每轮 Token ≈ 上下文大小 × 1.5（输入+输出）。例如，如果你的项目上下文约 20K tokens，每轮消耗约 30K，8 轮就是 240K。建议在此基础上加 50% 的安全余量。Claude Code 可以通过 claude config set maxTokens 设置全局上限。

Q3：无进展检测的"进展"怎么定义？

A：最简单也最可靠的方式：验证器的通过率或测试通过数。如果用连续 3 轮测试通过数没有增加（或分数波动 < 5%），就判定为无进展。更复杂的方式是对比代码变更的 diff hash——如果连续几轮改的是同一个文件同一区域，也是在打转。

最佳实践与避坑

• 永远设置最大迭代次数：这是最基本的防护，不要省略
• 无进展检测是最重要的智能停止：防止循环在死胡同中空转
• Token 预算要保守：实际消耗通常比预估多 30-50%
• 升级信息要丰富：当触发人工升级时，提供足够的上下文让人工快速理解问题
• 定期审查和调整：随着对循环行为的理解加深，优化停止参数

本节小结

本节详解了循环安全运行所需的五类停止条件：验证通过、迭代上限、无进展检测、Token/时间预算和人工升级。这些条件构成多层次断路器机制，确保循环在任何情况下都不会失控。记住 Osmani 的警告：当循环在你睡觉时运行时，一个可靠的断路器是唯一让你安心的东西。

至此，第二章"Loop Engineering 核心架构详解"全部完成。下一章我们将进入高级概念与技巧的学习。

延伸阅读

• Addy Osmani - Long Running Agents
• Tosea.ai - Complete Guide 2026
• 相关章节：3.1 状态持久化与上下文管理

关键词：Loop Engineering, 停止条件, 断路器, Circuit Breaker, 无进展检测, Token 预算, 人工升级, 循环安全, 教程, 架构

难度：进阶

预计阅读：14 分钟