Tuần này trên HN có một bài gây chú ý: một công ty lớn tiết kiệm hàng triệu đô nhờ không còn gửi mọi request tới model đắt nhất, mà dùng một lớp "decider" để chọn model phù hợp. Đọc xong mình thấy rất quen. Team mình cũng từng gửi 100% traffic tới model flagship, kể cả những câu đơn giản như "tóm tắt email này trong 1 câu" hay "trích số điện thoại từ đoạn text". Cuối tháng nhìn bill thì mới giật mình.

Trong bài này mình chia sẻ cách tự build một LLM router đơn giản bằng Python: request dễ thì gửi model nhỏ, request khó thì gửi model lớn, và có fallback khi model nhỏ trả lời không đạt. Không cần framework nặng nề, khoảng 150 dòng code là chạy được production.

Vì sao routing tiết kiệm được nhiều đến vậy?

Nếu nhìn vào log thật của các sản phẩm dùng LLM, bạn sẽ thấy phân bố request thường lệch rất mạnh:

  • Khoảng 60-80% là task đơn giản: classification, extraction, rewrite ngắn, trả lời FAQ.
  • Khoảng 15-30% là task trung bình: tóm tắt tài liệu dài, viết code nhỏ.
  • Chỉ 5-10% thật sự cần reasoning sâu: debug phức tạp, phân tích nhiều bước.

Giá giữa model nhỏ và model flagship thường chênh nhau 10-30 lần mỗi token. Chỉ cần đẩy được 70% traffic sang model nhỏ mà chất lượng không giảm đáng kể, chi phí đã giảm khoảng 60-70%. Latency cũng giảm theo, vì model nhỏ nhanh hơn hẳn.

Kiến trúc tổng quát trông như sau:

flowchart LR A[Client request] --> B[Router] B -->|easy| C[Small model] B -->|hard| D[Large model] C --> E{Validator pass?} E -->|yes| F[Response] E -->|no| D D --> F

Có 3 thành phần chính: Router quyết định gửi đi đâu, Validator kiểm tra output của model nhỏ, và Fallback (escalate) lên model lớn khi cần.

Bước 1: Router dựa trên heuristic (đừng vội dùng ML)

Sai lầm phổ biến là nhảy ngay vào train một classifier. Thực tế, heuristic đơn giản đã xử lý tốt phần lớn trường hợp, và quan trọng hơn là dễ debug. Mình bắt đầu với rule-based, sau đó mới thay bằng classifier khi đã có đủ data.

# router.py - Python 3.11+ from dataclasses import dataclass from enum import Enum class Tier(str, Enum): SMALL = 'small' LARGE = 'large' # Task type do caller khai báo; đây là tín hiệu mạnh nhất SIMPLE_TASKS = {'classify', 'extract', 'translate_short', 'faq'} HARD_KEYWORDS = ('debug', 'tại sao', 'why', 'phân tích', 'refactor', 'prove') @dataclass class RouteDecision: tier: Tier reason: str def route(task_type: str, prompt: str, history_len: int = 0) -> RouteDecision: tokens_est = len(prompt) // 4 # ước lượng thô, đủ dùng cho routing if task_type in SIMPLE_TASKS and tokens_est < 2000: return RouteDecision(Tier.SMALL, f'simple task: {task_type}') if tokens_est > 30_000: return RouteDecision(Tier.LARGE, 'long context') if history_len > 10: return RouteDecision(Tier.LARGE, 'long multi-turn conversation') lowered = prompt.lower() if any(k in lowered for k in HARD_KEYWORDS): return RouteDecision(Tier.LARGE, 'hard keyword detected') # Mặc định: thử model nhỏ trước, validator sẽ escalate nếu cần return RouteDecision(Tier.SMALL, 'default to small')

Điểm quan trọng: luôn trả về reason. Khi log lại, bạn sẽ biết rule nào đang route sai, chứ không phải đoán mò.

Bước 2: Cascade với validator, an toàn hơn routing thuần

Routing một lần rồi thôi thì rủi ro: router đoán sai thì user nhận câu trả lời tệ. Pattern an toàn hơn là cascade: thử model nhỏ, kiểm tra output, không đạt thì escalate.

Validator không cần thông minh. Với structured output, chỉ cần validate schema bằng Pydantic v2 là đã bắt được phần lớn lỗi:

# cascade.py import json import time import httpx from pydantic import BaseModel, ValidationError MODELS = { 'small': {'url': 'http://localhost:8000/v1/chat/completions', 'name': 'small-model'}, 'large': {'url': 'http://localhost:8001/v1/chat/completions', 'name': 'large-model'}, } class Invoice(BaseModel): vendor: str total: float currency: str def call_llm(tier: str, prompt: str, timeout: float = 30.0) -> str: cfg = MODELS[tier] resp = httpx.post(cfg['url'], timeout=timeout, json={ 'model': cfg['name'], 'messages': [{'role': 'user', 'content': prompt}], 'temperature': 0, }) resp.raise_for_status() return resp.json()['choices'][0]['message']['content'] def validate(raw: str) -> Invoice | None: try: data = json.loads(raw) inv = Invoice.model_validate(data) if inv.total < 0: # business rule đơn giản return None return inv except (json.JSONDecodeError, ValidationError): return None def extract_invoice(text: str) -> tuple[Invoice, dict]: prompt = f'Trích xuất JSON {{vendor, total, currency}} từ:\n{text}' meta = {'escalated': False} t0 = time.perf_counter() result = validate(call_llm('small', prompt)) meta['small_ms'] = round((time.perf_counter() - t0) * 1000) if result is None: meta['escalated'] = True result = validate(call_llm('large', prompt)) if result is None: raise ValueError('Both tiers failed validation') return result, meta

Với task dạng free text (tóm tắt, trả lời câu hỏi), validator có thể là: độ dài output có hợp lý không, có chứa các cụm như "I'm not sure" hay "không đủ thông tin" không, hoặc dùng chính model nhỏ làm judge chấm điểm 1-5. Đừng dùng model lớn làm judge cho mọi request, vì như vậy là mất luôn khoản tiết kiệm.

Luồng xử lý một request cụ thể:

sequenceDiagram participant C as Client participant R as Router participant S as Small model participant L as Large model C->>R: extract_invoice R->>S: prompt S-->>R: JSON lỗi schema R->>L: escalate cùng prompt L-->>R: JSON hợp lệ R-->>C: Invoice + meta escalated=true

Bước 3: Đo lường, vì không đo thì đừng tối ưu

Router chỉ tốt khi bạn biết nó đang làm gì. Mình log mỗi request ra một dòng JSONL gồm tier, reason, escalated, latency, input_tokens, output_tokens. Sau đó dùng jq để xem nhanh mà không cần dựng dashboard:

# Tỷ lệ escalate theo reason: reason nào escalate nhiều thì rule đó đang sai jq -r 'select(.tier=="small") | [.reason, .escalated] | @tsv' logs/llm-*.jsonl \ | sort | uniq -c | sort -rn | head -20 # Phân bố traffic giữa các tier jq -r '.final_tier' logs/llm-*.jsonl | sort | uniq -c # p95 latency của small tier (cần datamash: apt install datamash) jq -r 'select(.final_tier=="small") | .latency_ms' logs/llm-*.jsonl \ | datamash perc:95 1

Con số quan trọng nhất là escalation rate. Nếu một reason nào đó escalate trên 30%, nghĩa là route loại request đó sang model nhỏ đang lãng phí: bạn trả tiền 2 lần và user chờ lâu hơn. Lúc đó nên chuyển thẳng loại request này sang large.

Ngoài ra, hãy giữ một golden set khoảng 200-500 request thật đã được gán nhãn kết quả đúng. Mỗi lần sửa rule hoặc đổi model, chạy lại golden set và so sánh accuracy cùng cost. Nếu thiếu bước này, bạn sẽ không biết mình tiết kiệm tiền hay đang âm thầm làm giảm chất lượng sản phẩm.

Những cái bẫy mình đã dính

  • Prompt tối ưu cho model lớn không chạy tốt trên model nhỏ. Model nhỏ cần instruction rõ ràng hơn, thêm few-shot example, và output format chặt hơn. Nên tách prompt template theo tier.
  • Escalate làm tăng latency tail. Request bị escalate sẽ có latency bằng small + large. Với các flow cần realtime, đặt timeout ngắn cho small (ví dụ 3-5 giây) và escalate ngay khi timeout.
  • Cache theo tier. Nếu dùng semantic cache, đừng để response từ model nhỏ (chưa validate) lọt vào cache rồi phục vụ cho request cần model lớn.
  • Model thay đổi thì rule cũng phải thay đổi. Mỗi khi provider ra phiên bản mới (tuần nào cũng có), model nhỏ thế hệ mới có thể làm được việc mà trước đây phải dùng model lớn. Chạy lại golden set để cập nhật ngưỡng route.

Kết luận

LLM routing không phải kỹ thuật cao siêu. Đây là bài toán engineering cơ bản: phân loại workload rồi chọn công cụ có chi phí phù hợp. Checklist để bắt đầu ngay tuần này:

  1. Log trước, route sau: thêm task_type và token count vào log của mọi LLM call trong 1 tuần để biết phân bố thật.
  2. Bắt đầu bằng heuristic với task_type do caller khai báo. Đây là tín hiệu rẻ và chính xác nhất.
  3. Dùng cascade thay vì route một lần: model nhỏ, rồi validator (Pydantic schema là đủ cho structured output), rồi mới escalate.
  4. Theo dõi escalation rate theo từng reason và chuyển thẳng sang large những loại request escalate trên 30%.
  5. Xây golden set và chạy lại mỗi khi đổi rule hoặc đổi model.
  6. Chỉ cân nhắc train classifier (ví dụ logistic regression trên embedding) khi đã có vài nghìn request được gán nhãn từ log.

Với team nhỏ ở Việt Nam, nơi bill API tính bằng USD còn doanh thu tính bằng VND, giảm 50-70% chi phí LLM có thể là khác biệt giữa một feature lời và một feature lỗ. Bắt đầu từ 150 dòng code ở trên, đo kỹ, rồi tối ưu dần.