The Complete Production SOP: How to Deploy and Run Multi-Model Fallback Cascade & Cost Routing in Autonomous AI Agents

---
name: "multi-model-fallback-cascade"
description: "Route queries to lightweight models first, escalating to frontier models only on complexity."
metadata:
  author: "RouteLLM"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "setup multi-model cascade"
  - "route query by complexity"
  - "fallback from haiku to opus"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

from typing import Any, Type
from pydantic import BaseModel, ValidationError

class CascadeRouter:
    def __init__(self, primary_client: Any, frontier_client: Any):
        self.primary = primary_client
        self.frontier = frontier_client

    def execute(self, prompt: str, schema: Type[BaseModel]) -> BaseModel:
        try:
            output = self.primary.complete(prompt)
            return schema.model_validate_json(output)
        except (ValidationError, Exception) as err:
            escalated = f"{prompt}\n\nValidation failed: {err}. Recompute with exact formatting."
            return schema.model_validate_json(self.frontier.complete(escalated))

Please route query by complexity for processing support_tickets.csv. Use Claude Haiku as the baseline tier to extract structured JSON entities, and fallback from haiku to opus if regex/schema validation fails or confidence score < 0.85.

import pytest
from pydantic import BaseModel
from src.model_router import CascadeRouter

class SummarySchema(BaseModel):
    summary: str
    severity: int

class MockClient:
    def __init__(self, response: str):
        self.response = response
    def complete(self, prompt: str) -> str:
        return self.response

def test_cascade_escalation_on_invalid_json():
    haiku_mock = MockClient("Malformed output without JSON")
    opus_mock = MockClient('{"summary": "Critical outage", "severity": 1}')
    
    router = CascadeRouter(primary_client=haiku_mock, frontier_client=opus_mock)
    result = router.execute("Parse log", SummarySchema)
    
    assert result.severity == 1
    assert result.summary == "Critical outage"