Module 8: Unified AI Client — Final integrating project
Cost optimization
Fallback fixes failures. But in real production, you also want to actively optimize cost: routing to the cheapest provider when the task allows, and to the more expensive/better one when quality matters.
In this capsule you're going to add routing strategies to the UnifiedClient. The public interface barely changes — we add a priority parameter or a default configuration, and the client decides internally.
By the end you'll be able to:
- Configure multiple routing policies:
cost-first,quality-first,balanced - Change policy per request without re-instantiating the client
- Calculate real savings by applying smart routing
- Combine routing with fallback (compatible, not mutually exclusive)
Mental model: routing vs fallback
They're two distinct features that get confused:
| Feature | When it applies | Decision |
|---|---|---|
| Fallback | When a provider fails | Move to the next |
| Routing | Before sending, according to policy | Choose which provider to try first |
They're complementary. Routing decides who is "primary" for this request; if that primary fails, fallback keeps working as in capsule 04.
Simple policy: tag per provider
We annotate each provider in its config with tags that describe its characteristics:
providers:
openai-mini:
name: openai-mini
type: openai
model: gpt-4o-mini
api_key_env: OPENAI_API_KEY
tags: ["quality", "balanced"]
cost_tier: "medium"
openrouter-mistral:
name: openrouter-mistral
type: openai_compatible
model: mistralai/mistral-7b-instruct
api_key_env: OPENROUTER_API_KEY
base_url: https://openrouter.ai/api/v1
tags: ["cheap", "balanced"]
cost_tier: "low"
ollama-mistral:
name: ollama-mistral
type: openai_compatible
model: mistral
base_url: http://localhost:11434/v1
api_key_env: OLLAMA_DUMMY_KEY
tags: ["cheap", "privacy", "offline"]
cost_tier: "free"
openai-gpt4o:
name: openai-gpt4o
type: openai
model: gpt-4o
api_key_env: OPENAI_API_KEY
tags: ["quality"]
cost_tier: "high"
Predefined strategies
We define 3 commonly used strategies:
| Strategy | Logic |
|---|---|
cost-first | Sort providers by cost_tier ascending. Try the cheapest first. |
quality-first | Sort by cost_tier descending (proxy: more expensive = better quality). |
balanced | Only providers with the balanced tag, in declaration order. |
You can add more (e.g., latency-first, privacy-first with the privacy tag). The pattern is the same.
Implementation: extending UnifiedClient
Add this to the ProviderConfig model in models.py:
class ProviderConfig(BaseModel):
name: str
type: Literal["openai", "openai_compatible", "modal_custom"]
model: str | None = None
base_url: str | None = None
api_key_env: str | None = None
base_url_env: str | None = None
api_token_env: str | None = None
price_input_per_1m: float | None = None
price_output_per_1m: float | None = None
# New fields:
tags: list[str] = Field(default_factory=list)
cost_tier: Literal["free", "low", "medium", "high"] | None = None
Create unified_ai_client/routing.py:
# unified_ai_client/routing.py
from typing import Literal
from .adapters.base import BaseAdapter
from .models import ProviderConfig
Priority = Literal["cost-first", "quality-first", "balanced"]
_TIER_RANK = {"free": 0, "low": 1, "medium": 2, "high": 3, None: 999}
def sort_by_priority(
adapters: list[BaseAdapter],
priority: Priority,
) -> list[BaseAdapter]:
"""
Returns the list of adapters reordered according to the priority policy.
Doesn't filter; always returns the same adapters, in a different order.
"""
if priority == "cost-first":
return sorted(adapters, key=lambda a: _TIER_RANK[a.config.cost_tier])
if priority == "quality-first":
return sorted(adapters, key=lambda a: -_TIER_RANK[a.config.cost_tier])
if priority == "balanced":
# Keeps order, but filters to those marked as "balanced"
ordered = [a for a in adapters if "balanced" in a.config.tags]
if not ordered:
return adapters # fallback: none were marked balanced
return ordered
raise ValueError(f"Unknown priority: {priority}")
Update client.py:
# unified_ai_client/client.py — only changes
from .routing import sort_by_priority, Priority
class UnifiedClient:
def __init__(
self,
config: ClientConfig,
circuit_threshold: int = 3,
circuit_duration_s: int = 60,
default_priority: Priority | None = None,
):
# ... rest the same
self.default_priority = default_priority
# ...
def chat(
self,
prompt: str,
*,
system: str | None = None,
max_tokens: int = 256,
temperature: float = 0.7,
use_fallback: bool = True,
priority: Priority | None = None,
) -> ChatResponse:
messages = []
if system:
messages.append(Message(role="system", content=system))
messages.append(Message(role="user", content=prompt))
return self.chat_with_messages(
messages,
max_tokens=max_tokens,
temperature=temperature,
use_fallback=use_fallback,
priority=priority,
)
def chat_with_messages(
self,
messages: list[Message],
max_tokens: int = 256,
temperature: float = 0.7,
use_fallback: bool = True,
priority: Priority | None = None,
) -> ChatResponse:
# Apply routing policy
prio = priority or self.default_priority
if prio:
ordered_adapters = sort_by_priority(self.adapters, prio)
else:
ordered_adapters = self.adapters
# If fallback disabled, only the first
if not use_fallback:
ordered_adapters = ordered_adapters[:1]
errors: dict[str, Exception] = {}
for adapter in ordered_adapters:
if self._circuit_open(adapter.name):
errors[adapter.name] = ProviderError(adapter.name, "Circuit open")
continue
try:
return self._try_with_retry(adapter, messages, max_tokens, temperature)
except (AuthError, RateLimitError, TimeoutError, ProviderError) as e:
errors[adapter.name] = e
self._record_failure(adapter.name)
continue
raise AllProvidersFailedError(errors)
Verification
Create examples/test_routing.py:
# examples/test_routing.py
from unified_ai_client import UnifiedClient
client = UnifiedClient.from_yaml("examples/clients_fallback.yaml")
# Cost-first: starts with Ollama (free)
r_cost = client.chat("Say hi", max_tokens=20, priority="cost-first")
print(f"\ncost-first → tried first: {r_cost.provider}")
# Quality-first: starts with GPT-4o if it's in your config, otherwise gpt-4o-mini
r_qual = client.chat("Say hi", max_tokens=20, priority="quality-first")
print(f"quality-first → tried first: {r_qual.provider}")
# Balanced: providers with the "balanced" tag
r_bal = client.chat("Say hi", max_tokens=20, priority="balanced")
print(f"balanced → tried first: {r_bal.provider}")
Expected output (assuming all providers working):
cost-first → tried first: ollama-mistral
quality-first → tried first: openai-gpt4o (if it's in config) or openai-mini
balanced → tried first: openai-mini (or the first "balanced" in config)
Configurable policy: default priority at the client level
If your app always wants cost-first, configure it at the client level:
client = UnifiedClient.from_yaml(
"examples/clients_fallback.yaml",
default_priority="cost-first",
)
# All requests use cost-first without having to pass it
client.chat("Question 1")
client.chat("Question 2")
# A specific request can override
client.chat("Critical question", priority="quality-first")
Advanced patterns
Pattern 1 — Routing by request context
For cases where the needed quality depends on the request (casual chat → cheap; complex analysis → expensive), your app determines the priority and passes it:
def smart_chat(prompt: str) -> str:
if "explica" in prompt or "analiza" in prompt:
return client.chat(prompt, priority="quality-first").text
return client.chat(prompt, priority="cost-first").text
Pattern 2 — Routing by user / tier
Your product has free and paid users. Free uses the cheap model; paid the expensive one:
def chat_for_user(prompt: str, tier: str) -> str:
priority = "quality-first" if tier == "paid" else "cost-first"
return client.chat(prompt, priority=priority).text
Pattern 3 — Routing by time of day
During business hours you use managed (fast); at night you use your cheaper Modal:
from datetime import datetime
def chat_by_hour(prompt: str) -> str:
hour = datetime.now().hour
priority = "balanced" if 9 <= hour < 18 else "cost-first"
return client.chat(prompt, priority=priority).text
How much routing saves in practice
Assume a typical distribution for your product:
- 70% of requests are "casual" (priority="cost-first" → cheap Mistral)
- 30% of requests are "complex" (priority="quality-first" → GPT-4o-mini)
Comparison with "always OpenAI" for 1M req/month (500 tokens average):
- Without routing (always GPT-4o-mini): 1M × 700 × $0.50 / 1M = ~$350/month
- With routing (70% Mistral OR + 30% GPT-4o-mini):
- 700k × 700 × $0.07 / 1M = $34
- 300k × 700 × $0.50 / 1M = $105
- Total: $139/month
- Savings: ~60% without losing quality in the cases where it matters.
This is real money. It fully justifies the added complexity of routing.
Common traps
Trap 1 — "Routing by hour breaks the UX of a user active at night." If your user gets different quality depending on the hour, that's a bug, not a feature. Routing by hour applies to batch workloads (internal jobs), not end users.
Trap 2 — "cost-tier 'free' always wins in cost-first." "Free" local Ollama doesn't scale if your server can't handle traffic. If you have it at cost-tier free but it only supports 10 req/min, when you have 100 req/min you'll have a queue. Set limits or raise its cost-tier.
Trap 3 — "I forgot that routing and fallback interact."
With priority="cost-first", your client tries Ollama first. If Ollama is down, fallback goes to OpenRouter, then OpenAI. The fallback order is the post-priority order, not the declaration order. Verify that this is what you want.
Trap 4 — "Routing hides that my cheap provider is bad." If Mistral 7B answers incorrectly on 30% of your "casual" requests, you're giving a bad product to 30% of your users. Cost routing assumes that all providers meet your quality SLA — verify beforehand with capsule 04 of Module 7.
Trap 5 — "Cost-tier not updated."
If OpenAI lowers the price of gpt-4o-mini by 50% tomorrow, your cost-tier: medium is outdated and the routing rankings are incorrect. Review the cost-tiers quarterly.
Exercise
Add a new strategy: priority="privacy-first" that filters providers to only those that have the privacy tag (in your config: local Ollama). If no provider qualifies, raise ConfigError. Important difference: it's not "sorting" — it's "filtering".
See solution
In routing.py, add:
Priority = Literal["cost-first", "quality-first", "balanced", "privacy-first"]
def sort_by_priority(
adapters: list[BaseAdapter],
priority: Priority,
) -> list[BaseAdapter]:
if priority == "cost-first":
return sorted(adapters, key=lambda a: _TIER_RANK[a.config.cost_tier])
if priority == "quality-first":
return sorted(adapters, key=lambda a: -_TIER_RANK[a.config.cost_tier])
if priority == "balanced":
ordered = [a for a in adapters if "balanced" in a.config.tags]
return ordered if ordered else adapters
if priority == "privacy-first":
privacy = [a for a in adapters if "privacy" in a.config.tags]
if not privacy:
from .exceptions import ConfigError
raise ConfigError(
"priority='privacy-first' requires at least one provider with the 'privacy' tag"
)
return privacy
raise ValueError(f"Unknown priority: {priority}")
Usage:
r = client.chat("Process sensitive medical data", priority="privacy-first")
print(f"Resolved by (on-prem only): {r.provider}")
Summary
You learned:
- ✅ Routing is sorting/filtering providers by policy; fallback is reacting to failures
- ✅ Tags and
cost_tierin config enable declarative policies - ✅ Three predefined strategies:
cost-first,quality-first,balanced - ✅ Configure a default at the client level, override per request
- ✅ Combinable patterns: routing by prompt context, by user tier, by hour
- ✅ Routing and fallback are compatible and combine naturally
Checkpoint: if you can do client.chat("...", priority="cost-first") and see in the logs that it tried the cheapest provider first, you're ready.
Next capsule
06 — Monitoring and metrics. Your client decides well but you don't know what decisions it's making. We add tracking that tells you: how many requests per provider, aggregate latency, accumulated cost, errors by type. It's the missing piece to make your client truly production-ready.
Resources
- LiteLLM Router strategies — reference for how others do it.
- Langfuse — LLM cost tracking — tool for cost tracking in production.
- OpenRouter price comparison API — prices of all models in one endpoint.
- Anyscale routing patterns — perspectives from teams that route at scale.