Khi mình bắt đầu vận hành một hệ thống chatbot phục vụ hơn 12.000 người dùng mỗi ngày, mình nhận ra rằng không một mô hình đơn lẻ nào có thể đáp ứng mọi ngữ cảnh: Grok (xAI) xử lý lý luận và code tốt nhưng độ trễ ở châu Á chưa ổn định, còn MiniMax M2.7 lại phản hồi tiếng Việt rất mượt với chi phí rất thấp. Thay vì chọn một bên, mình xây một gateway nhỏ ở tầng giữa để định tuyến theo từng route, từng ngôn ngữ và từng ngân sách token. Bài viết này chia sẻ lại toàn bộ kiến trúc, code chạy được và những lỗi mình đã đốt hơn 600.000 VND tiền token mới rút ra được.

Bảng so sánh 5 phương án triển khai

Phương án Độ trễ trung bình (ms) Chi phí/tháng (≈10M token) Độ khó vận hành Khả năng tùy biến routing
Custom Python gateway (tự code) 180 – 420 ≈ $42 (Grok) + $9 (M2.7) Trung bình Rất cao
LiteLLM proxy self-host 210 – 460 ≈ $55 (bao gồm server) Trung bình – Khó Cao
Portkey.ai (managed) 240 – 510 ≈ $70 + phí SaaS Thấp Cao
OpenRouter (chỉ routing) 300 – 700 ≈ $90 (có markup 5%) Thấp Trung bình
Cloudflare Worker + KV cache 90 – 250 ≈ $5 (free tier đủ) Trung bình Rất cao

Dữ liệu đo từ log thật trong 14 ngày của mình (workload 60% tiếng Việt, 30% tiếng Anh, 10% code review). Bạn có thể thấy Cloudflare Worker nổi bật về chi phí, còn custom Python thắng về độ tùy biến.

Kiến trúc gateway mình đang chạy

Code 1 - Provider adapter thống nhất cho Grok và MiniMax M2.7

from dataclasses import dataclass
from typing import Iterator
import httpx, os, time

@dataclass
class ChatRequest:
    messages: list
    temperature: float = 0.7
    max_tokens: int = 1024
    route_hint: str = "default"

class BaseProvider:
    name: str = "base"
    def chat(self, req: ChatRequest) -> dict: ...
    def stream(self, req: ChatRequest) -> Iterator[str]: ...

class GrokProvider(BaseProvider):
    name = "grok-3"
    def __init__(self):
        self.base = os.environ["GROK_BASE_URL"]   # ví dụ: https://api.x.ai/v1
        self.key = os.environ["GROK_API_KEY"]

    def chat(self, req: ChatRequest) -> dict:
        with httpx.Client(timeout=30) as cli:
            r = cli.post(f"{self.base}/chat/completions",
                headers={"Authorization": f"Bearer {self.key}"},
                json={"model": self.name,
                      "messages": req.messages,
                      "temperature": req.temperature,
                      "max_tokens": req.max_tokens})
            r.raise_for_status()
            return r.json()

class M27Provider(BaseProvider):
    name = "M2-7-chat"
    def __init__(self):
        self.base = os.environ["M27_BASE_URL"]
        self.key = os.environ["M27_API_KEY"]

    def chat(self, req: ChatRequest) -> dict:
        with httpx.Client(timeout=30) as cli:
            r = cli.post(f"{self.base}/chat/completions",
                headers={"Authorization": f"Bearer {self.key}"},
                json={"model": self.name,
                      "messages": req.messages,
                      "temperature": req.temperature,
                      "max_tokens": req.max_tokens})
            r.raise_for_status()
            return r.json()

Điểm mấu chốt: hai class có cùng interface, vì vậy router không cần biết nó đang nói với provider nào. Khi mình muốn đổi sang model mới, mình chỉ thêm một class mới, sửa một dòng trong registry.

Code 2 - Router thông minh có fallback và ngân sách

PRIORITY = {
    "vi":    ["M2-7-chat", "grok-3"],
    "en":    ["grok-3", "M2-7-chat"],
    "code":  ["grok-3", "M2-7-chat"],
    "math":  ["grok-3", "M2-7-chat"],
    "default":["M2-7-chat", "grok-3"],
}

COST_PER_M = {"grok-3": 3.0, "M2-7-chat": 0.42}  # USD / 1M output token

def pick_model(req: ChatRequest, budget_left: float) -> str:
    for cand in PRIORITY[req.route_hint]:
        est_tokens = max_tokens_estimate(req.messages)
        if COST_PER_M[cand] * est_tokens / 1_000_000 <= budget_left:
            return cand
    return PRIORITY[req.route_hint][-1]  # vẫn trả lời để không drop traffic

def route(req: ChatRequest, providers: dict, budget: float):
    chosen = pick_model(req, budget)
    try:
        return providers[chosen].chat(req), chosen
    except (httpx.HTTPStatusError, httpx.TimeoutException) as e:
        # fallback sang model tiếp theo trong chuỗi ưu tiên
        for backup in PRIORITY[req.route_hint][PRIORITY[req.route_hint].index(chosen)+1:]:
            try:
                return providers[backup].chat(req), backup
            except Exception:
                continue
        raise RuntimeError(f"All providers failed: {e}")

Mình thiết kế hàm pick_model theo hai tiêu chí: ngôn ngữ/ngữ cảnh (qua route_hint) và ngân sách còn lại của tenant trong tháng. Nếu mọi provider đều fail, mình vẫn trả về lỗi có cấu trúc để frontend hiển thị retry, thay vì để request treo.

Code 3 - Cấu hình routing nâng cao với circuit breaker

import time
from collections import deque

class CircuitBreaker:
    def __init__(self, fail_threshold=5, cooldown_sec=60):
        self.fail_threshold = fail_threshold
        self.cooldown = cooldown_sec
        self.window = deque()
        self.opened_at = None

    def allow(self) -> bool:
        if self.opened_at and time.time() - self.opened_at < self.cooldown:
            return False
        if self.opened_at and time.time() - self.opened_at >= self.cooldown:
            self.opened_at = None
            self.window.clear()
        return True

    def record(self, ok: bool):
        self.window.append((time.time(), ok))
        while self.window and time.time() - self.window[0][0] > 60:
            self.window.popleft()
        fails = sum(1 for _, ok in self.window if not ok)
        if fails >= self.fail_threshold:
            self.opened_at = time.time()

breakers = {"grok-3": CircuitBreaker(), "M2-7-chat": CircuitBreaker()}

def safe_call(name, fn, req):
    if not breakers[name].allow():
        raise RuntimeError(f"circuit open: {name}")
    try:
        res = fn(req); breakers[name].record(True); return res
    except Exception:
        breakers[name].record(False); raise

Circuit breaker tránh tình trạng mình từng gặp: Grok ở khu vực EU bị quá tải 4 giờ liên tục, gateway lúc đó gọi đi gọi lại làm cháy thêm 8% quota. Bây giờ sau 5 lần lỗi trong 60 giây, breaker tự ngắt 1 phút rồi thử lại.

Đo đạc thực tế và phản hồi cộng đồng

Phù hợp / không phù hợp với ai?

Nên dùng custom gateway khi