feat(quality_router): expose routing decision in response headers

For transparency, expose the QualityRouter's routing decision in the
proxy response headers:

  x-litellm-quality-router-model       → picked model_name (e.g. "haiku-vision")
  x-litellm-quality-router-tier        → resolved quality tier (e.g. "1")
  x-litellm-quality-router-complexity  → ComplexityTier name (e.g. "SIMPLE")

Mechanism: the pre-routing hook stashes the decision in
request_kwargs["metadata"]["quality_router_decision"]. After the call
returns, Router.set_response_headers lifts the decision into
response._hidden_params["additional_headers"] alongside the existing
x-litellm-model-group / x-litellm-model-id headers. Existing metadata
keys (trace_id, user_id, etc.) are preserved.

Co-Authored-By: Claude Opus 4 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krrish Dholakia 2026-04-17 18:07:24 -07:00
parent 2a617d48e6
commit 103512f2f4
3 changed files with 217 additions and 2 deletions

View file

@ -5889,7 +5889,7 @@ class Router:
response = await response
## PROCESS RESPONSE HEADERS
response = await self.set_response_headers(
response=response, model_group=model_group
response=response, model_group=model_group, request_kwargs=kwargs
)
return response
@ -8208,7 +8208,10 @@ class Router:
return returned_dict
async def set_response_headers(
self, response: Any, model_group: Optional[str] = None
self,
response: Any,
model_group: Optional[str] = None,
request_kwargs: Optional[dict] = None,
) -> Any:
"""
Add the most accurate rate limit headers for a given model response.
@ -8229,6 +8232,33 @@ class Router:
additional_headers = response._hidden_params["additional_headers"] # type: ignore
# Lift QualityRouter routing decision into response headers for
# transparency. The decision is stashed in request_kwargs.metadata
# by QualityRouter.async_pre_routing_hook.
metadata = (
(request_kwargs.get("metadata") or {})
if isinstance(request_kwargs, dict)
else {}
)
decision = (
metadata.get("quality_router_decision")
if isinstance(metadata, dict)
else None
)
if isinstance(decision, dict):
if "routed_model" in decision:
additional_headers["x-litellm-quality-router-model"] = str(
decision["routed_model"]
)
if "quality_tier" in decision:
additional_headers["x-litellm-quality-router-tier"] = str(
decision["quality_tier"]
)
if "complexity_tier" in decision:
additional_headers["x-litellm-quality-router-complexity"] = str(
decision["complexity_tier"]
)
if (
"x-ratelimit-remaining-tokens" not in additional_headers
and "x-ratelimit-remaining-requests" not in additional_headers

View file

@ -319,6 +319,23 @@ class QualityRouter(CustomLogger):
f"routed_model={routed_model}"
)
# Stash the decision in request_kwargs.metadata so the Router can lift
# it into response headers (`x-litellm-quality-router-*`) for
# transparency. The same dict object flows from here through to
# `make_call.set_response_headers`.
if request_kwargs is not None:
metadata = request_kwargs.setdefault("metadata", {})
if isinstance(metadata, dict):
metadata["quality_router_decision"] = {
"router_model_name": self.model_name,
"routed_model": routed_model,
"quality_tier": int(quality_tier),
"complexity_tier": complexity_name,
"required_capabilities": (
sorted(required_capabilities) if required_capabilities else []
),
}
return PreRoutingHookResponse(
model=routed_model,
messages=messages,

View file

@ -365,3 +365,171 @@ class TestCapabilities:
assert resp is not None
# tier 1, no caps required → first registered model at tier 1.
assert resp.model == "haiku-text"
# ─── 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["required_capabilities"] == []
@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"]
@pytest.mark.asyncio
async def test_decision_metadata_includes_required_capabilities(
self, capability_router
):
request_kwargs: Dict[str, Any] = {
"litellm_capabilities": ["vision"],
}
await capability_router.async_pre_routing_hook(
model="qr",
request_kwargs=request_kwargs,
messages=[{"role": "user", "content": "hi"}],
)
decision = request_kwargs["metadata"]["quality_router_decision"]
assert decision["routed_model"] == "haiku-vision"
assert decision["required_capabilities"] == ["vision"]
# ─── 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": "haiku-vision",
"quality_tier": 1,
"complexity_tier": "SIMPLE",
"required_capabilities": ["vision"],
}
}
}
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"] == "haiku-vision"
assert headers["x-litellm-quality-router-tier"] == "1"
assert headers["x-litellm-quality-router-complexity"] == "SIMPLE"
# Existing x-litellm-model-group behavior is unchanged.
assert headers["x-litellm-model-group"] == "qr"
@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
assert "x-litellm-quality-router-complexity" not in headers