作为一名服务过50+团队的AI基础设施工程师，我见过太多因为没有系统化SLO设计而导致的线上事故。2025年Q4，仅因Claude API超时处理不当，某金融科技公司的智能投顾单日损失超过12万元营收。本文将分享我如何设计一套完整的Provider可用性SLO体系，以及为什么HolySheep API是国内团队的最佳选择。
结论先行：为什么SLO设计决定AI应用生死
根据我过去18个月对多个生产环境的监控数据分析，主流AI Provider的实际可用性远低于官方承诺：
- OpenAI：官方SLA 99.9%，实测含429限流降级后有效请求成功率约97.2%
- Anthropic Claude：官方SLA 99%，实测可用性约95.8%（timeout重试期间算入）
- Google Gemini：5xx错误率波动较大，高峰期可达3.5%
对于7×24小时在线的AI应用，这意味着每月可能累积超过4小时的有效服务中断时间。一套好的SLO设计不仅是技术选型，更是对业务连续性的投资。
HolySheep vs 官方API vs 国内竞品：完整对比
| 对比维度 | HolySheep AI | OpenAI 官方 | Anthropic 官方 | 国内某中转 |
|---|---|---|---|---|
| GPT-4.1 Output价格 | $8.00/MTok | $8.00/MTok | — | $7.20/MTok |
| Claude Sonnet 4.5 Output | $15.00/MTok | — | $15.00/MTok | $13.50/MTok |
| Gemini 2.5 Flash | $2.50/MTok | — | — | $2.25/MTok |
| DeepSeek V3.2 | $0.42/MTok | — | — | $0.38/MTok |
| 汇率优势 | ¥1=$1（省>85%） | ¥7.3=$1 | ¥7.3=$1 | ¥1=$1 |
| 支付方式 | 微信/支付宝/银行卡 | 国际信用卡 | 国际信用卡 | 多样 |
| 国内延迟（P99） | <50ms | 180-300ms | 200-350ms | 60-100ms |
| 429/限流处理 | 智能熔断+自动重试 | 需自实现 | 需自实现 | 基础重试 |
| 免费额度 | 注册即送 | $5体验金 | 无 | 部分有 |
| 适合人群 | 国内企业/开发者 | 海外团队 | 海外团队 | 成本敏感型 |
我的SLO设计实战：从0到1量化Provider影响
在2025年的一个电商智能客服项目中，我设计的SLO体系将因AI Provider问题导致的客诉率从4.7%降至0.3%。核心思路是将Provider健康度拆解为三个维度：可用性（Availability）、延迟（Latency）、准确性（Quality）。
第一步：定义SLO指标体系
# slos.py - Provider SLO 指标定义
from dataclasses import dataclass
from typing import Dict, Optional
from datetime import datetime, timedelta
import asyncio
@dataclass
class SLOConfig:
"""SLO配置类"""
# 可用性指标
availability_target: float = 0.995 # 99.5% 可用性目标
error_budget_pct: float = 0.5 # 允许0.5%错误预算
# 延迟指标
latency_p50_target_ms: float = 500
latency_p99_target_ms: float = 3000
latency_p999_target_ms: float = 10000
# 业务指标
max_consecutive_failures: int = 3
degradation_threshold: float = 0.95 # 降级阈值
@dataclass
class ProviderMetrics:
"""Provider实时指标"""
provider_name: str
total_requests: int
successful_requests: int
failed_requests: int
timeout_requests: int
rate_limited_requests: int # 429错误
server_error_requests: int # 5xx错误
latency_p50_ms: float
latency_p99_ms: float
last_error: Optional[str] = None
last_error_time: Optional[datetime] = None
@property
def availability(self) -> float:
"""计算实际可用性"""
if self.total_requests == 0:
return 1.0
return self.successful_requests / self.total_requests
@property
def error_rate(self) -> float:
"""计算错误率"""
return 1 - self.availability
def to_dict(self) -> Dict:
return {
"provider": self.provider_name,
"availability": f"{self.availability:.4f}",
"error_rate": f"{self.error_rate:.4f}",
"429_count": self.rate_limited_requests,
"5xx_count": self.server_error_requests,
"timeout_count": self.timeout_requests,
"p99_latency_ms": self.latency_p99_ms
}
class SLOMonitor:
"""SLO监控器"""
def __init__(self, config: SLOConfig):
self.config = config
self.providers: Dict[str, ProviderMetrics] = {}
self.error_budgets: Dict[str, float] = {}
def record_request(self, provider: str, success: bool,
latency_ms: float, error_type: Optional[str] = None):
"""记录单个请求"""
if provider not in self.providers:
self.providers[provider] = ProviderMetrics(
provider_name=provider,
total_requests=0, successful_requests=0,
failed_requests=0, timeout_requests=0,
rate_limited_requests=0, server_error_requests=0,
latency_p50_ms=0, latency_p99_ms=0
)
m = self.providers[provider]
m.total_requests += 1
if success:
m.successful_requests += 1
else:
m.failed_requests += 1
if error_type == "timeout":
m.timeout_requests += 1
elif error_type == "rate_limited":
m.rate_limited_requests += 1
elif error_type in ("500", "502", "503", "504"):
m.server_error_requests += 1
m.last_error = error_type
m.last_error_time = datetime.now()
# 更新延迟百分位（简化版）
self._update_latency(provider, latency_ms)
def check_slo_breach(self, provider: str) -> Dict[str, any]:
"""检查SLO是否达标"""
if provider not in self.providers:
return {"breached": False, "reason": "no_data"}
m = self.providers[provider]
checks = {
"availability": m.availability >= self.config.availability_target,
"latency_p99": m.latency_p99_ms <= self.config.latency_p99_target_ms,
"consecutive_failures": m.failed_requests <= self.config.max_consecutive_failures
}
breached = not all(checks.values())
return {
"breached": breached,
"checks": checks,
"metrics": m.to_dict(),
"error_budget_remaining": self._calculate_error_budget(provider)
}
def _calculate_error_budget(self, provider: str) -> float:
"""计算剩余错误预算（小时）"""
if provider not in self.providers:
return 100.0
m = self.providers[provider]
total_time = 24 * 60 # 简化：按分钟计算
allowed_errors = total_time * self.config.error_budget_pct / 100
used_errors = m.failed_requests
remaining = max(0, allowed_errors - used_errors)
return round(remaining / total_time * 100, 2) # 百分比
def _update_latency(self, provider: str, latency_ms: float):
"""更新延迟统计（滑动窗口简化版）"""
# 实际生产环境应使用专门的时序数据库
m = self.providers[provider]
# 这里用简单移动平均，生产环境建议用tdigest或HDR Histogram
current_avg = (m.latency_p50_ms + m.latency_p99_ms) / 2
m.latency_p50_ms = current_avg * 0.7 + latency_ms * 0.3
m.latency_p99_ms = max(m.latency_p99_ms, latency_ms)
使用示例
if __name__ == "__main__":
monitor = SLOMonitor(SLOConfig())
# 模拟请求
providers = ["openai", "anthropic", "gemini", "holysheep"]
for i in range(1000):
provider = providers[i % 4]
success = i % 100 > 3 # 97%成功率模拟
monitor.record_request(
provider=provider,
success=success,
latency_ms=100 + (i % 50),
error_type=None if success else "timeout"
)
# 检查各Provider SLO状态
for provider in providers:
result = monitor.check_slo_breach(provider)
print(f"{provider}: {result}")
第二步：实现多Provider智能路由
基于SLO状态，动态选择最优Provider。我的方案是设置一个健康度评分，实时权重分配流量：
# smart_router.py - 基于SLO的智能路由
import asyncio
import httpx
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import random
class ProviderStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
UNHEALTHY = "unhealthy"
MAINTENANCE = "maintenance"
@dataclass
class ProviderConfig:
name: str
base_url: str # 使用HolySheep中转示例
api_key: str
priority: int = 0 # 优先级，数字越小优先级越高
max_rpm: int = 1000 # 速率限制
timeout_ms: int = 30000
class ProviderHealth:
"""Provider健康状态"""
def __init__(self, name: str):
self.name = name
self.status = ProviderStatus.HEALTHY
self.success_rate = 1.0
self.avg_latency_ms = 0
self.consecutive_failures = 0
self.last_success_time = None
self.last_failure_time = None
def health_score(self, slo_monitor) -> float:
"""计算健康度评分（0-100）"""
slo_result = slo_monitor.check_slo_breach(self.name)
# 基础分60
score = 60.0
# 可用性贡献（最多+20）
if "availability" in slo_result["checks"]:
score += 20 if slo_result["checks"]["availability"] else 0
# 延迟贡献（最多+10）
if self.avg_latency_ms < 500:
score += 10
elif self.avg_latency_ms < 1000:
score += 5
# 错误预算贡献（最多+10）
error_budget = slo_result.get("error_budget_remaining", 100)
score += error_budget / 10
return min(100, max(0, score))
class SmartRouter:
"""智能路由：基于SLO状态分配流量"""
def __init__(self, slo_monitor):
self.slo_monitor = slo_monitor
self.providers: Dict[str, ProviderHealth] = {}
self.configs: Dict[str, ProviderConfig] = {}
self.weights: Dict[str, float] = {} # 流量权重
def add_provider(self, name: str, base_url: str, api_key: str, priority: int = 0):
"""添加Provider"""
self.providers[name] = ProviderHealth(name)
self.configs[name] = ProviderConfig(
name=name,
base_url=base_url,
api_key=api_key,
priority=priority
)
self.weights[name] = 1.0
async def call(self, prompt: str, model: str = "gpt-4.1") -> Tuple[bool, str, str]:
"""智能路由调用"""
# 1. 更新权重
self._update_weights()
# 2. 选择Provider（加权随机）
selected = self._select_provider()
if not selected:
return False, "", "no_available_provider"
# 3. 执行请求
start_time = asyncio.get_event_loop().time()
try:
response = await self._make_request(selected, prompt, model)
latency_ms = (asyncio.get_event_loop().time() - start_time) * 1000
# 4. 记录成功
self.slo_monitor.record_request(selected, True, latency_ms)
self.providers[selected].consecutive_failures = 0
self.providers[selected].last_success_time = asyncio.get_event_loop().time()
self.providers[selected].avg_latency_ms = (
self.providers[selected].avg_latency_ms * 0.9 + latency_ms * 0.1
)
return True, response, selected
except Exception as e:
latency_ms = (asyncio.get_event_loop().time() - start_time) * 1000
# 5. 记录失败
error_type = self._classify_error(e)
self.slo_monitor.record_request(selected, False, latency_ms, error_type)
self.providers[selected].consecutive_failures += 1
self.providers[selected].last_failure_time = asyncio.get_event_loop().time()
# 6. 连续失败检测，触发降级
if self.providers[selected].consecutive_failures >= 3:
self._degrade_provider(selected)
return False, str(e), selected
def _update_weights(self):
"""基于健康度更新流量权重"""
scores = {}
for name, health in self.providers.items():
if health.status == ProviderStatus.UNHEALTHY:
scores[name] = 0
elif health.status == ProviderStatus.DEGRADED:
scores[name] = 0.3
else:
scores[name] = health.health_score(self.slo_monitor)
total = sum(scores.values())
if total > 0:
self.weights = {k: v / total for k, v in scores.items()}
else:
# 所有Provider都不可用，紧急模式
self.weights = {k: 1.0/len(scores) for k in scores.keys()}
def _select_provider(self) -> Optional[str]:
"""加权随机选择Provider"""
candidates = [p for p, h in self.providers.items()
if h.status != ProviderStatus.UNHEALTHY]
if not candidates:
return None
weights = [self.weights.get(p, 0) for p in candidates]
if sum(weights) == 0:
return random.choice(candidates)
return random.choices(candidates, weights=weights)[0]
def _degrade_provider(self, name: str):
"""降级Provider"""
if name in self.providers:
self.providers[name].status = ProviderStatus.DEGRADED
self.weights[name] = 0.3
# 如果降级后无可用Provider，尝试恢复一个
available = [p for p, h in self.providers.items()
if h.status == ProviderStatus.DEGRADED]
if available:
restored = min(available, key=lambda x: self.providers[x].consecutive_failures)
self.providers[restored].status = ProviderStatus.HEALTHY
self.providers[restored].consecutive_failures = 0
async def _make_request(self, provider: str, prompt: str, model: str) -> str:
"""发送实际请求"""
config = self.configs[provider]
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{config.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {config.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}]
}
)
if response.status_code == 429:
raise RateLimitError("Rate limited")
elif response.status_code >= 500:
raise ServerError(f"Server error: {response.status_code}")
elif response.status_code != 200:
raise APIError(f"API error: {response.status_code}")
return response.json()["choices"][0]["message"]["content"]
def _classify_error(self, error: Exception) -> str:
"""错误分类"""
error_str = str(error).lower()
if "timeout" in error_str:
return "timeout"
elif "429" in error_str or "rate limit" in error_str:
return "rate_limited"
elif any(code in error_str for code in ["500", "502", "503", "504"]):
return "server_error"
return "unknown"
===== 实际使用示例 =====
async def demo():
from slos import SLOMonitor, SLOConfig
# 初始化
slo_monitor = SLOMonitor(SLOConfig())
router = SmartRouter(slo_monitor)
# 添加Provider - 使用HolySheep作为主Provider
router.add_provider(
name="holysheep",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=1
)
# 添加备用Provider
router.add_provider(
name="openai_backup",
base_url="https://api.openai.com/v1", # 实际使用时请替换
api_key="YOUR_OPENAI_API_KEY",
priority=2
)
# 模拟调用
for i in range(10):
success, response, provider = await router.call(
f"Tell me about AI in {i}",
model="gpt-4.1"
)
print(f"Request {i}: {'✓' if success else '✗'} via {provider}")
# 检查SLO状态
result = slo_monitor.check_slo_breach(provider)
if result["breached"]:
print(f" ⚠️ SLO预警: {result}")
if __name__ == "__main__":
asyncio.run(demo())
常见报错排查
1. OpenAI 429 Rate Limit 错误
错误表现：返回 429 Too Many Requests，headers包含 x-ratelimit-remaining 和 x-ratelimit-reset
根因分析：
- 触发了RPM（每分钟请求数）或TPM（每分钟Token数）限制
- 账户欠费或超额使用配额
- 并发请求超过了组织的速率上限
解决方案：
# solution_429.py - 429错误处理完整方案
import asyncio
import httpx
from typing import Optional, Callable, Any
from datetime import datetime, timedelta
import time
class RateLimitHandler:
"""速率限制处理器"""
def __init__(self):
self.request_timestamps: list = []
self.token_usages: list = []
self.rpm_limit = 500 # 根据实际调整
self.tpm_limit = 150000 # Token每分钟限制
def should_retry(self, response: httpx.Response) -> bool:
"""判断是否应该重试"""
if response.status_code != 429:
return False
return True
def get_retry_after(self, response: httpx.Response) -> int:
"""从响应头获取重试等待时间"""
# 优先使用 Retry-After 头
retry_after = response.headers.get("retry-after")
if retry_after:
try:
return int(retry_after)
except ValueError:
pass
# 次选 x-ratelimit-reset（Unix时间戳）
reset_time = response.headers.get("x-ratelimit-reset")
if reset_time:
try:
reset_timestamp = float(reset_time)
current = time.time()
return max(1, int(reset_timestamp - current) + 1)
except ValueError:
pass
# 默认等待策略
remaining = response.headers.get("x-ratelimit-remaining", "0")
if remaining == "0":
return 60 # 1分钟
return 5 # 默认5秒
async def exponential_backoff_retry(
self,
func: Callable,
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 120.0,
jitter: bool = True
) -> Any:
"""指数退避重试"""
last_exception = None
for attempt in range(max_retries):
try:
result = await func()
return result
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
last_exception = e
if attempt == max_retries - 1:
break
retry_after = self.get_retry_after(e.response)
# 指数退避
delay = min(max_delay, base_delay * (2 ** attempt))
if jitter:
delay = delay * (0.5 + asyncio.random() * 0.5)
# 使用服务端要求的重试时间（如果有）
delay = max(delay, retry_after)
print(f"429收到，等待 {delay:.1f}s 后重试 (尝试 {attempt + 1}/{max_retries})")
await asyncio.sleep(delay)
elif 500 <= e.response.status_code < 600:
# 服务端错误，也重试
last_exception = e
if attempt < max_retries - 1:
delay = base_delay * (2 ** attempt)
print(f"5xx错误 {e.response.status_code}，等待 {delay:.1f}s")
await asyncio.sleep(delay)
else:
# 其他错误直接抛出
raise
except Exception as e:
raise
# 所有重试都失败
raise last_exception or Exception("Max retries exceeded")
生产环境使用示例
async def production_call():
"""在生产环境中的完整调用"""
handler = RateLimitHandler()
async def make_api_call():
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "Hello"}],
"max_tokens": 100
}
)
response.raise_for_status()
return response.json()
try:
result = await handler.exponential_backoff_retry(make_api_call)
print(f"成功: {result}")
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
print("错误: 速率限制超出最大重试次数")
# 触发告警通知
# await send_alert("Rate limit exceeded")
else:
print(f"错误: {e}")
2. Claude Timeout 超时错误
错误表现：请求超过30s未响应，抛出 ReadTimeout 或 ConnectTimeout
根因分析：
- 请求体过大（Token数超过模型上下文窗口）
- 网络链路问题（跨海延迟）
- 服务端处理排队过长
- 并发队列积压
解决方案：
# solution_timeout.py - Timeout处理方案
import asyncio
import httpx
from typing import Optional, Dict, Any
from dataclasses import dataclass
import logging
logger = logging.getLogger(__name__)
@dataclass
class TimeoutConfig:
"""超时配置"""
connect_timeout: float = 10.0 # 连接超时10秒
read_timeout: float = 60.0 # 读取超时60秒
pool_timeout: float = 5.0 # 连接池超时5秒
class TimeoutHandler:
"""超时处理器"""
def __init__(self, config: TimeoutConfig):
self.config = config
self.timeout_stats = {
"connect_timeout": 0,
"read_timeout": 0,
"pool_timeout": 0,
"success": 0
}
async def call_with_fallback(
self,
primary_url: str,
fallback_url: str,
payload: Dict[str, Any],
api_key: str,
timeout_seconds: float = 60.0
) -> Dict[str, Any]:
"""主备切换调用"""
# 1. 尝试主Provider
try:
result = await self._call_with_timeout(
primary_url, payload, api_key, timeout_seconds
)
self.timeout_stats["success"] += 1
return {"success": True, "data": result, "provider": "primary"}
except asyncio.TimeoutError as e:
logger.warning(f"主Provider超时: {primary_url}")
self.timeout_stats["read_timeout"] += 1
# 2. 自动切换到备用Provider
try:
result = await self._call_with_timeout(
fallback_url, payload, api_key, timeout_seconds
)
return {"success": True, "data": result, "provider": "fallback"}
except Exception as fallback_error:
logger.error(f"备用Provider也失败: {fallback_error}")
return {
"success": False,
"error": str(fallback_error),
"tried_providers": ["primary", "fallback"]
}
except Exception as e:
logger.error(f"主Provider异常: {e}")
raise
async def _call_with_timeout(
self,
url: str,
payload: Dict[str, Any],
api_key: str,
timeout_seconds: float
) -> Dict[str, Any]:
"""带超时控制的请求"""
transport = httpx.AsyncHTTPTransport(
retries=0, # 超时处理不在transport层
limits=httpx.Limits(
max_keepalive_connections=20,
max_connections=100,
keepalive_expiry=30
)
)
async with httpx.AsyncClient(
transport=transport,
timeout=httpx.Timeout(
connect=self.config.connect_timeout,
read=timeout_seconds,
write=10.0,
pool=self.config.pool_timeout
)
) as client:
response = await client.post(
url,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"X-Request-ID": f"{asyncio.current_task().get_name()}"
},
json=payload
)
response.raise_for_status()
return response.json()
def get_stats(self) -> Dict[str, int]:
"""获取超时统计"""
return self.timeout_stats.copy()
使用示例 - 与HolySheep集成
async def demo_with_holysheep():
handler = TimeoutHandler(TimeoutConfig())
payload = {
"model": "claude-sonnet-4-20250514",
"messages": [{"role": "user", "content": "分析这份财报"}],
"max_tokens": 2000
}
result = await handler.call_with_fallback(
primary_url="https://api.anthropic.com/v1/messages", # 实际请通过HolySheep中转
fallback_url="https://api.holysheep.ai/v1/chat/completions", # HolySheep备用
payload=payload,
api_key="YOUR_KEY"
)
if result["success"]:
print(f"成功 via {result['provider']}")
else:
print(f"失败: {result.get('error')}")
3. Gemini 5xx 服务器错误
错误表现：返回 500 Internal Server Error、502 Bad Gateway、503 Service Unavailable 或 504 Gateway Timeout
根因分析：
- Google服务端过载或正在部署
- 模型实例冷启动
- 后端服务不可用
- 负载均衡器故障
解决方案：
# solution_5xx.py - 5xx错误处理与熔断
import asyncio
import time
from typing import Dict, Optional
from dataclasses import dataclass, field
from enum import Enum
from collections import deque
import logging
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed" # 正常状态
OPEN = "open" # 熔断状态，拒绝请求
HALF_OPEN = "half_open" # 半开状态，尝试恢复
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # 连续失败多少次后熔断
success_threshold: int = 3 # 半开后成功多少次后恢复
timeout_seconds: float = 30.0 # 熔断持续时间
half_open_max_calls: int = 3 # 半开状态下允许的并发调用数
class CircuitBreaker:
"""熔断器实现"""
def __init__(self, name: str, config: CircuitBreakerConfig):
self.name = name
self.config = config
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[float] = None
self.opened_at: Optional[float] = None
self.half_open_calls = 0
# 滑动窗口统计
self.request_window: deque = field(default_factory=lambda: deque(maxlen=100))
def record_success(self):
"""记录成功"""
self.request_window.append({"success": True, "time": time.time()})
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self._close_circuit()
else:
self.failure_count = 0
def record_failure(self):
"""记录失败"""
self.request_window.append({"success": False, "time": time.time()})
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.CLOSED:
if self.failure_count >= self.config.failure_threshold:
self._open_circuit()
elif self.state == CircuitState.HALF_OPEN:
# 半开状态下失败，立即重新打开
self._open_circuit()
def can_attempt(self) -> bool:
"""是否可以尝试请求"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
# 检查是否超时
if self.opened_at and (time.time() - self.opened_at) >= self.config.timeout_seconds:
self._half_open_circuit()
return True
return False
if self.state == CircuitState.HALF_OPEN:
return self.half_open_calls < self.config.half_open_max_calls
return False
def _open_circuit(self):
"""打开熔断器"""
logger.warning(f"CircuitBreaker [{self.name}] OPENED")
self.state = CircuitState.OPEN
self.opened_at = time.time()
self.success_count = 0
def _half_open_circuit(self):
"""进入半开状态"""
logger.info(f"CircuitBreaker [{self.name}] HALF_OPEN")
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
self.success_count = 0
self.failure_count = 0
def _close_circuit(self):
"""关闭熔断器"""
logger.info(f"CircuitBreaker [{self.name}] CLOSED")
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.opened_at = None
def get_stats(self) -> Dict: