cascade-llm 0.1.0

LLM Pipeline Query Optimizer — automatic model routing, cost optimization, and quality-aware escalation.

Cascade — LLM Pipeline Query Optimizer

Automatic model routing, cost optimization, and quality-aware escalation for LLM pipelines.

Cascade is the "query planner" for LLM systems. Just as a database query optimizer finds the cheapest execution plan that returns correct results, Cascade selects the cheapest model + prompt configuration that meets your quality threshold — per request, in real time.

Key Features

• Zero-Config Setup — Name the models you want (["gpt-4o-mini", "gpt-4o"]); prices come from a built-in catalog
• One-Line API — client.ask("...") for the simplest path, or the drop-in OpenAI-compatible interface
• Cost vs. Quality Modes — Per request, choose the cheapest good-enough model or the best model regardless of cost
• Budget-Aware Routing — max_cost_usd ceilings and min_tier floors for "best model under budget"
• Automatic Model Routing — Routes each request to the cheapest model tier likely to succeed
• Quality-Aware Escalation — Automatically cascades to a more capable model if quality is insufficient
• Thompson Sampling Planner — Online learning that improves routing decisions over time
• Difficulty Estimation — Sub-2ms request complexity scoring
• Cost Analytics — Full cost attribution, savings reports, and tier performance tracking
• HTTP Gateway — Optional FastAPI server for language-agnostic access

Installation

pip install -e .                    # Core
pip install -e ".[server]"          # + FastAPI gateway
pip install -e ".[all]"             # Everything
pip install -e ".[dev]"             # + test dependencies

Quick Start

from cascade import CascadeClient

# Just name the models — prices come from the built-in catalog.
client = CascadeClient(models=["gpt-4o-mini", "gpt-4o", "o1"])

# One-liner: ask and get an intelligently routed answer.
response = client.ask("Summarize this quarterly report...")

print(response.content)
print(response.cascade_metadata.model_used)   # e.g. 'gpt-4o-mini'
print(response.cascade_metadata.cost_usd)      # what you actually paid

That's the whole setup. Everything below is optional.

# Prefer the OpenAI-style interface? It's a drop-in — no other code changes:
response = client.chat.completions.create(
    messages=[{"role": "user", "content": "Summarize this report..."}],
    cascade_labels={"feature": "report_summary"},
)

print(response.cascade_metadata)
# model_used='gpt-4o-mini', quality_score=0.91, cost_usd=0.0003, savings_pct=98.8

Need exact prices or a custom model? Pass pricing explicitly (it overrides the catalog), or register a model once and reuse it by name:

from cascade import CascadeClient, register_model

register_model("my-llm", cost_per_1k_input=0.001, cost_per_1k_output=0.002)
client = CascadeClient(models=[
    "gpt-4o-mini",                                              # from catalog
    {"name": "gpt-4o", "cost_per_1k_input": 0.0025, "cost_per_1k_output": 0.01},  # explicit
    "my-llm",                                                   # registered above
])

Catalog prices are approximate public list prices and may be out of date — override them for billing-accurate routing. See cascade.available_models(), and cascade.pricing_updated_on() for when the catalog was last reviewed.

Routing Modes: Cost vs. Quality

Every request can pick its objective. Use cost mode (the default) when you want the cheapest model that still clears your quality bar, or quality mode when you want the most capable model and don't care about token spend.

# Cost-first (default): cheapest tier that meets the quality threshold,
# escalating only if needed.
response = client.chat.completions.create(
    messages=[{"role": "user", "content": "Summarize this email..."}],
    mode="cost",          # aliases: "cheap", "economy"
)

# Quality-first: route straight to the most capable tier, ignore cost.
response = client.chat.completions.create(
    messages=[{"role": "user", "content": "Draft a board-level strategy memo..."}],
    mode="quality",       # aliases: "best", "max", "premium"
)

print(response.cascade_metadata.routing_mode)  # 'cost' or 'quality'

You can also set the default for every request on the policy:

from cascade import QualityPolicy, RoutingMode

quality_policy = QualityPolicy(
    min_quality_score=0.85,
    mode=RoutingMode.QUALITY,   # all requests default to best-model routing
)

A per-request mode= always overrides the policy default. In quality mode the planner skips the cheap tiers and goes directly to the most capable model, so escalations is always 0 and savings_pct reflects that you opted out of cost savings in exchange for the best answer.

Budget-Aware Routing (best model under a cap)

For the middle ground, add hard per-request guardrails — no extra setup required:

• max_cost_usd — never spend more than this on a request.
• min_tier — never route below this model (a quality floor for critical traffic).

# "Give me the best answer you can for under 1 cent."
client.ask("Explain this contract clause", mode="quality", max_cost_usd=0.01)

# "Optimize for cost, but never use anything weaker than gpt-4o."
client.ask("Review this medical summary", min_tier="gpt-4o")

# Combine both: the best model at/above gpt-4o that still fits the budget.
client.ask("Draft the SLA section", mode="quality", min_tier="gpt-4o", max_cost_usd=0.05)

If the budget can't be met, Cascade falls back to the cheapest tier that satisfies min_tier and sets response.cascade_metadata.budget_exceeded = True so you can detect it. When min_tier and max_cost_usd conflict, min_tier (a correctness floor) wins. These options work on both ask() and chat.completions.create().

How It Works

Request ──▶ Difficulty Estimator (<2ms)
                │
                ▼
         Pipeline Planner (Thompson Sampling)
                │
                ▼
        ┌── Execute with cheapest viable tier
        │       │
        │   Quality Verifier
        │       │
        │   Quality OK? ──YES──▶ Return response
        │       │
        │      NO
        │       │
        └── Escalate to next tier (up to max_escalations)
                │
                ▼
         Update bandit state with outcome

1. Difficulty Estimator — Scores request complexity (0-1) using lightweight heuristic features: length, vocabulary richness, domain keywords, code presence, conversation depth.

2. Cascade Planner — Thompson Sampling bandit selects the cheapest model tier whose sampled success probability exceeds the difficulty-adjusted quality threshold.

3. Quality Verifier — Fast output scoring combining structural heuristics, optional user-registered evaluators, and optional LLM-as-judge.

4. Escalation — If quality falls below threshold, automatically retries with the next-tier model. Max escalations are configurable.

5. Feedback Loop — Every outcome updates the bandit's Beta distribution, so routing improves over time.

Custom Quality Evaluators

@client.quality_evaluator("domain_eval")
def evaluate(input_messages, output, model_used) -> float:
    # Return 0.0–1.0 quality score
    if "error" in output.lower():
        return 0.2
    return 0.9

Analytics

from cascade.analytics import Analytics

analytics = Analytics(client.telemetry)

# Cost breakdown by model
analytics.cost_report(group_by=["model"])

# How much Cascade saved vs. always using the most expensive model
analytics.savings_report()

HTTP Gateway

uvicorn cascade.gateway:app --host 0.0.0.0 --port 8000

Endpoints:

• POST /v1/chat/completions — Route a chat request
• GET /v1/stats — Planner bandit state
• GET /v1/analytics/savings — Savings report
• GET /v1/analytics/cost?group_by=model — Cost breakdown
• GET /health — Health check

Request body (POST /v1/chat/completions) — only messages is required:

{
  "messages": [{"role": "user", "content": "Summarize this report..."}],
  "mode": "quality",
  "max_cost_usd": 0.01,
  "min_tier": "gpt-4o",
  "cascade_labels": {"feature": "report_summary"},
  "temperature": 0.7,
  "max_tokens": 800
}

The response includes the same cascade_metadata as the Python client (model_used, routing_mode, cost_usd, budget_exceeded, savings_pct, …).

Running Tests

pip install -e ".[dev]"
pytest tests/ -v

Architecture

cascade/
├── __init__.py          # Public API exports
├── models.py            # Data models (ModelTier, QualityPolicy, RoutingMode, etc.)
├── catalog.py           # Built-in model price catalog (route by name)
├── estimator.py         # Difficulty Estimator
├── planner.py           # Thompson Sampling Planner
├── verifier.py          # Quality Verifier
├── engine.py            # Execution Engine (orchestration + escalation)
├── client.py            # CascadeClient (public API)
├── analytics.py         # Cost attribution & savings reports
├── gateway.py           # FastAPI HTTP gateway
└── storage/
    ├── __init__.py
    └── memory_store.py  # In-memory telemetry store

Configuration

Parameter	Default	Description
min_quality_score	0.85	Minimum acceptable quality (0-1)
max_escalations	2	Max tier escalations per request
latency_budget_ms	10000	Total latency budget before giving up
exploration_rate	0.05	Fraction of requests used for exploration
mode	cost	Default routing objective: cost (cheapest viable) or quality (best model)

Per-Request Options

Passed to client.ask(...) or client.chat.completions.create(...) — all optional:

Option	Description
mode	"cost" (cheapest viable) or "quality" (best model); overrides the policy default
max_cost_usd	Hard ceiling on estimated request cost
min_tier	Model-name floor; never route below this tier
cascade_labels / labels	Tags for cost attribution in analytics

License

MIT