Đêm qua, hệ thống chatbot của tôi ngừng hoạt động lúc 2:47 sáng. Không ai phát hiện cho đến khi khách hàng bắt đầu gọi điện phàn nàn. Tôi mở logs và thấy hàng trăm dòng: ConnectionError: timeout tiếp theo là 429 Too Many Requests. Nhưng điều tồi tệ nhất là tôi không có bất kỳ metrics nào để phân tích — không biết latency tăng bao nhiêu, không biết error rate ở mức nào, hoàn toàn mù về hành vi của API.

Bài học đắt giá: Không có monitoring, bạn đang điều khiển máy bay không có đồng hồ cao su.

Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống giám sát AI API hoàn chỉnh với Prometheus, tập trung vào bốn chỉ số vàng (Golden Signals): Latency, Traffic, Errors, và Saturation. Tất cả code sẽ sử dụng HolySheep AI làm ví dụ — nền tảng có độ trễ trung bình dưới 50ms và chi phí tiết kiệm đến 85% so với các provider khác.

Tại Sao Cần Monitor AI API?

Khi tích hợp AI API vào production, bạn cần biết:

Với HolySheep AI, bạn có thể theo dõi chi phí chi tiết với tỷ giá ¥1 = $1, hỗ trợ WeChat và Alipay, và nhận tín dụng miễn phí khi đăng ký tài khoản mới.

Cài Đặt Prometheus và Client Library

1. Cài đặt prometheus-client-py

# Cài đặt thư viện Prometheus
pip install prometheus-client>=0.19.0
pip install requests>=2.31.0


Cài đặt Prometheus Server (Docker)

docker pull prom/prometheus:latest

2. Cấu trúc Project

ai-api-monitor/
├── prometheus.yml
├── app.py
├── monitor.py
├── dashboards/
│   └── golden_signals.json
└── requirements.txt

Triển Khai Bốn Chỉ Số Vàng

3. Monitor Core — metrics_collection.py

"""
HolySheep AI API Monitor - Golden Signals Implementation
Monitor: Latency, Traffic, Errors, Saturation
"""

from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry, start_http_server
import requests
import time
import logging
from datetime import datetime


Tạo registry riêng để tránh conflict

registry = CollectorRegistry()


============================================


1. LATENCY METRICS (Histogram)


============================================

REQUEST_LATENCY = Histogram(
    'ai_api_request_latency_seconds',
    'Request latency in seconds',
    ['model', 'endpoint', 'status'],
    buckets=[0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 10.0],
    registry=registry
)


============================================


2. TRAFFIC METRICS (Counter)


============================================

REQUEST_COUNT = Counter(
    'ai_api_requests_total',
    'Total number of requests',
    ['model', 'endpoint', 'status_code'],
    registry=registry
)

TOKEN_USAGE = Counter(
    'ai_api_tokens_total',
    'Total tokens consumed',
    ['model', 'token_type'],  # token_type: prompt/completion
    registry=registry
)


============================================


3. ERROR METRICS (Counter)


============================================

ERROR_COUNT = Counter(
    'ai_api_errors_total',
    'Total number of errors',
    ['model', 'error_type'],  # error_type: timeout/auth/ratelimit/server
    registry=registry
)


============================================


4. SATURATION METRICS (Gauge)


============================================

RATE_LIMIT_REMAINING = Gauge(
    'ai_api_rate_limit_remaining',
    'Remaining rate limit quota',
    ['model'],
    registry=registry
)

API_HEALTH = Gauge(
    'ai_api_health_status',
    'API health status (1=healthy, 0=unhealthy)',
    ['endpoint'],
    registry=registry
)


class HolySheepAIMonitor:
    """Monitor class cho HolySheep AI API"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def chat_completions(self, model: str, messages: list, **kwargs):
        """Gọi chat completions API với metrics collection"""
        start_time = time.time()
        endpoint = "chat/completions"
        status_code = "unknown"
        
        try:
            payload = {
                "model": model,
                "messages": messages,
                **kwargs
            }
            
            response = self.session.post(
                f"{self.BASE_URL}/{endpoint}",
                json=payload,
                timeout=30
            )
            
            # Lấy status code
            status_code = str(response.status_code)
            latency = time.time() - start_time
            
            # Record metrics
            REQUEST_LATENCY.labels(model=model, endpoint=endpoint, status=status_code).observe(latency)
            REQUEST_COUNT.labels(model=model, endpoint=endpoint, status_code=status_code).inc()
            
            # Parse response để đếm tokens
            if response.status_code == 200:
                data = response.json()
                usage = data.get("usage", {})
                
                TOKEN_USAGE.labels(model=model, token_type="prompt").inc(usage.get("prompt_tokens", 0))
                TOKEN_USAGE.labels(model=model, token_type="completion").inc(usage.get("completion_tokens", 0))
                
                # Parse rate limit headers
                remaining = response.headers.get("X-RateLimit-Remaining", 999)
                RATE_LIMIT_REMAINING.labels(model=model).set(int(remaining))
                
                API_HEALTH.labels(endpoint=endpoint).set(1)
                
                return data
            else:
                # Xử lý errors
                self._handle_error(response, model, endpoint)
                return None
                
        except requests.exceptions.Timeout:
            ERROR_COUNT.labels(model=model, error_type="timeout").inc()
            API_HEALTH.labels(endpoint=endpoint).set(0)
            raise TimeoutError(f"Request timeout after 30s for model {model}")
            
        except requests.exceptions.ConnectionError as e:
            ERROR_COUNT.labels(model=model, error_type="connection").inc()
            API_HEALTH.labels(endpoint=endpoint).set(0)
            raise ConnectionError(f"Connection failed: {str(e)}")
            
        except Exception as e:
            ERROR_COUNT.labels(model=model, error_type="unknown").inc()
            raise
    
    def _handle_error(self, response, model, endpoint):
        """Xử lý các loại error từ API"""
        status = response.status_code
        
        if status == 401:
            ERROR_COUNT.labels(model=model, error_type="auth").inc()
            raise PermissionError("Invalid API key - Check your HolySheep AI credentials")
            
        elif status == 429:
            ERROR_COUNT.labels(model=model, error_type="ratelimit").inc()
            retry_after = response.headers.get("Retry-After", 60)
            raise RuntimeError(f"Rate limit exceeded. Retry after {retry_after}s")
            
        elif status == 500:
            ERROR_COUNT.labels(model=model, error_type="server").inc()
            raise RuntimeError("HolySheep AI server error")
            
        else:
            ERROR_COUNT.labels(model=model, error_type="http").inc()
            raise RuntimeError(f"HTTP {status}: {response.text}")



Khởi động Prometheus exporter

def start_exporter(port: int = 9090):
    """Khởi động HTTP server để Prometheus scrape metrics"""
    start_http_server(port, registry=registry)
    print(f"✅ Prometheus exporter started on port {port}")
    print(f"📊 Metrics available at: http://localhost:{port}/metrics")


if __name__ == "__main__":
    start_exporter(9090)
    
    # Demo usage
    monitor = HolySheepAIMonitor("YOUR_HOLYSHEEP_API_KEY")
    
    while True:
        try:
            result = monitor.chat_completions(
                model="gpt-4.1",
                messages=[{"role": "user", "content": "Hello"}],
                max_tokens=50
            )
            print(f"✅ Response received: {result['choices'][0]['message']['content'][:50]}...")
        except Exception as e:
            print(f"❌ Error: {e}")
        
        time.sleep(10)

4. Cấu hình Prometheus — prometheus.yml

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

alerting:
  alertmanagers:
    - static_configs:
        - targets: []

rule_files:
  - "alert_rules.yml"

scrape_configs:
  # AI API Metrics
  - job_name: 'ai-api-monitor'
    static_configs:
      - targets: ['localhost:9090']
    metrics_path: /metrics
    scrape_interval: 10s

  # Prometheus self-monitoring
  - job_name: 'prometheus'
    static_configs:
      - targets: ['localhost:9091']


Alert rules cho Golden Signals


alert_rules.yml

groups:
  - name: ai_api_alerts
    rules:
      # High Latency Alert (> 5s)
      - alert: AIAPILatencyHigh
        expr: histogram_quantile(0.95, rate(ai_api_request_latency_seconds_bucket[5m])) > 5
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "AI API latency above 5 seconds"
          description: "P95 latency is {{ $value }}s for model {{ $labels.model }}"

      # High Error Rate (> 5%)
      - alert: AIAPIErrorRateHigh
        expr: rate(ai_api_errors_total[5m]) / rate(ai_api_requests_total[5m]) > 0.05
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "AI API error rate above 5%"
          description: "Error rate is {{ $value | humanizePercentage }}"

      # Rate Limit Exhaustion
      - alert: AIAPIRateLimitLow
        expr: ai_api_rate_limit_remaining < 10
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "AI API rate limit running low"
          description: "Only {{ $value }} requests remaining for {{ $labels.model }}"

      # API Down
      - alert: AIAPIDown
        expr: ai_api_health_status == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "AI API is down"
          description: "API endpoint {{ $labels.endpoint }} is not responding"

Dashboard Grafana — Visualization

Sau khi cấu hình Prometheus, tạo dashboard trong Grafana với các panel sau:

5. Dashboard JSON — Golden Signals Panels

{
  "dashboard": {
    "title": "AI API Golden Signals Dashboard",
    "panels": [
      {
        "title": "P50/P95/P99 Latency",
        "type": "graph",
        "targets": [
          {
            "expr": "histogram_quantile(0.50, rate(ai_api_request_latency_seconds_bucket[5m]))",
            "legendFormat": "P50"
          },
          {
            "expr": "histogram_quantile(0.95, rate(ai_api_request_latency_seconds_bucket[5m]))",
            "legendFormat": "P95"
          },
          {
            "expr": "histogram_quantile(0.99, rate(ai_api_request_latency_seconds_bucket[5m]))",
            "legendFormat": "P99"
          }
        ],
        "gridPos": {"x": 0, "y": 0, "w": 12, "h": 8}
      },
      {
        "title": "Request Rate (RPM)",
        "type": "graph", 
        "targets": [
          {
            "expr": "rate(ai_api_requests_total[1m]) * 60",
            "legendFormat": "{{model}} - {{status_code}}"
          }
        ],
        "gridPos": {"x": 12, "y": 0, "w": 12, "h": 8}
      },
      {
        "title": "Error Rate by Type",
        "type": "piechart",
        "targets": [
          {
            "expr": "rate(ai_api_errors_total[5m])",
            "legendFormat": "{{error_type}}"
          }
        ],
        "gridPos": {"x": 0, "y": 8, "w": 8, "h": 8}
      },
      {
        "title": "Token Consumption",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(ai_api_tokens_total[1h])",
            "legendFormat": "{{model}} - {{token_type}}"
          }
        ],
        "gridPos": {"x": 8, "y": 8, "w": 8, "h": 8}
      },
      {
        "title": "Rate Limit Remaining",
        "type": "gauge",
        "targets": [
          {
            "expr": "ai_api_rate_limit_remaining",
            "legendFormat": "{{model}}"
          }
        ],
        "gridPos": {"x": 16, "y": 8, "w": 8, "h": 8}
      }
    ]
  }
}

Tính Năng Nâng Cao

6. Cost Tracking Module

"""
Cost Tracking cho AI API
Tính chi phí theo thời gian thực
"""

from prometheus_client import Counter, Gauge


Cost metrics (USD per 1M tokens)

MODEL_COSTS = {
    "gpt-4.1": {"prompt": 8.0, "completion": 8.0},      # $8/MTok
    "claude-sonnet-4.5": {"prompt": 15.0, "completion": 15.0},  # $15/MTok
    "gemini-2.5-flash": {"prompt": 2.50, "completion": 2.50},   # $2.50/MTok
    "deepseek-v3.2": {"prompt": 0.42, "completion": 0.42},     # $0.42/MTok
}

COST_ACCUMULATED = Counter(
    'ai_api_cost_usd_total',
    'Total cost in USD',
    ['model'],
    registry=registry
)

COST_HOURLY = Gauge(
    'ai_api_cost_usd_hourly',
    'Cost in USD for current hour',
    ['model'],
    registry=registry
)


def calculate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    """Tính chi phí request"""
    costs = MODEL_COSTS.get(model, {"prompt": 1.0, "completion": 1.0})
    
    prompt_cost = (prompt_tokens / 1_000_000) * costs["prompt"]
    completion_cost = (completion_tokens / 1_000_000) * costs["completion"]
    
    total_cost = prompt_cost + completion_cost
    
    # Update metrics
    COST_ACCUMULATED.labels(model=model).inc(total_cost)
    
    return total_cost


def estimate_monthly_cost(model: str, monthly_prompt_tokens: int, monthly_completion_tokens: int) -> dict:
    """Ước tính chi phí hàng tháng"""
    costs = MODEL_COSTS.get(model, {"prompt": 1.0, "completion": 1.0})
    
    monthly_cost = (
        (monthly_prompt_tokens / 1_000_000) * costs["prompt"] +
        (monthly_completion_tokens / 1_000_000) * costs["completion"]
    )
    
    return {
        "model": model,
        "prompt_cost": (monthly_prompt_tokens / 1_000_000) * costs["prompt"],
        "completion_cost": (monthly_completion_tokens / 1_000_000) * costs["completion"],
        "total_monthly_usd": round(monthly_cost, 2),
        "total_monthly_cny": round(monthly_cost, 2),  # ¥1 = $1
        "savings_vs_openai": round(monthly_cost * 0.15, 2) if model != "gpt-4.1" else 0  # 85% savings
    }



Demo cost calculation

if __name__ == "__main__":
    # So sánh chi phí giữa các provider
    test_tokens = 1_000_000  # 1M tokens
    
    print("=" * 60)
    print("SO SÁNH CHI PHÍ API (Per Million Tokens)")
    print("=" * 60)
    
    for model, costs in MODEL_COSTS.items():
        avg_cost = (costs["prompt"] + costs["completion"]) / 2
        print(f"{model:25} | ${avg_cost:6.2f}/MTok")
    
    print("-" * 60)
    print(f"{'DeepSeek V3.2 Savings':25} | ~95% so với GPT-4.1")
    print(f"{'HolySheep AI':25} | ¥1 = $1 (Tiết kiệm 85%+)")
    print("=" * 60)

Docker Compose Deployment

# docker-compose.yml
version: '3.8'

services:
  prometheus:
    image: prom/prometheus:latest
    container_name: prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - ./alert_rules.yml:/etc/prometheus/alert_rules.yml
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.path=/prometheus'
    restart: unless-stopped

  grafana:
    image: grafana/grafana:latest
    container_name: grafana
    ports:
      - "3000:3000"
    volumes:
      - grafana_data:/var/lib/grafana
      - ./dashboards:/etc/grafana/provisioning/dashboards
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
      - GF_USERS_ALLOW_SIGN_UP=false
    restart: unless-stopped

  ai-api-monitor:
    build: .
    container_name: ai-monitor
    ports:
      - "9091:9091"
    environment:
      - HOLYSHEEP_API_KEY=${HOLYSHEEP_API_KEY}
    restart: unless-stopped

volumes:
  prometheus_data:
  grafana_data:

Lỗi Thường Gặp và Cách Khắc Phục

1. Lỗi 401 Unauthorized

Mô tả: Khi khởi động monitor, bạn gặp lỗi 401 Unauthorized hoặc Invalid API key.

# ❌ SAI - Key bị trống hoặc sai định dạng
monitor = HolySheepAIMonitor("")


✅ ĐÚNG - Kiểm tra key trong environment

import os
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key

Tài nguyên liên quan
Bài viết liên quan