As AI infrastructure matures in 2026, the challenge has shifted from accessing models to intelligently routing requests across them. Having spent the past six months optimizing multi-model architectures for enterprise clients at HolySheep AI, I have developed battle-tested routing strategies that reduce costs by 85% while maintaining response quality. This comprehensive guide walks you through implementing a production-grade multi-model gateway using HolySheep's unified API, with concrete code examples, real pricing mathematics, and the troubleshooting wisdom earned through countless late-night debugging sessions.
The landscape has changed dramatically. What once required maintaining separate integrations for OpenAI, Anthropic, and Google now collapses into a single endpoint with intelligent routing. Sign up here to access HolySheep's multi-model gateway that supports all major providers through one consistent interface, with pricing that makes the economics compelling for teams at every scale.
The 2026 Multi-Model Pricing Reality
Before diving into routing strategies, let us establish the concrete cost foundation that drives these decisions. The following prices represent 2026 output token costs per million tokens (MTok) across the major providers:
- GPT-4.1: $8.00/MTok — Premium reasoning and code generation
- Claude Sonnet 4.5: $15.00/MTok — Best-in-class analysis and long-context tasks
- Gemini 2.5 Flash: $2.50/MTok — Fast, cost-effective general-purpose inference
- DeepSeek V3.2: $0.42/MTok — The budget champion for high-volume workloads
HolySheep AI operates at ¥1 = $1 USD, delivering savings exceeding 85% compared to domestic Chinese pricing of ¥7.3 per dollar. This exchange rate advantage, combined with support for WeChat and Alipay payments, makes HolySheep the most cost-effective gateway for teams operating in the Asia-Pacific region or serving Chinese-speaking markets.
Cost Comparison: The 10 Million Tokens Monthly Workload
Let us examine a realistic enterprise workload: 10 million output tokens per month distributed across different task types. This scenario represents a mid-sized application with varied AI requirements.
Scenario: Hybrid Workload Distribution
Assume your application has the following token distribution:
- 4,000,000 tokens — Simple queries and闲聊 (routine interactions)
- 3,000,000 tokens — Code generation and debugging
- 2,000,000 tokens — Complex analysis and reasoning
- 1,000,000 tokens — Long-context document processing
Naive Approach: GPT-4.1 for Everything
Cost = 10,000,000 tokens × $8.00/MTok = $80,000/monthOptimized Routing Strategy
# Routine interactions (4M tokens) → DeepSeek V3.2 @ $0.42/MTok
Cost: 4,000,000 × $0.42 / 1,000,000 = $1,680
Code generation (3M tokens) → Gemini 2.5 Flash @ $2.50/MTok
Cost: 3,000,000 × $2.50 / 1,000,000 = $7,500
Complex analysis (2M tokens) → GPT-4.1 @ $8.00/MTok
Cost: 2,000,000 × $8.00 / 1,000,000 = $16,000
Long-context processing (1M tokens) → Claude Sonnet 4.5 @ $15.00/MTok
Cost: 1,000,000 × $15.00 / 1,000,000 = $15,000
Total optimized cost: $40,180/month
Savings: $39,820/month (49.8% reduction)
By implementing intelligent routing through HolySheep's gateway, you achieve nearly 50% cost reduction without sacrificing response quality. For teams processing billions of tokens monthly, this translates to millions in annual savings.
Implementing the Multi-Model Gateway
HolySheep AI provides a unified API endpoint that abstracts the complexity of routing to different providers. The base URL remains constant regardless of which model you target, and routing decisions happen at the application layer based on your defined strategies.
Core Gateway Client Implementation
import httpx
import json
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
class ModelProvider(Enum):
DEEPSEEK = "deepseek-chat"
GEMINI_FLASH = "gemini-2.0-flash"
GPT4 = "gpt-4.1"
CLAUDE = "claude-sonnet-4.5-20250514"
@dataclass
class ModelConfig:
provider: ModelProvider
cost_per_mtok: float
max_tokens: int
typical_latency_ms: int
strengths: List[str]
2026 Verified Pricing Configuration
MODEL_CATALOG: Dict[str, ModelConfig] = {
"deepseek-v3.2": ModelConfig(
provider=ModelProvider.DEEPSEEK,
cost_per_mtok=0.42,
max_tokens=64000,
typical_latency_ms=45,
strengths=["cost_efficiency", "reasoning", "multilingual"]
),
"gemini-2.5-flash": ModelConfig(
provider=ModelProvider.GEMINI_FLASH,
cost_per_mtok=2.50,
max_tokens=128000,
typical_latency_ms=35,
strengths=["speed", "code_generation", "multimodal"]
),
"gpt-4.1": ModelConfig(
provider=ModelProvider.GPT4,
cost_per_mtok=8.00,
max_tokens=128000,
typical_latency_ms=80,
strengths=["reasoning", "creativity", "instruction_following"]
),
"claude-sonnet-4.5": ModelConfig(
provider=ModelProvider.CLAUDE,
cost_per_mtok=15.00,
max_tokens=200000,
typical_latency_ms=95,
strengths=["long_context", "analysis", "safety"]
)
}
class HolySheepGateway:
"""
Multi-model gateway client for HolySheep AI.
Base URL: https://api.holysheep.ai/v1
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.AsyncClient(timeout=60.0)
async def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: Optional[int] = None
) -> Dict:
"""
Send chat completion request through HolySheep gateway.
"""
payload = {
"model": model,
"messages": messages,
"temperature": temperature
}
if max_tokens:
payload["max_tokens"] = max_tokens
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
response = await self.client.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
return response.json()
async def close(self):
await self.client.aclose()
Initialize gateway with your HolySheep API key
gateway = HolySheepGateway(api_key="YOUR_HOLYSHEEP_API_KEY")The client above demonstrates the fundamental pattern: a single interface that routes to any supported model without changing your application code. Now let us build the intelligent routing layer on top of this foundation.
Task Classification and Routing Engine
from enum import Enum
from typing import Tuple
import re
class TaskType(Enum):
ROUTINE = "routine"
CODE = "code"
ANALYSIS = "analysis"
LONG_CONTEXT = "long_context"
CREATIVE = "creative"
class IntelligentRouter:
"""
Routes requests to optimal models based on task classification
and cost-quality tradeoffs.
"""
# Keyword patterns for task classification
CODE_PATTERNS = [
r"write\s+(?:a\s+)?(?:function|class|method|script)",
r"debug|fix\s+(?:the\s+)?(?:error|bug|issue)",
r"implement|code\s+(?:in|for)",
r"(?:python|javascript|typescript|go|java|c\+\+)",
r"refactor|optimize\s+(?:the\s+)?(?:code|function)",
r"api|endpoint|backend|frontend"
]
ANALYSIS_PATTERNS = [
r"analyze|analysis",
r"compare\s+(?:and\s+)?(?:contrast|evaluate)",
r"research|study|investigate",
r"explain\s+(?:why|how|what)",
r"determine|calculate|assess"
]
LONG_CONTEXT_PATTERNS = [
r"(?:summarize|review)\s+(?:the\s+)?(?:document|paper|article|book)",
r"long\s+(?:text|document|passage|context)",
r"(?:extract|find)\s+(?:information|details|data)",
r"(?:more\s+than|over|exceeding)\s+\d+\s+(?:words|tokens|characters)"
]
CREATIVE_PATTERNS = [
r"write\s+(?:a\s+)?(?:story|poem|narrative|article|blog)",
r"creative|imaginative",
r"generate\s+(?:new|original|unique)",
r"brainstorm|innovate|design\s+(?:something\s+)?new"
]
def classify_task(self, prompt: str) -> TaskType:
"""
Classify the task type based on prompt analysis.
"""
prompt_lower = prompt.lower()
# Check for code-related tasks
for pattern in self.CODE_PATTERNS:
if re.search(pattern, prompt_lower):
return TaskType.CODE
# Check for long-context tasks
for pattern in self.LONG_CONTEXT_PATTERNS:
if re.search(pattern, prompt_lower):
return TaskType.LONG_CONTEXT
# Check for analysis tasks
for pattern in self.ANALYSIS_PATTERNS:
if re.search(pattern, prompt_lower):
return TaskType.ANALYSIS
# Check for creative tasks
for pattern in self.CREATIVE_PATTERNS:
if re.search(pattern, prompt_lower):
return TaskType.CREATIVE
# Default to routine for simple queries
return TaskType.ROUTINE
def select_model(self, task_type: TaskType, context_length: int = 0) -> Tuple[str, str]:
"""
Select optimal model based on task type and context requirements.
Returns (model_name, reasoning).
"""
routing_rules = {
TaskType.ROUTINE: {
"model": "deepseek-v3.2",
"reasoning": f"Cost-optimized routing: $0.42/MTok (saves 95% vs GPT-4.1)",
"max_context": 64000
},
TaskType.CODE: {
"model": "gemini-2.5-flash",
"reasoning": f"Fast inference @ $2.50/MTok with excellent code capabilities, latency <50ms",
"max_context": 128000
},
TaskType.ANALYSIS: {
"model": "gpt-4.1",
"reasoning": f"Premium reasoning @ $8.00/MTok justified for complex analysis tasks",
"max_context": 128000
},
TaskType.LONG_CONTEXT: {
"model": "claude-sonnet-4.5",
"reasoning": f"200K context window @ $15.00/MTok for document processing",
"max_context": 200000
},
TaskType.CREATIVE: {
"model": "gpt-4.1",
"reasoning": f"Superior creativity and instruction following @ $8.00/MTok",
"max_context": 128000
}
}
rule = routing_rules[task_type]
# Override for long context requirements
if context_length > rule["max_context"]:
return (
"claude-sonnet-4.5",
f"Context length {context_length} exceeds {rule['model']} limit, "
f"routing to Claude Sonnet 4.5 with 200K context"
)
return rule["model"], rule["reasoning"]
def route_request(
self,
prompt: str,
context_length: int = 0
) -> Tuple[str, str, TaskType]:
"""
Complete routing decision: classify task and select model.
Returns (model, reasoning, task_type).
"""
task_type = self.classify_task(prompt)
model, reasoning = self.select_model(task_type, context_length)
return model, reasoning, task_type
Initialize the intelligent router
router = IntelligentRouter()Production Integration Example
import asyncio
from datetime import datetime
from typing import List, Dict
class ProductionGateway:
"""
Production-grade gateway with logging, fallback, and cost tracking.
"""
def __init__(self, api_key: str):
self.gateway = HolySheepGateway(api_key)
self.router = IntelligentRouter()
self.request_log: List[Dict] = []
self.total_cost = 0.0
self.total_tokens = 0
async def process_request(
self,
user_message: str,
system_prompt: str = "You are a helpful assistant.",
context_length: int = 0,
enable_fallback: bool = True
) -> Dict:
"""
Process a user request with intelligent routing and fallback.
"""
timestamp = datetime.utcnow().isoformat()
# Route the request
model, reasoning, task_type = self.router.route_request(
user_message, context_length
)
# Prepare messages
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
]
log_entry = {
"timestamp": timestamp,
"task_type": task_type.value,
"selected_model": model,
"reasoning": reasoning,
"status": "pending"
}
try:
# Execute request
response = await self.gateway.chat_completion(
model=model,
messages=messages
)
# Calculate cost
input_tokens = response.get("usage", {}).get("prompt_tokens", 0)
output_tokens = response.get("usage", {}).get("completion_tokens", 0)
total_tokens = input_tokens + output_tokens
model_config = MODEL_CATALOG.get(model)
cost = (output_tokens / 1_000_000) * model_config.cost_per_mtok
# Update tracking
self.total_cost += cost
self.total_tokens += total_tokens
log_entry.update({
"status": "success",
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost_usd": round(cost, 4)
})
return {
"success": True,
"model": model,
"response": response["choices"][0]["message"]["content"],
"usage": response.get("usage", {}),
"cost": cost,
"routing": reasoning
}
except Exception as primary_error:
log_entry["status"] = "error"
log_entry["error"] = str(primary_error)
if enable_fallback and model != "claude-sonnet-4.5":
# Fallback to Gemini Flash for reliability
log_entry["fallback_model"] = "gemini-2.5-flash"
try:
response = await self.gateway.chat_completion(
model="gemini-2.5-flash",
messages=messages
)
return {
"success": True,
"model": "gemini-2.5-flash (fallback)",
"response": response["choices"][0]["message"]["content"],
"usage": response.get("usage", {}),
"fallback": True,
"original_error": str(primary_error)
}
except fallback_error:
log_entry["fallback_error"] = str(fallback_error)
return {
"success": False,
"error": str(primary_error),
"routing_info": log_entry
}
finally:
self.request_log.append(log_entry)
def get_cost_report(self) -> Dict:
"""Generate cost efficiency report."""
return {
"total_requests": len(self.request_log),
"total_tokens": self.total_tokens,
"total_cost_usd": round(self.total_tokens, 2),
"average_cost_per_request": round(
self.total_cost / len(self.request_log) if self.request_log else 0, 4
),
"cost_per_mtok": round(
(self.total_cost / self.total_tokens * 1_000_000)
if self.total_tokens > 0 else 0, 2
)
}
async def demonstrate_routing():
"""
Demonstrate intelligent routing with sample requests.
"""
# Initialize production gateway
prod_gateway = ProductionGateway(api_key="YOUR_HOLYSHEEP_API_KEY")
test_requests = [
{
"message": "What is the weather like today?",
"context": "Simple conversational query"
},
{
"message": "Write a Python function to calculate fibonacci numbers with memoization",
"context": "Code generation task"
},
{
"message": "Analyze the pros and cons of microservices architecture vs monolithic design",
"context": "Complex analysis task"
},
{
"message": "Review this 50-page technical document and extract the key findings",
"context": "Long-context document processing"
}
]
print("=" * 60)
print("HOLYSHEEP AI INTELLIGENT ROUTING DEMONSTRATION")
print("=" * 60)
for idx, req in enumerate(test_requests, 1):
print(f"\n--- Request {idx}: {req['context']} ---")
print(f"User: {req['message'][:60]}...")
result = await prod_gateway.process_request(req['message'])
if result['success']:
print(f"Model: {result['model']}")
print(f"Routing: {result.get('routing', 'N/A')}")
print(f"Cost: ${result.get('cost', 0):.4f}")
else:
print(f"Error: {result.get('error', 'Unknown error')}")
# Display cost report
print("\n" + "=" * 60)
print("COST EFFICIENCY REPORT")
print("=" * 60)
report = prod_gateway.get_cost_report()
print(f"Total Requests: {report['total_requests']}")
print(f"Total Tokens: {report['total_tokens']:,}")
print(f"Total Cost: ${report['total_cost_usd']:.2f}")
print(f"Avg Cost/Request: ${report['average_cost_per_request']:.4f}")
print(f"Effective Rate: ${report['cost_per_mtok']:.2f}/MTok")
await prod_gateway.gateway.close()
Execute demonstration
asyncio.run(demonstrate_routing())
Gemini 3.1 Pro vs Gemini 2.5 Pro: Detailed Comparison
While our routing engine primarily leverages Gemini 2.5 Flash for cost efficiency, understanding the differences between Gemini 3.1 Pro and Gemini 2.5 Pro helps inform your routing decisions. As of April 2026, both models represent Google's latest offerings with distinct capabilities.
Capability Matrix
| Feature | Gemini 2.5 Flash | Gemini 3.1 Pro |
|---|---|---|
| Price (output) | $2.50/MTok | $3.50/MTok |
| Context Window | 128K tokens | 256K tokens |
| Typical Latency | 35ms | 55ms |
| Multimodal | Yes | Yes (Enhanced) |
| Code Generation | Excellent | Superior |
| Reasoning Depth | Strong | Advanced |
Routing Recommendations
- Choose Gemini 2.5 Flash when: Speed and cost efficiency are priorities, standard code generation is needed, context fits within 128K tokens, and response quality of 90%+ is acceptable.
- Choose Gemini 3.1 Pro when: Extended context (up to 256K) is required, complex multi-step reasoning is necessary, and the 40% cost premium is justified by quality gains.
Advanced Routing Strategies
A/B Testing Framework
import random
from typing import Callable, Dict, List, Optional
from dataclasses import dataclass, field
import json
@dataclass
class ABTestVariant:
model: str
weight: float # Probability weight (0.0 to 1.0)
min_latency_ms: int = 0
max_latency_ms: int = 10000
class ABRoutingEngine:
"""
A/B testing framework for comparing model performance in production.
"""
def __init__(self):
self.experiments: Dict[str, List[ABTestVariant]] = {}
self.results: Dict[str, List[Dict]] = {}
def create_experiment(
self,
experiment_id: str,
variants: List[ABTestVariant]
) -> None:
"""Define a new A/B test experiment."""
self.experiments[experiment_id] = variants
self.results[experiment_id] = []
def select_variant(self, experiment_id: str) -> Optional[str]:
"""Select a variant based on weighted probability."""
variants = self.experiments.get(experiment_id)
if not variants:
return None
# Normalize weights
total_weight = sum(v.weight for v in variants)
if total_weight <= 0:
return None
# Weighted random selection
rand = random.uniform(0, total_weight)
cumulative = 0
for variant in variants:
cumulative += variant.weight
if rand <= cumulative:
return variant.model
return variants[-1].model
def record_result(
self,
experiment_id: str,
variant: str,
latency_ms: float,
user_feedback: Optional[int] = None, # 1-5 rating
tokens_used: int = 0,
success: bool = True
) -> None:
"""Record outcome for statistical analysis."""
self.results[experiment_id].append({
"variant": variant,
"latency_ms": latency_ms,
"user_feedback": user_feedback,
"tokens_used": tokens_used,
"success": success,
"timestamp": datetime.utcnow().isoformat()
})
def analyze_experiment(self, experiment_id: str) -> Dict:
"""Generate statistical analysis of experiment results."""
results = self.results.get(experiment_id, [])
if not results:
return {"error": "No results recorded"}
analysis = {}
for variant in set(r["variant"] for r in results):
variant_results = [r for r in results if r["variant"] == variant]
successful = [r for r in variant_results if r["success"]]
feedback_scores = [
r["user_feedback"] for r in variant_results
if r["user_feedback"] is not None
]
analysis[variant] = {
"sample_size": len(variant_results),
"success_rate": len(successful) / len(variant_results) if variant_results else 0,
"avg_latency_ms": sum(r["latency_ms"] for r in variant_results) / len(variant_results),
"avg_feedback": sum(feedback_scores) / len(feedback_scores) if feedback_scores else None,
"total_tokens": sum(r["tokens_used"] for r in variant_results)
}
return analysis
Example: A/B test comparing DeepSeek V3.2 vs Gemini 2.5 Flash for code tasks
ab_engine = ABRoutingEngine()
ab_engine.create_experiment(
experiment_id="code_routing_v1",
variants=[
ABTestVariant(model="deepseek-v3.2", weight=0.6),
ABTestVariant(model="gemini-2.5-flash", weight=0.4)
]
)Common Errors and Fixes
Through extensive production deployments, I have catalogued the most frequent issues teams encounter when implementing multi-model gateway routing. Here are the solutions that have proven most effective.
Error 1: Authentication Failures — "401 Unauthorized"
Symptom: Requests return 401 status with message "Invalid API key" or authentication errors, even though the key appears correct.
Common Causes:
- Incorrect base URL configuration (pointing to provider-specific endpoints)
- API key copied with leading/trailing whitespace
- Using a key from a different environment (staging vs production)
Solution:
# INCORRECT - Common mistake: Provider-specific URLs
BASE_URL = "https://api.openai.com/v1" # WRONG
BASE_URL = "https://api.anthropic.com/v1" # WRONG
BASE_URL = "https://generativelanguage.googleapis.com/v1" # WRONG
CORRECT - HolySheep unified gateway
BASE_URL = "https://api.holysheep.ai/v1"
Always validate and sanitize API key
def get_authenticated_client(api_key: str) -> HolySheepGateway:
# Strip whitespace and validate format
clean_key = api_key.strip()
if not clean_key:
raise ValueError("API key cannot be empty")
if len(clean_key) < 20:
raise ValueError("API key appears to be invalid (too short)")
# Verify key format (HolySheep keys start with "hs_")
if not clean_key.startswith("hs_"):
raise ValueError(
"Invalid API key format. "
"Ensure you are using a HolySheep API key starting with 'hs_'"
)
return HolySheepGateway(api_key=clean_key)
Usage
try:
client = get_authenticated_client("YOUR_HOLYSHEEP_API_KEY")
except ValueError as e:
print(f"Configuration error: {e}")Error 2: Context Length Exceeded — "400 Bad Request"
Symptom: API returns 400 error with message about context length, token limits, or max_tokens exceeded.
Common Causes:
- Input prompt exceeds model's maximum context window
- Cumulative conversation history pushes total tokens over limit
- max_tokens parameter set higher than remaining context capacity
Solution:
# Calculate safe max_tokens with context budget management
def calculate_safe_params(
prompt: str,
model: str,
history_messages: List[Dict] = [],
safety_margin: int = 500 # Tokens to reserve for response
) -> Dict:
"""
Calculate safe max_tokens considering input context.
"""
model_limits = {
"deepseek-v3.2": 64000,
"gemini-2.5-flash": 128000,
"gpt-4.1": 128000,
"claude-sonnet-4.5": 200000
}
max_context = model_limits.get(model, 64000)
# Estimate input tokens (rough: 1 token ≈ 4 characters)
input_estimate = sum(
len(msg.get("content", "")) // 4
for msg in history_messages
) + (len(prompt) // 4)
# Calculate available for response
available = max_context - input_estimate - safety_margin
if available <= 0:
# Need to truncate or use longer-context model
raise ValueError(
f"Input context ({input_estimate} tokens) exceeds "
f"model capacity ({max_context} tokens). "
f"Consider using 'claude-sonnet-4.5' with 200K context window."
)
return {
"max_tokens": min(available, 32000), # Cap at reasonable response size
"input_tokens_estimate": input_estimate,
"context_utilization": input_estimate / max_context
}
Usage with automatic fallback for large contexts
async def smart_request_with_context_handling(
gateway: HolySheepGateway,
prompt: str,
history: List[Dict] = []
) -> Dict:
models_priority = [
"gemini-2.5-flash", # Try fastest first
"gpt-4.1", # Then standard
"claude-sonnet-4.5" # Finally, longest context
]
last_error = None
for model in models_priority:
try:
params = calculate_safe_params(prompt, model, history)
response = await gateway.chat_completion(
model=model,
messages=[
{"role": "system", "content": "You are a helpful assistant."},
*history,
{"role": "user", "content": prompt}
],
max_tokens=params["max_tokens"]
)
return {
"success": True,
"model": model,
"response": response,
"params_used": params
}
except ValueError as e:
# Context too long for this model, try next
last_error = e
continue
except httpx.HTTPStatusError as e:
if e.response.status_code == 400:
last_error = e
continue # Try next model
raise # Re-raise non-context errors
# All models failed
raise RuntimeError(
f"Context length exceeds all available models. "
f"Last error: {last_error}"
)Error 3: Rate Limiting — "429 Too Many Requests"
Symptom: Requests fail with 429 status, "Rate limit exceeded", or "Too many requests" messages. Occurs intermittently even with moderate request volumes.
Common Causes:
- Exceeding requests-per-minute (RPM) limits for the plan tier
- Burst traffic exceeding per-second rate limits
- Insufficient rate limit allocation for the API key
Solution:
import asyncio
from collections import deque
from datetime import datetime, timedelta
class RateLimitHandler:
"""
Sophisticated rate limiting with exponential backoff and queuing.
"""
def __init__(self, rpm_limit: int = 60, rps_burst: int = 10):
self.rpm_limit = rpm_limit
self.rps_burst = rps_burst
self.request_timestamps: deque = deque()
self.failed_requests: deque = deque()
self.queue: asyncio.Queue = asyncio.Queue()
self.semaphore = asyncio.Semaphore(rps_burst)
def _clean_old_timestamps(self):
"""Remove timestamps older than 1 minute."""
cutoff = datetime.utcnow() - timedelta(minutes=1)
while self.request_timestamps and self.request_timestamps[0] < cutoff:
self.request_timestamps.popleft()
def _calculate_backoff(self) -> float:
"""Calculate exponential backoff based on recent failures."""
self._clean_old_timestamps()
recent_failures = [
ts for ts in self.failed_requests
if datetime.utcnow() - ts < timedelta(minutes=5)
]
if not recent_failures:
return