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

Author: aabmass

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

@@ -77,3 +77,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

@@ -17,16 +17,20 @@
 import json
 import logging
 import os
+from re import S
 import time
 from typing import Any, AsyncIterator, Awaitable, Iterator, Optional, Union
 
-from google.genai.models import AsyncModels, Models
+from google.genai.models import AsyncModels, Models, t as transformers
 from google.genai.types import (
     BlockedReason,
     Candidate,
     Content,
     ContentListUnion,
+    GenerateContentConfig,
+    ContentUnion,
     ContentListUnionDict,
+    Content,
     ContentUnion,
     ContentUnionDict,
     GenerateContentConfig,
@@ -47,6 +51,12 @@
 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,
+)
 
 _logger = logging.getLogger(__name__)
 
@@ -143,6 +153,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 +263,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,14 +312,21 @@ 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)
@@ -373,6 +402,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:
+        def _transform_content(
+            content: Union[
+                ContentListUnion, ContentListUnionDict, Content, None
+            ],
+        ) -> list[Content]:
+            if content is None:
+                return []
+            return transformers.t_contents(
+                self._models_object._api_client, content
+            )
+
+        attributes = {
+            gen_ai_attributes.GEN_AI_SYSTEM: self._genai_system,
+        }
+
+        system_instruction = to_system_instruction(
+            contents=_transform_content(_config_to_system_instruction(config))
+        )
+        input_messages = to_input_messages(
+            contents=_transform_content(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,
+        )
+
     def _maybe_log_system_instruction(
         self, config: Optional[GenerateContentConfigOrDict] = None
     ):
@@ -596,7 +664,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 +711,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 +758,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 +819,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,118 @@
+# 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.
+
+import logging
+from .message_models import (
+    BlobPart,
+    ChatMessage,
+    Choice,
+    FileDataPart,
+    InputMessages,
+    MessagePart,
+    SystemMessage,
+    OutputMessages,
+    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],
+) -> InputMessages:
+    return InputMessages(
+        messages=[_to_chat_message(content) for content in contents]
+    )
+
+
+def to_output_message(
+    *,
+    candidates: list[Candidate],
+) -> OutputMessages:
+    choices = [
+        Choice(
+            finish_reason=candidate.finish_reason or "",
+            message=_to_chat_message(candidate.content)
+            if candidate.content
+            else None,
+        )
+        for candidate in candidates
+    ]
+    return OutputMessages(choices=choices)
+
+
+def to_system_instruction(
+    *,
+    contents: list[Content],
+) -> SystemMessage | None:
+    return SystemMessage(
+        messages=[_to_chat_message(content) for content in contents]
+    )
+
+
+def _to_chat_message(
+    content: Content,
+) -> ChatMessage:
+    parts = content.parts or []
+    return ChatMessage(role="system", parts=[_to_part(part) for part in parts])
+
+
+def _to_part(part: Part) -> MessagePart:
+    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 "",
+            name=response.name or "",
+            result=response.response,
+        )
+
+    _logger.info("Unknown part dropped from telemetry %s", part)
+    return UnknownPart()