全部技能
我安装的
error-recovery
去试试


自动化错误恢复机制 - 错误分类、自动修复、重试逻辑、降级方案、跨会话学习。触发词：错误恢复、自动修复、重试机制、降级方案、错误处理、错误分类、指数退避、抖动策略。适用于所有任务的错误处理和恢复。



name
error-recovery
description
自动化错误恢复机制 - 错误分类、自动修复、重试逻辑、降级方案、跨会话学习。触发词：错误恢复、自动修复、重试机制、降级方案、错误处理、错误分类、指数退避、抖动策略。适用于所有任务的错误处理和恢复。
disable
false
Error Recovery - 自动化错误恢复机制
版本：1.0
最后更新：2026-05-04
作者：小南（楠哥的 AI 助理）

一、错误分类体系（Error Classification）
1.1 错误类型定义
python
from enum import Enum

class ErrorType(Enum):
    """错误类型枚举"""
    
    # 文件操作错误
    FILE_NOT_FOUND = "文件不存在"
    FILE_PERMISSION_DENIED = "文件权限不足"
    FILE_ALREADY_EXISTS = "文件已存在"
    FILE_ENCODING_ERROR = "文件编码错误"
    
    # 网络请求错误
    NETWORK_TIMEOUT = "网络超时"
    NETWORK_CONNECTION_ERROR = "网络连接错误"
    API_RATE_LIMIT = "API 速率限制"
    API_AUTHENTICATION_ERROR = "API 认证错误"
    
    # 工具调用错误
    TOOL_NOT_FOUND = "工具不存在"
    TOOL_PARAMETER_ERROR = "工具参数错误"
    TOOL_EXECUTION_ERROR = "工具执行错误"
    
    # 依赖库错误
    MODULE_NOT_FOUND = "模块未找到"
    VERSION_CONFLICT = "版本冲突"
    DEPENDENCY_MISSING = "依赖缺失"
    
    # 数据格式错误
    INVALID_JSON = "JSON 格式无效"
    INVALID_XML = "XML 格式无效"
    ENCODING_ERROR = "编码错误"
    
    # 系统资源错误
    DISK_FULL = "磁盘空间不足"
    MEMORY_ERROR = "内存不足"
    CPU_OVERLOAD = "CPU 过载"
1.2 错误严重程度
python
class ErrorSeverity(Enum):
    """错误严重程度"""
    
    CRITICAL = "严重"  # 导致任务完全失败
    HIGH = "高"        # 导致部分功能失败
    MEDIUM = "中"      # 可以降级处理
    LOW = "低"         # 不影响主要功能
二、自动修复策略（Auto Recovery Strategies）
2.1 文件操作错误修复
策略 1：文件不存在 → 创建父目录

python
import os
from pathlib import Path

def auto_fix_file_not_found(file_path):
    """自动修复文件不存在错误"""
    try:
        # 检查文件是否存在
        if not os.path.exists(file_path):
            # 创建父目录
            parent_dir = Path(file_path).parent
            parent_dir.mkdir(parents=True, exist_ok=True)
            
            # 创建空文件
            with open(file_path, 'w', encoding='utf-8') as f:
                f.write("")
            
            return True, f"已创建文件：{file_path}"
        return True, "文件已存在"
    
    except Exception as e:
        return False, f"修复失败：{e}"
策略 2：文件权限不足 → 修改权限

python
import stat
import os

def auto_fix_permission_denied(file_path):
    """自动修复文件权限不足错误"""
    try:
        # 添加读写权限
        os.chmod(file_path, stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)
        return True, f"已修改文件权限：{file_path}"
    
    except Exception as e:
        return False, f"修复失败：{e}"
策略 3：文件编码错误 → 尝试多种编码

python
def auto_fix_encoding_error(file_path):
    """自动修复文件编码错误"""
    encodings = ['utf-8', 'gbk', 'gb2312', 'iso-8859-1']
    
    for encoding in encodings:
        try:
            with open(file_path, 'r', encoding=encoding) as f:
                content = f.read()
            return True, f"使用 {encoding} 编码成功读取文件"
        
        except UnicodeDecodeError:
            continue
    
    return False, "所有编码尝试均失败"
2.2 网络请求错误修复
策略 1：网络超时 → 指数退避重试

python
import time
import random
import requests
from typing import Callable, Any

def exponential_backoff_retry(
    func: Callable,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: bool = True
) -> Any:
    """
    指数退避重试
    
    Args:
        func: 要执行的函数
        max_retries: 最大重试次数
        base_delay: 基础延迟（秒）
        max_delay: 最大延迟（秒）
        jitter: 是否添加抖动（避免惊群效应）
    """
    for attempt in range(max_retries):
        try:
            return func()
        
        except (requests.Timeout, requests.ConnectionError) as e:
            if attempt == max_retries - 1:
                raise Exception(f"重试 {max_retries} 次后仍然失败：{e}")
            
            # 计算延迟时间
            delay = min(base_delay * (2 ** attempt), max_delay)
            
            # 添加抖动（0.5 ~ 1.5 倍）
            if jitter:
                delay = delay * (0.5 + random.random())
            
            print(f"第 {attempt + 1} 次尝试失败，{delay:.2f} 秒后重试...")
            time.sleep(delay)
    
    return None
策略 2：API 速率限制 → 等待后重试

python
import time
import requests

def handle_rate_limit(func, *args, **kwargs):
    """处理 API 速率限制"""
    max_retries = 3
    
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        
        except requests.HTTPError as e:
            if e.response.status_code == 429:  # Too Many Requests
                # 从响应头获取重试时间
                retry_after = int(e.response.headers.get('Retry-After', 60))
                
                if attempt < max_retries - 1:
                    print(f"遇到速率限制，等待 {retry_after} 秒...")
                    time.sleep(retry_after)
                else:
                    raise Exception(f"达到速率限制，重试 {max_retries} 次后仍然失败")
            else:
                raise e
    
    return None
2.3 依赖库错误修复
策略 1：模块未找到 → 自动安装

python
import subprocess
import sys
import importlib

def auto_install_missing_module(module_name):
    """自动安装缺失的模块"""
    try:
        # 尝试导入
        importlib.import_module(module_name)
        return True, f"模块已安装：{module_name}"
        
    except ImportError:
        try:
            # 自动安装
            print(f"模块 {module_name} 未找到，正在自动安装...")
            
            # 使用国内镜像加速
            result = subprocess.run(
                [
                    sys.executable,
                    "-m", "pip", "install",
                    module_name,
                    "-i", "https://pypi.tuna.tsinghua.edu.cn/simple"
                ],
                capture_output=True,
                text=True
            )
            
            if result.returncode == 0:
                return True, f"成功安装模块：{module_name}"
            else:
                return False, f"安装失败：{result.stderr}"
        
        except Exception as e:
            return False, f"自动安装失败：{e}"
策略 2：版本冲突 → 创建虚拟环境

python
import subprocess
import sys
import os

def create_virtual_env_with_correct_version(env_name, python_version):
    """创建虚拟环境并安装正确版本的依赖"""
    try:
        # 创建虚拟环境
        venv_path = f".venv_{env_name}"
        subprocess.run([sys.executable, "-m", "venv", venv_path])
        
        # 激活虚拟环境并安装依赖
        if os.name == 'nt':  # Windows
            pip_path = os.path.join(venv_path, "Scripts", "pip.exe")
        else:  # Unix
            pip_path = os.path.join(venv_path, "bin", "pip")
        
        # 安装指定版本的 Python
        subprocess.run([pip_path, "install", f"python=={python_version}"])
        
        return True, f"已创建虚拟环境：{venv_path}"
    
    except Exception as e:
        return False, f"创建虚拟环境失败：{e}"
三、降级方案（Fallback Strategies）
3.1 OCR 识别降级
python
def ocr_with_fallback(image_path):
    """OCR 识别（带降级）"""
    strategies = [
        ("easyocr", lambda: easyocr.readtext(image_path)),
        ("ocr_space_api", lambda: ocr_space_api(image_path)),
        ("baidu_ocr", lambda: baidu_ocr_api(image_path)),
        ("manual_input", lambda: ask_user_for_manual_input(image_path))
    ]
    
    for strategy_name, strategy_func in strategies:
        try:
            result = strategy_func()
            if result:
                return result
        except Exception as e:
            print(f"策略 {strategy_name} 失败：{e}")
            continue
    
    raise Exception("所有 OCR 策略均失败")
3.2 PDF 生成降级
python
def generate_pdf_with_fallback(content, output_path):
    """生成 PDF（带降级）"""
    strategies = [
        ("reportlab", lambda: generate_pdf_reportlab(content, output_path)),
        ("PyPDF2", lambda: generate_pdf_pypdf2(content, output_path)),
        ("html_to_pdf", lambda: generate_pdf_from_html(content, output_path)),
        ("docx_to_pdf", lambda: generate_pdf_via_docx(content, output_path))
    ]
    
    for strategy_name, strategy_func in strategies:
        try:
            strategy_func()
            if os.path.exists(output_path):
                return True, f"使用 {strategy_name} 成功生成 PDF"
        except Exception as e:
            print(f"策略 {strategy_name} 失败：{e}")
            continue
    
    return False, "所有 PDF 生成策略均失败"
3.3 中文文本处理降级
python
def handle_chinese_text_with_fallback(text):
    """处理中文文本（带降级）"""
    strategies = [
        ("SimSun", lambda: process_with_font(text, "SimSun.ttf")),
        ("MSYH", lambda: process_with_font(text, "MSYH.TTF")),
        ("SimHei", lambda: process_with_font(text, "SimHei.ttf")),
        ("image_fallback", lambda: convert_text_to_image(text))
    ]
    
    for strategy_name, strategy_func in strategies:
        try:
            result = strategy_func()
            return result
        except Exception as e:
            print(f"策略 {strategy_name} 失败：{e}")
            continue
    
    raise Exception("所有中文文本处理策略均失败")
四、错误记录与跨会话学习（Error Logging & Cross-Session Learning）
4.1 错误记录格式
python
import json
from datetime import datetime
from pathlib import Path

def log_error(error_type, error_msg, file_path=None, fix_strategy=None, success=False):
    """记录错误信息到 JSON 文件"""
    log_file = Path(".workbuddy/memory/error_log.json")
    
    # 读取现有日志
    if log_file.exists():
        with open(log_file, 'r', encoding='utf-8') as f:
            error_log = json.load(f)
    else:
        error_log = []
    
    # 添加新错误记录
    error_record = {
        "timestamp": datetime.now().isoformat(),
        "error_type": error_type,
        "error_message": str(error_msg),
        "file_path": file_path,
        "fix_strategy": fix_strategy,
        "success": success
    }
    error_log.append(error_record)
    
    # 保存到文件
    with open(log_file, 'w', encoding='utf-8') as f:
        json.dump(error_log, f, ensure_ascii=False, indent=2)
    
    return error_record
4.2 错误模式分析
python
def analyze_error_patterns():
    """分析错误模式，生成预防建议"""
    log_file = Path(".workbuddy/memory/error_log.json")
    
    if not log_file.exists():
        return []
    
    with open(log_file, 'r', encoding='utf-8') as f:
        error_log = json.load(f)
    
    # 统计错误类型频率
    error_counts = {}
    for record in error_log:
        error_type = record['error_type']
        if error_type not in error_counts:
            error_counts[error_type] = 0
        error_counts[error_type] += 1
    
    # 生成预防建议
    suggestions = []
    for error_type, count in sorted(error_counts.items(), key=lambda x: x[1], reverse=True):
        if count >= 3:  # 出现 3 次以上
            suggestion = generate_prevention_suggestion(error_type, count)
            suggestions.append(suggestion)
    
    return suggestions

def generate_prevention_suggestion(error_type, count):
    """生成预防性建议"""
    suggestions = {
        "FILE_NOT_FOUND": "建议：在执行文件操作前，先检查文件是否存在",
        "MODULE_NOT_FOUND": "建议：在项目开始时，先检查并安装所有依赖",
        "NETWORK_TIMEOUT": "建议：为所有网络请求添加超时和重试机制",
        "ENCODING_ERROR": "建议：统一使用 UTF-8 编码，并在打开文件时指定 encoding='utf-8'"
    }
    
    return {
        "error_type": error_type,
        "count": count,
        "suggestion": suggestions.get(error_type, "建议：查看错误日志获取详细信息")
    }
4.3 自动更新 MEMORY.md
python
def update_memory_with_error_pattern(error_type, fix_strategy):
    """将错误模式更新到 MEMORY.md"""
    memory_file = Path(".workbuddy/memory/MEMORY.md")
    
    if not memory_file.exists():
        return
    
    # 读取 MEMORY.md
    with open(memory_file, 'r', encoding='utf-8') as f:
        content = f.read()
    
    # 检查是否已存在该错误类型
    error_section = "## 六、常见错误与修复（Error Patterns）"
    
    if error_type not in content:
        # 添加新错误模式
        new_error_entry = f"""
### 6.{get_next_error_number()} {error_type}

- **错误**：{error_type}
- **修复策略**：{fix_strategy}
- **预防建议**：{generate_prevention_suggestion(error_type, 1)['suggestion']}
"""
        
        # 插入到错误章节
        content = content.replace(
            error_section,
            error_section + new_error_entry
        )
        
        # 保存更新后的内容
        with open(memory_file, 'w', encoding='utf-8') as f:
            f.write(content)
五、主动预防机制（Proactive Prevention）
5.1 预检查清单
python
def pre_check_before_task(task_type):
    """任务开始前预检查"""
    checklists = {
        "file_operation": [
            "检查文件是否存在",
            "检查文件权限",
            "检查磁盘空间",
            "备份重要文件"
        ],
        "network_request": [
            "检查网络连接",
            "检查 API Key 是否有效",
            "检查 API 速率限制",
            "设置超时时间"
        ],
        "document_generation": [
            "检查依赖库是否安装",
            "检查中文字体是否可用",
            "检查输出目录是否存在",
            "准备降级方案"
        ]
    }
    
    checklist = checklists.get(task_type, [])
    
    print(f"=== 预检查清单 ({task_type}) ===")
    for item in checklist:
        print(f"[ ] {item}")
    
    return checklist
5.2 自动修复建议
python
def suggest_auto_fix(error):
    """根据错误类型自动建议修复方案"""
    suggestions = {
        "ModuleNotFoundError": "建议：运行 `pip install <module_name>`",
        "FileNotFoundError": "建议：检查文件路径是否正确，或使用 `os.path.exists()` 检查",
        "PermissionError": "建议：检查文件权限，或使用 `os.chmod()` 修改权限",
        "TimeoutError": "建议：增加超时时间，或实现重试机制",
        "JSONDecodeError": "建议：检查 JSON 格式，或使用 `json.loads()` 前先验证"
    }
    
    error_type = type(error).__name__
    return suggestions.get(error_type, "建议：查看错误日志获取详细信息")
六、完整错误处理示例（Complete Example）
python
import functools
import json
from pathlib import Path

def with_error_recovery(error_types=None, max_retries=3, fallback_func=None):
    """
    错误处理装饰器
    
    Args:
        error_types: 要捕获的错误类型列表
        max_retries: 最大重试次数
        fallback_func: 降级函数
    """
    if error_types is None:
        error_types = [Exception]
    
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 预检查
            pre_check_before_task(func.__name__)
            
            # 重试逻辑
            for attempt in range(max_retries):
                try:
                    result = func(*args, **kwargs)
                    
                    # 成功：记录成功日志
                    log_error(
                        error_type="SUCCESS",
                        error_msg=f"{func.__name__} 执行成功",
                        success=True
                    )
                    
                    return result
                
                except tuple(error_types) as e:
                    # 记录错误
                    log_error(
                        error_type=type(e).__name__,
                        error_msg=str(e),
                        success=False
                    )
                    
                    # 尝试自动修复
                    fix_strategy = suggest_auto_fix(e)
                    print(f"尝试自动修复：{fix_strategy}")
                    
                    if attempt < max_retries - 1:
                        print(f"第 {attempt + 1} 次尝试失败，正在重试...")
                        time.sleep(2 ** attempt)  # 指数退避
                    else:
                        # 所有重试均失败，使用降级方案
                        if fallback_func:
                            print("使用降级方案...")
                            return fallback_func(*args, **kwargs)
                        else:
                            raise Exception(f"重试 {max_retries} 次后仍然失败：{e}")
        
        return wrapper
    return decorator

# 使用示例
@with_error_recovery(
    error_types=[FileNotFoundError, PermissionError],
    max_retries=3,
    fallback_func=lambda file_path: f"无法处理文件：{file_path}"
)
def process_file(file_path):
    """处理文件（带自动错误恢复）"""
    with open(file_path, 'r', encoding='utf-8') as f:
        content = f.read()
    return content
七、集成到 Best Practices
7.1 更新 best-practices SKILL.md
在 best-practices Skill 中添加以下章节：

markdown
## 十三、错误恢复最佳实践（Error Recovery）

### 13.1 使用错误恢复装饰器

```python
from error_recovery import with_error_recovery

@with_error_recovery(
    error_types=[FileNotFoundError, PermissionError],
    max_retries=3,
    fallback_func=my_fallback
)
def my_function():
    # 函数实现
    pass
13.2 实现降级方案
python
def my_function_with_fallback():
    strategies = [
        strategy_1,
        strategy_2,
        strategy_3
    ]
    
    for strategy in strategies:
        try:
            return strategy()
        except Exception as e:
            continue
    
    raise Exception("所有策略均失败")
13.3 记录错误用于学习
python
from error_recovery import log_error, analyze_error_patterns

try:
    result = risky_operation()
except Exception as e:
    log_error(
        error_type=type(e).__name__,
        error_msg=str(e),
        fix_strategy="自动重试",
        success=False
    )
    
    # 分析错误模式
    patterns = analyze_error_patterns()
    print("错误模式分析：", patterns)

---

## 八、总结（Summary）

### 核心功能

1. **错误分类** - 将错误分为文件操作、网络请求、工具调用等类型
2. **自动修复** - 针对每种错误类型，提供自动修复策略
3. **重试机制** - 实现指数退避、抖动等高级重试策略
4. **降级方案** - 当主要方案失败时，自动切换到备用方案
5. **错误记录** - 将错误信息记录到 `error_log.json`
6. **跨会话学习** - 分析错误模式，更新 MEMORY.md
7. **主动预防** - 根据历史错误模式，提前检查可能的问题

### 使用方式

1. **作为装饰器使用**：`@with_error_recovery`
2. **作为函数调用**：`auto_fix_file_not_found()`
3. **作为系统设计**：在任务开始前调用 `pre_check_before_task()`

### 文件清单

- `SKILL.md` - 本文件（完整文档）
- `scripts/error_recovery.py` - Python 实现
- `references/error_patterns.json` - 错误模式数据库
- `examples/usage_examples.py` - 使用示例

---

**最后更新**：2026-05-04 02:30  
**版本**：1.0  
**维护者**：小南（楠哥的 AI 助理）
