feat(serialization): introduce dedicated serialization module (#118)

Separates serialization logic from API clients (OpenAI, GoogleGenAI) and
make it more customizable per model or provider.
- Introduced a unified serialization pipeline through the BaseSerializer
and its implementations: OpenAICompletionSerializer,
ModelProxyOpenAISerializer, and GenAiSerializer.
- Removed inline message serialization loops (like _get_raw_messages)
from LLM classes, delegating it directly to the corresponding serializer
instances.

Author: s-alexey

src/kaggle_benchmarks/actors/llms.py

@@ -85,11 +85,9 @@ def invoke(self, messages, system: str = ""):
 
 """
 
-import base64
 import dataclasses
 import enum
 import json
-import mimetypes
 import typing
 from typing import TYPE_CHECKING, Any, Iterator, TypeVar
 
@@ -100,6 +98,8 @@ def invoke(self, messages, system: str = ""):
 from kaggle_benchmarks import actors, chats, messages, prompting, utils
 from kaggle_benchmarks._config import config
 from kaggle_benchmarks.content_types import images, videos
+from kaggle_benchmarks.serializers import genai as genai_serializer
+from kaggle_benchmarks.serializers import openai as openai_serializer
 
 if TYPE_CHECKING:
     from kaggle_benchmarks import llm_messages
@@ -278,6 +278,9 @@ def __init__(self, client: openai.OpenAI, model: str, **kwargs):
         super().__init__(**kwargs)
         self.model = model
         self.client = client
+        self.serializer = openai_serializer.ModelProxyOpenAISerializer(
+            roles_mapping={"tool": "system"}
+        )
 
     def _get_usage_meta(
         self, usage: openai.types.CompletionUsage | None
@@ -298,9 +301,12 @@ def _should_remove_seed(self) -> bool:
     def invoke(
         self, messages: list[messages.Message], system: str | None, **kwargs
     ) -> LLMResponse | Iterator[LLMResponse]:
-        raw_messages = self._get_raw_messages(messages)
         if system:
-            raw_messages = [{"role": "system", "content": system}] + raw_messages
+            from kaggle_benchmarks.messages import Message
+
+            messages = [Message(sender=actors.system, content=system)] + messages
+
+        raw_messages = list(self.serializer.dump_messages(messages))
 
         if self._should_remove_seed():
             # TODO(b/430112500): Remove once model proxy supports it for AIS backends.
@@ -309,17 +315,6 @@ def invoke(
 
         return self._call_api(raw_messages, **kwargs)
 
-    def _get_raw_messages(self, messages: list[messages.Message]):
-        return [
-            {
-                "role": message.sender.role
-                if message.sender.role != "tool"
-                else "system",  # TODO: Remove this renaming once ModelProxy supports tools
-                "content": message.payload,
-            }
-            for message in messages
-        ]
-
     def _get_stream_response(
         self, response_stream: openai.Stream
     ) -> Iterator[LLMResponse]:
@@ -390,6 +385,9 @@ def __init__(self, client: genai.Client, model: str, **kwargs):
         super().__init__(**kwargs)
         self.model = model
         self.client = client
+        self.serializer = genai_serializer.GenAISerializer(
+            roles_mapping={"assistant": "model", "system": "user", "tool": "user"}
+        )
 
     def _get_usage_meta(self, usage: types.UsageMetadata | None) -> dict[str, Any]:
         if usage is None:
@@ -400,60 +398,6 @@ def _get_usage_meta(self, usage: types.UsageMetadata | None) -> dict[str, Any]:
             **_extract_extra_usage_metadata(usage),
         }
 
-    def _get_raw_messages(self, messages: list[messages.Message]):
-        """Converts benchmark messages to Google GenAI's Content format."""
-        raw_messages = []
-        for message in messages:
-            role = "model" if message.sender.role == "assistant" else "user"
-            content = message.content
-            payload = message.payload
-
-            parts = []
-
-            # Video URLs are passed through directly for the model provider to resolve.
-            if isinstance(content, videos.VideoContent):
-                parts.append(
-                    types.Part.from_uri(
-                        file_uri=content.url, mime_type=content.mime_type
-                    )
-                )
-
-            elif isinstance(payload, str):
-                parts.append(types.Part(text=payload))
-
-            # Note: The Gemini API is smart enough to process image data URLs even when they are passed as part of a plain text string.
-            elif isinstance(payload, list) and payload and isinstance(payload[0], dict):
-                for item in payload:
-                    if item.get("type") == "image_url":
-                        url = item["image_url"]["url"]
-
-                        image_bytes = None
-                        mime_type = "image/jpeg"
-                        if url.startswith("data:"):
-                            # Handle base64 data URLs
-                            header, b64_string = url.split(",", 1)
-                            mime_type = header.split(";")[0].split(":")[1]
-                            image_bytes = base64.b64decode(b64_string)
-                        else:
-                            # Handle remote http/https URLs
-                            b64_string = images.image_url_to_base64(url)
-                            image_bytes = base64.b64decode(b64_string)
-                            mime_type = mimetypes.guess_type(url)[0] or "image/jpeg"
-
-                        if image_bytes:
-                            parts.append(
-                                types.Part.from_bytes(
-                                    data=image_bytes, mime_type=mime_type
-                                )
-                            )
-            else:
-                # Fallback for any other unexpected payload types
-                parts.append(types.Part(text=str(payload)))
-
-            raw_messages.append(types.Content(role=role, parts=parts))
-
-        return raw_messages
-
     def _get_stream_response(
         self, response_stream: Iterator[types.GenerateContentResponse]
     ) -> Iterator[LLMResponse]:
@@ -467,7 +411,7 @@ def _get_stream_response(
     def invoke(
         self, messages: list[messages.Message], system: str | None, **kwargs
     ) -> LLMResponse | Iterator[LLMResponse]:
-        raw_messages = self._get_raw_messages(messages)
+        raw_messages = list(self.serializer.dump_messages(messages))
 
         config_params = {}
         if system:

src/kaggle_benchmarks/serializers/__init__.py

@@ -0,0 +1,14 @@
+# Copyright 2026 Kaggle Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+

src/kaggle_benchmarks/serializers/base.py

@@ -0,0 +1,120 @@
+# Copyright 2026 Kaggle Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import copy
+import dataclasses
+import itertools
+import json
+
+import pydantic
+
+from kaggle_benchmarks import actors, chats, llm_messages, tools
+from kaggle_benchmarks import messages as msg
+from kaggle_benchmarks.content_types import images, videos
+
+
+class UnsupportedMessageFormat(ValueError):
+    pass
+
+
+def _copy_replace(message, **new_fields):
+    new = copy.copy(message)
+    for k, v in new_fields.items():
+        setattr(new, k, v)
+    return new
+
+
+class BaseSerializer:
+    """Base class for all message serializers.
+
+    Provides the core logic to map generic benchmark messages to provider-specific
+    formats. Subclasses must implement the specific `dump_*` methods.
+    """
+
+    def __init__(self, roles_mapping: dict[str, str] | None = None):
+        self.roles_mapping = roles_mapping or {}
+
+    def get_role(self, sender: actors.Actor):
+        """Resolves the provider-specific role for a given sender using roles_mapping."""
+        return self.roles_mapping.get(sender.role, sender.role)
+
+    def dump_chat(self, chat: chats.Chat):
+        """Serializes an entire chat history into a provider-specific format."""
+        return self.dump_messages(chat.messages)
+
+    def dump_messages(self, messages: list[msg.Message]):
+        """Serializes a list of messages into a provider-specific format."""
+        return itertools.chain(*(self.dump_message(message) for message in messages))
+
+    def dump_message(self, message: msg.Message):
+        """Dynamically dispatches serialization based on the message content type."""
+        if isinstance(message, llm_messages.LLMMessage):
+            try:
+                yield from self.dump_llm_message(message)
+                return
+            except NotImplementedError:
+                # Fallback if the subclass doesn't support explicit LLM messages
+                pass
+
+        content = message.content
+        if isinstance(content, str):
+            yield from self.dump_text_message(message)
+        elif isinstance(content, images.ImageContent):
+            yield from self.dump_image(message)
+        elif isinstance(content, videos.VideoContent):
+            yield from self.dump_video(message)
+        elif isinstance(content, dict):
+            yield from self.dump_json_message(message)
+        elif isinstance(content, tools.ToolInvocationResult):
+            yield from self.dump_tool_invocation(message)
+        elif isinstance(content, pydantic.BaseModel):
+            yield from self.dump_json_message(
+                _copy_replace(message, content=message.content.model_dump())
+            )
+        elif dataclasses.is_dataclass(content) and not isinstance(content, type):
+            yield from self.dump_json_message(
+                _copy_replace(message, content=dataclasses.asdict(content))
+            )
+        else:
+            yield from self._dump_message(message)
+
+    def _dump_message(self, message: msg.Message):
+        """Fallback method for unsupported message types. Override in subclass to handle."""
+        raise NotImplementedError(
+            f"Unsupported message format for: {type(message.content)}"
+        )
+
+    def dump_tool_invocation(self, message: msg.Message[tools.ToolInvocationResult]):
+        raise NotImplementedError()
+
+    def dump_llm_message(self, message: llm_messages.LLMMessage):
+        """Serializes an LLM message containing tools and complex outputs."""
+        raise NotImplementedError()
+
+    def dump_text_message(self, message: msg.Message[str]):
+        """Serializes a standard text message."""
+        raise NotImplementedError()
+
+    def dump_json_message(self, message: msg.Message[dict]):
+        """Serializes a JSON dictionary message by stringifying it as text by default."""
+        yield from self.dump_text_message(
+            _copy_replace(message, content=json.dumps(message.content))
+        )
+
+    def dump_image(self, message: msg.Message[images.ImageContent]):
+        """Serializes an image message."""
+        raise NotImplementedError()
+
+    def dump_video(self, message: msg.Message[videos.VideoContent]):
+        """Serializes a video message."""
+        raise NotImplementedError()