AI Agent Retry Strategies: Gracefully Handling Failures

AI Agent Retry Strategies: Gracefully Handling Failures

💡 What You Will Learn

AI Agent Retry Strategies: Gracefully Handling Failures

import time
import random
from functools import wraps

def retry_immediate(max_retries=3):
    """RetryDecorator"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except (TimeoutError, ConnectionError) as e:
                    if attempt == max_retries:
                        raise
                    print(f"[] {attempt}: {e}")
            return None
        return wrapper
    return decorator

def retry_exponential_backoff(max_retries=5, base_delay=1.0, max_delay=60.0, jitter=True):
    """Exponential BackoffRetry"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except (RateLimitError, ServiceUnavailable) as e:
                    if attempt == max_retries:
                        raise
                    delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
                    if jitter:
                        delay *= random.uniform(0.5, 1.5)
                    print(f"[] {attempt}{delay:.1f}")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

class FallbackModelRouter:
    def __init__(self):
        self.models = [
            ("gpt-4o-mini", OpenAILLM(model="gpt-4o-mini")),
            ("qwen3:32b", OllamaLLM(model="qwen3:32b")),
            ("deepseek-chat", DeepSeekLLM(model="deepseek-chat")),
        ]

    def invoke_with_fallback(self, prompt: str) -> str:
        last_error = None
        for model_name, model in self.models:
            try:
                print(f"[] : {model_name}")
                return model.invoke(prompt)
            except Exception as e:
                last_error = e
                print(f"[] {model_name}: {e}")
        raise RuntimeError(f": {last_error}")