Building your own AI model router is one of the highest-ROI projects you can undertake in 2026. With AI API costs eating into budgets and new models launching every week, the ability to automatically route requests to the right model can cut your costs by 50% or more while improving response quality.
In this tutorial, we'll build a production-ready AI model router in Python that can route requests based on task type, cost, and latency — with automatic failover.
Why Build Your Own Router?
Before we dive into code, let's understand why building your own router makes sense:
- Cost optimization — Route simple requests to cheap models, complex ones to flagships
- Redundancy — If one provider goes down, automatically switch to another
- Flexibility — Try new models without changing your application code
- Transparency — You control the routing logic and can audit every decision
Prerequisites
- Python 3.11+
- An API key from Qubax AI (gives you access to 370+ models through one API)
- Basic familiarity with async Python
Step 1: Set Up Your Environment
First, let's create our project and install dependencies:
mkdir ai-model-router && cd ai-model-router
python -m venv venv
source venv/bin/activate
pip install httpx pydantic asyncioStep 2: Define Your Model Registry
The heart of any router is its model registry — a list of available models with their capabilities, pricing, and performance characteristics.
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
class TaskType(Enum):
CODING = "coding"
WRITING = "writing"
REASONING = "reasoning"
SIMPLE_QA = "simple_qa"
MULTIMODAL = "multimodal"
@dataclass
class ModelConfig:
name: str
provider: str
input_price_per_1m: float # USD per 1M input tokens
output_price_per_1m: float # USD per 1M output tokens
max_tokens: int
supports_streaming: bool = True
strengths: list[TaskType] = field(default_factory=list)
avg_latency_ms: int = 1000
enabled: bool = True
# Real pricing from Qubax AI (as of August 2026)
MODEL_REGISTRY: dict[str, ModelConfig] = {
"gpt-5.6-sol": ModelConfig(
name="gpt-5.6-sol",
provider="openai",
input_price_per_1m=0.7121,
output_price_per_1m=4.2728,
max_tokens=128000,
strengths=[TaskType.CODING, TaskType.REASONING],
avg_latency_ms=1200,
),
"claude-opus-5": ModelConfig(
name="claude-opus-5",
provider="anthropic",
input_price_per_1m=0.675,
output_price_per_1m=3.375,
max_tokens=200000,
strengths=[TaskType.WRITING, TaskType.REASONING],
avg_latency_ms=1500,
),
"glm-5.2": ModelConfig(
name="glm-5.2",
provider="zhipu",
input_price_per_1m=0.0046,
output_price_per_1m=0.0145,
max_tokens=128000,
strengths=[TaskType.SIMPLE_QA, TaskType.WRITING],
avg_latency_ms=800,
),
"deepseek-v4-pro": ModelConfig(
name="deepseek-v4-pro",
provider="deepseek",
input_price_per_1m=0.12,
output_price_per_1m=0.24,
max_tokens=128000,
strengths=[TaskType.CODING, TaskType.REASONING],
avg_latency_ms=1000,
),
}Notice the price difference: GLM 5.2 costs $0.0046 per 1M input tokens, while GPT-5.6 Sol costs $0.7121 — that's 155x cheaper for input tokens. Routing simple questions to GLM 5.2 instead of GPT-5.6 Sol can save enormous amounts of money.
Step 3: Build the Routing Engine
Now let's build the core routing logic:
import re
from typing import Optional
class ModelRouter:
def __init__(self, registry: dict[str, ModelConfig]):
self.registry = registry
def detect_task_type(self, prompt: str) -> TaskType:
"""Classify the request to determine the best model type."""
prompt_lower = prompt.lower()
# Coding detection
if any(kw in prompt_lower for kw in [
"code", "function", "python", "javascript", "api", "bug",
"compile", "debug", "class", "method", "algorithm", "sql",
"react", "typescript", "rust", "golang"
]):
return TaskType.CODING
# Reasoning detection
if any(kw in prompt_lower for kw in [
"analyze", "compare", "why", "explain why", "reasoning",
"prove", "derive", "calculate", "step by step"
]):
return TaskType.REASONING
# Writing detection
if any(kw in prompt_lower for kw in [
"write", "essay", "article", "blog", "story", "poem",
"email", "letter", "summary", "paraphrase"
]):
return TaskType.WRITING
# Default to simple QA
return TaskType.SIMPLE_QA
def select_model(
self,
prompt: str,
prefer_cost: bool = False,
prefer_speed: bool = False,
exclude_models: set[str] = None,
) -> Optional[ModelConfig]:
"""Select the best model for the given prompt."""
exclude_models = exclude_models or set()
task_type = self.detect_task_type(prompt)
# Get candidates: enabled models not excluded
candidates = [
m for m in self.registry.values()
if m.enabled and m.name not in exclude_models
]
if not candidates:
return None
# Score each model
def score_model(model: ModelConfig) -> float:
score = 0.0
# Task match is most important
if task_type in model.strengths:
score += 100
# Cost preference
if prefer_cost:
max_input = max(m.input_price_per_1m for m in candidates)
max_output = max(m.output_price_per_1m for m in candidates)
cost_score = 100 * (1 - (
model.input_price_per_1m / max_input +
model.output_price_per_1m / max_output
) / 2)
score += cost_score
# Speed preference
if prefer_speed:
max_latency = max(m.avg_latency_ms for m in candidates)
score += 100 * (1 - model.avg_latency_ms / max_latency)
return score
# Return the highest-scoring model
best = max(candidates, key=score_model)
return best
def estimate_cost(
self, model: ModelConfig, input_tokens: int, output_tokens: int
) -> float:
"""Estimate the cost of a request in USD."""
return (
(input_tokens / 1_000_000) * model.input_price_per_1m +
(output_tokens / 1_000_000) * model.output_price_per_1m
)Step 4: Add API Integration with Failover
Now let's add the actual API calling logic with automatic failover:
import httpx
import asyncio
import json
from typing import AsyncGenerator
class QubaxRouter:
def __init__(self, api_key: str, base_url: str = "https://api.qubax.ai/v1"):
self.client = httpx.AsyncClient(
base_url=base_url,
headers={"Authorization": f"Bearer {api_key}"},
timeout=60.0,
)
self.router = ModelRouter(MODEL_REGISTRY)
self.failed_models: set[str] = set()
async def chat(
self,
messages: list[dict],
prefer_cost: bool = False,
prefer_speed: bool = False,
stream: bool = False,
) -> dict | AsyncGenerator:
"""Route a chat request to the best available model."""
prompt = " ".join(m.get("content", "") for m in messages)
# Try up to 3 models with failover
for attempt in range(3):
model = self.router.select_model(
prompt,
prefer_cost=prefer_cost,
prefer_speed=prefer_speed,
exclude_models=self.failed_models,
)
if model is None:
raise RuntimeError("No models available")
try:
response = await self._call_api(
model.name, messages, stream=stream
)
# Clear failures on success
self.failed_models.discard(model.name)
return response
except Exception as e:
print(f"Model {model.name} failed: {e}. Trying next...")
self.failed_models.add(model.name)
continue
raise RuntimeError("All models failed")
async def _call_api(
self, model: str, messages: list[dict], stream: bool = False
) -> dict:
"""Make the actual API call to Qubax."""
payload = {
"model": model,
"messages": messages,
"stream": stream,
}
response = await self.client.post("/chat/completions", json=payload)
response.raise_for_status()
return response.json()
async def close(self):
await self.client.aclose()Step 5: Add Cost Tracking
A good router tracks spending so you can see your savings:
from datetime import datetime
from collections import defaultdict
class CostTracker:
def __init__(self):
self.usage: dict[str, list[dict]] = defaultdict(list)
def record(
self,
model: str,
input_tokens: int,
output_tokens: int,
cost: float,
):
self.usage[model].append({
"timestamp": datetime.utcnow().isoformat(),
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost": cost,
})
def total_cost(self) -> float:
return sum(
entry["cost"]
for entries in self.usage.values()
for entry in entries
)
def cost_by_model(self) -> dict[str, float]:
return {
model: sum(e["cost"] for e in entries)
for model, entries in self.usage.items()
}
def savings_vs_single_model(
self, single_model: str, total_input_tokens: int, total_output_tokens: int
) -> float:
"""Calculate savings vs using a single expensive model for everything."""
if single_model not in MODEL_REGISTRY:
return 0.0
model = MODEL_REGISTRY[single_model]
single_cost = (
(total_input_tokens / 1e6) * model.input_price_per_1m +
(total_output_tokens / 1e6) * model.output_price_per_1m
)
return single_cost - self.total_cost()Step 6: Put It All Together
Here's how to use your complete AI model router:
import asyncio
import os
async def main():
router = QubaxRouter(api_key=os.environ["QUBAX_API_KEY"])
tracker = CostTracker()
# Example 1: A coding question (routes to GPT-5.6 Sol)
coding_response = await router.chat([
{"role": "user", "content": "Write a Python function to implement binary search"}
])
print(f"Coding response from: {coding_response['model']}")
# Example 2: A simple question (routes to GLM 5.2 for cost savings)
simple_response = await router.chat([
{"role": "user", "content": "What is the capital of France?"}
], prefer_cost=True)
print(f"Simple response from: {simple_response['model']}")
# Example 3: A writing task (routes to Claude Opus 5)
writing_response = await router.chat([
{"role": "user", "content": "Write a professional email to my boss requesting time off"}
])
print(f"Writing response from: {writing_response['model']}")
await router.close()
asyncio.run(main())Real Cost Savings Example
Let's say your app processes 100,000 requests per day:
- 30% are simple QA → routed to GLM 5.2 at ~$0.0001/request
- 40% are coding tasks → routed to GPT-5.6 Sol at ~$0.05/request
- 20% are writing tasks → routed to Claude Opus 5 at ~$0.04/request
- 10% are complex reasoning → routed to GPT-5.6 Sol at ~$0.08/request
Without a router (all GPT-5.6 Sol): ~$5,000/day With a router: ~$2,100/day Savings: 58% or $2,900/day = $87,000/month
Next Steps
- Sign up for a Qubax AI API key at qubax.ai/models
- Read the API documentation at qubax.ai/docs
- Add more models to your registry as new ones launch
- Implement streaming for real-time responses
- Add logging and analytics to track routing decisions over time
- Build a dashboard to visualize your cost savings
You can find the complete code and more examples in the Qubax AI documentation.
Building an AI model router is one of the highest-ROI projects you can undertake. With the pricing data from Qubax AI, you can make informed decisions about which models to use for each task — and save thousands of dollars per month.
FAQ
Q: How much does it cost to run an AI model router?
A: The router itself is just code — it's free to run. You only pay for the AI model API calls. By routing simple requests to cheaper models, the router typically saves 50-70% on API costs.
Q: Which AI models should I include in my router?
A: Start with a mix of price tiers: a cheap model like GLM 5.2 ($0.0046/1M input), a mid-tier model like DeepSeek V4 Pro ($0.12/1M input), and a flagship like GPT-5.6 Sol ($0.7121/1M input). Browse all available models at Qubax AI.
Q: How do I get an API key for multiple AI models?
A: Sign up at Qubax AI to get a single API key that works with 370+ models. You don't need separate accounts with each provider.
Q: What happens if a model provider goes down?
A: Your router automatically falls back to the next best model. The failover logic in our tutorial tries up to 3 models before giving up.
Q: Can I use this router in production?
A: Yes, but add proper error handling, logging, rate limiting, and monitoring. The tutorial code is a starting point — for production, consider using a managed service like Qubax AI which handles failover automatically.
Q: How do I know which model is best for each task type?
A: Check out our model comparison articles for head-to-head benchmarks. You can also A/B test models on your own data to find the best fit for your specific use case.
