mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-09 22:31:41 +00:00
6a0d03914c
* test: drop the cwd-relative sys.path.insert calls from the test suite
TQ003 stands at 1,077 across 1,058 files, and 1,015 of them are the same shape:
sys.path.insert(0, os.path.abspath("../..")) and its deeper siblings. The
argument resolves against the working directory rather than the file, so from
the repo root, where every job runs pytest, it inserts the directory two levels
above the checkout. It has never pointed at litellm. The package is installed
into the environment anyway, which is what actually makes the import work, and
what the rule's message has said all along.
Removing them leaves 1,634 imports of sys and os with no remaining reference,
and those go too, except where another test module imports the name back out of
the file. The rest of TQ003 is 62 call sites that resolve against __file__ or a
variable, which are a different question and are left alone.
Collection is identical either way: 45,871 tests and the same 51 pre-existing
collection errors before and after, and ruff reports no new undefined name.
* test: drop the duplicate imports the sys.path sweep exposed to F811
* test(pre-call-utils): restore the os import the new bedrock tests need
1132 lines
41 KiB
Python
1132 lines
41 KiB
Python
"""
|
|
Tests for the QualityRouter.
|
|
|
|
Covers:
|
|
- Tier index construction from `model_info.litellm_routing_preferences`.
|
|
- Quality-tier resolution (exact, round-up, default fallback).
|
|
- Keyword override (match, tiebreaking by quality + price).
|
|
- Pre-routing hook end-to-end.
|
|
- Decision metadata stash + Router.set_response_headers lift.
|
|
"""
|
|
|
|
from typing import Any, Dict, List
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
|
|
from litellm.router_strategy.quality_router.config import (
|
|
DEFAULT_COMPLEXITY_TO_QUALITY,
|
|
)
|
|
from litellm.router_strategy.quality_router.quality_router import QualityRouter
|
|
|
|
|
|
def _make_model_list(spec: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
"""
|
|
Build a router model_list from a compact spec.
|
|
|
|
spec entry shape: {
|
|
"model_name": str,
|
|
"quality_tier": Optional[int],
|
|
"keywords": Optional[List[str]],
|
|
"order": Optional[int],
|
|
"input_cost_per_token": Optional[float],
|
|
}
|
|
If quality_tier is None, the deployment is created without
|
|
`litellm_routing_preferences`.
|
|
"""
|
|
out: List[Dict[str, Any]] = []
|
|
for entry in spec:
|
|
model_info: Dict[str, Any] = {"id": f"id-{entry['model_name']}"}
|
|
if entry.get("quality_tier") is not None:
|
|
prefs: Dict[str, Any] = {"quality_tier": entry["quality_tier"]}
|
|
if "keywords" in entry:
|
|
prefs["keywords"] = entry["keywords"]
|
|
if "order" in entry:
|
|
prefs["order"] = entry["order"]
|
|
model_info["litellm_routing_preferences"] = prefs
|
|
if "input_cost_per_token" in entry:
|
|
model_info["input_cost_per_token"] = entry["input_cost_per_token"]
|
|
out.append(
|
|
{
|
|
"model_name": entry["model_name"],
|
|
"litellm_params": {"model": f"openai/{entry['model_name']}"},
|
|
"model_info": model_info,
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
@pytest.fixture
|
|
def four_tier_model_list() -> List[Dict[str, Any]]:
|
|
"""A standard haiku(1)/sonnet(2)/opus(3)/opus-next(4) model list."""
|
|
return _make_model_list(
|
|
[
|
|
{"model_name": "haiku", "quality_tier": 1},
|
|
{"model_name": "sonnet", "quality_tier": 2},
|
|
{"model_name": "opus", "quality_tier": 3},
|
|
{"model_name": "opus-next", "quality_tier": 4},
|
|
]
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_router(four_tier_model_list):
|
|
"""A MagicMock router preloaded with the four-tier model list."""
|
|
router = MagicMock()
|
|
router.model_list = four_tier_model_list
|
|
return router
|
|
|
|
|
|
@pytest.fixture
|
|
def quality_router(mock_router) -> QualityRouter:
|
|
"""Default QualityRouter wired to all four tiers."""
|
|
config = {
|
|
"available_models": ["haiku", "sonnet", "opus", "opus-next"],
|
|
"complexity_to_quality": DEFAULT_COMPLEXITY_TO_QUALITY,
|
|
}
|
|
return QualityRouter(
|
|
model_name="quality-router-test",
|
|
litellm_router_instance=mock_router,
|
|
default_model="haiku",
|
|
quality_router_config=config,
|
|
)
|
|
|
|
|
|
# ─── Tier index ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestTierIndex:
|
|
def test_builds_correct_tier_to_models_map(self, quality_router):
|
|
assert quality_router._tier_to_models == {
|
|
1: ["haiku"],
|
|
2: ["sonnet"],
|
|
3: ["opus"],
|
|
4: ["opus-next"],
|
|
}
|
|
|
|
def test_ignores_models_not_in_available_models(self, four_tier_model_list):
|
|
# Add a model the config doesn't list — it should be ignored.
|
|
extra = _make_model_list([{"model_name": "ghost", "quality_tier": 5}])
|
|
router = MagicMock()
|
|
router.model_list = four_tier_model_list + extra
|
|
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="haiku",
|
|
quality_router_config={
|
|
"available_models": ["haiku", "sonnet", "opus", "opus-next"]
|
|
},
|
|
)
|
|
|
|
for models in qr._tier_to_models.values():
|
|
assert "ghost" not in models
|
|
|
|
def test_raises_when_routing_preferences_missing(self):
|
|
# `sonnet` is in available_models but has no preferences.
|
|
ml = _make_model_list(
|
|
[
|
|
{"model_name": "haiku", "quality_tier": 1},
|
|
{"model_name": "sonnet", "quality_tier": None},
|
|
]
|
|
)
|
|
router = MagicMock()
|
|
router.model_list = ml
|
|
|
|
# Construction succeeds (tier index is lazy); the error surfaces on
|
|
# first use so the router entry doesn't have to appear after all of
|
|
# its referenced models in config.yaml.
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="haiku",
|
|
quality_router_config={"available_models": ["haiku", "sonnet"]},
|
|
)
|
|
with pytest.raises(ValueError, match="sonnet"):
|
|
_ = qr._tier_to_models
|
|
|
|
|
|
# ─── Resolve model for quality tier ─────────────────────────────────────────
|
|
|
|
|
|
class TestResolveModelForQualityTier:
|
|
def test_exact_match(self, quality_router):
|
|
assert quality_router._resolve_model_for_quality_tier(2) == "sonnet"
|
|
assert quality_router._resolve_model_for_quality_tier(4) == "opus-next"
|
|
|
|
def test_rounds_up_when_tier_missing(self, mock_router):
|
|
# Available tiers: 1, 3, 4. Asking for 2 should round up to 3.
|
|
spec = [
|
|
{"model_name": "haiku", "quality_tier": 1},
|
|
{"model_name": "opus", "quality_tier": 3},
|
|
{"model_name": "opus-next", "quality_tier": 4},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="haiku",
|
|
quality_router_config={"available_models": ["haiku", "opus", "opus-next"]},
|
|
)
|
|
|
|
assert qr._resolve_model_for_quality_tier(2) == "opus"
|
|
|
|
def test_rounds_down_when_no_higher_tier_exists(self):
|
|
# Only tier 1 available. Asking for tier 4 rounds up (nothing), then
|
|
# rounds DOWN to the closest lower tier — tier 1.
|
|
spec = [{"model_name": "haiku", "quality_tier": 1}]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="emergency-default",
|
|
quality_router_config={"available_models": ["haiku"]},
|
|
)
|
|
|
|
assert qr._resolve_model_for_quality_tier(4) == "haiku"
|
|
|
|
def test_rounds_down_prefers_closest_lower_tier(self):
|
|
# Available: 1, 2. Asking for 4 rounds down to tier 2 (not tier 1).
|
|
spec = [
|
|
{"model_name": "haiku", "quality_tier": 1},
|
|
{"model_name": "sonnet", "quality_tier": 2},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="emergency-default",
|
|
quality_router_config={"available_models": ["haiku", "sonnet"]},
|
|
)
|
|
|
|
assert qr._resolve_model_for_quality_tier(4) == "sonnet"
|
|
|
|
def test_prefers_round_up_over_round_down(self):
|
|
# Available: 1, 3. Asking for 2 rounds UP to 3, not DOWN to 1.
|
|
spec = [
|
|
{"model_name": "haiku", "quality_tier": 1},
|
|
{"model_name": "opus", "quality_tier": 3},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="emergency-default",
|
|
quality_router_config={"available_models": ["haiku", "opus"]},
|
|
)
|
|
|
|
assert qr._resolve_model_for_quality_tier(2) == "opus"
|
|
|
|
|
|
# ─── RoutingPreferences validation ─────────────────────────────────────────
|
|
|
|
|
|
class TestRoutingPreferencesValidation:
|
|
def test_invalid_quality_tier_type_raises_clear_error(self):
|
|
# quality_tier must be an int — pass a non-coercible string.
|
|
ml = [
|
|
{
|
|
"model_name": "haiku",
|
|
"litellm_params": {"model": "openai/gpt-4o-mini"},
|
|
"model_info": {
|
|
"id": "id-haiku",
|
|
"litellm_routing_preferences": {"quality_tier": "not-an-int"},
|
|
},
|
|
}
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = ml
|
|
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="haiku",
|
|
quality_router_config={"available_models": ["haiku"]},
|
|
)
|
|
with pytest.raises(ValueError, match="invalid litellm_routing_preferences"):
|
|
_ = qr._tier_to_models
|
|
|
|
|
|
# ─── Config-ordering independence (lazy index build) ───────────────────────
|
|
|
|
|
|
class TestConfigOrderingIndependence:
|
|
def test_router_can_be_instantiated_before_its_targets_exist(self):
|
|
# Build a router instance whose referenced model_list is EMPTY at
|
|
# construction time (simulating a config where the router entry
|
|
# appears before its target deployments). The tier index must not be
|
|
# built eagerly — it's deferred until first use.
|
|
router = MagicMock()
|
|
router.model_list = [] # <- targets haven't been added yet
|
|
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="haiku",
|
|
quality_router_config={"available_models": ["haiku", "sonnet", "opus"]},
|
|
)
|
|
|
|
# Now the targets come online. This mirrors the incremental add by
|
|
# `Router._create_deployment`.
|
|
router.model_list = _make_model_list(
|
|
[
|
|
{"model_name": "haiku", "quality_tier": 1},
|
|
{"model_name": "sonnet", "quality_tier": 2},
|
|
{"model_name": "opus", "quality_tier": 3},
|
|
]
|
|
)
|
|
|
|
# First access triggers the index build and sees the full list.
|
|
assert qr._tier_to_models == {
|
|
1: ["haiku"],
|
|
2: ["sonnet"],
|
|
3: ["opus"],
|
|
}
|
|
|
|
|
|
# ─── Router.set_model_list resets quality_routers (hot reload) ─────────────
|
|
|
|
|
|
class TestSetModelListResetsQualityRouters:
|
|
def test_set_model_list_clears_quality_routers_registry(self):
|
|
from litellm.router import Router
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "haiku",
|
|
"litellm_params": {
|
|
"model": "openai/gpt-4o-mini",
|
|
"api_key": "sk-test",
|
|
},
|
|
"model_info": {"litellm_routing_preferences": {"quality_tier": 1}},
|
|
},
|
|
{
|
|
"model_name": "my-qr",
|
|
"litellm_params": {
|
|
"model": "auto_router/quality_router",
|
|
"quality_router_default_model": "haiku",
|
|
"quality_router_config": {"available_models": ["haiku"]},
|
|
},
|
|
},
|
|
]
|
|
)
|
|
|
|
assert "my-qr" in router.quality_routers
|
|
|
|
# Hot-reload with a new model_list that doesn't define the router.
|
|
router.set_model_list(
|
|
[
|
|
{
|
|
"model_name": "haiku",
|
|
"litellm_params": {
|
|
"model": "openai/gpt-4o-mini",
|
|
"api_key": "sk-test",
|
|
},
|
|
}
|
|
]
|
|
)
|
|
|
|
# Stale router from before must be cleared.
|
|
assert "my-qr" not in router.quality_routers
|
|
|
|
|
|
# ─── Pre-routing hook ───────────────────────────────────────────────────────
|
|
|
|
|
|
class TestPreRoutingHook:
|
|
@pytest.mark.asyncio
|
|
async def test_simple_message_routes_to_tier_1(self, quality_router):
|
|
messages = [{"role": "user", "content": "hi"}]
|
|
resp = await quality_router.async_pre_routing_hook(
|
|
model="quality-router-test",
|
|
request_kwargs={},
|
|
messages=messages,
|
|
)
|
|
assert resp is not None
|
|
assert resp.model == "haiku"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reasoning_message_routes_to_tier_4(self, quality_router):
|
|
# Two reasoning markers triggers ComplexityTier.REASONING → quality 4.
|
|
messages = [
|
|
{
|
|
"role": "user",
|
|
"content": (
|
|
"Think step by step and reason through this problem. "
|
|
"Analyze this carefully and break down each component."
|
|
),
|
|
}
|
|
]
|
|
resp = await quality_router.async_pre_routing_hook(
|
|
model="quality-router-test",
|
|
request_kwargs={},
|
|
messages=messages,
|
|
)
|
|
assert resp is not None
|
|
assert resp.model == "opus-next"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_messages_returns_none(self, quality_router):
|
|
resp = await quality_router.async_pre_routing_hook(
|
|
model="quality-router-test",
|
|
request_kwargs={},
|
|
messages=[],
|
|
)
|
|
assert resp is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_only_system_message_routes_to_default(self, quality_router):
|
|
messages = [{"role": "system", "content": "You are a helpful assistant."}]
|
|
resp = await quality_router.async_pre_routing_hook(
|
|
model="quality-router-test",
|
|
request_kwargs={},
|
|
messages=messages,
|
|
)
|
|
assert resp is not None
|
|
assert resp.model == "haiku" # the configured default_model
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trivial_message_not_escalated_by_agent_system_prompt(self, quality_router):
|
|
"""QualityRouter delegates to ComplexityRouter's shared scorer
|
|
(`self._scorer.classify`), so a system-prompt scoring bug there is inherited here
|
|
too. A real agent-harness system prompt (tool-use rules, git workflow, markdown
|
|
formatting -- ordinary CLI-agent boilerplate, ~1.6KB) must not push a trivial "hi"
|
|
past tier 1: the system prompt is a per-session constant, identical on every
|
|
request in the session, and carries no signal about how requests differ. Before
|
|
the fix this system prompt alone supplied 5 codePresence + 2 technicalTerms
|
|
keyword matches, saturating both dimensions and crossing the default
|
|
simple_medium boundary (0.15) purely from harness text, independent of the ask."""
|
|
agent_system_prompt = (
|
|
"You are Claude Code, Anthropic's official CLI for Claude.\n"
|
|
"You are an interactive agent that helps users with software engineering tasks.\n\n"
|
|
"IMPORTANT: Assist with authorized security testing, defensive security, CTF challenges,\n"
|
|
"and educational contexts. Refuse requests for destructive techniques. Dual-use security\n"
|
|
"tools (C2 frameworks, credential testing, exploit development) require authorization.\n\n"
|
|
"# Harness\n"
|
|
"- Text you output outside of tool use is displayed as Github-flavored markdown.\n"
|
|
"- Tools run behind a user-selected permission mode; a denied call means the user declined.\n"
|
|
"- The system may send updates or reminders. Hooks may intercept tool calls.\n"
|
|
"- Prefer the dedicated file/search tools over shell commands when one fits. Independent\n"
|
|
" tool calls can run in parallel in one response.\n"
|
|
"- Reference code as `file_path:line_number` - it is clickable.\n\n"
|
|
"Write code that reads like the surrounding code: match its comment density, naming, idiom.\n\n"
|
|
"For actions that are hard to reverse, confirm first unless durably authorized. Before\n"
|
|
"deleting or overwriting, look at the target. Report outcomes faithfully: if tests fail,\n"
|
|
"say so with the output; if a step was skipped, say that.\n\n"
|
|
"# Git\n"
|
|
"- Interactive flags (-i, e.g. git rebase -i, git add -i) are not supported.\n"
|
|
"- Use the `gh` CLI for GitHub operations (PRs, issues, API).\n"
|
|
"- Commit or push only when the user asks. If on the default branch, branch first.\n"
|
|
"- End git commit messages with a Co-Authored-By trailer.\n"
|
|
"- End PR bodies with a generated-with footer.\n\n"
|
|
"# Environment\n"
|
|
"- Primary working directory: /Users/tin\n"
|
|
"- Is a git repository: false\n"
|
|
"- Platform: darwin\n"
|
|
"- You are powered by the model claude-opus-5.\n"
|
|
)
|
|
messages = [
|
|
{"role": "system", "content": agent_system_prompt},
|
|
{"role": "user", "content": "hi"},
|
|
]
|
|
resp = await quality_router.async_pre_routing_hook(
|
|
model="quality-router-test",
|
|
request_kwargs={},
|
|
messages=messages,
|
|
)
|
|
assert resp is not None
|
|
assert resp.model == "haiku" # tier 1, same as with no system prompt at all
|
|
|
|
|
|
# ─── Keyword override ──────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def keyword_router():
|
|
"""
|
|
Router where multiple deployments declare overlapping keywords so we can
|
|
exercise the (quality DESC, price ASC) tiebreak.
|
|
|
|
- cheap-coder tier 2, keywords [code, python], cost 0.000001
|
|
- smart-coder tier 3, keywords [code, python], cost 0.000010
|
|
- law-bot tier 2, keywords [legal, contract], cost 0.000005
|
|
- default-haiku tier 1, no keywords, cost 0.0000005
|
|
"""
|
|
spec = [
|
|
{
|
|
"model_name": "default-haiku",
|
|
"quality_tier": 1,
|
|
"keywords": [],
|
|
"input_cost_per_token": 0.0000005,
|
|
},
|
|
{
|
|
"model_name": "cheap-coder",
|
|
"quality_tier": 2,
|
|
"keywords": ["code", "python"],
|
|
"input_cost_per_token": 0.000001,
|
|
},
|
|
{
|
|
"model_name": "smart-coder",
|
|
"quality_tier": 3,
|
|
"keywords": ["code", "python"],
|
|
"input_cost_per_token": 0.000010,
|
|
},
|
|
{
|
|
"model_name": "law-bot",
|
|
"quality_tier": 2,
|
|
"keywords": ["legal", "contract"],
|
|
"input_cost_per_token": 0.000005,
|
|
},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
return QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="default-haiku",
|
|
quality_router_config={
|
|
"available_models": [
|
|
"default-haiku",
|
|
"cheap-coder",
|
|
"smart-coder",
|
|
"law-bot",
|
|
],
|
|
},
|
|
)
|
|
|
|
|
|
class TestKeywordOverride:
|
|
def test_no_keyword_in_message_returns_none(self, keyword_router):
|
|
assert keyword_router._keyword_override("hello there") is None
|
|
|
|
def test_single_match_returns_that_model(self, keyword_router):
|
|
# Only law-bot declares "legal".
|
|
assert keyword_router._keyword_override("review this legal doc") == (
|
|
"law-bot",
|
|
"legal",
|
|
)
|
|
|
|
def test_case_insensitive_match(self, keyword_router):
|
|
assert keyword_router._keyword_override("LEGAL question") == (
|
|
"law-bot",
|
|
"legal",
|
|
)
|
|
|
|
def test_overlap_picks_highest_quality_tier(self, keyword_router):
|
|
# Both cheap-coder (tier 2) and smart-coder (tier 3) declare "code".
|
|
# Quality wins over price → smart-coder.
|
|
assert keyword_router._keyword_override("write some code for me") == (
|
|
"smart-coder",
|
|
"code",
|
|
)
|
|
|
|
def test_same_tier_picks_cheapest(self):
|
|
# Two models at the same tier, both matching "data" — cheapest wins.
|
|
spec = [
|
|
{
|
|
"model_name": "expensive",
|
|
"quality_tier": 2,
|
|
"keywords": ["data"],
|
|
"input_cost_per_token": 0.000050,
|
|
},
|
|
{
|
|
"model_name": "cheap",
|
|
"quality_tier": 2,
|
|
"keywords": ["data"],
|
|
"input_cost_per_token": 0.000005,
|
|
},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="cheap",
|
|
quality_router_config={"available_models": ["expensive", "cheap"]},
|
|
)
|
|
match = qr._keyword_override("show me the data")
|
|
assert match == ("cheap", "data")
|
|
|
|
def test_unpriced_loses_to_priced_at_same_tier(self):
|
|
# Same quality tier, one has cost, one doesn't → priced wins.
|
|
spec = [
|
|
{
|
|
"model_name": "no-price",
|
|
"quality_tier": 2,
|
|
"keywords": ["data"],
|
|
# input_cost_per_token deliberately omitted
|
|
},
|
|
{
|
|
"model_name": "with-price",
|
|
"quality_tier": 2,
|
|
"keywords": ["data"],
|
|
"input_cost_per_token": 0.000005,
|
|
},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="no-price",
|
|
quality_router_config={"available_models": ["no-price", "with-price"]},
|
|
)
|
|
match = qr._keyword_override("show me the data")
|
|
assert match == ("with-price", "data")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hook_short_circuits_complexity_on_keyword_match(
|
|
self, keyword_router
|
|
):
|
|
# A reasoning-style prompt would normally route to a high-quality model
|
|
# via the complexity flow — but the keyword "code" should short-circuit
|
|
# to smart-coder (highest tier among "code" models).
|
|
messages = [
|
|
{
|
|
"role": "user",
|
|
"content": (
|
|
"Think step by step and reason through this code problem. "
|
|
"Analyze this carefully and break down each component."
|
|
),
|
|
}
|
|
]
|
|
request_kwargs: Dict[str, Any] = {}
|
|
resp = await keyword_router.async_pre_routing_hook(
|
|
model="qr",
|
|
request_kwargs=request_kwargs,
|
|
messages=messages,
|
|
)
|
|
assert resp is not None
|
|
assert resp.model == "smart-coder"
|
|
|
|
decision = request_kwargs["metadata"]["quality_router_decision"]
|
|
assert decision["routed_via"] == "keyword"
|
|
assert decision["matched_keyword"] == "code"
|
|
assert decision["complexity_tier"] is None # short-circuited
|
|
|
|
def test_quality_wins_over_explicit_order(self):
|
|
# Quality always beats order. A tier-3 model with no `order` wins over
|
|
# a tier-2 model with `order=1`.
|
|
spec = [
|
|
{
|
|
"model_name": "ordered-tier2",
|
|
"quality_tier": 2,
|
|
"keywords": ["code"],
|
|
"order": 1,
|
|
"input_cost_per_token": 0.000010,
|
|
},
|
|
{
|
|
"model_name": "implicit-tier3",
|
|
"quality_tier": 3,
|
|
"keywords": ["code"],
|
|
"input_cost_per_token": 0.000005,
|
|
},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="ordered-tier2",
|
|
quality_router_config={
|
|
"available_models": ["ordered-tier2", "implicit-tier3"]
|
|
},
|
|
)
|
|
match = qr._keyword_override("write some code")
|
|
assert match == ("implicit-tier3", "code")
|
|
|
|
def test_order_breaks_tie_within_same_quality_tier(self):
|
|
# Two tier-3 models, both match "code". Lower `order` wins.
|
|
spec = [
|
|
{
|
|
"model_name": "preferred",
|
|
"quality_tier": 3,
|
|
"keywords": ["code"],
|
|
"order": 1,
|
|
"input_cost_per_token": 0.000050, # more expensive
|
|
},
|
|
{
|
|
"model_name": "default-tier3",
|
|
"quality_tier": 3,
|
|
"keywords": ["code"],
|
|
"input_cost_per_token": 0.000005, # cheaper
|
|
},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="default-tier3",
|
|
quality_router_config={"available_models": ["preferred", "default-tier3"]},
|
|
)
|
|
match = qr._keyword_override("write some code")
|
|
assert match == ("preferred", "code")
|
|
|
|
def test_explicit_order_overrides_price(self):
|
|
# Same tier, but the more expensive one has a lower `order` and wins.
|
|
spec = [
|
|
{
|
|
"model_name": "expensive-but-preferred",
|
|
"quality_tier": 2,
|
|
"keywords": ["data"],
|
|
"order": 1,
|
|
"input_cost_per_token": 0.000050,
|
|
},
|
|
{
|
|
"model_name": "cheap-default",
|
|
"quality_tier": 2,
|
|
"keywords": ["data"],
|
|
"input_cost_per_token": 0.000005,
|
|
},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="cheap-default",
|
|
quality_router_config={
|
|
"available_models": ["expensive-but-preferred", "cheap-default"]
|
|
},
|
|
)
|
|
match = qr._keyword_override("show me the data")
|
|
assert match == ("expensive-but-preferred", "data")
|
|
|
|
def test_lower_order_wins_between_two_explicitly_ordered(self):
|
|
spec = [
|
|
{
|
|
"model_name": "second",
|
|
"quality_tier": 2,
|
|
"keywords": ["data"],
|
|
"order": 5,
|
|
},
|
|
{
|
|
"model_name": "first",
|
|
"quality_tier": 2,
|
|
"keywords": ["data"],
|
|
"order": 1,
|
|
},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="first",
|
|
quality_router_config={"available_models": ["first", "second"]},
|
|
)
|
|
match = qr._keyword_override("show me the data")
|
|
assert match == ("first", "data")
|
|
|
|
def test_same_order_falls_through_to_quality_then_price(self):
|
|
# All three models share order=1 → tiebreak falls through to
|
|
# (quality DESC, cost ASC).
|
|
spec = [
|
|
{
|
|
"model_name": "low-tier",
|
|
"quality_tier": 1,
|
|
"keywords": ["data"],
|
|
"order": 1,
|
|
"input_cost_per_token": 0.000001,
|
|
},
|
|
{
|
|
"model_name": "high-tier-cheap",
|
|
"quality_tier": 3,
|
|
"keywords": ["data"],
|
|
"order": 1,
|
|
"input_cost_per_token": 0.000005,
|
|
},
|
|
{
|
|
"model_name": "high-tier-expensive",
|
|
"quality_tier": 3,
|
|
"keywords": ["data"],
|
|
"order": 1,
|
|
"input_cost_per_token": 0.000050,
|
|
},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="low-tier",
|
|
quality_router_config={
|
|
"available_models": [
|
|
"low-tier",
|
|
"high-tier-cheap",
|
|
"high-tier-expensive",
|
|
]
|
|
},
|
|
)
|
|
match = qr._keyword_override("show me the data")
|
|
assert match == ("high-tier-cheap", "data")
|
|
|
|
def test_order_is_used_in_tier_resolution_too(self):
|
|
# Two models at the same tier. Explicit `order=1` on the second one
|
|
# should make _resolve_model_for_quality_tier(2) pick it.
|
|
spec = [
|
|
{
|
|
"model_name": "default-pick",
|
|
"quality_tier": 2,
|
|
},
|
|
{
|
|
"model_name": "preferred-pick",
|
|
"quality_tier": 2,
|
|
"order": 1,
|
|
},
|
|
]
|
|
router = MagicMock()
|
|
router.model_list = _make_model_list(spec)
|
|
qr = QualityRouter(
|
|
model_name="qr",
|
|
litellm_router_instance=router,
|
|
default_model="default-pick",
|
|
quality_router_config={
|
|
"available_models": ["default-pick", "preferred-pick"]
|
|
},
|
|
)
|
|
assert qr._resolve_model_for_quality_tier(2) == "preferred-pick"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hook_falls_back_to_complexity_when_no_keyword(self, keyword_router):
|
|
# No declared keyword in the message → complexity-based routing.
|
|
# "hi" is SIMPLE → quality 1 → default-haiku (the only tier-1 model).
|
|
messages = [{"role": "user", "content": "hi"}]
|
|
request_kwargs: Dict[str, Any] = {}
|
|
resp = await keyword_router.async_pre_routing_hook(
|
|
model="qr",
|
|
request_kwargs=request_kwargs,
|
|
messages=messages,
|
|
)
|
|
assert resp is not None
|
|
assert resp.model == "default-haiku"
|
|
|
|
decision = request_kwargs["metadata"]["quality_router_decision"]
|
|
assert decision["routed_via"] == "quality_tier"
|
|
assert decision["matched_keyword"] is None
|
|
assert decision["complexity_tier"] == "SIMPLE"
|
|
|
|
|
|
# ─── Routing-decision metadata (powers x-litellm-quality-router-* headers) ──
|
|
|
|
|
|
class TestDecisionMetadata:
|
|
@pytest.mark.asyncio
|
|
async def test_hook_stashes_decision_in_request_kwargs_metadata(
|
|
self, quality_router
|
|
):
|
|
# Reasoning prompt → REASONING → quality tier 4 → opus-next.
|
|
messages = [
|
|
{
|
|
"role": "user",
|
|
"content": (
|
|
"Think step by step and reason through this problem. "
|
|
"Analyze this carefully and break down each component."
|
|
),
|
|
}
|
|
]
|
|
request_kwargs: Dict[str, Any] = {}
|
|
|
|
resp = await quality_router.async_pre_routing_hook(
|
|
model="quality-router-test",
|
|
request_kwargs=request_kwargs,
|
|
messages=messages,
|
|
)
|
|
assert resp is not None and resp.model == "opus-next"
|
|
|
|
decision = request_kwargs["metadata"]["quality_router_decision"]
|
|
assert decision["routed_model"] == "opus-next"
|
|
assert decision["quality_tier"] == 4
|
|
assert decision["complexity_tier"] == "REASONING"
|
|
assert decision["router_model_name"] == "quality-router-test"
|
|
assert decision["routed_via"] == "quality_tier"
|
|
assert decision["matched_keyword"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decision_metadata_preserves_existing_metadata(self, quality_router):
|
|
request_kwargs: Dict[str, Any] = {
|
|
"metadata": {"trace_id": "abc-123", "user_id": "u-1"}
|
|
}
|
|
|
|
await quality_router.async_pre_routing_hook(
|
|
model="quality-router-test",
|
|
request_kwargs=request_kwargs,
|
|
messages=[{"role": "user", "content": "hi"}],
|
|
)
|
|
|
|
# Existing metadata keys are intact and the decision is added alongside.
|
|
assert request_kwargs["metadata"]["trace_id"] == "abc-123"
|
|
assert request_kwargs["metadata"]["user_id"] == "u-1"
|
|
assert "quality_router_decision" in request_kwargs["metadata"]
|
|
|
|
|
|
# ─── Router.set_response_headers lifts decision into x-litellm-quality-* ────
|
|
|
|
|
|
class TestSetResponseHeadersLiftsDecision:
|
|
"""
|
|
Verify the Router.set_response_headers helper turns a stashed quality-router
|
|
decision into x-litellm-quality-router-* headers on the response.
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lifts_decision_into_additional_headers(self):
|
|
from pydantic import BaseModel
|
|
|
|
from litellm.router import Router
|
|
|
|
class FakeResponse(BaseModel):
|
|
model_config = {"arbitrary_types_allowed": True}
|
|
_hidden_params: Dict[str, Any] = {}
|
|
|
|
# Build a real Router with a tiny model_list — enough to satisfy
|
|
# set_response_headers without needing the rest of the router stack.
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "haiku",
|
|
"litellm_params": {
|
|
"model": "openai/gpt-4o-mini",
|
|
"api_key": "sk-test",
|
|
},
|
|
}
|
|
]
|
|
)
|
|
|
|
response = FakeResponse()
|
|
response._hidden_params = {}
|
|
|
|
request_kwargs = {
|
|
"metadata": {
|
|
"quality_router_decision": {
|
|
"router_model_name": "qr",
|
|
"routed_model": "smart-coder",
|
|
"routed_via": "keyword",
|
|
"matched_keyword": "code",
|
|
"quality_tier": 3,
|
|
"complexity_tier": None,
|
|
}
|
|
}
|
|
}
|
|
|
|
await router.set_response_headers(
|
|
response=response,
|
|
model_group="qr",
|
|
request_kwargs=request_kwargs,
|
|
)
|
|
|
|
headers = response._hidden_params["additional_headers"]
|
|
assert headers["x-litellm-quality-router-model"] == "smart-coder"
|
|
assert headers["x-litellm-quality-router-tier"] == "3"
|
|
assert headers["x-litellm-quality-router-via"] == "keyword"
|
|
assert headers["x-litellm-quality-router-keyword"] == "code"
|
|
# Keyword route short-circuits classification → no complexity header.
|
|
assert "x-litellm-quality-router-complexity" not in headers
|
|
# Existing x-litellm-model-group behavior is unchanged.
|
|
assert headers["x-litellm-model-group"] == "qr"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_quality_tier_route_emits_complexity_not_keyword(self):
|
|
from pydantic import BaseModel
|
|
|
|
from litellm.router import Router
|
|
|
|
class FakeResponse(BaseModel):
|
|
model_config = {"arbitrary_types_allowed": True}
|
|
_hidden_params: Dict[str, Any] = {}
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "haiku",
|
|
"litellm_params": {
|
|
"model": "openai/gpt-4o-mini",
|
|
"api_key": "sk-test",
|
|
},
|
|
}
|
|
]
|
|
)
|
|
|
|
response = FakeResponse()
|
|
response._hidden_params = {}
|
|
|
|
request_kwargs = {
|
|
"metadata": {
|
|
"quality_router_decision": {
|
|
"router_model_name": "qr",
|
|
"routed_model": "haiku",
|
|
"routed_via": "quality_tier",
|
|
"matched_keyword": None,
|
|
"quality_tier": 1,
|
|
"complexity_tier": "SIMPLE",
|
|
}
|
|
}
|
|
}
|
|
|
|
await router.set_response_headers(
|
|
response=response,
|
|
model_group="qr",
|
|
request_kwargs=request_kwargs,
|
|
)
|
|
|
|
headers = response._hidden_params["additional_headers"]
|
|
assert headers["x-litellm-quality-router-via"] == "quality_tier"
|
|
assert headers["x-litellm-quality-router-complexity"] == "SIMPLE"
|
|
# Quality-tier route → no keyword header.
|
|
assert "x-litellm-quality-router-keyword" not in headers
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_decision_leaves_quality_router_headers_unset(self):
|
|
from pydantic import BaseModel
|
|
|
|
from litellm.router import Router
|
|
|
|
class FakeResponse(BaseModel):
|
|
model_config = {"arbitrary_types_allowed": True}
|
|
_hidden_params: Dict[str, Any] = {}
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "haiku",
|
|
"litellm_params": {
|
|
"model": "openai/gpt-4o-mini",
|
|
"api_key": "sk-test",
|
|
},
|
|
}
|
|
]
|
|
)
|
|
|
|
response = FakeResponse()
|
|
response._hidden_params = {}
|
|
|
|
await router.set_response_headers(
|
|
response=response,
|
|
model_group="haiku",
|
|
request_kwargs={}, # no quality_router_decision
|
|
)
|
|
|
|
headers = response._hidden_params["additional_headers"]
|
|
assert "x-litellm-quality-router-model" not in headers
|
|
assert "x-litellm-quality-router-tier" not in headers
|
|
|
|
|
|
class TestRouterQualityDeploymentMethods:
|
|
"""Tests for Router._is_quality_router_deployment and Router.init_quality_router_deployment."""
|
|
|
|
def test_is_quality_router_deployment_true(self):
|
|
"""_is_quality_router_deployment returns True for quality router models."""
|
|
from litellm.router import Router
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "gpt-4o-mini",
|
|
"litellm_params": {"model": "openai/gpt-4o-mini"},
|
|
}
|
|
]
|
|
)
|
|
params = LiteLLM_Params(model="auto_router/quality_router/my-router")
|
|
assert router._is_quality_router_deployment(params) is True
|
|
|
|
def test_is_quality_router_deployment_false(self):
|
|
"""_is_quality_router_deployment returns False for regular models."""
|
|
from litellm.router import Router
|
|
from litellm.types.router import LiteLLM_Params
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "gpt-4o-mini",
|
|
"litellm_params": {"model": "openai/gpt-4o-mini"},
|
|
}
|
|
]
|
|
)
|
|
params = LiteLLM_Params(model="openai/gpt-4o-mini")
|
|
assert router._is_quality_router_deployment(params) is False
|
|
|
|
def test_init_quality_router_deployment(self):
|
|
"""init_quality_router_deployment registers a QualityRouter."""
|
|
from litellm.router import Router
|
|
from litellm.types.router import Deployment, LiteLLM_Params
|
|
|
|
router = Router(
|
|
model_list=[
|
|
{
|
|
"model_name": "gpt-4o-mini",
|
|
"litellm_params": {"model": "openai/gpt-4o-mini"},
|
|
}
|
|
]
|
|
)
|
|
deployment = Deployment(
|
|
model_name="auto_router/quality_router/test-router",
|
|
litellm_params=LiteLLM_Params(
|
|
model="auto_router/quality_router/test-router",
|
|
quality_router_default_model="gpt-4o-mini",
|
|
),
|
|
model_info={"id": "test-id"},
|
|
)
|
|
router.init_quality_router_deployment(deployment)
|
|
assert "auto_router/quality_router/test-router" in router.quality_routers
|
|
|
|
|
|
class TestRoutingDecisionProvenance:
|
|
"""Every quality-router path must attach a routing_decision to its hook response,
|
|
including the no-user-message default path that previously recorded nothing."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_quality_tier_decision(self, quality_router):
|
|
resp = await quality_router.async_pre_routing_hook(
|
|
model="quality-router-test",
|
|
request_kwargs={},
|
|
messages=[{"role": "user", "content": "hi"}],
|
|
)
|
|
assert resp is not None
|
|
decision = resp.routing_decision
|
|
assert decision is not None
|
|
assert decision["router_model_name"] == "quality-router-test"
|
|
assert decision["router_type"] == "quality"
|
|
assert decision["cause"] == "quality_tier"
|
|
assert decision["routed_model"] == "haiku"
|
|
assert decision["tier"] == "1"
|
|
assert isinstance(decision["score"], float)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_keyword_decision_carries_matched_keyword(self, keyword_router):
|
|
resp = await keyword_router.async_pre_routing_hook(
|
|
model="quality-router-test",
|
|
request_kwargs={},
|
|
messages=[{"role": "user", "content": "please write python for me"}],
|
|
)
|
|
assert resp is not None
|
|
decision = resp.routing_decision
|
|
assert decision is not None
|
|
assert decision["cause"] == "keyword"
|
|
assert decision["matched_keyword"] == "python"
|
|
assert decision["routed_model"] == resp.model
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_user_message_decision_is_default_fallback(self, quality_router):
|
|
resp = await quality_router.async_pre_routing_hook(
|
|
model="quality-router-test",
|
|
request_kwargs={},
|
|
messages=[{"role": "system", "content": "You are a helpful assistant."}],
|
|
)
|
|
assert resp is not None
|
|
decision = resp.routing_decision
|
|
assert decision is not None
|
|
assert decision["cause"] == "default_fallback"
|
|
assert decision["routed_model"] == "haiku"
|
|
assert "tier" not in decision
|