Menguji smart model router di pipeline Continuous Integration (CI) sering terkendala flakiness, non-determinisme respons upstream, dan pemborosan biaya token akibat ketergantungan pada API live (seperti OpenAI, Anthropic, atau provider lokal). Routing engine pada dasarnya adalah sistem pengambilan keputusan berbasis aturan dan metadata: mengevaluasi kompleksitas prompt, estimasi token, batasan latensi, dan ketersediaan provider untuk menentukan downstream target terbaik.

Pengujian komponen ini harus 100% deterministik, berjalan di memori runner CI tanpa network call eksternal, dan memverifikasi bahwa keputusan dispatch konsisten terhadap skenario beban kerja dan kegagalan upstream.

Arsitektur Router & Strategi Isolasi Environment

Smart router memisahkan fase evaluasi kebijakan routing dari fase transport eksekusi. Isolasi environment dilakukan dengan mengabstraksi transport layer menggunakan mock dispatcher berbasis contract interface. Dispatcher tiruan ini mencatat riwayat dispatch (audit trail) dan mengembalikan respons terprogram tanpa payload LLM sebenarnya.

Prinsip isolasi di CI:

  • Zero Outbound Network: Blokir soket jaringan eksternal pada runner untuk menjamin tidak ada kebocoran request ke endpoint produksi.
  • State Injection: Injeksi status sirkuit (health, cooldown, error rate) secara langsung ke in-memory storage router daripada mengandalkan akumulasi traffic runtime.
  • Frozen Clock: Gunakan clock mocking untuk pengujian TTL cache, timeout, dan cooldown circuit breaker agar deterministik hingga satuan milidetik.

1. Setup Mock Dispatcher dan Golden Fixture Dataset

Golden fixtures adalah representasi statis dari prompt input nyata yang telah diberi label rute target yang diharapkan. Dataset ini memverifikasi bahwa regresi heuristik tidak terjadi ketika aturan router diperbarui.

from dataclasses import dataclass
from enum import Enum
import time
from typing import Dict, Optional

class ModelTier(str, Enum):
    TIER_1_FAST = "tier-1-fast"       # e.g., lightweight/edge model
    TIER_2_BALANCED = "tier-2-balanced" # e.g., general purpose model
    TIER_3_REASONING = "tier-3-reasoning" # e.g., complex reasoning model

@dataclass
class PromptRequest:
    prompt: str
    estimated_tokens: int
    max_latency_ms: int
    require_reasoning: bool = False

@dataclass
class RouteDecision:
    selected_tier: ModelTier
    provider_id: str
    reason: str

class MockDispatcher:
    def __init__(self):
        self.dispatched_routes = []
        self.provider_responses: Dict[str, dict] = {}

    def set_response(self, provider_id: str, status_code: int, latency_ms: float):
        self.provider_responses[provider_id] = {
            "status_code": status_code,
            "latency_ms": latency_ms
        }

    def dispatch(self, decision: RouteDecision, payload: PromptRequest) -> dict:
        self.dispatched_routes.append((decision, payload))
        res = self.provider_responses.get(decision.provider_id, {"status_code": 200, "latency_ms": 5.0})
        if res["status_code"] != 200:
            raise RuntimeError(f"Upstream provider {decision.provider_id} failed with HTTP {res['status_code']}")
        return {"status": "ok", "tier": decision.selected_tier.value}

Fixture pengujian didefinisikan dalam JSON statis untuk memvalidasi akurasi klasifikasi tanpa biaya token:

[
  {
    "test_case": "simple_classification",
    "prompt": "Klasifikasikan sentimen: 'Aplikasi ini responsif'",
    "estimated_tokens": 25,
    "max_latency_ms": 500,
    "require_reasoning": false,
    "expected_tier": "tier-1-fast"
  },
  {
    "test_case": "code_synthesis_reasoning",
    "prompt": "Tulis parser compiler PEG dalam Rust dengan zero-copy.",
    "estimated_tokens": 1200,
    "max_latency_ms": 4000,
    "require_reasoning": true,
    "expected_tier": "tier-3-reasoning"
  }
]

2. Boundary Testing: Token Context, Latency Budget, & Fallback

Logika router harus tahan terhadap edge case pada batas limit konfigurasi model, seperti ukuran context window yang hampir meluap atau target latensi yang sangat ketat.

class RouterEngine:
    def __init__(self, provider_health: Optional[Dict[str, bool]] = None):
        self.provider_health = provider_health or {
            "provider-fast": True,
            "provider-balanced": True,
            "provider-reasoning": True
        }

    def route(self, req: PromptRequest) -> RouteDecision:
        # Boundary 1: Kebutuhan eksplisit reasoning
        if req.require_reasoning:
            if self.provider_health.get("provider-reasoning"):
                return RouteDecision(ModelTier.TIER_3_REASONING, "provider-reasoning", "explicit_reasoning_required")
            # Fallback ke tier-2 jika tier-3 unhealthy
            return RouteDecision(ModelTier.TIER_2_BALANCED, "provider-balanced", "fallback_reasoning_unavailable")

        # Boundary 2: Token context limit (> 8000 tokens dialihkan ke model kapasitas besar)
        if req.estimated_tokens > 8000:
            return RouteDecision(ModelTier.TIER_2_BALANCED, "provider-balanced", "high_token_context")

        # Boundary 3: Budget latency rendah (< 300ms)
        if req.max_latency_ms < 300 and self.provider_health.get("provider-fast"):
            return RouteDecision(ModelTier.TIER_1_FAST, "provider-fast", "low_latency_budget")

        # Default fallback
        if self.provider_health.get("provider-balanced"):
            return RouteDecision(ModelTier.TIER_2_BALANCED, "provider-balanced", "default_tier")
        return RouteDecision(ModelTier.TIER_1_FAST, "provider-fast", "emergency_fallback")

Uji skenario boundary edge case menggunakan test runner (misal: pytest):

import pytest

def test_context_window_boundary_dispatch():
    engine = RouterEngine()
    
    # Tepat di bawah boundary 8000 token
    req_sub = PromptRequest(prompt="text", estimated_tokens=8000, max_latency_ms=200)
    decision_sub = engine.route(req_sub)
    assert decision_sub.selected_tier == ModelTier.TIER_1_FAST
    
    # Tepat di atas boundary 8000 token (memaksa alokasi ke model kapasitas seimbang)
    req_super = PromptRequest(prompt="text", estimated_tokens=8001, max_latency_ms=200)
    decision_super = engine.route(req_super)
    assert decision_super.selected_tier == ModelTier.TIER_2_BALANCED

3. Verifikasi Circuit Breaker (Timeout & HTTP 429)

Smart router wajib mendeteksi upstream degradasi secara deterministik. Ketika provider utama memicu error 429 (Rate Limited) atau timeout, circuit breaker harus berpindah state (CLOSED ke OPEN) dan mendispatch request ke provider cadangan.

class CircuitBreakerOpenException(Exception):
    pass

class CircuitBreaker:
    def __init__(self, threshold: int = 3, cooldown_seconds: float = 60.0):
        self.threshold = threshold
        self.cooldown_seconds = cooldown_seconds
        self.failure_count = 0
        self.state = "CLOSED"
        self.last_failure_time = 0.0

    def record_failure(self, current_time: float):
        self.failure_count += 1
        self.last_failure_time = current_time
        if self.failure_count >= self.threshold:
            self.state = "OPEN"

    def allow_execution(self, current_time: float) -> bool:
        if self.state == "CLOSED":
            return True
        if self.state == "OPEN":
            if current_time - self.last_failure_time >= self.cooldown_seconds:
                self.state = "HALF-OPEN"
                return True
            return False
        return True  # HALF-OPEN

    def record_success(self):
        self.failure_count = 0
        self.state = "CLOSED"

def test_circuit_breaker_trip_on_consecutive_429():
    cb = CircuitBreaker(threshold=2, cooldown_seconds=30.0)
    simulated_now = 1000.0

    assert cb.allow_execution(simulated_now) is True

    # Error 429 pertama
    cb.record_failure(simulated_now)
    assert cb.state == "CLOSED"
    assert cb.allow_execution(simulated_now) is True

    # Error 429 kedua (mencapai threshold)
    cb.record_failure(simulated_now + 1.0)
    assert cb.state == "OPEN"
    
    # Request berikutnya harus ditolak oleh sirkuit
    assert cb.allow_execution(simulated_now + 2.0) is False

    # Setelah cooldown terlewati, sirkuit masuk ke HALF-OPEN
    assert cb.allow_execution(simulated_now + 31.0) is True
    assert cb.state == "HALF-OPEN"

4. Benchmark Routing Overhead (<10ms)

Router engine menambahkan overhead latensi di depan setiap request. Evaluasi aturan, parsing token, dan pemilihan endpoint harus selesai dalam durasi < 10ms pada lingkungan CI agar tidak mendegradasi pipeline end-to-end.

Implementasi benchmark regression test tanpa dependensi framework benchmark eksternal:

def test_routing_overhead_sla():
    engine = RouterEngine()
    req = PromptRequest(
        prompt="Refactor this function to be thread-safe",
        estimated_tokens=350,
        max_latency_ms=800,
        require_reasoning=False
    )
    
    # Warm-up phase untuk meminimalkan cold-cache artifacts
    for _ in range(100):
        _ = engine.route(req)

    iterations = 5000
    start_ns = time.perf_counter_ns()
    
    for _ in range(iterations):
        decision = engine.route(req)
        assert decision is not None

    total_time_ms = (time.perf_counter_ns() - start_ns) / 1_000_000
    avg_latency_ms = total_time_ms / iterations

    # Assertion SLA: Overhead per-route wajib di bawah 10ms (toleransi CI: rata-rata < 0.5ms)
    assert avg_latency_ms < 0.5, f"Overhead router terlalu tinggi: {avg_latency_ms:.4f}ms"

Trade-offs & Rekomendasi Pemeliharaan

  • Mock Drift vs Real Testing: Mock dispatcher tidak menguji perubahan format payload aktual API provider. Mitigasi dengan menjalankan contract smoke test terjadwal secara terpisah di luar alur utama CI PR.
  • Fixture Decay: Distribusi prompt aktual pengguna dapat bergeser dari fixture awal. Lakukan pembaruan dataset golden fixture secara berkala dari sampel log produksi yang telah disanitasi.
  • Deterministik vs Fleksibilitas: Hindari penyematan model probabilistik (seperti LLM evaluasi lain) di dalam router core. Gunakan algoritma klasifikasi statis, rule-based heuristics, atau lightweight embedding classifiers untuk menjaga evaluasi tetap deterministik di CI.