fix(types): suppress Pydantic serialization warnings on ModelResponse choices

Pydantic v2's Union serializer for `List[Union[Choices, StreamingChoices]]`
tries both branches when serializing, emitting spurious
`PydanticSerializationUnexpectedValue` warnings (field count mismatch on
`Message` and type mismatch `Expected StreamingChoices but got Choices`).

Add a `WrapSerializer` on the `choices` field that serializes each item
individually via its own `model_dump()`, bypassing the Union dispatch
entirely while correctly propagating `exclude_none`, `exclude_unset`,
and `exclude_defaults` from the parent serialization context.
This commit is contained in:
Chesars 2026-02-19 21:43:19 -03:00
parent 2d39825868
commit 3564b8d83b
2 changed files with 100 additions and 4 deletions

View file

@ -22,8 +22,9 @@ from openai.types.moderation_create_response import Moderation as Moderation
from openai.types.moderation_create_response import (
ModerationCreateResponse as ModerationCreateResponse,
)
from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, model_validator
from typing_extensions import Required, TypedDict
from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, SerializationInfo, model_validator
from pydantic.functional_serializers import WrapSerializer
from typing_extensions import Annotated, Required, TypedDict
from litellm._uuid import uuid
from litellm.types.llms.base import (
@ -1640,6 +1641,34 @@ class StreamingChatCompletionChunk(OpenAIChatCompletionChunk):
super().__init__(**kwargs)
def _serialize_choices_list(
choices: list, handler, info: SerializationInfo
) -> list:
"""Serialize each choice individually to avoid Union serializer warnings.
Pydantic's Union serializer for ``List[Union[Choices, StreamingChoices]]``
may try the wrong branch first, emitting spurious
``PydanticSerializationUnexpectedValue`` warnings. By serializing each
item via its own ``model_dump()`` we bypass the Union dispatch entirely.
"""
kwargs: Dict[str, Any] = {}
if info.exclude_none:
kwargs["exclude_none"] = True
if info.exclude_unset:
kwargs["exclude_unset"] = True
if info.exclude_defaults:
kwargs["exclude_defaults"] = True
result = []
for choice in choices:
if hasattr(choice, "model_dump"):
result.append(choice.model_dump(**kwargs))
elif isinstance(choice, dict):
result.append(choice)
else:
result.append(choice)
return result
class ModelResponseBase(OpenAIObject):
id: str
"""A unique identifier for the completion."""
@ -1748,7 +1777,10 @@ class ModelResponseStream(ModelResponseBase):
class ModelResponse(ModelResponseBase):
choices: List[Union[Choices, StreamingChoices]]
choices: Annotated[
List[Union[Choices, StreamingChoices]],
WrapSerializer(_serialize_choices_list, return_type=list),
]
"""The list of completion choices the model generated for the input prompt."""
def __init__( # noqa: PLR0915

View file

@ -2,7 +2,7 @@ import warnings
import pytest
from litellm.types.utils import Choices, Message, ModelResponse
from litellm.types.utils import Choices, Delta, Message, ModelResponse, StreamingChoices
def test_modelresponse_normalizes_openai_base_models() -> None:
@ -59,3 +59,67 @@ def test_modelresponse_serialization_avoids_pydantic_warnings() -> None:
or "Pydantic serializer warnings" in str(w.message)
for w in captured
)
def test_modelresponse_model_dump_json_no_pydantic_warnings() -> None:
"""model_dump_json() bypasses the Python model_dump() override and uses
Pydantic's C-level serializer directly. The Union[Choices, StreamingChoices]
field previously triggered PydanticSerializationUnexpectedValue warnings via
this path."""
response = ModelResponse(
model="test-model",
choices=[
Choices(
finish_reason="stop",
index=0,
message=Message(content="hello", role="assistant"),
)
],
)
with warnings.catch_warnings(record=True) as captured:
warnings.simplefilter("always")
_ = response.model_dump_json()
_ = response.model_dump()
_ = response.model_dump(exclude_none=True)
pydantic_warnings = [
w
for w in captured
if "PydanticSerializationUnexpectedValue" in str(w.message)
or "Pydantic serializer warnings" in str(w.message)
]
assert pydantic_warnings == [], (
f"Unexpected Pydantic serialization warnings: {pydantic_warnings}"
)
def test_streaming_modelresponse_no_pydantic_warnings() -> None:
"""Streaming responses use StreamingChoices in the Union field and should
also serialize without warnings."""
response = ModelResponse(
model="test-model",
choices=[
StreamingChoices(
finish_reason="stop",
index=0,
delta=Delta(content="hello", role="assistant"),
)
],
stream=True,
)
with warnings.catch_warnings(record=True) as captured:
warnings.simplefilter("always")
_ = response.model_dump_json()
_ = response.model_dump()
pydantic_warnings = [
w
for w in captured
if "PydanticSerializationUnexpectedValue" in str(w.message)
or "Pydantic serializer warnings" in str(w.message)
]
assert pydantic_warnings == [], (
f"Unexpected Pydantic serialization warnings: {pydantic_warnings}"
)