เปิดเรื่องด้วยเหตุการณ์จริงใน Production

เมื่อเดือนที่แล้ว ระบบแชทบอทของลูกค้ารายหนึ่งเกิดอาการ "ค้าง" ทุกช่วง 14:00–16:00 ของวันทำงาน หน้าจอมอนิเตอร์เต็มไปด้วย log สีแดงที่ผมเห็นแล้วใจหาย:

openai.APITimeoutError: Request timed out (timeout=30s)
  File "gateway/router.py", line 142, in route_request
    response = await client.chat.completions.create(model="gpt-4.1", ...)

anthropic.RateLimitError: 429 - Too Many Requests
  File "gateway/fallback.py", line 78, in handle_fallback
    return await anthropic_client.messages.create(model="claude-sonnet-4.5", ...)

ทีมของผมใช้เวลาเกือบ 6 ชั่วโมงในการวิเคราะห์จนพบว่า ปัญหาไม่ได้อยู่ที่โมเดลใดโมเดลหนึ่ง แต่เป็นเพราะเราส่งทุก request ไปที่ GPT-4.1 ตัวเดียว ทั้งที่ในช่วงเวลาเดียวกัน Claude Sonnet 4.5 ว่างอยู่ และ Gemini 2.5 Flash ตอบกลับใน 178ms ขณะที่ GPT-4.1 ตอบ 2,180ms บทความนี้คือบทเรียนที่ผมแปลงเป็น API Gateway แบบ dual-objective (latency + cost) ที่ใช้งานจริงในโปรเจกต์ปัจจุบัน โดยใช้ endpoint เดียวจาก HolySheep AI ซึ่งรวม Claude/GPT/Gemini/DeepSeek ไว้ใน base_url เดียวกัน (latency <50ms ที่ edge, อัตรา ¥1=$1 ประหยัด 85%+ รองรับ WeChat/Alipay)

สถาปัตยกรรม Gateway แบบ Dual-Objective Routing

แนวคิดคือ ทุก request เข้ามาที่ Gateway ตัวเดียว แล้ว Gateway จะตัดสินใจว่าจะส่งไปที่ผู้ให้บริการรายใด โดยใช้ "คะแนน" ที่ผสมระหว่าง:

โค้ด Gateway ฉบับ Production-Ready (คัดลอกและรันได้)

ตัวอย่างแรกคือคลาส SmartRouter ที่คำนวณคะแนนแบบ normalized แล้วเลือก provider ที่เหมาะสมที่สุด พร้อม fallback อัตโนมัติ:

import time
import asyncio
import httpx
from dataclasses import dataclass, field
from collections import deque
from typing import Dict, List
from openai import AsyncOpenAI

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY  = "YOUR_HOLYSHEEP_API_KEY"

@dataclass
class ProviderStats:
    name: str
    cost_per_mtok: float
    success_rate: float = 1.0
    latency_window: deque = field(default_factory=lambda: deque(maxlen=200))

    def record(self, ms: float):
        self.latency_window.append((time.time(), ms))

    @property
    def p95_latency_ms(self) -> float:
        if not self.latency_window: return 200.0
        vals = sorted(v for _, v in self.latency_window)
        return vals[int(len(vals) * 0.95)]

@dataclass
class RouteRequest:
    task_type: str          # "coding" | "vision" | "chat" | "budget"
    estimated_tokens: int = 1000

class SmartRouter:
    PROVIDERS = {
        "gpt-4.1":           ProviderStats("gpt-4.1",           8.00),
        "claude-sonnet-4.5": ProviderStats("claude-sonnet-4.5", 15.00),
        "gemini-2.5-flash":  ProviderStats("gemini-2.5-flash",  2.50),
        "deepseek-v3.2":     ProviderStats("deepseek-v3.2",     0.42),
    }
    PREFERENCE = {
        "coding": ["claude-sonnet-4.5", "gpt-4.1", "deepseek-v3.2"],
        "vision": ["gemini-2.5-flash",  "gpt-4.1", "claude-sonnet-4.5"],
        "chat":   ["gpt-4.1",           "claude-sonnet-4.5", "gemini-2.5-flash"],
        "budget": ["deepseek-v3.2",     "gemini-2.5-flash",  "gpt-4.1"],
    }

    def __init__(self, w_latency: float = 0.6, w_cost: float = 0.4):
        self.w_latency, self.w_cost = w_latency, w_cost
        self.client = AsyncOpenAI(
            base_url=BASE_URL, api_key=API_KEY,
            timeout=httpx.Timeout(8.0, connect=3.0), max_retries=2,
        )

    def score(self, s: ProviderStats) -> float:
        return (self.w_latency * (s.p95_latency_ms / 3000.0)
              + self.w_cost    * (s.cost_per_mtok / 15.0)) / max(s.success_rate, 0.01)

    def pick(self, req: RouteRequest) -> str:
        prefs = self.PREFERENCE.get(req.task_type, list(self.PROVIDERS.keys()))
        return sorted(self.PROVIDERS[m] for m in prefs, key=self.score)[0].name

    async def chat(self, req: RouteRequest, messages: list):
        chain = self.PREFERENCE.get(req.task_type, list(self.PROVIDERS.keys()))
        for i, model in enumerate(chain):
            t0 = time.perf_counter()
            try:
                resp = await self.client.chat.completions.create(
                    model=model, messages=messages

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