Uploader class for _ref events

WIP use newest iteration of semconv

Author: aabmass

instrumentation-genai/opentelemetry-instrumentation-google-genai/pyproject.toml

@@ -37,9 +37,10 @@ classifiers = [
   "Programming Language :: Python :: 3.12"
 ]
 dependencies = [
+  "fsspec>=2025.5.1",
   "opentelemetry-api >=1.31.1, <2",
   "opentelemetry-instrumentation >=0.52b1, <2",
-  "opentelemetry-semantic-conventions >=0.52b1, <2"
+  "opentelemetry-semantic-conventions >=0.52b1, <2",
 ]
 
 [project.optional-dependencies]

instrumentation-genai/opentelemetry-instrumentation-google-genai/src/opentelemetry/instrumentation/google_genai/generate_content.py

@@ -12,25 +12,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from __future__ import annotations
+
 import copy
 import functools
 import json
 import logging
 import os
-from re import S
 import time
 from typing import Any, AsyncIterator, Awaitable, Iterator, Optional, Union
+from uuid import uuid4
 
-from google.genai.models import AsyncModels, Models, t as transformers
+from google.genai.models import AsyncModels, Models
+from google.genai.models import t as transformers
 from google.genai.types import (
     BlockedReason,
     Candidate,
     Content,
     ContentListUnion,
-    GenerateContentConfig,
-    ContentUnion,
     ContentListUnionDict,
-    Content,
     ContentUnion,
     ContentUnionDict,
     GenerateContentConfig,
@@ -49,14 +49,13 @@
 from .custom_semconv import GCP_GENAI_OPERATION_CONFIG
 from .dict_util import flatten_dict
 from .flags import is_content_recording_enabled
-from .otel_wrapper import OTelWrapper
-from .tool_call_wrapper import wrapped as wrapped_tool
 from .message import (
-    ContentUnion,
     to_input_messages,
     to_output_message,
     to_system_instruction,
 )
+from .otel_wrapper import OTelWrapper
+from .tool_call_wrapper import wrapped as wrapped_tool
 
 _logger = logging.getLogger(__name__)
 
@@ -330,7 +329,7 @@ def process_completion(
         self._update_finish_reasons(response)
         self._maybe_update_token_counts(response)
         self._maybe_update_error_type(response)
-        self._maybe_log_response(response)
+        # self._maybe_log_response(response)
         self._response_index += 1
 
     def process_error(self, e: Exception):
@@ -409,24 +408,17 @@ def _maybe_log_completion_details(
         request: Union[ContentListUnion, ContentListUnionDict],
         response: GenerateContentResponse,
     ) -> None:
-        def _transform_content(
-            content: Union[
-                ContentListUnion, ContentListUnionDict, Content, None
-            ],
-        ) -> list[Content]:
-            if content is None:
-                return []
-            return transformers.t_contents(content)
-
         attributes = {
             gen_ai_attributes.GEN_AI_SYSTEM: self._genai_system,
         }
 
-        system_instruction = to_system_instruction(
-            contents=_transform_content(_config_to_system_instruction(config))
-        )
+        system_instruction = None
+        if system_content := _config_to_system_instruction(config):
+            system_instruction = to_system_instruction(
+                content=transformers.t_contents(system_content)[0]
+            )
         input_messages = to_input_messages(
-            contents=_transform_content(request)
+            contents=transformers.t_contents(request)
         )
         output_message = to_output_message(
             candidates=response.candidates or []
@@ -439,6 +431,15 @@ def _transform_content(
             attributes=attributes,
         )
 
+        # Forward looking remote storage refs
+        self._otel_wrapper.log_completion_details_refs(
+            system_instructions=system_instruction,
+            input_messages=input_messages,
+            output_messages=output_message,
+            attributes=attributes,
+            response_id=response.response_id or str(uuid4()),
+        )
+
     def _maybe_log_system_instruction(
         self, config: Optional[GenerateContentConfigOrDict] = None
     ):

instrumentation-genai/opentelemetry-instrumentation-google-genai/src/opentelemetry/instrumentation/google_genai/message.py

@@ -12,85 +12,83 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from __future__ import annotations
+
 import logging
+
+from google.genai import types as genai_types
+
 from .message_models import (
     BlobPart,
     ChatMessage,
-    Choice,
     FileDataPart,
+    FinishReason,
     InputMessages,
     MessagePart,
-    SystemMessage,
+    OutputMessage,
     OutputMessages,
+    Role,
     TextPart,
     ToolCallPart,
     ToolCallResponsePart,
-    UnknownPart,
-)
-
-from google.genai.models import t as transfomers
-
-from google.genai.types import (
-    BlockedReason,
-    Candidate,
-    Content,
-    ContentListUnion,
-    ContentListUnionDict,
-    ContentUnion,
-    GenerateContentConfig,
-    ContentUnionDict,
-    GenerateContentConfig,
-    GenerateContentConfigOrDict,
-    GenerateContentResponse,
-    Part,
-    PartUnion,
 )
 
 _logger = logging.getLogger(__name__)
 
 
 def to_input_messages(
     *,
-    contents: list[Content],
+    contents: list[genai_types.Content],
 ) -> InputMessages:
-    return InputMessages(
-        messages=[_to_chat_message(content) for content in contents]
-    )
+    return InputMessages([_to_chat_message(content) for content in contents])
 
 
 def to_output_message(
     *,
-    candidates: list[Candidate],
+    candidates: list[genai_types.Candidate],
 ) -> OutputMessages:
-    choices = [
-        Choice(
-            finish_reason=candidate.finish_reason or "",
-            message=_to_chat_message(candidate.content)
-            if candidate.content
-            else None,
+    def content_to_output_message(
+        candidate: genai_types.Candidate,
+    ) -> OutputMessage | None:
+        if not candidate.content:
+            return None
+
+        message = _to_chat_message(candidate.content)
+        return OutputMessage(
+            finish_reason=_to_finish_reason(candidate.finish_reason),
+            role=message.role,
+            parts=message.parts,
         )
-        for candidate in candidates
-    ]
-    return OutputMessages(choices=choices)
+
+    messages = (
+        content_to_output_message(candidate) for candidate in candidates
+    )
+    return OutputMessages(
+        [message for message in messages if message is not None]
+    )
 
 
+# TODO: re-using ChatMessage for now but it is defined as any in
+# https://github.com/open-telemetry/semantic-conventions/pull/2179. I prefer ChatMessage.
 def to_system_instruction(
     *,
-    contents: list[Content],
-) -> SystemMessage | None:
-    return SystemMessage(
-        messages=[_to_chat_message(content) for content in contents]
-    )
+    content: genai_types.Content,
+) -> ChatMessage | None:
+    return _to_chat_message(content)
 
 
 def _to_chat_message(
-    content: Content,
+    content: genai_types.Content,
 ) -> ChatMessage:
-    parts = content.parts or []
-    return ChatMessage(role="system", parts=[_to_part(part) for part in parts])
+    parts = (_to_part(part) for part in (content.parts or []))
+    return ChatMessage(
+        role=_to_role(content.role),
+        # filter Nones
+        parts=[part for part in parts if part is not None],
+    )
 
 
-def _to_part(part: Part) -> MessagePart:
+def _to_part(part: genai_types.Part) -> MessagePart | None:
     if (text := part.text) is not None:
         return TextPart(content=text)
 
@@ -110,9 +108,35 @@ def _to_part(part: Part) -> MessagePart:
     if response := part.function_response:
         return ToolCallResponsePart(
             id=response.id or "",
-            name=response.name or "",
             result=response.response,
         )
 
     _logger.info("Unknown part dropped from telemetry %s", part)
-    return UnknownPart()
+    return None
+
+
+def _to_role(role: str | None) -> Role | str:
+    if role == "user":
+        return Role.USER
+    elif role == "model":
+        return Role.ASSISTANT
+    return ""
+
+
+def _to_finish_reason(
+    finish_reason: genai_types.FinishReason | None,
+) -> FinishReason | str:
+    if finish_reason is None:
+        return ""
+    if (
+        finish_reason is genai_types.FinishReason.FINISH_REASON_UNSPECIFIED
+        or finish_reason is genai_types.FinishReason.OTHER
+    ):
+        return FinishReason.ERROR
+    if finish_reason is genai_types.FinishReason.STOP:
+        return FinishReason.STOP
+    if finish_reason is genai_types.FinishReason.MAX_TOKENS:
+        return FinishReason.LENGTH
+
+    # If there is no 1:1 mapping to an OTel preferred enum value, use the exact vertex reason
+    return finish_reason.name