Grafana + Prometheus วิธีติดตาม AI API Calls แบบมืออาชีพ

ในบทความนี้ผมจะเล่าประสบการณ์ตรงจากการ deploy ระบบ monitor สำหรับ AI API ที่ใช้งานจริงในองค์กร E-commerce ขนาดใหญ่แห่งหนึ่ง ตอนนั้นเราต้องรับมือกับ AI ที่ตอบคำถามลูกค้ากว่า 50,000 คำถามต่อวัน และต้องการ visibility เต็มรูปแบบเรื่อง latency, token usage และ cost

ทำไมต้อง Monitor AI API?

หลายคนอาจคิดว่าแค่เรียก API ได้ก็พอแล้ว แต่ในความเป็นจริงมีเรื่องที่ต้องควบคุม

• Cost Control — AI token มีราคาต่อหน่วย ถ้าไม่ monitor เดือนนึงอาจบิลเกินงบได้ง่ายๆ
• Latency Alert — Response time ที่สูงผิดปกติอาจทำให้ UX แย่ลง
• Usage Pattern — รู้ว่า peak time อยู่ช่วงไหน เพื่อวางแผน scale
• Error Rate — API error ที่ไม่ได้รับการแจ้งเตือนจะกระทบ business โดยไม่รู้ตัว

การติดตั้ง Prometheus + Grafana Stack

เริ่มจากติดตั้ง stack พื้นฐานที่จำเป็น

# docker-compose.yml
version: '3.8'
services:
  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    volumes:
      - grafana_data:/var/lib/grafana
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin

  alertmanager:
    image: prom/alertmanager:latest
    ports:
      - "9093:9093"
    volumes:
      - ./alertmanager.yml:/etc/alertmanager/alertmanager.yml

volumes:
  prometheus_data:
  grafana_data:

Client Library สำหรับ Track Metrics

ผมสร้าง Python client ที่ wrap API call และส่ง metrics ไปยัง Prometheus โดยอัตโนมัติ

# ai_monitor_client.py
import time
import tiktoken
from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry
from openai import OpenAI
from typing import Optional, Dict, Any

สร้าง registry แยกสำหรับ AI metrics
ai_registry = CollectorRegistry()

กำหนด metrics ที่ต้องการ track
REQUEST_COUNT = Counter(
    'ai_api_requests_total',
    'Total AI API requests',
    ['model', 'status'],
    registry=ai_registry
)

REQUEST_LATENCY = Histogram(
    'ai_api_request_duration_seconds',
    'AI API request latency',
    ['model'],
    buckets=[0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0],
    registry=ai_registry
)

TOKEN_USAGE = Counter(
    'ai_api_tokens_used_total',
    'Total tokens used',
    ['model', 'token_type'],
    registry=ai_registry
)

COST_ESTIMATE = Counter(
    'ai_api_cost_usd_total',
    'Estimated cost in USD',
    ['model'],
    registry=ai_registry
)

REQUEST_IN_PROGRESS = Gauge(
    'ai_api_requests_in_progress',
    'Requests currently being processed',
    ['model'],
    registry=ai_registry
)

กำหนด pricing ต่อ million tokens (2026)
MODEL_PRICING = {
    'gpt-4.1': {'input': 2.0, 'output': 8.0},        # $2/$8 per MTok
    'claude-sonnet-4.5': {'input': 3.0, 'output': 15.0},  # $3/$15 per MTok
    'gemini-2.5-flash': {'input': 0.35, 'output': 2.50},  # $0.35/$2.50 per MTok
    'deepseek-v3.2': {'input': 0.07, 'output': 0.42},    # $0.07/$0.42 per MTok
}

class AIMonitorClient:
    def __init__(self, api_key: str, model: str = 'deepseek-v3.2'):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"  # HolySheep AI endpoint
        )
        self.model = model
        self.encoding = tiktoken.encoding_for_model("gpt-4")
    
    def count_tokens(self, text: str) -> int:
        """นับจำนวน tokens ในข้อความ"""
        return len(self.encoding.encode(text))
    
    def calculate_cost(self, prompt_tokens: int, completion_tokens: int) -> float:
        """คำนวณค่าใช้จ่ายเป็น USD"""
        pricing = MODEL_PRICING.get(self.model, MODEL_PRICING['deepseek-v3.2'])
        input_cost = (prompt_tokens / 1_000_000) * pricing['input']
        output_cost = (completion_tokens / 1_000_000) * pricing['output']
        return input_cost + output_cost
    
    def chat(self, messages: list, **kwargs) -> Dict[str, Any]:
        """เรียก API พร้อม track metrics"""
        start_time = time.time()
        REQUEST_IN_PROGRESS.labels(model=self.model).inc()
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                **kwargs
            )
            
            # คำนวณ tokens และ cost
            prompt_tokens = self.count_tokens(
                messages[0]['content'] if messages else ''
            )
            completion_tokens = self.count_tokens(response.choices[0].message.content)
            cost = self.calculate_cost(prompt_tokens, completion_tokens)
            
            # อัพเดท metrics
            REQUEST_COUNT.labels(model=self.model, status='success').inc()
            REQUEST_LATENCY.labels(model=self.model).observe(time.time() - start_time)
            TOKEN_USAGE.labels(model=self.model, token_type='prompt').inc(prompt_tokens)
            TOKEN_USAGE.labels(model=self.model, token_type='completion').inc(completion_tokens)
            COST_ESTIMATE.labels(model=self.model).inc(cost)
            
            return {
                'content': response.choices[0].message.content,
                'usage': response.usage.dict() if response.usage else {},
                'cost_usd': round(cost, 6),
                'latency_ms': round((time.time() - start_time) * 1000, 2)
            }
            
        except Exception as e:
            REQUEST_COUNT.labels(model=self.model, status='error').inc()
            raise
        finally:
            REQUEST_IN_PROGRESS.labels(model=self.model).dec()

Configuration สำหรับ Prometheus

กำหนด scrape config เพื่อดึง metrics จาก application

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

alerting:
  alertmanagers:
    - static_configs:
        - targets:
          - alertmanager:9093

rule_files:
  - "alert_rules.yml"

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

  # AI API Application
  - job_name: 'ai-api-app'
    static_configs:
      - targets: ['ai-app:8000']
    metrics_path: '/metrics'
    scrape_interval: 10s

alert_rules.yml
groups:
  - name: ai_api_alerts
    rules:
      - alert: HighLatency
        expr: histogram_quantile(0.95, rate(ai_api_request_duration_seconds_bucket[5m])) > 2
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "AI API latency สูงกว่า 2 วินาที (P95)"
          
      - alert: HighErrorRate
        expr: rate(ai_api_requests_total{status="error"}[5m]) / rate(ai_api_requests_total[5m]) > 0.05
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "AI API error rate เกิน 5%"
          
      - alert: BudgetOverspend
        expr: increase(ai_api_cost_usd_total[1h]) > 100
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "ค่าใช้จ่าย AI API เกิน $100/ชั่วโมง"
          
      - alert: TokenSpike
        expr: rate(ai_api_tokens_used_total[15m]) > 1000000
        for: 10m
        labels:
          severity: info
        annotations:
          summary: "Token usage สูงผิดปกติ"

FastAPI Integration พร้อม Metrics Endpoint

สร้าง FastAPI app ที่ expose metrics endpoint สำหรับ Prometheus

# main.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import Response
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
from ai_monitor_client import AIMonitorClient
import os

app = FastAPI(title="AI API Monitor")

Initialize client - ดึง key จาก environment
ai_client = AIMonitorClient(
    api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
    model="deepseek-v3.2"  # โมเดลที่ประหยัดที่สุด
)

@app.get("/metrics")
async def metrics():
    """Prometheus metrics endpoint"""
    return Response(
        content=generate_latest(ai_client.client._registry if hasattr(ai_client.client, '_registry') else None),
        media_type=CONTENT_TYPE_LATEST
    )

@app.post("/chat")
async def chat(message: dict):
    """AI Chat endpoint with monitoring"""
    try:
        result = ai_client.chat(
            messages=[{"role": "user", "content": message.get("content")}]
        )
        return {
            "success": True,
            "data": result,
            "model": ai_client.model
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "healthy", "service": "ai-monitor"}

รันด้วย: uvicorn main:app --host 0.0.0.0 --port 8000

Dashboard ตัวอย่างสำหรับ Grafana

Import JSON dashboard นี้เข้า Grafana เพื่อดู overview ที่ครอบคลุม

{
  "dashboard": {
    "title": "AI API Monitoring Dashboard",
    "panels": [
      {
        "title": "Request Rate (req/s)",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(ai_api_requests_total[1m])",
            "legendFormat": "{{model}} - {{status}}"
          }
        ]
      },
      {
        "title": "P95 Latency (ms)",
        "type": "graph", 
        "targets": [
          {
            "expr": "histogram_quantile(0.95, rate(ai_api_request_duration_seconds_bucket[5m])) * 1000",
            "legendFormat": "{{model}}"
          }
        ]
      },
      {
        "title": "Token Usage Today",
        "type": "stat",
        "targets": [
          {
            "expr": "sum(increase(ai_api_tokens_used_total[24h])) by (token_type)",
            "legendFormat": "{{token_type}}"
          }
        ]
      },
      {
        "title": "Estimated Cost Today",
        "type": "gauge",
        "targets": [
          {
            "expr": "sum(increase(ai_api_cost_usd_total[24h]))"
          }
        ],
        "fieldConfig": {
          "defaults": {
            "unit": "currencyUSD"
          }
        }
      },
      {
        "title": "Error Rate %",
        "type": "gauge",
        "targets": [
          {
            "expr": "100 * rate(ai_api_requests_total{status='error'}[5m]) / rate(ai_api_requests_total[5m])"
          }
        ]
      }
    ]
  }
}

Use Case: ระบบ RAG ขององค์กรขนาดใหญ่

กรณีศึกษานี้มาจากองค์กรที่ deploy ระบบ RAG (Retrieval-Augmented Generation) สำหรับค้นหาเอกสารภายใน ปัญหาหลักคือ:

• Latency ไม่คงที่ — บางครั้ง 3 วินาที บางครั้ง 15 วินาที
• Cost ไม่คาดคิด — เดือนแรกบิลเกินงบไป 300%
• ไม่รู้ว่า query ไหนกิน token เยอะ

หลังจากติดตั้ง monitoring stack นี้ พวกเขาค้นพบว่า:

# ก่อน optimize
avg_latency: 4,500ms
avg_tokens_per_query: 8,200
daily_cost: $847

หลัง optimize (เปลี่ยน chunk size และใช้ reranking)
avg_latency: 1,200ms  
avg_tokens_per_query: 3,100
daily_cost: $156

ประหยัดได้: 82%

ทำไมต้อง HolySheep AI?

จากประสบการณ์ที่ใช้งานหลาย provider พบว่า HolySheep AI ให้ความคุ้มค่าที่สุดสำหรับ use case นี้

• ราคาถูกกว่า 85% — อัตราแลกเปลี่ยน ¥1=$1 ทำให้ค่าใช้จ่ายต่ำมาก
• Latency ต่ำกว่า 50ms — เหมาะสำหรับ real-time application
• รองรับหลายโมเดล — DeepSeek V3.2 ราคาเพียง $0.42/MTok output
• ชำระเงินง่าย — รองรับ WeChat และ Alipay

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. Metrics ไม่ถูก scrape โดย Prometheus

# ปัญหา: Prometheus ไม่เห็น metrics endpoint
สาเหตุ: ตรวจสอบ network ระหว่าง Prometheus และ app

แก้ไข: เพิ่ม network ใน docker-compose
services:
  prometheus:
    networks:
      - ai-network
  ai-api-app:
    networks:
      - ai-network

networks:
  ai-network:
    driver: bridge

2. Token Count ไม่แม่นยำ

# ปัญหา: ใช้ tiktoken แต่โมเดลไม่ตรงกับ tokenizer
สาเหตุ: tiktoken ใช้ tokenizer ของ GPT-4 เสมอ

แก้ไข: ใช้ tokenizer ที่ตรงกับโมเดล
from anthropic import Anthropic

def count_tokens_anthropic(text: str) -> int:
    """ใช้ Claude tokenizer สำหรับ Claude models"""
    anthropic = Anthropic()
    return len(anthropic.count_tokens(text))

หรือใช้ approximate formula ที่ general
def approximate_tokens(text: str) -> int:
    """Approximate: 1 token ≈ 4 characters สำหรับ Thai/English mixed"""
    return len(text) // 4

3. Cost Calculation ผิดเพราะโมเดลไม่ตรง

# ปัญหา: cost ค
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
RAG 生产部署清单：从 PoC 到上线的 20 项检查
สร้างระบบคัดกรองเรซูเม HR ด้วย AI API สำหรับนักพัฒนาไทย
AI Video Generation API 2026: คู่มือเปลี่ยนจาก Sora 2 / Veo