As applications increasingly rely on AI services, understanding and monitoring Service Level Agreements (SLAs) becomes critical for maintaining reliable products. In this hands-on tutorial, I will walk you through setting up comprehensive SLA monitoring for your AI API integrations from absolute zero knowledge. Whether you are building a startup MVP or managing enterprise AI infrastructure, this guide will help you track, measure, and report on your service quality metrics effectively.
What is SLA Monitoring and Why Should You Care?
Service Level Agreement monitoring measures whether your AI API provider delivers on their promised performance standards. Think of it like checking the speedometer while driving—without monitoring, you are flying blind. A typical SLA includes commitments on uptime percentage, response latency, error rates, and throughput capacity. When I first started building AI-powered applications, I learned this lesson the hard way when a third-party API outage cost me an entire weekend of debugging without any visibility into what went wrong.
For HolySheep AI, the service guarantees less than 50ms latency on most requests, which means your applications can deliver responsive user experiences. By monitoring these metrics yourself, you can catch degradation before it impacts users, validate whether your provider meets their commitments, and gather data for negotiating better terms or switching providers if necessary. Sign up here to access their high-performance API with transparent pricing at just $1 per dollar (saving 85% compared to typical ¥7.3 rates) and start monitoring your SLA today.
Understanding Key SLA Metrics
Before diving into code, let us understand the metrics that matter most for AI API services.
Uptime Percentage
Uptime measures the percentage of time your API service is available and responding successfully. The industry standard is often 99.9% ("three nines"), which translates to approximately 8.7 hours of allowable downtime per year. HolySheep AI provides robust uptime guarantees, but you should always verify this independently through your own monitoring.
Response Latency
Latency measures the time between sending a request and receiving a response. HolySheep AI advertises less than 50ms latency, but real-world measurements vary based on request complexity, network conditions, and server load. For AI APIs specifically, you will encounter two types of latency: Time to First Byte (TTFB) for initial response and Total Response Time for complete responses.
Error Rate
Error rate is the percentage of requests that fail or return error responses. Common error types include HTTP 4xx client errors (bad request, unauthorized) and HTTP 5xx server errors (internal server error, service unavailable). A healthy SLA typically targets less than 0.1% error rate for server-side issues.
Throughput and Rate Limits
Throughput measures how many requests your application can process per second, while rate limits define the maximum allowed requests within a time window. HolySheep AI offers flexible pricing tiers that accommodate everything from development testing to production-scale deployments with competitive rates like DeepSeek V3.2 at just $0.42 per million tokens.
Setting Up Your Monitoring Environment
You will need Python installed on your system, along with a few essential libraries. Install the required packages using pip:
# Install monitoring dependencies
pip install requests pandas matplotlib time datetime
For this tutorial, we will create a comprehensive monitoring script that tracks all critical SLA metrics. Create a new Python file called sla_monitor.py and add the following foundation code:
# sla_monitor.py
import requests
import time
import json
from datetime import datetime
from collections import defaultdict
class SLAProbe:
"""
HolySheep AI SLA Monitoring Probe
This class measures real-world API performance metrics
"""
def __init__(self, api_key, base_url="https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.metrics = {
'latencies': [],
'errors': [],
'successes': 0,
'failures': 0,
'rate_limit_hits': 0
}
def check_health(self):
"""
Check API health endpoint - should return within milliseconds
"""
start = time.perf_counter()
try:
response = requests.get(
f"{self.base_url}/health",
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=5
)
latency = (time.perf_counter() - start) * 1000 # Convert to ms
if response.status_code == 200:
return {'success': True, 'latency': latency, 'data': response.json()}
else:
return {'success': False, 'latency': latency, 'error': response.text}
except Exception as e:
return {'success': False, 'latency': None, 'error': str(e)}
def probe_chat_completion(self, prompt="Hello, test message"):
"""
Test actual AI completion endpoint and measure performance
"""
start = time.perf_counter()
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 50
},
timeout=30
)
latency = (time.perf_counter() - start) * 1000
if response.status_code == 200:
self.metrics['successes'] += 1
self.metrics['latencies'].append(latency)
return {'success': True, 'latency': latency, 'response': response.json()}
elif response.status_code == 429:
self.metrics['rate_limit_hits'] += 1
return {'success': False, 'latency': latency, 'error': 'Rate limited'}
else:
self.metrics['failures'] += 1
self.metrics['errors'].append({'status': response.status_code, 'body': response.text})
return {'success': False, 'latency': latency, 'error': response.text}
except requests.exceptions.Timeout:
self.metrics['failures'] += 1
return {'success': False, 'latency': None, 'error': 'Request timeout'}
except Exception as e:
self.metrics['failures'] += 1
return {'success': False, 'latency': None, 'error': str(e)}
def calculate_sla_metrics(self):
"""
Calculate SLA metrics from collected data
"""
total_requests = self.metrics['successes'] + self.metrics['failures']
if total_requests == 0:
return None
uptime = (self.metrics['successes'] / total_requests) * 100
error_rate = (self.metrics['failures'] / total_requests) * 100
avg_latency = sum(self.metrics['latencies']) / len(self.metrics['latencies']) if self.metrics['latencies'] else 0
p50_latency = sorted(self.metrics['latencies'])[len(self.metrics['latencies']) // 2] if self.metrics['latencies'] else 0
p95_latency = sorted(self.metrics['latencies'])[int(len(self.metrics['latencies']) * 0.95)] if self.metrics['latencies'] else 0
p99_latency = sorted(self.metrics['latencies'])[int(len(self.metrics['latencies']) * 0.99)] if self.metrics['latencies'] else 0
return {
'total_requests': total_requests,
'successes': self.metrics['successes'],
'failures': self.metrics['failures'],
'uptime_percentage': round(uptime, 4),
'error_rate_percentage': round(error_rate, 4),
'average_latency_ms': round(avg_latency, 2),
'p50_latency_ms': round(p50_latency, 2),
'p95_latency_ms': round(p95_latency, 2),
'p99_latency_ms': round(p99_latency, 2),
'rate_limit_hits': self.metrics['rate_limit_hits']
}
def generate_report(self):
"""
Generate human-readable SLA report
"""
metrics = self.calculate_sla_metrics()
if not metrics:
return "No metrics collected yet"
report = f"""
=== HolySheep AI SLA Monitoring Report ===
Generated: {datetime.now().isoformat()}
Total Requests: {metrics['total_requests']}
Successful: {metrics['successes']}
Failed: {metrics['failures']}
UPTIME: {metrics['uptime_percentage']}% {'✓ MET' if metrics['uptime_percentage'] >= 99.9 else '✗ BELOW TARGET'}
ERROR RATE: {metrics['error_rate_percentage']}% {'✓ MET' if metrics['error_rate_percentage'] <= 0.1 else '✗ EXCEEDED'}
LATENCY METRICS (in milliseconds):
Average: {metrics['average_latency_ms']}ms
P50 (Median): {metrics['p50_latency_ms']}ms
P95: {metrics['p95_latency_ms']}ms
P99: {metrics['p99_latency_ms']}ms
Target (<50ms): {'✓ MET' if metrics['average_latency_ms'] < 50 else '✗ EXCEEDED'}
RATE LIMIT HITS: {metrics['rate_limit_hits']}
=========================================
"""
return report
Example usage
if __name__ == "__main__":
PROBE_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
monitor = SLAProbe(PROBE_API_KEY)
# Run 10 probe requests
print("Running SLA probes...")
for i in range(10):
result = monitor.probe_chat_completion(f"Test request {i+1}")
print(f"Request {i+1}: {'✓' if result['success'] else '✗'} - {result.get('latency', 'N/A')}ms")
time.sleep(0.5) # Rate limiting
# Generate and display report
print(monitor.generate_report())
Run this script with your HolySheep API key to start collecting baseline metrics. You should see output similar to this (actual numbers will vary based on network conditions):
Running SLA probes...
Request 1: ✓ - 127.43ms
Request 2: ✓ - 98.21ms
Request 3: ✓ - 112.87ms
Request 4: ✓ - 134.56ms
Request 5: ✓ - 108.92ms
Request 6: ✓ - 99.34ms
Request 7: ✓ - 145.23ms
Request 8: ✓ - 101.45ms
Request 9: ✓ - 117.88ms
Request 10: ✓ - 95.67ms
Building Automated Continuous Monitoring
One-time probes do not give you the complete picture. Production systems require continuous monitoring that runs periodically and stores historical data. Let me show you how to build a monitoring daemon that runs continuously in the background.
# continuous_monitor.py
import requests
import time
import json
import sqlite3
from datetime import datetime, timedelta
from threading import Thread, Event
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
class ContinuousMonitor:
"""
Continuous SLA monitoring with persistent storage
Runs background probes and maintains historical records
"""
def __init__(self, api_key, interval_seconds=60):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.interval = interval_seconds
self.stop_event = Event()
self.db_path = "sla_metrics.db"
self.init_database()
def init_database(self):
"""Create SQLite database for storing metrics"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS probe_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
endpoint TEXT NOT NULL,
success INTEGER NOT NULL,
latency_ms REAL,
status_code INTEGER,
error_message TEXT
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS aggregated_metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
window_start TEXT NOT NULL,
window_end TEXT NOT NULL,
total_requests INTEGER,
successes INTEGER,
failures INTEGER,
avg_latency_ms REAL,
p95_latency_ms REAL,
error_rate REAL
)
""")
conn.commit()
conn.close()
def store_result(self, endpoint, success, latency, status_code, error):
"""Store probe result in database"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO probe_results (timestamp, endpoint, success, latency_ms, status_code, error_message)
VALUES (?, ?, ?, ?, ?, ?)
""", (datetime.now().isoformat(), endpoint, int(success), latency, status_code, error))
conn.commit()
conn.close()
def probe_endpoint(self, endpoint, method="GET", payload=None):
"""Execute single probe against endpoint"""
url = f"{self.base_url}/{endpoint}"
headers = {"Authorization": f"Bearer {self.api_key}"}
if payload:
headers["Content-Type"] = "application/json"
start = time.perf_counter()
try:
if method == "GET":
response = requests.get(url, headers=headers, timeout=10)
else:
response = requests.post(url, headers=headers, json=payload, timeout=30)
latency = (time.perf_counter() - start) * 1000
success = 200 <= response.status_code < 300
self.store_result(endpoint, success, latency, response.status_code, None if success else response.text)
return success, latency, response.status_code
except Exception as e:
latency = (time.perf_counter() - start) * 1000
self.store_result(endpoint, False, latency, None, str(e))
return False, latency, None
def run_probe_cycle(self):
"""Execute one complete probe cycle"""
print(f"[{datetime.now().strftime('%H:%M:%S')}] Running probe cycle...")
# Probe health endpoint (fast, lightweight)
success, latency, status = self.probe_endpoint("health")
print(f" Health: {'✓' if success else '✗'} {status} - {latency:.2f}ms")
# Probe actual AI completion (measures real-world latency)
success, latency, status = self.probe_endpoint(
"chat/completions",
method="POST",
payload={"model": "gpt-4.1", "messages": [{"role": "user", "content": "Ping"}], "max_tokens": 10}
)
print(f" AI Completion: {'✓' if success else '✗'} {status} - {latency:.2f}ms")
def calculate_hourly_stats(self):
"""Calculate aggregated statistics for the last hour"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
one_hour_ago = (datetime.now() - timedelta(hours=1)).isoformat()
cursor.execute("""
SELECT
COUNT(*) as total,
SUM(success) as successes,
COUNT(*) - SUM(success) as failures,
AVG(latency_ms) as avg_latency,
AVG(CASE WHEN latency_ms IS NOT NULL THEN latency_ms ELSE NULL END) as avg_non_null
FROM probe_results
WHERE timestamp > ?
""", (one_hour_ago,))
row = cursor.fetchone()
conn.close()
if row and row[0] > 0:
total, successes, failures, avg_lat, avg_non_null = row
success_rate = (successes / total * 100) if total > 0 else 0
error_rate = (failures / total * 100) if total > 0 else 0
return {
'total': total,
'successes': successes,
'failures': failures,
'success_rate': success_rate,
'error_rate': error_rate,
'avg_latency': avg_non_null or avg_lat or 0
}
return None
def monitor_loop(self):
"""Main monitoring loop - runs until stop_event is set"""
print("Starting continuous monitoring...")
print(f"Probe interval: {self.interval} seconds")
print("Press Ctrl+C to stop\n")
while not self.stop_event.is_set():
self.run_probe_cycle()
self.stop_event.wait(self.interval)
def start(self):
"""Start monitoring in background thread"""
self.thread = Thread(target=self.monitor_loop)
self.thread.daemon = True
self.thread.start()
return self
def stop(self):
"""Stop monitoring"""
print("\nStopping monitor...")
self.stop_event.set()
self.thread.join(timeout=5)
Interactive demo
if __name__ == "__main__":
monitor = ContinuousMonitor(
api_key="YOUR_HOLYSHEEP_API_KEY",
interval_seconds=30
)
try:
monitor.start()
# Run for 5 minutes for demo, remove this in production
time.sleep(300)
except KeyboardInterrupt:
pass
finally:
monitor.stop()
# Display hourly summary
stats = monitor.calculate_hourly_stats()
if stats:
print("\n=== Hourly Summary ===")
print(f"Total Probes: {stats['total']}")
print(f"Success Rate: {stats['success_rate']:.2f}%")
print(f"Error Rate: {stats['error_rate']:.2f}%")
print(f"Average Latency: {stats['avg_latency']:.2f}ms")
This continuous monitor stores all results in a SQLite database, allowing you to query historical performance and identify patterns over time. For a production deployment, consider using this alongside a proper metrics platform like Prometheus or Grafana for advanced alerting and visualization.
Creating SLA Compliance Reports
Now that you have data flowing in, let us create a report generator that calculates SLA compliance over defined periods. This is essential for business stakeholders who need to verify service quality guarantees.
# sla_report_generator.py
import sqlite3
from datetime import datetime, timedelta
import json
class SLAReportGenerator:
"""
Generate SLA compliance reports for HolySheep AI
"""
def __init__(self, db_path="sla_metrics.db"):
self.db_path = db_path
def get_data_for_period(self, start_date, end_date, endpoint=None):
"""Query probe results for a specific time period"""
conn = sqlite3.connect(self.db_path)
if endpoint:
query = """
SELECT * FROM probe_results
WHERE timestamp >= ? AND timestamp <= ? AND endpoint = ?
ORDER BY timestamp
"""
results = conn.execute(query, (start_date.isoformat(), end_date.isoformat(), endpoint)).fetchall()
else:
query = """
SELECT * FROM probe_results
WHERE timestamp >= ? AND timestamp <= ?
ORDER BY timestamp
"""
results = conn.execute(query, (start_date.isoformat(), end_date.isoformat())).fetchall()
conn.close()
return results
def calculate_sla_metrics(self, results):
"""Calculate SLA metrics from raw probe results"""
if not results:
return None
total = len(results)
successes = sum(1 for r in results if r[2]) # success column
failures = total - successes
latencies = [r[3] for r in results if r[3] is not None]
latencies.sort()
uptime_percentage = (successes / total) * 100
error_rate = (failures / total) * 100
return {
'total_requests': total,
'successful_requests': successes,
'failed_requests': failures,
'uptime_percentage': uptime_percentage,
'error_rate_percentage': error_rate,
'avg_latency_ms': sum(latencies) / len(latencies) if latencies else 0,
'p50_latency_ms': latencies[len(latencies) // 2] if latencies else 0,
'p95_latency_ms': latencies[int(len(latencies) * 0.95)] if latencies else 0,
'p99_latency_ms': latencies[int(len(latencies) * 0.99)] if latencies else 0,
'min_latency_ms': min(latencies) if latencies else 0,
'max_latency_ms': max(latencies) if latencies else 0
}
def check_sla_compliance(self, metrics, sla_targets):
"""Check if metrics meet SLA targets"""
compliance = {
'uptime': {
'target': sla_targets.get('min_uptime', 99.9),
'actual': metrics['uptime_percentage'],
'met': metrics['uptime_percentage'] >= sla_targets.get('min_uptime', 99.9)
},
'error_rate': {
'target': sla_targets.get('max_error_rate', 0.1),
'actual': metrics['error_rate_percentage'],
'met': metrics['error_rate_percentage'] <= sla_targets.get('max_error_rate', 0.1)
},
'latency': {
'target': sla_targets.get('max_latency_ms', 100),
'actual': metrics['avg_latency_ms'],
'met': metrics['avg_latency_ms'] <= sla_targets.get('max_latency_ms', 100)
}
}
compliance['all_met'] = all(c['met'] for c in compliance.values())
return compliance
def generate_weekly_report(self):
"""Generate comprehensive weekly SLA report"""
end_date = datetime.now()
start_date = end_date - timedelta(days=7)
results = self.get_data_for_period(start_date, end_date)
metrics = self.calculate_sla_metrics(results)
if not metrics:
return "No data available for this period"
sla_targets = {
'min_uptime': 99.9,
'max_error_rate': 0.1,
'max_latency_ms': 50
}
compliance = self.check_sla_compliance(metrics, sla_targets)
report = f"""
================================================================================
HOLYSHEEP AI - WEEKLY SLA REPORT
================================================================================
Report Period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}
Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
--------------------------------------------------------------------------------
EXECUTIVE SUMMARY
--------------------------------------------------------------------------------
SLA Compliance: {'✓ FULLY COMPLIANT' if compliance['all_met'] else '⚠ PARTIALLY NON-COMPLIANT'}
SERVICE AVAILABILITY
Target: {sla_targets['min_uptime']}% uptime
Actual: {metrics['uptime_percentage']:.4f}%
Status: {'✓ MET' if compliance['uptime']['met'] else '✗ BREACH'}
ERROR PERFORMANCE
Target: ≤{sla_targets['max_error_rate']}% error rate
Actual: {metrics['error_rate_percentage']:.4f}%
Status: {'✓ MET' if compliance['error_rate']['met'] else '✗ BREACH'}
RESPONSE LATENCY
Target: ≤{sla_targets['max_latency_ms']}ms average
Actual: {metrics['avg_latency_ms']:.2f}ms average
Status: {'✓ MET' if compliance['latency']['met'] else '✗ BREACH'}
--------------------------------------------------------------------------------
DETAILED METRICS
--------------------------------------------------------------------------------
Total Probe Requests: {metrics['total_requests']}
Successful Requests: {metrics['successful_requests']}
Failed Requests: {metrics['failed_requests']}
Latency Distribution:
Minimum: {metrics['min_latency_ms']:.2f}ms
Average: {metrics['avg_latency_ms']:.2f}ms
P50 (Median): {metrics['p50_latency_ms']:.2f}ms
P95: {metrics['p95_latency_ms']:.2f}ms
P99: {metrics['p99_latency_ms']:.2f}ms
Maximum: {metrics['max_latency_ms']:.2f}ms
--------------------------------------------------------------------------------
PRICING REFERENCE
--------------------------------------------------------------------------------
HolySheep AI Current Rates (2026):
GPT-4.1: $8.00 per 1M tokens
Claude Sonnet 4.5: $15.00 per 1M tokens
Gemini 2.5 Flash: $2.50 per 1M tokens
DeepSeek V3.2: $0.42 per 1M tokens
Cost Advantage: ¥1 = $1 (85%+ savings vs. typical ¥7.3 rates)
Payment Methods: WeChat Pay, Alipay, Credit Card
--------------------------------------------------------------------------------
================================================================================
"""
return report
Generate sample report
if __name__ == "__main__":
generator = SLAReportGenerator()
print(generator.generate_weekly_report())
# Export to JSON for integration with other tools
# json_output = generator.export_to_json(period_days=30)
Pro tip: Schedule this report to run automatically every week using cron jobs (Linux/Mac) or Task Scheduler (Windows) and email the output to your stakeholders. This creates a paper trail of SLA compliance that proves valuable during contract renewals or service credit negotiations.
Setting Up Real-Time Alerts
Reports are great for historical analysis, but you need real-time alerts to catch issues before they become major problems. Here is a simple alerting system that notifies you when SLA thresholds are breached:
# sla_alerting.py
import requests
import time
import sqlite3
from datetime import datetime
from threading import Thread, Lock
class SLAAlertManager:
"""
Real-time alerting for SLA breaches
Supports multiple notification channels
"""
def __init__(self, api_key, config=None):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.db_path = "sla_metrics.db"
self.config = config or self.default_config()
self.alert_state = {}
self.state_lock = Lock()
self.cooldown_period = 300 # 5 minutes between repeat alerts
def default_config(self):
"""Default alerting thresholds"""
return {
'uptime_threshold': 99.0, # Alert if uptime drops below 99%
'error_rate_threshold': 1.0, # Alert if error rate exceeds 1%
'latency_threshold': 200, # Alert if latency exceeds 200ms
'consecutive_failures': 3, # Alert after 3 consecutive failures
'webhook_url': None, # Optional webhook for Slack/Discord
'email_to': None # Optional email notifications
}
def check_current_health(self):
"""Perform health check on HolySheep API"""
start = time.perf_counter()
try:
response = requests.get(
f"{self.base_url}/health",
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=5
)
latency = (time.perf_counter() - start) * 1000
return {
'success': response.status_code == 200,
'latency': latency,
'timestamp': datetime.now()
}
except Exception as e:
return {
'success': False,
'latency': None,
'timestamp': datetime.now(),
'error': str(e)
}
def get_recent_error_rate(self, minutes=5):
"""Calculate error rate from recent probes"""
conn = sqlite3.connect(self.db_path)
since = (datetime.now().timestamp() - minutes * 60) * 1000
cursor = conn.cursor()
cursor.execute("""
SELECT COUNT(*) as total,
SUM(CASE WHEN success = 1 THEN 1 ELSE 0 END) as successes
FROM probe_results
WHERE timestamp > datetime('now', '-' || ? || ' minutes')
""", (minutes,))
result = cursor.fetchone()
conn.close()
if result and result[0] > 0:
error_rate = ((result[0] - result[1]) / result[0]) * 100
return {'error_rate': error_rate, 'total_requests': result[0]}
return {'error_rate': 0, 'total_requests': 0}
def send_alert(self, alert_type, message, severity="warning"):
"""Send alert through configured channels"""
alert = {
'timestamp': datetime.now().isoformat(),
'type': alert_type,
'severity': severity,
'message': message
}
print(f"[ALERT {severity.upper()}] {alert_type}: {message}")
# Send to webhook if configured
if self.config.get('webhook_url'):
try:
requests.post(
self.config['webhook_url'],
json=alert,
timeout=5
)
except Exception as e:
print(f"Failed to send webhook: {e}")
return alert
def check_and_alert(self):
"""Check current metrics and fire alerts if thresholds breached"""
alerts_triggered = []
# Check consecutive failures
health = self.check_current_health()
if not health['success']:
with self.state_lock:
self.alert_state['consecutive_failures'] = self.alert_state.get('consecutive_failures', 0) + 1
last_alert = self.alert_state.get('last_failure_alert', 0)
if (self.alert_state['consecutive_failures'] >= self.config['consecutive_failures'] and
time.time() - last_alert > self.cooldown_period):
alerts_triggered.append(
self.send_alert(
'CONSECUTIVE_FAILURES',
f"{self.alert_state['consecutive_failures']} consecutive API failures detected",
severity='critical'
)
)
self.alert_state['last_failure_alert'] = time.time()
else:
with self.state_lock:
self.alert_state['consecutive_failures'] = 0
# Check latency
if health.get('latency') and health['latency'] > self.config['latency_threshold']:
with self.state_lock:
last_alert = self.alert_state.get('last_latency_alert', 0)
if time.time() - last_alert > self.cooldown_period:
alerts_triggered.append(
self.send_alert(
'HIGH_LATENCY',
f"Latency {health['latency']:.2f}ms exceeds threshold {self.config['latency_threshold']}ms",
severity='warning'
)
)
self.alert_state['last_latency_alert'] = time.time()
# Check error rate (5-minute window)
recent = self.get_recent_error_rate(minutes=5)
if recent['total_requests'] > 0 and recent['error_rate'] > self.config['error_rate_threshold']:
with self.state_lock:
last_alert = self.alert_state.get('last_error_rate_alert', 0)
if time.time() - last_alert > self.cooldown_period:
alerts_triggered.append(
self.send_alert(
'HIGH_ERROR_RATE',
f"Error rate {recent['error_rate']:.2f}% exceeds threshold {self.config['error_rate_threshold']}%",
severity='warning'
)
)
self.alert_state['last_error_rate_alert'] = time.time()
return alerts_triggered
Example usage with monitoring loop
if __name__ == "__main__":
alert_manager = SLAAlertManager(
api_key="YOUR_HOLYSHEEP_API_KEY",
config={
'latency_threshold': 150, # Alert if latency exceeds 150ms
'error_rate_threshold': 2.0, # Alert if error rate exceeds 2%
'webhook_url': 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL'
}
)
print("Starting alert monitoring (Ctrl+C to stop)...")
try:
while True:
alerts = alert_manager.check_and_alert()
if alerts:
print(f"Triggered {len(alerts)} alert(s)")
time.sleep(30) # Check every 30 seconds
except KeyboardInterrupt:
print("\nAlert monitoring stopped.")
Common Errors and Fixes
Throughout my experience implementing SLA monitoring for AI APIs, I have encountered numerous pitfalls. Here are the most common issues and their solutions:
Error 401: Authentication Failed
This error indicates that your API key is invalid, missing, or malformed. Double-check that you have properly set the Authorization header with the Bearer prefix.
# WRONG - Missing Bearer prefix
headers = {"Authorization": "YOUR_HOLYSHEEP_API_KEY"}
CORRECT - Proper Bearer token format
headers = {"Authorization": f"Bearer {api_key}"}
If you see this error, verify:
1. API key is correctly copied (no extra spaces or characters)
2. Key has not expired or been revoked
3. Key has appropriate permissions for the endpoint
Error 429: Rate Limit Exceeded
Rate limiting is common when running continuous probes. Implement exponential backoff to handle this gracefully without overwhelming the API.
import time
import random
def probe_with_backoff(monitor, max_retries=5):
"""
Probe with exponential backoff on rate limit errors
"""
for attempt in range(max_retries):
result = monitor.probe_chat_completion("Test")
if result['success']:
return result
elif 'Rate limited' in str(result.get('error', '')):
# Exponential backoff: 1s, 2s, 4s, 8s, 16s
wait_time = (2 ** attempt) + random.uniform(0, 1)
print(f"Rate limited. Waiting {wait_time:.2f}s before retry...")
time.sleep(wait_time)
else:
raise Exception(f"Non-retryable error: {result.get('error')}")
raise Exception("Max retries exceeded due to rate limiting")
Timeout Errors: Request Timeout
Timeout errors can indicate network issues, server overload, or requests that are too complex. Configure appropriate timeout values and implement circuit breakers for resilience.