📚 课程目录第3章确认重试策略

3. 确认重试策略

任务目标

在Agent执行关键操作前需要用户确认，但用户可能因误操作、网络延迟或注意力分散导致确认失败或超时。本案例实现一个带重试上限的确认机制：用户可重试，但最多允许N次（本案例设为3次），超过后终止操作并返回失败原因。

核心实现

import time
from enum import Enum

class ConfirmResult(Enum):
    APPROVED = "approved"
    REJECTED = "rejected"
    TIMEOUT = "timeout"
    EXCEEDED = "exceeded_max_retries"

class ConfirmationRetry:
    def __init__(self, max_retries=3, timeout_seconds=5):
        self.max_retries = max_retries
        self.timeout_seconds = timeout_seconds
        self.retry_count = 0

    def _get_user_input(self):
        """模拟用户输入，实际场景可替换为UI/API调用"""
        try:
            return input(f"确认执行操作？(y/n，{self.timeout_seconds}秒内响应): ").strip().lower()
        except Exception:
            return ""

    def confirm(self, action_desc):
        """执行带重试上限的确认流程"""
        while self.retry_count <= self.max_retries:
            remaining = self.max_retries - self.retry_count
            print(f"[{action_desc}] 剩余重试次数: {remaining}")
            
            start = time.time()
            user_input = self._get_user_input()
            elapsed = time.time() - start

            if elapsed >= self.timeout_seconds:
                reason = "超时未响应"
            elif user_input == "y":
                return ConfirmResult.APPROVED, self.retry_count
            elif user_input == "n":
                return ConfirmResult.REJECTED, self.retry_count
            else:
                reason = f"无效输入: '{user_input}'"

            self.retry_count += 1
            if self.retry_count > self.max_retries:
                break
            print(f"确认失败（{reason}），请重试")

        return ConfirmResult.EXCEEDED, self.retry_count

def execute_operation():
    """模拟需要确认的敏感操作"""
    print("正在执行敏感操作：删除生产数据库记录...")

def main():
    confirm_mgr = ConfirmationRetry(max_retries=3, timeout_seconds=5)
    result, attempts = confirm_mgr.confirm("删除生产数据库记录")
    
    if result == ConfirmResult.APPROVED:
        execute_operation()
        print(f"操作已执行（第{attempts + 1}次确认通过）")
    elif result == ConfirmResult.REJECTED:
        print("用户拒绝，操作已取消")
    elif result == ConfirmResult.TIMEOUT:
        print("确认超时，操作已终止")
    else:
        print(f"重试次数超过上限（{attempts}次），操作已终止")

if __name__ == "__main__":
    main()

运行与验证

运行方式：

python confirmation_retry.py

验证路径：分别测试以下场景，观察输出是否符合预期：

成功确认：输入y → 应输出“操作已执行”
用户拒绝：输入n → 应输出“用户拒绝，操作已取消”
无效输入后重试成功：先输入abc，再输入y → 应提示重试并最终执行
超过重试上限：连续输入abc共4次（首次+3次重试） → 应输出“重试次数超过上限（4次），操作已终止”
超时终止：等待超过5秒不输入 → 应输出“确认失败（超时未响应）”并进入重试，连续超时4次后终止

判定标准：代码中max_retries=3表示首次确认失败后最多允许3次重试，总计最多4次确认机会。当第4次仍失败时，retry_count变为4，循环条件retry_count <= max_retries不再满足，返回EXCEEDED。观察输出中“剩余重试次数”从3递减到0，且最终正确终止操作，即为通过。

关键设计说明

超时检测：使用time.time()计算输入耗时，而非依赖输入函数自身的超时机制，保证跨平台行为一致。
重试计数：retry_count记录已发生的失败次数，首次失败后变为1，达到max_retries+1时终止。
实际接入：将_get_user_input()替换为实际UI对话框、API回调或消息队列等待即可用于生产环境，重试逻辑无需改动。