AI Agent 实战（六）：容错与自愈——构建高可用的 Agent 系统

第六章：容错与自愈——构建高可用的 Agent 系统

本章是《AI Agent 实战》系列第六篇，主题是容错与自愈：把‘可能失败’当成常态，用重试、熔断、降级、自动修复把系统稳定性拉满。目标只有一个：让 Agent 在真实世界里跑得久、扛得住、还能自己爬起来。

6.1 为什么 Agent 需要容错？

在真实环境中，Agent 的每一次工具调用都可能失败：

网络抖动：API 超时、连接重置。
依赖缺失：command not found、缺少 Python 包。
权限问题：文件无法写入、API Key 过期。
数据错误：输入参数格式错误、JSON 解析失败。

如果 Agent 遇到错误就直接崩溃，那么它将无法完成任何复杂任务。因此，容错与自愈是构建高可用 Agent 系统的核心能力。

6.2 错误分类：瞬时错误 vs 持久错误

6.2.1 瞬时错误（Transient Errors）

特点：暂时性故障，重试后可能成功。

示例：

网络超时（TimeoutError）。
连接重置（ECONNRESET）。
速率限制（429 Too Many Requests）。
服务器临时不可用（503 Service Unavailable）。

策略：自动重试（指数退避）。

6.2.2 持久错误（Persistent Errors）

特点：根本性问题，重试无法解决。

示例：

参数错误（400 Bad Request）。
权限不足（403 Forbidden）。
资源不存在（404 Not Found）。
代码逻辑错误（SyntaxError）。

策略：人工干预或调整策略。

6.3 重试机制：指数退避

6.3.1 什么是指数退避？

指数退避（Exponential Backoff） 是一种智能重试策略：

第一次失败：等待 1 秒后重试。
第二次失败：等待 2 秒后重试。
第三次失败：等待 4 秒后重试。
...
第 N 次失败：等待 2^(N-1) 秒后重试。

优点：

避免频繁重试导致服务器压力。
给系统足够时间恢复。

6.3.2 实战：实现指数退避重试

import time
import requests
from requests.exceptions import Timeout, ConnectionError

def retry_with_backoff(func, max_retries=3, base_delay=1):
    """带指数退避的重试装饰器"""
    for attempt in range(max_retries):
        try:
            return func()
        except (Timeout, ConnectionError) as e:
            if attempt == max_retries - 1:
                raise e
            delay = base_delay * (2 ** attempt)
            print(f"重试 {attempt+1}/{max_retries}，等待 {delay} 秒...")
            time.sleep(delay)

# 使用示例
def fetch_data(url):
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()

# 调用
try:
    result = retry_with_backoff(lambda: fetch_data("https://api.example.com/data"))
    print("成功获取数据:", result)
except Exception as e:
    print("重试失败:", e)

6.4 自愈策略：自动修复常见错误

6.4.1 场景：command not found

错误：

$ jq --version
bash: jq: command not found

自愈流程：

识别错误：解析错误信息，提取缺失的工具名（jq）。
选择包管理器：根据系统选择（apt-get、yum、brew）。
安装工具：执行 apt-get install -y jq。
验证安装：再次运行 jq --version。
重试原命令：如果安装成功，重新执行原命令。

代码示例：

import subprocess
import re

def auto_repair_command(error_message):
    """自动修复 command not found 错误"""
    # 1. 提取工具名
    match = re.search(r"bash: (\w+): command not found", error_message)
    if not match:
        return False
    
    tool_name = match.group(1)
    print(f"检测到缺失工具：{tool_name}，尝试安装...")
    
    # 2. 选择包管理器
    try:
        subprocess.run(["apt-get", "--version"], check=True, capture_output=True)
        pkg_manager = "apt-get"
    except:
        try:
            subprocess.run(["yum", "--version"], check=True, capture_output=True)
            pkg_manager = "yum"
        except:
            print("未找到支持的包管理器")
            return False
    
    # 3. 安装工具
    try:
        subprocess.run([pkg_manager, "install", "-y", tool_name], check=True)
        print(f"成功安装 {tool_name}")
        return True
    except Exception as e:
        print(f"安装失败：{e}")
        return False

# 使用
error = "bash: jq: command not found"
if auto_repair_command(error):
    print("已修复，请重试原命令")

6.4.2 场景：Python 包缺失

错误：

ModuleNotFoundError: No module named 'requests'

自愈流程：

识别错误：提取缺失的包名（requests）。
安装包：执行 pip install requests。
重试导入：重新运行代码。

代码示例：

import subprocess
import re

def auto_install_package(error_message):
    """自动安装缺失的 Python 包"""
    match = re.search(r"No module named '([\w-]+)'", error_message)
    if not match:
        return False
    
    package_name = match.group(1)
    print(f"检测到缺失包：{package_name}，尝试安装...")
    
    try:
        subprocess.run(["pip3", "install", package_name], check=True)
        print(f"成功安装 {package_name}")
        return True
    except Exception as e:
        print(f"安装失败：{e}")
        return False

# 使用
error = "ModuleNotFoundError: No module named 'requests'"
if auto_install_package(error):
    print("已修复，请重试原代码")

6.5 实战：构建自愈 Agent

将上述容错机制集成到 OpenClaw 中，实现自愈 Agent。

# 自愈 Agent 核心逻辑
class SelfHealingAgent:
    def __init__(self):
        self.max_retries = 3
    
    def execute_tool(self, tool_name, params):
        """执行工具，带容错"""
        for attempt in range(self.max_retries):
            try:
                result = self.call_tool(tool_name, params)
                return result
            except TimeoutError as e:
                if attempt == self.max_retries - 1:
                    raise e
                time.sleep(2 ** attempt)
            except subprocess.CalledProcessError as e:
                error_msg = e.stderr.decode()
                if "command not found" in error_msg:
                    if auto_repair_command(error_msg):
                        continue
                raise e
            except ModuleNotFoundError as e:
                if auto_install_package(str(e)):
                    continue
                raise e
        
        raise Exception("工具执行失败，超出最大重试次数")
    
    def call_tool(self, tool_name, params):
        """实际的工具调用逻辑"""
        # 调用 mcporter 或其他工具
        cmd = ["mcporter", "call", tool_name] + [f"{k}={v}" for k, v in params.items()]
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        return result.stdout

# 使用
agent = SelfHealingAgent()
try:
    result = agent.execute_tool("weather-mcp-server", {"city": "Beijing"})
    print("工具执行成功:", result)
except Exception as e:
    print("工具执行失败:", e)

6.6 深度思考：自愈的边界在哪里？

6.6.1 过度自动化的风险

掩盖问题：自动修复可能掩盖系统性问题（如配置错误）。
安全风险：自动安装软件可能引入恶意代码。
资源浪费：频繁重试可能消耗大量计算资源。

6.6.2 最佳实践

限制范围：仅对已知安全的问题（如网络超时、包缺失）进行自动修复。
人工确认：对高风险操作（如系统级安装）要求人工确认。
日志记录：记录所有修复操作，便于审计和排查。
监控告警：当修复失败时，及时通知人工介入。

结论：自愈是 Agent 系统的重要能力，但必须在安全边界内运行。过度自动化可能导致不可预知的后果，因此需要谨慎设计。

（第六章完）

下一章预告：第七章将探讨OpenClaw 社区生态与AI Agent 的未来趋势。