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:

FeatureWhen it appliesDecision
FallbackWhen a provider failsMove to the next
RoutingBefore sending, according to policyChoose 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:

StrategyLogic
cost-firstSort providers by cost_tier ascending. Try the cheapest first.
quality-firstSort by cost_tier descending (proxy: more expensive = better quality).
balancedOnly 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_tier in 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

  1. LiteLLM Router strategies — reference for how others do it.
  2. Langfuse — LLM cost tracking — tool for cost tracking in production.
  3. OpenRouter price comparison API — prices of all models in one endpoint.
  4. Anyscale routing patterns — perspectives from teams that route at scale.