Fix add_model_file_id_mappings when router returns single deployment dict

When model_info.id equals model_name (common for batch models), the router
resolves via has_model_id and returns one deployment dict instead of a list.
The dict branch incorrectly iterated deployment keys (model_name,
litellm_params, model_info), producing non-string values that broke
LiteLLM_ManagedFileTable validation on managed file upload.

Normalize list vs dict by wrapping single deployments and extracting
model_info.id for each response pair.

Add regression tests including the batch model id == model_name case.

Made-with: Cursor
This commit is contained in:
shivam
2026-04-30 18:04:47 -07:00
parent 326bcd6cec
commit e72eac9176
2 changed files with 127 additions and 10 deletions
+17 -10
View File
@@ -23,21 +23,28 @@ def add_model_file_id_mappings(
healthy_deployments: Union[List[Dict], Dict], responses: List["OpenAIFileObject"]
) -> dict:
"""
Create a mapping of model name to file id
Create a mapping of model id to file id
{
"model_id": "file_id",
"model_id": "file_id",
}
`healthy_deployments` may be either a list of deployment dicts (multiple
matched deployments) or a single deployment dict (when the router resolved
a specific deployment, e.g. because the requested model matched a
`model_info.id`). Both shapes must be handled by extracting
`model_info.id` from each deployment.
"""
model_file_id_mapping = {}
if isinstance(healthy_deployments, list):
for deployment, response in zip(healthy_deployments, responses):
model_file_id_mapping[deployment.get("model_info", {}).get("id")] = (
response.id
)
elif isinstance(healthy_deployments, dict):
for model_id, file_id in healthy_deployments.items():
model_file_id_mapping[model_id] = file_id
model_file_id_mapping: Dict[str, str] = {}
deployments_list: List[Dict] = (
healthy_deployments
if isinstance(healthy_deployments, list)
else [healthy_deployments]
)
for deployment, response in zip(deployments_list, responses):
model_id = deployment.get("model_info", {}).get("id")
if model_id is not None:
model_file_id_mapping[model_id] = response.id
return model_file_id_mapping
@@ -6,6 +6,7 @@ import pytest
from litellm import Router
from litellm.router_utils.common_utils import (
_deployment_supports_web_search,
add_model_file_id_mappings,
filter_team_based_models,
filter_web_search_deployments,
)
@@ -362,3 +363,112 @@ def test_invalidate_model_group_info_cache():
# Invalidate and verify cache is cleared
router._invalidate_model_group_info_cache()
assert router._cached_get_model_group_info.cache_info().currsize == 0
class TestAddModelFileIdMappings:
"""Test cases for add_model_file_id_mappings.
The router may pass either a list of deployment dicts (multiple matched
deployments) or a single deployment dict (when a specific deployment was
resolved, e.g. because the requested model matched a `model_info.id`).
Both shapes must produce a `{model_id: file_id}` mapping by extracting
`model_info.id` from each deployment.
"""
@staticmethod
def _make_response(file_id: str):
response = Mock()
response.id = file_id
return response
def test_should_map_each_deployment_id_when_given_list(self):
deployments = [
{
"model_name": "gpt-4",
"litellm_params": {"model": "gpt-4"},
"model_info": {"id": "deployment-1"},
},
{
"model_name": "gpt-4",
"litellm_params": {"model": "gpt-4"},
"model_info": {"id": "deployment-2"},
},
]
responses = [self._make_response("file-1"), self._make_response("file-2")]
result = add_model_file_id_mappings(deployments, responses)
assert result == {"deployment-1": "file-1", "deployment-2": "file-2"}
def test_should_extract_model_info_id_when_given_single_deployment_dict(self):
"""Regression test: when `_common_checks_available_deployment` resolves
a specific deployment (returned as a dict, not a list), the function
must still extract `model_info.id` rather than iterate over the
deployment's own keys (`model_name`, `litellm_params`, `model_info`).
"""
deployment = {
"model_name": "gpt-4",
"litellm_params": {"model": "gpt-4", "api_key": "sk-test"},
"model_info": {"id": "deployment-1", "mode": "chat"},
}
responses = [self._make_response("file-1")]
result = add_model_file_id_mappings(deployment, responses)
assert result == {"deployment-1": "file-1"}
assert all(isinstance(v, str) for v in result.values())
def test_should_handle_batch_model_when_id_matches_model_name(self):
"""Regression test for the batch-model case: when `model_info.id` is
intentionally set equal to `model_name`, the router resolves a single
deployment via `has_model_id` and returns it as a dict. The mapping
must contain only `{id: file_id}` with string values so the resulting
`LiteLLM_ManagedFileTable` Pydantic validation passes.
"""
deployment = {
"model_name": "openai/openai/gpt-5.5-batch",
"litellm_params": {
"model": "openai/gpt-5.5",
"api_key": "sk-test",
"tpm": 40000000,
"rpm": 15000,
},
"model_info": {
"id": "openai/openai/gpt-5.5-batch",
"mode": "batch",
"base_model": "gpt-5.5",
"access_groups": ["default-models"],
},
}
responses = [self._make_response("file-batch-1")]
result = add_model_file_id_mappings(deployment, responses)
# Bug case would have produced keys ["model_name", "litellm_params",
# "model_info"] with non-string values.
assert result == {"openai/openai/gpt-5.5-batch": "file-batch-1"}
assert "litellm_params" not in result
assert "model_info" not in result
def test_should_skip_deployment_when_model_info_id_missing(self):
deployments = [
{
"model_name": "gpt-4",
"litellm_params": {"model": "gpt-4"},
"model_info": {},
},
{
"model_name": "gpt-4",
"litellm_params": {"model": "gpt-4"},
"model_info": {"id": "deployment-2"},
},
]
responses = [self._make_response("file-1"), self._make_response("file-2")]
result = add_model_file_id_mappings(deployments, responses)
assert result == {"deployment-2": "file-2"}
def test_should_return_empty_mapping_when_given_empty_list(self):
result = add_model_file_id_mappings([], [])
assert result == {}