feat: serialize LangChain multimodal content to ingest content blocks

[pratduv]

User description

Summary

• Convert LangChain's multimodal message format (image_url, audio_url, video_url, etc.) into Galileo's IngestContentBlock schema (TextContentBlock, DataContentBlock) instead of flattening list content to a plain string
• Add IngestTraces client for the orbit ingest service (opt-in via GALILEO_INGEST_URL env var)
• Add debug logging at serialization and ingest payload boundaries

Changes

Serialization (src/galileo/utils/serialization.py)

• New _convert_langchain_content_block() maps LangChain's {"type": "image_url", "image_url": {"url": "..."}} format to DataContentBlock(type="image", url="..."), with base64 data URI detection
• EventSerializer now converts list content on both AIMessage and BaseMessage subclasses to content block arrays instead of extracting only the first text element

Ingest client (src/galileo/traces.py)

• New IngestTraces class sends traces directly to the orbit ingest service via httpx.AsyncClient
• _log_ingest_content_blocks() helper logs content block types at DEBUG level before HTTP POST
• New Routes.ingest_traces constant

Logger integration (src/galileo/logger/logger.py)

• GalileoLogger creates an IngestTraces client when GALILEO_INGEST_URL is set, preferring it over the API-proxied path

Handler (src/galileo/handlers/langchain/handler.py)

• Debug log in on_chat_model_start showing multimodal message count

Test plan

• TestConvertLangchainContentBlock -- unit tests for text, image URL, base64, audio, unknown fallback
• TestMultimodalContentSerialization -- round-trip serialization of AIMessage/HumanMessage with multimodal content
• test_on_chat_model_start_multimodal -- integration test verifying LangChain callback produces structured content blocks
• Updated existing Responses API test to expect content block arrays instead of flattened string

Dependencies

Requires galileo-core#feature/sc-56110 to be merged and released first (adds IngestContentBlock, TextContentBlock, DataContentBlock to galileo_core.schemas.shared.content_blocks).

Made with Cursor

---

Generated description

Below is a concise technical summary of the changes proposed in this PR:
Convert LangChain message serialization in EventSerializer to emit TextContentBlock/DataContentBlock arrays so multimodal content_blocks map directly to the ingest schema while preserving string-only legacy flows and surfacing debug logs for multimodal callbacks. Introduce the IngestTraces client along with the GALILEO_INGEST_URL wiring so GalileoLogger can post structured traces directly to the ingest service when available.

Topic	Details
Ingest Client	Add the IngestTraces client, routes, and logger wiring so structured traces can be sent to the ingest service via GALILEO_INGEST_URL, including dependency updates. Modified files (5) poetry.lock pyproject.toml src/galileo/constants/routes.py src/galileo/logger/logger.py src/galileo/traces.py Latest Contributors(2) User Commit Date ci@rungalileo.io chore-release-v1.49.0 March 06, 2026 Focadecombate feat-support-crewai-1.... February 25, 2026
Multimodal Serialization	Convert LangChain content blocks into IngestContentBlock arrays through EventSerializer, ensure handlers report multimodal counts, and cover the flow with new serialization tests and assertions in both sync/async callbacks. Modified files (4) src/galileo/utils/serialization.py tests/test_langchain.py tests/test_langchain_async.py tests/utils/test_serialization.py Latest Contributors(2) User Commit Date Focadecombate feat-add-proto-plus-me... February 25, 2026 mikebranc fix-Tool-calls-not-app... November 18, 2025

This pull request is reviewed by Baz. Review like a pro on (Baz).

[baz-reviewer]

IngestTraces.ingest_traces now reimplements the experiment/log_stream wiring (and by extension the model_dump/_log_ingest_content_blocks flow) that already exists in Traces.ingest_traces; keeping two copies means every future change to request preparation or logging must be applied twice. Can we extract a shared helper that sets experiment_id/log_stream_id, dumps the request, logs the content blocks, and optionally sets logging_method, then call it from both ingestion clients so we only maintain that plumbing in one place?

_{Finding type: Code Dedup and Conventions}

---

• Apply fix with Baz

[pratduv]

Will clean this up in a follow-up, keeping it as-is for now.

[baz-reviewer]

Understood, we can leave it as-is for now and plan the cleanup for a follow-up; I’ll remember this once the PR is merged.

[pratduv]

Only created when this URL is set

[pratduv]

"""
Multimodal content ingestion via LangChain + Galileo callback.

Invokes the model with REAL calls for every LangChain message type that can
carry multimodal content, then flushes to the ingest service. A successful
flush (no errors) proves the Go ingest service can parse every content-block
variant.

Tests:
  1. Plain text messages
  2. HumanMessage with image_url
  3. HumanMessage with base64 image
  4. Streaming with multimodal input
  5. ToolMessage with multimodal content in conversation history
  6. AIMessage with list content in conversation history

Expects these env vars in ../.env (or the logstreams/.env):
    GALILEO_CONSOLE_URL, GALILEO_API_KEY, GALILEO_PROJECT, GALILEO_LOG_STREAM, OPENAI_API_KEY

Optional:
    GALILEO_INGEST_URL  - if set, uses the orbit ingest service instead of v2 API

Usage:
    cd nbs/examples/py_files/logstreams
    python langchain_multimodal_example.py
"""

import logging
from pathlib import Path

from dotenv import load_dotenv

load_dotenv(Path(__file__).resolve().parents[3] / ".env")

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s [%(levelname)s] %(message)s")

import contextlib  # noqa: E402

from langchain_core.messages import (  # noqa: E402
    AIMessage,
    HumanMessage,
    SystemMessage,
    ToolMessage,
)
from langchain_openai import ChatOpenAI  # noqa: E402

from galileo import galileo_context  # noqa: E402
from galileo.handlers.langchain import GalileoCallback  # noqa: E402

SAMPLE_IMAGE_URL = (
    "https://upload.wikimedia.org/wikipedia/commons/thumb/4/47/"
    "PNG_transparency_demonstration_1.png/280px-PNG_transparency_demonstration_1.png"
)

RED_SQUARE_B64 = (
    "data:image/png;base64,"
    "iVBORw0KGgoAAAANSUhEUgAAAGQAAABkCAIAAAD/gAIDAAABFUlEQVR4nO3OUQkAIABEsetfWiv4"
    "Nx4IC7Cd7XvkByF+EOIHIX4Q4gchfhDiByF+EOIHIX4Q4gchfhDiByF+EOIHIX4Q4gchfhDiByF+"
    "EOIHIX4Q4gchfhDiByF+EOIHIX4Q4gchfhDiByF+EOIHIX4Q4gchfhDiByF+EOIHIX4Q4gchfhDi"
    "ByF+EOIHIX4Q4gchfhDiByF+EOIHIX4Q4gchfhDiByF+EOIHIX4Q4gchfhDiByF+EOIHIX4Q4gch"
    "fhDiByF+EOIHIX4Q4gchfhDiByF+EOIHIX4Q4gchfhDiByF+EOIHIX4Q4gchfhDiByF+EOIHIX4Q"
    "4gchfhDiByF+EOIHIX4Q4gchfhDiByF+EOIHIReeLesrH9s1agAAAABJRU5ErkJggg=="
)


def run_llm_tests() -> None:
    """Run live LLM calls through the Galileo callback, covering all message types."""
    galileo_context.init()

    callback = GalileoCallback()
    model = ChatOpenAI(model="gpt-4o", temperature=0.0)

    # --- Test 1: Plain text ---
    print("\n--- Test 1: Text-only ---")  # noqa: T201
    resp = model.invoke(
        [SystemMessage(content="You are a helpful assistant. Be concise."), HumanMessage(content="What is 2 + 2?")],
        config={"callbacks": [callback]},
    )
    print(f"Response: {resp.content}")  # noqa: T201

    # --- Test 2: HumanMessage with image URL ---
    print("\n--- Test 2: HumanMessage + image URL ---")  # noqa: T201
    resp = model.invoke(
        [
            SystemMessage(content="You are a helpful assistant that can analyze images. Be concise."),
            HumanMessage(
                content=[
                    {"type": "text", "text": "What do you see in this image? Describe it briefly."},
                    {"type": "image_url", "image_url": {"url": SAMPLE_IMAGE_URL}},
                ]
            ),
        ],
        config={"callbacks": [callback]},
    )
    print(f"Response: {resp.content}")  # noqa: T201

    # --- Test 3: HumanMessage with base64 image ---
    print("\n--- Test 3: HumanMessage + base64 image ---")  # noqa: T201
    with contextlib.suppress(Exception):
        resp = model.invoke(
            [
                SystemMessage(content="You are a helpful assistant. Be concise."),
                HumanMessage(
                    content=[
                        {"type": "text", "text": "What color is this image?"},
                        {"type": "image_url", "image_url": {"url": RED_SQUARE_B64}},
                    ]
                ),
            ],
            config={"callbacks": [callback]},
        )
        print(f"Response: {resp.content}")  # noqa: T201

    # --- Test 4: Streaming with multimodal input ---
    print("\n--- Test 4: Streaming multimodal ---")  # noqa: T201
    chunks = []
    for chunk in model.stream(
        [
            SystemMessage(content="You are a helpful assistant that can analyze images. Be concise."),
            HumanMessage(
                content=[
                    {"type": "text", "text": "Describe this image in one sentence."},
                    {"type": "image_url", "image_url": {"url": SAMPLE_IMAGE_URL}},
                ]
            ),
        ],
        config={"callbacks": [callback]},
    ):
        chunks.append(chunk)
    print(f"Streaming: {len(chunks)} chunks received")  # noqa: T201

    # --- Test 5: ToolMessage with multimodal content in conversation history ---
    # LangChain passes the entire conversation via on_chat_model_start, which
    # serializes every message through EventSerializer. Placing a ToolMessage
    # with list[dict] content in history tests that code path end-to-end.
    print("\n--- Test 5: ToolMessage (multimodal) in history ---")  # noqa: T201
    resp = model.invoke(
        [
            SystemMessage(content="You are a helpful assistant. Be concise."),
            HumanMessage(content="Generate a chart of our sales data."),
            AIMessage(
                content="",
                tool_calls=[{"id": "call_chart", "name": "generate_chart", "args": {"data": "sales"}}],
            ),
            ToolMessage(
                content=[
                    {"type": "text", "text": "Chart generated successfully:"},
                    {"type": "image_url", "image_url": {"url": SAMPLE_IMAGE_URL}},
                ],
                tool_call_id="call_chart",
            ),
            HumanMessage(content="Summarize what you see in the chart you generated."),
        ],
        config={"callbacks": [callback]},
    )
    print(f"Response: {resp.content}")  # noqa: T201

    # --- Test 6: AIMessage with list content in conversation history ---
    # Tests serialization of AIMessage where .content is a list of dicts
    # (e.g., from a model that returns structured multimodal output).
    print("\n--- Test 6: AIMessage (list content) in history ---")  # noqa: T201
    resp = model.invoke(
        [
            SystemMessage(content="You are a helpful assistant. Be concise."),
            HumanMessage(
                content=[
                    {"type": "text", "text": "What is in this image?"},
                    {"type": "image_url", "image_url": {"url": SAMPLE_IMAGE_URL}},
                ]
            ),
            AIMessage(
                content=[
                    {"type": "text", "text": "The image shows overlapping colored dice on a transparent background."},
                ]
            ),
            HumanMessage(content="Are the dice all the same color?"),
        ],
        config={"callbacks": [callback]},
    )
    print(f"Response: {resp.content}")  # noqa: T201

    # --- Flush to Galileo ---
    print("\n--- Flushing to Galileo ---")  # noqa: T201
    galileo_context.flush()
    print("Flush complete.")  # noqa: T201


def main() -> None:
    run_llm_tests()


if __name__ == "__main__":
    main()

[baz-reviewer]

The same metadata normalization block (if metadata: metadata = {k: GalileoLogger._convert_metadata_value(v) ...}) appears in add_single_llm_span_trace, add_llm_span, add_tool_span, add_protect_span, add_agent_span, add_workflow_span, and add_retriever_span; keeping 7 copies in sync if the MetadataValue handling changes is error prone. Can we extract a helper like _normalize_metadata(metadata) and call it before building each kwargs dict so all spans share the logic?

_{Finding type: Code Dedup and Conventions | Severity: 🟢 Low}

---

Want Baz to fix this for you? Activate Fixer

[pratduv]

Pre-existing pattern, out of scope for this PR.

[baz-reviewer]

Understood, I’ll treat this as an existing pattern and out of scope for this PR, and I’ll remember this for future reviews once the PR is merged.

[baz-reviewer]

Message.content is being converted to structured content blocks (TextContentBlock / DataContentBlock) when LangChain multimodal messages are encountered. The new functions _convert_langchain_content_block and _normalize_multimodal_content convert list-of-dict LangChain multimodal content into IngestContentBlock instances (TextContentBlock/DataContentBlock). This causes the runtime payloads to include arrays/objects for message.content rather than the legacy string. The OpenAPI schema (openapi.yaml) still declares galileo_core__schemas__logging__llm__Message.content as a string, so traces with multimodal content will no longer conform to the public ingest contract and may be rejected. The schema and downstream validation should be updated to accept content block arrays (or the serialization should continue sending a string representation). Also review the EventSerializer branches that call _normalize_multimodal_content (further down in this file) as related call sites.

_{Finding type: Breaking Changes | Severity: 🔴 High}

---

Want Baz to fix this for you? Activate Fixer

Other fix methods

Prompt for AI Agents:

In src/galileo/utils/serialization.py around lines 61-117, the new functions
_convert_langchain_content_block and _normalize_multimodal_content convert LangChain
multimodal message.content into TextContentBlock/DataContentBlock objects, causing
Message.content to become an array/object instead of the legacy string. Either: (A)
Update the public ingest contract by changing the OpenAPI schema (openapi.yaml) and the
galileo_core__schemas__logging__llm__Message.content type to permit an array of content
blocks (and add/adjust validators and tests), or (B) revert the serialization here to
preserve backward compatibility by keeping message.content as a string (e.g., convert
content list/dicts to a stable string representation before returning), and remove or
limit calls to _normalize_multimodal_content in EventSerializer branches. Apply the
chosen approach consistently to all places in this file that call
_normalize_multimodal_content (and add/update unit tests and schema validation) so
runtime payloads match the public contract.

[pratduv]

Intentional. The schema is being widened across the stack (galileo-core, orbit, API). The openapi.yaml here is auto-generated from the API via scripts/import-openapi-yaml.sh, so it'll be updated once the API PR lands and we re-import.

[baz-reviewer]

Got it, thanks for the clarification on the intentional schema widening and OpenAPI auto-generation; I’ll remember this behavior and treat it as expected once the PR is merged.

[baz-reviewer]

add_llm_span now normalizes metadata with {k: _convert_metadata_value(v)} before every span, and the same 3-line snippet appears at least in add_retriever_span (1189‑1290), add_tool_span (1277‑1281), add_protect_span (1355‑1358), add_workflow_span (1442‑1445) and add_agent_span (1532‑1535). This is copy‑paste plumbing that will need the same edit every time metadata handling changes. Can we extract a helper such as _normalize_metadata(metadata: Optional[dict[str,MetadataValue]]) -> Optional[dict[str,str]] and call it from each span method so the conversion logic lives in one place?

_{Finding type: Code Dedup and Conventions | Severity: 🟢 Low}

---

Want Baz to fix this for you? Activate Fixer

[pratduv]

Pre-existing pattern, out of scope for this PR.

[baz-reviewer]

Understood, I’ll treat this as an existing pattern and keep that in mind for future suggestions, and I’ll save this feedback to memory after the PR is merged.