Reliability Patterns: Building Dependable Systems

Overview

Reliability means systems work consistently, recover gracefully, and don't lose data when things go wrong.

---

Error Handling Strategy

class ReliableAgent:
    def execute_with_recovery(self, task):
        """Execute with error handling and recovery"""

        for attempt in range(self.max_retries):
            try:
                result = self.execute(task)
                return result

            except TransientError as e:
                # Retry with backoff
                wait_time = self.calculate_backoff(attempt)
                time.sleep(wait_time)

            except PermanentError as e:
                # Don't retry, fallback instead
                return self.fallback(task, e)

            except Exception as e:
                # Unknown error
                self.log_unexpected(e)
                return {"error": str(e)}

    def calculate_backoff(self, attempt):
        """Exponential backoff: 1s, 2s, 4s, 8s..."""
        return min(2 ** attempt, 60)  # Cap at 60s

---

Circuit Breaker

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_count = 0
        self.threshold = failure_threshold
        self.state = 'closed'  # closed=healthy, open=failing
        self.reset_timeout = reset_timeout
        self.last_failure_time = None

    def call(self, func, *args, **kwargs):
        if self.state == 'open':
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = 'half-open'  # Try again
            else:
                raise CircuitBreakerOpen("Service temporarily unavailable")

        try:
            result = func(*args, **kwargs)
            self.on_success()
            return result

        except Exception as e:
            self.on_failure()
            raise

    def on_success(self):
        self.failure_count = 0
        self.state = 'closed'

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

        if self.failure_count >= self.threshold:
            self.state = 'open'  # Stop trying

---

Health Checks

class HealthChecker:
    def check_agent_health(self) -> dict:
        """Check if agent is healthy"""

        health = {
            'status': 'healthy',
            'checks': {}
        }

        # Check 1: LLM connection
        if not self.test_llm_connection():
            health['status'] = 'degraded'
            health['checks']['llm'] = 'failed'
        else:
            health['checks']['llm'] = 'ok'

        # Check 2: Tool connectivity
        for tool in self.tools:
            if not self.test_tool(tool):
                health['status'] = 'degraded'
                health['checks'][tool.name] = 'failed'

        # Check 3: Memory availability
        if not self.has_memory_available():
            health['status'] = 'unhealthy'
            health['checks']['memory'] = 'critical'

        return health

---

3 Warnings ⚠️

Warning 1: Retry Forever

# ❌ WRONG
while True:
    try:
        result = api.call()
        break
    except:
        continue  # Infinite retry loop!

# ✅ RIGHT
max_retries = 3
for attempt in range(max_retries):
    try:
        result = api.call()
        break
    except TransientError:
        if attempt < max_retries - 1:
            time.sleep(2 ** attempt)

Warning 2: Silent Failures

# ❌ WRONG
try:
    critical_operation()
except:
    pass  # Ignore error, continue

# Data corruption!

# ✅ RIGHT
try:
    critical_operation()
except Exception as e:
    self.log_error(e)
    self.alert('Critical operation failed')
    return {"error": str(e)}

Warning 3: No Monitoring of Recovery

# ❌ WRONG
for attempt in range(max_retries):
    if operation():
        return
# No insight into how many retries needed

# ✅ RIGHT
for attempt in range(max_retries):
    if operation():
        self.metrics.record_retries_needed(attempt)
        return
# Track retry frequency for improvement

---

Last Updated: August 9, 2026