This commit is contained in:
Ahmed Hassan 2026-08-27 20:01:56 -05:00 committed by GitHub
commit f2b384dc6d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 732 additions and 0 deletions

View file

@ -0,0 +1,337 @@
"""Static production-debt analysis for a LiteLLM proxy config.yaml.
Nothing in the proxy validates a config for these patterns today:
`router_settings`/`litellm_settings` are only checked for unknown keys
against `Router.__init__`'s parameter list (see `proxy_server.py`'s
`router_settings` handling), never for whether the *combination* of
values is safe. Each of these compiles and runs fine, then turns into a
real incident once a deployment starts failing or a spend cap was never
set:
- a deployment with no fallback and no default_fallbacks: one failing
deployment means requests for that model_name just error out
- a high num_retries with no allowed_fails/allowed_fails_policy: a
persistently-failing deployment gets retried on every single
request instead of cooling down, multiplying cost and latency
- a deployment with no max_budget and no global litellm_settings
max_budget: spend on it has no automatic cutoff
- a deployment pointing at a model litellm's own pricing data already
marks as deprecated
"""
from __future__ import annotations
import datetime
import warnings
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from enum import Enum
from types import MappingProxyType
from typing import Final, TypeAlias
from pydantic import TypeAdapter
from typing_extensions import ReadOnly, TypedDict
class Severity(str, Enum):
"""Risk tier of a `Finding`.
CRITICAL: can directly cause a cost or availability incident on its
own (e.g. retries multiplying spend against a dead deployment).
WARNING: a real gap, but one a team may have intentionally covered
another way (provider-side billing alerts, accepting a hard
failure instead of a fallback).
"""
CRITICAL = "critical"
WARNING = "warning"
@dataclass(frozen=True, slots=True)
class Finding:
rule: str
severity: Severity
model_name: str | None
message: str
_SEVERITY_WEIGHT: Final[Mapping[Severity, int]] = MappingProxyType(
{Severity.CRITICAL: 10, Severity.WARNING: 3}
)
_SEVERITY_ORDER: Final[Mapping[Severity, int]] = MappingProxyType(
{Severity.CRITICAL: 0, Severity.WARNING: 1}
)
# Below this, we don't flag num_retries: a couple of retries on a
# transient error is normal and not, by itself, a cost-multiplication
# risk worth surfacing.
_HIGH_RETRY_THRESHOLD: Final = 3
class LiteLLMParamsShape(TypedDict, total=False):
model: ReadOnly[str]
num_retries: ReadOnly[int]
max_budget: ReadOnly[float]
class DeploymentShape(TypedDict):
model_name: ReadOnly[str]
litellm_params: ReadOnly[LiteLLMParamsShape]
FallbackEntry: TypeAlias = Mapping[str, Sequence[str]]
class SettingsShape(TypedDict, total=False):
fallbacks: ReadOnly[Sequence[FallbackEntry]]
context_window_fallbacks: ReadOnly[Sequence[FallbackEntry]]
content_policy_fallbacks: ReadOnly[Sequence[FallbackEntry]]
default_fallbacks: ReadOnly[Sequence[str]]
num_retries: ReadOnly[int]
allowed_fails: ReadOnly[int]
allowed_fails_policy: ReadOnly[Mapping[str, int]]
max_budget: ReadOnly[float]
class ProxyConfigShape(TypedDict, total=False):
model_list: ReadOnly[Sequence[DeploymentShape]]
litellm_settings: ReadOnly[SettingsShape]
router_settings: ReadOnly[SettingsShape]
ModelCost: TypeAlias = Mapping[str, Mapping[str, object]]
with warnings.catch_warnings():
# pydantic warns that ReadOnly (PEP 705) isn't enforced against mutation at
# runtime, which is expected here: we use it only for the static contract.
warnings.simplefilter("ignore", UserWarning)
_CONFIG_ADAPTER: Final[TypeAdapter[ProxyConfigShape]] = TypeAdapter(ProxyConfigShape)
def validate_config(raw: object) -> ProxyConfigShape:
"""Validate an arbitrary loaded config.yaml value into `ProxyConfigShape`.
`ProxyConfig.get_config` returns a loosely-typed `dict`; this is the
boundary where that untyped value is validated into the shape every
detector below assumes. Raises `pydantic.ValidationError` on a
malformed config.
"""
return _CONFIG_ADAPTER.validate_python(raw)
def production_debt_score(findings: Sequence[Finding]) -> int:
"""Severity-weighted sum of findings (critical=10, warning=3).
A simple, transparent, reproducible heuristic for trending a
config's structural health over time -- not a standardized metric.
"""
return sum(_SEVERITY_WEIGHT[f.severity] for f in findings)
def _model_names(config: ProxyConfigShape) -> frozenset[str]:
"""Real (non-wildcard) model_name values declared in model_list."""
return frozenset(
entry["model_name"]
for entry in config.get("model_list") or ()
if "*" not in entry["model_name"]
)
def _fallback_source_keys(fallback_list: Sequence[FallbackEntry] | None) -> frozenset[str]:
"""Source model_names covered by a fallbacks-shaped list.
`fallbacks`/`context_window_fallbacks`/`content_policy_fallbacks`
all share the same shape: [{"model_name": ["fallback_1", ...]}, ...]
(confirmed against Router.__init__ and the example proxy config).
"""
return frozenset(key for entry in fallback_list or () for key in entry)
def _fallback_target_names(fallback_list: Sequence[FallbackEntry] | None) -> frozenset[str]:
"""Model names appearing as a fallback *target* anywhere in the list.
A model whose sole purpose is to serve as another model's fallback
destination is not itself missing coverage just because nothing
falls back to it in turn.
"""
return frozenset(
target for entry in fallback_list or () for targets in entry.values() for target in targets
)
def _global_settings(config: ProxyConfigShape) -> SettingsShape:
"""Merge litellm_settings and router_settings.
Both accept fallbacks/num_retries/allowed_fails/max_budget (the
proxy validates router_settings keys against Router.__init__'s own
parameter list), so a value set in either place is equally real.
router_settings wins on overlap since it is the more specific,
router-scoped block.
"""
litellm_settings: Final = config.get("litellm_settings")
router_settings: Final = config.get("router_settings")
merged: Final[SettingsShape] = {
**(litellm_settings or {}),
**(router_settings or {}),
}
return merged
def detect_missing_fallback_coverage(config: ProxyConfigShape) -> tuple[Finding, ...]:
settings: Final = _global_settings(config)
if settings.get("default_fallbacks"):
# A catch-all default_fallbacks covers every model_group.
return ()
covered: Final = (
_fallback_source_keys(settings.get("fallbacks"))
| _fallback_source_keys(settings.get("context_window_fallbacks"))
| _fallback_source_keys(settings.get("content_policy_fallbacks"))
| _fallback_target_names(settings.get("fallbacks"))
| _fallback_target_names(settings.get("context_window_fallbacks"))
| _fallback_target_names(settings.get("content_policy_fallbacks"))
)
return tuple(
Finding(
rule="no-fallback-coverage",
severity=Severity.WARNING,
model_name=name,
message=(
f"model_name '{name}' has no fallbacks, context_window_fallbacks, "
"content_policy_fallbacks, or default_fallbacks entry -- if every "
"deployment behind it fails, requests for it error out with nothing "
"to fall back to"
),
)
for name in sorted(_model_names(config))
if name not in covered
)
def detect_retry_without_cooldown(config: ProxyConfigShape) -> tuple[Finding, ...]:
settings: Final = _global_settings(config)
has_cooldown: Final = (
settings.get("allowed_fails") is not None or settings.get("allowed_fails_policy") is not None
)
if has_cooldown:
return ()
global_retries: Final = settings.get("num_retries")
global_findings: Final = (
(
Finding(
rule="retry-without-cooldown",
severity=Severity.CRITICAL,
model_name=None,
message=(
f"global num_retries={global_retries} is set with no "
"allowed_fails/allowed_fails_policy configured -- a "
"persistently-failing deployment is retried on every request "
"instead of being cooled down, multiplying cost and latency "
"on every call that hits it"
),
),
)
if isinstance(global_retries, int) and global_retries >= _HIGH_RETRY_THRESHOLD
else ()
)
deployment_findings: Final = tuple(
Finding(
rule="retry-without-cooldown",
severity=Severity.CRITICAL,
model_name=entry["model_name"],
message=(
f"deployment '{entry['model_name']}' sets num_retries={retries} with "
"no allowed_fails/allowed_fails_policy configured anywhere -- a "
"persistently-failing deployment is retried on every request instead "
"of being cooled down, multiplying cost and latency on every call "
"that hits it"
),
)
for entry in config.get("model_list") or ()
for retries in (entry["litellm_params"].get("num_retries"),)
if isinstance(retries, int) and retries >= _HIGH_RETRY_THRESHOLD
)
return global_findings + deployment_findings
def detect_missing_budget_cap(config: ProxyConfigShape) -> tuple[Finding, ...]:
settings: Final = _global_settings(config)
if settings.get("max_budget") is not None:
return ()
return tuple(
Finding(
rule="missing-deployment-budget-cap",
severity=Severity.WARNING,
model_name=entry["model_name"],
message=(
f"deployment '{entry['model_name']}' has no max_budget, and no global "
"litellm_settings.max_budget is set either -- spend on this "
"deployment has no automatic cutoff"
),
)
for entry in config.get("model_list") or ()
if "*" not in entry["model_name"] and entry["litellm_params"].get("max_budget") is None
)
def _parse_date(value: str) -> datetime.date | None:
try:
return datetime.date.fromisoformat(value)
except ValueError:
return None
def detect_deprecated_model_references(
config: ProxyConfigShape,
model_cost: ModelCost,
today: datetime.date | None = None,
) -> tuple[Finding, ...]:
"""Flag deployments whose underlying model litellm's own pricing data
already marks as deprecated (`litellm.model_cost[...]["deprecation_date"]`,
a real field litellm ships and uses for cost/capability data)."""
resolved_today: Final = today if today is not None else datetime.date.today()
return tuple(
Finding(
rule="deprecated-model-reference",
severity=Severity.WARNING,
model_name=entry["model_name"],
message=(
f"deployment '{entry['model_name']}' points at '{model}', deprecated "
f"since {dep_date_str} per litellm's own model pricing data"
),
)
for entry in config.get("model_list") or ()
for model in (entry["litellm_params"].get("model"),)
if isinstance(model, str)
for info in (model_cost.get(model),)
if isinstance(info, Mapping)
for dep_date_str in (info.get("deprecation_date"),)
if isinstance(dep_date_str, str)
for dep_date in (_parse_date(dep_date_str),)
if dep_date is not None and dep_date <= resolved_today
)
def analyze(config: ProxyConfigShape, model_cost: ModelCost | None = None) -> tuple[Finding, ...]:
"""Run all detectors and return every finding, most-critical first.
`model_cost` is optional (pass `litellm.model_cost` for the
deprecated-model-reference check); the other three detectors need
only the config itself.
"""
deprecated_findings: Final = (
detect_deprecated_model_references(config, model_cost) if model_cost is not None else ()
)
findings: Final = (
*detect_retry_without_cooldown(config),
*detect_missing_budget_cap(config),
*detect_missing_fallback_coverage(config),
*deprecated_findings,
)
return tuple(sorted(findings, key=lambda f: (_SEVERITY_ORDER[f.severity], f.rule, f.model_name or "")))

View file

@ -239,6 +239,35 @@ class ProxyInitializationHelpers:
)
print(completion_response)
@staticmethod
def _run_config_lint(config_path: str | None) -> None:
"""Check --config for production-debt gaps and print findings, without starting the server."""
import asyncio
from litellm.proxy.production_debt_linter import Severity, analyze, production_debt_score, validate_config
from litellm.proxy.proxy_server import ProxyConfig
if config_path is None:
raise click.UsageError("--lint_config requires --config <path to config.yaml>")
proxy_config: Final = ProxyConfig()
loaded_config: Final = asyncio.run(proxy_config.get_config(config_file_path=config_path))
validated_config: Final = validate_config(loaded_config)
findings: Final = analyze(validated_config, model_cost=litellm.model_cost)
if not findings:
click.echo(f"\nLiteLLM: no production-debt findings in {config_path}\n")
return
click.echo(f"\nLiteLLM: production-debt findings in {config_path}\n")
for finding in findings:
target = finding.model_name or "<global>"
click.echo(f" [{finding.severity.value.upper()}] {finding.rule} ({target}): {finding.message}")
click.echo(f"\nproduction_debt_score={production_debt_score(findings)}\n")
if any(f.severity is Severity.CRITICAL for f in findings):
sys.exit(1)
@staticmethod
def _get_default_unvicorn_init_args(
host: str,
@ -759,6 +788,16 @@ class ProxyInitializationHelpers:
default=False,
help="Run the interactive setup wizard to configure providers and generate a config file",
)
@click.option(
"--lint_config",
is_flag=True,
default=False,
help=(
"Check --config for production-debt gaps (deployments with no fallback "
"coverage, retries with no cooldown, no budget cap, or a deprecated model) "
"and exit without starting the server"
),
)
@click.option(
"--version",
"-v",
@ -961,6 +1000,7 @@ def run_server(
use_queue,
health,
setup,
lint_config,
version,
run_gunicorn,
run_hypercorn,
@ -1039,6 +1079,9 @@ def run_server(
if test is True:
ProxyInitializationHelpers._run_test_chat_completion(host, port, model, test)
return
if lint_config is True:
ProxyInitializationHelpers._run_config_lint(config)
return
else:
if headers:
headers = json.loads(headers)

View file

@ -0,0 +1,352 @@
"""Tests + labelled benchmark corpus for litellm.proxy.production_debt_linter.
Each corpus config is labelled with exactly the rule names it should (for
adversarial configs) or should not (for clean configs) trigger. The
parametrized tests below turn that corpus into a real, reproducible
recall/false-positive measurement rather than an asserted claim:
- test_adversarial_config_detected: every injected defect must be
caught (recall).
- test_clean_config_has_no_findings: no clean, realistic config may
produce a finding (false positives).
"""
from __future__ import annotations
import datetime
import pytest
from litellm.proxy.production_debt_linter import (
Finding,
Severity,
analyze,
detect_deprecated_model_references,
detect_missing_budget_cap,
detect_missing_fallback_coverage,
detect_retry_without_cooldown,
production_debt_score,
)
# A small, fixed model_cost stand-in so the deprecated-model tests don't
# drift with the real dataset's dates. The real litellm.model_cost
# integration is exercised separately below.
FAKE_MODEL_COST = {
"openai/gpt-4o": {"deprecation_date": "2099-01-01"},
"openai/old-model": {"deprecation_date": "2020-01-01"},
}
FIXED_TODAY = datetime.date(2026, 1, 1)
def _deployment(model_name: str, model: str, **extra_params) -> dict:
return {
"model_name": model_name,
"litellm_params": {"model": model, **extra_params},
}
# ---------------------------------------------------------------------------
# Clean corpus: realistic configs that must produce zero findings.
# ---------------------------------------------------------------------------
def clean_single_model_with_fallback_and_budget() -> dict:
return {
"model_list": [
_deployment("gpt-4o", "openai/gpt-4o", max_budget=100.0),
_deployment("gpt-4o-mini", "openai/gpt-4o-mini", max_budget=50.0),
],
"litellm_settings": {
"fallbacks": [{"gpt-4o": ["gpt-4o-mini"]}],
},
}
def clean_global_budget_and_default_fallbacks() -> dict:
return {
"model_list": [
_deployment("gpt-4o", "openai/gpt-4o"),
_deployment("claude", "anthropic/claude-3-5-sonnet"),
],
"litellm_settings": {
"max_budget": 500.0,
"default_fallbacks": ["gpt-4o"],
},
}
def clean_bounded_retries_with_allowed_fails() -> dict:
return {
"model_list": [
_deployment("gpt-4o", "openai/gpt-4o", max_budget=100.0, num_retries=5),
],
"router_settings": {
"allowed_fails": 3,
"fallbacks": [{"gpt-4o": ["gpt-4o"]}],
},
}
def clean_low_retry_count_needs_no_cooldown() -> dict:
return {
"model_list": [
_deployment("gpt-4o", "openai/gpt-4o", max_budget=100.0, num_retries=1),
],
"litellm_settings": {
"default_fallbacks": ["gpt-4o"],
},
}
def clean_context_window_fallback_counts_as_coverage() -> dict:
return {
"model_list": [
_deployment("gpt-3.5-turbo", "openai/gpt-3.5-turbo", max_budget=10.0),
_deployment("gpt-3.5-turbo-large", "openai/gpt-4.1", max_budget=10.0),
],
"litellm_settings": {
"context_window_fallbacks": [
{"gpt-3.5-turbo": ["gpt-3.5-turbo-large"]}
],
},
}
CLEAN_CORPUS = {
"single_model_with_fallback_and_budget": clean_single_model_with_fallback_and_budget,
"global_budget_and_default_fallbacks": clean_global_budget_and_default_fallbacks,
"bounded_retries_with_allowed_fails": clean_bounded_retries_with_allowed_fails,
"low_retry_count_needs_no_cooldown": clean_low_retry_count_needs_no_cooldown,
"context_window_fallback_counts_as_coverage": clean_context_window_fallback_counts_as_coverage,
}
# ---------------------------------------------------------------------------
# Adversarial corpus: one deliberately injected defect each, labelled with
# the exact rule name that must fire.
# ---------------------------------------------------------------------------
def bad_no_fallback_coverage() -> dict:
return {
"model_list": [
_deployment("gpt-4o", "openai/gpt-4o", max_budget=100.0),
],
"litellm_settings": {},
}
def bad_high_retries_no_cooldown_global() -> dict:
return {
"model_list": [
_deployment("gpt-4o", "openai/gpt-4o", max_budget=100.0),
],
"litellm_settings": {
"num_retries": 5,
"default_fallbacks": ["gpt-4o"],
},
}
def bad_high_retries_no_cooldown_deployment() -> dict:
return {
"model_list": [
_deployment(
"gpt-4o", "openai/gpt-4o", max_budget=100.0, num_retries=10
),
],
"litellm_settings": {
"default_fallbacks": ["gpt-4o"],
},
}
def bad_missing_budget_cap() -> dict:
return {
"model_list": [
_deployment("gpt-4o", "openai/gpt-4o"),
],
"litellm_settings": {
"default_fallbacks": ["gpt-4o"],
},
}
def bad_deprecated_model_reference() -> dict:
return {
"model_list": [
_deployment(
"legacy", "openai/old-model", max_budget=10.0
),
],
"litellm_settings": {
"default_fallbacks": ["legacy"],
},
}
ADVERSARIAL_CORPUS = {
"no_fallback_coverage": (bad_no_fallback_coverage, "no-fallback-coverage"),
"high_retries_no_cooldown_global": (
bad_high_retries_no_cooldown_global,
"retry-without-cooldown",
),
"high_retries_no_cooldown_deployment": (
bad_high_retries_no_cooldown_deployment,
"retry-without-cooldown",
),
"missing_budget_cap": (bad_missing_budget_cap, "missing-deployment-budget-cap"),
"deprecated_model_reference": (
bad_deprecated_model_reference,
"deprecated-model-reference",
),
}
# ---------------------------------------------------------------------------
# Benchmark tests
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("name", CLEAN_CORPUS)
def test_clean_config_has_no_findings(name: str) -> None:
config = CLEAN_CORPUS[name]()
findings = analyze(config, model_cost=FAKE_MODEL_COST)
assert findings == (), f"{name} produced unexpected findings: {findings}"
@pytest.mark.parametrize("name", ADVERSARIAL_CORPUS)
def test_adversarial_config_detected(name: str) -> None:
factory, expected_rule = ADVERSARIAL_CORPUS[name]
config = factory()
findings = analyze(config, model_cost=FAKE_MODEL_COST)
rules = {f.rule for f in findings}
assert expected_rule in rules, f"{name}: expected '{expected_rule}' not in {rules}"
def test_benchmark_summary(capsys: pytest.CaptureFixture[str]) -> None:
"""Not an assertion -- prints the real recall/false-positive numbers.
Run with `-s` to see the summary; the two tests above are what
actually enforce recall == 100% and false positives == 0 per-case.
"""
total_clean = len(CLEAN_CORPUS)
fp_count = sum(
1
for factory in CLEAN_CORPUS.values()
if analyze(factory(), model_cost=FAKE_MODEL_COST)
)
total_adversarial = len(ADVERSARIAL_CORPUS)
detected = 0
for factory, expected_rule in ADVERSARIAL_CORPUS.values():
rules = {f.rule for f in analyze(factory(), model_cost=FAKE_MODEL_COST)}
if expected_rule in rules:
detected += 1
print(
f"\nbenchmark: recall={detected}/{total_adversarial} "
f"false_positives={fp_count}/{total_clean}"
)
# ---------------------------------------------------------------------------
# Unit-level tests for individual detectors and the score function
# ---------------------------------------------------------------------------
def test_production_debt_score_weights() -> None:
findings = [
Finding("r1", Severity.CRITICAL, None, "x"),
Finding("r2", Severity.WARNING, None, "y"),
]
assert production_debt_score(findings) == 10 + 3
assert production_debt_score([]) == 0
def test_no_fallback_coverage_ignores_wildcard_models() -> None:
config = {
"model_list": [_deployment("*", "openai/*", max_budget=100.0)],
"litellm_settings": {"max_budget": 100.0},
}
assert detect_missing_fallback_coverage(config) == ()
def test_retry_without_cooldown_respects_allowed_fails_policy() -> None:
config = {
"model_list": [
_deployment("gpt-4o", "openai/gpt-4o", num_retries=10),
],
"litellm_settings": {
"allowed_fails_policy": {"AuthenticationErrorRetries": 0},
},
}
assert detect_retry_without_cooldown(config) == ()
def test_missing_budget_cap_respects_global_budget() -> None:
config = {
"model_list": [_deployment("gpt-4o", "openai/gpt-4o")],
"router_settings": {"max_budget": 1000.0},
}
assert detect_missing_budget_cap(config) == ()
def test_deprecated_model_reference_only_flags_past_dates() -> None:
config = {
"model_list": [
_deployment("current", "openai/gpt-4o"),
_deployment("legacy", "openai/old-model"),
],
}
findings = detect_deprecated_model_references(
config, FAKE_MODEL_COST, today=FIXED_TODAY
)
assert len(findings) == 1
assert findings[0].model_name == "legacy"
def test_analyze_sorts_critical_first() -> None:
config = {
"model_list": [
_deployment("gpt-4o", "openai/gpt-4o", num_retries=10),
],
"litellm_settings": {"default_fallbacks": ["gpt-4o"]},
}
findings = analyze(config)
assert findings[0].severity == Severity.CRITICAL
def test_analyze_works_without_model_cost() -> None:
# model_cost is optional; the other three detectors don't need it.
config = {"model_list": [_deployment("gpt-4o", "openai/gpt-4o")]}
findings = analyze(config)
assert all(f.rule != "deprecated-model-reference" for f in findings)
def test_real_litellm_model_cost_integration() -> None:
"""Exercises the real litellm.model_cost data, not the fake stand-in,
to prove the integration point (not just the detector logic) works."""
import litellm
deprecated_model = next(
(
model
for model, info in litellm.model_cost.items()
if isinstance(info, dict)
and isinstance(info.get("deprecation_date"), str)
and info["deprecation_date"] not in ("", "date when the model becomes deprecated in the format YYYY-MM-DD")
),
None,
)
assert deprecated_model is not None, "expected at least one model with a real deprecation_date in litellm.model_cost"
config = {
"model_list": [_deployment("legacy", deprecated_model, max_budget=1.0)],
"litellm_settings": {"default_fallbacks": ["legacy"]},
}
findings = detect_deprecated_model_references(
config, litellm.model_cost, today=datetime.date(2099, 1, 1)
)
assert len(findings) == 1
assert findings[0].model_name == "legacy"