How to Test Model Routing for Answer Quality, Session Cost, and Routing Latency

Evaluate model routing on three separate tracks: answer quality, billed session cost after caching, and routing latency. GMI Router returns selection metadata for attribution, while fixed-model baselines and per-arm API keys share GMI Cloud's billing views.

GMI Router selects a model per request, with no judge model and a published routing time under 200 ms. The plan below covers measurements, sample sizes, billing, and latency attribution.

Why does a single routing scorecard give you an unusable answer?

A composite score hides task-level quality differences, uses the wrong cost unit for cached sessions, and mixes model speed with routing overhead.

Report quality per task bucket: a router can improve reasoning while weakening simple rewrites, leaving the overall mean unchanged.

Report cost per completed session. Per-call pricing misses cache reuse across a multi-turn conversation.

A 90 ms end-to-end delay may come from routing or selection of a slower model. Matched-model comparisons separate those effects.

Which managed services do you need for this test?

Use GMI Router for the routed arm, MaaS endpoints for fixed baselines, the Console playground for smoke tests, and Console usage for billing. Keep the arms under the same accounting rules.

The routed arm calls https://console.gmicloud.ai/api/v1/ie/recommendation/autoroute with a bearer key, messages, and optional cost/balanced/quality mode. Fixed arms use POST /v1/chat/completions on https://api.gmi-serving.com/v1. See the GMI Router documentation and GMI Cloud MaaS page.

Smoke-test five representative prompts in the playground. Routing is free for a limited preview period, so the cost comparison currently measures model spend.

How do you measure answer quality against a fixed-model baseline?

Use blind paired comparisons within each task bucket. Apply five controls:

1. Define buckets from actual product tasks, including deliberately easy and hard cases. GMI's leaderboard categories offer a starting taxonomy: Agentic, Coding, Instruction, Math, Reasoning, Knowledge, Language, and Data-Analysis. Its benchmark chart includes GPQA Diamond, IFBench, LiveBench, LiveCodeBench, Terminal-Bench, and WildBench; use your own prompts rather than copying benchmark content.

2. For each prompt, grade routed and baseline answers head to head as win, loss, or tie, avoiding an absolute score that varies across graders.

3. Blind the grader: randomize A/B order and remove model names and routing metadata. Use a judge outside both arms' model pools, and score each pair twice with the sides swapped to control position bias.

4. Use two fixed baselines: the frontier model otherwise chosen for production, such as openai/gpt-5.5, and a cheaper candidate. Compare routed cost with the frontier and routed quality with the cheap baseline.

5. Send byte-identical messages. Router documents messages, mode, and stream; unknown fields are ignored. Leave baseline sampling at endpoint defaults and disclose that sampling is uncontrolled across interfaces rather than claiming it is matched.

GMI reports a +2.4-point Quality-mode improvement over GPT-5.5 in its internal benchmark across eight task categories. This is directional, not independently audited, and limited to the allowed pool. Results vary with workload, prompt length, pool, and settings; measure your own gain. (GMI Router product page)

How many prompts does each task bucket need?

Choose the smallest win-rate difference that would change the deployment decision, then size each bucket. The examples below require 79–1,305 decisive comparisons.

The counts use a two-sided normal approximation to a binomial test against a 50% win rate, with 5% significance and 80% power. Exclude ties from decisive comparisons.

A 55% target across six buckets needs roughly 7,800 graded pairs. If that is unaffordable, prioritize three high-traffic buckets or a larger effect: detecting 65% needs 142 prompts. Explicitly assign unmeasured buckets to routing or a fixed endpoint until evaluated.

For 400 prompts per arm at 3,000 input and 600 output tokens, GPT-5.5 list rates of $5/$30 per 1M tokens yield $6.00 input + $7.20 output = $13.20 per arm, or about $40 for three arms at those illustrative rates. Rates are from October 15, 2026; recheck the model library.

Two swapped-side grades per pair require 800 judge calls. At 4,200 input and 100 output tokens each, the same illustrative rates give $16.80 + $2.40, about $19 per baseline comparison. Two baselines add about $38 in grading to $40 in generation: roughly $78 total. The judge must sit outside both arms' pools; budget its actual catalog rate.

How do you measure session cost after caching?

Use settled Console billing filtered by an API key dedicated to each arm, divided by completed sessions.

The usage object reports prompt_tokens, completion_tokens, and total_tokens, not cache-priced tokens. On October 15, 2026, GPT-5.5 cache reads listed at $0.50 per 1M tokens versus $5.00 for fresh input. Token counts times list rates miss that tenfold difference; report the settled bill. (model library; billing usage views)

Keep the billing split reproducible:

1. Issue one API key per arm. The Console usage filters include API Key as a dimension, so a dedicated key per arm splits the ledger cleanly without any time-window arithmetic. Never run a test arm on a key that also serves production traffic.
2. Replay whole sessions in order, not shuffled requests. The Router page states that KV-cache reuse is preserved across turns, so the savings only appear when a multi-turn session is sent turn by turn as it would be in production. The router itself is stateless, and the docs are explicit: "Resend the full conversation on every request." Your harness holds the conversation; the cache economics still land at the serving layer.
3. Set the usage filters deliberately. The filter panel exposes timezone (local or UTC), time range, granularity (daily or hourly), API key, and model name. Set the timezone to UTC, match it to the timestamps your harness logs, and use hourly granularity so a run that straddles midnight does not get smeared across two days.
4. Divide by sessions, not requests. Cost per completed session is the number that maps to your unit economics. Cost per request is not comparable across arms whose sessions differ in length or turn count, and that is exactly the condition a routing change creates.

Inspect routed spend by model. If one model receives 90% of spend, the breakdown also reveals how the pool fits your task mix.

How do you separate routing latency from model latency?

Measure p50 and p95 TTFT deltas on pairs where routed selected_model equals the fixed baseline. Log request-sent, headers-received, and first-token timestamps; TTFT is first token minus send time. With fewer than about 20 matched pairs, p95 reflects only the largest observations.

Log selected_model, task_type, selected_mode, fallback_models, and recommendation_id from routing_metadata. Non-streaming returns an object; streaming emits metadata before [DONE]. Capture X-Recommendation-ID at header arrival where available; it is absent on auth and early validation errors and may still help attribute a failed stream.

Compute routed-minus-baseline TTFT for identical prompts on the same serving model. This estimates routed-path overhead; unmatched-model latency also reflects model selection and belongs in the broader product evaluation.

Apply three latency controls:

• Report p50 and p95, never the mean. One cold connection or one retried request moves a mean enough to invent a problem that is not there.
• Exclude retried requests from the latency distribution and count them separately as an error rate. The documented fallback behavior is specific: in streaming mode, "One backup model may be tried before the first token on a 5xx, provider error, 429, or a 10-second first-token timeout", and once tokens start flowing there is no further fallback. In non-streaming mode, the primary model plus one backup are tried on a transient failure, with a combined timeout of 10 minutes. That 10 minutes is the service's ceiling, not a target; the harness below sets its own idle read timeout at 120 seconds on purpose, because a latency experiment that waits ten minutes for one row has stopped being a latency experiment. A request that fell back has a legitimately different latency profile; mixing it into the p95 tells you nothing you can act on.
• Run the arms interleaved, not sequentially. Alternate routed and baseline requests inside the same wall-clock window so that upstream load affects both arms equally.

GMI publishes routing time under 200 ms before model execution. Your matched-model delta also includes network, queue, and cache differences; report a distribution. For a substantially larger delta, inspect connections and fallback rows before blaming routing. GMI plans fuller production metrics, but deployment should use your own measured latency budget. (GMI Router page)

What does the three-track script need to handle?

The harness streams both endpoint shapes, records routing metadata and TTFT, retains partial output, bounds 429 retries, distinguishes empty from incomplete responses, and validates prompts before sending them.

import codecs, json, os, time, uuid
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
import requests

ROUTER_URL = "https://console.gmicloud.ai/api/v1/ie/recommendation/autoroute"
MAAS_URL = "https://api.gmi-serving.com/v1/chat/completions"
CONNECT_TIMEOUT, READ_TIMEOUT = 10, 120   # connect and idle read, not a wall-clock deadline
MAX_RETRIES = 2
RETRY_BUDGET_S = 30                       # total wait allowed across retries, not per attempt
META_FIELDS = ("selected_model", "task_type", "selected_mode", "fallback_models")

def _validate(messages):
    if not isinstance(messages, list) or not messages:
        raise ValueError("messages must be a non-empty list")
    has_user = False
    for m in messages:
        if not isinstance(m, dict):
            raise ValueError("each message must be a dict")
        content = m.get("content")
        if not isinstance(content, str) or not content.strip():
            raise ValueError("each message needs non-empty string content")
        has_user = has_user or m.get("role") == "user"
    if not has_user:
        raise ValueError("at least one user message is required")

def _payload(arm, messages):
    _validate(messages)                   # raises before any network call
    body = {"messages": messages, "stream": True}
    if arm["kind"] == "router":
        body["mode"] = arm["mode"]        # cost | balanced | quality
    else:
        body["model"] = arm["model"]      # baseline stays on endpoint defaults
    return body

def _retry_delay(header, attempt):
    """Retry-After is delay-seconds or an HTTP-date. Anything else falls back to backoff."""
    fallback = float(2 ** attempt)
    if not header:
        return fallback
    try:
        return max(0.0, float(header))
    except (TypeError, ValueError):
        pass
    try:
        when = parsedate_to_datetime(header)
    except (TypeError, ValueError, IndexError):
        return fallback
    if when is None:
        return fallback
    if when.tzinfo is None:
        when = when.replace(tzinfo=timezone.utc)
    return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())

def _sse_lines(resp):
    """Yield SSE lines as they arrive. iter_lines buffers and would inflate TTFT.
    Decode UTF-8 here rather than passing decode_unicode=True: Requests falls back
    to ISO-8859-1 when a text response omits a charset, which mangles non-ASCII
    answers, and an incremental decoder survives a character split across chunks."""
    decode = codecs.getincrementaldecoder("utf-8")("replace").decode
    buf = ""
    for chunk in resp.iter_content(chunk_size=None):
        if not chunk:
            continue
        buf += decode(chunk)
        while "\n" in buf:
            line, buf = buf.split("\n", 1)
            yield line.rstrip("\r")
    buf += decode(b"", True)
    if buf:
        yield buf

def _read_stream(resp, t0, state):
    """Writes into state as it goes, so a mid-stream failure keeps what already arrived."""
    event, parts = None, []
    for line in _sse_lines(resp):
        if not line:
            event = None
            continue
        if line.startswith("event:"):
            event = line[6:].strip()
            continue
        if not line.startswith("data:"):
            continue
        data = line[5:].strip()
        if data == "[DONE]":
            state["closed"] = True
            break
        try:
            obj = json.loads(data)
        except json.JSONDecodeError:
            continue                      # keep-alive or other non-JSON event
        if not isinstance(obj, dict):
            continue                      # null or scalar event payload
        if event == "error" or "error" in obj:
            state["stream_error"] = str(obj.get("error", obj))[:200]
            continue
        if event == "routing_metadata" or "routing_metadata" in obj:
            payload = obj.get("routing_metadata", obj)
            if isinstance(payload, dict):
                state["meta"] = payload   # handles both the nested and the flat event shape
            continue
        for choice in obj.get("choices") or []:
            piece = ((choice or {}).get("delta") or {}).get("content") or ""
            if piece:
                if state["ttft_ms"] is None:
                    state["ttft_ms"] = (time.perf_counter() - t0) * 1000
                parts.append(piece)
                state["answer"] = "".join(parts)

def _is_timeout(exc):
    """A read timeout mid-stream reaches you as ConnectionError wrapping urllib3's
    error, sometimes only through args rather than __cause__."""
    level, depth = [exc], 0
    while level and depth < 6:
        nxt = []
        for err in level:
            if not isinstance(err, BaseException):
                continue
            if isinstance(err, requests.Timeout) or "Timeout" in type(err).__name__:
                return True
            nxt += [err.__cause__, err.__context__, *getattr(err, "args", ())]
        level, depth = nxt, depth + 1
    return False

def _finish(row, state, t0, t_start, status, error):
    now = time.perf_counter()
    row["attempt_ms"] = (now - t0) * 1000       # last attempt only
    row["total_ms"] = (now - t_start) * 1000    # includes earlier attempts and backoff
    row["ttft_ms"] = state["ttft_ms"]
    row["answer"] = state["answer"]             # partial output is kept, never discarded
    for field in META_FIELDS:                   # keep the whole decision, not just the model
        value = state["meta"].get(field)
        if value is not None:                   # an empty fallback list is a real answer
            row[field] = value
    row["recommendation_id"] = row["recommendation_id"] or state["meta"].get("recommendation_id")
    row["status"], row["error"] = status, error
    return row

def call_once(arm, messages, session_id, turn):
    row = {"request_id": str(uuid.uuid4()), "arm": arm["name"], "session_id": session_id,
           "turn": turn, "attempts": 0, "status": "error", "ttft_ms": None,
           "attempt_ms": None, "total_ms": None, "selected_model": arm.get("model"),
           "task_type": None, "selected_mode": None, "fallback_models": None,
           "recommendation_id": None, "answer": "", "error": None}
    url = ROUTER_URL if arm["kind"] == "router" else MAAS_URL
    body = _payload(arm, messages)              # raises on invalid input, before the network
    headers = {"Authorization": f"Bearer {os.environ[arm['key_env']]}",   # one key per arm
               "Content-Type": "application/json"}
    t_start, waited = time.perf_counter(), 0.0

    for attempt in range(MAX_RETRIES + 1):
        row["attempts"] = attempt + 1
        state = {"ttft_ms": None, "answer": "", "meta": {},
                 "closed": False, "stream_error": None}
        t0 = time.perf_counter()
        try:
            with requests.post(url, json=body, headers=headers, stream=True,
                               timeout=(CONNECT_TIMEOUT, READ_TIMEOUT)) as resp:
                row["recommendation_id"] = resp.headers.get("X-Recommendation-ID")
                if resp.status_code == 429:
                    if attempt == MAX_RETRIES:
                        return _finish(row, state, t0, t_start, "rate_limited",
                                       "429 after retries")
                    wait = _retry_delay(resp.headers.get("Retry-After"), attempt)
                    if waited + wait > RETRY_BUDGET_S:   # budget spans every retry
                        return _finish(row, state, t0, t_start, "rate_limited",
                                       f"{waited + wait:.0f}s of waiting over budget")
                    resp.close()
                    time.sleep(wait)
                    waited += wait
                    continue
                if resp.status_code >= 400:
                    return _finish(row, state, t0, t_start, "http_error",
                                   f"{resp.status_code}: {resp.text[:200]}")
                _read_stream(resp, t0, state)
        except requests.RequestException as exc:
            kind = "timeout" if _is_timeout(exc) else "transport_error"
            return _finish(row, state, t0, t_start, kind, str(exc)[:200])

        if state["stream_error"]:
            return _finish(row, state, t0, t_start, "partial", state["stream_error"])
        if not state["closed"]:
            return _finish(row, state, t0, t_start, "incomplete", "stream ended without [DONE]")
        return _finish(row, state, t0, t_start,
                       "ok" if state["answer"].strip() else "empty", None)
    return row

def percentile(values, pct):
    clean = sorted(v for v in values if v is not None)
    if not clean:
        return None                             # no comparable requests, say so
    k = (len(clean) - 1) * pct / 100
    lo, hi = int(k), min(int(k) + 1, len(clean) - 1)
    return clean[lo] + (clean[hi] - clean[lo]) * (k - lo)

The implementation makes six checks explicit:

• Only ok rows go into the quality and latency stats. empty, partial, incomplete, timeout, rate_limited and http_error are separate statuses, so a truncated answer never scores as a complete one.
• total_ms spans every attempt and the backoff between them, while attempt_ms covers the last one. Averaging a field that silently drops retried time is how a rate-limited run ends up looking fast.
• The SSE reader writes into a shared state object and decodes UTF-8 itself. When a stream dies mid-answer, the partial text and the first-token time survive. Decoding explicitly matters because the standard client falls back to Latin-1 when a text response arrives without a charset, which quietly corrupts every non-ASCII answer before a grader ever sees it.
• All four routing metadata fields land in the row, not just the model name. task_type and selected_mode are what let you check later that a bucket was classified the way you assumed, and fallback_models tells you what the router would have reached for.
• The arm supplies the name of its own key environment variable. A single import-time key would defeat the per-arm billing split the cost track depends on.
• percentile returns None on an empty list instead of raising or reporting 0 ms. That is what happens when the matched-model filter leaves you with nothing comparable, and you want to see it before it reaches a chart.

Replay each session in turn order, alternate arms, and append results to JSONL keyed by session_id and turn. Log model, bucket, arm, and retries there; obtain dollar cost from Console billing.

How do you read the results and make the call?

Set per-bucket thresholds before testing: at least 50% wins in decisive comparisons, billed session cost below the frontier baseline by the required margin, and matched-model p95 within the product latency budget. Use GMI's published sub-200 ms routing time as context, not a substitute for measurement.

Act per bucket: mode travels with each request from the client; the organization owner sets the workspace-wide Allowed Model Pool. These are separate controls, so request policy and catalog governance need separate owners.

Check quality first. A bucket that fails its quality threshold does not qualify on cost savings alone.

How do you keep the result from going stale?

Re-run the three highest-traffic buckets monthly using stored prompts and fixed baselines. Model catalog changes can alter selections; retain recommendation IDs and metadata to audit older runs.

If one bucket repeatedly selects the same model at predictable load, evaluate Prime Inference: reserved single-tenant GPUs keep weights pre-loaded and bill per GPU-hour. Routing and dedicated serving can share the same account. (GMI Cloud Prime Inference)

FAQ

Can I evaluate routing quality without building a harness? Yes, for a smoke test. Use the Console playground to check task classification; use a scripted harness for defensible paired comparisons and sample counts.

Why can't I just multiply tokens by list price to get session cost? The usage object does not attribute cache-priced tokens. Multiplying prompt_tokens and completion_tokens by list rates misses cache savings. Use settled Console billing by arm key and model.

How do I know which model answered a routed request? Use selected_model in routing_metadata. Non-streaming includes metadata in the response; streaming emits it before [DONE]. Log the remaining routing fields and X-Recommendation-ID when present. Mark missing-model rows unattributed rather than assigning them to a baseline.

What does the routing step itself cost? Routing and recommendation are free for a limited preview period; pay the selected model's token charges. Recheck the commercial terms before a production commitment.

How long should a routing evaluation take? Allow roughly a week: two days for buckets and the harness, one for interleaved runs, two for grading, and one for billing reconciliation. Set sample sizes before starting; the illustrated range is 79–1,305 decisive comparisons per bucket.

Does routing work for multi-turn sessions, or only single prompts? Test complete sessions in turn order. GMI Router is cache-aware but stateless, so the client must resend the full conversation on every request.

Start the test

Create one API key per arm, smoke-test five prompts in the playground, then run /autoroute against two fixed-model baselines. GMI's Router launch article and Dynamic Model Escalation for Coding Agents provide background on selection and its two-harness experiment. (GMI Router playground; Automatic Model Selection for Every Prompt: GMI Router Is Live)

For help sizing task buckets or the evaluation budget, contact GMI Cloud with your prompt mix and current model spend. (contact the GMI Cloud team)