scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-07 08:26:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

fix(auto-router): correct Responses API tool_choice shape and propagate alias litellm_params (#32974)

Browse source 
* fix(anthropic-messages): send bare-string tool_choice to Responses API, propagate router-alias litellm_params

The Anthropic /v1/messages -> Responses API adapter always wrapped
tool_choice in an object ({"type": "auto"}, {"type": "required"}), but
the Responses API's tool_choice schema for these cases is a bare
string ("auto"/"required"/"none"). Sending the object shape to an
OpenAI-compatible backend (e.g. vLLM) fails Pydantic validation with a
400. The "none" case also fell through to "auto" instead of mapping to
"none".

Separately, litellm_params configured directly on a router-alias
deployment (auto_router/complexity_router, adaptive_router,
quality_router, or semantic auto_router) - e.g.
cache_control_injection_points, drop_params - were silently dropped
for every request through that alias. async_pre_routing_hook swaps
`model` from the alias name to the selected tier/route's model before
the deployment lookup runs, so the outbound call only ever merged in
the tier deployment's own litellm_params, never the alias's. Register
non-routing-config litellm_params from the alias deployment and apply
them to the request whenever a pre-routing hook substitutes the model.

* fix: satisfy ruff-strict-budget UP006 and router coverage checker

Use builtin dict[...] generics instead of typing.Dict for the two new
annotations introduced in the previous commit, since they pushed
UP006 over the codebase ceiling in ruff-strict-budget.json. Add a
direct unit test for _register_pre_routing_alias_overrides so the
text-based router_code_coverage.py checker sees it exercised by name.

* fix(router): replace alias-param denylist with a tight allowlist

_PRE_ROUTING_ALIAS_RESERVED_PARAMS excluded router-init-only keys from
the alias's litellm_params before forwarding the rest as request
kwargs, but GenericLiteLLMParams also holds deployment-management
fields (tpm, rpm, weight, tags, max_budget, budget_duration,
use_in_pass_through, litellm_credential_name, ...) on the same object.
Any of those left off the denylist would get silently forwarded as if
they were request kwargs.

Replace the denylist with a tight allowlist of exactly the two
request-shaping params this feature exists for - drop_params and
cache_control_injection_points - so unrelated management fields never
reach the outbound call regardless of what else GenericLiteLLMParams
grows to hold.

* fix(router): re-register adaptive-alias overrides on set_model_list reload

set_model_list() unconditionally clears pre_routing_alias_overrides on
every call (e.g. /config/reload), but _finalize_adaptive_router_if_configured()
skips rebuilding an AdaptiveRouter whose model_name already exists in
self.adaptive_routers - so _register_pre_routing_alias_overrides() never
ran again for an auto_router/adaptive_router alias after a reload,
silently dropping its drop_params/cache_control_injection_points.

Build the Deployment unconditionally and re-register its overrides even
on the skip-existing-router path; only the (expensive) AdaptiveRouter
construction itself stays skipped.

* style: ruff format after merging litellm_internal_staging

* fix(router): drop the alias-param allowlist, exclude only model

Per review discussion: instead of a router.py-local allowlist of exactly
which litellm_params an alias (auto_router/complexity_router,
adaptive_router, quality_router, semantic auto_router) can forward to
the request it routes, _register_pre_routing_alias_overrides now
forwards everything except `model` (the alias marker itself, e.g.
auto_router/complexity_router, never a real provider model).

Router-init-only fields (complexity_router_config,
complexity_router_default_model, auto_router_config,
auto_router_config_path, auto_router_default_model,
auto_router_embedding_model, adaptive_router_config,
adaptive_router_default_model, quality_router_config,
quality_router_default_model) now flow into request_kwargs unfiltered
too. That's safe because litellm.completion()/acompletion() already
strips anything in litellm.types.utils.all_litellm_params before
building the provider request - added these 10 keys there, alongside
the deployment-management fields (tpm, rpm, weight, ...) already listed.
Verified live: without that addition, complexity_router_config lands in
extra_body and ships raw to the provider; with it, it's stripped.

This moves the "which fields aren't real LLM params" list from a
router.py-local allowlist to the single existing global list every
completion() call already depends on, instead of maintaining two.

* refactor(router): look up alias litellm_params on demand instead of caching them

_register_pre_routing_alias_overrides cached each alias's litellm_params
into self.pre_routing_alias_overrides at deployment-init time, which
required keeping that cache in sync with set_model_list() reloads - the
exact bug the previous adaptive-router-reload fix was patching around
(AdaptiveRouter survives a reload, but the cache didn't always get
refreshed to match).

Delete the cache and the registration method entirely. async_pre_routing_hook
now looks up the alias's own litellm_params directly from self.model_list
via self.model_name_to_deployment_indices at request time, the same
model_list that's already correctly rebuilt on every set_model_list()
call. No second piece of state to invalidate, so the reload staleness
bug class isn't possible anymore, and it's less code than before.
This commit is contained in:
Krrish Dholakia 2026-07-13 15:11:48 -07:00 • committed by GitHub
parent 3f897b29ae
commit 39e0efa11d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 317 additions and 298 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py
View file

@ -198,14 +198,16 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
@staticmethod
def translate_tool_choice_to_responses_api(
tool_choice: AnthropicMessagesToolChoice,
) -> Dict[str, Any]:
) -> Union[str, dict[str, Any]]:
"""Convert Anthropic tool_choice to Responses API tool_choice."""
tc_type = tool_choice.get("type")
if tc_type == "any":
return {"type": "required"}
return "required"
elif tc_type == "tool":
return {"type": "function", "name": tool_choice.get("name", "")}
return {"type": "auto"}
elif tc_type == "none":
return "none"
return "auto"
@staticmethod
def translate_context_management_to_responses_api(

81
litellm/router.py
View file

@ -7605,13 +7605,13 @@ class Router:
model_name = entry.get("model_name") if isinstance(entry, dict) else entry.model_name
if not model_name or not lp:
continue
if model_name in self.adaptive_routers:
continue
deployment = Deployment(
model_name=model_name,
litellm_params=(lp if not isinstance(lp, dict) else LiteLLM_Params(**lp)),
model_info=(entry.get("model_info") if isinstance(entry, dict) else entry.model_info),
)
if model_name in self.adaptive_routers:
continue
self.init_adaptive_router_deployment(deployment=deployment)
for model_name, complexity_router in self.complexity_routers.items():
@ -10707,56 +10707,39 @@ class Router:
if self.routing_plugins:
await self._run_routing_plugins(model=model, request_kwargs=request_kwargs, messages=messages)
#########################################################
# Check if any auto-router should be used
#########################################################
if model in self.auto_routers:
return await self.auto_routers[model].async_pre_routing_hook(
model=model,
request_kwargs=request_kwargs,
messages=messages,
input=input,
specific_deployment=specific_deployment,
)
router_strategy = (
self.auto_routers.get(model)
or self.complexity_routers.get(model)
or self.adaptive_routers.get(model)
or self.quality_routers.get(model)
)
if router_strategy is None:
return None
#########################################################
# Check if any complexity-router should be used
#########################################################
if model in self.complexity_routers:
return await self.complexity_routers[model].async_pre_routing_hook(
model=model,
request_kwargs=request_kwargs,
messages=messages,
input=input,
specific_deployment=specific_deployment,
)
pre_routing_hook_response = await router_strategy.async_pre_routing_hook(
model=model,
request_kwargs=request_kwargs,
messages=messages,
input=input,
specific_deployment=specific_deployment,
)
#########################################################
# Check if an adaptive-router should be used
#########################################################
adaptive_router = self.adaptive_routers.get(model)
if adaptive_router is not None:
return await adaptive_router.async_pre_routing_hook(
model=model,
request_kwargs=request_kwargs,
messages=messages,
input=input,
specific_deployment=specific_deployment,
)
# `model` (the alias, e.g. "smart-router") is never the deployment actually
# called - apply the alias's own litellm_params (besides `model` itself,
# which is just the alias marker) to the request, since the tier/route
# deployment the hook selected won't have them. Router-only fields
# (tpm, rpm, weight, complexity_router_config, ...) are excluded from the
# actual outbound LLM call downstream by litellm.types.utils.all_litellm_params,
# not here.
if pre_routing_hook_response is not None:
alias_index = self.model_name_to_deployment_indices.get(model, [])
if alias_index:
alias_litellm_params = self.model_list[alias_index[0]].get("litellm_params", {})
for key, value in alias_litellm_params.items():
if key != "model" and value is not None:
request_kwargs.setdefault(key, value)
#########################################################
# Check if any quality-router should be used
#########################################################
if model in self.quality_routers:
return await self.quality_routers[model].async_pre_routing_hook(
model=model,
request_kwargs=request_kwargs,
messages=messages,
input=input,
specific_deployment=specific_deployment,
)
return None
return pre_routing_hook_response
def get_available_deployment(
self,

10
litellm/types/utils.py
View file

@ -3210,6 +3210,16 @@ all_litellm_params = (
"_litellm_tpm_reserved_model",
"_litellm_tpm_reserved_scopes",
"_litellm_tpm_reservation_released",
"auto_router_config_path",
"auto_router_config",
"auto_router_default_model",
"auto_router_embedding_model",
"complexity_router_config",
"complexity_router_default_model",
"adaptive_router_config",
"adaptive_router_default_model",
"quality_router_config",
"quality_router_default_model",
]
+ list(StandardCallbackDynamicParams.__annotations__.keys())
+ list(CustomPricingLiteLLMParams.model_fields.keys())

99
tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py
View file

@ -99,17 +99,11 @@ class TestContextManagementConversion:
}
)
kwargs = _ADAPTER.translate_request(req)
assert kwargs["context_management"] == [
{"type": "compaction", "compact_threshold": 100000}
]
assert kwargs["context_management"] == [{"type": "compaction", "compact_threshold": 100000}]
def test_translate_request_drops_anthropic_only_context_management(self):
"""context_management with only unknown edit types is omitted from kwargs."""
req = _make_request(
context_management={
"edits": [{"type": "clear_thinking_20251015", "keep": "all"}]
}
)
req = _make_request(context_management={"edits": [{"type": "clear_thinking_20251015", "keep": "all"}]})
kwargs = _ADAPTER.translate_request(req)
assert "context_management" not in kwargs
@ -134,9 +128,7 @@ class TestOutputConfigStructuredOutput:
def test_output_config_format_json_schema_converted(self):
"""output_config.format.json_schema is converted to OpenAI text.format."""
req = _make_request(
output_config={"format": {"type": "json_schema", "schema": self._SCHEMA}}
)
req = _make_request(output_config={"format": {"type": "json_schema", "schema": self._SCHEMA}})
kwargs = _ADAPTER.translate_request(req)
assert "text" in kwargs
fmt = kwargs["text"]["format"]
@ -153,9 +145,7 @@ class TestOutputConfigStructuredOutput:
def test_output_format_still_works(self):
"""The original output_format field still takes precedence when present."""
req = _make_request(
output_format={"type": "json_schema", "schema": self._SCHEMA}
)
req = _make_request(output_format={"type": "json_schema", "schema": self._SCHEMA})
kwargs = _ADAPTER.translate_request(req)
assert "text" in kwargs
assert kwargs["text"]["format"]["type"] == "json_schema"
@ -250,9 +240,7 @@ class TestTranslateMessagesToResponsesInput:
]
result = _translate_messages(messages)
assert len(result) == 1
assert result[0]["content"] == [
{"type": "input_image", "image_url": "data:image/png;base64,abc123"}
]
assert result[0]["content"] == [{"type": "input_image", "image_url": "data:image/png;base64,abc123"}]
def test_user_url_image(self):
"""User message with URL image source becomes input_image with the URL."""
@ -268,9 +256,7 @@ class TestTranslateMessagesToResponsesInput:
}
]
result = _translate_messages(messages)
assert result[0]["content"] == [
{"type": "input_image", "image_url": "https://example.com/img.jpg"}
]
assert result[0]["content"] == [{"type": "input_image", "image_url": "https://example.com/img.jpg"}]
def test_user_base64_image_empty_data_skipped(self):
"""Base64 image with empty data is skipped (no URL can be formed)."""
@ -341,9 +327,7 @@ class TestTranslateMessagesToResponsesInput:
messages = [
{
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "call_null", "content": None}
],
"content": [{"type": "tool_result", "tool_use_id": "call_null", "content": None}],
}
]
result = _translate_messages(messages)
@ -370,9 +354,7 @@ class TestTranslateMessagesToResponsesInput:
}
]
result = _translate_messages(messages)
assert result[0]["content"] == [
{"type": "output_text", "text": "Here is the answer."}
]
assert result[0]["content"] == [{"type": "output_text", "text": "Here is the answer."}]
def test_assistant_tool_use_becomes_function_call(self):
"""Assistant tool_use block becomes a top-level function_call item."""
@ -404,15 +386,11 @@ class TestTranslateMessagesToResponsesInput:
messages = [
{
"role": "assistant",
"content": [
{"type": "thinking", "thinking": "Let me reason step by step."}
],
"content": [{"type": "thinking", "thinking": "Let me reason step by step."}],
}
]
result = _translate_messages(messages)
assert result[0]["content"] == [
{"type": "output_text", "text": "Let me reason step by step."}
]
assert result[0]["content"] == [{"type": "output_text", "text": "Let me reason step by step."}]
def test_assistant_empty_thinking_block_skipped(self):
"""Assistant thinking block with empty thinking text is skipped."""
@ -584,28 +562,27 @@ class TestTranslateToolsToResponsesAPI:
class TestTranslateToolChoiceToResponsesAPI:
"""Anthropic tool_choice -> Responses API tool_choice."""
"""Anthropic tool_choice -> Responses API tool_choice.
def test_auto_maps_to_auto(self):
assert _ADAPTER.translate_tool_choice_to_responses_api({"type": "auto"}) == {
"type": "auto"
}
The Responses API's tool_choice schema (openai.types.responses.tool_choice_options)
is a bare Literal["none", "auto", "required"] for these simple cases - not an
object like {"type": "auto"}. Sending the object shape to an OpenAI-compatible
server gets rejected with a pydantic validation error.
"""
def test_any_maps_to_required(self):
assert _ADAPTER.translate_tool_choice_to_responses_api({"type": "any"}) == {
"type": "required"
}
def test_auto_maps_to_bare_string_auto(self):
assert _ADAPTER.translate_tool_choice_to_responses_api({"type": "auto"}) == "auto"
def test_any_maps_to_bare_string_required(self):
assert _ADAPTER.translate_tool_choice_to_responses_api({"type": "any"}) == "required"
def test_none_maps_to_bare_string_none(self):
assert _ADAPTER.translate_tool_choice_to_responses_api({"type": "none"}) == "none"
def test_specific_tool_maps_to_function(self):
result = _ADAPTER.translate_tool_choice_to_responses_api(
{"type": "tool", "name": "get_weather"}
)
result = _ADAPTER.translate_tool_choice_to_responses_api({"type": "tool", "name": "get_weather"})
assert result == {"type": "function", "name": "get_weather"}
def test_unknown_type_defaults_to_auto(self):
result = _ADAPTER.translate_tool_choice_to_responses_api({"type": "none"})
assert result == {"type": "auto"}
# ---------------------------------------------------------------------------
# translate_thinking_to_reasoning
@ -616,17 +593,13 @@ class TestTranslateThinkingToReasoning:
"""Anthropic thinking param -> Responses API reasoning param."""
def test_budget_high_effort(self):
result = _ADAPTER.translate_thinking_to_reasoning(
{"type": "enabled", "budget_tokens": 10000}
)
result = _ADAPTER.translate_thinking_to_reasoning({"type": "enabled", "budget_tokens": 10000})
# Default (reasoning_auto_summary=False): only effort, no summary
assert result == {"effort": "high"}
assert result is not None and "summary" not in result
def test_budget_above_threshold_high_effort(self):
result = _ADAPTER.translate_thinking_to_reasoning(
{"type": "enabled", "budget_tokens": 50000}
)
result = _ADAPTER.translate_thinking_to_reasoning({"type": "enabled", "budget_tokens": 50000})
assert result is not None
assert result["effort"] == "high"
assert "summary" not in result
@ -652,9 +625,7 @@ class TestTranslateThinkingToReasoning:
assert result is not None and "summary" not in result
def test_budget_minimal_effort(self):
result = _ADAPTER.translate_thinking_to_reasoning(
{"type": "enabled", "budget_tokens": 500}
)
result = _ADAPTER.translate_thinking_to_reasoning({"type": "enabled", "budget_tokens": 500})
assert result == {"effort": "minimal"}
assert result is not None and "summary" not in result
@ -707,9 +678,7 @@ class TestTranslateThinkingToReasoning:
original = litellm.reasoning_auto_summary
try:
litellm.reasoning_auto_summary = True
result = _ADAPTER.translate_thinking_to_reasoning(
{"type": "enabled", "budget_tokens": 10000}
)
result = _ADAPTER.translate_thinking_to_reasoning({"type": "enabled", "budget_tokens": 10000})
assert result == {"effort": "high", "summary": "detailed"}
finally:
litellm.reasoning_auto_summary = original
@ -789,11 +758,7 @@ class TestTranslateRequestBroaderCoverage:
assert kwargs["top_p"] == 0.9
def test_tools_translated(self):
req = _make_request(
tools=[
{"name": "calculator", "description": "Does math.", "input_schema": {}}
]
)
req = _make_request(tools=[{"name": "calculator", "description": "Does math.", "input_schema": {}}])
kwargs = _ADAPTER.translate_request(req)
assert len(kwargs["tools"]) == 1
assert kwargs["tools"][0]["name"] == "calculator"
@ -929,9 +894,7 @@ class TestTranslateResponse:
def test_multiple_text_parts(self):
"""Multiple output_text parts become multiple text content blocks."""
response = _make_mock_response(
output=[_make_output_message(["Part 1", "Part 2"])]
)
response = _make_mock_response(output=[_make_output_message(["Part 1", "Part 2"])])
result: Any = _ADAPTER.translate_response(response)
assert len(result["content"]) == 2
assert result["content"][0]["text"] == "Part 1"

417
tests/test_litellm/router_strategy/test_complexity_router.py
View file

@ -14,9 +14,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from pydantic import ValidationError
sys.path.insert(
0, os.path.abspath("../../..")
) # Adds the parent directory to the system path
sys.path.insert(0, os.path.abspath("../../..")) # Adds the parent directory to the system path
import litellm
from litellm import Router
@ -130,9 +128,7 @@ class TestTokenScoring:
tier, score, signals = complexity_router.classify("What is Python?")
# Should be classified as SIMPLE due to short length and simple indicator
assert tier == ComplexityTier.SIMPLE
assert any("short" in s.lower() for s in signals) or any(
"simple" in s.lower() for s in signals
)
assert any("short" in s.lower() for s in signals) or any("simple" in s.lower() for s in signals)
def test_long_prompt_positive_score(self, complexity_router):
"""Long prompts should get positive scores (complex indicator)."""
@ -143,9 +139,7 @@ class TestTokenScoring:
tier, score, signals = complexity_router.classify(long_prompt)
# Should have positive score and detect long token count or technical terms
assert score > 0, f"Expected positive score for long prompt, got {score}"
assert any("long" in s.lower() for s in signals) or any(
"technical" in s.lower() for s in signals
)
assert any("long" in s.lower() for s in signals) or any("technical" in s.lower() for s in signals)
class TestCodePresenceScoring:
@ -220,9 +214,7 @@ class TestMultiStepPatterns:
def test_first_then_pattern(self, complexity_router):
"""'First...then' patterns should increase complexity."""
prompt = (
"First analyze the data, then create a visualization, then write a report"
)
prompt = "First analyze the data, then create a visualization, then write a report"
tier, score, signals = complexity_router.classify(prompt)
assert any("multi-step" in s.lower() for s in signals)
@ -266,9 +258,7 @@ class TestTierAssignment:
)
tier, score, signals = complexity_router.classify(prompt)
# Should detect technical terms
assert any(
"technical" in s.lower() for s in signals
), f"Expected technical signals, got {signals}"
assert any("technical" in s.lower() for s in signals), f"Expected technical signals, got {signals}"
# Score should be positive due to technical content
assert score > 0, f"Expected positive score, got {score}"
@ -468,13 +458,9 @@ class TestConfigOverrides:
complexity_router_config=config,
)
# With very low thresholds, even neutral prompts should be COMPLEX or higher
tier, score, signals = router.classify(
"Explain how HTTP works with REST APIs and distributed systems"
)
tier, score, signals = router.classify("Explain how HTTP works with REST APIs and distributed systems")
# With boundaries this low, should be at least MEDIUM (anything above -0.5)
assert (
tier != ComplexityTier.SIMPLE
), f"Expected non-SIMPLE tier, got {tier} with score {score}"
assert tier != ComplexityTier.SIMPLE, f"Expected non-SIMPLE tier, got {tier} with score {score}"
def test_custom_token_thresholds(self, mock_router_instance):
"""Test custom token thresholds work correctly."""
@ -499,9 +485,7 @@ class TestConfigOverrides:
long_prompt = "This is a test prompt " * 30 # ~120 tokens
tier, score, signals = router.classify(long_prompt)
# Should get token length signal indicating "long"
assert any(
"long" in s.lower() if s else False for s in signals
), f"Expected 'long' signal, got {signals}"
assert any("long" in s.lower() if s else False for s in signals), f"Expected 'long' signal, got {signals}"
class TestCustomTechnicalKeywords:
@ -516,9 +500,7 @@ class TestCustomTechnicalKeywords:
)
assert router.technical_keywords == DEFAULT_TECHNICAL_KEYWORDS + ["udp", "kafka"]
def test_custom_keywords_appended_to_technical_keywords_override(
self, mock_router_instance
):
def test_custom_keywords_appended_to_technical_keywords_override(self, mock_router_instance):
"""Custom keywords should be appended to a technical_keywords override."""
router = ComplexityRouter(
model_name="test-router",
@ -535,9 +517,7 @@ class TestCustomTechnicalKeywords:
router = ComplexityRouter(
model_name="test-router",
litellm_router_instance=mock_router_instance,
complexity_router_config={
"custom_technical_keywords": ["TCP", "udp", "UDP", "kafka"]
},
complexity_router_config={"custom_technical_keywords": ["TCP", "udp", "UDP", "kafka"]},
)
lowered = [kw.lower() for kw in router.technical_keywords]
assert lowered == [kw.lower() for kw in DEFAULT_TECHNICAL_KEYWORDS] + [
@ -560,9 +540,7 @@ class TestCustomTechnicalKeywords:
assert router_absent.technical_keywords == DEFAULT_TECHNICAL_KEYWORDS
assert router_none.technical_keywords == DEFAULT_TECHNICAL_KEYWORDS
def test_prompt_with_only_custom_keywords_scores_technical(
self, mock_router_instance, basic_config
):
def test_prompt_with_only_custom_keywords_scores_technical(self, mock_router_instance, basic_config):
"""A prompt matching only custom keywords should score higher on technicalTerms."""
prompt = "Configure udp multicast between kafka brokers"
baseline_router = ComplexityRouter(
@ -581,9 +559,7 @@ class TestCustomTechnicalKeywords:
_, baseline_score, baseline_signals = baseline_router.classify(prompt)
_, custom_score, custom_signals = custom_router.classify(prompt)
assert not any("technical" in s.lower() for s in baseline_signals)
assert any(
"technical" in s.lower() for s in custom_signals
), f"Expected technical signal, got {custom_signals}"
assert any("technical" in s.lower() for s in custom_signals), f"Expected technical signal, got {custom_signals}"
assert custom_score > baseline_score
@ -776,9 +752,7 @@ class TestKeywordFalsePositives:
prompt = "What is the capital of France?"
tier, score, signals = complexity_router.classify(prompt)
# Should NOT detect code presence from 'api' in 'capital'
assert not any(
"code" in s.lower() for s in signals
), "False positive: got code signal from 'capital'"
assert not any("code" in s.lower() for s in signals), "False positive: got code signal from 'capital'"
# Should be SIMPLE (definition question)
assert tier == ComplexityTier.SIMPLE
@ -787,9 +761,7 @@ class TestKeywordFalsePositives:
prompt = "Explain digital marketing strategies"
tier, score, signals = complexity_router.classify(prompt)
# Should NOT detect code presence from 'git' in 'digital'
assert not any(
"code" in s.lower() for s in signals
), "False positive: got code signal from 'digital'"
assert not any("code" in s.lower() for s in signals), "False positive: got code signal from 'digital'"
def test_try_not_in_entry(self, complexity_router):
"""'try' should not match in 'entry'."""
@ -803,43 +775,33 @@ class TestKeywordFalsePositives:
"""'error' should not match in 'terrorism'."""
prompt = "The country is dealing with terrorism"
tier, score, signals = complexity_router.classify(prompt)
assert not any(
"code" in s.lower() for s in signals
), "False positive: got code signal from 'terrorism'"
assert not any("code" in s.lower() for s in signals), "False positive: got code signal from 'terrorism'"
def test_class_not_in_classical(self, complexity_router):
"""'class' should not match in 'classical'."""
prompt = "I enjoy listening to classical music"
tier, score, signals = complexity_router.classify(prompt)
assert not any(
"code" in s.lower() for s in signals
), "False positive: got code signal from 'classical'"
assert not any("code" in s.lower() for s in signals), "False positive: got code signal from 'classical'"
def test_merge_not_in_emerged(self, complexity_router):
"""'merge' should not match in 'emerged'."""
prompt = "A new leader emerged from the crowd"
tier, score, signals = complexity_router.classify(prompt)
assert not any(
"code" in s.lower() for s in signals
), "False positive: got code signal from 'emerged'"
assert not any("code" in s.lower() for s in signals), "False positive: got code signal from 'emerged'"
def test_actual_api_keyword_detected(self, complexity_router):
"""Actual 'api' usage should be detected."""
prompt = "How do I call the REST api endpoint?"
tier, score, signals = complexity_router.classify(prompt)
# Should detect code presence from actual 'api' usage
assert any(
"code" in s.lower() for s in signals
), f"Expected code signal for 'api', got {signals}"
assert any("code" in s.lower() for s in signals), f"Expected code signal for 'api', got {signals}"
def test_actual_git_keyword_detected(self, complexity_router):
"""Actual 'git' usage should be detected."""
prompt = "How do I use git to commit changes?"
tier, score, signals = complexity_router.classify(prompt)
# Should detect code presence from actual 'git' usage
assert any(
"code" in s.lower() for s in signals
), f"Expected code signal for 'git', got {signals}"
assert any("code" in s.lower() for s in signals), f"Expected code signal for 'git', got {signals}"
class TestEdgeCases:
@ -859,9 +821,7 @@ class TestEdgeCases:
# Should have positive score due to length
assert score > 0, f"Expected positive score for very long prompt, got {score}"
# Should detect long token count
assert any(
"long" in s.lower() for s in signals
), f"Expected 'long' signal, got {signals}"
assert any("long" in s.lower() for s in signals), f"Expected 'long' signal, got {signals}"
def test_unicode_prompt(self, complexity_router):
"""Test handling of unicode characters."""
@ -879,9 +839,7 @@ class TestEdgeCases:
"""
tier, score, signals = complexity_router.classify(prompt)
# The "step N" pattern should be detected
assert any(
"multi-step" in s.lower() for s in signals
), f"Expected multi-step signal, got {signals}"
assert any("multi-step" in s.lower() for s in signals), f"Expected multi-step signal, got {signals}"
class TestRouterComplexityDeploymentMethods:
@ -1019,9 +977,7 @@ class TestAsyncPreRoutingHookMultiFormat:
assert result.messages is not None
@pytest.mark.asyncio
async def test_should_route_with_responses_api_string_input(
self, complexity_router
):
async def test_should_route_with_responses_api_string_input(self, complexity_router):
"""Test routing with Responses API string input via handler dispatch."""
from litellm.llms.openai.responses.guardrail_translation.handler import (
OpenAIResponsesHandler,
@ -1109,9 +1065,7 @@ class TestAsyncPreRoutingHookMultiFormat:
assert result.model is not None
@pytest.mark.asyncio
async def test_should_return_none_when_no_messages_or_input(
self, complexity_router
):
async def test_should_return_none_when_no_messages_or_input(self, complexity_router):
"""Test that None is returned when neither messages nor input is available."""
result = await complexity_router.async_pre_routing_hook(
model="test-model",
@ -1122,9 +1076,7 @@ class TestAsyncPreRoutingHookMultiFormat:
assert result is None
@pytest.mark.asyncio
async def test_should_prefer_original_messages_over_conversion(
self, complexity_router
):
async def test_should_prefer_original_messages_over_conversion(self, complexity_router):
"""Test that original messages are used when both messages and input are available."""
messages = [{"role": "user", "content": "What is 2+2?"}]
result = await complexity_router.async_pre_routing_hook(
@ -1136,9 +1088,7 @@ class TestAsyncPreRoutingHookMultiFormat:
assert result.messages == messages
@pytest.mark.asyncio
async def test_should_include_instructions_in_classification(
self, complexity_router
):
async def test_should_include_instructions_in_classification(self, complexity_router):
"""Test that Responses API instructions influence classification via system message."""
from litellm.llms.openai.responses.guardrail_translation.handler import (
OpenAIResponsesHandler,
@ -1175,9 +1125,7 @@ class TestExtractUserMessageAndSystemPrompt:
{"role": "assistant", "content": "Hi!"},
{"role": "user", "content": "How are you?"},
]
user_msg, sys_prompt = ComplexityRouter._extract_user_message_and_system_prompt(
messages
)
user_msg, sys_prompt = ComplexityRouter._extract_user_message_and_system_prompt(messages)
assert user_msg == "How are you?"
assert sys_prompt == "You are helpful."
@ -1187,9 +1135,7 @@ class TestExtractUserMessageAndSystemPrompt:
{"role": "system", "content": "You are helpful."},
{"role": "assistant", "content": "Hi!"},
]
user_msg, sys_prompt = ComplexityRouter._extract_user_message_and_system_prompt(
messages
)
user_msg, sys_prompt = ComplexityRouter._extract_user_message_and_system_prompt(messages)
assert user_msg is None
assert sys_prompt == "You are helpful."
@ -1207,17 +1153,13 @@ class TestExtractUserMessageAndSystemPrompt:
],
}
]
user_msg, sys_prompt = ComplexityRouter._extract_user_message_and_system_prompt(
messages
)
user_msg, sys_prompt = ComplexityRouter._extract_user_message_and_system_prompt(messages)
assert user_msg == "Describe this image"
assert sys_prompt is None
def test_should_handle_empty_messages(self):
"""Test with empty messages list."""
user_msg, sys_prompt = ComplexityRouter._extract_user_message_and_system_prompt(
[]
)
user_msg, sys_prompt = ComplexityRouter._extract_user_message_and_system_prompt([])
assert user_msg is None
assert sys_prompt is None
@ -1282,17 +1224,13 @@ class TestLLMClassifier:
assert tier == ComplexityTier.SIMPLE
@pytest.mark.asyncio
async def test_aclassify_llm_success_routes_by_llm_verdict(
self, llm_complexity_router, mock_router_instance
):
async def test_aclassify_llm_success_routes_by_llm_verdict(self, llm_complexity_router, mock_router_instance):
"""A well-formed structured LLM response should decide the tier directly.
Uses a prompt that heuristic scoring alone would classify as SIMPLE, to prove
the LLM verdict -- not the heuristic scorer -- is what decided the tier.
"""
mock_router_instance.acompletion = AsyncMock(
return_value=_llm_response('{"tier": "COMPLEX"}')
)
mock_router_instance.acompletion = AsyncMock(return_value=_llm_response('{"tier": "COMPLEX"}'))
tier, score, signals = await llm_complexity_router.aclassify("hi")
assert tier == ComplexityTier.COMPLEX
assert "llm-classifier:COMPLEX" in signals
@ -1311,13 +1249,9 @@ class TestLLMClassifier:
sees no user_api_key/team_id/user_id and silently drops all spend logging
and budget accounting for the classifier call.
"""
mock_router_instance.acompletion = AsyncMock(
return_value=_llm_response('{"tier": "SIMPLE"}')
)
mock_router_instance.acompletion = AsyncMock(return_value=_llm_response('{"tier": "SIMPLE"}'))
request_metadata = {"user_api_key": "sk-abc", "user_api_key_team_id": "team-1"}
await llm_complexity_router.aclassify(
"hi", request_kwargs={"litellm_metadata": request_metadata}
)
await llm_complexity_router.aclassify("hi", request_kwargs={"litellm_metadata": request_metadata})
call_kwargs = mock_router_instance.acompletion.call_args.kwargs
assert call_kwargs["metadata"] == request_metadata
@ -1333,18 +1267,14 @@ class TestLLMClassifier:
business touching, so it must be stripped while the rest of the attribution
metadata (key/team) is preserved.
"""
mock_router_instance.acompletion = AsyncMock(
return_value=_llm_response('{"tier": "SIMPLE"}')
)
mock_router_instance.acompletion = AsyncMock(return_value=_llm_response('{"tier": "SIMPLE"}'))
request_metadata = {
"user_api_key": "sk-abc",
"user_api_key_team_id": "team-1",
"user_api_key_budget_reservation": {"reserved_cost": 1.0},
"user_api_key_auth": {"models": ["gpt-4o"], "budget_reservation": {"reserved_cost": 1.0}},
}
await llm_complexity_router.aclassify(
"hi", request_kwargs={"litellm_metadata": request_metadata}
)
await llm_complexity_router.aclassify("hi", request_kwargs={"litellm_metadata": request_metadata})
call_kwargs = mock_router_instance.acompletion.call_args.kwargs
# user_api_key_budget_reservation is stripped (budget enforcement) while
# user_api_key_auth is kept so _filter_deployments_by_model_access_groups
@ -1391,13 +1321,9 @@ class TestLLMClassifier:
assert tier == ComplexityTier.SIMPLE
@pytest.mark.asyncio
async def test_pre_routing_hook_uses_llm_classifier_end_to_end(
self, llm_complexity_router, mock_router_instance
):
async def test_pre_routing_hook_uses_llm_classifier_end_to_end(self, llm_complexity_router, mock_router_instance):
"""The full pre-routing hook should route using the LLM classifier's verdict."""
mock_router_instance.acompletion = AsyncMock(
return_value=_llm_response('{"tier": "REASONING"}')
)
mock_router_instance.acompletion = AsyncMock(return_value=_llm_response('{"tier": "REASONING"}'))
request_metadata = {"user_api_key": "sk-abc", "user_api_key_team_id": "team-1"}
result = await llm_complexity_router.async_pre_routing_hook(
model="test-model",
@ -1410,6 +1336,186 @@ class TestLLMClassifier:
assert call_kwargs["metadata"] == request_metadata
class TestRouterPreRoutingAliasOverrides:
"""
Regression tests for: litellm_params configured on a complexity-router alias
entry (e.g. `cache_control_injection_points`, `drop_params`) were silently
dropped, because `async_pre_routing_hook` swaps `model` from the alias name
to the selected tier's model *before* the deployment lookup - so the actual
outbound call only ever merges in the tier deployment's own litellm_params,
never the alias's.
"""
def _make_router(self) -> Router:
return Router(
model_list=[
{
"model_name": "smart-router",
"litellm_params": {
"model": "auto_router/complexity_router",
"drop_params": True,
"cache_control_injection_points": [{"location": "message", "role": "system"}],
"complexity_router_config": {
"tiers": {
"SIMPLE": "gpt-4o-mini",
"MEDIUM": "gpt-4o",
}
},
"complexity_router_default_model": "gpt-4o",
},
},
{
"model_name": "gpt-4o-mini",
"litellm_params": {"model": "openai/gpt-4o-mini"},
},
{
"model_name": "gpt-4o",
"litellm_params": {"model": "openai/gpt-4o"},
},
]
)
@pytest.mark.asyncio
async def test_alias_litellm_params_applied_to_request_kwargs(self):
"""cache_control_injection_points/drop_params set on the alias entry
reach the outbound request even though the tier deployment is what
actually gets called."""
router = self._make_router()
request_kwargs: Dict = {}
result = await router.async_pre_routing_hook(
model="smart-router",
request_kwargs=request_kwargs,
messages=[{"role": "user", "content": "hi"}],
)
assert result is not None
assert request_kwargs["drop_params"] is True
assert request_kwargs["cache_control_injection_points"] == [{"location": "message", "role": "system"}]
@pytest.mark.asyncio
async def test_alias_overrides_exclude_only_model(self):
"""`model` (the alias marker, e.g. auto_router/complexity_router) is
excluded since it's never a real provider model. Router-only fields
like complexity_router_config DO flow through into request_kwargs at
this layer - they're filtered from the actual outbound LLM call
downstream by litellm.types.utils.all_litellm_params instead, not by
the router's pre-routing hook. See test_router_init_only_params_are_
never_sent_to_a_provider for the guard on that downstream filter."""
router = self._make_router()
request_kwargs: Dict = {}
await router.async_pre_routing_hook(
model="smart-router",
request_kwargs=request_kwargs,
messages=[{"role": "user", "content": "hi"}],
)
assert "model" not in request_kwargs
assert request_kwargs["complexity_router_config"] == {
"tiers": {
"SIMPLE": "gpt-4o-mini",
"MEDIUM": "gpt-4o",
}
}
assert request_kwargs["complexity_router_default_model"] == "gpt-4o"
def test_router_init_only_params_are_never_sent_to_a_provider(self):
"""The router's pre-routing hook only excludes `model` (see
test_alias_overrides_exclude_only_model above) - every other alias
litellm_param, including router-init-only fields like
complexity_router_config, flows into request_kwargs unfiltered. That's
only safe because litellm.completion()/acompletion() itself strips
anything listed in all_litellm_params before building the provider
request. If one of these keys is ever removed from that list, it
ships raw to the real provider as extra_body - verified live via
litellm.completion(..., complexity_router_config={...}) landing in
extra_body before this list included it."""
from litellm.types.utils import all_litellm_params
router_init_only_params = (
"auto_router_config_path",
"auto_router_config",
"auto_router_default_model",
"auto_router_embedding_model",
"complexity_router_config",
"complexity_router_default_model",
"adaptive_router_config",
"adaptive_router_default_model",
"quality_router_config",
"quality_router_default_model",
)
for param in router_init_only_params:
assert param in all_litellm_params, (
f"{param} must stay in litellm.types.utils.all_litellm_params - "
"removing it means it ships raw to the real provider as extra_body"
)
@pytest.mark.asyncio
async def test_caller_supplied_kwargs_are_not_overwritten(self):
"""A value the caller already passed for this request takes
precedence over the alias's configured default."""
router = self._make_router()
request_kwargs: Dict = {"drop_params": False}
await router.async_pre_routing_hook(
model="smart-router",
request_kwargs=request_kwargs,
messages=[{"role": "user", "content": "hi"}],
)
assert request_kwargs["drop_params"] is False
@pytest.mark.asyncio
async def test_non_alias_model_is_untouched(self):
"""A plain (non-router-alias) model name is not affected by the
alias-override merge at all."""
router = self._make_router()
request_kwargs: Dict = {}
result = await router.async_pre_routing_hook(
model="gpt-4o-mini",
request_kwargs=request_kwargs,
messages=[{"role": "user", "content": "hi"}],
)
assert result is None
assert request_kwargs == {}
@pytest.mark.asyncio
async def test_adaptive_router_alias_overrides_survive_reload(self):
"""Alias litellm_params are read fresh from self.model_list at request
time (not cached at init), so a set_model_list() reload (e.g.
/config/reload) - which rebuilds self.model_list but leaves an
already-built AdaptiveRouter alone - can't leave them stale."""
model_list = [
{
"model_name": "smart-router",
"litellm_params": {
"model": "auto_router/adaptive_router",
"drop_params": True,
"adaptive_router_config": {"available_models": ["gpt-4o-mini"]},
},
},
{
"model_name": "gpt-4o-mini",
"litellm_params": {"model": "openai/gpt-4o-mini"},
},
]
router = Router(model_list=model_list)
router.set_model_list(model_list)
assert "smart-router" in router.adaptive_routers
request_kwargs: Dict = {}
await router.async_pre_routing_hook(
model="smart-router",
request_kwargs=request_kwargs,
messages=[{"role": "user", "content": "hi"}],
)
assert request_kwargs["drop_params"] is True
class TestAdaptiveSoftFloors:
def test_adaptive_defaults_use_cost_weighted_cold_policy(self):
config = ComplexityRouterConfig(
@ -1430,9 +1536,7 @@ class TestAdaptiveSoftFloors:
"model": "openai/gpt-4o-mini",
"input_cost_per_token": 0.00000015,
},
"model_info": {
"adaptive_router_preferences": {"quality_tier": 1, "strengths": []}
},
"model_info": {"adaptive_router_preferences": {"quality_tier": 1, "strengths": []}},
},
{
"model_name": "premium",
@ -1440,9 +1544,7 @@ class TestAdaptiveSoftFloors:
"model": "openai/gpt-4o",
"input_cost_per_token": 0.000005,
},
"model_info": {
"adaptive_router_preferences": {"quality_tier": 3, "strengths": []}
},
"model_info": {"adaptive_router_preferences": {"quality_tier": 3, "strengths": []}},
},
]
router.model_name_to_deployment_indices = {"cheap": [0], "premium": [1]}
@ -1467,9 +1569,7 @@ class TestAdaptiveSoftFloors:
with pytest.raises(ValidationError):
ComplexityRouterConfig(adaptive=True, tiers={"SIMPLE": []})
def test_cold_start_randomly_samples_unobserved_classified_tier_models(
self, adaptive_router_instance
):
def test_cold_start_randomly_samples_unobserved_classified_tier_models(self, adaptive_router_instance):
cr = ComplexityRouter(
model_name="hybrid",
litellm_router_instance=adaptive_router_instance,
@ -1498,9 +1598,7 @@ class TestAdaptiveSoftFloors:
"premium",
}
def test_get_model_for_tier_list_without_adaptive_random_choice(
self, mock_router_instance
):
def test_get_model_for_tier_list_without_adaptive_random_choice(self, mock_router_instance):
router = ComplexityRouter(
model_name="test",
litellm_router_instance=mock_router_instance,
@ -1519,9 +1617,7 @@ class TestAdaptiveSoftFloors:
choice.assert_called_once_with(pool)
assert router.get_model_for_tier(ComplexityTier.MEDIUM) == "mid"
def test_soft_floor_prefers_home_tier_when_posteriors_equal(
self, adaptive_router_instance, hybrid_config
):
def test_soft_floor_prefers_home_tier_when_posteriors_equal(self, adaptive_router_instance, hybrid_config):
from litellm.router_strategy.adaptive_router.bandit import BanditCell
from litellm.types.router import RequestType
@ -1533,9 +1629,7 @@ class TestAdaptiveSoftFloors:
adaptive = cr._ensure_adaptive_router()
assert adaptive is not None
for model in ("cheap", "premium"):
adaptive._cells[(RequestType.GENERAL, model)] = BanditCell(
alpha=5.0, beta=5.0
)
adaptive._cells[(RequestType.GENERAL, model)] = BanditCell(alpha=5.0, beta=5.0)
# Equal quality samples; home-tier penalty should favor cheap for SIMPLE.
with patch(
@ -1545,9 +1639,7 @@ class TestAdaptiveSoftFloors:
picked = cr._soft_floor_pick(ComplexityTier.SIMPLE, "hi")
assert picked == "cheap"
def test_soft_floor_allows_cross_tier_when_posterior_dominates(
self, adaptive_router_instance, hybrid_config
):
def test_soft_floor_allows_cross_tier_when_posterior_dominates(self, adaptive_router_instance, hybrid_config):
from litellm.router_strategy.adaptive_router.bandit import BanditCell
from litellm.types.router import RequestType
@ -1558,12 +1650,8 @@ class TestAdaptiveSoftFloors:
)
adaptive = cr._ensure_adaptive_router()
assert adaptive is not None
adaptive._cells[(RequestType.GENERAL, "cheap")] = BanditCell(
alpha=1.0, beta=20.0
)
adaptive._cells[(RequestType.GENERAL, "premium")] = BanditCell(
alpha=20.0, beta=1.0
)
adaptive._cells[(RequestType.GENERAL, "cheap")] = BanditCell(alpha=1.0, beta=20.0)
adaptive._cells[(RequestType.GENERAL, "premium")] = BanditCell(alpha=20.0, beta=1.0)
with patch(
"litellm.router_strategy.adaptive_router.bandit.thompson_sample",
@ -1572,9 +1660,7 @@ class TestAdaptiveSoftFloors:
picked = cr._soft_floor_pick(ComplexityTier.SIMPLE, "hi")
assert picked == "premium"
def test_reused_model_has_zero_distance_in_each_configured_tier(
self, adaptive_router_instance
):
def test_reused_model_has_zero_distance_in_each_configured_tier(self, adaptive_router_instance):
from litellm.router_strategy.adaptive_router.bandit import BanditCell
from litellm.types.router import RequestType
@ -1593,9 +1679,7 @@ class TestAdaptiveSoftFloors:
adaptive = cr._ensure_adaptive_router()
assert adaptive is not None
for model in ("cheap", "premium"):
adaptive._cells[(RequestType.GENERAL, model)] = BanditCell(
alpha=6.0, beta=5.0
)
adaptive._cells[(RequestType.GENERAL, model)] = BanditCell(alpha=6.0, beta=5.0)
request_kwargs: Dict = {"metadata": {}}
with patch(
@ -1604,20 +1688,14 @@ class TestAdaptiveSoftFloors:
):
cr._soft_floor_pick(ComplexityTier.MEDIUM, "hi", request_kwargs)
candidates = request_kwargs["metadata"]["adaptive_router_decision"][
"candidates"
]
assert {
candidate["model"]: candidate["tier_distance"] for candidate in candidates
} == {
candidates = request_kwargs["metadata"]["adaptive_router_decision"]["candidates"]
assert {candidate["model"]: candidate["tier_distance"] for candidate in candidates} == {
"cheap": 0,
"premium": 0,
}
@pytest.mark.asyncio
async def test_pre_routing_hook_adaptive_stashes_chosen_model(
self, adaptive_router_instance, hybrid_config
):
async def test_pre_routing_hook_adaptive_stashes_chosen_model(self, adaptive_router_instance, hybrid_config):
cr = ComplexityRouter(
model_name="hybrid",
litellm_router_instance=adaptive_router_instance,
@ -1631,10 +1709,7 @@ class TestAdaptiveSoftFloors:
)
assert result is not None
assert result.model in {"cheap", "premium"}
assert (
request_kwargs["metadata"].get("adaptive_router_chosen_model")
== result.model
)
assert request_kwargs["metadata"].get("adaptive_router_chosen_model") == result.model
decision = request_kwargs["metadata"]["adaptive_router_decision"]
assert decision["phase"] == "cold_start"
assert decision["classified_tier"] == "SIMPLE"
@ -1657,9 +1732,7 @@ class TestLexicalKeywordTierRules:
}
@pytest.mark.asyncio
async def test_matching_rule_overrides_scoring(
self, mock_router_instance, rule_config
):
async def test_matching_rule_overrides_scoring(self, mock_router_instance, rule_config):
"""A prompt hitting a rule keyword routes to that tier, not the scored tier."""
router = ComplexityRouter(
model_name="test-router",
@ -1745,9 +1818,7 @@ class TestLexicalKeywordTierRules:
assert router._lexical_tier_override("nothing relevant here") is None
@pytest.mark.asyncio
async def test_no_rule_match_falls_back_to_scoring(
self, mock_router_instance, basic_config
):
async def test_no_rule_match_falls_back_to_scoring(self, mock_router_instance, basic_config):
"""A prompt that matches no rule is classified by the scorer as usual."""
config = {
**basic_config,
@ -1768,9 +1839,7 @@ class TestLexicalKeywordTierRules:
assert result is not None
assert result.model == "gpt-4o-mini" # SIMPLE via scoring, rule did not fire
def test_word_boundary_avoids_substring_false_positive(
self, mock_router_instance, basic_config
):
def test_word_boundary_avoids_substring_false_positive(self, mock_router_instance, basic_config):
"""A single-word rule keyword must not match inside a larger word."""
config = {
**basic_config,
@ -1788,10 +1857,7 @@ class TestLexicalKeywordTierRules:
def _make_embedding_response(vectors: List[List[float]]) -> "litellm.EmbeddingResponse":
return litellm.EmbeddingResponse(
model="fake-embed",
data=[
{"embedding": vec, "index": idx, "object": "embedding"}
for idx, vec in enumerate(vectors)
],
data=[{"embedding": vec, "index": idx, "object": "embedding"} for idx, vec in enumerate(vectors)],
object="list",
)
@ -1818,8 +1884,7 @@ class FakeEmbeddingRouter:
def _vectors(self, docs: List[str]) -> List[List[float]]:
return [
[1.0, 0.0] if any(marker in doc.lower() for marker in self._CLUSTER_MARKERS) else [0.0, 1.0]
for doc in docs
[1.0, 0.0] if any(marker in doc.lower() for marker in self._CLUSTER_MARKERS) else [0.0, 1.0] for doc in docs
]
@staticmethod
@ -2358,9 +2423,7 @@ class TestRoutingDecisionCauseLogging:
verbose_router_logger.removeHandler(caplog.handler)
@pytest.mark.asyncio
async def test_literal_keyword_match_logs_its_cause(
self, mock_router_instance, basic_config, router_log_capture
):
async def test_literal_keyword_match_logs_its_cause(self, mock_router_instance, basic_config, router_log_capture):
config = {
**basic_config,
"keyword_tier_rules": [{"keywords": ["deploy to k8s"], "tier": "REASONING"}],
@ -2406,9 +2469,7 @@ class TestRoutingDecisionCauseLogging:
assert "cause=literal_keyword_match" not in router_log_capture.text
@pytest.mark.asyncio
async def test_complexity_scorer_logs_its_cause(
self, mock_router_instance, basic_config, router_log_capture
):
async def test_complexity_scorer_logs_its_cause(self, mock_router_instance, basic_config, router_log_capture):
# No keyword rules -> the scorer decides, and its line must be tagged as such.
router = ComplexityRouter(
model_name="test-router",