Trong quá trình xây dựng hệ thống AI Agent cho startup của mình, tôi đã trải qua không ít lần "đau đầu" với các vấn đề về SLA, rate limiting, retry logic và multi-model fallback. Bài viết này sẽ chia sẻ checklist SRE mà tôi đã đúc kết được, kèm theo hướng dẫn triển khai thực tế với HolySheep AI — nền tảng mà team đã chọn để giải quyết bài toán này một cách hiệu quả.
Mục lục
- Vấn đề thực tế khi vận hành AI Agent SaaS
- SLA Monitoring với HolySheep
- Rate Limiting — Triển khai tầng bảo vệ đầu tiên
- Retry Logic thông minh
- Multi-Model Fallback và Circuit Breaker
- Giá và ROI
- Phù hợp / không phù hợp với ai
- Vì sao chọn HolySheep
- Lỗi thường gặp và cách khắc phục
- Đăng ký và bắt đầu
Vấn đề thực tế khi vận hành AI Agent SaaS
Khi xây dựng AI Agent SaaS, đội ngũ kỹ thuật thường gặp những thách thức cốt lõi sau:
- Latency không ổn định: API của OpenAI/Anthropic có độ trễ trung bình 800-2000ms, ảnh hưởng trực tiếp đến trải nghiệm người dùng.
- Cost explosion: Không kiểm soát được chi phí khi user gọi API liên tục, đặc biệt với các model đắt đỏ như GPT-4o.
- Single point of failure: Phụ thuộc vào một provider duy nhất, khi provider gặp sự cố, toàn bộ hệ thống ngừng hoạt động.
- Retry storm: Khi hệ thống có lỗi, các retry request từ client có thể gây ra "thác lũ" (cascade failure).
Từ kinh nghiệm vận hành hệ thống phục vụ hơn 50,000 request/ngày, tôi đã xây dựng bộ công cụ SRE với HolySheep AI — nền tảng cung cấp latency trung bình dưới 50ms, hỗ trợ WeChat/Alipay thanh toán, và tỷ giá ¥1 = $1 giúp tiết kiệm chi phí đến 85%.
SLA Monitoring với HolySheep
HolySheep cung cấp dashboard theo dõi SLA với các chỉ số quan trọng. Tuy nhiên, để chủ động hơn, bạn nên triển khai monitoring riêng.
Metrics cần theo dõi
- Success Rate: Target > 99.5%
- P50/P95/P99 Latency: Target P99 < 200ms
- Error Rate by Model: Phát hiện model có vấn đề
- Token Usage: Theo dõi chi phí theo thời gian thực
#!/usr/bin/env python3
"""
SLA Monitoring Client cho HolySheep AI
Metrics: Success Rate, Latency, Error Classification
"""
import time
import asyncio
import aiohttp
from dataclasses import dataclass
from typing import Optional, List
from collections import defaultdict
import statistics
@dataclass
class RequestMetrics:
timestamp: float
latency_ms: float
success: bool
model: str
error_type: Optional[str] = None
tokens_used: Optional[int] = None
class HolySheepSLAMonitor:
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.metrics: List[RequestMetrics] = []
self.error_counts = defaultdict(int)
self.total_requests = 0
async def call_with_metrics(
self,
model: str,
prompt: str,
max_tokens: int = 1000
) -> dict:
"""Gọi API với metrics tracking"""
start_time = time.perf_counter()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
latency_ms = (time.perf_counter() - start_time) * 1000
if response.status == 200:
data = await response.json()
tokens = data.get("usage", {}).get("total_tokens", 0)
self._record_success(model, latency_ms, tokens)
return {"success": True, "data": data, "latency_ms": latency_ms}
else:
error_body = await response.text()
self._record_error(model, latency_ms, response.status, error_body)
return {"success": False, "error": error_body, "latency_ms": latency_ms}
except asyncio.TimeoutError:
latency_ms = (time.perf_counter() - start_time) * 1000
self._record_error(model, latency_ms, "TIMEOUT", "Request timeout")
return {"success": False, "error": "timeout", "latency_ms": latency_ms}
except Exception as e:
latency_ms = (time.perf_counter() - start_time) * 1000
self._record_error(model, latency_ms, "EXCEPTION", str(e))
return {"success": False, "error": str(e), "latency_ms": latency_ms}
def _record_success(self, model: str, latency_ms: float, tokens: int):
self.metrics.append(RequestMetrics(
timestamp=time.time(),
latency_ms=latency_ms,
success=True,
model=model,
tokens_used=tokens
))
self.total_requests += 1
def _record_error(self, model: str, latency_ms: float, error_code, error_msg: str):
self.metrics.append(RequestMetrics(
timestamp=time.time(),
latency_ms=latency_ms,
success=False,
model=model,
error_type=f"{error_code}: {error_msg}"
))
self.error_counts[error_code] += 1
self.total_requests += 1
def get_sla_report(self, window_seconds: int = 300) -> dict:
"""Generate SLA report for monitoring window"""
cutoff_time = time.time() - window_seconds
recent_metrics = [m for m in self.metrics if m.timestamp >= cutoff_time]
if not recent_metrics:
return {"error": "No data in window"}
total = len(recent_metrics)
successful = sum(1 for m in recent_metrics if m.success)
success_rate = (successful / total) * 100
latencies = [m.latency_ms for m in recent_metrics]
latencies.sort()
return {
"window_seconds": window_seconds,
"total_requests": total,
"successful_requests": successful,
"success_rate_pct": round(success_rate, 3),
"sla_compliant": success_rate >= 99.5,
"latency": {
"p50": round(latencies[int(len(latencies) * 0.50)], 2),
"p95": round(latencies[int(len(latencies) * 0.95)], 2),
"p99": round(latencies[int(len(latencies) * 0.99)], 2),
"avg": round(statistics.mean(latencies), 2),
"max": round(max(latencies), 2)
},
"error_breakdown": dict(self.error_counts),
"target_met": success_rate >= 99.5
}
Sử dụng
monitor = HolySheepSLAMonitor(api_key="YOUR_HOLYSHEEP_API_KEY")
async def main():
# Test với các model khác nhau
models = ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"]
for model in models:
result = await monitor.call_with_metrics(
model=model,
prompt="Hello, tell me a short joke"
)
print(f"{model}: {result}")
# In SLA report
report = monitor.get_sla_report(window_seconds=300)
print(f"\n=== SLA Report ===")
print(f"Success Rate: {report['success_rate_pct']}%")
print(f"SLA Compliant: {report['sla_compliant']}")
print(f"P99 Latency: {report['latency']['p99']}ms")
if __name__ == "__main__":
asyncio.run(main())Prometheus Integration
# prometheus.yml
scrape_configs:
- job_name: 'holysheep-sla-monitor'
static_configs:
- targets: ['your-monitor-host:9090']
metrics_path: '/metrics'
Exporters cần expose các metrics:
- holysheep_request_total{status="success|error",model="xxx"}
- holysheep_request_latency_seconds{quantile="0.99",model="xxx"}
- holysheep_token_usage_total{model="xxx"}
- holysheep_cost_estimate_usd{model="xxx"}
Rate Limiting — Triển khai tầng bảo vệ đầu tiên
Rate limiting là lớp bảo vệ quan trọng nhất để ngăn chặn cost explosion và abuse. Tôi khuyến nghị triển khai multi-tier rate limiting.
Architecture Rate Limiting
#!/usr/bin/env python3
"""
Multi-Tier Rate Limiting với Token Bucket + Sliding Window
Tier 1: Per-User (100 req/min)
Tier 2: Per-API-Key (1000 req/min)
Tier 3: Per-Model (500 req/min)
"""
import time
import asyncio
import hashlib
from dataclasses import dataclass, field
from typing import Dict, Optional, Tuple
from collections import defaultdict
import threading
@dataclass
class RateLimitConfig:
requests_per_minute: int
burst_size: int
window_seconds: int = 60
class TokenBucket:
"""Token Bucket implementation với refill logic"""
def __init__(self, capacity: int, refill_rate: float):
self.capacity = capacity
self.refill_rate = refill_rate # tokens per second
self.tokens = float(capacity)
self.last_refill = time.time()
self._lock = threading.Lock()
def consume(self, tokens: int = 1) -> Tuple[bool, float]:
"""Try to consume tokens, returns (allowed, wait_time_ms)"""
with self._lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True, 0.0
# Calculate wait time
tokens_needed = tokens - self.tokens
wait_seconds = tokens_needed / self.refill_rate
return False, wait_seconds * 1000
def _refill(self):
"""Refill tokens based on elapsed time"""
now = time.time()
elapsed = now - self.last_refill
if elapsed > 0:
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
class SlidingWindowCounter:
"""Sliding Window Counter cho accurate rate limiting"""
def __init__(self, window_seconds: int = 60, max_requests: int = 100):
self.window_seconds = window_seconds
self.max_requests = max_requests
self.requests: Dict[str, list] = defaultdict(list)
self._lock = threading.Lock()
def is_allowed(self, key: str, request_count: int = 1) -> Tuple[bool, int, float]:
"""
Check if request is allowed
Returns: (allowed, current_count, retry_after_seconds)
"""
with self._lock:
now = time.time()
window_start = now - self.window_seconds
# Clean old requests
self.requests[key] = [
ts for ts in self.requests[key]
if ts > window_start
]
current_count = len(self.requests[key])
if current_count + request_count <= self.max_requests:
self.requests[key].extend([now] * request_count)
return True, current_count + request_count, 0.0
# Calculate retry-after
oldest_in_window = min(self.requests[key]) if self.requests[key] else now
retry_after = (oldest_in_window + self.window_seconds) - now
return False, current_count, max(0, retry_after)
class MultiTierRateLimiter:
"""
Multi-tier rate limiter
Check order: User -> API Key -> Model -> Global
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
# Tier configurations
self.tier_configs = {
"user": RateLimitConfig(requests_per_minute=100, burst_size=20),
"api_key": RateLimitConfig(requests_per_minute=1000, burst_size=100),
"model": RateLimitConfig(requests_per_minute=500, burst_size=50),
}
# Sliding window counters
self.sliding_windows = {
tier: SlidingWindowCounter(
window_seconds=cfg.window_seconds,
max_requests=cfg.requests_per_minute
)
for tier, cfg in self.tier_configs.items()
}
# Token buckets for burst handling
self.token_buckets = {
tier: TokenBucket(
capacity=cfg.burst_size,
refill_rate=cfg.requests_per_minute / 60.0
)
for tier, cfg in self.tier_configs.items()
}
def _get_user_id(self, request_context: dict) -> str:
"""Extract user identifier from request context"""
return hashlib.sha256(
f"{request_context.get('user_id', 'anonymous')}"
.encode()
).hexdigest()[:16]
def _get_model_id(self, model: str) -> str:
"""Get model-specific identifier"""
return f"model:{model}"
async def check_rate_limit(
self,
request_context: dict,
model: str,
tokens: int = 1
) -> Tuple[bool, Optional[float], dict]:
"""
Check all rate limit tiers
Returns: (allowed, retry_after_ms, breakdown)
"""
user_id = self._get_user_id(request_context)
api_key_id = self.api_key[:8] # Partial key for identification
model_id = self._get_model_id(model)
tiers_to_check = [
("user", user_id),
("api_key", api_key_id),
("model", model_id)
]
breakdown = {}
max_retry_after = 0.0
for tier, key in tiers_to_check:
# Check sliding window
allowed, count, retry_after = self.sliding_windows[tier].is_allowed(key)
# Check token bucket for burst
bucket_allowed, wait_ms = self.token_buckets[tier].consume()
total_wait = max(retry_after, wait_ms / 1000)
breakdown[tier] = {
"allowed": allowed and bucket_allowed,
"count": count,
"retry_after_s": total_wait
}
if not (allowed and bucket_allowed):
max_retry_after = max(max_retry_after, total_wait)
is_allowed = all(b["allowed"] for b in breakdown.values())
return is_allowed, max_retry_after if not is_allowed else None, breakdown
async def call_with_rate_limit(
self,
request_context: dict,
model: str,
prompt: str,
**kwargs
) -> dict:
"""Make API call với rate limit enforcement"""
allowed, retry_after, breakdown = await self.check_rate_limit(
request_context, model
)
if not allowed:
return {
"success": False,
"error": "rate_limit_exceeded",
"retry_after_ms": retry_after * 1000,
"breakdown": breakdown
}
# Proceed with API call through HolySheep
import aiohttp
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
**kwargs
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
return await response.json()
Sử dụng
limiter = MultiTierRateLimiter(api_key="YOUR_HOLYSHEEP_API_KEY")
async def protected_endpoint(user_id: str, model: str, prompt: str):
context = {"user_id": user_id}
result = await limiter.call_with_rate_limit(
request_context=context,
model=model,
prompt=prompt
)
if not result.get("success", True):
print(f"Rate limited! Retry after: {result.get('retry_after_ms')}ms")
return result
print(f"Success! Tokens: {result.get('usage', {}).get('total_tokens')}")
return result
Test
asyncio.run(protected_endpoint("user_123", "deepseek-v3.2", "Hello!"))Retry Logic thông minh
Retry logic cần được thiết kế cẩn thận để tránh cascade failure. Dưới đây là implementation với exponential backoff và jitter.
#!/usr/bin/env python3
"""
Smart Retry với Exponential Backoff, Jitter và Circuit Breaker
Hỗ trợ HolySheep AI với multi-model fallback
"""
import time
import asyncio
import random
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Optional, List, Any
from collections import defaultdict
import aiohttp
class RetryStrategy(Enum):
EXPONENTIAL = "exponential"
LINEAR = "linear"
FIBONACCI = "fibonacci"
@dataclass
class RetryConfig:
max_attempts: int = 3
base_delay_ms: int = 100
max_delay_ms: int = 5000
strategy: RetryStrategy = RetryStrategy.EXPONENTIAL
jitter: bool = True
jitter_factor: float = 0.3
retryable_status_codes: List[int] = field(
default_factory=lambda: [408, 429, 500, 502, 503, 504]
)
retryable_exceptions: tuple = (
aiohttp.ClientError,
asyncio.TimeoutError,
ConnectionError
)
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject requests
HALF_OPEN = "half_open" # Testing if recovery possible
@dataclass
class CircuitBreaker:
name: str
failure_threshold: int = 5
recovery_timeout_seconds: float = 30.0
half_open_max_calls: int = 3
state: CircuitState = field(default=CircuitState.CLOSED)
failure_count: int = field(default=0)
success_count: int = field(default=0)
last_failure_time: float = field(default=0)
half_open_calls: int = field(default=0)
async def call(self, func: Callable, *args, **kwargs) -> Any:
"""Execute function with circuit breaker protection"""
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time >= self.recovery_timeout_seconds:
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
return await func(*args, **kwargs)
else:
raise CircuitBreakerOpenError(f"Circuit {self.name} is OPEN")
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls >= self.half_open_max_calls:
raise CircuitBreakerOpenError(
f"Circuit {self.name} half-open limit reached"
)
self.half_open_calls += 1
try:
result = await func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.half_open_max_calls:
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
else:
self.failure_count = max(0, self.failure_count - 1)
def _on_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
self.success_count = 0
elif self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
def get_status(self) -> dict:
return {
"name": self.name,
"state": self.state.value,
"failure_count": self.failure_count,
"success_count": self.success_count
}
class CircuitBreakerOpenError(Exception):
pass
class SmartRetry:
"""Smart retry với exponential backoff, jitter và circuit breaker"""
def __init__(self, config: RetryConfig, circuit_breaker: Optional[CircuitBreaker] = None):
self.config = config
self.circuit_breaker = circuit_breaker
self.call_counts = defaultdict(int)
def _calculate_delay(self, attempt: int) -> float:
"""Calculate delay with exponential backoff and jitter"""
if self.config.strategy == RetryStrategy.EXPONENTIAL:
delay = self.config.base_delay_ms * (2 ** attempt)
elif self.config.strategy == RetryStrategy.LINEAR:
delay = self.config.base_delay_ms * (attempt + 1)
elif self.config.strategy == RetryStrategy.FIBONACCI:
a, b = 1, 1
for _ in range(attempt):
a, b = b, a + b
delay = self.config.base_delay_ms * a
else:
delay = self.config.base_delay_ms
# Cap at max delay
delay = min(delay, self.config.max_delay_ms)
# Add jitter
if self.config.jitter:
jitter_range = delay * self.config.jitter_factor
delay = delay + random.uniform(-jitter_range, jitter_range)
return delay / 1000 # Convert to seconds
def _is_retryable(self, error: Exception, response: Optional[dict] = None) -> bool:
"""Determine if error/reponse is retryable"""
if isinstance(error, self.config.retryable_exceptions):
return True
if response and isinstance(response, dict):
status = response.get("status") or response.get("status_code")
if status in self.config.retryable_status_codes:
return True
return False
async def execute(
self,
func: Callable,
*args,
fallback_func: Optional[Callable] = None,
**kwargs
) -> Any:
"""
Execute function với smart retry logic
"""
last_error = None
for attempt in range(self.config.max_attempts):
self.call_counts["attempt"] += 1
try:
if self.circuit_breaker:
result = await self.circuit_breaker.call(func, *args, **kwargs)
else:
result = await func(*args, **kwargs)
# Check if response indicates error
if isinstance(result, dict):
status = result.get("status") or result.get("status_code")
if status in self.config.retryable_status_codes:
raise aiohttp.ClientResponseError(
request_info=None,
history=None,
status=status,
message=result.get("error", "Unknown error")
)
return result
except Exception as e:
last_error = e
self.call_counts["error"] += 1
# Check if should retry
if not self._is_retryable(e) or attempt == self.config.max_attempts - 1:
break
# Calculate delay
delay = self._calculate_delay(attempt)
print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f}s...")
await asyncio.sleep(delay)
# Fallback if available
if fallback_func:
self.call_counts["fallback"] += 1
print("Primary failed, attempting fallback...")
return await fallback_func()
raise last_error
class HolySheepMultiModelRetry:
"""
Multi-model retry với automatic fallback
Priority: Primary Model -> Fallback 1 -> Fallback 2 -> Error
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
# Model priority chain (cheapest first for cost optimization)
self.model_chain = [
("deepseek-v3.2", RetryConfig(max_attempts=3)),
("gemini-2.5-flash", RetryConfig(max_attempts=2)),
("claude-sonnet-4.5", RetryConfig(max_attempts=2)),
("gpt-4.1", RetryConfig(max_attempts=1)),
]
# Circuit breakers per model
self.circuit_breakers = {
model: CircuitBreaker(
name=model,
failure_threshold=3,
recovery_timeout_seconds=60
)
for model, _ in self.model_chain
}
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async def _call_model(self, model: str, prompt: str, **kwargs) -> dict:
"""Call specific model"""
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
**kwargs
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
return await response.json()
async def call_with_fallback(
self,
prompt: str,
primary_model: Optional[str] = None,
**kwargs
) -> dict:
"""
Call với automatic model fallback
Tries models in chain until success
"""
errors = []
# Build effective model chain
if primary_model:
chain = [(m, cfg) for m, cfg in self.model_chain if m == primary_model]
chain += [(m, cfg) for m, cfg in self.model_chain if m != primary_model]
else:
chain = self.model_chain
for model, retry_config in chain:
breaker = self.circuit_breakers[model]
retry = SmartRetry(retry_config, breaker)
try:
result = await retry.execute(
self._call_model,
model=model,
prompt=prompt,
**kwargs
)
# Add metadata
result["_meta"] = {
"model_used": model,
"circuit_status": breaker.get_status()
}
return result
except Exception as e:
errors.append({"model": model, "error": str(e)})
print(f"Model {model} failed: {e}")
continue
# All models failed
return {
"error": "All models failed",
"errors": errors
}
def get_health_report(self) -> dict:
"""Get health status of all models"""
return {
model: breaker.get_status()
for model, breaker in self.circuit_breakers.items()
}
Sử dụng
client = HolySheepMultiModelRetry(api_key="YOUR_HOLYSHEEP_API_KEY")
async def main():
# Single call với automatic fallback
result = await client.call_with_fallback(
prompt="Explain quantum computing in simple terms",
temperature=0.7
)
if "error" in result:
print(f"All models failed: {result['errors']}")
else:
print(f"Success with {result['_meta']['model_used']}")
print(f"Response: {result['choices'][0]['message']['content']}")
# Check model health
print("\nModel Health Report:")
for model, status in client.get_health_report().items():
print(f" {model}: {status['state']}")
asyncio.run(main())Multi-Model Fallback và Circuit Breaker
Đây là phần quan trọng nhất trong SRE checklist. Tôi đã triển khai hệ thống circuit breaker cho từng model và automatic fallback chain.
Model Fallback Chain Strategy
#!/usr/bin/env python3
"""
Production Multi-Model Fallback System
Priority: deepseek-v3.2 (cheapest) -> gemini-2.5-flash -> claude-sonnet-4.5 -> gpt-4.1
HolySheep Pricing 2026 (per 1M tokens):
- deepseek-v3.2: $0.42 (INPUT) / $0.42 (OUTPUT)
- gemini-2.5-flash: $2.50 (INPUT) / $10.00 (OUTPUT)
- claude-sonnet-4.5: $15.00 (INPUT) / $15.00 (OUTPUT)
- gpt-4.1: $8.00 (INPUT) / $8.00 (OUTPUT)
"""
import asyncio
import aiohttp
import time
from dataclasses import dataclass
from typing import List, Optional, Tuple
from enum import Enum
class ModelTier(Enum):
TIER_1_CHEAP = "deepseek-v3.2" # $0.42/M
TIER_2_BALANCED = "gemini-2.5-flash" # $2.50/M
TIER_3_PREMIUM = "claude-sonnet-4.5" # $15.00/M
TIER_4_MAX = "gpt-4.1" # $8.00/M
@dataclass
class ModelConfig:
name: str
max_tokens: int
timeout_seconds: float
failure_threshold: int
recovery_timeout_seconds: float
input_cost_per_m: float
output_cost_per_m: float
MODEL_CONFIGS = {
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
max_tokens=8192,
timeout_seconds=10.0,
failure_threshold=5,
recovery_timeout_seconds=60.0,
input_cost_per_m=0.42,
output_cost_per_m=0.42
),
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
max_tokens=8192,
timeout_seconds=15.0,
failure_threshold=3,
recovery_timeout_seconds=120.0,
input_cost