Reliability

Patterns for building reliable agent systems: circuit breakers, retries, timeouts, and failure handling.

Reliability Patterns

Retry Patterns

Exponential Backoff with Jitter

import asyncio
import random
from typing import TypeVar, Callable, Awaitable

T = TypeVar('T')

class RetryPolicy:
    def __init__(
        self,
        max_attempts: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        exponential_base: float = 2.0,
        jitter: bool = True,
        retryable_exceptions: tuple = (Exception,)
    ):
        self.max_attempts = max_attempts
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.exponential_base = exponential_base
        self.jitter = jitter
        self.retryable_exceptions = retryable_exceptions

    async def execute(self, func: Callable[..., Awaitable[T]], *args, **kwargs) -> T:
        last_exception = None

        for attempt in range(self.max_attempts):
            try:
                return await func(*args, **kwargs)
            except self.retryable_exceptions as e:
                last_exception = e

                if attempt == self.max_attempts - 1:
                    break

                delay = min(
                    self.base_delay * (self.exponential_base ** attempt),
                    self.max_delay
                )

                if self.jitter:
                    delay *= random.uniform(0.5, 1.5)

                await asyncio.sleep(delay)

        raise last_exception

# Usage
retry_policy = RetryPolicy(
    max_attempts=3,
    base_delay=1.0,
    max_delay=30.0,
    retryable_exceptions=(httpx.TimeoutException, httpx.ConnectError, 
                          httpx.HTTPStatusError)
)

result = await retry_policy.execute(llm_client.generate, prompt)

Retry with Conditions

class ConditionalRetryPolicy(RetryPolicy):
    def __init__(self, *args, should_retry: Callable[[Exception], bool] = None, **kwargs):
        super().__init__(*args, **kwargs)
        self.should_retry = should_retry or (lambda e: True)

    async def execute(self, func: Callable[..., Awaitable[T]], *args, **kwargs) -> T:
        for attempt in range(self.max_attempts):
            try:
                return await func(*args, **kwargs)
            except Exception as e:
                if not self.should_retry(e) or attempt == self.max_attempts - 1:
                    raise

                delay = self.calculate_delay(attempt)
                await asyncio.sleep(delay)

        raise last_exception

# Usage: only retry on specific conditions
def is_retryable(e: Exception) -> bool:
    if isinstance(e, httpx.HTTPStatusError):
        return e.response.status_code >= 500 or e.response.status_code == 429
    return isinstance(e, (httpx.TimeoutException, httpx.ConnectError))

retry_policy = ConditionalRetryPolicy(
    max_attempts=3,
    should_retry=is_retryable
)

Timeout Patterns

Per-Operation Timeout

async def with_timeout(coro: Awaitable[T], timeout: float) -> T:
    try:
        return await asyncio.wait_for(coro, timeout=timeout)
    except asyncio.TimeoutError:
        raise TimeoutError(f"Operation timed out after {timeout}s")

# Usage
result = await with_timeout(llm_client.generate(prompt), timeout=30.0)

Timeout Budget (End-to-End)

class TimeoutBudget:
    def __init__(self, total_timeout: float):
        self.total_timeout = total_timeout
        self.start_time = time.time()
        self.spent = 0.0

    @property
    def remaining(self) -> float:
        elapsed = time.time() - self.start_time
        return max(0, self.total_timeout - elapsed)

    def check(self) -> float:
        remaining = self.remaining
        if remaining <= 0:
            raise TimeoutError("Total timeout budget exceeded")
        return remaining

    @contextmanager
    def operation(self, name: str):
        start = time.time()
        try:
            yield self.remaining
        finally:
            elapsed = time.time() - start
            self.spent += elapsed

# Usage
budget = TimeoutBudget(30.0)

with budget.operation("planning") as remaining:
    plan = await planner.plan(input, timeout=remaining)

with budget.operation("execution") as remaining:
    result = await executor.execute(plan, timeout=remaining)

Timeout Propagation (Context)

import contextvars

timeout_budget: contextvars.ContextVar[TimeoutBudget] = contextvars.ContextVar('timeout_budget')

def get_timeout_budget() -> TimeoutBudget:
    return timeout_budget.get()

@contextmanager
def timeout_budget_context(budget: TimeoutBudget):
    token = timeout_budget.set(budget)
    try:
        yield
    finally:
        timeout_budget.reset(token)

# In agent code
async def agent_operation(input: str):
    budget = get_timeout_budget()
    remaining = budget.check()

    with budget.operation("llm_call"):
        result = await llm.generate(prompt, timeout=remaining)

    return result

Circuit Breaker

State Machine

from enum import Enum
from dataclasses import dataclass, field
import time

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing recovery

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5
    success_threshold: int = 2
    timeout: float = 30.0
    half_open_requests: int = 3

class CircuitBreaker:
    def __init__(self, name: str, config: CircuitBreakerConfig = None):
        self.name = name
        self.config = config or CircuitBreakerConfig()
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = 0
        self.half_open_requests = 0

    async def call(self, func: Callable[..., Awaitable[T]], *args, **kwargs) -> T:
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.config.timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_requests = 0
            else:
                raise CircuitOpenError(f"Circuit breaker {self.name} is OPEN")

        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_requests >= self.config.half_open_requests:
                raise CircuitOpenError("Half-open request limit reached")
            self.half_open_requests += 1

        try:
            result = await func()
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise

    def on_success(self):
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.config.success_threshold:
                self.state = CircuitState.CLOSED
                self.success_count = 0

    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        self.success_count = 0

        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
        elif self.failure_count >= self.config.failure_threshold:
            self.state = CircuitState.OPEN

class CircuitOpenError(Exception):
    pass

Circuit Breaker Registry

class CircuitBreakerRegistry:
    def __init__(self):
        self.breakers: dict[str, CircuitBreaker] = {}

    def get_or_create(self, name: str, config: CircuitBreakerConfig = None) -> CircuitBreaker:
        if name not in self.breakers:
            self.breakers[name] = CircuitBreaker(name, config)
        return self.breakers[name]

    def get_all_status(self) -> dict[str, dict]:
        return {
            name: {
                "state": cb.state.value,
                "failure_count": cb.failure_count,
                "success_count": cb.success_count,
                "last_failure": cb.last_failure_time
            }
            for name, cb in self.breakers.items()
        }

    async def reset_all(self):
        for cb in self.breakers.values():
            cb.state = CircuitState.CLOSED
            cb.failure_count = 0
            cb.success_count = 0

# Global registry
circuit_breaker_registry = CircuitBreakerRegistry()

Circuit Breaker for LLM Calls

class LLMCircuitBreaker:
    def __init__(self):
        self.breakers: dict[str, CircuitBreaker] = {}

    def get_breaker(self, model: str) -> CircuitBreaker:
        if model not in self.breakers:
            self.breakers[model] = CircuitBreaker(
                f"llm-{model}",
                CircuitBreakerConfig(
                    failure_threshold=3,
                    success_threshold=2,
                    timeout=60.0
                )
            )
        return self.breakers[model]

    async def call_with_breaker(self, model: str, func: Callable, *args, **kwargs):
        breaker = self.get_breaker(model)
        return await breaker.call(func, *args, **kwargs)

# Usage
llm_breaker = LLMCircuitBreaker()

result = await llm_breaker.call_with_breaker(
    "gpt-4o",
    llm_client.generate,
    prompt
)

Bulkhead Pattern

Thread Pool Isolation

from concurrent.futures import ThreadPoolExecutor
import asyncio

class BulkheadExecutor:
    def __init__(self, name: str, max_workers: int = 10):
        self.name = name
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.semaphore = asyncio.Semaphore(max_workers)
        self.active_tasks = 0

    async def execute(self, func: Callable, *args, **kwargs):
        async with self.semaphore:
            self.active_tasks += 1
            try:
                loop = asyncio.get_event_loop()
                return await loop.run_in_executor(self.executor, func, *args, **kwargs)
            finally:
                self.active_tasks -= 1

    def get_stats(self) -> dict:
        return {
            "name": self.name,
            "max_workers": self.executor._max_workers,
            "active_tasks": self.active_tasks,
            "queue_size": len(self.executor._work_queue)
        }

# Usage: Isolate different tool categories
bulkheads = {
    "llm": BulkheadExecutor("llm", max_workers=20),
    "code_exec": BulkheadExecutor("code_exec", max_workers=5),
    "web_search": BulkheadExecutor("web_search", max_workers=10),
    "database": BulkheadExecutor("database", max_workers=10),
    "file_io": BulkheadExecutor("file_io", max_workers=10)
}

async def execute_tool(tool_name: str, func: Callable, *args, **kwargs):
    bulkhead = bulkheads.get(tool_name, bulkheads["default"])
    return await bulkhead.execute(func)

Connection Pool Bulkhead

class ConnectionPoolBulkhead:
    def __init__(self, name: str, max_connections: int = 100):
        self.name = name
        self.max_connections = max_connections
        self.semaphore = asyncio.Semaphore(max_connections)
        self.active_connections = 0

    @asynccontextmanager
    async def connection(self):
        await self.semaphore.acquire()
        self.active_connections += 1
        try:
            yield self
        finally:
            self.active_connections -= 1
            self.semaphore.release()

    def get_stats(self) -> dict:
        return {
            "name": self.name,
            "max_connections": self.max_connections,
            "active": self.active_connections,
            "available": self.max_connections - self.active_connections,
            "utilization": self.active_connections / self.max_connections
        }

# Per-service connection pools
http_pools = {
    "openai": ConnectionPoolBulkhead("openai", max_connections=50),
    "anthropic": ConnectionPoolBulkhead("anthropic", max_connections=30),
    "database": ConnectionPoolBulkhead("database", max_connections=20),
    "redis": ConnectionPoolBulkhead("redis", max_connections=50),
    "external_api": ConnectionPoolBulkhead("external", max_connections=10)
)

async def http_request(service: str, method: str, url: str, **kwargs):
    pool = http_pools.get(service)
    if not pool:
        raise ValueError(f"Unknown service: {service}")

    async with pool.connection():
        async with httpx.AsyncClient() as client:
            return await client.request(method, url, **kwargs)

Idempotency

Idempotency Keys

import hashlib
import json

class IdempotencyManager:
    def __init__(self, redis: Redis, ttl: int = 86400):  # 24 hours
        self.redis = redis
        self.ttl = ttl

    def generate_key(self, operation: str, payload: dict) -> str:
        payload_str = json.dumps(payload, sort_keys=True)
        return f"idempotency:{operation}:{hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()[:16]}"

    async def check_and_store(self, key: str, result: Any) -> tuple[bool, Any]:
        """Returns (is_new, result)"""
        # Check if exists
        existing = await self.redis.get(key)
        if existing:
            return False, json.loads(existing)

        # Store result
        await self.redis.setex(key, self.ttl, json.dumps(result, default=str))
        return True, None

    def generate_key(self, operation: str, *args, **kwargs) -> str:
        payload = {"args": args, "kwargs": kwargs}
        return self.generate_key(operation, payload)

# Usage
idempotency = IdempotencyManager(redis_client)

async def idempotent_operation(operation: str, func: Callable, *args, **kwargs):
    key = idempotency.generate_key("operation", *args, **kwargs)
    is_new, cached = await idempotency.check_and_store(key, None)

    if not is_new:
        return cached  # Return cached result

    result = await func(*args, **kwargs)
    await idempotency.redis.setex(key, 86400, json.dumps(result, default=str))
    return result

Idempotent HTTP Calls

class IdempotentHTTPClient:
    def __init__(self, client: httpx.AsyncClient, idempotency: IdempotencyManager):
        self.client = client
        self.idempotency = idempotency

    async def post(self, url: str, json: dict, idempotency_key: str = None) -> httpx.Response:
        key = idempotency_key or self.idempotency.generate_key("http_post", {"url": url, "json": json})

        is_new, cached = await self.idempotency.check_and_store(key, None)
        if not is_new:
            return cached  # Return cached response

        # Add idempotency key header
        headers = {"Idempotency-Key": key}
        response = await self.client.post(url, json=json, headers=headers)

        # Cache successful responses
        if response.is_success:
            await self.idempotency.redis.setex(
                key, 86400, json.dumps({
                    "status": response.status_code,
                    "body": response.json()
                }, default=str)
            )

        return response

Graceful Degradation

Fallback Patterns

class FallbackManager:
    def __init__(self):
        self.fallbacks: dict[str, Callable] = {}

    def register_fallback(self, operation: str, fallback: Callable):
        self.fallbacks[operation] = fallback

    async def execute_with_fallback(self, operation: str, primary: Callable, *args, **kwargs):
        try:
            return await primary()
        except Exception as e:
            if operation in self.fallbacks:
                logger.warning(f"Primary failed, using fallback for {operation}: {e}")
                return await self.fallbacks[operation]()
            raise

# Usage
fallback_manager = FallbackManager()

fallback_manager.register_fallback(
    "llm_generate",
    lambda: "I'm experiencing high demand. Please try again later."
)

fallback_manager.register_fallback(
    "web_search",
    lambda: [{"title": "Cached result", "url": "https://example.com", "snippet": "Cached result"}]
)

# Usage
result = await fallback_manager.execute_with_fallback(
    "llm_generate",
    lambda: llm.generate(prompt)
)

Cached Fallback

class CachedFallback:
    def __init__(self, cache: Cache, ttl: int = 300):
        self.cache = cache
        self.ttl = ttl

    async def get_or_compute(self, key: str, compute_fn: Callable) -> Any:
        # Try cache first
        cached = await self.cache.get(key)
        if cached is not None:
            return cached

        try:
            result = await compute_fn()
            await self.cache.set(key, result, ttl=self.ttl)
            return result
        except Exception as e:
            # Try stale cache on failure
            stale = await self.cache.get(key, allow_stale=True)
            if stale is not None:
                logger.warning(f"Using stale cache for {key}: {e}")
                return stale
            raise

Feature Flags for Degradation

class FeatureFlagDegradation:
    def __init__(self, flag_client: FeatureFlagClient):
        self.flags = flag_client

    async def execute_with_degradation(self, feature: str, 
                                        full_fn: Callable, 
                                        degraded_fn: Callable,
                                        minimal_fn: Callable) -> Any:
        # Check feature flags
        if await self.flags.is_enabled(f"{feature}.full"):
            return await full_fn()
        elif await self.flags.is_enabled(f"{feature}.degraded"):
            return await degraded_fn()
        else:
            return await minimal_fn()

# Usage
degradation = FeatureFlagDegradation(flag_client)

result = await degradation.execute_with_degradation(
    "code_generation",
    full_fn=lambda: agent.generate_code(prompt, full_context=True),
    degraded_fn=lambda: agent.generate_code(prompt, max_tokens=1000),
    minimal_fn=lambda: "Code generation temporarily limited. Please try again later."
)

Dead Letter Queue

DLQ Implementation

class DeadLetterQueue:
    def __init__(self, redis: Redis, queue_name: str):
        self.redis = redis
        self.dlq_name = f"{queue_name}:dlq"
        self.max_retries = 3

    async def add(self, task: Task, error: Exception, attempts: int):
        dlq_entry = {
            "task": task.to_dict(),
            "error": str(error),
            "error_type": type(error).__name__,
            "attempts": attempts,
            "failed_at": datetime.utcnow().isoformat(),
            "stack_trace": traceback.format_exc()
        }
        await self.redis.lpush(self.dlq_name, json.dumps(dlq_entry))

    async def get_failed(self, limit: int = 100) -> list[DLQEntry]:
        entries = await self.redis.lrange(self.dlq_name, 0, limit - 1)
        return [DLQEntry(**json.loads(e)) for e in entries]

    async def retry(self, entry: DLQEntry) -> bool:
        """Move back to main queue for retry"""
        task = Task.from_dict(entry["task"])
        await self.main_queue.enqueue(task)
        await self.redis.lrem(self.dlq_name, 1, json.dumps(entry.to_dict()))
        return True

    async def purge(self, older_than_days: int = 30):
        cutoff = datetime.utcnow() - timedelta(days=older_than_days)
        # Lua script for atomic purge
        lua_script = """
        local entries = redis.call('LRANGE', KEYS[1], 0, -1)
        local removed = 0
        for i, entry in ipairs(entries) do
            local data = cjson.decode(entry)
            if data.failed_at < ARGV[1] then
                redis.call('LREM', KEYS[1], 1, entry)
                removed = removed + 1
            end
        end
        return removed
        """
        await self.redis.eval(lua_script, 1, self.dlq_name, cutoff.isoformat())

DLQ Monitoring

class DLQMonitor:
    def __init__(self, dlq: DeadLetterQueue):
        self.dlq = dlq

    async def check_health(self) -> DLQHealth:
        entries = await self.dlq.get_failed(limit=1000)

        by_error_type = {}
        for entry in entries:
            error_type = entry.error_type
            by_error_type[error_type] = by_error_type.get(error_type, 0) + 1

        return DLQHealth(
            total_entries=len(entries),
            by_error_type=by_error_type,
            oldest_entry=min(e.failed_at for e in entries) if entries else None,
            newest_entry=max(e.failed_at for e in entries) if entries else None,
            retryable_count=sum(1 for e in entries if e.attempts < 3)
        )

Health Checks

Liveness & Readiness

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class HealthResponse(BaseModel):
    status: str  # healthy, degraded, unhealthy
    checks: dict[str, CheckResult]
    timestamp: datetime

class CheckResult(BaseModel):
    status: str  # pass, warn, fail
    component: str
    message: str
    latency_ms: float = None

@app.get("/health/live")
async def liveness():
    """Liveness: Is the process alive?"""
    return HealthResponse(
        status="healthy",
        checks={"process": CheckResult(status="pass", component="process", message="Alive")},
        timestamp=datetime.utcnow()
    )

@app.get("/health/ready")
async def readiness():
    """Readiness: Can serve traffic?"""
    checks = {}

    # Database
    try:
        start = time.time()
        await db.execute("SELECT 1")
        checks["database"] = CheckResult(
            status="pass", component="database", 
            message="Connected", latency_ms=(time.time()-start)*1000
        )
    except Exception as e:
        checks["database"] = CheckResult(
            status="fail", component="database", message=str(e)
        )

    # Redis
    try:
        start = time.time()
        await redis.ping()
        checks["redis"] = CheckResult(
            status="pass", component="redis",
            message="Connected", latency_ms=(time.time()-start)*1000
        )
    except Exception as e:
        checks["redis"] = CheckResult(status="fail", component="redis", message=str(e))

    # LLM API
    try:
        start = time.time()
        await llm.health_check()
        checks["llm"] = CheckResult(
            status="pass", component="llm", message="Available",
            latency_ms=(time.time()-start)*1000
        )
    except Exception as e:
        checks["llm"] = CheckResult(status="warn", component="llm", message=str(e))

    # Overall status
    statuses = [c.status for c in checks.values()]
    if all(s == "pass" for s in statuses):
        overall = "healthy"
    elif any(s == "fail" for s in statuses):
        overall = "unhealthy"
    else:
        overall = "degraded"

    return HealthResponse(
        status=overall,
        checks=checks,
        timestamp=datetime.utcnow()
    )

Chaos Engineering

Failure Injection

class ChaosEngine:
    def __init__(self):
        self.faults: dict[str, FaultConfig] = {}

    def inject_latency(self, service: str, latency_ms: int, rate: float = 1.0):
        self.faults[f"latency:{service}"] = FaultConfig(
            type="latency", latency_ms=latency_ms, rate=rate
        )

    def inject_error(self, service: str, error_rate: float, 
                     status_code: int = 500):
        self.faults[f"error:{service}"] = FaultConfig(
            type="error", error_rate=error_rate, status_code=status_code
        )

    def inject_failure(self, service: str, exception: Exception):
        self.faults[f"failure:{service}"] = FaultConfig(
            type="exception", exception=exception
        )

    def clear(self, service: str = None):
        if service:
            keys = [k for k in self.faults if k.endswith(f":{service}")]
            for k in keys:
                del self.faults[k]
        else:
            self.faults.clear()

    def should_fail(self, service: str) -> tuple[bool, Any]:
        for key, fault in self.faults.items():
            if key.endswith(f":{service}"):
                if random.random() < fault.rate:
                    if fault.type == "latency":
                        return True, ("delay", fault.latency_ms)
                    elif fault.type == "error":
                        return True, ("error", fault.status_code)
                    elif fault.type == "exception":
                        return True, ("exception", fault.exception)
        return False, None

# Usage in middleware
@app.middleware("http")
async def chaos_middleware(request: Request, call_next):
    should_fail, fault = chaos_engine.should_fail(request.url.path)
    if should_fail:
        if fault[0] == "delay":
            await asyncio.sleep(fault[1] / 1000)
        elif fault[0] == "error":
            return Response(status_code=fault[1], content="Injected error")
        elif fault[0] == "exception":
            raise fault[1]
    return await call_next(request)

Related Notes

• Deployment Architecture — Deployment patterns
• Scaling Strategies — Scaling strategies
• Observability — Monitoring, alerting
• Security — Security
• Cost Optimization — Cost optimization
• Compliance — Compliance
• Deployment Architecture — Deployment
• CI CD Best Practices — CI/CD
• Infrastructure as Code — IaC
• Observability — Observability
• Git Workflow — GitOps
• Supply Chain Security — Supply chain security
• Secure Coding — Secure coding
• Research Directions — Reliability research
• Safety Coordination — Multi-agent safety
• Coordination Patterns — Coordination reliability
• Red Teaming — Red teaming for reliability

Sources: "Release It!" (Nygard), "Site Reliability Engineering" (Google), "Building Microservices" (Newman), "Designing Data-Intensive Applications" (Kleppmann), "Patterns of Enterprise Application Architecture" (Fowler), Hystrix, Resilience4j, Polly, Circuit Breaker pattern, Bulkhead pattern, Retry patterns, Timeout patterns, Netflix Hystrix, AWS Well-Architected Reliability Pillar, Azure Reliability Patterns, Google SRE Book, Chaos Engineering (Netflix), Gremlin, Chaos Monkey, Litmus, PowerfulSeal, Istio Fault Injection, Linkerd Failure Injection.