Dealing with financial and failover side!
Introduction
Static routing rules are insufficient when building production-grade AI pipelines. If a local model container experiences a transient failure or a sudden spike in request volume, the entire application can stall. Furthermore, static quality thresholds fail to adapt to live performance conditions.
To solve this, we need a system that can handle infrastructure failures gracefully while dynamically adjusting its routing parameters based on real-time telemetry.
This post walks through a standalone, personal proof-of-concept — routing-v4 — built to demonstrate the immense financial and operational logic behind the multi-model orchestration paradigm. By layering a custom, telemetry-driven decision framework on top of an enterprise gateway (LiteLLM Router), we achieve a self-healing, self-optimizing AI routing pipeline.
The Dual-Layer Resilience Architecture
The system splits operational responsibilities into two distinct, isolated layers:
- The Strategic Layer (router.py** + telemetry.py)**: Runs a continuous feedback loop. It analyzes incoming sub-tasks using an economic utility function (U=β⋅Q−α⋅C) where the cost weight (α) and quality threshold are continuously adjusted on the fly by a live telemetry engine.
- The Tactical Layer (failover_manager.py** + llm_client.py)**: Protects application execution using per-model circuit breakers, explicit failover chains, and a unified connection layer.
+-----------------------------------+
| FastAPI Payload Intake Engine |
+-----------------+-----------------+
|
v
+-----------------------------------+
| Telemetry Engine & Param Tuner |
| (Live Ring Buffers & EMA) |
+-----------------+-----------------+
|
[Dynamic α and β Weights]
|
v
+-----------------------------------+
| Utility-Based Task Router |
| U = β(t)·Q − α(t)·C |
+-----------------+-----------------+
|
[Primary Model Nominee]
|
v
+-----------------------------------+
| Circuit-Breaker Failover Mgr |
| (CLOSED / OPEN / HALF_OPEN) |
+-----------------+-----------------+
|
[Resolved Execution Target]
|
v
+-----------------------------------+
| LiteLLM Transport Gateway |
+-----------------------------------+
Deep-Diving the Core Technical Core
State-Driven Circuit Breakers
To prevent requests from hanging on unresponsive endpoints, the FailoverManager tracks each model label inside an isolated state machine. Repeated failures trip the circuit into an OPEN state, instantly bypassing the model and routing requests to an explicit fallback chain.
# From src/failover_manager.py
class CircuitState(Enum):
CLOSED = "CLOSED" # Normal operations
OPEN = "OPEN" # Outage detected; block traffic
HALF_OPEN = "HALF_OPEN" # Testing recovery with probe calls
class CircuitBreaker:
# ... tracking logic ...
def record_failure(self) -> None:
with self._lock:
self._consecutive_failures += 1
# Update health score using an Exponential Moving Average
self.health_score = 0.7 * self.health_score + 0.3 * 0.0
if self._state == CircuitState.CLOSED and self._consecutive_failures >= self._failure_threshold:
self._state = CircuitState.OPEN
self._open_since = time.time()
When a primary model nominee is selected, the manager intercepts the call. If the primary target’s circuit is broken, it traverses the task-specific failover prioritization matrix:
# From src/failover_manager.py
def resolve(self, primary: ModelSpec, task_type: str) -> tuple[ModelSpec, Optional[FailoverEvent]]:
"""Resolves the best available model, fallback chain, or healthy asset."""
if self.is_callable(primary.label):
return primary, None
# Primary is blocked. Intercept and walk the fallback chain
for fallback_label, reason in primary.failover_chain:
if self.is_callable(fallback_label):
fb_model = MODEL_BY_LABEL.get(fallback_label)
if fb_model:
event = FailoverEvent(
primary_label=primary.label,
fallback_label=fallback_label,
reason=reason,
)
return fb_model, event
# Last resort: fallback to whichever model has the highest health score
best = max(MODEL_REGISTRY, key=lambda m: self._circuits[m.label].health_score)
return best, FailoverEvent(primary.label, best.label, "all_chains_exhausted")
Telemetry-Driven Parameter Tuning
The system does not use hardcoded thresholds. Instead, a TelemetryEngine maintains a fixed-size ring buffer for each model-task combination, computing p95 latencies and running quality scores.
If p95 latency spikes beyond a target SLA, the system calculates latency_pressure and automatically dials down the cost weight (α). This adjustments shifts preference toward faster, more expensive tiers to safeguard response speeds.
# From src/telemetry.py
def _tune_params(self, key: tuple[str, str], buf: Deque[TelemetryRecord]) -> DynamicParams:
n = len(buf)
q_ema = sum(r.quality_eval for r in buf) / n
# Calculate Latency Percentiles
latencies = sorted(r.latency_ms for r in buf)
p95 = latencies[min(n - 1, int(n * 0.95))] if n else 0
# Dynamic Threshold Calibration
delta = q_ema - _BASE_THRESHOLD
new_thr = _BASE_THRESHOLD + 0.5 * delta
new_thr = max(_MIN_THRESHOLD, min(_MAX_THRESHOLD, new_thr))
# Dynamic Cost Weight Modulation based on SLA Latency Pressure
latency_pressure = p95 / _TARGET_LATENCY_MS
cost_w = _BASE_COST_WEIGHT * (1.0 - min(0.5, latency_pressure * 0.3))
cost_w = max(_MIN_COST_WEIGHT, min(_MAX_COST_WEIGHT, cost_w))
return DynamicParams(
model_label=key[0],
task_type=key[1],
quality_threshold=round(new_thr, 4),
cost_weight=round(cost_w, 4),
quality_ema=round(q_ema, 3),
p95_latency_ms=p95
)
Integrated Dynamic Execution Loop
The router.py module consumes these live-tuned parameters directly during the routing cycle. It evaluates Candidate models based on the current cost weights and quality thresholds before calling the failover resolver:
"""
router.py — routing-v4
=========================
Core routing logic enhanced with:
· Dynamic quality thresholds from the telemetry engine
· Failover-aware model resolution via the failover manager
· Per-task utility function U = β(t)·Q − α(t)·C
where α and β are continuously adjusted per (model, task_type) pair
"""
from __future__ import annotations
from src.cost_registry import MODEL_REGISTRY, MODEL_BY_LABEL, TASK_FAILOVER_PRIORITY, ModelSpec
from src.task_classifier import TaskProfile
from src.failover_manager import failover_manager, FailoverEvent
from src.telemetry import telemetry_engine
# ── Static fallback parameters (used before telemetry has data) ───────────────
BASE_QUALITY_THRESHOLD = 0.72
BASE_COST_WEIGHT = 0.40
BASE_QUALITY_WEIGHT = 0.60
STATIC_TASK_THRESHOLDS: dict[str, float] = {
"documentation": 0.60,
"summarisation": 0.55,
"qa_simple": 0.55,
"code_generation": 0.72,
"code_review": 0.85,
"reasoning": 0.80,
}
SECURITY_OVERRIDES: set[str] = {"code_review"}
# ── Utility function ──────────────────────────────────────────────────────────
def _utility(model: ModelSpec, profile: TaskProfile) -> float:
"""
U = β·Q − α·C with dynamic α and β from the telemetry engine.
"""
alpha = telemetry_engine.get_cost_weight(model.label, profile.task_type.value)
beta = 1.0 - alpha
q = model.quality_score(profile.complexity)
c = model.cost_per_1k
return beta * q - alpha * c
def _effective_threshold(model_label: str, task_type: str) -> float:
"""
Blend the static task-level threshold with the live telemetry threshold.
The dynamic value wins once there are enough samples.
"""
static_thr = STATIC_TASK_THRESHOLDS.get(task_type, BASE_QUALITY_THRESHOLD)
dynamic_thr = telemetry_engine.get_threshold(model_label, task_type)
return round((static_thr + dynamic_thr) / 2.0, 4)
# ── Primary routing ───────────────────────────────────────────────────────────
def route(profile: TaskProfile) -> tuple[ModelSpec, FailoverEvent | None]:
"""
Return (selected_model, failover_event_or_None) for the given TaskProfile.
Selection rules:
1. Security overrides → always tier-4 (then failover if circuit open)
2. Filter by capability + context window
3. Filter by blended quality threshold
4. Best utility wins
5. Safety net: highest quality model if no candidate passes threshold
6. Resolve failover if the chosen model's circuit is not callable
"""
ttype = profile.task_type.value
if ttype in SECURITY_OVERRIDES:
primary = next(m for m in MODEL_REGISTRY if m.tier == 4)
return failover_manager.resolve(primary, ttype)
threshold = _effective_threshold("", ttype) # task-level threshold (no model prefix)
candidates = [
m for m in MODEL_REGISTRY
if ttype in m.capable_types
and m.max_tokens >= profile.token_estimate
and m.quality_score(profile.complexity) >= threshold
]
if not candidates:
candidates = sorted(
MODEL_REGISTRY,
key=lambda m: m.quality_score(profile.complexity),
reverse=True,
)
primary = max(candidates, key=lambda m: _utility(m, profile))
return failover_manager.resolve(primary, ttype)
def dispatch(
profiles: list[TaskProfile],
) -> dict[str, tuple[TaskProfile, ModelSpec, FailoverEvent | None]]:
"""
Route each sub-task independently.
Returns mapping of unique_key → (profile, selected_model, failover_event).
"""
result: dict[str, tuple[TaskProfile, ModelSpec, FailoverEvent | None]] = {}
type_counts: dict[str, int] = {}
for profile in profiles:
key = profile.task_type.value
count = type_counts.get(key, 0)
type_counts[key] = count + 1
unique_key = key if count == 0 else f"{key}_{count}"
model, failover_event = route(profile)
result[unique_key] = (profile, model, failover_event)
return result
Verifying the Flow Execution with Examples
To test the end-to-end orchestration pipeline, copy .env.example to configuration state, start your local Ollama nodes, and run the terminal test scenario:
# Launch the pre-configured demonstration script
./scripts/demo.sh
This demo triggers a complex compound request: “Write the API reference documentation AND implement the OAuth2 middleware”.
The system fragments the string into its constituent clauses, routes the documentation task to a light, high-efficiency instance (e.g., granite4.1:3b), and dynamically assigns the complex security implementation to a heavy tier (e.g., mistral-small3.2).
Conclusion
By coupling LiteLLM’s multi-provider connection layer with a self-correcting Circuit Breaker and a live Telemetry Engine, this design proves that multi-model orchestration is highly viable for production.
The architecture guarantees high availability while continuously minimizing token costs, demonstrating how automated routing workflows can maximize efficiency in the software development lifecycle.
Up Next — Part 5: Breaking the Speed Barrier with Go 🏎️
While Python’s asynchronous event loop handles our v4 dual-layer routing beautifully at moderate scale, production systems tracking hundreds of live telemetry ring buffers and real-time streaming updates face a common bottleneck: runtime synchronization overhead.
In the next part of this series, we take this exact architectural blueprint and port it entirely to Go. We will explore how replacing single-threaded event loops with native goroutines and lock-free thread-safe primitives allows us to scale routing throughput, lower memory footprints, and achieve predictable, bare-metal efficiency. Stay tuned for the code drop!
🎯 That’s a wrap 💯 for part 4 and thanks for reading!
Stay tuned for the next episode… 📺
Links
- Code repository for this post: https://github.com/aairom/multillm-taskrouting/tree/main/routing-v4
- IBM Bob: https://bob.ibm.com/
Top comments (0)