AI Agent Fault Tolerance: Keep Running When Things Fail

AI Agent Fault Tolerance: Keep Running When Things Fail

💡 What You Will Learn

AI Agent Fault Tolerance: Keep Running When Things Fail

import time
import random
from functools import wraps

def retry_with_backoff(max_retries=3, base_delay=2):
    """Exponential BackoffRetryDecorator"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                    print(f"{attempt+1}{delay:.1f}... : {e}")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3, base_delay=2)
def call_llm_api(prompt):
    # APITimeout
    resp = openai_client.chat.completions.create(
        model="gpt-4o", messages=[{"role": "user", "content": prompt}],
        timeout=30
    )
    return resp.choices[0].message.content

# APIAuto/AutomaticRetry2s→4s→8s
# 95%3Retry99.99%

class ModelFallback:
    """→→"""
    MODELS = [
        {"provider": "openai", "model": "gpt-4o", "priority": 1},
        {"provider": "deepseek", "model": "deepseek-chat", "priority": 2},
        {"provider": "local", "model": "qwen2.5:7b", "priority": 3},
    ]

    def query(self, prompt, max_attempts=3):
        errors = []
        for model in self.MODELS[:max_attempts]:
            try:
                if model['provider'] == 'openai':
                    return call_openai(prompt)
                elif model['provider'] == 'deepseek':
                    return call_deepseek(prompt)
                else:
                    return call_local_ollama(prompt)
            except Exception as e:
                errors.append(f"{model['model']}: {e}")
                print(f"{model['model']}...")
                continue
        raise Exception(f": {errors}")

# 
# |  |  |  |  |
# |:----|:----|:----|:----|
# |  | GPT-4o | 1-2s | 100% |
# |  | DeepSeek | 0.8-1.5s | 95% |
# |  | Qwen2.5:7B | 3-5s | 80% |

import asyncio

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

    async def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "HALF_OPEN"
            else:
                raise Exception("")

        try:
            result = await asyncio.wait_for(func(*args, **kwargs), timeout=30)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
                print(f"{self.failure_count}")
            raise e

|:----|:------|:------|:----|