Lab 009：Agent 自动重试几次最合适？成功率与成本曲线

实验假设：
  H1：重试 1 次成功率提升最大，之后递减（待验证）
  H2：重试 3 次后成功率接近饱和（待验证）
  H3：成本随重试次数线性增长（待验证）
  H4：给 Agent 失败原因后重试比盲目重试效果好（待验证）

# app/retry/retry_controller.py
"""
Agent 自动重试控制器。
支持多种重试策略和停止条件。
"""
import json
import time
from dataclasses import dataclass, field
from enum import Enum

class RetryStrategy(Enum):
    BLIND = "blind"               # 盲重试
    WITH_FEEDBACK = "feedback"    # 带反馈重试
    UPGRADE = "upgrade"           # 升级模型重试
    DECOMPOSE = "decompose"       # 分解后重试

@dataclass
class RetryConfig:
    max_retries: int = 3
    strategy: RetryStrategy = RetryStrategy.WITH_FEEDBACK
    budget_limit: float = 1.0     # 总预算上限
    stop_on_success: bool = True
    stop_condition: str = "quality_score >= 0.7"
    upgrade_model: str = "opus"   # 升级重试的目标模型
    cooldown_seconds: float = 0   # 重试间隔

@dataclass
class RetryResult:
    task_id: str
    attempts: list = field(default_factory=list)
    final_success: bool = False
    total_cost: float = 0.0
    total_tokens: int = 0
    total_time: float = 0.0
    strategy_used: str = ""
    
@dataclass
class AttemptResult:
    attempt_number: int
    success: bool
    quality_score: float
    cost: float
    tokens: int
    time_seconds: float
    failure_reason: str = ""
    model_used: str = ""

class RetryController:
    def __init__(self, config: RetryConfig, agent_runner):
        self.config = config
        self.runner = agent_runner
    
    async def execute_with_retry(self, task: dict) -> RetryResult:
        """带重试的执行"""
        result = RetryResult(task_id=task["id"])
        total_cost = 0.0
        previous_failures = []
        
        for attempt_num in range(self.config.max_retries + 1):
            # 检查预算
            if total_cost >= self.config.budget_limit:
                break
            
            # 构建 Prompt（根据策略）
            prompt = self._build_prompt(task, attempt_num, previous_failures)
            model = self._select_model(attempt_num)
            
            # 执行
            start = time.time()
            try:
                response = await self.runner.execute(prompt, model=model)
                elapsed = time.time() - start
                
                # 质量评估
                quality = await self._evaluate_quality(task, response)
                success = quality >= 0.7
                
                attempt = AttemptResult(
                    attempt_number=attempt_num + 1,
                    success=success,
                    quality_score=quality,
                    cost=response.cost,
                    tokens=response.tokens,
                    time_seconds=elapsed,
                    model_used=model,
                )
                
                if not success:
                    attempt.failure_reason = self._extract_failure_reason(response)
                    previous_failures.append(attempt)
                
                result.attempts.append(attempt)
                total_cost += response.cost
                result.total_tokens += response.tokens
                result.total_time += elapsed
                
                if success and self.config.stop_on_success:
                    result.final_success = True
                    break
                
            except Exception as e:
                elapsed = time.time() - start
                attempt = AttemptResult(
                    attempt_number=attempt_num + 1,
                    success=False,
                    quality_score=0.0,
                    cost=0.0,
                    tokens=0,
                    time_seconds=elapsed,
                    failure_reason=str(e),
                    model_used=model,
                )
                result.attempts.append(attempt)
                previous_failures.append(attempt)
            
            # 冷却间隔
            if self.config.cooldown_seconds > 0:
                await asyncio.sleep(self.config.cooldown_seconds)
        
        result.total_cost = total_cost
        result.strategy_used = self.config.strategy.value
        return result
    
    def _build_prompt(self, task: dict, attempt_num: int, failures: list) -> str:
        """根据重试策略构建 Prompt"""
        base_prompt = task["prompt"]
        
        if attempt_num == 0:
            return base_prompt
        
        if self.config.strategy == RetryStrategy.BLIND:
            return base_prompt  # 盲重试，Prompt 不变
        
        elif self.config.strategy == RetryStrategy.WITH_FEEDBACK:
            feedback = "上次尝试失败了，以下是失败信息：\n\n"
            for f in failures[-2:]:  # 最多参考最近 2 次失败
                feedback += f"尝试 {f.attempt_number}：{f.failure_reason}\n"
            feedback += "\n请分析失败原因，换一种方式解决这个问题。"
            return f"{base_prompt}\n\n{feedback}"
        
        elif self.config.strategy == RetryStrategy.DECOMPOSE:
            return f"{base_prompt}\n\n请将问题分解为更小的步骤，逐步解决。"
        
        return base_prompt
    
    def _select_model(self, attempt_num: int) -> str:
        """根据重试策略选择模型"""
        if self.config.strategy == RetryStrategy.UPGRADE and attempt_num > 0:
            return self.config.upgrade_model
        return "sonnet"  # 默认模型
    
    async def _evaluate_quality(self, task: dict, response) -> float:
        """评估响应质量"""
        if task.get("test_command"):
            result = await run_tests(task["test_command"])
            return result.passed / result.total
        return 0.5  # 无法评估时返回中间值
    
    def _extract_failure_reason(self, response) -> str:
        """从失败响应中提取失败原因"""
        if hasattr(response, "test_output"):
            return response.test_output.failure_summary[:500]
        return "Unknown failure"

# retry_experiment.py
"""
重试实验 Runner。
固定 100 个任务，分别测试 0/1/2/3/5 次最大重试。
"""

async def run_retry_experiment(tasks: list, max_retries_list: list):
    """运行完整重试实验"""
    all_results = {}
    
    for max_retries in max_retries_list:
        print(f"\n{'='*60}")
        print(f"测试最大重试次数: {max_retries}")
        print(f"{'='*60}")
        
        # 4 种策略分别测试
        for strategy in RetryStrategy:
            config = RetryConfig(
                max_retries=max_retries,
                strategy=strategy,
                budget_limit=10.0,  # 不设严格限制
            )
            controller = RetryController(config, agent_runner)
            
            results = []
            for task in tasks:
                result = await controller.execute_with_retry(task)
                results.append(result)
                status = "✅" if result.final_success else "❌"
                print(f"  [{strategy.value}] {task['id']} {status} "
                      f"attempts={len(result.attempts)} cost=${result.total_cost:.4f}")
            
            key = f"retries_{max_retries}_{strategy.value}"
            all_results[key] = results
    
    return all_results

累积成功率曲线（带反馈重试）：

100% ┤
     │                                    ●━━━━●━━━━ 饱和区
 85% ┤                              ●────●
     │                         ●────╯
 80% ┤                    ●────╯
     │               ●────╯
 75% ┤          ●────╯
     │     ●────╯
 70% ┤     ╱
     │   ╱  ← 最大收益区间
 65% ┤  ╱
     │╱
 60% ┤
     │
 50% ┤●  ← 首次尝试
     ┼────┬────┬────┬────┬────┬──
     0    1    2    3    4    5
              重试次数

成本效率排名（成功率增量 / 成本增量）：
  1. 带反馈重试（第 1 次）：效率 2.44 ★★★
  2. 升级重试（第 1 次）：效率 2.22 ★★☆
  3. 分解重试（第 1 次）：效率 1.78 ★★☆
  4. 带反馈重试（第 2 次）：效率 1.00 ★☆☆
  5. 盲重试（任何次数）：效率 < 0.5 ☆☆☆

# recommended-retry-config.yaml
retry_policy:
  default:
    max_retries: 2                    # 默认最多重试 2 次
    strategy: "with_feedback"          # 带反馈重试
    stop_on_success: true
    
  by_task_type:
    simple_bugfix:
      max_retries: 1                   # 简单 Bug 只重试 1 次
      reason: "1 次重试已覆盖 90% 收益"
    
    complex_bugfix:
      max_retries: 3                   # 复杂 Bug 可重试 3 次
      strategy: "upgrade_on_second"    # 第 2 次升级模型
      reason: "复杂任务需要更强模型"
    
    code_review:
      max_retries: 1                   # Review 只重试 1 次
      reason: "Review 结果不稳定，多次重试意义不大"
    
    test_generation:
      max_retries: 2
      strategy: "with_feedback"
      reason: "测试生成重试效果好"
  
  budget_guard:
    max_cost_multiplier: 3.0           # 总成本不超过首次的 3 倍
    alert_on_max_retries: true         # 达到最大重试时告警

def should_retry(attempt_results: list, config: dict) -> bool:
    """是否应该继续重试"""
    if not attempt_results:
        return True
    
    last = attempt_results[-1]
    
    # 1. 已成功 → 停止
    if last.success:
        return False
    
    # 2. 达到最大重试次数 → 停止
    if len(attempt_results) >= config["max_retries"]:
        return False
    
    # 3. 成本超出预算 → 停止
    total_cost = sum(a.cost for a in attempt_results)
    if total_cost >= config["budget_limit"]:
        return False
    
    # 4. 质量没有提升趋势 → 停止
    if len(attempt_results) >= 2:
        prev = attempt_results[-2]
        if last.quality_score <= prev.quality_score * 0.8:
            # 质量反而下降了，可能是方向错了
            return False
    
    return True