Prompt Caching + Model Routing: วิธีลดค่า AI API ได้ 90%

ทีมส่วนใหญ่จ่ายค่า AI แพงเกิน 5-10 เท่า ไม่ใช่เพราะเลือก model ผิด — แต่เพราะใช้ model แพงตัวเดียวทำทุกอย่าง แล้วยังส่ง context เดิมซ้ำทุก request

Prompt caching ตัดค่า input ที่ซ้ำซ้อนทิ้ง Model routing ส่ง query ง่ายๆ ไปให้ model ถูก ใช้ทั้งสองอย่างพร้อมกันแล้ว bill ลดได้ 80-90% บทความนี้มี working code ให้ทั้งคู่

Part 1: Prompt Caching เจาะลึก#

วิธีทำงาน

ไม่มี cache ทุก API call จะ process system prompt ใหม่ทั้งหมด prompt 2,000 token คูณ 10,000 request/วัน = 20M input token ที่ process จากศูนย์

มี cache แล้ว provider เก็บ prompt ที่ process ไว้ request ถัดๆ ไปที่ hit cache จ่ายในราคาเศษเสี้ยว:

Without caching:
  Request 1: [System: 2000 tok] + [User: 200 tok] → 2200 input tokens billed
  Request 2: [System: 2000 tok] + [User: 150 tok] → 2150 input tokens billed
  Total: 4,350 tokens at full price

With caching:
  Request 1: [System: 2000 tok → WRITE CACHE] + [User: 200 tok] → 2200 at full price
  Request 2: [System: CACHE HIT] + [User: 150 tok] → 150 full-price + 2000 cached (90% off)
  Total: 2,350 full-price + 2,000 cached tokens

เปรียบเทียบ Provider

Anthropic Implementation

Anthropic ให้ส่วนลดมากที่สุด (90%) แต่ต้อง mark cache เอง TTL 5 นาทีจะ reset ทุกครั้งที่ hit — เหมาะมากสำหรับ app ที่มี traffic เยอะ

import anthropic

client = anthropic.Anthropic()

SYSTEM_PROMPT = """You are a senior code reviewer for a Python codebase.
Review code for: security vulnerabilities, performance issues,
readability problems, and adherence to PEP 8.
Always provide specific line references and suggested fixes.
Rate severity as: critical, warning, or info.
... (imagine 1500+ tokens of detailed instructions here)
"""

def review_code(code_snippet: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=[{
            "type": "text",
            "text": SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"}  # enables caching
        }],
        messages=[
            {"role": "user", "content": f"Review this code:\n```python\n{code_snippet}\n```"}
        ]
    )
    usage = response.usage
    print(f"Input: {usage.input_tokens} | Cache read: {usage.cache_read_input_tokens} | Cache write: {usage.cache_creation_input_tokens}")
    return response.content[0].text

# First call: cache write
result = review_code("def add(a, b): return a + b")
# Input: 1700 | Cache read: 0 | Cache write: 1500

# Second call within 5 min: cache hit — 90% cheaper on cached tokens
result = review_code("def multiply(x, y): return x * y")
# Input: 200 | Cache read: 1500 | Cache write: 0

OpenAI Implementation (อัตโนมัติ)

OpenAI cache ให้อัตโนมัติสำหรับ prompt ที่เกิน 1,024 token ไม่ต้องแก้ code — แค่ verify:

from openai import OpenAI
client = OpenAI()

def query_openai(user_message: str) -> str:
    response = client.chat.completions.create(
        model="gpt-5",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message}
        ]
    )
    cached = getattr(response.usage, "prompt_tokens_details", None)
    if cached:
        print(f"Cached tokens: {cached.cached_tokens}")  # > 0 = cache hit
    return response.choices[0].message.content

DeepSeek (Automatic Prefix Caching)

DeepSeek ให้ส่วนลด 90% ด้วย automatic prefix-based caching ผ่าน disk-based system แค่ให้ system prompt คงที่ — DeepSeek จัดการที่เหลือ:

client = OpenAI(api_key="your-deepseek-key", base_url="https://api.deepseek.com")

def query_deepseek(user_message: str) -> str:
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message}
        ]
    )
    print(f"Cache hit tokens: {getattr(response.usage, 'prompt_cache_hit_tokens', 0)}")
    return response.choices[0].message.content

คำนวณการประหยัดจริง

Part 2: Model Routing เจาะลึก#

ทำไม query ส่วนใหญ่ไม่ต้องการ Frontier Model

70% ของ traffic AI API ทั่วไปเป็นงานง่ายๆ — classification, extraction, reformatting, basic Q&A ส่งพวกนี้ไปให้ GPT-5 หรือ Claude Opus ก็เหมือนจ้าง PhD มา sort จดหมาย

สร้าง Model Router

router นี้จำแนกความซับซ้อนของ query และส่งแต่ละ request ไปให้ tier ที่ถูกต้อง:

import anthropic
from openai import OpenAI
from dataclasses import dataclass
from enum import Enum

class Tier(Enum):
    NANO = "nano"        # Classification, extraction
    MID = "mid"          # Summarization, Q&A
    FRONTIER = "frontier" # Reasoning, code gen, analysis

@dataclass
class ModelConfig:
    provider: str
    model: str
    cost_per_1k_input: float
    cost_per_1k_output: float

MODEL_TIERS: dict[Tier, ModelConfig] = {
    Tier.NANO: ModelConfig("openai", "gpt-5-nano", 0.00010, 0.00040),
    Tier.MID: ModelConfig("deepseek", "deepseek-chat", 0.00014, 0.00028),
    Tier.FRONTIER: ModelConfig("anthropic", "claude-sonnet-4-6", 0.003, 0.015),
}

COMPLEXITY_KEYWORDS = {
    "high": ["analyze", "compare", "debug", "refactor", "architect",
             "design", "optimize", "explain why", "trade-off", "reason"],
    "low": ["classify", "extract", "format", "convert", "translate",
            "summarize briefly", "yes or no", "list the", "parse"],
}

def classify_complexity(query: str) -> Tier:
    query_lower = query.lower()
    word_count = len(query.split())
    high = sum(1 for kw in COMPLEXITY_KEYWORDS["high"] if kw in query_lower)
    low = sum(1 for kw in COMPLEXITY_KEYWORDS["low"] if kw in query_lower)

    if high >= 2 or (word_count > 200 and high >= 1):
        return Tier.FRONTIER
    if low >= 1 and word_count < 50:
        return Tier.NANO
    return Tier.MID

# Provider clients
clients = {
    "anthropic": anthropic.Anthropic(),
    "openai": OpenAI(),
    "deepseek": OpenAI(api_key="deepseek-key", base_url="https://api.deepseek.com"),
}

def route_and_query(query: str, system_prompt: str = "") -> dict:
    tier = classify_complexity(query)
    config = MODEL_TIERS[tier]

    if config.provider == "anthropic":
        resp = clients["anthropic"].messages.create(
            model=config.model, max_tokens=1024,
            system=[{"type": "text", "text": system_prompt,
                     "cache_control": {"type": "ephemeral"}}] if system_prompt else [],
            messages=[{"role": "user", "content": query}]
        )
        text, inp, out = resp.content[0].text, resp.usage.input_tokens, resp.usage.output_tokens
    else:
        resp = clients[config.provider].chat.completions.create(
            model=config.model,
            messages=[*([{"role": "system", "content": system_prompt}] if system_prompt else []),
                      {"role": "user", "content": query}]
        )
        text, inp, out = resp.choices[0].message.content, resp.usage.prompt_tokens, resp.usage.completion_tokens

    cost = inp / 1000 * config.cost_per_1k_input + out / 1000 * config.cost_per_1k_output
    return {"tier": tier.value, "model": config.model, "response": text, "cost": cost}

# Simple extraction → nano ($0.0001/1K tokens)
result = route_and_query("Extract all email addresses from this text: ...")
# Routed to: nano (gpt-5-nano) — Cost: $0.000024

# Complex reasoning → frontier
result = route_and_query("Analyze the trade-offs between microservices and monolith and design an architecture...")
# Routed to: frontier (claude-sonnet-4-6) — Cost: $0.018500

Cost-per-Success Framework

ราคาต่อ token ตรงๆ มันหลอก model ถูกที่ fail 40% ของเวลาจริงๆ แล้วแพงกว่า model แพงที่ทำสำเร็จทุกครั้ง ใช้ Cost-per-Success (CPS):

CPS = total_cost / successful_outputs

from dataclasses import dataclass, field

@dataclass
class CostPerSuccessTracker:
    results: dict = field(default_factory=lambda: {
        "nano": {"cost": 0.0, "success": 0, "total": 0},
        "mid": {"cost": 0.0, "success": 0, "total": 0},
        "frontier": {"cost": 0.0, "success": 0, "total": 0},
    })

    def record(self, tier: str, cost: float, success: bool):
        self.results[tier]["cost"] += cost
        self.results[tier]["total"] += 1
        if success:
            self.results[tier]["success"] += 1

    def cps(self, tier: str) -> float:
        r = self.results[tier]
        return r["cost"] / r["success"] if r["success"] > 0 else float("inf")

    def report(self):
        for tier, r in self.results.items():
            rate = r["success"] / r["total"] * 100 if r["total"] else 0
            print(f"{tier:<10} {r['total']:>5} reqs | {rate:.0f}% success | CPS: ${self.cps(tier):.6f}")

หลังรัน 1,000 mixed query:

Part 3: ใช้ทั้งสองเทคนิคร่วมกัน#

Routing อย่างเดียวประหยัด 70% Caching อย่างเดียวประหยัด 75% ใช้ด้วยกันแล้วมัน compound:

การ implement ตรงไปตรงมา — ใช้ router จาก Part 2 แล้วเพิ่ม cache_control ให้ทุก Anthropic call (แสดงใน route_and_query ข้างบนแล้ว) OpenAI และ DeepSeek cache ให้อัตโนมัติ

Part 4: Batch API สำหรับงาน Offline#

ไม่ใช่ทุกอย่างต้องการ response แบบ real-time Batch API ให้ส่วนลด 50% สำหรับ async processing:

from openai import OpenAI
import json

client = OpenAI()

def submit_batch(queries: list[str], system_prompt: str) -> str:
    # Build JSONL batch file
    requests = [
        {"custom_id": f"req-{i}", "method": "POST", "url": "/v1/chat/completions",
         "body": {"model": "gpt-5-mini", "max_tokens": 512,
                  "messages": [{"role": "system", "content": system_prompt},
                               {"role": "user", "content": q}]}}
        for i, q in enumerate(queries)
    ]
    with open("/tmp/batch.jsonl", "w") as f:
        for r in requests:
            f.write(json.dumps(r) + "\n")

    batch_file = client.files.create(file=open("/tmp/batch.jsonl", "rb"), purpose="batch")
    job = client.batches.create(
        input_file_id=batch_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h"
    )
    print(f"Batch {job.id} submitted — 50% cheaper, results within 24h")
    return job.id

def get_results(batch_id: str) -> list[dict] | None:
    batch = client.batches.retrieve(batch_id)
    if batch.status == "completed":
        content = client.files.content(batch.output_file_id)
        return [json.loads(line) for line in content.text.strip().split("\n")]
    print(f"Status: {batch.status}")
    return None

Part 5: Cost Tracking — พิสูจน์การประหยัดของคุณ#

import json
from datetime import datetime
from collections import defaultdict

class CostTracker:
    def __init__(self):
        self.records = defaultdict(lambda: {
            "requests": 0, "cost": 0.0, "cache_savings": 0.0, "routing_savings": 0.0
        })

    def record(self, model: str, cost: float, cache_savings: float = 0, routing_savings: float = 0):
        key = f"{datetime.now():%Y-%m-%d}:{model}"
        self.records[key]["requests"] += 1
        self.records[key]["cost"] += cost
        self.records[key]["cache_savings"] += cache_savings
        self.records[key]["routing_savings"] += routing_savings

    def summary(self) -> dict:
        total_cost = sum(v["cost"] for v in self.records.values())
        saved_cache = sum(v["cache_savings"] for v in self.records.values())
        saved_route = sum(v["routing_savings"] for v in self.records.values())
        reqs = sum(v["requests"] for v in self.records.values())
        baseline = total_cost + saved_cache + saved_route
        return {
            "total_cost": round(total_cost, 2),
            "total_savings": round(saved_cache + saved_route, 2),
            "effective_discount": f"{(saved_cache + saved_route) / max(baseline, 0.01) * 100:.1f}%",
            "total_requests": reqs,
            "avg_cost_per_request": round(total_cost / max(reqs, 1), 6),
        }

tracker = CostTracker()
# After a day of traffic:
# {"total_cost": 15.42, "total_savings": 128.76, "effective_discount": "89.3%", ...}

สรุปย่อ#

แหล่งข้อมูล#

1. Anthropic — Prompt Caching docs — ส่วนลด 90%, TTL 5 นาที
2. OpenAI — Prompt Caching guide — อัตโนมัติ, ส่วนลด 50%
3. DeepSeek — KV Cache docs — Automatic prefix caching, ส่วนลด 90%
4. Google — Context Caching for Gemini — TTL ตั้งเองได้, ส่วนลด 90%
5. OpenAI — Batch API reference — ส่วนลด 50%, window 24 ชม.
6. Anthropic — Message Batches API — ส่วนลด 50%, window 24 ชม.
7. TokenTab — Live Model Pricing — ราคา real-time สำหรับ model 1,800+ ตัว