refactor(proxy): align model deployment validation with router admission rules

This commit is contained in:
aoright 2026-09-28 14:56:27 +08:00
parent 58ecf91d4b
commit 021b68fb8b
2 changed files with 74 additions and 22 deletions

View file

@ -347,9 +347,9 @@ def _validate_routable_model_deployment(
"""Validate that incoming litellm_params specifies a routable LLM provider before DB insertion.
When adding or patching a model deployment, the router must be able to resolve its provider.
Unroutable models (such as typesafe/* without a chat completion provider) get dropped by the
router upon reload, causing a 500 degraded serving error and leaving dead zombie rows in
LiteLLM_ProxyModelTable. Pre-validating the provider returns a clean 400 Bad Request instead.
Unroutable models get dropped by the router upon reload, causing a 500 degraded serving error
and leaving dead zombie rows in LiteLLM_ProxyModelTable. Pre-validating the provider returns
a clean 400 Bad Request instead.
"""
if incoming_params is None:
return None
@ -361,39 +361,45 @@ def _validate_routable_model_deployment(
if not model or not isinstance(model, str):
return None
# Auto-router and strategy-router pseudo-models are validated separately
if model.startswith("auto_router/") or model.startswith("strategy_router/"):
# Auto-router pseudo-models are validated separately
if model.startswith("auto_router/"):
return None
# Prompt management models resolve provider dynamically at runtime
if "/" in model:
split_model = model.split("/")[0]
if split_model in litellm._known_custom_logger_compatible_callbacks:
return None
custom_llm_provider = getattr(incoming_params, "custom_llm_provider", None) or (
getattr(existing_params, "custom_llm_provider", None) if existing_params else None
)
api_base = getattr(incoming_params, "api_base", None) or (
getattr(existing_params, "api_base", None) if existing_params else None
)
api_key = getattr(incoming_params, "api_key", None) or (
getattr(existing_params, "api_key", None) if existing_params else None
)
try:
from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
from litellm.litellm_core_utils.get_llm_provider_logic import (
get_llm_provider,
is_registered_custom_provider,
)
from litellm.llms.openai_like.json_loader import JSONProviderRegistry
get_llm_provider(
_model, resolved_provider, _dynamic_api_key, _api_base = get_llm_provider(
model=model,
custom_llm_provider=custom_llm_provider,
api_base=api_base,
api_key=api_key,
litellm_params=incoming_params,
)
except Exception as e:
err_msg = str(e)
if "typesafe" in model.lower() or (custom_llm_provider and "typesafe" in custom_llm_provider.lower()):
if (
resolved_provider not in litellm.provider_list
and not JSONProviderRegistry.exists(resolved_provider)
and not is_registered_custom_provider(resolved_provider)
):
return (
f"Model '{model}' cannot be registered as a chat completion deployment: 'typesafe' is not a routable chat provider. "
"TypeSafe Jev is supported via pass-through endpoints (/typesafe/*) or as an auto-router complexity classifier "
f"(router_settings.classifier_type: 'jev'). Provider error: {err_msg}"
f"Model '{model}' is not a supported or routable provider: unsupported provider '{resolved_provider}'"
)
return f"Model '{model}' is not a routable provider: {err_msg}"
except Exception as e:
return f"Model '{model}' is not a routable provider: {e}"
return None

View file

@ -5840,7 +5840,7 @@ class TestStrategyRouterWriteValidation:
is None
)
def test_typesafe_model_rejected_with_helpful_guidance(self):
def test_unroutable_model_rejected(self):
from litellm.proxy.management_endpoints.model_management_endpoints import (
_validate_routable_model_deployment,
)
@ -5852,8 +5852,7 @@ class TestStrategyRouterWriteValidation:
)
violation = _validate_routable_model_deployment(params)
assert violation is not None
assert "typesafe" in violation.lower()
assert "not a routable chat provider" in violation or "pass-through" in violation
assert "not a supported or routable provider" in violation or "not a routable provider" in violation
def test_routable_model_accepted(self):
from litellm.proxy.management_endpoints.model_management_endpoints import (
@ -5879,6 +5878,53 @@ class TestStrategyRouterWriteValidation:
)
assert _validate_routable_model_deployment(params) is None
def test_prompt_management_model_skipped(self):
from litellm.proxy.management_endpoints.model_management_endpoints import (
_validate_routable_model_deployment,
)
from litellm.types.router import LiteLLM_Params
params = LiteLLM_Params(
model="dotprompt/hello-world",
api_key="sk-test",
)
assert _validate_routable_model_deployment(params) is None
@pytest.mark.asyncio
async def test_add_new_model_rejects_unroutable_model_before_db_write(self):
from litellm.proxy._types import LitellmUserRoles, ProxyException, UserAPIKeyAuth
from litellm.proxy.management_endpoints.model_management_endpoints import (
add_new_model,
)
from litellm.types.router import Deployment, LiteLLM_Params
admin = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
mock_prisma = MagicMock()
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_prisma),
patch("litellm.proxy.proxy_server.store_model_in_db", True),
patch("litellm.proxy.proxy_server.premium_user", True),
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call",
new=AsyncMock(return_value=None),
),
):
with pytest.raises(ProxyException) as exc_info:
await add_new_model(
model_params=Deployment(
model_name="test-unroutable-model",
litellm_params=LiteLLM_Params(model="typesafe/jev-latest"),
model_info={"id": "test-unroutable-id"},
),
user_api_key_dict=admin,
)
assert exc_info.value.code == 400
assert "not a supported or routable provider" in str(
exc_info.value.message
) or "not a routable provider" in str(exc_info.value.message)
mock_prisma.db.litellm_proxymodeltable.create.assert_not_called()
@staticmethod
def _live_router_holding_one_capability(limit: int | None, config: Mapping[str, object]) -> Router:
return Router(