ในยุคที่ AI กลายเป็นหัวใจสำคัญของธุรกิจดิจิทัล การติดตามและบันทึกการใช้งาน LLM API ไม่ใช่ทางเลือกอีกต่อไป แต่เป็นความจำเป็นเชิงกฎหมาย ความปลอดภัย และการควบคุมต้นทุน บทความนี้จะพาคุณสร้างระบบ Audit Log ที่พร้อมสำหรับ Production ตั้งแต่ศูนย์

ทำไมต้องมี Audit Log สำหรับ LLM API

จากประสบการณ์การ Deploy ระบบ AI หลายสิบโปรเจกต์ ผมพบว่าบริษัทที่ไม่มี Audit Log มักเจอปัญหาหลายอย่าง:

กรณีศึกษา: ระบบ AI Customer Service ของร้าน E-Commerce

ร้านค้าออนไลน์แห่งหนึ่งใช้ LLM ตอบคำถามลูกค้า 24/7 แต่พบว่า:

การแก้ปัญหาคือสร้าง Audit Logging Middleware ที่บันทึกทุก Request และ Response อย่างเป็นระบบ

การสร้าง Audit Logger ด้วย Python

โค้ดด้านล่างนี้ใช้งานได้จริงกับ HolySheep AI ซึ่งให้บริการ API หลากหลาย Models ด้วยราคาที่ประหยัด (DeepSeek V3.2 เพียง $0.42/MTok) และ Latency ต่ำกว่า 50ms ทำให้เหมาะสำหรับ Production ที่ต้องการ Logging แบบ Real-time

import json
import logging
import hashlib
from datetime import datetime, timezone
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, asdict, field
from pathlib import Path
import sqlite3

@dataclass
class AuditLogEntry:
    """โครงสร้างข้อมูลสำหรับบันทึกตรวจสอบ"""
    log_id: str
    timestamp: str
    user_id: Optional[str]
    session_id: str
    model: str
    provider: str
    
    # Request Data
    request_tokens: int
    request_cost: float
    prompt_preview: str  # ตัดยาวให้เหลือ 200 ตัวอักษร
    
    # Response Data  
    response_tokens: int
    response_cost: float
    response_preview: str
    full_response_hash: str
    
    # Metadata
    latency_ms: float
    status: str  # success, error, timeout
    error_message: Optional[str] = None
    ip_address: Optional[str] = None
    user_agent: Optional[str] = None
    
    def __post_init__(self):
        """สร้าง Log ID อัตโนมัติถ้ายังไม่มี"""
        if not self.log_id:
            self.log_id = hashlib.sha256(
                f"{self.timestamp}{self.session_id}{self.prompt_preview}".encode()
            ).hexdigest()[:16]

class LLMAuditLogger:
    """ระบบบันทึกตรวจสอบสำหรับ LLM API ทุกตัว"""
    
    def __init__(self, db_path: str = "llm_audit.db"):
        self.db_path = db_path
        self._init_database()
        
        # ตั้งค่า Logger
        self.logger = logging.getLogger("LLMAudit")
        self.logger.setLevel(logging.INFO)
        
        # Handler สำหรับไฟล์
        file_handler = logging.FileHandler("llm_audit.log")
        file_handler.setFormatter(
            logging.Formatter('%(asctime)s | %(levelname)s | %(message)s')
        )
        self.logger.addHandler(file_handler)
    
    def _init_database(self):
        """สร้างตาราง SQLite สำหรับเก็บ Log"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS audit_logs (
                log_id TEXT PRIMARY KEY,
                timestamp TEXT NOT NULL,
                user_id TEXT,
                session_id TEXT NOT NULL,
                model TEXT NOT NULL,
                provider TEXT NOT NULL,
                request_tokens INTEGER,
                request_cost REAL,
                prompt_preview TEXT,
                response_tokens INTEGER,
                response_cost REAL,
                response_preview TEXT,
                full_response_hash TEXT,
                latency_ms REAL,
                status TEXT,
                error_message TEXT,
                ip_address TEXT,
                user_agent TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        # สร้าง Index เพื่อความเร็วในการ Query
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_timestamp ON audit_logs(timestamp)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_user ON audit_logs(user_id)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_session ON audit_logs(session_id)')
        
        conn.commit()
        conn.close()
    
    def _calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int, provider: str) -> tuple[float, float]:
        """คำนวณค่าใช้จ่ายตาม Model และ Provider"""
        # ราคาจาก HolySheep AI (USD per Million Tokens)
        prices = {
            "holysheep": {
                "gpt-4.1": {"prompt": 8.0, "completion": 8.0},
                "claude-sonnet-4.5": {"prompt": 15.0, "completion": 15.0},
                "gemini-2.5-flash": {"prompt": 2.5, "completion": 2.5},
                "deepseek-v3.2": {"prompt": 0.42, "completion": 0.42},
            },
            "openai": {
                "gpt-4": {"prompt": 30.0, "completion": 60.0},
                "gpt-4-turbo": {"prompt": 10.0, "completion": 30.0},
            }
        }
        
        provider_prices = prices.get(provider, prices["holysheep"])
        model_prices = provider_prices.get(model.lower(), {"prompt": 10.0, "completion": 10.0})
        
        prompt_cost = (prompt_tokens / 1_000_000) * model_prices["prompt"]
        completion_cost = (completion_tokens / 1_000_000) * model_prices["completion"]
        
        return prompt_cost, completion_cost
    
    def _hash_content(self, content: str) -> str:
        """สร้าง Hash ของเนื้อหาเพื่อไม่ต้องเก็บ Response ทั้งหมด"""
        return hashlib.sha256(content.encode()).hexdigest()[:32]
    
    def _truncate(self, text: str, length: int = 200) -> str:
        """ตัดข้อความยาวให้สั้นลง"""
        if len(text) <= length:
            return text
        return text[:length-3] + "..."
    
    def log_request(self, 
                   session_id: str,
                   model: str,
                   provider: str,
                   prompt: str,
                   user_id: Optional[str] = None,
                   ip_address: Optional[str] = None,
                   user_agent: Optional[str] = None) -> AuditLogEntry:
        """บันทึก Request ที่กำลังจะส่งไป API"""
        
        timestamp = datetime.now(timezone.utc).isoformat()
        
        entry = AuditLogEntry(
            log_id="",  # จะถูกสร้างใน __post_init__
            timestamp=timestamp,
            user_id=user_id,
            session_id=session_id,
            model=model,
            provider=provider,
            request_tokens=0,  # จะอัพเดทหลังได้ response
            request_cost=0.0,
            prompt_preview=self._truncate(prompt),
            response_tokens=0,
            response_cost=0.0,
            response_preview="PENDING",
            full_response_hash="",
            latency_ms=0.0,
            status="in_progress",
            ip_address=ip_address,
            user_agent=user_agent
        )
        
        return entry
    
    def log_response(self, 
                    entry: AuditLogEntry,
                    response: str,
                    prompt_tokens: int,
                    completion_tokens: int,
                    latency_ms: float,
                    status: str = "success",
                    error_message: Optional[str] = None):
        """บันทึก Response ที่ได้รับจาก API"""
        
        prompt_cost, completion_cost = self._calculate_cost(
            entry.model, prompt_tokens, completion_tokens, entry.provider
        )
        
        entry.request_tokens = prompt_tokens
        entry.request_cost = prompt_cost
        entry.response_tokens = completion_tokens
        entry.response_cost = completion_cost
        entry.response_preview = self._truncate(response)
        entry.full_response_hash = self._hash_content(response)
        entry.latency_ms = latency_ms
        entry.status = status
        entry.error_message = error_message
        
        # บันทึกลง Database
        self._save_to_db(entry)
        
        # บันทึกลง Log file
        log_msg = f"{entry.status} | {entry.model} | {entry.user_id or 'anonymous'} | "
        log_msg += f"req:{prompt_tokens}tok ${prompt_cost:.4f} | "
        log_msg += f"res:{completion_tokens}tok ${completion_cost:.4f} | "
        log_msg += f"{latency_ms:.0f}ms"
        
        if status == "error":
            self.logger.error(log_msg + f" | ERROR: {error_message}")
        else:
            self.logger.info(log_msg)
    
    def _save_to_db(self, entry: AuditLogEntry):
        """บันทึก Log Entry ลง SQLite"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT OR REPLACE INTO audit_logs 
            (log_id, timestamp, user_id, session_id, model, provider,
             request_tokens, request_cost, prompt_preview,
             response_tokens, response_cost, response_preview, full_response_hash,
             latency_ms, status, error_message, ip_address, user_agent)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ''', (
            entry.log_id, entry.timestamp, entry.user_id, entry.session_id,
            entry.model, entry.provider, entry.request_tokens, entry.request_cost,
            entry.prompt_preview, entry.response_tokens, entry.response_cost,
            entry.response_preview, entry.full_response_hash, entry.latency_ms,
            entry.status, entry.error_message, entry.ip_address, entry.user_agent
        ))
        
        conn.commit()
        conn.close()
    
    def get_cost_summary(self, 
                        start_date: Optional[str] = None,
                        end_date: Optional[str] = None,
                        user_id: Optional[str] = None) -> Dict[str, Any]:
        """ดึงสรุปค่าใช้จ่าย"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        query = '''
            SELECT model, provider, 
                   SUM(request_tokens) as total_req_tokens,
                   SUM(response_tokens) as total_res_tokens,
                   SUM(request_cost + response_cost) as total_cost,
                   COUNT(*) as total_requests,
                   AVG(latency_ms) as avg_latency
            FROM audit_logs
            WHERE status = 'success'
        '''
        params = []
        
        if start_date:
            query += " AND timestamp >= ?"
            params.append(start_date)
        if end_date:
            query += " AND timestamp <= ?"
            params.append(end_date)
        if user_id:
            query += " AND user_id = ?"
            params.append(user_id)
        
        query += " GROUP BY model, provider ORDER BY total_cost DESC"
        
        cursor.execute(query, params)
        results = cursor.fetchall()
        conn.close()
        
        return {
            "summary": [
                {
                    "model": r[0],
                    "provider": r[1],
                    "total_prompt_tokens": r[2],
                    "total_completion_tokens": r[3],
                    "total_cost_usd": round(r[4], 4),
                    "total_requests": r[5],
                    "avg_latency_ms": round(r[6], 2)
                }
                for r in results
            ]
        }


วิธีใช้งาน

logger = LLMAuditLogger("production_audit.db")
print("Audit Logger initialized successfully!")

Integration กับ HolySheep AI API

ด้านล่างคือโค้ดที่พร้อมใช้งานจริงสำหรับเชื่อมต่อกับ HolySheep AI ซึ่งรองรับ Models ยอดนิยมทั้ง GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash และ DeepSeek V3.2 โดยมีอัตราแลกเปลี่ยนที่คุ้มค่ามาก (¥1 = $1 หรือประหยัดกว่า 85% เมื่อเทียบกับราคาตลาดอื่น)

import openai
import time
import tiktoken
from typing import Optional

class HolySheepAIClient:
    """Client สำหรับเชื่อมต่อ HolySheep AI พร้อม Audit Logging"""
    
    def __init__(self, api_key: str, model: str = "deepseek-v3.2"):
        self.client = openai.OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"  # URL ของ HolySheep AI เท่านั้น
        )
        self.model = model
        self.encoding = tiktoken.get_encoding("cl100k_base")  # สำหรับ GPT-4
        self.audit_logger = LLMAuditLogger("holysheep_audit.db")
    
    def _count_tokens(self, text: str) -> int:
        """นับจำนวน Token ในข้อความ"""
        return len(self.encoding.encode(text))
    
    def chat(self, 
             messages: list,
             session_id: str,
             user_id: Optional[str] = None,
             temperature: float = 0.7,
             max_tokens: int = 2048) -> dict:
        """ส่ง Chat Request พร้อมบันทึก Audit Log"""
        
        # รวมข้อความสำหรับ Logging
        full_prompt = "\n".join([f"{m['role']}: {m['content']}" for m in messages])
        
        # สร้าง Log Entry ก่อน Request
        log_entry = self.audit_logger.log_request(
            session_id=session_id,
            model=self.model,
            provider="holysheep",
            prompt=full_prompt,
            user_id=user_id
        )
        
        start_time = time.time()
        
        try:
            # ส่ง Request ไปยัง HolySheep AI
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens
            )
            
            latency_ms = (time.time() - start_time) * 1000
            
            # ดึงข้อมูล Response
            assistant_message = response.choices[0].message.content
            
            # นับ Tokens
            prompt_tokens = self._count_tokens(full_prompt)
            completion_tokens = self._count_tokens(assistant_message)
            
            # บันทึก Response
            self.audit_logger.log_response(
                entry=log_entry,
                response=assistant_message,
                prompt_tokens=prompt_tokens,
                completion_tokens=completion_tokens,
                latency_ms=latency_ms,
                status="success"
            )
            
            return {
                "success": True,
                "message": assistant_message,
                "usage": {
                    "prompt_tokens": prompt_tokens,
                    "completion_tokens": completion_tokens,
                    "total_tokens": prompt_tokens + completion_tokens
                },
                "latency_ms": latency_ms,
                "log_id": log_entry.log_id
            }
            
        except Exception as e:
            latency_ms = (time.time() - start_time) * 1000
            
            # บันทึก Error
            self.audit_logger.log_response(
                entry=log_entry,
                response="",
                prompt_tokens=self._count_tokens(full_prompt),
                completion_tokens=0,
                latency_ms=latency_ms,
                status="error",
                error_message=str(e)
            )
            
            return {
                "success": False,
                "error": str(e),
                "latency_ms": latency_ms,
                "log_id": log_entry.log_id
            }


ตัวอย่างการใช้งาน

if __name__ == "__main__":
    # สร้าง Client (ใช้ API Key จาก HolySheep AI)
    client = HolySheepAIClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",  # แทนที่ด้วย API Key จริง
        model="deepseek-v3.2"  # Model ราคาถูก แต่คุณภาพดี
    )
    
    # ทดสอบ Chat
    result = client.chat(
        messages=[
            {"role": "system", "content": "คุณเป็นผู้ช่วยแนะนำสินค้า"},
            {"role": "user", "content": "แนะนำโทรศัพท์มือถือราคา 10,000 บาท"}
        ],
        session_id="user123-session456",
        user_id="user123"
    )
    
    if result["success"]:
        print(f"Response: {result['message']}")
        print(f"Tokens used: {result['usage']['total_tokens']}")
        print(f"Latency: {result['latency_ms']:.2f}ms")
        print(f"Log ID: {result['log_id']}")
    else:
        print(f"Error: {result['error']}")

Dashboard สำหรับ Monitor ค่าใช้จ่ายแบบ Real-time

import streamlit as st
import pandas as pd
from datetime import datetime, timedelta
import plotly.express as px

def render_audit_dashboard(db_path: str = "holysheep_audit.db"):
    """สร้าง Dashboard สำหรับ Monitor การใช้งาน API"""
    
    st.set_page_config(page_title="LLM API Audit Dashboard", layout="wide")
    
    st.title("📊 LLM API Audit Dashboard")
    st.markdown("ติดตามการใช้งานและค่าใช้จ่ายแบบ Real-time")
    
    # เชื่อมต่อ Database
    import sqlite3
    conn = sqlite3.connect(db_path)
    
    # Sidebar filters
    st.sidebar.header("ตัวกรอง")
    
    # เลือกช่วงเวลา
    time_range = st.sidebar.selectbox(
        "ช่วงเวลา",
        ["วันนี้", "7 วัน", "30 วัน", "ทั้งหมด"]
    )
    
    days_map = {"วันนี้": 1, "7 วัน": 7, "30 วัน": 30, "ทั้งหมด": 365}
    days = days_map[time_range]
    
    # ดึงข้อมูล Summary
    df = pd.read_sql_query(f'''
        SELECT 
            DATE(timestamp) as date,
            model,
            provider,
            SUM(request_tokens + response_tokens) as total_tokens,
            SUM(request_cost + response_cost) as total_cost,
            COUNT(*) as total_requests,
            AVG(latency_ms) as avg_latency,
            SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) as success_count,
            SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as error_count
        FROM audit_logs
        WHERE timestamp >= datetime('now', '-{days} days')
        GROUP BY DATE(timestamp), model, provider
        ORDER BY date DESC
    ''', conn)
    
    # Metrics Cards
    col1, col2, col3, col4 = st.columns(4)
    
    total_cost = df['total_cost'].sum()
    total_tokens = df['total_tokens'].sum()
    total_requests = df['total_requests'].sum()
    avg_latency = df['avg_latency'].mean()
    
    with col1:
        st.metric("💰 ค่าใช้จ่ายรวม", f"${total_cost:.2f}")
    with col2:
        st.metric("🔢 Tokens ที่ใช้", f"{total_tokens:,}")
    with col3:
        st.metric("📝 Requests", f"{total_requests:,}")
    with col4:
        st.metric("⚡ Latency เฉลี่ย", f"{avg_latency:.0f}ms")
    
    # Charts
    st.subheader("📈 กราฟค่าใช้จ่ายรายวัน")
    
    if not df.empty:
        # กราฟค่าใช้จ่ายตาม Model
        fig1 = px.bar(
            df.groupby('model')['total_cost'].sum().reset_index(),
            x='model', y='total_cost',
            title='ค่าใช้จ่ายแยกตาม Model',
            color='model'
        )
        st.plotly_chart(fig1, use_container_width=True)
        
        # กราฟ Requests ตามเวลา
        fig2 = px.line(
            df.groupby('date')['total_requests'].sum().reset_index(),
            x='date', y='total_requests',
            title='จำนวน Requests รายวัน'
        )
        st.plotly_chart(fig2, use_container_width=True)
    
    # Recent Logs Table
    st.subheader("📋 Log ล่าสุด")
    
    recent_logs = pd.read_sql_query('''
        SELECT 
            timestamp,
            user_id,
            model,
            request_tokens,
            response_tokens,
            request_cost + response_cost as cost,
            latency_ms,
            status
        FROM audit_logs
        ORDER BY timestamp DESC
        LIMIT 100
    ''', conn)
    
    st.dataframe(recent_logs, use_container_width=True)
    
    conn.close()


รัน Dashboard

if __name__ == "__main__":
    render_audit_dashboard()

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

1. Error: "Connection timeout" หรือ Latency สูงผิดปกติ

สาเหตุ: การเชื่อมต่อ API โดยไม่มี Retry Logic หรือ Timeout ที่เหมาะสม

# โค้ดที่ผิด - ไม่มี Timeout
response = openai.ChatCompletion.create(
    model="deepseek-v3.2",
    messages=messages
)


โค้ดที่ถูกต้อง - มี Timeout และ Retry

from openai import OpenAI
from tenacity import retry, stop_after_attempt, wait_exponential

client = OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    timeout=30.0,  # Timeout 30 วินาที
    max_retries=3  # Retry สูงสุด 3 ครั้ง
)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_with_retry(messages):
    return client.chat.completions.create(
        model="deepseek-v3.2",
        messages=messages,
        timeout=30.0
    )

2. Error: "Invalid API Key" หรือ Authentication Failed

สาเหตุ: API Key ไม่ถูกต้อง หรือถูก Limit Quota

# โค้ดที่ผิด - ไม่ตรวจสอบ Key ก่อนใช้งาน
client = OpenAI(
    api_key="sk-xxx",  # Key อาจหมดอายุ
    base_url="https://api.holysheep.ai/v1"
)


โค้ดที่ถูกต้อง - ตรวจสอบ Key ก่อน

import os

def validate_api_key(api_key: str) -> bool:
    """ตรวจสอบความถูกต้องของ API Key"""
    if not api_key or len(api_key) < 10:
        return False
    
    try:
        test_client = OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"
        )
        # ทดสอบด้วย Request เล็กๆ
        test_client.models.list()
        return True
    except Exception as e:
        print(f"API Key validation failed: {e}")
        return False


ใช้งาน

API_KEY = os.environ.get("HOLYSHEEP_API_KEY")
if validate_api_key(API_KEY):
    client = OpenAI(
        api_key=API_KEY,
        base_url="https://api.holysheep.ai/v1"
    )
else:
    raise ValueError("Invalid or expired API Key")

3. Error: "Token limit exceeded" หรือ Context Overflow

สาเหตุ: Prompt หรือ History ยาวเกิน Context Window ของ Model

# โค้ดที่ผิด - ไม่จำกัด Context Length
def build_prompt(messages):
    return "\n".join([f"{m['role']}: {m['content']}" for m in messages])


โค้ดที่ถูกต้อง - จำกัด Context และ Summarize

def build_prompt_safe(messages: list, max_tokens: int = 3000) -> str:
    """สร้าง Prompt ที่ควบคุม Context Length ได้"""
    import tiktoken
    
    encoding = tiktoken.get_encoding("cl100k_base")
    
    # คำนวณ Token ที่เหลือสำหรับ History
    system_tokens = 200  # เก็บไว้สำหรับ System Prompt
    
    result = []
    remaining_tokens = max_tokens - system_tokens
    
    # วนจากข้อความล่าสุดขึ้นไป
    for message in reversed(messages):
        if message["role"] == "system":
            continue
            
        content_tokens = len(encoding.encode(message["content

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง