深入解析Agent异常终止问题：从错误处理到自动恢复机制

1次阅读

没有评论

共计 3231 个字符，预计需要花费 9 分钟才能阅读完成。

在 AI Agent 的实际运行中，异常终止是开发者经常遇到的问题。这些异常通常由以下几种场景触发：

API 限流：当调用第三方 API 时，超过请求速率限制会导致服务暂时不可用
资源耗尽：内存泄漏或 CPU 过载可能导致进程崩溃
网络波动：不稳定的网络连接会中断 Agent 与服务端的通信
代码缺陷：未处理的边界条件或逻辑错误引发运行时异常

这些异常如果不妥善处理，会导致用户体验下降、数据丢失，甚至引发连锁故障。特别是在生产环境中，手动恢复的成本极高。

指数退避（Exponential Backoff）
优点：避免在服务暂时不可用时造成请求风暴
缺点：恢复延迟较长，不适合对延迟敏感的场景
典型实现：初始间隔 1 秒，最大间隔 32 秒，最多重试 5 次
固定间隔重试（Fixed Interval）
优点：实现简单，响应时间可预测
缺点：可能加剧服务压力，特别是在服务尚未完全恢复时
典型实现：每隔 5 秒重试，最多 3 次

内存存储
优点：访问速度快，实现简单
缺点：进程终止后状态丢失，不适合关键业务
数据库存储
优点：状态可持久化，支持分布式恢复
缺点：引入额外依赖，性能开销较大
折衷方案：Redis 等内存数据库兼顾速度与持久性

from typing import Optional, Dict, Any
from enum import Enum
import logging

class ErrorType(Enum):
    TRANSIENT = 1  # 可恢复错误(如网络超时)
    PERMANENT = 2  # 不可恢复错误(如权限拒绝)
    RESOURCE = 3   # 资源问题(如内存不足)

def classify_error(exc: Exception) -> ErrorType:
    if isinstance(exc, (TimeoutError, ConnectionError)):
        return ErrorType.TRANSIENT
    elif isinstance(exc, (MemoryError, RuntimeError)):
        return ErrorType.RESOURCE
    return ErrorType.PERMANENT

import pickle
from datetime import datetime

class ContextManager:
    def __init__(self, storage_backend):
        self.storage = storage_backend

    def save_context(self, session_id: str, context: Dict[str, Any]) -> bool:
        try:
            snapshot = {'timestamp': datetime.utcnow().isoformat(),
                'data': pickle.dumps(context)
            }
            self.storage.set(f"ctx:{session_id}", snapshot)
            return True
        except Exception as e:
            logging.error(f"Failed to save context: {e}")
            return False

    def load_context(self, session_id: str) -> Optional[Dict[str, Any]]:
        try:
            snapshot = self.storage.get(f"ctx:{session_id}")
            if snapshot:
                return pickle.loads(snapshot['data'])
        except Exception as e:
            logging.error(f"Failed to load context: {e}")
        return None

import time
from functools import wraps

def retry(
    max_attempts: int = 3,
    initial_delay: float = 1.0,
    max_delay: float = 30.0,
    factor: float = 2.0
):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            delay = initial_delay
            last_error = None

            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    if classify_error(e) != ErrorType.TRANSIENT:
                        break

                    attempts += 1
                    if attempts < max_attempts:
                        time.sleep(min(delay, max_delay))
                        delay *= factor

            raise last_error if last_error else RuntimeError("Unknown error")
        return wrapper
    return decorator

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"

    def record_success(self):
        self.failure_count = 0
        self.state = "CLOSED"

    def is_request_allowed(self):
        if self.state == "CLOSED":
            return True

        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "HALF_OPEN"
                return True
            return False

        return True  # HALF_OPEN 状态允许试探请求