Reliability Patterns

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