เมื่อเดือนที่แล้ว ทีมของเราเจอปัญหาสำคัญ: 401 Unauthorized ปรากฏขึ้นกลางคืนขณะระบบทำงาน production แต่ไม่มีใครรู้ว่า API key ถูก revoke ตอนไหน หรือใครเป็นคน revoke การไม่มี audit trail ที่ดีทำให้การแก้ปัญหาใช้เวลากว่า 4 ชั่วโมง บทความนี้จะสอนวิธีสร้างระบบ audit trail ที่ครบถ้วนสำหรับ การใช้งาน AI API อย่างปลอดภัย
ทำไมต้องมี Audit Trail?
- การตรวจสอบย้อนกลับ (Traceability): รู้ว่าใครเรียก API ตอนไหน ด้วย parameter อะไร
- ความปลอดภัย: ตรวจจับการใช้งานผิดปกติหรือ key ที่รั่วไหลได้ทันที
- การปฏิบัติตามกฎระเบียบ (Compliance): หลายอุตสาหกรรมต้องมี log การใช้งานอย่างน้อย 90 วัน
- การวิเคราะห์ค่าใช้จ่าย: เข้าใจ pattern การใช้งานเพื่อ optimize ต้นทุน
การตั้งค่า Audit Logger พื้นฐาน
เริ่มต้นด้วยการสร้าง class สำหรับ log ทุก request ไปยัง HolySheep AI ระบบ audit trail ที่ดีต้องบันทึก timestamp, request details, response status, latency และ cost อย่างครบถ้วน
import requests
import json
from datetime import datetime
from typing import Optional, Dict, Any
import hashlib
class HolySheepAuditLogger:
"""Audit trail logger สำหรับ HolySheep AI API"""
def __init__(self, api_key: str, log_file: str = "audit_log.jsonl"):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.log_file = log_file
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def _calculate_cost(self, model: str, tokens: int) -> float:
"""คำนวณค่าใช้จ่ายจริงตามราคาปี 2026"""
pricing = {
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4.5": 15.0, # $15/MTok
"gemini-2.5-flash": 2.50, # $2.50/MTok
"deepseek-v3.2": 0.42 # $0.42/MTok
}
return (tokens / 1_000_000) * pricing.get(model, 0)
def _log_request(self, entry: Dict[str, Any]):
"""บันทึก log ลงไฟล์ JSONL"""
entry["timestamp"] = datetime.utcnow().isoformat()
entry["checksum"] = hashlib.sha256(
json.dumps(entry, sort_keys=True).encode()
).hexdigest()[:16]
with open(self.log_file, "a") as f:
f.write(json.dumps(entry) + "\n")
def chat_completion(
self,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: int = 1000
) -> Dict[str, Any]:
"""เรียก API พร้อมบันทึก audit trail"""
start_time = datetime.utcnow()
request_id = hashlib.md5(
f"{start_time.isoformat()}{model}".encode()
).hexdigest()[:12]
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=30
)
latency_ms = (datetime.utcnow() - start_time).total_seconds() * 1000
result = response.json()
usage = result.get("usage", {})
total_tokens = usage.get("total_tokens", 0)
log_entry = {
"request_id": request_id,
"event": "api_call",
"model": model,
"status_code": response.status_code,
"latency_ms": round(latency_ms, 2),
"input_tokens": usage.get("prompt_tokens", 0),
"output_tokens": usage.get("completion_tokens", 0),
"total_tokens": total_tokens,
"cost_usd": round(self._calculate_cost(model, total_tokens), 4),
"success": response.ok
}
self._log_request(log_entry)
if not response.ok:
log_entry["error"] = result.get("error", {})
self._log_request(log_entry)
return result
except requests.exceptions.Timeout:
self._log_request({
"request_id": request_id,
"event": "timeout",
"model": model,
"latency_ms": 30000,
"success": False,
"error": "ConnectionTimeout: Request exceeded 30s"
})
raise
def get_usage_summary(self, days: int = 7) -> Dict[str, Any]:
"""สรุปการใช้งานจาก log file"""
summary = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_cost_usd": 0.0,
"total_tokens": 0,
"model_breakdown": {},
"avg_latency_ms": 0
}
latencies = []
try:
with open(self.log_file, "r") as f:
for line in f:
entry = json.loads(line)
if entry["event"] != "api_call":
continue
summary["total_requests"] += 1
if entry["success"]:
summary["successful_requests"] += 1
else:
summary["failed_requests"] += 1
summary["total_cost_usd"] += entry["cost_usd"]
summary["total_tokens"] += entry["total_tokens"]
latencies.append(entry["latency_ms"])
model = entry["model"]
if model not in summary["model_breakdown"]:
summary["model_breakdown"][model] = {
"requests": 0, "tokens": 0, "cost": 0
}
summary["model_breakdown"][model]["requests"] += 1
summary["model_breakdown"][model]["tokens"] += entry["total_tokens"]
summary["model_breakdown"][model]["cost"] += entry["cost_usd"]
summary["avg_latency_ms"] = round(
sum(latencies) / len(latencies), 2
) if latencies else 0
except FileNotFoundError:
pass
return summary
ตัวอย่างการใช้งาน
logger = HolySheepAuditLogger(
api_key="YOUR_HOLYSHEEP_API_KEY",
log_file="holysheep_audit.jsonl"
)
messages = [{"role": "user", "content": "อธิบาย audit trail"}]
response = logger.chat_completion(
model="deepseek-v3.2",
messages=messages,
max_tokens=500
)
print(f"Response: {response['choices'][0]['message']['content']}")
ตรวจสอบค่าใช้จ่าย
summary = logger.get_usage_summary(days=7)
print(f"Total cost this week: ${summary['total_cost_usd']:.4f}")การตรวจจับความผิดปกติแบบ Real-time
นอกจาก log พื้นฐาน เราควรมีระบบ alert เมื่อมีสิ่งผิดปกติ เช่น จำนวน request ที่พุ่งสูงผิดปกติ หรือ latency ที่สูงเกินปกติ ระบบ HolySheep AI มี latency เฉลี่ยต่ำกว่า 50ms ดังนั้นถ้าเริ่มเห็นค่าเฉลี่ยสูงขึ้น แสดงว่ามีปัญหา
import time
from collections import deque
from threading import Lock
class AnomalyDetector:
"""ตรวจจับความผิดปกติใน API usage patterns"""
def __init__(
self,
rate_limit_per_minute: int = 100,
max_avg_latency_ms: float = 100,
error_threshold_percent: float = 5.0
):
self.rate_limit = rate_limit_per_minute
self.max_latency = max_avg_latency_ms
self.error_threshold = error_threshold_percent
self.request_timestamps = deque(maxlen=1000)
self.latencies = deque(maxlen=100)
self.errors = deque(maxlen=100)
self._lock = Lock()
self.alerts = []
def record_request(self, latency_ms: float, success: bool):
"""บันทึก metrics ล่าสุด"""
with self._lock:
now = time.time()
self.request_timestamps.append(now)
self.latencies.append(latency_ms)
self.errors.append(0 if success else 1)
self._check_anomalies()
def _check_anomalies(self):
"""ตรวจสอบทุกเงื่อนไขที่ผิดปกติ"""
now = time.time()
# ตรวจ: Rate limit
recent = [
t for t in self.request_timestamps
if now - t < 60
]
if len(recent) > self.rate_limit:
self.alerts.append({
"type": "RATE_LIMIT_EXCEEDED",
"message": f"Requests per minute: {len(recent)} (limit: {self.rate_limit})",
"timestamp": datetime.utcnow().isoformat()
})
# ตรวจ: Latency
if len(self.latencies) >= 10:
avg_latency = sum(self.latencies) / len(self.latencies)
if avg_latency > self.max_latency:
self.alerts.append({
"type": "HIGH_LATENCY",
"message": f"Avg latency: {avg_latency:.2f}ms (threshold: {self.max_latency}ms)",
"timestamp": datetime.utcnow().isoformat()
})
# ตรวจ: Error rate
if len(self.errors) >= 10:
error_rate = (sum(self.errors) / len(self.errors)) * 100
if error_rate > self.error_threshold:
self.alerts.append({
"type": "HIGH_ERROR_RATE",
"message": f"Error rate: {error_rate:.2f}% (threshold: {self.error_threshold}%)",
"timestamp": datetime.utcnow().isoformat()
})
def get_status(self) -> Dict[str, Any]:
"""สถานะปัจจุบันของระบบ"""
with self._lock:
now = time.time()
recent_requests = [
t for t in self.request_timestamps
if now - t < 60
]
return {
"requests_last_minute": len(recent_requests),
"avg_latency_ms": round(
sum(self.latencies) / len(self.latencies), 2
) if self.latencies else 0,
"error_rate_percent": round(
(sum(self.errors) / len(self.errors)) * 100, 2
) if self.errors else 0,
"recent_alerts": self.alerts[-5:],
"healthy": len(self.alerts) < 3
}
ใช้งานร่วมกับ API logger
detector = AnomalyDetector(
rate_limit_per_minute=60,
max_avg_latency_ms=80, # HolySheep ปกติ <50ms
error_threshold_percent=2.0
)
ทดสอบการตรวจจับ
detector.record_request(latency_ms=45.2, success=True)
detector.record_request(latency_ms=120.5, success=False) # สูงผิดปกติ
detector.record_request(latency_ms=130.0, success=False) # สูงต่อเนื่อง
status = detector.get_status()
print(f"System status: {status}")
print(f"Recent alerts: {status['recent_alerts']}")การสร้าง Compliance Report อัตโนมัติ
องค์กรหลายแห่งต้องมี compliance report สำหรับ GDPR, SOC2 หรือ HIPAA ด้านล่างคือ script ที่สร้าง report อัตโนมัติพร้อม export เป็น CSV และ PDF summary
from datetime import datetime, timedelta
import csv
class ComplianceReporter:
"""สร้าง compliance report อัตโนมัติ"""
def __init__(self, log_file: str = "holysheep_audit.jsonl"):
self.log_file = log_file
def generate_report(
self,
start_date: datetime,
end_date: datetime,
output_csv: str = "compliance_report.csv"
) -> Dict[str, Any]:
"""สร้าง report ตามช่วงเวลาที่กำหนด"""
report = {
"period": f"{start_date.date()} to {end_date.date()}",
"total_api_calls": 0,
"successful_calls": 0,
"failed_calls": 0,
"total_cost_usd": 0.0,
"tokens_by_model": {},
"daily_breakdown": {},
"security_events": [],
"pii_accessed": False
}
with open(self.log_file, "r") as f:
for line in f:
entry = json.loads(line)
entry_time = datetime.fromisoformat(entry["timestamp"])
# กรองเฉพาะช่วงเวลาที่ต้องการ
if not (start_date <= entry_time <= end_date):
continue
if entry.get("event") != "api_call":
continue
report["total_api_calls"] += 1
if entry["success"]:
report["successful_calls"] += 1
else:
report["failed_calls"] += 1
report["security_events"].append({
"timestamp": entry["timestamp"],
"request_id": entry["request_id"],
"error": entry.get("error", {})
})
report["total_cost_usd"] += entry["cost_usd"]
# แยกตาม model
model = entry["model"]
if model not in report["tokens_by_model"]:
report["tokens_by_model"][model] = 0
report["tokens_by_model"][model] += entry["total_tokens"]
# แยกตามวัน
day = entry_time.date().isoformat()
if day not in report["daily_breakdown"]:
report["daily_breakdown"][day] = {
"calls": 0, "tokens": 0, "cost": 0
}
report["daily_breakdown"][day]["calls"] += 1
report["daily_breakdown"][day]["tokens"] += entry["total_tokens"]
report["daily_breakdown"][day]["cost"] += entry["cost_usd"]
# Export CSV
self._export_csv(output_csv, report["daily_breakdown"])
return report
def _export_csv(self, filename: str, daily_data: Dict):
"""Export daily breakdown เป็น CSV"""
with open(filename, "w", newline="") as f:
writer = csv.writer(f)
writer.writerow(["Date", "API Calls", "Total Tokens", "Cost (USD)"])
for date, data in sorted(daily_data.items()):
writer.writerow([
date,
data["calls"],
data["tokens"],
f"{data['cost']:.4f}"
])
def get_security_summary(self) -> Dict[str, Any]:
"""สรุป security events ทั้งหมด"""
events = []
try:
with open(self.log_file, "r") as f:
for line in f:
entry = json.loads(line)
if entry.get("event") == "api_call" and not entry["success"]:
events.append(entry)
except FileNotFoundError:
pass
return {
"total_security_events": len(events),
"recent_events": events[-10:],
"most_common_errors": self._count_errors(events)
}
def _count_errors(self, events: list) -> Dict[str, int]:
"""นับประเภทของ error"""
counts = {}
for event in events:
error_type = event.get("error", {}).get("type", "unknown")
counts[error_type] = counts.get(error_type, 0) + 1
return counts
สร้าง report รายเดือน
reporter = ComplianceReporter()
start = datetime.utcnow() - timedelta(days=30)
end = datetime.utcnow()
monthly_report = reporter.generate_report(
start_date=start,
end_date=end,
output_csv="monthly_compliance.csv"
)
print(f"Monthly Summary:")
print(f"- Total calls: {monthly_report['total_api_calls']}")
print(f"- Success rate: {monthly_report['successful_calls']/monthly_report['total_api_calls']*100:.1f}%")
print(f"- Total cost: ${monthly_report['total_cost_usd']:.2f}")
print(f"- Security events: {len(monthly_report['security_events'])}")
security = reporter.get_security_summary()
print(f"- Most common errors: {security['most_common_errors']}")ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
กรณีที่ 1: 401 Unauthorized — Invalid API Key
อาการ: ได้รับ error {"error": {"code": 401, "message": "Invalid API key"}} ทุกครั้งที่เรียก API แม้ว่าจะตั้งค่า key ถูกต้อง
สาเหตุที่พบบ่อย:
- API key ถูก revoke ไปแล้ว
- ใส่ช่องว่างหรือตัวอักษรพิเศษผิด
- key หมดอายุ (expired)
- เรียกใช้ model ที่ไม่มีสิทธิ์เข้าถึง
วิธีแก้ไข:
# ตรวจสอบ API key ก่อนใช้งาน
import requests
def verify_api_key(api_key: str) -> bool:
"""ตรวจสอบความถูกต้องของ API key"""
response = requests.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"},
timeout=10
)
if response.status_code == 401:
print("❌ API key ไม่ถูกต้องหรือถูก revoke แล้ว")
return False
if response.status_code == 200:
print("✅ API key ถูกต้อง")
return True
print(f"⚠️ Error: {response.status_code} - {response.text}")
return False
ใช้งาน
if verify_api_key("YOUR_HOLYSHEEP_API_KEY"):
logger = HolySheepAuditLogger(api_key="YOUR_HOLYSHEEP_API_KEY")
else:
# Redirect ไปสมัคร key ใหม่
print("กรุณาสร้าง API key ใหม่ที่: https://www.holysheep.ai/register")กรณีที่ 2: ConnectionError: timeout ต่อเนื่อง
อาการ: ได้รับ requests.exceptions.ReadTimeout หรือ ConnectionError: timeout ซ้ำๆ กัน
สาเหตุที่พบบ่อย:
- Network timeout ต่ำเกินไป
- Request body ใหญ่เกินไป (prompt ยาวมาก)
- Server มีปัญหาหรือ maintenance
- Firewall หรือ proxy บล็อก request
วิธีแก้ไข:
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import requests
def create_resilient_session() -> requests.Session:
"""สร้าง session ที่ retry อัตโนมัติเมื่อ timeout"""
session = requests.Session()
# ตั้งค่า retry strategy
retry_strategy = Retry(
total=3,
backoff_factor=1, # รอ 1s, 2s, 4s ระหว่าง retry
status_forcelist=[408, 429, 500, 502, 503, 504],
allowed_methods=["POST", "GET"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
session.headers.update({
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"
})
return session
ใช้งานพร้อม timeout ที่เหมาะสม
session = create_resilient_session()
try:
response = session.post(
"https://api.holysheep.ai/v1/chat/completions",
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": "ทดสอบ"}],
"max_tokens": 100
},
timeout=(10, 60) # (connect_timeout, read_timeout)
)
print(f"Response: {response.json()}")
except requests.exceptions.Timeout:
print("❌ Request timeout - ลองลด max_tokens หรือแบ่ง request")
except requests.exceptions.ConnectionError as e:
print(f"❌ Connection error: {e}")กรณีที่ 3: Quota Exceeded — ค่าใช้จ่ายพุ่งสูงผิดปกติ
อาการ: ได้รับ 429 Too Many Requests หรือค่าใช้จ่ายในเดือนนี้สูงกว่าปกติมาก
สาเหตุที่พบบ่อย:
- Token usage สูงเกิน limit ที่กำหนด
- API key รั่วไหลถูกนำไปใช้โดยไม่ได้รับอนุญาต
- Code มี infinite loop หรือ recursive ที่เรียก API ซ้ำ
- Cache ไม่ทำงาน ทำให้ถามซ้ำๆ กัน
วิธีแก้ไข:
# ตรวจสอบและจำกัดการใช้งานด้วย Budget Controller
from datetime import datetime, timedelta
class BudgetController:
"""ควบคุมงบประมาณ API อัตโนมัติ"""
def __init__(
self,
monthly_budget_usd: float = 100.0,
daily_limit_usd: float = 10.0
):
self.monthly_budget = monthly_budget_usd
self.daily_limit = daily_limit_usd
self.monthly_spent = 0.0
self.daily_spent = 0.0
self.daily_reset_date = datetime.utcnow().date()
def _reset_daily_if_needed(self):
"""รีเซ็ต daily counter ทุกวัน"""
today = datetime.utcnow().date()
if today > self.daily_reset_date:
self.daily_spent = 0.0
self.daily_reset_date = today
def check_and_record(self, cost_usd: float) -> bool:
"""ตรวจสอบก่อนเรียก API - return True ถ้าผ่าน"""
self._reset_daily_if_needed()
if self.monthly_spent + cost_usd > self.monthly_budget:
print(f"❌ เกิน monthly budget: ${self.monthly_spent:.2f}/${self.monthly_budget}")
return False
if self.daily_spent + cost_usd > self.daily_limit:
print(f"❌ เกิน daily limit: ${self.daily_spent:.2f}/${self.daily_limit}")
return False
self.monthly_spent += cost_usd
self.daily_spent += cost_usd
return True
def get_remaining(self) -> Dict[str, float]:
"""ดูยอดคงเหลือ"""
self._reset_daily_if_needed()
return {
"monthly_remaining": round(self.monthly_budget - self.monthly_spent, 2),
"daily_remaining": round(self.daily_limit - self.daily_spent, 2),
"monthly_spent": round(self.monthly_spent, 2),
"daily_spent": round(self.daily_spent, 2)
}
ใช้งาน
budget = BudgetController(monthly_budget_usd=50.0, daily_limit_usd=5.0)
def safe_api_call(model: str, messages: list, cost_estimate: float):
"""เรียก API เมื่อผ่านการตรวจสอบงบประมาณ"""
if not budget.check_and_record(cost_estimate):
raise Exception(f"Budget limit exceeded - Remaining: {budget.get_remaining()}")
# ดำเนินการ API call จริง
logger = HolySheepAuditLogger(api_key="YOUR_HOLYSHEEP_API_KEY")
return logger.chat_completion(model=model, messages=messages)
ตัวอย่าง: คำนวณค่าใช้จ่ายล่วงหน้า
estimated_cost = 0.00042 # deepseek-v3.2: 500 tokens
if safe_api_call("deepseek-v3.2", messages, estimated_cost):
print("API call สำเร็จ")
print(f"งบคงเหลือ: {budget.get_remaining()}")สรุป
การสร้าง audit trail ที่ดีไม่ใช่แค่การบันทึก log แต่รวมถึงการตรวจจับความผิดปกติ การควบคุมงบประมาณ และการสร้าง report สำหรับ compliance อัตโนมัติ ระบบที่อธิบายในบทความนี้ช่วยให้:
- แก้ป