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

feat(router): tag routing denylist support via ! prefix (#31728)

Browse source 
Adds `!` prefix negation to tag-based routing so callers can exclude deployments by exact tag value without enumerating every allowed alternative. `!provider:anthropic` removes all deployments tagged exactly `provider:anthropic` before routing, and positive and negation tags compose. Matching is exact literal membership (frozenset intersection), so there is no regex or ReDoS surface for client-supplied tags. Ban-only requests that carry only negation tags stay within the default pool, mirroring untagged-request semantics so callers can't use negation to escape it. Fallback chains keep working because get_deployments_for_tag runs on each routing hop

Copy of #31680; implementation credit to @deepanshululla

Co-authored-by: deepanshululla <15312873+deepanshululla@users.noreply.github.com>
This commit is contained in:
Mateo Wang 2026-06-30 11:00:30 -07:00 • committed by GitHub
parent 1815636e1c
commit fecaf5c9e5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 674 additions and 95 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

107
litellm/router_strategy/tag_based_routing.py
View file

@ -7,7 +7,7 @@ Use this to route requests between Teams
"""
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 @@ else:
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.
@ -43,7 +43,7 @@ def _is_valid_deployment_tag_regex(
return None
def is_valid_deployment_tag(deployment_tags: List[str], request_tags: List[str], match_any: bool = True) -> bool:
def is_valid_deployment_tag(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
"""
@ -71,10 +71,10 @@ def is_valid_deployment_tag(deployment_tags: List[str], request_tags: List[str],
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.
@ -87,8 +87,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:
@ -114,11 +114,46 @@ 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}. 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",
):
"""
@ -136,13 +171,8 @@ 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:
verbose_logger.debug("get_deployments_for_tag: empty candidate set; skipping tag filter")
if not healthy_deployments:
verbose_logger.debug("get_deployments_for_tag: empty or None healthy_deployments; skipping tag filter")
return healthy_deployments
verbose_logger.debug("request metadata: %s", request_kwargs.get(metadata_variable_name))
@ -154,30 +184,36 @@ 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)
has_regex_deployments = any(d.get("litellm_params", {}).get("tag_regex") for d in candidates)
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] = []
# 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)
has_tag_filter = bool(request_tags) or (bool(header_strings) and has_regex_deployments)
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,
)
@ -189,10 +225,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"),
@ -208,7 +240,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 new_healthy_deployments if len(new_healthy_deployments) > 0 else default_deployments
@ -231,9 +264,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

662
tests/test_litellm/router_strategy/test_router_tag_routing.py
View file

@ -1,29 +1,17 @@
#### What this tests ####
# This tests litellm router
import asyncio
import os
import sys
import time
import traceback
import openai
import pytest
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 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 +54,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 +67,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 +123,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 +136,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 +188,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 +201,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 +214,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 +224,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 +257,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 +269,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
@ -332,16 +294,10 @@ def test_tag_routing_with_list_of_tags_match_all():
from litellm.router_strategy.tag_based_routing import is_valid_deployment_tag
assert is_valid_deployment_tag(["teamA", "teamB"], ["teamA"], match_any=False)
assert is_valid_deployment_tag(
["teamA", "teamB"], ["teamA", "teamB"], match_any=False
)
assert not is_valid_deployment_tag(
["teamA", "teamB", "teamC"], ["teamA", "teamD"], match_any=False
)
assert is_valid_deployment_tag(["teamA", "teamB"], ["teamA", "teamB"], match_any=False)
assert not is_valid_deployment_tag(["teamA", "teamB", "teamC"], ["teamA", "teamD"], match_any=False)
assert not is_valid_deployment_tag(["teamA"], ["teamA", "teamB"], match_any=False)
assert not is_valid_deployment_tag(
["teamA", "teamB"], ["teamA", "teamC"], match_any=False
)
assert not is_valid_deployment_tag(["teamA", "teamB"], ["teamA", "teamC"], match_any=False)
assert not is_valid_deployment_tag(["teamA", "teamB"], [], match_any=False)
assert not is_valid_deployment_tag(["default"], ["teamA"], match_any=False)
@ -413,10 +369,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 +382,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"
@ -455,9 +405,7 @@ def test_get_tags_from_request_kwargs_various_inputs():
assert _get_tags_from_request_kwargs({"metadata": None}) == []
# Indirect via "litellm_params" - metadata inside
assert _get_tags_from_request_kwargs(
{"litellm_params": {"metadata": {"tags": ["paid"]}}}
) == ["paid"]
assert _get_tags_from_request_kwargs({"litellm_params": {"metadata": {"tags": ["paid"]}}}) == ["paid"]
assert _get_tags_from_request_kwargs({"litellm_params": {"metadata": None}}) == []
assert _get_tags_from_request_kwargs({"litellm_params": {}}) == []
@ -473,3 +421,601 @@ 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"