[BUG]: google_adk LLMObs agent span records every streaming partial as a separate message

[JonathanvanWestendorp]

Summary

In streaming mode, the google_adk integration writes one message per ADK event to the agent span's output, including every partial=True chunk. A streamed answer therefore appears in LLM Observability split into its chunks and repeated in full by the final aggregated event, instead of once.

For a multi-turn agent this multiplies: an answer streamed in 40 chunks becomes 41 messages in the span, which makes the conversation view unreadable and inflates the span payload toward the event size limit.

Version / environment

• ddtrace: 4.12.2 (also present on main - the offending function is unchanged)
• google-adk: 2.5.0
• Python: 3.11.15
• Export mode: agentless (DD_LLMOBS_AGENTLESS_ENABLED=1), though the extraction path is export-mode independent

Root cause

_traced_agent_run_async accumulates every yielded event (ddtrace/contrib/internal/google_adk/patch.py:53-56):

async for event in agen:
    response_events.append(event)
    yield event

_llmobs_set_tags_agent passes that list to extract_messages_from_adk_events (ddtrace/llmobs/_integrations/google_utils.py:330-361), which iterates every event with no check on Event.partial. ADK sets partial=True on streaming chunks and then yields a final event carrying the assembled text with partial unset, so both forms are recorded.

Reproduction

import asyncio
import json
import os

os.environ.update(DD_LLMOBS_ENABLED="1", DD_LLMOBS_AGENTLESS_ENABLED="1",
                  DD_API_KEY="unused", DD_APM_TRACING_ENABLED="false")
import ddtrace.auto
from ddtrace.llmobs import LLMObs
from google.adk.agents import Agent
from google.adk.agents.run_config import RunConfig, StreamingMode
from google.adk.models.base_llm import BaseLlm
from google.adk.models.llm_response import LlmResponse
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

spans = []
LLMObs._instance._llmobs_span_writer.enqueue = spans.append


class StreamingLlm(BaseLlm):
    """Stands in for Gemini in SSE mode: partial chunks, then the assembled turn."""

    async def generate_content_async(self, llm_request, stream: bool = False):
        for chunk in ["Hello ", "from ", "ADK."]:
            yield LlmResponse(content=types.Content(role="model", parts=[types.Part(text=chunk)]),
                              partial=True)
        yield LlmResponse(content=types.Content(role="model",
                                                parts=[types.Part(text="Hello from ADK.")]))


runner = Runner(app_name="demo", agent=Agent(name="demo", model=StreamingLlm(model="fake")),
                session_service=InMemorySessionService())


async def main():
    await runner.session_service.create_session(app_name="demo", user_id="u", session_id="s")
    async for _ in runner.run_async(
        user_id="u", session_id="s",
        new_message=types.Content(role="user", parts=[types.Part(text="hi")]),
        run_config=RunConfig(streaming_mode=StreamingMode.SSE),
    ):
        pass


asyncio.run(main())
agent_span = next(s for s in spans if s["meta"]["span"]["kind"] == "agent")
print(json.dumps(json.loads(agent_span["meta"]["output"]["value"]), indent=2))

Actual output

[
  {"content": "Hello ",          "role": "assistant"},
  {"content": "from ",           "role": "assistant"},
  {"content": "ADK.",            "role": "assistant"},
  {"content": "Hello from ADK.", "role": "assistant"}
]

Expected output

[
  {"content": "Hello from ADK.", "role": "assistant"}
]

Suggested fix

Drop partial events in extract_messages_from_adk_events, since the final event of each turn already carries the assembled content. run_live is wrapped by the same _traced_agent_run_async, so guarding on the presence of an assembled event keeps live runs from losing their output entirely:

if isinstance(events, list) and any(not _get_attr(e, "partial", False) for e in events):
    events = [e for e in events if not _get_attr(e, "partial", False)]

Working around this downstream is awkward: LLMObs.register_processor is the supported hook, but an agent span's output reaches it as a single Message whose content is a JSON blob (_llmobs.py:386 routes messages only for kind == "llm"), and Event.partial is gone by then - so the only reliable filter point is inside the integration.