How to route by latency, cost, and capability simultaneously

When you run agentic workloads in production, pinning every call to a single flagship model is either a latency and cost tax or a reliability risk. To ship efficiently you need to route by latency cost capability at the request level: a trivial classification task should hit a fast cheap model, while a complex planning step demands deeper reasoning. This guide gives you a concrete, code-backed process to build that routing layer and operate it with fallback and metering.

Step 1: Catalog models with objective metadata

You cannot route without a registry. Build a static JSON file that lists every model your gateway can reach, tagged with the attributes your router cares about. Keep capability as explicit tags, and use relative tiers for latency and cost rather than hardcoding prices that drift.

{
  "models": [
    {
      "id": "anthropic/claude-3-haiku",
      "provider": "anthropic",
      "capabilities": ["chat", "json", "summarize"],
      "latency_tier": 1,
      "cost_tier": 1
    },
    {
      "id": "openai/gpt-4o-mini",
      "provider": "openai",
      "capabilities": ["chat", "code", "json", "vision"],
      "latency_tier": 2,
      "cost_tier": 2
    },
    {
      "id": "openai/gpt-4o",
      "provider": "openai",
      "capabilities": ["chat", "code", "json", "vision", "reasoning"],
      "latency_tier": 3,
      "cost_tier": 4
    },
    {
      "id": "meta/llama-3.1-70b-instruct",
      "provider": "meta",
      "capabilities": ["chat", "code", "json"],
      "latency_tier": 2,
      "cost_tier": 2
    }
  ]
}

Latency and cost tiers are integers where lower is better. You can later replace tiers with live measurements from your metering step.

Step 2: Define a constraint-and-score policy

Pure cost minimization breaks tasks that need reasoning. Pure capability matching ignores spend. The pattern that works: apply hard constraints first, then score the survivors.

Define required capabilities per request. Then among models that satisfy them, compute a normalized score:

score = w_latency * (latency_tier / max_tier) + w_cost * (cost_tier / max_tier)

Pick the lowest score. Weights are tunable per traffic class. For a user-facing chat agent, w_latency might be 0.6 and w_cost 0.4. For a background batch job, flip them.

Step 3: Implement the selector

Write a small function that loads the registry and returns the best model id. This is plain Python, no external deps.

import json

def load_registry(path="models.json"):
    with open(path) as f:
        return json.load(f)["models"]

def select_model(models, required_caps, w_latency=0.5, w_cost=0.5):
    max_tier = max(max(m["latency_tier"], m["cost_tier"]) for m in models)
    eligible = [
        m for m in models
        if set(required_caps).issubset(set(m["capabilities"]))
    ]
    if not eligible:
        raise ValueError(f"No model satisfies {required_caps}")
    scored = []
    for m in eligible:
        lat = m["latency_tier"] / max_tier
        cost = m["cost_tier"] / max_tier
        score = w_latency * lat + w_cost * cost
        scored.append((score, m["id"]))
    scored.sort(key=lambda x: x[0])
    return scored[0][1]

models = load_registry()
# Route a simple JSON extraction task
model_id = select_model(models, required_caps=["json", "chat"], w_latency=0.7, w_cost=0.3)
print(model_id)  # anthropic/claude-3-haiku

The selector is deterministic and unit-testable. It never touches the network.

Step 4: Call the model through a unified endpoint

Your agent code should speak OpenAI-compatible HTTP. A gateway such as n4n.ai provides one OpenAI-compatible endpoint covering 240+ models and honors client routing directives, so the model string from your selector drops straight into the request. You avoid per-provider SDK sprawl.

from openai import OpenAI

client = OpenAI(
    base_url="https://api.n4n.ai/v1",  # single endpoint, 240+ models
    api_key="YOUR_KEY",
)

def complete(model_id, prompt):
    resp = client.chat.completions.create(
        model=model_id,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
    )
    return resp.choices[0].message.content, resp.usage

If you self-host a similar gateway, swap the base_url. The router logic stays identical.

Step 5: Add automatic fallback for degradation

Providers rate-limit and degrade. Wrap the call so that on a retryable error you pick the next best model from the same eligible set, excluding the failed one.

import openai

def complete_with_fallback(models, required_caps, prompt, max_attempts=3):
    eligible = [m for m in models if set(required_caps).issubset(m["capabilities"])]
    attempted = set()
    last_err = None
    for _ in range(max_attempts):
        if not eligible:
            raise RuntimeError("No models left to try")
        # re-score each attempt in case tiers shift
        cand = select_model(eligible, required_caps)
        if cand in attempted:
            continue
        attempted.add(cand)
        try:
            return complete(cand, prompt)
        except openai.RateLimitError as e:
            last_err = e
            eligible = [m for m in eligible if m["id"] != cand]
        except openai.APIConnectionError as e:
            last_err = e
            eligible = [m for m in eligible if m["id"] != cand]
    raise last_err

This gives you automatic fallback when a provider is rate-limited or degraded without changing the calling code.

Step 6: Meter and observe

Per-token usage and latency are the feedback loop that keeps your tiers honest. Log the usage object and round-trip time on every call.

import time, logging

def metered_complete(model_id, prompt):
    t0 = time.perf_counter()
    content, usage = complete(model_id, prompt)
    dt = time.perf_counter() - t0
    logging.info({
        "model": model_id,
        "prompt_tokens": usage.prompt_tokens,
        "completion_tokens": usage.completion_tokens,
        "latency_ms": round(dt * 1000, 1),
    })
    return content

Aggregate these logs by model id. After a week you will see real latency distributions and token costs that let you replace static tiers with empirical percentiles.

Step 7: Verify the router end to end

Verification has two layers. First, unit-test the selector against the registry to prove constraints and weighting work.

def test_selector_prefers_cheap_for_simple():
    models = load_registry()
    picked = select_model(models, ["chat"], w_latency=0.1, w_cost=0.9)
    assert picked in {"anthropic/claude-3-haiku", "openai/gpt-4o-mini", "meta/llama-3.1-70b-instruct"}

def test_selector_requires_reasoning():
    models = load_registry()
    picked = select_model(models, ["reasoning"])
    assert picked == "openai/gpt-4o"

Second, run an integration script that fires one request per capability class through complete_with_fallback and prints the chosen model and latency. If the script returns coherent output and the logs show the expected model ids, the router is working.

Success criteria: the simple task never lands on gpt-4o when a tier-1 model is available, the reasoning task always picks a model with the reasoning tag, and a forced provider outage (simulate by using a bad key) still yields a response via fallback.

Tuning weights in production

Static weights are a starting point. Feed your meter logs into a daily job that adjusts w_latency and w_cost per route based on p95 latency breaches and budget burn. For interactive agents, bias hard toward latency; for nightly summarization, bias toward cost. The router code does not change—only the numbers passed in.

Route by latency cost capability simultaneously by keeping the policy explicit, the registry honest, and the fallback path tested. That is the difference between a demo and a system that survives real traffic.