ในฐานะ DevOps Engineer ที่ดูแลระบบ AI Service มากว่า 3 ปี ผมเชื่อว่าหลายคนคงประสบปัญหาเดียวกัน — ต้องการ Monitor AI API Calls ทั้ง Request/Response แต่ไม่รู้จะเริ่มจากตรงไหน โดยเฉพาะเมื่อใช้ HolySheep AI ที่มีโมเดลหลากหลายและราคาประหยัดกว่า 85%
ทำไมต้องวิเคราะห์ Log AI API ด้วย ELK?
- Debug ง่าย — ดู Response Time, Error Rate แต่ละ Request
- Cost Tracking — คำนวณค่าใช้จ่ายตาม Token ที่ใช้จริง
- Performance Optimization — ระบุ Latency ที่สูงผิดปกติ
- Alerting — แจ้งเตือนเมื่อ Error Rate เกิน 5% หรือ Latency เกิน 500ms
สถาปัตยกรรมระบบที่ใช้จริง
┌─────────────────┐ ┌──────────────┐ ┌───────────────┐ ┌─────────────┐
│ Application │───▶│ Filebeat │───▶│ Elasticsearch │───▶│ Kibana │
│ (AI API Logs) │ │ (Collector) │ │ (Storage) │ │ (Dashboard) │
└─────────────────┘ └──────────────┘ └───────────────┘ └─────────────┘
│ ▲
│ ┌──────────────┐ │
└─────────────▶│ Logstash │◀──────────────┘
│ (Processing) │
└──────────────┘
การสร้าง Python Client สำหรับ Log ทุก API Call
จากประสบการณ์ ผมแนะนำให้สร้าง Wrapper Client ที่ทำ Log อัตโนมัติทุกครั้งที่เรียก API วิธีนี้ทำให้ควบคุมได้ทั้ง Log Format และ Volume
import json
import time
import logging
from datetime import datetime
from typing import Optional, Dict, Any
from elasticsearch import Elasticsearch
Logging Configuration
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("ai_api_monitor")
class AILogger:
"""AI API Logger with ELK Integration"""
def __init__(
self,
es_host: str = "http://localhost:9200",
index_prefix: str = "ai-api-logs",
base_url: str = "https://api.holysheep.ai/v1",
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
):
self.base_url = base_url
self.api_key = api_key
self.es = Elasticsearch([es_host])
self.index_prefix = index_prefix
# สร้าง Index Template อัตโนมัติ
self._create_index_template()
def _create_index_template(self):
"""สร้าง Index Template สำหรับ AI Logs"""
template = {
"index_patterns": [f"{self.index_prefix}-*"],
"template": {
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0,
"index.lifecycle.name": "ai-logs-policy"
},
"mappings": {
"properties": {
"timestamp": {"type": "date"},
"model": {"type": "keyword"},
"provider": {"type": "keyword"},
"latency_ms": {"type": "float"},
"prompt_tokens": {"type": "integer"},
"completion_tokens": {"type": "integer"},
"total_tokens": {"type": "integer"},
"cost_usd": {"type": "float"},
"status_code": {"type": "integer"},
"error": {"type": "text"},
"success": {"type": "boolean"},
"request_id": {"type": "keyword"}
}
}
}
}
try:
self.es.indices.put_index_template(
name="ai-api-template",
body=template
)
logger.info("Index template created successfully")
except Exception as e:
logger.warning(f"Template creation failed: {e}")
def _get_index_name(self) -> str:
"""สร้างชื่อ Index ตามวันที่"""
return f"{self.index_prefix}-{datetime.now().strftime('%Y.%m.%d')}"
def log_request(
self,
model: str,
prompt_tokens: int,
completion_tokens: int,
latency_ms: float,
status_code: int,
error: Optional[str] = None,
request_id: Optional[str] = None
) -> Dict[str, Any]:
"""บันทึก API Request ไปยัง Elasticsearch"""
# คำนวณค่าใช้จ่ายตามโมเดล
cost_per_mtok = self._get_model_cost(model)
total_tokens = prompt_tokens + completion_tokens
cost_usd = (total_tokens / 1_000_000) * cost_per_mtok
document = {
"timestamp": datetime.utcnow().isoformat(),
"model": model,
"provider": "holysheep",
"latency_ms": latency_ms,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": total_tokens,
"cost_usd": round(cost_usd, 6),
"status_code": status_code,
"error": error,
"success": status_code == 200,
"request_id": request_id or f"req_{int(time.time() * 1000)}"
}
# บันทึกลง Elasticsearch
self.es.index(
index=self._get_index_name(),
document=document
)
return document
def _get_model_cost(self, model: str) -> float:
"""ราคาต่อ Million Tokens (USD)"""
costs = {
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42,
# เพิ่มโมเดลอื่นตามต้องการ
}
return costs.get(model.lower(), 1.00)
ตัวอย่างการใช้งาน
ai_logger = AILogger(
es_host="http://localhost:9200",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
การสร้าง AI API Client พร้อม Auto-Logging
import requests
import json
import time
from typing import Dict, Any, Optional
class HolySheepAIClient:
"""HolySheep AI Client with Integrated ELK Logging"""
def __init__(
self,
api_key: str = "YOUR_HOLYSHEEP_API_KEY",
base_url: str = "https://api.holysheep.ai/v1",
es_host: str = "http://localhost:9200",
auto_log: bool = True
):
self.api_key = api_key
self.base_url = base_url
self.auto_log = auto_log
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
# Lazy import เพื่อไม่ต้องติดตั้ง elasticsearch ถ้าไม่ต้องการ
if auto_log:
from elasticsearch import Elasticsearch
self.es = Elasticsearch([es_host])
def chat_completions(
self,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: Optional[int] = None
) -> Dict[str, Any]:
"""เรียก Chat Completions API พร้อมบันทึก Log"""
start_time = time.time()
request_id = f"req_{int(start_time * 1000)}"
payload = {
"model": model,
"messages": messages,
"temperature": temperature
}
if max_tokens:
payload["max_tokens"] = max_tokens
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=30
)
latency_ms = (time.time() - start_time) * 1000
result = response.json()
# ดึง Token Usage จาก Response
usage = result.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
if self.auto_log:
self._log_to_elasticsearch(
model=model,
latency_ms=latency_ms,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
status_code=response.status_code,
request_id=request_id
)
return {
"success": True,
"data": result,
"latency_ms": round(latency_ms, 2),
"tokens": {
"prompt": prompt_tokens,
"completion": completion_tokens,
"total": prompt_tokens + completion_tokens
},
"cost_usd": self._calculate_cost(
model, prompt_tokens, completion_tokens
)
}
except requests.exceptions.Timeout:
return self._handle_error(
model, request_id, "Request Timeout (>30s)", start_time
)
except requests.exceptions.RequestException as e:
return self._handle_error(
model, request_id, str(e), start_time
)
def _log_to_elasticsearch(
self,
model: str,
latency_ms: float,
prompt_tokens: int,
completion_tokens: int,
status_code: int,
request_id: str
):
"""บันทึก Log ไปยัง Elasticsearch"""
try:
document = {
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime()),
"model": model,
"provider": "holysheep",
"latency_ms": round(latency_ms, 2),
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens,
"cost_usd": self._calculate_cost(
model, prompt_tokens, completion_tokens
),
"status_code": status_code,
"success": status_code == 200,
"request_id": request_id
}
index_name = f"ai-api-{time.strftime('%Y.%m.%d')}"
self.es.index(index=index_name, document=document)
except Exception as e:
print(f"Logging failed: {e}")
def _calculate_cost(
self, model: str, prompt_tokens: int, completion_tokens: int
) -> float:
"""คำนวณค่าใช้จ่าย USD"""
costs = {
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
rate = costs.get(model, 1.00)
total_tokens = prompt_tokens + completion_tokens
return round((total_tokens / 1_000_000) * rate, 6)
def _handle_error(
self, model: str, request_id: str, error: str, start_time: float
) -> Dict[str, Any]:
"""จัดการ Error Case"""
latency_ms = (time.time() - start_time) * 1000
if self.auto_log:
self._log_to_elasticsearch(
model=model,
latency_ms=latency_ms,
prompt_tokens=0,
completion_tokens=0,
status_code=500,
request_id=request_id
)
return {
"success": False,
"error": error,
"latency_ms": round(latency_ms, 2),
"request_id": request_id
}
ตัวอย่างการใช้งาน
if __name__ == "__main__":
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# ทดสอบเรียก API
response = client.chat_completions(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello!"}
]
)
print(json.dumps(response, indent=2, ensure_ascii=False))
การตั้งค่า Elasticsearch Index Lifecycle Management
สร้าง ILM Policy สำหรับ AI Logs
PUT _ilm/policy/ai-logs-policy
{
"policy": {
"phases": {
"hot": {
"min_age": "0ms",
"actions": {
"rollover": {
"max_age": "1d",
"max_primary_shard_size": "50gb"
},
"set_priority": {
"priority": 100
}
}
},
"warm": {
"min_age": "7d",
"actions": {
"shrink": {
"number_of_shards": 1
},
"forcemerge": {
"max_num_segments": 1
},
"set_priority": {
"priority": 50
}
}
},
"delete": {
"min_age": "30d",
"actions": {
"delete": {}
}
}
}
}
}
สร้าง Index Template พร้อม ILM
PUT _index_template/ai-api-template
{
"index_patterns": ["ai-api-*"],
"template": {
"settings": {
"number_of_shards": 1,
"number_of_replicas": 1,
"index.lifecycle.name": "ai-logs-policy",
"index.lifecycle.rollover_alias": "ai-api-logs"
},
"mappings": {
"properties": {
"timestamp": {
"type": "date"
},
"model": {
"type": "keyword"
},
"provider": {
"type": "keyword"
},
"latency_ms": {
"type": "float"
},
"prompt_tokens": {
"type": "integer"
},
"completion_tokens": {
"type": "integer"
},
"total_tokens": {
"type": "integer"
},
"cost_usd": {
"type": "float"
},
"status_code": {
"type": "integer"
},
"success": {
"type": "boolean"
},
"request_id": {
"type": "keyword"
},
"error": {
"type": "text"
}
}
}
}
}
Kibana Dashboard สำหรับ AI API Monitoring
จากการใช้งานจริง ผมแนะนำ Dashboard ที่ครอบคลุม Metrics สำคัญดังนี้
- Overview Panel — Total Requests, Success Rate, Avg Latency, Total Cost
- Latency Distribution — Histogram แสดง Latency รายนาที
- Token Usage by Model — Pie Chart แสดงสัดส่วน Token ตามโมเดล
- Error Rate Timeline — Line Chart แสดง Error Rate ตามเวลา
- Cost by Day — Bar Chart แสดงค่าใช้จ่ายรายวัน
การตั้งค่า Alerting Rules
Alert: Error Rate > 5%
POST _Watcher/watch/error-rate-alert
{
"trigger": {
"schedule": {
"interval": "5m"
}
},
"input": {
"search": {
"request": {
"indices": ["ai-api-*"],
"body": {
"size": 0,
"query": {
"range": {
"timestamp": {
"gte": "now-5m"
}
}
},
"aggs": {
"total": {
"value_count": {
"field": "request_id"
}
},
"errors": {
"filter": {
"term": {
"success": false
}
}
}
}
}
}
}
},
"condition": {
"script": {
"source": "return ctx.payload.aggs.errors.doc_count > 0 && (ctx.payload.aggs.errors.doc_count / ctx.payload.aggs.total.value) > 0.05",
"lang": "painless"
}
},
"actions": {
"log_error": {
"logging": {
"text": "AI API Error Rate Alert: {{ctx.payload.aggs.errors.doc_count}} errors in last 5 minutes"
}
},
"webhook_alert": {
"webhook": {
"scheme": "https",
"host": "hooks.slack.com",
"port": 443,
"method": "post",
"path": "/services/xxx",
"body": "{\"text\": \"AI API Error Rate Alert: {{ctx.payload.aggs.errors.doc_count}} errors (>5%)\"}"
}
}
}
}
Alert: High Latency > 500ms
POST _Watcher/watch/high-latency-alert
{
"trigger": {
"schedule": {
"interval": "5m"
}
},
"input": {
"search": {
"request": {
"indices": ["ai-api-*"],
"body