feat(conventions): added otel v1.37 conventions mapping

Author: poshinchen

src/strands_evals/mappers/__init__.py

@@ -1,6 +1,6 @@
 """Converters for transforming telemetry data to Session format."""
 
 from .session_mapper import SessionMapper
-from .strands_in_memory_session_mapper import StrandsInMemorySessionMapper
+from .strands_in_memory_session_mapper import GenAIConventionVersion, StrandsInMemorySessionMapper
 
-__all__ = ["SessionMapper", "StrandsInMemorySessionMapper"]
+__all__ = ["GenAIConventionVersion", "SessionMapper", "StrandsInMemorySessionMapper"]

src/strands_evals/mappers/strands_in_memory_session_mapper.py

@@ -2,6 +2,7 @@
 import logging
 from collections import defaultdict
 from datetime import datetime, timezone
+from enum import Enum
 from typing import Any
 
 from opentelemetry.sdk.trace import ReadableSpan
@@ -27,10 +28,52 @@
 logger = logging.getLogger(__name__)
 
 
+class GenAIConventionVersion(Enum):
+    """GenAI semantic convention versions following OTEL_SEMCONV_STABILITY_OPT_IN.
+
+    This enum aligns with OpenTelemetry's semantic convention stability options
+    as defined in OTEL_SEMCONV_STABILITY_OPT_IN environment variable.
+
+    Attributes:
+        LEGACY: Use legacy conventions (v1.36.0 or prior) with gen_ai.system attribute
+            and separate message events (gen_ai.user.message, gen_ai.choice, etc.)
+        LATEST_EXPERIMENTAL: Use latest experimental conventions (v1.37+) with
+            gen_ai.provider.name attribute and unified gen_ai.client.inference.operation.details events.
+            Corresponds to OTEL's "gen_ai_latest_experimental" stability option.
+        AUTO: Automatically detect convention version from span attributes.
+            Checks for presence of gen_ai.provider.name to determine version.
+    """
+
+    LEGACY = "legacy"
+    LATEST_EXPERIMENTAL = "gen_ai_latest_experimental"
+    AUTO = "auto"
+
+
 class StrandsInMemorySessionMapper(SessionMapper):
-    """Maps OpenTelemetry in-memory spans to Session format for evaluation."""
+    """Maps OpenTelemetry in-memory spans to Session format for evaluation.
+
+    Supports both legacy and latest GenAI semantic conventions:
+    - Latest (v1.37+): gen_ai.provider.name with unified gen_ai.client.inference.operation.details events
+    - Legacy: gen_ai.system with separate message events (gen_ai.user.message, gen_ai.choice, etc.)
+
+    The mapper can auto-detect the convention version or accept an explicit setting. Default to Legacy.
+    """
+
+    def __init__(self, convention_version: GenAIConventionVersion = GenAIConventionVersion.AUTO):
+        """Initialize the mapper.
+
+        Args:
+            convention_version: GenAI semantic convention version to use.
+                Defaults to AUTO which auto-detects from span attributes.
+        """
+        self.convention_version = convention_version
 
     def map_to_session(self, otel_spans: list[ReadableSpan], session_id: str) -> Session:
+        # Auto-detect convention version if set to AUTO
+        if otel_spans and self.convention_version == GenAIConventionVersion.AUTO:
+            detected_version = self._detect_convention_version(otel_spans[0])
+            self.convention_version = detected_version
+
         traces_by_id = defaultdict(list)
         for span in otel_spans:
             trace_id = format(span.context.trace_id, "032x")
@@ -44,6 +87,26 @@ def map_to_session(self, otel_spans: list[ReadableSpan], session_id: str) -> Ses
 
         return Session(traces=traces, session_id=session_id)
 
+    def _detect_convention_version(self, span: ReadableSpan) -> GenAIConventionVersion:
+        """Detect which GenAI semantic convention version is being used.
+
+        Returns:
+            GenAIConventionVersion.LATEST_EXPERIMENTAL if using latest conventions,
+            GenAIConventionVersion.LEGACY otherwise
+        """
+        if span.attributes and "gen_ai.provider.name" in span.attributes:
+            return GenAIConventionVersion.LATEST_EXPERIMENTAL
+
+        return GenAIConventionVersion.LEGACY
+
+    def _use_latest_conventions(self) -> bool:
+        """Helper method to determine if latest conventions should be used.
+
+        Returns:
+            True if LATEST_EXPERIMENTAL, False if LEGACY
+        """
+        return self.convention_version == GenAIConventionVersion.LATEST_EXPERIMENTAL
+
     def _convert_trace(self, trace_id: str, otel_spans: list[ReadableSpan], session_id: str) -> Trace:
         converted_spans: list[InferenceSpan | ToolExecutionSpan | AgentInvocationSpan] = []
 
@@ -126,6 +189,16 @@ def _process_tool_results(self, content_list: list[dict[str, Any]]) -> list[Text
 
     def _convert_inference_span(self, span: ReadableSpan, session_id: str) -> InferenceSpan:
         span_info = self._create_span_info(span, session_id)
+
+        if self._use_latest_conventions():
+            messages = self._extract_messages_from_inference_details(span)
+        else:
+            messages = self._extract_messages_from_events(span)
+
+        return InferenceSpan(span_info=span_info, messages=messages, metadata={})
+
+    def _extract_messages_from_events(self, span: ReadableSpan) -> list[UserMessage | AssistantMessage]:
+        """Extract messages from legacy event format (gen_ai.user.message, etc.)."""
         messages: list[UserMessage | AssistantMessage] = []
 
         for event in span.events:
@@ -156,29 +229,160 @@ def _convert_inference_span(self, span: ReadableSpan, session_id: str) -> Infere
             except Exception as e:
                 logger.warning(f"Failed to process event {event.name}: {e}")
 
-        return InferenceSpan(span_info=span_info, messages=messages, metadata={})
+        return messages
+
+    def _extract_messages_from_inference_details(self, span: ReadableSpan) -> list[UserMessage | AssistantMessage]:
+        """Extract messages from latest event format (gen_ai.client.inference.operation.details)."""
+        messages: list[UserMessage | AssistantMessage] = []
+
+        for event in span.events:
+            try:
+                if event.name == "gen_ai.client.inference.operation.details":
+                    event_attributes = event.attributes
+                    if not event_attributes:
+                        continue
+                    # Check for input messages
+                    if "gen_ai.input.messages" in event_attributes:
+                        input_messages = self._parse_json_attr(event_attributes, "gen_ai.input.messages")
+                        for msg in input_messages:
+                            input_content = self._convert_inference_messages(msg)
+                            if input_content:
+                                messages.append(input_content)
+
+                    # Check for output messages
+                    if "gen_ai.output.messages" in event_attributes:
+                        output_messages = self._parse_json_attr(event_attributes, "gen_ai.output.messages")
+                        for msg in output_messages:
+                            output_content = self._convert_inference_messages(msg)
+                            if output_content:
+                                messages.append(output_content)
+            except Exception as e:
+                logger.warning(f"Failed to process inference details event: {e}")
+
+        return messages
+
+    def _convert_inference_messages(self, otel_msg: dict[str, Any]) -> UserMessage | AssistantMessage | None:
+        """Convert OTEL message format (with parts) to internal message types.
+
+        Args:
+            otel_msg: Message in OTEL format with 'role' and 'parts' fields
+
+        Returns:
+            UserMessage or AssistantMessage, or None if conversion fails
+        """
+        try:
+            role = otel_msg.get("role", "")
+            parts = otel_msg.get("parts", [])
+
+            if role == "assistant":
+                assistant_content: list[TextContent | ToolCallContent] = []
+
+                for part in parts:
+                    part_type = part.get("type", "")
+
+                    if part_type == "text":
+                        assistant_content.append(TextContent(text=part.get("content", "")))
+
+                    elif part_type == "tool_call":
+                        assistant_content.append(
+                            ToolCallContent(
+                                name=part.get("name", ""),
+                                arguments=part.get("arguments", {}),
+                                tool_call_id=part.get("id"),
+                            )
+                        )
+                return AssistantMessage(content=assistant_content) if assistant_content else None
+
+            # Tool messages are represented as UserMessage with ToolResultContent
+            content: list[TextContent | ToolResultContent] = []
+
+            for part in parts:
+                part_type = part.get("type", "")
+
+                if part_type == "text":
+                    content.append(TextContent(text=part.get("content", "")))
+
+                if part_type == "tool_call_response":
+                    # Extract text from response array if present
+                    response = part.get("response", [])
+                    response_text = ""
+
+                    ## To-do: Compare the differences for multiple toolResults
+                    if isinstance(response, list) and response:
+                        response_text = (
+                            response[0].get("text", "") if isinstance(response[0], dict) else str(response[0])
+                        )
+                    elif isinstance(response, str):
+                        response_text = response
+
+                    content.append(
+                        ToolResultContent(
+                            content=response_text,
+                            tool_call_id=part.get("id"),
+                        )
+                    )
+            return UserMessage(content=content) if content else None
+
+        except Exception as e:
+            logger.warning(f"Failed to convert OTEL message: {e}")
+            return None
 
     def _convert_tool_execution_span(self, span: ReadableSpan, session_id: str) -> ToolExecutionSpan:
         span_info = self._create_span_info(span, session_id)
         attrs = span.attributes or {}
 
         tool_name = str(attrs.get("gen_ai.tool.name", ""))
         tool_call_id = str(attrs.get("gen_ai.tool.call.id", ""))
-        tool_status = attrs.get("tool.status", "")
+        tool_status = attrs.get("gen_ai.tool.status", attrs.get("tool.status", ""))
         tool_error = None if tool_status == "success" else (str(tool_status) if tool_status else None)
 
         tool_arguments = {}
         tool_result_content = ""
 
-        for event in span.events:
-            try:
-                if event.name == "gen_ai.tool.message":
-                    tool_arguments = self._parse_json_attr(event.attributes, "content", "{}")
-                elif event.name == "gen_ai.choice":
-                    message_list = self._parse_json_attr(event.attributes, "message")
-                    tool_result_content = message_list[0].get("text", "") if message_list else ""
-            except Exception as e:
-                logger.warning(f"Failed to process tool event {event.name}: {e}")
+        if self._use_latest_conventions():
+            # Extract from gen_ai.client.inference.operation.details events
+            for event in span.events:
+                try:
+                    if event.name == "gen_ai.client.inference.operation.details":
+                        event_attributes = event.attributes
+                        if not event_attributes:
+                            continue
+                        if "gen_ai.input.messages" in event_attributes:
+                            input_messages = self._parse_json_attr(event_attributes, "gen_ai.input.messages")
+                            if input_messages and input_messages[0].get("parts"):
+                                part = input_messages[0]["parts"][0]
+                                if part.get("type") == "tool_call":
+                                    tool_arguments = part.get("arguments", {})
+
+                        if "gen_ai.output.messages" in event_attributes:
+                            output_messages = self._parse_json_attr(event_attributes, "gen_ai.output.messages")
+                            if output_messages and output_messages[0].get("parts"):
+                                part = output_messages[0]["parts"][0]
+                                if part.get("type") == "tool_call_response":
+                                    response = part.get("response", [])
+                                    if isinstance(response, list) and response:
+                                        tool_result_content = (
+                                            response[0].get("text", "")
+                                            if isinstance(response[0], dict)
+                                            else str(response[0])
+                                        )
+                                    elif isinstance(response, str):
+                                        tool_result_content = response
+                except Exception as e:
+                    logger.warning(f"Failed to process tool event {event.name}: {e}")
+        else:
+            for event in span.events:
+                try:
+                    event_attributes = event.attributes
+                    if not event_attributes:
+                        continue
+                    if event.name == "gen_ai.tool.message":
+                        tool_arguments = self._parse_json_attr(event_attributes, "content", "{}")
+                    elif event.name == "gen_ai.choice":
+                        message_list = self._parse_json_attr(event_attributes, "message")
+                        tool_result_content = message_list[0].get("text", "") if message_list else ""
+                except Exception as e:
+                    logger.warning(f"Failed to process tool event {event.name}: {e}")
 
         tool_call = ToolCall(name=tool_name, arguments=tool_arguments, tool_call_id=tool_call_id)
         tool_result = ToolResult(content=tool_result_content, error=tool_error, tool_call_id=tool_call_id)
@@ -198,16 +402,46 @@ def _convert_agent_invocation_span(self, span: ReadableSpan, session_id: str) ->
         except Exception as e:
             logger.warning(f"Failed to parse available tools: {e}")
 
-        for event in span.events:
-            try:
-                if event.name == "gen_ai.user.message":
-                    content_list = self._parse_json_attr(event.attributes, "content")
-                    user_prompt = content_list[0].get("text", "") if content_list else ""
-                elif event.name == "gen_ai.choice":
-                    msg = event.attributes.get("message", "") if event.attributes else ""
-                    agent_response = str(msg)
-            except Exception as e:
-                logger.warning(f"Failed to process agent event {event.name}: {e}")
+        if self._use_latest_conventions():
+            for event in span.events:
+                try:
+                    if event.name == "gen_ai.client.inference.operation.details":
+                        event_attributes = event.attributes
+                        if not event_attributes:
+                            continue
+                        if "gen_ai.input.messages" in event_attributes:
+                            input_messages = self._parse_json_attr(event_attributes, "gen_ai.input.messages")
+                            if input_messages and input_messages[0].get("parts"):
+                                parts = input_messages[0]["parts"]
+                                for part in parts:
+                                    if part.get("type") == "text":
+                                        user_prompt = part.get("content", "")
+                                        break
+
+                        if "gen_ai.output.messages" in event_attributes:
+                            output_messages = self._parse_json_attr(event_attributes, "gen_ai.output.messages")
+                            if output_messages and output_messages[0].get("parts"):
+                                parts = output_messages[0]["parts"]
+                                for part in parts:
+                                    if part.get("type") == "text":
+                                        agent_response = part.get("content", "")
+                                        break
+                except Exception as e:
+                    logger.warning(f"Failed to process agent event {event.name}: {e}")
+        else:
+            for event in span.events:
+                try:
+                    event_attributes = event.attributes
+                    if not event_attributes:
+                        continue
+                    if event.name == "gen_ai.user.message":
+                        content_list = self._parse_json_attr(event_attributes, "content")
+                        user_prompt = content_list[0].get("text", "") if content_list else ""
+                    elif event.name == "gen_ai.choice":
+                        msg = event_attributes.get("message", "") if event_attributes else ""
+                        agent_response = str(msg)
+                except Exception as e:
+                    logger.warning(f"Failed to process agent event {event.name}: {e}")
 
         return AgentInvocationSpan(
             span_info=span_info,

src/strands_evals/types/trace.py

@@ -8,7 +8,7 @@
 from enum import Enum
 
 from pydantic import BaseModel, field_serializer
-from typing_extensions import TypeAlias, Union
+from typing_extensions import Mapping, Sequence, TypeAlias, Union
 
 
 class Role(str, Enum):
@@ -191,3 +191,10 @@ class EvaluatorResult(BaseModel):
 
 class EvaluationResponse(BaseModel):
     evaluator_results: list[EvaluatorResult]
+
+
+AttributeValue = Mapping[
+    str, str | bool | int | float | Sequence[str] | Sequence[bool] | Sequence[int] | Sequence[float]
+]
+
+Attributes = Mapping[str, AttributeValue] | None