GenAI Utils | Embedding Type and Span Creation

util/opentelemetry-util-genai/CHANGELOG.md

@@ -25,6 +25,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3891](#3891))
 - Add parent class genAI invocation
   ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3889](#3889))
+- Add EmbeddingInvocation span lifecycle support
+  ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4219](#4219))
 
 ## Version 0.2b0 (2025-10-14)
 
@@ -37,8 +39,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Make inputs / outputs / system instructions optional params to `on_completion`,
   ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3802](#3802)).
   - Use a SHA256 hash of the system instructions as it's upload filename, and check
-  if the file exists before re-uploading it, ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3814](#3814)).
-
+    if the file exists before re-uploading it, ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3814](#3814)).
 
 ## Version 0.1b0 (2025-09-25)
 

util/opentelemetry-util-genai/README.rst

@@ -36,6 +36,19 @@ This package provides these span attributes:
 - `gen_ai.output.messages`: Str('[{"role": "AI", "parts": [{"content": "hello back", "type": "text"}], "finish_reason": "stop"}]')
 - `gen_ai.system_instructions`: Str('[{"content": "You are a helpful assistant.", "type": "text"}]') (when system instruction is provided)
 
+This package also supports embedding invocation spans via
+`EmbeddingInvocation` and `TelemetryHandler.start_embedding/stop_embedding/fail_embedding`.
+For embedding invocations, common attributes include:
+
+- `gen_ai.provider.name`: Str(openai)
+- `gen_ai.operation.name`: Str(embeddings)
+- `gen_ai.request.model`: Str(text-embedding-3-small)
+- `gen_ai.embeddings.dimension.count`: Int(1536)
+- `gen_ai.request.encoding_formats`: Slice(["float"])
+- `gen_ai.usage.input_tokens`: Int(24)
+- `server.address`: Str(api.openai.com)
+- `server.port`: Int(443)
+
 When `EVENT_ONLY` or `SPAN_AND_EVENT` mode is enabled and a LoggerProvider is configured, 
 the package also emits `gen_ai.client.inference.operation.details` events with structured 
 message content (as dictionaries instead of JSON strings). Note that when using `EVENT_ONLY` 

util/opentelemetry-util-genai/pyproject.toml

@@ -26,8 +26,8 @@ classifiers = [
   "Programming Language :: Python :: 3.14",
 ]
 dependencies = [
-  "opentelemetry-instrumentation ~= 0.58b0",
-  "opentelemetry-semantic-conventions ~= 0.58b0",
+  "opentelemetry-instrumentation ~= 0.60b0",
+  "opentelemetry-semantic-conventions ~= 0.60b0",
   "opentelemetry-api>=1.31.0",
 ]
 
@@ -46,10 +46,7 @@ Repository = "https://github.com/open-telemetry/opentelemetry-python-contrib"
 path = "src/opentelemetry/util/genai/version.py"
 
 [tool.hatch.build.targets.sdist]
-include = [
-  "/src",
-  "/tests",
-]
+include = ["/src", "/tests"]
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/opentelemetry"]

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

@@ -62,7 +62,7 @@
 
 import timeit
 from contextlib import contextmanager
-from typing import Iterator
+from typing import Iterator, TypeVar
 
 from opentelemetry import context as otel_context
 from opentelemetry._logs import (
@@ -78,15 +78,24 @@
     get_tracer,
     set_span_in_context,
 )
+from opentelemetry.trace.status import Status, StatusCode
 from opentelemetry.util.genai.metrics import InvocationMetricsRecorder
 from opentelemetry.util.genai.span_utils import (
+    _apply_embedding_finish_attributes,
     _apply_error_attributes,
     _apply_llm_finish_attributes,
     _maybe_emit_llm_event,
 )
-from opentelemetry.util.genai.types import Error, LLMInvocation
+from opentelemetry.util.genai.types import (
+    EmbeddingInvocation,
+    Error,
+    GenAIInvocation,
+    LLMInvocation,
+)
 from opentelemetry.util.genai.version import __version__
 
+_T = TypeVar("_T", bound=GenAIInvocation)
+
 
 class TelemetryHandler:
     """
@@ -131,14 +140,22 @@ def _record_llm_metrics(
             error_type=error_type,
         )
 
-    def start_llm(
+    def _record_embedding_metrics(
         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
+        invocation: EmbeddingInvocation,
+        span: Span | None = None,
+        *,
+        error_type: str | None = None,
+    ) -> None:
+        # Metrics recorder currently supports LLMInvocation fields only.
+        # Keep embedding metrics as a no-op until dedicated embedding
+        # metric support is added.
+        return
+
+    def _start(self, invocation: _T) -> _T:
+        """Start a GenAI invocation and create a pending span entry."""
         span = self._tracer.start_span(
-            name=f"{invocation.operation_name} {invocation.request_model}",
+            name=invocation.operation_name or "",
             kind=SpanKind.CLIENT,
         )
         # Record a monotonic start timestamp (seconds) for duration
@@ -150,40 +167,108 @@ def start_llm(
         )
         return invocation
 
-    def stop_llm(self, invocation: LLMInvocation) -> LLMInvocation:  # pylint: disable=no-self-use
-        """Finalize an LLM invocation successfully and end its span."""
+    def _stop(self, invocation: _T) -> _T:
+        """Finalize a GenAI invocation successfully and end its span."""
         if invocation.context_token is None or invocation.span is None:
             # TODO: Provide feedback that this invocation was not started
             return invocation
 
         span = invocation.span
-        _apply_llm_finish_attributes(span, invocation)
-        self._record_llm_metrics(invocation, span)
-        _maybe_emit_llm_event(self._logger, span, invocation)
-        # Detach context and end span
-        otel_context.detach(invocation.context_token)
-        span.end()
+        try:
+            if isinstance(invocation, LLMInvocation):
+                _apply_llm_finish_attributes(span, invocation)
+                self._record_llm_metrics(invocation, span)
+                _maybe_emit_llm_event(self._logger, span, invocation)
+            elif isinstance(invocation, EmbeddingInvocation):
+                _apply_embedding_finish_attributes(span, invocation)
+                self._record_embedding_metrics(invocation, span)
+            else:
+                span.set_status(
+                    Status(
+                        StatusCode.ERROR,
+                        f"Unsupported invocation type: {type(invocation)!r}",
+                    )
+                )
+                raise TypeError(
+                    f"Unsupported invocation type: {type(invocation)!r}"
+                )
+        finally:
+            # Detach context and end span even if finishing fails
+            otel_context.detach(invocation.context_token)
+            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."""
+    def _fail(self, invocation: _T, error: Error) -> _T:
+        """Fail a GenAI invocation and end its span with error status."""
         if invocation.context_token is None or invocation.span is None:
             # TODO: Provide feedback that this invocation was not started
             return invocation
 
         span = invocation.span
-        _apply_llm_finish_attributes(invocation.span, invocation)
-        _apply_error_attributes(invocation.span, error)
-        error_type = getattr(error.type, "__qualname__", None)
-        self._record_llm_metrics(invocation, span, error_type=error_type)
-        _maybe_emit_llm_event(self._logger, span, invocation, error)
-        # Detach context and end span
-        otel_context.detach(invocation.context_token)
-        span.end()
+        try:
+            if isinstance(invocation, LLMInvocation):
+                _apply_llm_finish_attributes(span, invocation)
+                _apply_error_attributes(span, error)
+                error_type = getattr(error.type, "__qualname__", None)
+                self._record_llm_metrics(
+                    invocation, span, error_type=error_type
+                )
+                _maybe_emit_llm_event(self._logger, span, invocation, error)
+            elif isinstance(invocation, EmbeddingInvocation):
+                _apply_embedding_finish_attributes(span, invocation)
+                _apply_error_attributes(span, error)
+                error_type = getattr(error.type, "__qualname__", None)
+                self._record_embedding_metrics(
+                    invocation, span, error_type=error_type
+                )
+            else:
+                span.set_status(
+                    Status(
+                        StatusCode.ERROR,
+                        f"Unsupported invocation type: {type(invocation)!r}",
+                    )
+                )
+                raise TypeError(
+                    f"Unsupported invocation type: {type(invocation)!r}"
+                )
+        finally:
+            # Detach context and end span even if finishing fails
+            otel_context.detach(invocation.context_token)
+            span.end()
         return invocation
 
+    def start(
+        self,
+        invocation: _T,
+    ) -> _T:
+        """Start a GenAI invocation and create a pending span entry."""
+        return self._start(invocation)
+
+    def stop(self, invocation: _T) -> _T:
+        """Finalize a GenAI invocation successfully and end its span."""
+        return self._stop(invocation)
+
+    def fail(self, invocation: _T, error: Error) -> _T:
+        """Fail a GenAI invocation and end its span with error status."""
+        return self._fail(invocation, error)
+
+    def start_llm(
+        self,
+        invocation: LLMInvocation,
+    ) -> LLMInvocation:
+        """Start an LLM invocation and create a pending span entry."""
+        return self.start(invocation)
+
+    def stop_llm(self, invocation: LLMInvocation) -> LLMInvocation:
+        """Finalize an LLM invocation successfully and end its span."""
+        return self.stop(invocation)
+
+    def fail_llm(
+        self, invocation: LLMInvocation, error: Error
+    ) -> LLMInvocation:
+        """Fail an LLM invocation and end its span with error status."""
+        return self.fail(invocation, error)
+
     @contextmanager
     def llm(
         self, invocation: LLMInvocation | None = None
@@ -208,6 +293,48 @@ def llm(
             raise
         self.stop_llm(invocation)
 
+    @contextmanager
+    def embedding(
+        self, invocation: EmbeddingInvocation | None = None
+    ) -> Iterator[EmbeddingInvocation]:
+        """Context manager for Embedding 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 = EmbeddingInvocation()
+        self.start_embedding(invocation)
+        try:
+            yield invocation
+        except Exception as exc:
+            self.fail_embedding(
+                invocation, Error(message=str(exc), type=type(exc))
+            )
+            raise
+        self.stop_embedding(invocation)
+
+    def start_embedding(
+        self, invocation: EmbeddingInvocation
+    ) -> EmbeddingInvocation:
+        """Start an embedding invocation and create a pending span entry."""
+        return self.start(invocation)
+
+    def stop_embedding(
+        self, invocation: EmbeddingInvocation
+    ) -> EmbeddingInvocation:
+        """Finalize an embedding invocation successfully and end its span."""
+        return self.stop(invocation)
+
+    def fail_embedding(
+        self, invocation: EmbeddingInvocation, error: Error
+    ) -> EmbeddingInvocation:
+        """Fail an embedding invocation and end its span with error status."""
+        return self.fail(invocation, error)
+
 
 def get_telemetry_handler(
     tracer_provider: TracerProvider | None = None,