GenAI Utils | Inference Type and Span Creation

docs/nitpick-exceptions.ini

@@ -45,6 +45,7 @@ py-class=
     psycopg.AsyncConnection
     ObjectProxy
     fastapi.applications.FastAPI
+    _contextvars.Token
 
 any=
     ; API

util/opentelemetry-util-genai/CHANGELOG.md

@@ -24,3 +24,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3763](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3763))
 - Add a utility to parse the `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT` environment variable.
   Add `gen_ai_latest_experimental` as a new value to the Sem Conv stability flag ([#3716](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3716)).
+
+### Added
+
+- Generate Spans for LLM invocations
+- Helper functions for starting and finishing LLM invocations

util/opentelemetry-util-genai/README.rst

@@ -6,6 +6,25 @@ The GenAI Utils package will include boilerplate and helpers to standardize inst
 This package will provide APIs and decorators to minimize the work needed to instrument genai libraries, 
 while providing standardization for generating both types of otel, "spans and metrics" and "spans, metrics and events"
 
+This package relies on environment variables to configure capturing of message content. 
+By default, message content will not be captured.
+Set the environment variable `OTEL_SEMCONV_STABILITY_OPT_IN` to `gen_ai_latest_experimental` to enable experimental features.
+And set the environment variable `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT` to `SPAN_ONLY` or `SPAN_AND_EVENT` to capture message content in spans.
+
+This package provides these span attributes:
+
+- `gen_ai.provider.name`: Str(openai)
+- `gen_ai.operation.name`: Str(chat)
+- `gen_ai.request.model`: Str(gpt-3.5-turbo)
+- `gen_ai.response.finish_reasons`: Slice(["stop"])
+- `gen_ai.response.model`: Str(gpt-3.5-turbo-0125)
+- `gen_ai.response.id`: Str(chatcmpl-Bz8yrvPnydD9pObv625n2CGBPHS13)
+- `gen_ai.usage.input_tokens`: Int(24)
+- `gen_ai.usage.output_tokens`: Int(7)
+- `gen_ai.input.messages`: Str('[{"role": "Human", "parts": [{"content": "hello world", "type": "text"}]}]')
+- `gen_ai.output.messages`: Str('[{"role": "AI", "parts": [{"content": "hello back", "type": "text"}], "finish_reason": "stop"}]')
+
+
 Installation
 ------------
 

util/opentelemetry-util-genai/pyproject.toml

@@ -8,7 +8,7 @@ dynamic = ["version"]
 description = "OpenTelemetry GenAI Utils"
 readme = "README.rst"
 license = "Apache-2.0"
-requires-python = ">=3.8"
+requires-python = ">=3.9"
 authors = [
   { name = "OpenTelemetry Authors", email = "[email protected]" },
 ]
@@ -25,8 +25,8 @@ classifiers = [
   "Programming Language :: Python :: 3.13",
 ]
 dependencies = [
-  "opentelemetry-instrumentation ~= 0.51b0",
-  "opentelemetry-semantic-conventions ~= 0.51b0",
+  "opentelemetry-instrumentation ~= 0.57b0",
+  "opentelemetry-semantic-conventions ~= 0.57b0",
   "opentelemetry-api>=1.31.0",
 ]
 

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

@@ -0,0 +1,13 @@
+# 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.

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

@@ -0,0 +1,180 @@
+# 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.
+
+"""
+Telemetry handler for GenAI invocations.
+
+This module exposes the `TelemetryHandler` class, which manages the lifecycle of
+GenAI (Generative AI) invocations and emits telemetry data (spans and related attributes).
+It supports starting, stopping, and failing LLM invocations.
+
+Classes:
+    - TelemetryHandler: Manages GenAI invocation lifecycles and emits telemetry.
+
+Functions:
+    - get_telemetry_handler: Returns a singleton `TelemetryHandler` instance.
+
+Usage:
+    handler = get_telemetry_handler()
+
+    # Create an invocation object with your request data
+    # The span and context_token attributes are set by the TelemetryHandler, and
+    # managed by the TelemetryHandler during the lifecycle of the span.
+
+    # Use the context manager to manage the lifecycle of an LLM invocation.
+    with handler.llm(invocation) as invocation:
+        # Populate outputs and any additional attributes
+        invocation.output_messages = [...]
+        invocation.attributes.update({"more": "attrs"})
+
+    # Or, if you prefer to manage the lifecycle manually
+    invocation = LLMInvocation(
+        request_model="my-model",
+        input_messages=[...],
+        provider="my-provider",
+        attributes={"custom": "attr"},
+    )
+
+    # Start the invocation (opens a span)
+    handler.start_llm(invocation)
+
+    # Populate outputs and any additional attributes, then stop (closes the span)
+    invocation.output_messages = [...]
+    invocation.attributes.update({"more": "attrs"})
+    handler.stop_llm(invocation)
+
+    # Or, in case of error
+    handler.fail_llm(invocation, Error(type="...", message="..."))
+"""
+
+import time
+from contextlib import contextmanager
+from typing import Any, Iterator, Optional
+
+from opentelemetry import context as otel_context
+from opentelemetry import trace
+from opentelemetry.semconv._incubating.attributes import (
+    gen_ai_attributes as GenAI,
+)
+from opentelemetry.semconv.schemas import Schemas
+from opentelemetry.trace import (
+    SpanKind,
+    Tracer,
+    get_tracer,
+    set_span_in_context,
+)
+from opentelemetry.util.genai.span_utils import (
+    _apply_error_attributes,
+    _apply_finish_attributes,
+)
+from opentelemetry.util.genai.types import Error, LLMInvocation
+from opentelemetry.util.genai.version import __version__
+
+
+class TelemetryHandler:
+    """
+    High-level handler managing GenAI invocation lifecycles and emitting
+    them as spans, metrics, and events.
+    """
+
+    def __init__(self, **kwargs: Any):
+        tracer_provider = kwargs.get("tracer_provider")
+        tracer = get_tracer(
+            __name__,
+            __version__,
+            tracer_provider,
+            schema_url=Schemas.V1_36_0.value,
+        )
+        self._tracer: Tracer = tracer or trace.get_tracer(__name__)
+
+    def start_llm(
+        self,
+        invocation: LLMInvocation,
+    ) -> LLMInvocation:
+        """Start an LLM invocation and create a pending span entry."""
+        # Create a span and attach it as current; keep the token to detach later
+        span = self._tracer.start_span(
+            name=f"{GenAI.GenAiOperationNameValues.CHAT.value} {invocation.request_model}",
+            kind=SpanKind.CLIENT,
+        )
+        invocation.span = span
+        invocation.context_token = otel_context.attach(
+            set_span_in_context(span)
+        )
+        return invocation
+
+    def stop_llm(self, invocation: LLMInvocation) -> LLMInvocation:  # pylint: disable=no-self-use
+        """Finalize an LLM invocation successfully and end its span."""
+        invocation.end_time = time.time()
+        if invocation.context_token is None or invocation.span is None:
+            # TODO: Provide feedback that this invocation was not started
+            return invocation
+
+        _apply_finish_attributes(invocation.span, invocation)
+        # Detach context and end span
+        otel_context.detach(invocation.context_token)
+        invocation.span.end()
+        return invocation
+
+    def fail_llm(  # pylint: disable=no-self-use
+        self, invocation: LLMInvocation, error: Error
+    ) -> LLMInvocation:
+        """Fail an LLM invocation and end its span with error status."""
+        invocation.end_time = time.time()
+        if invocation.context_token is None or invocation.span is None:
+            # TODO: Provide feedback that this invocation was not started
+            return invocation
+
+        _apply_error_attributes(invocation.span, error)
+        # Detach context and end span
+        otel_context.detach(invocation.context_token)
+        invocation.span.end()
+        return invocation
+
+    @contextmanager
+    def llm(
+        self, invocation: Optional[LLMInvocation] = None
+    ) -> Iterator[LLMInvocation]:
+        """Context manager for LLM invocations.
+
+        Only set data attributes on the invocation object, do not modify the span or context.
+
+        Starts the span on entry. On normal exit, finalizes the invocation and ends the span.
+        If an exception occurs inside the context, marks the span as error, ends it, and
+        re-raises the original exception.
+        """
+        if invocation is None:
+            invocation = LLMInvocation(
+                request_model="",
+            )
+        self.start_llm(invocation)
+        try:
+            yield invocation
+        except Exception as exc:
+            self.fail_llm(invocation, Error(message=str(exc), type=type(exc)))
+            raise
+        self.stop_llm(invocation)
+
+
+def get_telemetry_handler(**kwargs: Any) -> TelemetryHandler:
+    """
+    Returns a singleton TelemetryHandler instance.
+    """
+    handler: Optional[TelemetryHandler] = getattr(
+        get_telemetry_handler, "_default_handler", None
+    )
+    if handler is None:
+        handler = TelemetryHandler(**kwargs)
+        setattr(get_telemetry_handler, "_default_handler", handler)
+    return handler

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

@@ -0,0 +1,134 @@
+# 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 json
+from dataclasses import asdict
+from typing import Any, Dict, List
+
+from opentelemetry.semconv._incubating.attributes import (
+    gen_ai_attributes as GenAI,
+)
+from opentelemetry.semconv.attributes import (
+    error_attributes as ErrorAttributes,
+)
+from opentelemetry.trace import (
+    Span,
+)
+from opentelemetry.trace.status import Status, StatusCode
+from opentelemetry.util.genai.types import (
+    Error,
+    InputMessage,
+    LLMInvocation,
+    OutputMessage,
+)
+from opentelemetry.util.genai.utils import (
+    ContentCapturingMode,
+    get_content_capturing_mode,
+    is_experimental_mode,
+)
+
+
+def _apply_common_span_attributes(
+    span: Span, invocation: LLMInvocation
+) -> None:
+    """Apply attributes shared by finish() and error() and compute metrics.
+
+    Returns (genai_attributes) for use with metrics.
+    """
+    request_model = invocation.request_model
+    provider = invocation.provider
+    span.update_name(
+        f"{GenAI.GenAiOperationNameValues.CHAT.value} {request_model}".strip()
+    )
+    span.set_attribute(
+        GenAI.GEN_AI_OPERATION_NAME, GenAI.GenAiOperationNameValues.CHAT.value
+    )
+    if request_model:
+        span.set_attribute(GenAI.GEN_AI_REQUEST_MODEL, request_model)
+    if provider is not None:
+        # TODO: clean provider name to match GenAiProviderNameValues?
+        span.set_attribute(GenAI.GEN_AI_PROVIDER_NAME, provider)
+
+    finish_reasons = [gen.finish_reason for gen in invocation.output_messages]
+    if finish_reasons:
+        span.set_attribute(
+            GenAI.GEN_AI_RESPONSE_FINISH_REASONS, finish_reasons
+        )
+
+    if invocation.response_model_name is not None:
+        span.set_attribute(
+            GenAI.GEN_AI_RESPONSE_MODEL, invocation.response_model_name
+        )
+    if invocation.response_id is not None:
+        span.set_attribute(GenAI.GEN_AI_RESPONSE_ID, invocation.response_id)
+    if invocation.input_tokens is not None:
+        span.set_attribute(
+            GenAI.GEN_AI_USAGE_INPUT_TOKENS, invocation.input_tokens
+        )
+    if invocation.output_tokens is not None:
+        span.set_attribute(
+            GenAI.GEN_AI_USAGE_OUTPUT_TOKENS, invocation.output_tokens
+        )
+
+
+def _maybe_set_span_messages(
+    span: Span,
+    input_messages: List[InputMessage],
+    output_messages: List[OutputMessage],
+) -> None:
+    if not is_experimental_mode() or get_content_capturing_mode() not in (
+        ContentCapturingMode.SPAN_ONLY,
+        ContentCapturingMode.SPAN_AND_EVENT,
+    ):
+        return
+    if input_messages:
+        span.set_attribute(
+            GenAI.GEN_AI_INPUT_MESSAGES,
+            json.dumps([asdict(message) for message in input_messages]),
+        )
+    if output_messages:
+        span.set_attribute(
+            GenAI.GEN_AI_OUTPUT_MESSAGES,
+            json.dumps([asdict(message) for message in output_messages]),
+        )
+
+
+def _set_span_extra_attributes(
+    span: Span,
+    attributes: Dict[str, Any],
+) -> None:
+    for key, value in attributes.items():
+        span.set_attribute(key, value)
+
+
+def _apply_finish_attributes(span: Span, invocation: LLMInvocation) -> None:
+    """Apply attributes/messages common to finish() paths."""
+    _apply_common_span_attributes(span, invocation)
+    _maybe_set_span_messages(
+        span, invocation.input_messages, invocation.output_messages
+    )
+    _set_span_extra_attributes(span, invocation.attributes)
+
+
+def _apply_error_attributes(span: Span, error: Error) -> None:
+    """Apply status and error attributes common to error() paths."""
+    span.set_status(Status(StatusCode.ERROR, error.message))
+    if span.is_recording():
+        span.set_attribute(ErrorAttributes.ERROR_TYPE, error.type.__qualname__)
+
+
+__all__ = [
+    "_apply_finish_attributes",
+    "_apply_error_attributes",
+]