refactor(otel): Add pure functions for constructing stable and experimental semconv logs

Co-authored-by: Max Ind <maxind@google.com>
PiperOrigin-RevId: 933035924

Author: RKest

src/google/adk/telemetry/_experimental_semconv.py

@@ -19,7 +19,6 @@
 import logging
 import sys
 import time
-from typing import Any
 from typing import AsyncIterator
 from typing import TYPE_CHECKING
 
@@ -94,8 +93,8 @@ def record_llm_response(
 def _record_agent_metrics(
     agent_name: str,
     elapsed_s: float,
-    user_content: Any,
-    events: Any,
+    user_content: object,
+    events: object,
     caught_error: Exception | None,
 ) -> None:
   try:
@@ -143,7 +142,7 @@ async def record_agent_invocation(
 async def record_tool_execution(
     tool: BaseTool,
     agent: BaseAgent,
-    function_args: dict[str, Any],
+    function_args: dict[str, object],
     invocation_context: InvocationContext | None = None,
 ) -> AsyncIterator[TelemetryContext]:
   """Unified context manager for consolidated tool execution telemetry."""

src/google/adk/telemetry/_instrumentation.py

@@ -0,0 +1,61 @@
+# Copyright 2026 Google LLC
+#
+# 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.
+
+"""Shared serialization helpers used by telemetry modules."""
+
+from __future__ import annotations
+
+import json
+
+from google.genai import types
+from opentelemetry.util.types import AnyValue
+from pydantic import BaseModel
+
+
+def safe_json_serialize(obj: object) -> str:
+  """Convert any Python object to a JSON-serializable type or string.
+
+  Args:
+    obj: The object to serialize.
+
+  Returns:
+    The JSON-serialized object string or `<not serializable>` if the object
+    cannot be serialized.
+  """
+  try:
+    return json.dumps(
+        obj, ensure_ascii=False, default=lambda o: "<not serializable>"
+    )
+  except (TypeError, ValueError, OverflowError):
+    return "<not serializable>"
+
+
+def serialize_content(content: types.ContentUnion | None) -> AnyValue:
+  """Serialize a `types.ContentUnion` value into an OTel-friendly form.
+
+  - `None` is preserved.
+  - Pydantic models are dumped via `model_dump()`.
+  - Strings are returned as-is.
+  - Lists are recursively serialized.
+  - Anything else falls back to `safe_json_serialize`.
+  """
+  if content is None:
+    return None
+  if isinstance(content, BaseModel):
+    return content.model_dump()
+  if isinstance(content, str):
+    return content
+  if isinstance(content, list):
+    return [serialize_content(part) for part in content]
+  return safe_json_serialize(content)

src/google/adk/telemetry/_serialization.py

@@ -0,0 +1,146 @@
+# Copyright 2026 Google LLC
+#
+# 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.
+
+"""Helpers for building log bodies that follow the stable OTel GenAI semconv.
+
+This module centralizes the construction of `gen_ai.system.message`,
+`gen_ai.user.message`, and `gen_ai.choice` log bodies so that both the
+tracing layer (which emits the logs) and the ADK Web UI exporter (which
+rebuilds the bodies after elision) share the same shape.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import TYPE_CHECKING
+
+from opentelemetry.util.types import AnyValue
+
+from ._serialization import serialize_content
+from .context import TelemetryConfig
+
+if TYPE_CHECKING:
+  from google.genai import types
+
+  from ..models.llm_request import LlmRequest
+  from ..models.llm_response import LlmResponse
+
+# Stable OTel GenAI semantic-convention event names.
+GEN_AI_SYSTEM_MESSAGE_EVENT = "gen_ai.system.message"
+GEN_AI_USER_MESSAGE_EVENT = "gen_ai.user.message"
+GEN_AI_CHOICE_EVENT = "gen_ai.choice"
+
+# Standard OTEL env variable that controls whether prompt/response content is
+# included in log bodies. When unset/false, content is replaced with <elided>.
+OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT = (
+    "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
+)
+
+USER_CONTENT_ELIDED = "<elided>"
+
+
+def _serialize_content_with_optional_elision(
+    content: types.ContentUnion | None, *, capture_content: bool
+) -> AnyValue:
+  if not capture_content:
+    return USER_CONTENT_ELIDED
+  if content is None:
+    return None
+  return serialize_content(content)
+
+
+def system_message_body(
+    llm_request: LlmRequest,
+    telemetry_config: TelemetryConfig,
+    *,
+    do_not_elide: bool = False,
+) -> Mapping[str, AnyValue]:
+  """Builds the body for a `gen_ai.system.message` log event.
+
+  Args:
+    llm_request: The LLM request whose system instruction should be logged.
+    do_not_elide_content: When True, always include the content regardless of
+      the `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT` env var. The Web
+      UI exporter sets this to True because the UI needs the full content.
+  """
+  system_instruction = None
+  if llm_request.config is not None:
+    system_instruction = llm_request.config.system_instruction
+  return {
+      "content": _serialize_content_with_optional_elision(
+          system_instruction,
+          capture_content=do_not_elide
+          or telemetry_config.should_add_content_to_logs,
+      )
+  }
+
+
+def user_message_body(
+    content: types.ContentUnion | None,
+    telemetry_config: TelemetryConfig,
+    *,
+    do_not_elide: bool = False,
+) -> Mapping[str, AnyValue]:
+  """Builds the body for a single `gen_ai.user.message` log event.
+
+  Args:
+    content: The user content for this message. Callers that emit multiple user
+      messages (e.g. tracing's per-content loop) call this builder once per
+      content.
+    do_not_elide_content: When True, always include the content regardless of
+      the `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT` env var.
+  """
+  return {
+      "content": _serialize_content_with_optional_elision(
+          content,
+          capture_content=do_not_elide
+          or telemetry_config.should_add_content_to_logs,
+      )
+  }
+
+
+def choice_body(
+    llm_response: LlmResponse | None,
+    telemetry_config: TelemetryConfig,
+    *,
+    do_not_elide: bool = False,
+) -> Mapping[str, AnyValue]:
+  """Builds the body for a `gen_ai.choice` log event.
+
+  ADK always returns a single candidate, so `index` is always 0.
+  `finish_reason` is included only when present on the response.
+
+  Args:
+    llm_response: The LLM response describing the choice.
+    do_not_elide_content: When True, always include the content regardless of
+      the `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT` env var.
+  """
+  if llm_response is None:
+    return {"content": None, "index": 0}
+  body: dict[str, AnyValue] = {
+      "content": _serialize_content_with_optional_elision(
+          llm_response.content,
+          capture_content=do_not_elide
+          or telemetry_config.should_add_content_to_logs,
+      ),
+      "index": 0,  # ADK always returns a single candidate.
+  }
+  if llm_response.finish_reason is not None:
+    finish_reason = llm_response.finish_reason
+    body["finish_reason"] = (
+        finish_reason.value
+        if hasattr(finish_reason, "value")
+        else str(finish_reason)
+    )
+  return body

src/google/adk/telemetry/_stable_semconv.py

@@ -20,7 +20,6 @@
 import logging
 import sqlite3
 import threading
-from typing import Any
 from typing import Iterable
 from typing import Optional
 from typing import Sequence
@@ -104,7 +103,7 @@ def _ensure_schema(self) -> None:
       conn.execute(_CREATE_TRACE_INDEX)
       conn.commit()
 
-  def _serialize_attributes(self, attributes: dict[str, Any]) -> str:
+  def _serialize_attributes(self, attributes: dict[str, object]) -> str:
     try:
       return json.dumps(
           attributes,
@@ -115,7 +114,9 @@ def _serialize_attributes(self, attributes: dict[str, Any]) -> str:
       logger.debug("Failed to serialize span attributes: %r", e)
       return "{}"
 
-  def _deserialize_attributes(self, attributes_json: Any) -> dict[str, Any]:
+  def _deserialize_attributes(
+      self, attributes_json: object
+  ) -> dict[str, object]:
     if not attributes_json:
       return {}
     try:
@@ -129,7 +130,7 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
     try:
       with self._lock:
         conn = self._get_connection()
-        rows: list[tuple[Any, ...]] = []
+        rows: list[tuple[object, ...]] = []
         for span in spans:
           attributes = dict(span.attributes) if span.attributes else {}
           session_id = attributes.get(
@@ -168,7 +169,7 @@ def shutdown(self) -> None:
   def force_flush(self, timeout_millis: int = 30000) -> bool:
     return True
 
-  def _query(self, sql: str, params: Iterable[Any]) -> list[sqlite3.Row]:
+  def _query(self, sql: str, params: Iterable[object]) -> list[sqlite3.Row]:
     with self._lock:
       conn = self._get_connection()
       cur = conn.execute(sql, tuple(params))