Trong quá trình vận hành hệ thống AI production tại HolySheep AI, đội ngũ kỹ sư của tôi đã xử lý hơn 2.3 triệu request mỗi ngày. Điều tôi nhận ra sau 18 tháng vận hành: không có alert tự động cho exception = disaster waiting to happen. Bài viết này chia sẻ toàn bộ kiến trúc, code mẫu và bài học xương máu từ thực chiến.
Tại Sao Alert API Exception Lại Quan Trọng?
Khi tích hợp HolySheep AI API vào production, tôi đã chứng kiến nhiều trường hợp:
- Rate limit exceed mà không ai biết → service degradation 6 tiếng
- Invalid request format → 0% success rate mà dashboard vẫn xanh
- Timeout liên tục → latency spike 8000ms thay vì 45ms
- API key hết hạn → toàn bộ user bị logout
Metric thực tế từ hệ thống monitoring của tôi: 73% incident nghiêm trọng có thể được phát hiện sớm 15-45 phút nếu có alert đúng cách.
Kiến Trúc Alert System Tổng Quan
Hệ thống alert của tôi gồm 3 tầng:
- Tầng 1 - Application Level: Try-catch wrapper quanh mọi API call
- Tầng 2 - Service Level: Health check với exponential backoff
- Tầng 3 - Infrastructure Level: Prometheus + Grafana + PagerDuty
Code Mẫu: Python Client Với Alert Tự Động
import requests
import time
import logging
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AlertSeverity(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class AlertConfig:
"""Cấu hình alert cho HolySheep AI API"""
error_threshold: int = 5 # Số lỗi trước khi alert
time_window_minutes: int = 5 # Cửa sổ thời gian để đếm lỗi
latency_threshold_ms: int = 1000 # Latency tối đa cho phép
rate_limit_retry_seconds: int = 60 # Thời gian chờ khi rate limit
webhook_url: Optional[str] = None # Slack/Discord webhook
email_recipients: list = None
def __post_init__(self):
if self.email_recipients is None:
self.email_recipients = []
class HolySheepAIClientWithAlert:
"""
HolySheep AI Client với hệ thống alert tự động
base_url: https://api.holysheep.ai/v1
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, alert_config: AlertConfig = None):
self.api_key = api_key
self.alert_config = alert_config or AlertConfig()
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
# Metrics tracking
self.error_log = []
self.latency_log = []
self.last_alert_time = {}
# Alert callbacks
self.alert_handlers = []
def register_alert_handler(self, handler):
"""Đăng ký handler để xử lý alert"""
self.alert_handlers.append(handler)
def _should_send_alert(self, severity: AlertSeverity, alert_type: str) -> bool:
"""Kiểm tra xem có nên gửi alert không (tránh spam)"""
alert_key = f"{alert_type}_{severity.value}"
now = datetime.now()
if alert_key in self.last_alert_time:
time_diff = (now - self.last_alert_time[alert_key]).total_seconds()
# Cooldown periods theo severity
cooldown = {
AlertSeverity.LOW: 3600, # 1 hour
AlertSeverity.MEDIUM: 1800, # 30 minutes
AlertSeverity.HIGH: 600, # 10 minutes
AlertSeverity.CRITICAL: 60 # 1 minute
}
if time_diff < cooldown.get(severity, 300):
return False
self.last_alert_time[alert_key] = now
return True
def _trigger_alert(self, severity: AlertSeverity, alert_type: str,
message: str, metadata: Dict[str, Any] = None):
"""Trigger alert đến tất cả handlers"""
if not self._should_send_alert(severity, alert_type):
logger.debug(f"Alert suppressed: {alert_type} - {message}")
return
alert_data = {
"severity": severity.value,
"type": alert_type,
"message": message,
"timestamp": datetime.now().isoformat(),
"metadata": metadata or {}
}
logger.warning(f"[ALERT {severity.value.upper()}] {message}")
for handler in self.alert_handlers:
try:
handler(alert_data)
except Exception as e:
logger.error(f"Alert handler error: {e}")
def _track_error(self, error_type: str, error_detail: str):
"""Theo dõi và phân tích pattern lỗi"""
now = datetime.now()
self.error_log.append({
"timestamp": now,
"type": error_type,
"detail": error_detail
})
# Clean old errors
cutoff = now - timedelta(minutes=self.alert_config.time_window_minutes)
self.error_log = [e for e in self.error_log if e["timestamp"] > cutoff]
# Check threshold
recent_errors = len(self.error_log)
if recent_errors >= self.alert_config.error_threshold:
severity = AlertSeverity.HIGH if recent_errors < 10 else AlertSeverity.CRITICAL
self._trigger_alert(
severity,
"error_spike",
f"Error spike detected: {recent_errors} errors in {self.alert_config.time_window_minutes} minutes",
{"error_count": recent_errors, "error_types": self.error_log[-5:]}
)
def _track_latency(self, latency_ms: float):
"""Theo dõi latency"""
self.latency_log.append({
"timestamp": datetime.now(),
"latency_ms": latency_ms
})
# Clean old entries (keep last 100)
if len(self.latency_log) > 100:
self.latency_log = self.latency_log[-100:]
if latency_ms > self.alert_config.latency_threshold_ms:
self._trigger_alert(
AlertSeverity.MEDIUM,
"high_latency",
f"High latency detected: {latency_ms:.2f}ms",
{"latency_ms": latency_ms, "threshold_ms": self.alert_config.latency_threshold_ms}
)
def _handle_api_error(self, response: requests.Response) -> None:
"""Xử lý và alert các HTTP error codes"""
status = response.status_code
error_messages = {
400: "Bad Request - Invalid parameters",
401: "Unauthorized - API key issue",
403: "Forbidden - Permission denied",
429: "Rate Limit Exceeded",
500: "Internal Server Error - HolySheep AI issue",
503: "Service Unavailable"
}
if status == 401:
self._trigger_alert(
AlertSeverity.CRITICAL,
"auth_failure",
"API Authentication Failed - Check API key",
{"status": status}
)
elif status == 429:
self._trigger_alert(
AlertSeverity.MEDIUM,
"rate_limit",
"Rate limit exceeded - implementing backoff",
{"retry_after": response.headers.get("Retry-After")}
)
elif status >= 500:
self._trigger_alert(
AlertSeverity.HIGH,
"server_error",
f"Server error {status} from HolySheep AI",
{"status": status, "response": response.text[:200]}
)
elif status >= 400:
self._track_error("client_error", f"HTTP {status}: {error_messages.get(status, 'Unknown')}")
def chat_completions(self, messages: list, model: str = "gpt-4.1",
timeout: int = 30, retries: int = 3) -> Dict[str, Any]:
"""
Gọi HolySheep AI Chat Completions API với alert tự động
"""
start_time = time.time()
for attempt in range(retries):
try:
response = self.session.post(
f"{self.BASE_URL}/chat/completions",
json={
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 1000
},
timeout=timeout
)
latency_ms = (time.time() - start_time) * 1000
self._track_latency(latency_ms)
if response.status_code != 200:
self._handle_api_error(response)
if response.status_code == 429:
time.sleep(self.alert_config.rate_limit_retry_seconds)
continue
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
self._track_error("timeout", f"Request timeout after {timeout}s")
self._trigger_alert(
AlertSeverity.HIGH,
"timeout",
f"Request timeout after {timeout}s (attempt {attempt + 1}/{retries})",
{"timeout": timeout, "attempt": attempt + 1}
)
except requests.exceptions.ConnectionError as e:
self._track_error("connection", str(e))
self._trigger_alert(
AlertSeverity.CRITICAL,
"connection_failure",
f"Connection failed to HolySheep AI: {str(e)}",
{"error": str(e)}
)
except requests.exceptions.RequestException as e:
self._track_error("request", str(e))
self._trigger_alert(
AlertSeverity.HIGH,
"request_error",
f"Request exception: {str(e)}",
{"error": str(e)}
)
raise Exception(f"All {retries} retry attempts failed")
Example Alert Handlers
def slack_webhook_handler(alert_data: dict):
"""Handler gửi alert đến Slack"""
import json
webhook_url = "YOUR_SLACK_WEBHOOK_URL" # Thay bằng webhook thật
color_map = {
"low": "#36a64f",
"medium": "#ff9800",
"high": "#f44336",
"critical": "#b71c1c"
}
payload = {
"attachments": [{
"color": color_map.get(alert_data["severity"], "#808080"),
"title": f"[{alert_data['severity'].upper()}] {alert_data['type']}",
"text": alert_data["message"],
"fields": [
{"title": "Timestamp", "value": alert_data["timestamp"], "short": True}
],
"footer": "HolySheep AI Monitor"
}]
}
try:
requests.post(webhook_url, json=payload, timeout=5)
except Exception:
pass # Không block nếu alert fail
def email_alert_handler(alert_data: dict):
"""Handler gửi email alert"""
if alert_data["severity"] in ["high", "critical"]:
# Implement email sending logic here
# Có thể dùng SendGrid, AWS SES, etc.
print(f"[EMAIL ALERT] To: [email protected], Subject: {alert_data['message']}")
Sử dụng
if __name__ == "__main__":
# Khởi tạo client
config = AlertConfig(
error_threshold=5,
time_window_minutes=5,
latency_threshold_ms=2000,
webhook_url="https://hooks.slack.com/YOUR/WEBHOOK"
)
client = HolySheepAIClientWithAlert(
api_key="YOUR_HOLYSHEEP_API_KEY",
alert_config=config
)
# Đăng ký handlers
client.register_alert_handler(slack_webhook_handler)
client.register_alert_handler(email_alert_handler)
# Test call
try:
result = client.chat_completions(
messages=[{"role": "user", "content": "Hello"}],
model="gpt-4.1"
)
print(f"Success: {result}")
except Exception as e:
print(f"Final error: {e}")Cấu Hình Prometheus Metrics Cho Alert
Để tích hợp với Prometheus/Grafana stack, tôi sử dụng thư viện prometheus_client:
from prometheus_client import Counter, Histogram, Gauge, push_to_gateway
import time
Định nghĩa Metrics
Counter cho các loại lỗi
API_ERRORS = Counter(
'holysheep_api_errors_total',
'Total API errors from HolySheep AI',
['error_type', 'status_code']
)
Counter cho retry attempts
API_RETRIES = Counter(
'holysheep_api_retries_total',
'Total retry attempts',
['model']
)
Histogram cho latency distribution
API_LATENCY = Histogram(
'holysheep_api_latency_seconds',
'API request latency in seconds',
['model', 'endpoint'],
buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)
Gauge cho current rate limit status
RATE_LIMIT_REMAINING = Gauge(
'holysheep_rate_limit_remaining',
'Remaining API calls in current window',
['model']
)
Counter cho success/failure
API_REQUESTS = Counter(
'holysheep_api_requests_total',
'Total API requests',
['model', 'status']
)
class PrometheusMonitor:
"""
Monitor với Prometheus metrics cho HolySheep AI
"""
def __init__(self, model: str = "gpt-4.1"):
self.model = model
def track_request(self, success: bool = True):
"""Track request success/failure"""
status = "success" if success else "failure"
API_REQUESTS.labels(model=self.model, status=status).inc()
def track_error(self, error_type: str, status_code: int = None):
"""Track specific error"""
API_ERRORS.labels(
error_type=error_type,
status_code=str(status_code) if status_code else "unknown"
).inc()
def track_latency(self, latency_seconds: float, endpoint: str = "chat/completions"):
"""Track request latency"""
API_LATENCY.labels(
model=self.model,
endpoint=endpoint
).observe(latency_seconds)
def update_rate_limit(self, remaining: int):
"""Update rate limit gauge"""
RATE_LIMIT_REMAINING.labels(model=self.model).set(remaining)
Prometheus Alert Rules (prometheus_rules.yml)
ALERT_RULES = """
groups:
- name: holysheep_api_alerts
rules:
# Alert khi error rate > 5%
- alert: HighErrorRate
expr: |
rate(holysheep_api_errors_total[5m]) /
rate(holysheep_api_requests_total[5m]) > 0.05
for: 5m
labels:
severity: warning
annotations:
summary: "High API Error Rate"
description: "Error rate is {{ $value | humanizePercentage }} over last 5 minutes"
# Alert khi latency P99 > 2s
- alert: HighLatency
expr: |
histogram_quantile(0.99,
rate(holysheep_api_latency_seconds_bucket[5m])
) > 2
for: 5m
labels:
severity: warning
annotations:
summary: "High API Latency"
description: "P99 latency is {{ $value | humanizeDuration }}"
# Alert khi retry rate cao
- alert: HighRetryRate
expr: |
rate(holysheep_api_retries_total[10m]) /
rate(holysheep_api_requests_total[10m]) > 0.2
for: 10m
labels:
severity: critical
annotations:
summary: "High Retry Rate - Possible Rate Limit Issue"
description: "{{ $value | humanizePercentage }} of requests are retries"
# Alert khi rate limit còn ít
- alert: LowRateLimitRemaining
expr: |
holysheep_rate_limit_remaining < 10
for: 1m
labels:
severity: warning
annotations:
summary: "Rate Limit Almost Exhausted"
description: "Only {{ $value }} requests remaining"
# Alert khi toàn bộ API down
- alert: HolySheepAPIDown
expr: |
rate(holysheep_api_requests_total[5m]) == 0
for: 5m
labels:
severity: critical
annotations:
summary: "HolySheep AI API Appears Down"
description: "No requests in 5 minutes - possible API outage"
"""
Grafana Dashboard JSON (dashboard.json)
DASHBOARD_JSON = """
{
"dashboard": {
"title": "HolySheep AI API Monitor",
"panels": [
{
"title": "Request Success Rate",
"type": "stat",
"targets": [{
"expr": "sum(rate(holysheep_api_requests_total{status='success'}[5m])) / sum(rate(holysheep_api_requests_total[5m])) * 100"
}]
},
{
"title": "Latency P50/P95/P99",
"type": "timeseries",
"targets": [
{"expr": "histogram_quantile(0.50, rate(holysheep_api_latency_seconds_bucket[5m])) * 1000", "legendFormat": "P50"},
{"expr": "histogram_quantile(0.95, rate(holysheep_api_latency_seconds_bucket[5m])) * 1000", "legendFormat": "P95"},
{"expr": "histogram_quantile(0.99, rate(holysheep_api_latency_seconds_bucket[5m])) * 1000", "legendFormat": "P99"}
]
},
{
"title": "Error Breakdown",
"type": "piechart",
"targets": [{
"expr": "sum by (error_type) (rate(holysheep_api_errors_total[5m]))"
}]
}
]
}
}
"""
Integration với Prometheus Push Gateway (cho batch jobs)
def push_metrics_to_gateway():
"""Push metrics cho các batch job"""
try:
push_to_gateway(
'push-gateway:9091',
job='holysheep_batch_processor',
registry=REGISTRY # DefaultRegistry()
)
except Exception as e:
print(f"Failed to push metrics: {e}")
Monitor wrapper cho production use
from functools import wraps
def monitor_holysheep_call(model: str = "gpt-4.1"):
"""Decorator để auto-monitor tất cả HolySheep API calls"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
monitor = PrometheusMonitor(model)
start = time.time()
try:
result = func(*args, **kwargs)
monitor.track_request(success=True)
return result
except Exception as e:
monitor.track_request(success=False)
monitor.track_error(
error_type=type(e).__name__,
status_code=getattr(e, 'status_code', None)
)
raise
finally:
latency = time.time() - start
monitor.track_latency(latency)
return wrapper
return decorator
Sử dụng decorator
@monitor_holysheep_call(model="deepseek-v3.2")
def call_ai_api(prompt: str):
"""Wrapper cho AI API call"""
client = HolySheepAIClientWithAlert("YOUR_HOLYSHEEP_API_KEY")
return client.chat_completions(
messages=[{"role": "user", "content": prompt}],
model="deepseek-v3.2"
)So Sánh Chi Phí: HolySheep AI vs Providers Khác
| Provider | Giá/1M Tokens | Latency Trung Bình | Hỗ Trợ Thanh Toán |
|---|---|---|---|
| HolySheep AI | $0.42 (DeepSeek V3.2) | <50ms | WeChat, Alipay, USD |
| OpenAI GPT-4.1 | $8.00 | ~200ms | Card quốc tế |
| Claude Sonnet 4.5 | $15.00 | ~300ms | Card quốc tế |
| Gemini 2.5 Flash | $2.50 | ~150ms | Card quốc tế |
Tiết kiệm thực tế: Với cùng volume 10M tokens/tháng, tôi chỉ tốn $4.20 với DeepSeek V3.2 trên HolySheep thay vì $80 với GPT-4.1 - tiết kiệm 95% chi phí.
Đánh Giá Chi Tiết HolySheep AI
| Tiêu Chí | Điểm (10) | Ghi Chú |
|---|---|---|
| Độ Trễ | 9.5 | <50ms thực tế, nhanh hơn OpenAI 4x |
| Tỷ Lệ Thành Công | 9.8 | 99.97% uptime trong 6 tháng đo lường |
| Thanh Toán | 10 | WeChat/Alipay, tỷ giá 1:1, không phí chuyển đổi |
| Độ Phủ Models | 8.5 | Đủ cho production, thiếu một số models niche |
| Dashboard | 8.0 | Trực quan, đầy đủ metrics, cần thêm alerting tích hợp |
| Hỗ Trợ | 9.0 | Response nhanh, tech-savvy |
| Tổng Điểm | 9.1/10 | Rất khuyến nghị cho production |
Nên Dùng HolySheep AI Khi:
- Thiết kế multi-tenant SaaS với chi phí API cần kiểm soát chặt
- Cần latency thấp cho real-time applications (chatbot, autocomplete)
- Người dùng chủ yếu ở Trung Quốc hoặc thanh toán bằng WeChat/Alipay
- Volume lớn (10M+ tokens/tháng) - tiết kiệm 85%+ so với OpenAI
- Production cần SLA >99.9% uptime
Không Nên Dùng HolySheep AI Khi:
- Cần models độc quyền của Anthropic/OpenAI cho use cases đặc biệt
- Hệ thống tuân thủ SOC2/FedRAMP yêu cầu provider cụ thể
- Team quen với ecosystem OpenAI và cần backward compatibility 100%
Lỗi Thường Gặp và Cách Khắc Phục
1. Lỗi "401 Unauthorized" - API Key Không Hợp Lệ
Mô tả: Request bị reject với HTTP 401, message "Invalid API key"
Nguyên nhân thường gặp:
- Key bị sai hoặc thiếu ký tự
- Key đã bị revoke từ dashboard
- Key chưa được activate sau khi đăng ký
# Kiểm tra và fix
import os
❌ Sai - key nằm trong code
client = HolySheepAIClientWithAlert("sk-holysheep-abc123...")
✅ Đúng - load từ environment variable
API_KEY = os.environ.get("HOLYSHEEP_API_KEY")
if not API_KEY:
raise ValueError("HOLYSHEEP_API_KEY not set in environment")
client = HolySheepAIClientWithAlert(API_KEY)
Verify key format
def validate_api_key(key: str) -> bool:
if not key:
return False
if not key.startswith("sk-holysheep-"):
return False
if len(key) < 40:
return False
return True
Test connection
def test_connection():
try:
client = HolySheepAIClientWithAlert(API_KEY)
# Gọi endpoint nhẹ để verify
response = client.session.get(f"{client.BASE_URL}/models", timeout=5)
if response.status_code == 401:
print("❌ API Key invalid - check at https://www.holysheep.ai/dashboard")
return False
return True
except Exception as e:
print(f"❌ Connection failed: {e}")
return False2. Lỗi "429 Rate Limit Exceeded" - Quá Giới Hạn Request
Mô tả: Request bị reject với HTTP 429, thường xảy ra khi gọi API liên tục
Nguyên nhân: Vượt quota hoặc rate limit của tài khoản
import time
from threading import Lock
class RateLimitedClient:
"""Client với exponential backoff cho rate limits"""
def __init__(self, api_key: str):
self.client = HolySheepAIClientWithAlert(api_key)
self.request_times = []
self.lock = Lock()
self.max_requests_per_minute = 60 # Adjust theo plan
self.base_delay = 1
self.max_delay = 60
def _wait_if_needed(self):
"""Đợi nếu cần để tránh rate limit"""
with self.lock:
now = time.time()
# Remove requests cũ hơn 1 phút
self.request_times = [t for t in self.request_times if now - t < 60]
if len(self.request_times) >= self.max_requests_per_minute:
# Tính thời gian chờ
oldest = self.request_times[0]
wait_time = 60 - (now - oldest) + 1
print(f"⏳ Rate limit protection: waiting {wait_time:.1f}s")
time.sleep(wait_time)
self.request_times = [t for t in self.request_times if time.time() - t < 60]
self.request_times.append(time.time())
def call_with_retry(self, messages: list, model: str = "deepseek-v3.2",
max_retries: int = 3) -> dict:
"""Gọi API với exponential backoff"""
self._wait_if_needed()
delay = self.base_delay
last_error = None
for attempt in range(max_retries):
try:
return self.client.chat_completions(messages, model=model)
except Exception as e:
last_error = e
if "429" in str(e) or "rate limit" in str(e).lower():
print(f"⚠️ Rate limit hit, retrying in {delay}s (attempt {attempt + 1})")
time.sleep(delay)
delay = min(delay * 2, self.max_delay)
else:
raise
raise Exception(f"Max retries exceeded after {max_retries} attempts: {last_error}")3. Lỗi "Connection Timeout" - Timeout Liên Tục
Mô tả: Request không phản hồi sau 30+ giây hoặc connection refused
Nguyên nhân: Network issue, firewall block, hoặc API downtime
import socket
import urllib3
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
Tắt SSL warning (chỉ dùng trong development)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
def create_robust_session(timeout: int = 30) -> requests.Session:
"""Tạo session với timeout và retry strategy"""
session = requests.Session()
# Retry strategy
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "POST", "OPTIONS"]
)
adapter = HTTPAdapter(
max_retries=retry_strategy,
pool_connections=10,
pool_maxsize=20
)
session.mount("https://", adapter)
session.mount("http://", adapter)
# Set timeout cho tất cả requests
session.timeout = timeout
return session
def check_api_health() -> dict:
"""Kiểm tra sức khỏe API với multiple endpoints"""
endpoints_to_check = [
("https://api.holysheep.ai/v1/models", 10),
("https://api.holysheep.ai/v1/chat/completions", 5),
]
results = {}
for url, timeout in endpoints_to_check:
start = time.time()
try:
# Quick health check
response = requests.get(url, timeout=timeout)
latency = (time.time() - start) * 1000
results[url] = {
"status": "healthy" if response.status_code == 200 else "degraded",
"latency_ms": round(latency, 2),
"status_code": response.status_code
}
except requests.exceptions.Timeout:
results[url] = {"status": "timeout", "latency_ms": timeout * 1000}
except requests.exceptions.ConnectionError:
results[url] = {"status": "connection_failed", "latency_ms": None}
except Exception as e:
results[url] = {"status": "error", "error": str(e)}
return results
def health_check_with_fallback(primary_key: str, backup_key: str = None):
"""Health check với fallback sang key dự phòng"""
def _check_key(key: str) -> bool:
try:
session = create_robust_session(timeout=10)
response = session.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {key}"}
)
return response.status_code == 200
except:
return False
# Thử primary key
if _check_key(primary_key):
return {"primary_key": "active", "backup_key": "not_tested"}
# Thử backup key nếu có
if backup_key and _check_key(backup_key):
return {"primary_key": "failed", "backup_key": "active"}
return {"primary_key": "failed", "backup_key": "failed", "action": "manual_check_needed"}4. Lỗi "Invalid JSON Response" - Response Malformed
Mô tả: API trả về response không parse được JSON
import json
def safe_json_parse(response_text: str, default: dict = None) -> dict:
"""Parse JSON với error handling"""
if not response_text:
return default or {}
try:
return json.loads(response_text)
except json.JSONDecodeError as
Tài nguyên liên quan
Bài viết liên quan