fix(genai-utils): move get_content_attributes to _invocation.py, add test coverage

Author: etserend

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py

@@ -22,8 +22,11 @@
 )
 from opentelemetry.semconv.attributes import server_attributes
 from opentelemetry.trace import SpanKind, Tracer
-from opentelemetry.util.genai._content import _get_content_attributes
-from opentelemetry.util.genai._invocation import Error, GenAIInvocation
+from opentelemetry.util.genai._invocation import (
+    Error,
+    GenAIInvocation,
+    get_content_attributes,
+)
 from opentelemetry.util.genai.metrics import InvocationMetricsRecorder
 from opentelemetry.util.genai.types import (
     InputMessage,
@@ -32,6 +35,17 @@
     ToolDefinition,
 )
 
+_GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS: str = getattr(
+    GenAI,
+    "GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS",
+    "gen_ai.usage.cache_creation.input_tokens",
+)
+_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS: str = getattr(
+    GenAI,
+    "GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS",
+    "gen_ai.usage.cache_read.input_tokens",
+)
+
 
 class AgentInvocation(GenAIInvocation):
     """Represents a single agent invocation (invoke_agent span).
@@ -143,18 +157,18 @@ def _get_usage_attributes(self) -> dict[str, Any]:
             (GenAI.GEN_AI_USAGE_INPUT_TOKENS, self.input_tokens),
             (GenAI.GEN_AI_USAGE_OUTPUT_TOKENS, self.output_tokens),
             (
-                GenAI.GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS,
+                _GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS,
                 self.cache_creation_input_tokens,
             ),
             (
-                GenAI.GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS,
+                _GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS,
                 self.cache_read_input_tokens,
             ),
         )
         return {k: v for k, v in optional_attrs if v is not None}
 
     def _get_content_attributes_for_span(self) -> dict[str, Any]:
-        return _get_content_attributes(
+        return get_content_attributes(
             input_messages=self.input_messages,
             output_messages=self.output_messages,
             system_instruction=self.system_instruction,

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_content.py

@@ -23,8 +23,11 @@
 )
 from opentelemetry.semconv.attributes import server_attributes
 from opentelemetry.trace import INVALID_SPAN, Span, SpanKind, Tracer
-from opentelemetry.util.genai._content import _get_content_attributes
-from opentelemetry.util.genai._invocation import Error, GenAIInvocation
+from opentelemetry.util.genai._invocation import (
+    Error,
+    GenAIInvocation,
+    get_content_attributes,
+)
 from opentelemetry.util.genai.metrics import InvocationMetricsRecorder
 from opentelemetry.util.genai.types import (
     InputMessage,
@@ -37,6 +40,17 @@
     should_emit_event,
 )
 
+_GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS: str = getattr(
+    GenAI,
+    "GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS",
+    "gen_ai.usage.cache_creation.input_tokens",
+)
+_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS: str = getattr(
+    GenAI,
+    "GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS",
+    "gen_ai.usage.cache_read.input_tokens",
+)
+
 
 class InferenceInvocation(GenAIInvocation):
     """Represents a single LLM chat/completion call.
@@ -118,7 +132,7 @@ def __init__(  # pylint: disable=too-many-locals
         self._start()
 
     def _get_message_attributes(self, *, for_span: bool) -> dict[str, Any]:
-        return _get_content_attributes(
+        return get_content_attributes(
             input_messages=self.input_messages,
             output_messages=self.output_messages,
             system_instruction=self.system_instruction,
@@ -166,11 +180,11 @@ def _get_attributes(self) -> dict[str, Any]:
             (GenAI.GEN_AI_USAGE_INPUT_TOKENS, self.input_tokens),
             (GenAI.GEN_AI_USAGE_OUTPUT_TOKENS, self.output_tokens),
             (
-                GenAI.GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS,
+                _GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS,
                 self.cache_creation_input_tokens,
             ),
             (
-                GenAI.GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS,
+                _GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS,
                 self.cache_read_input_tokens,
             ),
         )

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_inference_invocation.py

@@ -18,17 +18,33 @@
 from abc import ABC, abstractmethod
 from contextlib import contextmanager
 from contextvars import Token
-from typing import TYPE_CHECKING, Any, Iterator
+from dataclasses import asdict
+from typing import TYPE_CHECKING, Any, Iterator, Sequence
 
 from typing_extensions import Self, TypeAlias
 
 from opentelemetry._logs import Logger
 from opentelemetry.context import Context, attach, detach
+from opentelemetry.semconv._incubating.attributes import (
+    gen_ai_attributes as GenAI,
+)
 from opentelemetry.semconv.attributes import error_attributes
 from opentelemetry.trace import INVALID_SPAN as _INVALID_SPAN
 from opentelemetry.trace import Span, SpanKind, Tracer, set_span_in_context
 from opentelemetry.trace.status import Status, StatusCode
-from opentelemetry.util.genai.types import Error
+from opentelemetry.util.genai.types import (
+    Error,
+    InputMessage,
+    MessagePart,
+    OutputMessage,
+    ToolDefinition,
+)
+from opentelemetry.util.genai.utils import (
+    ContentCapturingMode,
+    gen_ai_json_dumps,
+    get_content_capturing_mode,
+    is_experimental_mode,
+)
 
 if TYPE_CHECKING:
     from opentelemetry.util.genai.metrics import InvocationMetricsRecorder
@@ -138,3 +154,64 @@ def _managed(self) -> Iterator[Self]:
             self.fail(exc)
             raise
         self.stop()
+
+
+def get_content_attributes(
+    *,
+    input_messages: Sequence[InputMessage],
+    output_messages: Sequence[OutputMessage],
+    system_instruction: Sequence[MessagePart],
+    tool_definitions: Sequence[ToolDefinition] | None,
+    for_span: bool,
+) -> dict[str, Any]:
+    """Serialize messages, system instructions, and tool definitions into attributes.
+
+    Args:
+        input_messages: Input messages to serialize.
+        output_messages: Output messages to serialize.
+        system_instruction: System instructions to serialize.
+        tool_definitions: Tool definitions to serialize (may be None).
+        for_span: If True, serialize for span attributes (JSON string);
+                  if False, serialize for event attributes (list of dicts).
+    """
+    if not is_experimental_mode():
+        return {}
+
+    mode = get_content_capturing_mode()
+    allowed_modes = (
+        (
+            ContentCapturingMode.SPAN_ONLY,
+            ContentCapturingMode.SPAN_AND_EVENT,
+        )
+        if for_span
+        else (
+            ContentCapturingMode.EVENT_ONLY,
+            ContentCapturingMode.SPAN_AND_EVENT,
+        )
+    )
+    if mode not in allowed_modes:
+        return {}
+
+    def serialize(items: Sequence[Any]) -> Any:
+        dicts = [asdict(item) for item in items]
+        return gen_ai_json_dumps(dicts) if for_span else dicts
+
+    optional_attrs = (
+        (
+            GenAI.GEN_AI_INPUT_MESSAGES,
+            serialize(input_messages) if input_messages else None,
+        ),
+        (
+            GenAI.GEN_AI_OUTPUT_MESSAGES,
+            serialize(output_messages) if output_messages else None,
+        ),
+        (
+            GenAI.GEN_AI_SYSTEM_INSTRUCTIONS,
+            serialize(system_instruction) if system_instruction else None,
+        ),
+        (
+            GenAI.GEN_AI_TOOL_DEFINITIONS,
+            serialize(tool_definitions) if tool_definitions else None,
+        ),
+    )
+    return {key: value for key, value in optional_attrs if value is not None}