feat: serialize LangChain multimodal content to ingest content blocks

pyproject.toml

@@ -22,7 +22,7 @@ langchain-core = { version = ">=0.3.68", optional = true }
 langchain = { version = ">=0.0.0,<2.0.0", optional = true }
 openai = { version = "<1.96.0", optional = true }
 openai-agents = { version = "<0.2.1", optional = true }
-galileo-core = "^3.82.1"
+galileo-core = {git = "https://github.com/rungalileo/galileo-core.git", branch = "feature/sc-56110/-sdk-python-use-new-content-schema-in-langchain"}
 starlette = { version = ">=0.27.0", optional = true }
 backoff = "^2.2.1"
 crewai = { version = ">=0.152.0,<2.0.0", optional = true, python = ">=3.10,<3.14" }
@@ -38,7 +38,7 @@ pytest-xdist = "^3.7.0"
 pytest-socket = "^0.7"
 pytest-asyncio = "^1.0.0"
 requests-mock = "^1.11.0"
-galileo-core = { extras = ["testing"], version = "^3.82.1" }
+galileo-core = {git = "https://github.com/rungalileo/galileo-core.git", branch = "feature/sc-56110/-sdk-python-use-new-content-schema-in-langchain", extras = ["testing"]}
 
 pytest-env = "^1.1.5"
 langchain-core = ">=0.3.68"

src/galileo/constants/routes.py

@@ -23,3 +23,5 @@ class Routes(str, Enum):
 
     sessions = "/v2/projects/{project_id}/sessions"
     sessions_search = "/v2/projects/{project_id}/sessions/search"
+
+    ingest_traces = "/ingest/traces/{project_id}"

src/galileo/logger/logger.py

@@ -4,6 +4,7 @@
 import inspect
 import json
 import logging
+import os
 import time
 import uuid
 from datetime import datetime
@@ -30,7 +31,7 @@
     TracesIngestRequest,
     TraceUpdateRequest,
 )
-from galileo.traces import Traces
+from galileo.traces import IngestTraces, Traces
 from galileo.utils.decorators import (
     async_warn_catch_exception,
     nop_async,
@@ -153,6 +154,7 @@ class GalileoLogger(TracesLogger):
 
     _logger = logging.getLogger("galileo.logger")
     _traces_client: Optional["Traces"] = None
+    _ingest_client: Optional["IngestTraces"] = None
     _task_handler: ThreadPoolTaskHandler
     _trace_completion_submitted: bool
 
@@ -305,6 +307,11 @@ def __init__(
                 self._traces_client = Traces(project_id=self.project_id, log_stream_id=self.log_stream_id)
             elif self.experiment_id:
                 self._traces_client = Traces(project_id=self.project_id, experiment_id=self.experiment_id)
+
+            if os.environ.get("GALILEO_INGEST_URL"):
+                self._ingest_client = IngestTraces(
+                    project_id=self.project_id, log_stream_id=self.log_stream_id, experiment_id=self.experiment_id
+                )
         else:
             # ingestion_hook path: Traces client not created eagerly.
             # If the user later calls ingest_traces(), it will be created lazily.
@@ -474,6 +481,8 @@ def _ingest_trace_streaming(self, trace: Trace, is_complete: bool = False) -> No
         )
         @retry_on_transient_http_error
         async def ingest_traces_with_backoff(request: Any) -> None:
+            if self._ingest_client:
+                return await self._ingest_client.ingest_traces(request)
             return await self._traces_client.ingest_traces(request)
 
         self._task_handler.submit_task(
@@ -1837,6 +1846,8 @@ async def _flush_batch(self) -> list[Trace]:
                 await self._ingestion_hook(traces_ingest_request)
             else:
                 self._ingestion_hook(traces_ingest_request)
+        elif self._ingest_client:
+            await self._ingest_client.ingest_traces(traces_ingest_request)
         else:
             await self._traces_client.ingest_traces(traces_ingest_request)
 

src/galileo/traces.py

@@ -1,10 +1,14 @@
 import logging
+import os
 from typing import Any, Optional
 from uuid import UUID
 
+import httpx
+
 from galileo.config import GalileoPythonConfig
 from galileo.constants.routes import Routes
 from galileo.schema.trace import (
+    LoggingMethod,
     LogRecordsSearchRequest,
     SessionCreateRequest,
     SpansIngestRequest,
@@ -19,6 +23,8 @@
 
 _logger = logging.getLogger(__name__)
 
+INGEST_SERVICE_TIMEOUT_SECONDS = 120.0
+
 
 class Traces:
     """
@@ -159,3 +165,62 @@ async def get_span(self, span_id: str) -> dict[str, str]:
         return await self._make_async_request(
             RequestMethod.GET, endpoint=Routes.span.format(project_id=self.project_id, span_id=span_id)
         )
+
+
+class IngestTraces:
+    """Client for the orbit ingest service (``/ingest/traces/:project_id``).
+
+    The ingest service accepts multimodal content blocks natively and
+    runs on a separate URL from the main Galileo API.
+
+    The service URL is resolved from ``GALILEO_INGEST_URL`` env var.
+    If not set, it falls back to ``{api_url}/ingest/traces/{project_id}``.
+    """
+
+    def __init__(self, project_id: str, log_stream_id: Optional[str] = None, experiment_id: Optional[str] = None):
+        self.config = GalileoPythonConfig.get()
+        self.project_id = project_id
+        self.log_stream_id = log_stream_id
+        self.experiment_id = experiment_id
+
+        if self.log_stream_id is None and self.experiment_id is None:
+            raise ValueError("log_stream_id or experiment_id must be set")
+
+    def _get_ingest_base_url(self) -> str:
+        explicit = os.environ.get("GALILEO_INGEST_URL")
+        if explicit:
+            return explicit.rstrip("/")
+        return str(self.config.api_url or self.config.console_url).rstrip("/")
+
+    def _get_auth_headers(self) -> dict[str, str]:
+        headers: dict[str, str] = {"Content-Type": "application/json", "X-Galileo-SDK": get_sdk_header()}
+        if self.config.api_key:
+            headers["Galileo-API-Key"] = self.config.api_key.get_secret_value()
+        elif self.config.jwt_token:
+            headers["Authorization"] = f"Bearer {self.config.jwt_token.get_secret_value()}"
+        return headers
+
+    @async_warn_catch_exception(logger=_logger)
+    async def ingest_traces(self, traces_ingest_request: TracesIngestRequest) -> dict[str, Any]:
+        if self.experiment_id:
+            traces_ingest_request.experiment_id = UUID(self.experiment_id)
+        elif self.log_stream_id:
+            traces_ingest_request.log_stream_id = UUID(self.log_stream_id)
+
+        traces_ingest_request.logging_method = LoggingMethod.python_client
+
+        base_url = self._get_ingest_base_url()
+        url = f"{base_url}{Routes.ingest_traces.format(project_id=self.project_id)}"
+        json_body = traces_ingest_request.model_dump(mode="json")
+
+        _logger.info(
+            "Sending traces to ingest service",
+            extra={"url": url, "project_id": self.project_id, "num_traces": len(traces_ingest_request.traces)},
+        )
+
+        async with httpx.AsyncClient(
+            timeout=httpx.Timeout(INGEST_SERVICE_TIMEOUT_SECONDS, connect=10.0), verify=self.config.ssl_context
+        ) as client:
+            response = await client.post(url, json=json_body, headers=self._get_auth_headers())
+            response.raise_for_status()
+            return response.json()

src/galileo/utils/serialization.py

@@ -14,6 +14,7 @@
 from pydantic import BaseModel
 
 from galileo.utils.dependencies import is_langchain_available, is_langgraph_available, is_proto_plus_available
+from galileo_core.schemas.shared.content_blocks import DataContentBlock, IngestContentBlock, TextContentBlock
 
 _logger = logging.getLogger(__name__)
 
@@ -42,10 +43,79 @@ def _serialize_zoned_datetime(v: dt.datetime) -> str:
 
 
 def map_langchain_role(role: str) -> str:
-    role_map = {"ai": "assistant", "human": "user"}
+    # Non-chunk types like "system", "tool", "function", "chat" pass through unchanged.
+    # Chunk classes set type to the class name (e.g. "AIMessageChunk") so we map those too.
+    role_map = {
+        "ai": "assistant",
+        "AIMessageChunk": "assistant",
+        "human": "user",
+        "HumanMessageChunk": "user",
+        "SystemMessageChunk": "system",
+        "ToolMessageChunk": "tool",
+        "FunctionMessageChunk": "function",
+        "ChatMessageChunk": "chat",
+    }
     return role_map.get(role, role)
 
 
+# LangChain multimodal message format mapping.
+# See https://python.langchain.com/docs/concepts/multimodality/
+# LangChain uses {"type": "<modality>_url", "<modality>_url": {"url": "..."}} for media,
+# and {"type": "text", "text": "..."} for text segments.
+_LANGCHAIN_TYPE_TO_MODALITY = {
+    "image_url": "image",
+    "audio_url": "audio",
+    "video_url": "video",
+    "document_url": "document",
+    "input_image": "image",
+    "input_audio": "audio",
+}
+
+
+def _convert_langchain_content_block(block: dict) -> IngestContentBlock:
+    """Convert a single LangChain content block dict to a Galileo ingest content block.
+
+    LangChain multimodal format (https://python.langchain.com/docs/concepts/multimodality/):
+        {"type": "text", "text": "hello"}
+        {"type": "image_url", "image_url": {"url": "https://..."}}
+        {"type": "input_audio", "input_audio": {"data": "base64...", "format": "wav"}}
+
+    Returns either a TextContentBlock or DataContentBlock instance.
+    """
+    block_type = block.get("type", "")
+    if block_type == "text" or (not block_type and "text" in block):
+        return TextContentBlock(text=block.get("text", ""))
+    modality = _LANGCHAIN_TYPE_TO_MODALITY.get(block_type)
+    if modality:
+        nested = block.get(block_type, {})
+        if isinstance(nested, dict):
+            url = nested.get("url", "")
+            raw_data = nested.get("data", "")
+            fmt = nested.get("format", "")
+        else:
+            url = str(nested)
+            raw_data = ""
+            fmt = ""
+        kwargs: dict[str, Any] = {"type": modality}
+        if url:
+            if url.startswith("data:"):
+                kwargs["base64"] = url
+            else:
+                kwargs["url"] = url
+        elif raw_data:
+            mime = f"{modality}/{fmt}" if fmt else modality
+            kwargs["base64"] = f"data:{mime};base64,{raw_data}"
+        return DataContentBlock(**kwargs)
+    return TextContentBlock(text=str(block))
+
+
+def _normalize_multimodal_content(dumped: dict) -> None:
+    """If ``dumped["content"]`` is a list of dicts, convert each to an ingest content block in-place."""
+    content = dumped.get("content")
+    if isinstance(content, list) and content and isinstance(content[0], dict):
+        dumped["content"] = [_convert_langchain_content_block(b) for b in content]
+
+
 class EventSerializer(JSONEncoder):
     """Custom JSON encoder to assist in the serialization of a wide range of objects."""
 
@@ -96,6 +166,14 @@ def default(self, obj: Any) -> Any:
                     return self.default(obj.message)
                 if isinstance(obj, LLMResult):
                     return self.default(obj.generations[0])
+                # LangChain message type multimodal audit (all 12 types accounted for):
+                #   Branch 1 (AIMessageChunk, AIMessage): explicit — tool_calls + multimodal
+                #   Branch 2 (ToolMessage, ToolMessageChunk via inheritance): explicit — status + multimodal
+                #   Branch 3 (BaseMessage catch-all): HumanMessage, HumanMessageChunk,
+                #     SystemMessage, SystemMessageChunk, ChatMessage, ChatMessageChunk,
+                #     FunctionMessage (deprecated), FunctionMessageChunk (deprecated)
+                # _normalize_multimodal_content() is called in every branch so list[dict]
+                # content is converted to IngestContentBlock for all message types.
                 if isinstance(obj, (AIMessageChunk, AIMessage)):
                     # Map the `type` to `role`.
                     if hasattr(obj, "model_dump"):
@@ -105,12 +183,7 @@ def default(self, obj: Any) -> Any:
                     else:
                         # Fallback to using the dict method if model_dump is not available i.e pydantic v1
                         dumped = obj.dict(include={"content", "type", "additional_kwargs", "tool_calls"})
-                    content = dumped.get("content")
-                    if isinstance(content, list):
-                        # Responses API returns content as a list of dicts
-                        # Convert list content to string format for consistency
-                        if content and isinstance(content[0], dict):
-                            dumped["content"] = content[0].get("text", "")
+                    _normalize_multimodal_content(dumped)
                     dumped["role"] = map_langchain_role(dumped.pop("type"))
                     additional_kwargs = dumped.pop("additional_kwargs", {})
                     # Check both direct attribute and additional_kwargs for tool_calls
@@ -156,6 +229,7 @@ def default(self, obj: Any) -> Any:
                     else:
                         # Fallback to using the dict method if model_dump is not available i.e pydantic v1
                         dumped = obj.dict(include={"content", "type", "status", "tool_call_id"})
+                    _normalize_multimodal_content(dumped)
                     dumped["role"] = map_langchain_role(dumped.pop("type"))
                     return dumped
                 if isinstance(obj, BaseMessage):
@@ -165,6 +239,7 @@ def default(self, obj: Any) -> Any:
                     else:
                         # Fallback to using the dict method if model_dump is not available i.e pydantic v1
                         dumped = obj.dict(include={"content", "type"})
+                    _normalize_multimodal_content(dumped)
                     dumped["role"] = map_langchain_role(dumped.pop("type"))
                     return dumped
 
@@ -212,6 +287,9 @@ def default(self, obj: Any) -> Any:
                         return f"<{obj.__name__}>"
                 return f"<{obj.__name__}>"
 
+            if isinstance(obj, (TextContentBlock, DataContentBlock)):
+                return obj.model_dump(mode="json", exclude_none=True)
+
             if isinstance(obj, BaseModel):
                 if hasattr(obj, "model_dump"):
                     return self.default(

tests/test_langchain.py

@@ -230,6 +230,44 @@ def test_on_chat_model_start(self, callback: GalileoCallback) -> None:
         assert input_data[1]["role"] == "user"
         assert input_data[2]["role"] == "assistant"
 
+    def test_on_chat_model_start_multimodal(self, callback: GalileoCallback) -> None:
+        """Test that multimodal HumanMessage produces structured content blocks, not stringified JSON."""
+        parent_id = uuid.uuid4()
+        run_id = uuid.uuid4()
+
+        callback.on_chain_start(serialized={}, inputs={"query": "test"}, run_id=parent_id)
+
+        # Given: a HumanMessage with an image_url content block (LangChain multimodal format)
+        multimodal_message = HumanMessage(
+            content=[
+                {"type": "text", "text": "What is in this image?"},
+                {"type": "image_url", "image_url": {"url": "https://example.com/cat.png"}},
+            ]
+        )
+
+        # When: on_chat_model_start processes the multimodal message
+        callback.on_chat_model_start(
+            serialized={},
+            messages=[[multimodal_message]],
+            run_id=run_id,
+            parent_run_id=parent_id,
+            invocation_params={"model": "gpt-4o"},
+        )
+
+        # Then: the input content is a list of structured content blocks, not a flat string
+        node = callback._handler.get_node(run_id)
+        assert node is not None
+        input_data = node.span_params["input"]
+        assert isinstance(input_data, list)
+        assert len(input_data) == 1
+        content = input_data[0]["content"]
+        assert isinstance(content, list), "Multimodal content should be a list of blocks, not a string"
+        assert len(content) == 2
+        assert content[0]["type"] == "text"
+        assert content[0]["text"] == "What is in this image?"
+        assert content[1]["type"] == "image"
+        assert content[1]["url"] == "https://example.com/cat.png"
+
     def test_on_chat_model_start_end_with_tools(self, callback: GalileoCallback, galileo_logger: GalileoLogger) -> None:
         """Test chat model start and end callbacks with tools"""
         run_id = uuid.uuid4()
@@ -923,11 +961,11 @@ def test_ai_message_with_list_content(self, callback: GalileoCallback, galileo_l
         assert node is not None
         assert node.node_type == "chat"
 
-        # Check that content was properly converted from list to string
+        # Check that content was properly converted to structured content blocks
         input_data = node.span_params["input"]
         assert isinstance(input_data, list)
         assert len(input_data) == 1
-        assert input_data[0]["content"] == "This is a response from the Responses API"
+        assert input_data[0]["content"] == [{"type": "text", "text": "This is a response from the Responses API"}]
         assert input_data[0]["role"] == "assistant"
 
     def test_ai_message_with_reasoning(self, callback: GalileoCallback, galileo_logger: GalileoLogger) -> None:

tests/test_langchain_async.py

@@ -900,11 +900,11 @@ async def test_ai_message_with_list_content(
         assert node is not None
         assert node.node_type == "chat"
 
-        # Check that content was properly converted from list to string
+        # Check that content was properly converted to structured content blocks
         input_data = node.span_params["input"]
         assert isinstance(input_data, list)
         assert len(input_data) == 1
-        assert input_data[0]["content"] == "This is a response from the Responses API"
+        assert input_data[0]["content"] == [{"type": "text", "text": "This is a response from the Responses API"}]
         assert input_data[0]["role"] == "assistant"
 
     @mark.asyncio