LLM Agent异常终止处理指南：如何优雅应对’agent terminated due to error’问题

1次阅读

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

在日常开发中，LLM Agent 的异常终止是影响业务连续性的主要风险之一。当出现 ”agent terminated due to error” 提示时，不仅会中断当前任务，还可能导致数据丢失或状态不一致。常见的触发原因包括 API 限流、上下文窗口溢出、网络波动以及模型内部错误等。下面我们就来深入探讨如何构建健壮的容错机制。

重试是处理瞬时错误的第一道防线，但简单的立即重试可能加剧服务压力。一个完善的方案需要考虑：

退避算法选择 ：指数退避（Exponential Backoff）能有效避免请求风暴
重试条件判断 ：仅对可重试错误（如 429、5xx）触发机制
最大尝试次数 ：通常 3 - 5 次为宜，需考虑业务场景

from datetime import timedelta
import random

def exponential_backoff(
    retries: int,
    base_delay: float = 1.0,
    max_delay: float = 30.0
) -> float:
    """计算指数退避延迟时间"""
    delay = min(base_delay * (2 ** (retries - 1)), max_delay)
    return delay * (1 + random.random())  # 添加抖动避免同步重试

async def call_with_retry(
    func: Callable,
    max_retries: int = 3,
    **kwargs
):
    last_error = None
    for attempt in range(1, max_retries + 1):
        try:
            return await func(**kwargs)
        except RetryableError as e:
            last_error = e
            if attempt == max_retries:
                break
            delay = exponential_backoff(attempt)
            await asyncio.sleep(delay)
    raise RetryExhaustedError(f"After {max_retries} attempts") from last_error

对于长时间运行的 Agent，定期保存状态至关重要：

快照策略 ：在关键检查点（如每 N 轮对话）保存状态
序列化格式 ：推荐使用 JSON 或 Protocol Buffers
存储选择 ：根据性能要求选用内存 /Redis/ 数据库

from dataclasses import dataclass, asdict
import json
from typing import Dict, Any

@dataclass
class AgentState:
    conversation_history: List[Dict[str, str]]
    current_task: str
    context_window: int

class StateManager:
    def __init__(self, storage_path: str):
        self.storage_path = storage_path

    def save_state(self, state: AgentState) -> None:
        """保存状态到磁盘"""
        with open(self.storage_path, 'w') as f:
            json.dump(asdict(state), f)

    def load_state(self) -> AgentState:
        """从磁盘加载状态"""
        try:
            with open(self.storage_path, 'r') as f:
                data = json.load(f)
            return AgentState(**data)
        except FileNotFoundError:
            return AgentState([], "", 0)

当错误率超过阈值时，应触发熔断机制：

断路器模式 ：参考 Circuit Breaker 设计模式
健康度指标 ：统计最近 N 次请求的错误率
半开状态 ：允许部分请求试探服务恢复

from collections import deque
from time import time

class CircuitBreaker:
    def __init__(self, 
                 failure_threshold: float = 0.5,
                 recovery_timeout: int = 60):
        self.window = deque(maxlen=100)  # 滑动窗口
        self.threshold = failure_threshold
        self.timeout = recovery_timeout
        self.tripped_time = 0

    def record_result(self, success: bool) -> None:
        self.window.append(success)

    def should_block(self) -> bool:
        if not self.window:
            return False

        failure_rate = 1 - sum(self.window)/len(self.window)
        if failure_rate >= self.threshold:
            self.tripped_time = time()
            return True

        # 检查是否在冷却期
        if self.tripped_time > 0:
            if (time() - self.tripped_time) < self.timeout:
                return True
            # 进入半开状态
            if len(self.window) < 10:  # 样本不足
                return False
            return failure_rate >= self.threshold/2
        return False

幂等性处理
为每个请求生成唯一 ID
服务端实现请求去重
写操作前先检查状态
上下文窗口管理
实时计算 token 消耗
实现 LRU 缓存淘汰策略
对长对话自动分片

监控指标设计

# Prometheus 监控示例
from prometheus_client import Counter, Histogram

REQUEST_ERRORS = Counter('agent_errors', 'Error types', ['type'])
RESPONSE_TIME = Histogram('agent_latency', 'Request latency')

@RESPONSE_TIME.time()
async def process_request():
    try:
        # 处理逻辑
        pass
    except APIError as e:
        REQUEST_ERRORS.labels(type=e.__class__.__name__).inc()