บทนำ: วันที่ระบบล่มเพราะไม่มี Log
คืนหนึ่งตอนตี 3 เว็บไซต์ AI ของผมล่ม ผู้ใช้งานติดต่อเข้ามาเต็ม พอเช็ค log กลับพบว่าไฟล์บันทึกถูกลบไปเองเพราะดิสก์เต็ม พอพยายามหาสาเหตุว่า API ตอบกลับอะไรมาก่อนหน้านั้น ก็หายไปหมด ต้องนั่งไล่ดูโค้ดเก่าๆ เพื่อ Replay request อีกครั้ง ใช้เวลาแก้ปัญหา 4 ชั่วโมงแทนที่จะแก้ได้ใน 10 นาทีถ้ามี log ที่ดี
บทความนี้จะสอนวิธีสร้างระบบบันทึกคำขอและการตอบกลับของ AI API ตั้งแต่เริ่มต้น ครอบคลุมทั้ง PostgreSQL, MongoDB, Elasticsearch และวิธีที่เหมาะกับงบประมาณต่างๆ
ทำไมต้องจัดเก็บ AI API Log
ก่อนจะลงมือทำ มาดูเหตุผลที่ระบบบันทึก log สำคัญมาก
- แก้ปัญหาข้อผิดพลาดได้รวดเร็ว — เช่น เมื่อได้รับ
401 Unauthorized หรือ ConnectionError: timeout จะดู log ย้อนหลังได้ทันที
- วิเคราะห์พฤติกรรมผู้ใช้ — ดูว่า prompt ประเภทไหนใช้บ่อย ผู้ใช้ถามเรื่องอะไรมากที่สุด
- ตรวจสอบค่าใช้จ่าย — รู้ว่า token ใช้ไปเท่าไหร่ แต่ละ request มี cost เท่าไร
- เป็นหลักฐานทางธุรกิจ — ถ้าระบบตอบผิดพลาด มี log ยืนยันว่าส่งอะไรไป ได้รับอะไรกลับมา
- ปฏิบัติตามกฎหมาย — ข้อมูลบางอย่างต้องเก็บ audit trail
โซลูชันการจัดเก็บ Log ที่นิยมใช้
การจัดเก็บ AI API log มีหลายวิธี แต่ละวิธีมีข้อดีข้อเสียต่างกัน
1. วิธีบันทึกลงไฟล์ (File-based Logging)
วิธีที่ง่ายที่สุด เหมาะกับโปรเจกต์เล็กหรือต้องการทดสอบอย่างรวดเร็ว
import json
import logging
from datetime import datetime
from pathlib import Path
ตั้งค่า logging
log_dir = Path("logs/api_logs")
log_dir.mkdir(parents=True, exist_ok=True)
สร้าง logger ที่บันทึกลงไฟล์
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(log_dir / f"api_{datetime.now().strftime('%Y%m%d')}.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger("AI_API_Logger")
def log_request(endpoint, request_data, response_data, status_code, latency_ms):
"""บันทึกคำขอและการตอบกลับทุกครั้ง"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"endpoint": endpoint,
"request": request_data,
"response": response_data,
"status_code": status_code,
"latency_ms": latency_ms
}
logger.info(json.dumps(log_entry, ensure_ascii=False, indent=2))
ตัวอย่างการใช้งาน
log_request(
endpoint="https://api.holysheep.ai/v1/chat/completions",
request_data={"model": "gpt-4.1", "messages": [{"role": "user", "content": "สวัสดี"}]},
response_data={"choices": [{"message": {"content": "สวัสดีครับ"}}]},
status_code=200,
latency_ms=245
)
2. วิธีบันทึกลง PostgreSQL
เหมาะกับระบบที่ต้องการ query ข้อมูลได้ยืดหยุ่น และมีความสัมพันธ์ระหว่างข้อมูล
import psycopg2
from psycopg2.extras import Json
from datetime import datetime
import os
class APILogDatabase:
def __init__(self):
self.conn = psycopg2.connect(
host=os.getenv("DB_HOST", "localhost"),
port=os.getenv("DB_PORT", "5432"),
database=os.getenv("DB_NAME", "ai_logs"),
user=os.getenv("DB_USER", "postgres"),
password=os.getenv("DB_PASSWORD", "")
)
self.create_table()
def create_table(self):
"""สร้างตารางสำหรับเก็บ log"""
with self.conn.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS api_request_logs (
id SERIAL PRIMARY KEY,
request_id UUID DEFAULT gen_random_uuid(),
endpoint VARCHAR(500) NOT NULL,
model VARCHAR(100),
request_payload JSONB NOT NULL,
response_payload JSONB,
status_code INTEGER,
latency_ms FLOAT,
token_usage JSONB,
error_message TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_logs_created_at ON api_request_logs(created_at);
CREATE INDEX IF NOT EXISTS idx_logs_endpoint ON api_request_logs(endpoint);
""")
self.conn.commit()
def save_log(self, endpoint, model, request_payload, response_payload,
status_code, latency_ms, token_usage=None, error_message=None):
"""บันทึก log ลงฐานข้อมูล"""
with self.conn.cursor() as cur:
cur.execute("""
INSERT INTO api_request_logs
(endpoint, model, request_payload, response_payload,
status_code, latency_ms, token_usage, error_message)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
RETURNING request_id;
""", (
endpoint, model, Json(request_payload), Json(response_payload),
status_code, latency_ms, Json(token_usage) if token_usage else None,
error_message
))
result = cur.fetchone()
self.conn.commit()
return result[0] if result else None
ตัวอย่างการใช้งาน
db = APILogDatabase()
log_id = db.save_log(
endpoint="https://api.holysheep.ai/v1/chat/completions",
model="gpt-4.1",
request_payload={"messages": [{"role": "user", "content": "อธิบาย AI"}]},
response_payload={"choices": [{"message": {"content": "AI คือ..."}}]},
status_code=200,
latency_ms=180,
token_usage={"prompt_tokens": 15, "completion_tokens": 120, "total_tokens": 135}
)
3. วิธีบันทึกลง Elasticsearch
เหมาะกับระบบใหญ่ที่ต้องรองรับ log ปริมาณมาก และต้องการค้นหาแบบ Full-text search
from elasticsearch import Elasticsearch
from datetime import datetime
import json
class ElasticsearchLogger:
def __init__(self, hosts=None):
self.es = Elasticsearch(hosts or ["http://localhost:9200"])
self.index_prefix = "ai-api-logs"
def create_index_template(self):
"""สร้าง index template พร้อม mapping ที่เหมาะสม"""
template = {
"index_patterns": [f"{self.index_prefix}-*"],
"template": {
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0
},
"mappings": {
"properties": {
"timestamp": {"type": "date"},
"request_id": {"type": "keyword"},
"endpoint": {"type": "keyword"},
"model": {"type": "keyword"},
"prompt_tokens": {"type": "integer"},
"completion_tokens": {"type": "integer"},
"total_tokens": {"type": "integer"},
"latency_ms": {"type": "float"},
"status_code": {"type": "integer"},
"user_query": {"type": "text", "analyzer": "thai"},
"ai_response": {"type": "text", "analyzer": "thai"},
"error": {"type": "text"},
"metadata": {"type": "object", "enabled": False}
}
}
}
}
self.es.indices.put_index_template(name="ai-logs-template", body=template)
def log_request(self, endpoint, model, user_query, ai_response,
token_usage, latency_ms, status_code, error=None, metadata=None):
"""บันทึก log ลง Elasticsearch"""
import uuid
doc = {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid.uuid4()),
"endpoint": endpoint,
"model": model,
"user_query": user_query,
"ai_response": ai_response,
"prompt_tokens": token_usage.get("prompt_tokens") if token_usage else 0,
"completion_tokens": token_usage.get("completion_tokens") if token_usage else 0,
"total_tokens": token_usage.get("total_tokens") if token_usage else 0,
"latency_ms": latency_ms,
"status_code": status_code,
"error": str(error) if error else None,
"metadata": metadata
}
index_name = f"{self.index_prefix}-{datetime.now().strftime('%Y.%m')}"
return self.es.index(index=index_name, document=doc)
def search_logs(self, query=None, model=None, start_date=None, end_date=None, size=100):
"""ค้นหา log ตามเงื่อนไข"""
must_conditions = []
if query:
must_conditions.append({
"multi_match": {
"query": query,
"fields": ["user_query", "ai_response"]
}
})
if model:
must_conditions.append({"term": {"model": model}})
if start_date or end_date:
date_range = {}
if start_date:
date_range["gte"] = start_date
if end_date:
date_range["lte"] = end_date
must_conditions.append({"range": {"timestamp": date_range}})
body = {
"query": {
"bool": {
"must": must_conditions if must_conditions else [{"match_all": {}}]
}
},
"sort": [{"timestamp": {"order": "desc"}}],
"size": size
}
return self.es.search(index=f"{self.index_prefix}-*", body=body)
ตัวอย่างการใช้งาน
es_logger = ElasticsearchLogger(["http://localhost:9200"])
es_logger.create_index_template()
result = es_logger.log_request(
endpoint="https://api.holysheep.ai/v1/chat/completions",
model="deepseek-v3.2",
user_query="สอนเขียน Python หน่อย",
ai_response="Python เป็นภาษาเขียนโปรแกรมที่...",
token_usage={"prompt_tokens": 25, "completion_tokens": 200, "total_tokens": 225},
latency_ms=320,
status_code=200
)
การสร้าง Middleware สำหรับ Log อัตโนมัติ
เมื่อใช้ HolySheep API โดยตรง สามารถสร้าง wrapper ที่จับ request/response อัตโนมัติ
import requests
import time
import json
from datetime import datetime
from typing import Dict, Any, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class HolySheepAPIClient:
"""Client สำหรับ HolySheep API พร้อมระบบบันทึก log"""
def __init__(self, api_key: str, log_handler=None):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.log_handler = log_handler
def _log_request(self, endpoint: str, request_data: Dict,
response_data: Optional[Dict], status_code: int,
latency_ms: float, error: Optional[str] = None):
"""บันทึก request/response อัตโนมัติ"""
if self.log_handler:
self.log_handler({
"timestamp": datetime.now().isoformat(),
"endpoint": endpoint,
"model": request_data.get("model", "unknown"),
"request": request_data,
"response": response_data,
"status_code": status_code,
"latency_ms": latency_ms,
"error": error
})
def chat_completions(self, messages: list, model: str = "gpt-4.1",
temperature: float = 0.7, max_tokens: int = 1000) -> Dict:
"""เรียกใช้ Chat Completions API พร้อมบันทึก log"""
start_time = time.time()
endpoint = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
response = requests.post(endpoint, headers=headers, json=payload, timeout=30)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
result = response.json()
self._log_request(endpoint, payload, result, response.status_code, latency_ms)
return result
else:
error_data = response.json() if response.content else {}
self._log_request(endpoint, payload, error_data, response.status_code, latency_ms,
error=f"HTTP {response.status_code}")
raise Exception(f"API Error: {response.status_code} - {error_data}")
except requests.exceptions.Timeout:
latency_ms = (time.time() - start_time) * 1000
self._log_request(endpoint, payload, None, 408, latency_ms,
error="ConnectionError: timeout")
raise Exception("Request timeout - การเชื่อมต่อใช้เวลานานเกินไป")
except requests.exceptions.ConnectionError as e:
latency_ms = (time.time() - start_time) * 1000
self._log_request(endpoint, payload, None, 503, latency_ms,
error=f"ConnectionError: {str(e)}")
raise Exception("ConnectionError - ไม่สามารถเชื่อมต่อกับ API")
ตัวอย่างการใช้งาน
def simple_log_handler(log_entry):
"""Handler สำหรับบันทึก log แบบง่าย"""
logger.info(f"[API LOG] {json.dumps(log_entry, ensure_ascii=False)}")
สร้าง client
client = HolySheepAPIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
log_handler=simple_log_handler
)
เรียกใช้งาน - log จะถูกบันทึกอัตโนมัติ
response = client.chat_completions(
messages=[{"role": "user", "content": "ทำไมฟ้าถึงเป็นสีฟ้า?"}],
model="deepseek-v3.2",
max_tokens=500
)
การจัดการข้อมูลที่ละเอียดอ่อน
เมื่อบันทึก log ต้องระวังเรื่องข้อมูลส่วนตัวด้วย
import re
import hashlib
from typing import Dict, Any
class PIIRedactor:
"""ระบบซ่อนข้อมูลส่วนตัวใน log"""
PATTERNS = {
"email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
"phone": r'\b\d{10,11}\b',
"credit_card": r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',
"id_card": r'\b\d{13}\b'
}
@classmethod
def redact(cls, text: str) -> str:
"""ซ่อนข้อมูลส่วนตัวในข้อความ"""
if not text:
return text
result = text
for pii_type, pattern in cls.PATTERNS.items():
result = re.sub(pattern, f"[REDACTED_{pii_type.upper()}]", result)
return result
@classmethod
def hash_user_id(cls, user_id: str) -> str:
"""แปลง user ID เป็น hash เพื่อไม่ต้องเก็บข้อมูลจริง"""
return hashlib.sha256(str(user_id).encode()).hexdigest()[:16]
@classmethod
def sanitize_log(cls, log_entry: Dict[str, Any]) -> Dict[str, Any]:
"""ทำความสะอาด log entry ก่อนบันทึก"""
sanitized = log_entry.copy()
# แปลง user ID
if "user_id" in sanitized:
sanitized["user_id_hash"] = cls.hash_user_id(sanitized.pop("user_id"))
# ซ่อนข้อมูลใน request/response
if "request" in sanitized and isinstance(sanitized["request"], dict):
for key in ["content", "prompt", "query"]:
if key in sanitized["request"]:
sanitized["request"][key] = cls.redact(str(sanitized["request"][key]))
if "response" in sanitized and isinstance(sanitized["response"], dict):
for key in ["content", "response", "answer"]:
if key in sanitized["response"]:
sanitized["response"][key] = cls.redact(str(sanitized["response"][key]))
return sanitized
ตัวอย่างการใช้งาน
raw_log = {
"user_id": "user_123456789",
"request": {
"model": "gpt-4.1",
"content": "ส่งข้อมูลบัตรประชาชน 1234567890123 ไปที่ [email protected]"
},
"response": {
"content": "ได้รับข้อมูลแล้ว ตอบกลับไปที่ [email protected]"
}
}
sanitized = PIIRedactor.sanitize_log(raw_log)
print(f"User ID ถูกแปลงเป็น: {sanitized['user_id_hash']}")
print(f"Request หลัง sanitize: {sanitized['request']['content']}")
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
ในการใช้งาน AI API และระบบบันทึก log มีข้อผิดพลาดที่พบบ่อยดังนี้
1. 401 Unauthorized - API Key ไม่ถูกต้อง
# ปัญหา: ได้รับข้อผิดพลาด 401 Unauthorized
สาเหตุที่พบบ่อย:
1. API Key หมดอายุหรือถูก revoke
2. ใส่ API Key ผิด format
3. Authorization header ไม่ถูกต้อง
วิธีแก้ไข - ตรวจสอบ API Key
import os
def verify_api_key(api_key: str) -> bool:
"""ตรวจสอบความถูกต้องของ API Key"""
import requests
if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
print("❌ กรุณาใส่ API Key ที่ถูกต้อง")
print(" สมัครได้ที่: https://www.holysheep.ai/register")
return False
# ทดสอบเรียก API ด้วย key ที่ได้รับ
response = requests.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"},
timeout=10
)
if response.status_code == 200:
print("✅ API Key ถูกต้อง")
return True
elif response.status_code == 401:
print("❌ API Key ไม่ถูกต้องหรือหมดอายุ")
return False
else:
print(f"⚠️ เกิดข้อผิดพลาด: {response.status_code}")
return False
วิธีแก้ไข - ตรวจสอบว่า key มาจาก environment variable
ไม่ควร hardcode API Key ในโค้ด
API_KEY = os.environ.get("HOLYSHEEP_API_KEY") # วิธีที่ถูกต้อง
API_KEY = "sk-xxx..." # ❌ ไม่ควรทำ
2. ConnectionError: timeout - การเชื่อมต่อใช้เวลานานเกินไป
# ปัญหา: requests.exceptions.ReadTimeout หรือ ConnectionError
สาเหตุที่พบบ่อย:
1. Request timeout สั้นเกินไป
2. เครือข่ายช้าหรือไม่เสถียร
3. Server ปลายทางมีปัญหา
วิธีแก้ไข - ปรับ timeout และเพิ่ม retry logic
import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_resilient_session():
"""สร้าง session ที่มี retry และ timeout ที่เหมาะสม"""
session = requests.Session()
# Retry strategy - ลองใหม่สูงสุด 3 ครั้ง
retry_strategy = Retry(
total=3,
backoff_factor=1, # รอ 1, 2, 4 วินาที (exponential backoff)
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def call_api_with_retry(api_key: str, payload: dict, max_retries: int = 3) -> dict:
"""เรียก API พร้อม retry logic"""
session = create_resilient_session()
url = "https://api.holysheep.ai/v1/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
for attempt in range(max_retries):
try:
# timeout = (connect_timeout, read_timeout)
response = session.post(
url,
headers=headers,
json=payload,
timeout=(10, 60) # 10 วินาทีสำหรับเชื่อมต่อ, 60 วินาทีสำหรับรอ response
)
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
print(f"⏰ ลองใหม่ครั้งที่ {attempt + 1}/{max_retries} - timeout")
if attempt < max_retries - 1:
time.sleep(2 ** attempt) # Exponential backoff
else:
raise Exception("เกินจำนวนครั้งที่กำหนด - timeout")
except requests.exceptions.ConnectionError as e:
print(f"🔌 ลองใหม่ครั้งที่ {attempt + 1}/{max_retries} - connection error")
if attempt < max_retries - 1:
time.sleep(2 ** attempt)
else:
raise Exception(f"ไม่สามารถเชื่อมต่อ: {str(e)}")
3. Rate Limit Exceeded - เรียกใช้ API เกินจำนวนที่กำหนด
# ปัญหา: ได้รับข้อผิดพลาด 429 Too Many Requests
สา
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง