คู่มือฉบับสมบูรณ์: จัดเก็บ Log คำขอ AI API แบบมีโครงสร้างด้วย ClickHouse เพื่อ Observability และการแก้ปัญหา

ผมเคยเจอเหตุการณ์ลูกค้าแจ้งว่า "AI ตอบช้ามากช่วงบ่ายสาม" แต่ผมเปิด Dashboard ของระบบ proxy ทั่วไปดูไม่ออกเลยว่าคำขอไหนคอขวด ต้นทุนเท่าไหร่ โมเดลไหนแพงเกิน จนวันหนึ่งผมย้าย Log ทั้งหมดเข้า ClickHouse แล้วออกแบบ schema ใหม่ — เวลาแก้ปัญหาลดจาก 2 ชั่วโมงเหลือ 4 นาที บทความนี้คือสิ่งที่ผมอยากแชร์ พร้อมเปรียบเทียบกับการใช้ Official API และ Relay อื่นๆ รวมถึงเหตุผลที่ผมเลือกใช้บริการของ สมัครที่นี่ เป็นตัวหลักในการทดสอบ

ตารางเปรียบเทียบ: HolySheep vs Official API vs Relay ทั่วไป

ทำไมต้องจัดเก็บ Log แบบมีโครงสร้าง (Structured Logging)

Log แบบ raw text เช่น INFO 2026-01-12 10:03:21 request_id=abc model=gpt-4.1 tokens=1240 latency=0.42s เก็บง่าย แต่ค้นหายากและแยก field ไม่ได้ ClickHouse เป็น columnar database ที่ออกแบบมาเพื่อ time-series โดยเฉพาะ เมื่อเทียบกับ Elasticsearch:

• อัตราการบีบอัด 10–20 เท่า (ZSTD) — Log 1 ปี ขนาด 50 GB เหลือ 3 GB
• ความเร็วในการ INSERT สูงถึง 1 ล้านแถวต่อวินาทีต่อ shard
• ค่าใช้จ่าย storage ต่ำกว่า Elasticsearch ประมาณ 5–10 เท่า
• SQL มาตรฐาน ใช้ร่วมกับ Grafana/Superset ได้ทันที

ClickHouse Schema สำหรับ AI API Logs

ผมออกแบบ schema ให้รองรับทั้ง OpenAI-compatible endpoint และ Anthropic-compatible endpoint โดยใช้ MergeTree engine พร้อม partition รายวัน:

-- สร้าง database
CREATE DATABASE IF NOT EXISTS ai_observability;

-- ตารางหลักสำหรับ log ทุกคำขอ
CREATE TABLE IF NOT EXISTS ai_observability.api_requests
(
    request_id        String,
    trace_id          String,
    span_id           String,
    ts                DateTime64(3, 'UTC'),
    user_id           String,
    api_key_hash      String,
    model             LowCardinality(String),
    provider          LowCardinality(String),  -- holy_sheep, openai, anthropic
    endpoint          LowCardinality(String),  -- /v1/chat/completions
    prompt_tokens     UInt32,
    completion_tokens UInt32,
    total_tokens      UInt32,
    cost_usd          Decimal(10, 6),
    latency_ms        UInt32,
    ttfb_ms           UInt32,
    status_code       UInt16,
    error_type        LowCardinality(String) DEFAULT '',
    error_message     String DEFAULT '',
    prompt_hash       String,                 -- SHA256 ของ prompt เพื่อ dedupe
    is_streaming      UInt8,
    is_retry          UInt8,
    user_agent        String DEFAULT '',
    ip_country        LowCardinality(String) DEFAULT 'XX',
    metadata          JSON                    -- field เสริม เช่น temperature, top_p
) ENGINE = MergeTree
PARTITION BY toYYYYMMDD(ts)
ORDER BY (provider, model, ts, request_id)
TTL toDate(ts) + INTERVAL 90 DAY;

-- Materialized view สำหรับสรุปต้นทุนรายชั่วโมง
CREATE MATERIALIZED VIEW IF NOT EXISTS ai_observability.hourly_cost_mv
ENGINE = SummingMergeTree
PARTITION BY toYYYYMMDD(hour)
ORDER BY (provider, model, hour)
AS
SELECT
    toStartOfHour(ts) AS hour,
    provider,
    model,
    count() AS request_count,
    sum(prompt_tokens) AS sum_prompt,
    sum(completion_tokens) AS sum_completion,
    sum(total_tokens) AS sum_total,
    sum(cost_usd) AS total_cost,
    quantile(0.50)(latency_ms) AS p50_latency,
    quantile(0.95)(latency_ms) AS p95_latency,
    quantile(0.99)(latency_ms) AS p99_latency,
    countIf(status_code >= 400) AS error_count
FROM ai_observability.api_requests
GROUP BY hour, provider, model;

Python Middleware: เก็บ Log อัตโนมัติขณะเรียก API

ตัวอย่างนี้เป็น FastAPI middleware ที่ทำหน้าที่ proxy ไปยัง https://api.holysheep.ai/v1 พร้อมเก็บ log ทุก request/response เข้า ClickHouse แบบ async (ใช้ buffering เพื่อไม่ให้กระทบ latency):

import os
import time
import uuid
import json
import hashlib
import asyncio
from typing import Optional
import httpx
import clickhouse_driver
from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse

HOLYSHEEP_BASE = "https://api.holysheep.ai/v1"
HOLYSHEEP_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")

CH_HOST = os.environ.get("CLICKHOUSE_HOST", "localhost")
CH_PORT = int(os.environ.get("CLICKHOUSE_PORT", "9000"))
CH_USER = os.environ.get("CLICKHOUSE_USER", "default")
CH_PASS = os.environ.get("CLICKHOUSE_PASSWORD", "")

PRICING_USD_PER_MTOK = {
    "gpt-4.1": 8.00,
    "claude-sonnet-4.5": 15.00,
    "gemini-2.5-flash": 2.50,
    "deepseek-v3.2": 0.42,
}

app = FastAPI()
ch_client = clickhouse_driver.Client(
    host=CH_HOST, port=CH_PORT, user=CH_USER, password=CH_PASS
)
log_buffer: list[dict] = []
BUFFER_FLUSH_SIZE = 200


def calc_cost_usd(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    rate = PRICING_USD_PER_MTOK.get(model, 5.0)
    return round((prompt_tokens + completion_tokens) * rate / 1_000_000, 6)


async def flush_logs():
    while True:
        await asyncio.sleep(2.0)
        if not log_buffer:
            continue
        batch, log_buffer[:] = log_buffer, []
        try:
            ch_client.execute(
                "INSERT INTO ai_observability.api_requests VALUES",
                batch,
            )
        except Exception as e:
            print(f"[clickhouse] flush failed: {e}")


@app.on_event("startup")
async def startup_event():
    asyncio.create_task(flush_logs())


async def proxy_and_log(request: Request) -> Response:
    body_bytes = await request.body()
    try:
        payload = json.loads(body_bytes)
    except Exception:
        payload = {}
    model = payload.get("model", "unknown")
    is_stream = bool(payload.get("stream", False))
    prompt_hash = hashlib.sha256(body_bytes).hexdigest()[:32]
    request_id = str(uuid.uuid4())
    trace_id = request.headers.get("x-trace-id", request_id)

    headers = {
        "Authorization": f"Bearer {HOLYSHEEP_KEY}",
        "Content-Type": "application/json",
        "X-Request-Id": request_id,
    }
    start = time.perf_counter()
    ttfb_ms = 0
    status_code = 0
    completion_text = ""
    usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}

    async with httpx.AsyncClient(timeout=120.0) as client:
        if is_stream:
            async def stream_gen():
                nonlocal ttfb_ms, status_code, usage, completion_text
                first = True
                async with client.stream(
                    "POST",
                    f"{HOLYSHEEP_BASE}/chat/completions",
                    content=body_bytes,
                    headers=headers,
                ) as r:
                    status_code = r.status_code
                    async for chunk in r.aiter_bytes():
                        if first:
                            ttfb_ms = int((time.perf_counter() - start) * 1000)
                            first = False
                        try:
                            txt = chunk.decode("utf-8", errors="ignore")
                            completion_text += txt
                        except Exception:
                            pass
                        yield chunk
                await _record(request_id, trace_id, model, is_stream, status_code,
                              usage, ttfb_ms, start, prompt_hash, request)

            return StreamingResponse(stream_gen(), media_type="text/event-stream")
        else:
            r = await client.post(
                f"{HOLYSHEEP_BASE}/chat/completions",
                content=body_bytes,
                headers=headers,
            )
            status_code = r.status_code
            ttfb_ms = int((time.perf_counter() - start) * 1000)
            try:
                data = r.json()
                usage = data.get("usage", usage)
            except Exception:
                data = {"raw": r.text
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
deepseek v4 pro bianmanengli terminal bench shicec
gemini 25 pro 1m shangxiawenchuangkouzaizhongzhuan
function calling toolchoice canshuyichangpaichayuz