diff --git a/litellm/router_strategy/tag_based_routing.py b/litellm/router_strategy/tag_based_routing.py index 07143af38a20..fdaabcf302c6 100644 --- a/litellm/router_strategy/tag_based_routing.py +++ b/litellm/router_strategy/tag_based_routing.py @@ -7,7 +7,7 @@ """ import re -from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Union +from typing import TYPE_CHECKING, Any, Literal, Optional, Union from litellm._logging import verbose_logger from litellm.types.router import RouterErrors @@ -21,8 +21,8 @@ def _is_valid_deployment_tag_regex( - tag_regexes: List[str], - header_strings: List[str], + tag_regexes: list[str], + header_strings: list[str], ) -> Optional[str]: """ Test compiled regex patterns against "Header-Name: value" strings. @@ -44,7 +44,7 @@ def _is_valid_deployment_tag_regex( def is_valid_deployment_tag( - deployment_tags: List[str], request_tags: List[str], match_any: bool = True + deployment_tags: list[str], request_tags: list[str], match_any: bool = True ) -> bool: """ Check if a tag is valid, the matching can be either any or all based on `match_any` flag @@ -73,10 +73,10 @@ def is_valid_deployment_tag( def _match_deployment( deployment: Any, - request_tags: Optional[List[str]], - header_strings: List[str], + request_tags: Optional[list[str]], + header_strings: list[str], match_any: bool, -) -> Optional[Dict[str, str]]: +) -> Optional[dict[str, str]]: """ Determine whether *deployment* matches the current request. @@ -89,8 +89,8 @@ def _match_deployment( ran and failed, so the regex cannot override strict-tag policy. """ litellm_params = deployment.get("litellm_params", {}) - deployment_tags: Optional[List[str]] = litellm_params.get("tags") - deployment_tag_regex: Optional[List[str]] = litellm_params.get("tag_regex") + deployment_tags: Optional[list[str]] = litellm_params.get("tags") + deployment_tag_regex: Optional[list[str]] = litellm_params.get("tag_regex") # 1. Exact tag match (existing behaviour). if deployment_tags and request_tags: @@ -118,11 +118,55 @@ def _match_deployment( return None +def _split_tags(tags: list[str]) -> tuple[list[str], list[str]]: + positive = [t for t in tags if not t.startswith("!")] + excluded = [tag[1:] for tag in tags if tag.startswith("!") and len(tag) > 1] + return positive, excluded + + +def _exclude_deployments( + deployments: Union[list[Any], dict[Any, Any]], + excluded_set: frozenset[str], +) -> list[Any]: + if not excluded_set: + return list(deployments) + return [ + d + for d in deployments + if not excluded_set.intersection(d.get("litellm_params", {}).get("tags") or []) + ] + + +def _require_candidates( + candidates: list[Any], + model: str, + request_tags: Any, +) -> list[Any]: + if not candidates: + raise ValueError( + f"{RouterErrors.no_deployments_with_tag_routing.value}." + f" Passed model={model} and tags={request_tags}" + ) + return candidates + + +def _ban_only_base_pool( + deployments: Union[list[Any], dict[Any, Any]], +) -> list[Any]: + # Mirrors untagged-request semantics so callers can't use !tags to escape the default pool. + defaults = [ + d + for d in deployments + if "default" in (d.get("litellm_params", {}).get("tags") or []) + ] + return defaults if defaults else list(deployments) + + async def get_deployments_for_tag( llm_router_instance: LitellmRouter, model: str, # used to raise the correct error - healthy_deployments: Union[List[Any], Dict[Any, Any]], - request_kwargs: Optional[Dict[Any, Any]] = None, + healthy_deployments: Union[list[Any], dict[Any, Any]], + request_kwargs: Optional[dict[Any, Any]] = None, metadata_variable_name: Literal["metadata", "litellm_metadata"] = "metadata", ): """ @@ -140,16 +184,9 @@ async def get_deployments_for_tag( ) return healthy_deployments - if healthy_deployments is None: - verbose_logger.debug( - "get_deployments_for_tag: healthy_deployments is None returning healthy_deployments" - ) - return healthy_deployments - - # Tag filtering applies only when there is at least one deployment to evaluate. - if isinstance(healthy_deployments, list) and len(healthy_deployments) == 0: + if not healthy_deployments: verbose_logger.debug( - "get_deployments_for_tag: empty candidate set; skipping tag filter" + "get_deployments_for_tag: empty or None healthy_deployments; skipping tag filter" ) return healthy_deployments @@ -164,34 +201,42 @@ async def get_deployments_for_tag( # Build header strings for regex matching from what the proxy already stores. # Currently we match against User-Agent; format matches "^User-Agent: claude-code/..." user_agent = metadata.get("user_agent", "") - header_strings: List[str] = [f"User-Agent: {user_agent}"] if user_agent else [] + header_strings: list[str] = [f"User-Agent: {user_agent}"] if user_agent else [] - new_healthy_deployments: List[Any] = [] - default_deployments: List[Any] = [] + positive_tags, excluded_patterns = _split_tags(request_tags or []) + + excluded_set = frozenset(excluded_patterns) + candidates = _exclude_deployments(healthy_deployments, excluded_set) - # Only activate header-based regex filtering when at least one deployment in - # the candidate set has tag_regex configured. This preserves existing - # behaviour for operators who use plain tags: a request that carries a - # User-Agent (all proxy requests do) but targets deployments with no - # tag_regex will continue to use the original tag-only code path. has_regex_deployments = any( - d.get("litellm_params", {}).get("tag_regex") for d in healthy_deployments + d.get("litellm_params", {}).get("tag_regex") for d in candidates ) - has_tag_filter = bool(request_tags) or ( + has_tag_filter = bool(positive_tags) or ( bool(header_strings) and has_regex_deployments ) + ban_only = bool(excluded_set) and not has_tag_filter + + if ban_only: + pool = _exclude_deployments( + _ban_only_base_pool(healthy_deployments), excluded_set + ) + return _require_candidates(pool, model, request_tags) + + new_healthy_deployments: list[Any] = [] + default_deployments: list[Any] = [] + if has_tag_filter: verbose_logger.debug( "get_deployments_for_tag routing: request_tags=%s user_agent=%s", request_tags, user_agent, ) - for deployment in healthy_deployments: + for deployment in candidates: deployment_tags = deployment.get("litellm_params", {}).get("tags") match_result = _match_deployment( deployment=deployment, - request_tags=request_tags, + request_tags=positive_tags, header_strings=header_strings, match_any=match_any, ) @@ -203,10 +248,6 @@ async def get_deployments_for_tag( match_result["matched_via"], match_result["matched_value"], ) - # Record provenance in metadata so it flows to SpendLogs. - # Written only for the first match — load balancer selects one - # deployment from new_healthy_deployments, so overwriting on - # subsequent matches would produce misleading observability data. if "tag_routing" not in metadata: metadata["tag_routing"] = { "matched_deployment": deployment.get("model_name"), @@ -222,7 +263,8 @@ async def get_deployments_for_tag( if len(new_healthy_deployments) == 0 and len(default_deployments) == 0: raise ValueError( - f"{RouterErrors.no_deployments_with_tag_routing.value}. Passed model={model} and tags={request_tags}" + f"{RouterErrors.no_deployments_with_tag_routing.value}." + f" Passed model={model} and tags={request_tags}" ) return ( @@ -249,9 +291,9 @@ async def get_deployments_for_tag( def _get_tags_from_request_kwargs( - request_kwargs: Optional[Dict[Any, Any]] = None, + request_kwargs: Optional[dict[Any, Any]] = None, metadata_variable_name: Literal["metadata", "litellm_metadata"] = "metadata", -) -> List[str]: +) -> list[str]: """ Helper to get tags from request kwargs diff --git a/tests/test_litellm/router_strategy/test_router_tag_routing.py b/tests/test_litellm/router_strategy/test_router_tag_routing.py index a6e39ec3c0ad..c257acd4b897 100644 --- a/tests/test_litellm/router_strategy/test_router_tag_routing.py +++ b/tests/test_litellm/router_strategy/test_router_tag_routing.py @@ -1,13 +1,9 @@ #### What this tests #### # This tests litellm router -import asyncio import os import sys -import time -import traceback -import openai import pytest sys.path.insert( @@ -15,15 +11,9 @@ ) # Adds the parent directory to the system path import logging import os -from collections import defaultdict -from concurrent.futures import ThreadPoolExecutor -from unittest.mock import AsyncMock, MagicMock, patch -import httpx -from dotenv import load_dotenv import litellm -from litellm import Router from litellm._logging import verbose_logger @@ -66,10 +56,7 @@ async def test_router_free_paid_tier(): mock_response="Tell me a joke.", ) - print("Response: ", response) - response_extra_info = response._hidden_params - print("response_extra_info: ", response_extra_info) assert response_extra_info["model_id"] == "very-cheap-model" @@ -82,10 +69,7 @@ async def test_router_free_paid_tier(): mock_response="Tell me a joke.", ) - print("Response: ", response) - response_extra_info = response._hidden_params - print("response_extra_info: ", response_extra_info) assert response_extra_info["model_id"] == "very-expensive-model" @@ -141,10 +125,7 @@ async def test_router_free_paid_tier_embeddings(): mock_response=[1, 2, 3], ) - print("Response: ", response) - response_extra_info = response._hidden_params - print("response_extra_info: ", response_extra_info) assert response_extra_info["model_id"] == "very-cheap-model" @@ -157,10 +138,7 @@ async def test_router_free_paid_tier_embeddings(): mock_response=[1, 2, 3], ) - print("Response: ", response) - response_extra_info = response._hidden_params - print("response_extra_info: ", response_extra_info) assert response_extra_info["model_id"] == "very-expensive-model" @@ -212,10 +190,7 @@ async def test_default_tagged_deployments(): mock_response="Tell me a joke.", ) - print("Response: ", response) - response_extra_info = response._hidden_params - print("response_extra_info: ", response_extra_info) assert response_extra_info["model_id"] == "default-model" @@ -228,10 +203,7 @@ async def test_default_tagged_deployments(): mock_response="Tell me a joke.", ) - print("Response: ", response) - response_extra_info = response._hidden_params - print("response_extra_info: ", response_extra_info) assert response_extra_info["model_id"] == "default-model" @@ -244,10 +216,7 @@ async def test_default_tagged_deployments(): mock_response="Tell me a joke.", ) - print("Response: ", response) - response_extra_info = response._hidden_params - print("response_extra_info: ", response_extra_info) assert response_extra_info["model_id"] == "default-model" @@ -257,10 +226,6 @@ async def test_error_from_tag_routing(): """ Tests the correct error raised when no deployments found for tag """ - import logging - - from litellm._logging import verbose_logger - verbose_logger.setLevel(logging.DEBUG) router = litellm.Router( model_list=[ @@ -294,7 +259,7 @@ async def test_error_from_tag_routing(): ) try: - response = await router.acompletion( + await router.acompletion( model="gpt-4", messages=[{"role": "user", "content": "Tell me a joke."}], metadata={"tags": ["paid"]}, @@ -306,7 +271,6 @@ async def test_error_from_tag_routing(): from litellm.types.router import RouterErrors assert RouterErrors.no_deployments_with_tag_routing.value in str(e) - print("got expected exception = ", e) pass @@ -413,10 +377,7 @@ async def test_router_free_paid_tier_with_responses_api(): mock_response="Tell me a joke.", ) - print("Response: ", response) - response_extra_info = response._hidden_params - print("response_extra_info: ", response_extra_info) assert response_extra_info["model_id"] == "very-cheap-model" @@ -429,10 +390,7 @@ async def test_router_free_paid_tier_with_responses_api(): mock_response="Tell me a joke.", ) - print("Response: ", response) - response_extra_info = response._hidden_params - print("response_extra_info: ", response_extra_info) assert response_extra_info["model_id"] == "very-expensive-model" @@ -473,3 +431,603 @@ def test_get_tags_from_request_kwargs_various_inputs(): # No relevant keys present assert _get_tags_from_request_kwargs({"foo": "bar"}) == [] + + +# --- _split_tags unit tests --- + + +def test_split_tags_positive_only(): + from litellm.router_strategy.tag_based_routing import _split_tags + + positive, excluded = _split_tags(["paid", "teamA"]) + assert positive == ["paid", "teamA"] + assert excluded == [] + + +def test_split_tags_negation_only(): + from litellm.router_strategy.tag_based_routing import _split_tags + + positive, excluded = _split_tags(["!provider:anthropic"]) + assert positive == [] + assert excluded == ["provider:anthropic"] + + +def test_split_tags_mixed(): + from litellm.router_strategy.tag_based_routing import _split_tags + + positive, excluded = _split_tags( + ["paid", "!provider:anthropic", "!inference:cerebras"] + ) + assert positive == ["paid"] + assert len(excluded) == 2 + + +def test_split_tags_bare_bang_skipped(): + from litellm.router_strategy.tag_based_routing import _split_tags + + # A bare "!" with nothing after it is not a valid negation tag; skip it + positive, excluded = _split_tags(["paid", "!"]) + assert positive == ["paid"] + assert excluded == [] + + +def test_split_tags_empty(): + from litellm.router_strategy.tag_based_routing import _split_tags + + positive, excluded = _split_tags([]) + assert positive == [] + assert excluded == [] + + +# --- get_deployments_for_tag negation integration tests --- + + +@pytest.mark.asyncio() +async def test_negation_excludes_matching_deployments(): + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:anthropic", "model:claude-sonnet-4-6"], + }, + "model_info": {"id": "anthropic-model"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:openai", "model:gpt-4o"], + }, + "model_info": {"id": "openai-model"}, + }, + ], + enable_tag_filtering=True, + ) + + for _ in range(5): + response = await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!provider:anthropic"]}, + mock_response="hi", + ) + assert response._hidden_params["model_id"] == "openai-model" + + +@pytest.mark.asyncio() +async def test_negation_multiple_tags_exclude_multiple_providers(): + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:anthropic"], + }, + "model_info": {"id": "anthropic-model"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:openai"], + }, + "model_info": {"id": "openai-model"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:vertex"], + }, + "model_info": {"id": "vertex-model"}, + }, + ], + enable_tag_filtering=True, + ) + + for _ in range(5): + response = await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!provider:anthropic", "!provider:openai"]}, + mock_response="hi", + ) + assert response._hidden_params["model_id"] == "vertex-model" + + +@pytest.mark.asyncio() +async def test_negation_with_positive_tag(): + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["paid", "provider:anthropic"], + }, + "model_info": {"id": "anthropic-paid"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["paid", "provider:openai"], + }, + "model_info": {"id": "openai-paid"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["free", "provider:openai"], + }, + "model_info": {"id": "openai-free"}, + }, + ], + enable_tag_filtering=True, + ) + + for _ in range(5): + response = await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["paid", "!provider:anthropic"]}, + mock_response="hi", + ) + assert response._hidden_params["model_id"] == "openai-paid" + + +@pytest.mark.asyncio() +async def test_negation_all_excluded_raises(): + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:anthropic"], + }, + "model_info": {"id": "anthropic-model"}, + }, + ], + enable_tag_filtering=True, + ) + + with pytest.raises(Exception) as exc_info: + await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!provider:anthropic"]}, + mock_response="hi", + ) + + from litellm.types.router import RouterErrors + + assert RouterErrors.no_deployments_with_tag_routing.value in str(exc_info.value) + + +@pytest.mark.asyncio() +async def test_negation_ban_only_cannot_escape_default_pool(): + # A ban-only request must not route to tagged deployments outside the default pool. + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["default"], + }, + "model_info": {"id": "default-model"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["paid"], + }, + "model_info": {"id": "paid-model"}, + }, + ], + enable_tag_filtering=True, + ) + + # Sending only "!default" must NOT route to the paid deployment. + # The base pool for ban-only is the default pool; banning the only + # default deployment should raise rather than falling through to paid. + with pytest.raises(Exception) as exc_info: + await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!default"]}, + mock_response="hi", + ) + + from litellm.types.router import RouterErrors + + assert RouterErrors.no_deployments_with_tag_routing.value in str(exc_info.value) + + +@pytest.mark.asyncio() +async def test_negation_ban_only_respects_default_pool(): + # A ban-only request stays within the default pool; non-default deployments + # remain unreachable even when the negation tag is unrelated to the default. + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["default"], + }, + "model_info": {"id": "default-model"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["paid"], + }, + "model_info": {"id": "paid-model"}, + }, + ], + enable_tag_filtering=True, + ) + + # "!paid" bans the paid deployment, but the base pool for ban-only is + # already restricted to defaults; default-model must still be returned. + for _ in range(5): + response = await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!paid"]}, + mock_response="hi", + ) + assert response._hidden_params["model_id"] == "default-model" + + +@pytest.mark.asyncio() +async def test_negation_untagged_deployment_kept(): + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:anthropic"], + }, + "model_info": {"id": "anthropic-model"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + }, + "model_info": {"id": "untagged-model"}, + }, + ], + enable_tag_filtering=True, + ) + + for _ in range(5): + response = await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!provider:anthropic"]}, + mock_response="hi", + ) + assert response._hidden_params["model_id"] == "untagged-model" + + +@pytest.mark.asyncio() +async def test_negation_literal_only_no_partial_match(): + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:anthropic-haiku"], + }, + "model_info": {"id": "anthropic-haiku-model"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:openai"], + }, + "model_info": {"id": "openai-model"}, + }, + ], + enable_tag_filtering=True, + ) + + # "!provider:anthropic" should NOT match "provider:anthropic-haiku" — exact tag match only + for _ in range(5): + response = await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!provider:anthropic"]}, + mock_response="hi", + ) + assert response._hidden_params["model_id"] in ( + "anthropic-haiku-model", + "openai-model", + ) + + +@pytest.mark.asyncio() +async def test_negation_regex_pattern_treated_as_literal(): + # "!provider:(anthropic|openai)" looks like a regex but is treated as a literal string. + # It does NOT exclude deployments tagged "provider:anthropic" or "provider:openai". + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:anthropic"], + }, + "model_info": {"id": "anthropic-model"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:openai"], + }, + "model_info": {"id": "openai-model"}, + }, + ], + enable_tag_filtering=True, + ) + + # The regex-like string matches no deployment tag literally, so all + # candidates survive and both model IDs are reachable. + seen_ids = set() + for _ in range(10): + response = await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!provider:(anthropic|openai)"]}, + mock_response="hi", + ) + seen_ids.add(response._hidden_params["model_id"]) + + assert seen_ids == {"anthropic-model", "openai-model"} + + +@pytest.mark.asyncio() +async def test_positive_tags_unchanged_by_negation(): + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["free"], + }, + "model_info": {"id": "free-model"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["paid"], + }, + "model_info": {"id": "paid-model"}, + }, + ], + enable_tag_filtering=True, + ) + + for _ in range(5): + response = await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["free"]}, + mock_response="hi", + ) + assert response._hidden_params["model_id"] == "free-model" + + +@pytest.mark.asyncio() +async def test_negation_skips_banned_group_and_uses_fallback(): + router = litellm.Router( + model_list=[ + { + "model_name": "primary", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:anthropic"], + }, + "model_info": {"id": "anthropic-primary"}, + }, + { + "model_name": "fallback", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:openai"], + }, + "model_info": {"id": "openai-fallback"}, + }, + ], + fallbacks=[{"primary": ["fallback"]}], + enable_tag_filtering=True, + ) + + response = await router.acompletion( + model="primary", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!provider:anthropic"]}, + mock_response="hi", + ) + assert response._hidden_params["model_id"] == "openai-fallback" + + +@pytest.mark.asyncio() +async def test_negation_exhausts_entire_fallback_chain(): + router = litellm.Router( + model_list=[ + { + "model_name": "primary", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:anthropic"], + }, + "model_info": {"id": "anthropic-primary"}, + }, + { + "model_name": "fallback", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:anthropic"], + }, + "model_info": {"id": "anthropic-fallback"}, + }, + ], + fallbacks=[{"primary": ["fallback"]}], + enable_tag_filtering=True, + ) + + with pytest.raises(Exception) as exc_info: + await router.acompletion( + model="primary", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!provider:anthropic"]}, + mock_response="hi", + ) + + from litellm.types.router import RouterErrors + + assert RouterErrors.no_deployments_with_tag_routing.value in str(exc_info.value) + + +@pytest.mark.asyncio() +async def test_tag_regex_survives_when_negation_removes_other_deployment(): + # Negation removes a plain-tagged deployment; the surviving tag_regex deployment + # is still matched by User-Agent and selected. + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tag_regex": ["^User-Agent: claude-code\\/"], + }, + "model_info": {"id": "claude-code-deployment"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:anthropic"], + }, + "model_info": {"id": "anthropic-deployment"}, + }, + ], + enable_tag_filtering=True, + tag_filtering_match_any=True, + ) + + for _ in range(5): + response = await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!provider:anthropic"], "user_agent": "claude-code/1.2.3"}, + mock_response="hi", + ) + assert response._hidden_params["model_id"] == "claude-code-deployment" + + +@pytest.mark.asyncio() +async def test_negation_removes_tag_regex_deployment_falls_to_ban_only(): + # When a negation tag removes the only tag_regex deployment, no regex deployments + # remain in the candidate pool. has_tag_filter becomes False, ban_only fires, + # and the remaining plain-tagged deployment is returned via the ban-only path. + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tag_regex": ["^User-Agent: claude-code\\/"], + "tags": ["group:claude"], + }, + "model_info": {"id": "claude-code-deployment"}, + }, + { + "model_name": "gpt-4", + "litellm_params": { + "model": "gpt-4o-mini", + "api_base": "https://exampleopenaiendpoint-production.up.railway.app/", + "tags": ["provider:openai"], + }, + "model_info": {"id": "openai-deployment"}, + }, + ], + enable_tag_filtering=True, + tag_filtering_match_any=True, + ) + + # !group:claude removes the tag_regex deployment from candidates, so no regex + # deployments remain. The ban-only path fires and returns the openai deployment. + for _ in range(5): + response = await router.acompletion( + model="gpt-4", + messages=[{"role": "user", "content": "hi"}], + metadata={"tags": ["!group:claude"], "user_agent": "claude-code/1.2.3"}, + mock_response="hi", + ) + assert response._hidden_params["model_id"] == "openai-deployment"