fix(genai-utils): use AttributeValue instead of Any for attributes/metric_attributes

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

@@ -3,8 +3,6 @@
 
 from __future__ import annotations
 
-from typing import Any
-
 from opentelemetry._logs import Logger
 from opentelemetry.semconv._incubating.attributes import (
     gen_ai_attributes as GenAI,
@@ -24,6 +22,7 @@
     OutputMessage,
     ToolDefinition,
 )
+from opentelemetry.util.types import AttributeValue
 
 
 class AgentInvocation(GenAIInvocation):
@@ -101,7 +100,7 @@ def __init__(
 
         self._start(self._get_base_attributes())
 
-    def _get_base_attributes(self) -> dict[str, Any]:
+    def _get_base_attributes(self) -> dict[str, AttributeValue]:
         """Return sampling-relevant attributes available at span creation time."""
         optional_attrs = (
             (GenAI.GEN_AI_PROVIDER_NAME, self.provider),
@@ -115,7 +114,7 @@ def _get_base_attributes(self) -> dict[str, Any]:
             **{k: v for k, v in optional_attrs if v is not None},
         }
 
-    def _get_common_attributes(self) -> dict[str, Any]:
+    def _get_common_attributes(self) -> dict[str, AttributeValue]:
         optional_attrs = (
             (GenAI.GEN_AI_PROVIDER_NAME, self.provider),
             (GenAI.GEN_AI_REQUEST_MODEL, self.request_model),
@@ -131,7 +130,7 @@ def _get_common_attributes(self) -> dict[str, Any]:
             **{k: v for k, v in optional_attrs if v is not None},
         }
 
-    def _get_request_attributes(self) -> dict[str, Any]:
+    def _get_request_attributes(self) -> dict[str, AttributeValue]:
         optional_attrs = (
             (GenAI.GEN_AI_CONVERSATION_ID, self.conversation_id),
             (GenAI.GEN_AI_DATA_SOURCE_ID, self.data_source_id),
@@ -147,12 +146,12 @@ def _get_request_attributes(self) -> dict[str, Any]:
         )
         return {k: v for k, v in optional_attrs if v is not None}
 
-    def _get_response_attributes(self) -> dict[str, Any]:
+    def _get_response_attributes(self) -> dict[str, AttributeValue]:
         if self.finish_reasons:
             return {GenAI.GEN_AI_RESPONSE_FINISH_REASONS: self.finish_reasons}
         return {}
 
-    def _get_usage_attributes(self) -> dict[str, Any]:
+    def _get_usage_attributes(self) -> dict[str, AttributeValue]:
         optional_attrs = (
             (GenAI.GEN_AI_USAGE_INPUT_TOKENS, self.input_tokens),
             (GenAI.GEN_AI_USAGE_OUTPUT_TOKENS, self.output_tokens),
@@ -167,7 +166,7 @@ def _get_usage_attributes(self) -> dict[str, Any]:
         )
         return {k: v for k, v in optional_attrs if v is not None}
 
-    def _get_content_attributes_for_span(self) -> dict[str, Any]:
+    def _get_content_attributes_for_span(self) -> dict[str, AttributeValue]:
         return get_content_attributes(
             input_messages=self.input_messages,
             output_messages=self.output_messages,
@@ -176,14 +175,14 @@ def _get_content_attributes_for_span(self) -> dict[str, Any]:
             for_span=True,
         )
 
-    def _get_metric_attributes(self) -> dict[str, Any]:
+    def _get_metric_attributes(self) -> dict[str, AttributeValue]:
         optional_attrs = (
             (GenAI.GEN_AI_PROVIDER_NAME, self.provider),
             (GenAI.GEN_AI_REQUEST_MODEL, self.request_model),
             (server_attributes.SERVER_ADDRESS, self.server_address),
             (server_attributes.SERVER_PORT, self.server_port),
         )
-        attrs: dict[str, Any] = {
+        attrs: dict[str, AttributeValue] = {
             GenAI.GEN_AI_OPERATION_NAME: self._operation_name,
             **{k: v for k, v in optional_attrs if v is not None},
         }
@@ -204,7 +203,7 @@ def _apply_finish(self, error: Error | None = None) -> None:
         if error is not None:
             self._apply_error_attributes(error)
 
-        attributes: dict[str, Any] = {}
+        attributes: dict[str, AttributeValue] = {}
         attributes.update(self._get_common_attributes())
         attributes.update(self._get_request_attributes())
         attributes.update(self._get_response_attributes())

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

@@ -3,8 +3,6 @@
 
 from __future__ import annotations
 
-from typing import Any
-
 from opentelemetry._logs import Logger
 from opentelemetry.semconv._incubating.attributes import (
     gen_ai_attributes as GenAI,
@@ -60,7 +58,7 @@ def __init__(
         self.response_model_name: str | None = None
         self._start(self._get_base_attributes())
 
-    def _get_base_attributes(self) -> dict[str, Any]:
+    def _get_base_attributes(self) -> dict[str, AttributeValue]:
         """Return sampling-relevant attributes available at span creation time."""
         optional_attrs = (
             (GenAI.GEN_AI_REQUEST_MODEL, self.request_model),
@@ -73,7 +71,7 @@ def _get_base_attributes(self) -> dict[str, Any]:
             **{k: v for k, v in optional_attrs if v is not None},
         }
 
-    def _get_metric_attributes(self) -> dict[str, Any]:
+    def _get_metric_attributes(self) -> dict[str, AttributeValue]:
         optional_attrs = (
             (GenAI.GEN_AI_PROVIDER_NAME, self.provider),
             (GenAI.GEN_AI_REQUEST_MODEL, self.request_model),
@@ -104,7 +102,7 @@ def _apply_finish(self, error: Error | None = None) -> None:
             (GenAI.GEN_AI_RESPONSE_MODEL, self.response_model_name),
             (GenAI.GEN_AI_USAGE_INPUT_TOKENS, self.input_tokens),
         )
-        attributes: dict[str, Any] = {
+        attributes: dict[str, AttributeValue] = {
             GenAI.GEN_AI_OPERATION_NAME: self._operation_name,
             **{
                 key: value

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

@@ -4,7 +4,6 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Any
 
 from opentelemetry._logs import Logger, LogRecord
 from opentelemetry.semconv._incubating.attributes import (
@@ -28,6 +27,7 @@
 from opentelemetry.util.genai.utils import (
     should_emit_event,
 )
+from opentelemetry.util.types import AttributeValue
 
 # TODO: Migrate to GenAI constants once available in semconv package
 _GEN_AI_REASONING_OUTPUT_TOKENS = "gen_ai.usage.reasoning.output_tokens"
@@ -97,7 +97,7 @@ def __init__(
         self.output_type: str | None = None
         self._start(self._get_base_attributes())
 
-    def _get_message_attributes(self, *, for_span: bool) -> dict[str, Any]:
+    def _get_message_attributes(self, *, for_span: bool) -> dict[str, AttributeValue]:
         return get_content_attributes(
             input_messages=self.input_messages,
             output_messages=self.output_messages,
@@ -118,7 +118,7 @@ def _get_finish_reasons(self) -> list[str] | None:
             return reasons or None
         return None
 
-    def _get_base_attributes(self) -> dict[str, Any]:
+    def _get_base_attributes(self) -> dict[str, AttributeValue]:
         optional_attrs = (
             (GenAI.GEN_AI_REQUEST_MODEL, self.request_model),
             (GenAI.GEN_AI_PROVIDER_NAME, self.provider),
@@ -130,7 +130,7 @@ def _get_base_attributes(self) -> dict[str, Any]:
             **{k: v for k, v in optional_attrs if v is not None},
         }
 
-    def _get_attributes(self) -> dict[str, Any]:
+    def _get_attributes(self) -> dict[str, AttributeValue]:
         attrs = self._get_base_attributes()
         if self.output_tokens is None and self.thinking_tokens is None:
             output_tokens = None
@@ -170,7 +170,7 @@ def _get_attributes(self) -> dict[str, Any]:
         attrs.update({k: v for k, v in optional_attrs if v is not None})
         return attrs
 
-    def _get_metric_attributes(self) -> dict[str, Any]:
+    def _get_metric_attributes(self) -> dict[str, AttributeValue]:
         attrs = self._get_base_attributes()
         if self.response_model_name is not None:
             attrs[GenAI.GEN_AI_RESPONSE_MODEL] = self.response_model_name
@@ -243,9 +243,9 @@ class LLMInvocation:
     finish_reasons: list[str] | None = None
     input_tokens: int | None = None
     output_tokens: int | None = None
-    attributes: dict[str, Any] = field(default_factory=dict)  # pyright: ignore[reportUnknownVariableType]
+    attributes: dict[str, AttributeValue] = field(default_factory=dict)  # pyright: ignore[reportUnknownVariableType]
     """Additional attributes to set on spans and/or events. Not set on metrics."""
-    metric_attributes: dict[str, Any] = field(default_factory=dict)  # pyright: ignore[reportUnknownVariableType]
+    metric_attributes: dict[str, AttributeValue] = field(default_factory=dict)  # pyright: ignore[reportUnknownVariableType]
     """Additional attributes to set on metrics. Must be low cardinality. Not set on spans or events."""
     temperature: float | None = None
     top_p: float | None = None
@@ -331,4 +331,4 @@ def span(self) -> Span:
             self._inference_invocation.span
             if self._inference_invocation is not None
             else INVALID_SPAN
-        )
+        )

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

@@ -33,6 +33,7 @@
     gen_ai_json_dumps,
     get_content_capturing_mode,
 )
+from opentelemetry.util.types import AttributeValue
 
 if TYPE_CHECKING:
     from opentelemetry.util.genai.metrics import InvocationMetricsRecorder
@@ -61,19 +62,19 @@ def __init__(
         operation_name: str,
         span_name: str,
         span_kind: SpanKind = SpanKind.CLIENT,
-        attributes: dict[str, Any] | None = None,
-        metric_attributes: dict[str, Any] | None = None,
+        attributes: dict[str, AttributeValue] | None = None,
+        metric_attributes: dict[str, AttributeValue] | None = None,
     ) -> None:
         self._tracer = tracer
         self._metrics_recorder = metrics_recorder
         self._logger = logger
         self._completion_hook = completion_hook
         self._operation_name: str = operation_name
-        self.attributes: dict[str, Any] = (
+        self.attributes: dict[str, AttributeValue] = (
             {} if attributes is None else attributes
         )
         """Additional attributes to set on spans and/or events. Not set on metrics."""
-        self.metric_attributes: dict[str, Any] = (
+        self.metric_attributes: dict[str, AttributeValue] = (
             {} if metric_attributes is None else metric_attributes
         )
         """Additional attributes to set on metrics. Must be low cardinality. Not set on spans or events."""
@@ -84,7 +85,7 @@ def __init__(
         self._context_token: ContextToken | None = None
         self._monotonic_start_s: float | None = None
 
-    def _start(self, attributes: dict[str, Any] | None = None) -> None:
+    def _start(self, attributes: dict[str, AttributeValue] | None = None) -> None:
         """Start the invocation span and attach it to the current context.
 
         Args:
@@ -99,7 +100,7 @@ def _start(self, attributes: dict[str, Any] | None = None) -> None:
         self._monotonic_start_s = timeit.default_timer()
         self._context_token = attach(self._span_context)
 
-    def _get_metric_attributes(self) -> dict[str, Any]:
+    def _get_metric_attributes(self) -> dict[str, AttributeValue]:
         """Return low-cardinality attributes for metric recording."""
         return self.metric_attributes
 

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

@@ -3,8 +3,6 @@
 
 from __future__ import annotations
 
-from typing import Any
-
 from opentelemetry._logs import Logger
 from opentelemetry.semconv._incubating.attributes import (
     gen_ai_attributes as GenAI,
@@ -74,7 +72,7 @@ def __init__(
         self.tool_description = tool_description
         self._start(self._get_base_attributes())
 
-    def _get_base_attributes(self) -> dict[str, Any]:
+    def _get_base_attributes(self) -> dict[str, AttributeValue]:
         """Return sampling-relevant attributes available at span creation time."""
         optional_attrs = (
             (GenAI.GEN_AI_TOOL_NAME, self.name),
@@ -87,8 +85,8 @@ def _get_base_attributes(self) -> dict[str, Any]:
             **{k: v for k, v in optional_attrs if v is not None},
         }
 
-    def _get_metric_attributes(self) -> dict[str, Any]:
-        attrs: dict[str, Any] = {
+    def _get_metric_attributes(self) -> dict[str, AttributeValue]:
+        attrs: dict[str, AttributeValue] = {
             GenAI.GEN_AI_OPERATION_NAME: self._operation_name,
         }
         attrs.update(self.metric_attributes)
@@ -115,7 +113,7 @@ def _apply_finish(self, error: Error | None = None) -> None:
                 else None,
             ),
         )
-        attributes: dict[str, Any] = {
+        attributes: dict[str, AttributeValue] = {
             GenAI.GEN_AI_OPERATION_NAME: self._operation_name,
             **{k: v for k, v in optional_attrs if v is not None},
         }

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

@@ -4,7 +4,6 @@
 from __future__ import annotations
 
 from dataclasses import asdict
-from typing import Any
 
 from opentelemetry._logs import Logger
 from opentelemetry.semconv._incubating.attributes import (
@@ -22,6 +21,7 @@
     gen_ai_json_dumps,
     should_capture_content_on_spans,
 )
+from opentelemetry.util.types import AttributeValue
 
 
 class WorkflowInvocation(GenAIInvocation):
@@ -57,14 +57,14 @@ def __init__(
         self.output_messages: list[OutputMessage] = []
         self._start(self._get_base_attributes())
 
-    def _get_base_attributes(self) -> dict[str, Any]:
+    def _get_base_attributes(self) -> dict[str, AttributeValue]:
         """Return sampling-relevant attributes available at span creation time."""
-        attrs: dict[str, Any] = {
+        attrs: dict[str, AttributeValue] = {
             GenAI.GEN_AI_OPERATION_NAME: self._operation_name,
         }
         return attrs
 
-    def _get_messages_for_span(self) -> dict[str, Any]:
+    def _get_messages_for_span(self) -> dict[str, AttributeValue]:
         if not should_capture_content_on_spans():
             return {}
         optional_attrs = (
@@ -86,8 +86,8 @@ def _get_messages_for_span(self) -> dict[str, Any]:
         }
 
     def _apply_finish(self, error: Error | None = None) -> None:
-        attributes: dict[str, Any] = {
-            GenAI.GEN_AI_OPERATION_NAME: self._operation_name
+        attributes: dict[str, AttributeValue] = {
+            GenAI.GEN_AI_OPERATION_NAME: self._operation_name,
         }
         attributes.update(self._get_messages_for_span())
         if error is not None:
@@ -98,4 +98,4 @@ def _apply_finish(self, error: Error | None = None) -> None:
             inputs=self.input_messages,
             outputs=self.output_messages,
         )
-        # TODO: Add workflow metrics when supported
+        # TODO: Add workflow metrics when supported