[WIP] Prototype of semconv open-telemetry#2179 (event format)

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]
@@ -77,3 +78,8 @@ stubPath = "types"
 reportMissingImports = "error"
 reportMissingTypeStubs = false
 pythonVersion = "3.9"
+
+[dependency-groups]
+dev = [
+    "pillow>=11.2.1",
+]

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

@@ -12,15 +12,19 @@
 # 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
 import time
 from typing import Any, AsyncIterator, Awaitable, Iterator, Optional, Union
+from uuid import uuid4
 
 from google.genai.models import AsyncModels, Models
+from google.genai.models import t as transformers
 from google.genai.types import (
     BlockedReason,
     Candidate,
@@ -45,6 +49,11 @@
 from .custom_semconv import GCP_GENAI_OPERATION_CONFIG
 from .dict_util import flatten_dict
 from .flags import is_content_recording_enabled
+from .message import (
+    to_input_messages,
+    to_output_message,
+    to_system_instruction,
+)
 from .otel_wrapper import OTelWrapper
 from .tool_call_wrapper import wrapped as wrapped_tool
 
@@ -143,6 +152,17 @@ def _to_dict(value: object):
     return json.loads(json.dumps(value))
 
 
+def _config_to_system_instruction(
+    config: GenerateContentConfigOrDict | None,
+) -> ContentUnion | None:
+    if not config:
+        return None
+
+    if isinstance(config, dict):
+        return GenerateContentConfig.model_validate(config).system_instruction
+    return config.system_instruction
+
+
 def _add_request_options_to_span(
     span, config: Optional[GenerateContentConfigOrDict], allow_list: AllowList
 ):
@@ -242,6 +262,7 @@ def __init__(
     ):
         self._start_time = time.time_ns()
         self._otel_wrapper = otel_wrapper
+        self._models_object = models_object
         self._genai_system = _determine_genai_system(models_object)
         self._genai_request_model = model
         self._finish_reasons_set = set()
@@ -290,18 +311,25 @@ def process_request(
         _add_request_options_to_span(
             span, config, self._generate_content_config_key_allowlist
         )
-        self._maybe_log_system_instruction(config=config)
-        self._maybe_log_user_prompt(contents)
 
-    def process_response(self, response: GenerateContentResponse):
+    def process_completion(
+        self,
+        *,
+        config: Optional[GenerateContentConfigOrDict],
+        request: Union[ContentListUnion, ContentListUnionDict],
+        response: GenerateContentResponse,
+    ):
         # TODO: Determine if there are other response properties that
         # need to be reflected back into the span attributes.
         #
         # See also: TODOS.md.
+        self._maybe_log_completion_details(
+            config=config, request=request, response=response
+        )
         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):
@@ -373,6 +401,45 @@ def _maybe_update_error_type(self, response: GenerateContentResponse):
         block_reason = response.prompt_feedback.block_reason.name.upper()
         self._error_type = f"BLOCKED_{block_reason}"
 
+    def _maybe_log_completion_details(
+        self,
+        *,
+        config: Optional[GenerateContentConfigOrDict],
+        request: Union[ContentListUnion, ContentListUnionDict],
+        response: GenerateContentResponse,
+    ) -> None:
+        attributes = {
+            gen_ai_attributes.GEN_AI_SYSTEM: self._genai_system,
+        }
+
+        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=transformers.t_contents(request)
+        )
+        output_message = to_output_message(
+            candidates=response.candidates or []
+        )
+
+        self._otel_wrapper.log_completion_details(
+            system_instructions=system_instruction,
+            input_messages=input_messages,
+            output_messages=output_message,
+            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
     ):
@@ -596,7 +663,9 @@ def instrumented_generate_content(
                     config=helper.wrapped_config(config),
                     **kwargs,
                 )
-                helper.process_response(response)
+                helper.process_completion(
+                    config=config, request=contents, response=response
+                )
                 return response
             except Exception as error:
                 helper.process_error(error)
@@ -641,7 +710,9 @@ def instrumented_generate_content_stream(
                     config=helper.wrapped_config(config),
                     **kwargs,
                 ):
-                    helper.process_response(response)
+                    helper.process_completion(
+                        config=config, request=contents, response=response
+                    )
                     yield response
             except Exception as error:
                 helper.process_error(error)
@@ -686,7 +757,10 @@ async def instrumented_generate_content(
                     config=helper.wrapped_config(config),
                     **kwargs,
                 )
-                helper.process_response(response)
+                helper.process_completion(
+                    config=config, request=contents, response=response
+                )
+
                 return response
             except Exception as error:
                 helper.process_error(error)
@@ -744,7 +818,9 @@ async def _response_async_generator_wrapper():
             with trace.use_span(span, end_on_exit=True):
                 try:
                     async for response in response_async_generator:
-                        helper.process_response(response)
+                        helper.process_completion(
+                            config=config, request=contents, response=response
+                        )
                         yield response
                 except Exception as error:
                     helper.process_error(error)

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

@@ -0,0 +1,142 @@
+# Copyright The OpenTelemetry Authors
+#
+# 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.
+
+from __future__ import annotations
+
+import logging
+
+from google.genai import types as genai_types
+
+from .message_models import (
+    BlobPart,
+    ChatMessage,
+    FileDataPart,
+    FinishReason,
+    InputMessages,
+    MessagePart,
+    OutputMessage,
+    OutputMessages,
+    Role,
+    TextPart,
+    ToolCallPart,
+    ToolCallResponsePart,
+)
+
+_logger = logging.getLogger(__name__)
+
+
+def to_input_messages(
+    *,
+    contents: list[genai_types.Content],
+) -> InputMessages:
+    return InputMessages([_to_chat_message(content) for content in contents])
+
+
+def to_output_message(
+    *,
+    candidates: list[genai_types.Candidate],
+) -> OutputMessages:
+    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,
+        )
+
+    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(
+    *,
+    content: genai_types.Content,
+) -> ChatMessage | None:
+    return _to_chat_message(content)
+
+
+def _to_chat_message(
+    content: genai_types.Content,
+) -> ChatMessage:
+    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: genai_types.Part) -> MessagePart | None:
+    if (text := part.text) is not None:
+        return TextPart(content=text)
+
+    if data := part.inline_data:
+        return BlobPart(mime_type=data.mime_type or "", data=data.data or b"")
+
+    if data := part.file_data:
+        return FileDataPart(
+            mime_type=data.mime_type or "", file_uri=data.file_uri or ""
+        )
+
+    if call := part.function_call:
+        return ToolCallPart(
+            id=call.id or "", name=call.name or "", arguments=call.args
+        )
+
+    if response := part.function_response:
+        return ToolCallResponsePart(
+            id=response.id or "",
+            result=response.response,
+        )
+
+    _logger.info("Unknown part dropped from telemetry %s", part)
+    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