Fix DeepEvalAdapter to extract multi-turn conversations from service-normalized SESSION format

Author: ybdarrenwang

src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/registry.py

@@ -125,15 +125,63 @@ def map_spans(
     return result
 
 
+def _extract_message_text(messages: List[Dict[str, Any]]) -> Optional[str]:
+    """Extract text content from service message format.
+
+    Handles the nested structure: [{role: ..., content: {content: [{text: ...}]}}]
+    as well as the variant: [{role: ..., content: {message: [{text: ...}]}}]
+    """
+    for msg in messages:
+        content = msg.get("content", msg.get("message", {}))
+        if isinstance(content, dict):
+            # Unwrap nested content/message key
+            content = content.get("content", content.get("message", []))
+        if isinstance(content, list):
+            text = " ".join(c.get("text", "") for c in content if isinstance(c, dict)).strip()
+            if text:
+                return text
+        elif isinstance(content, str) and content.strip():
+            return content.strip()
+    return None
+
+
 def _extract_from_service_format(session_spans: List[Dict[str, Any]]) -> Optional[SpanMapResult]:
     """Extract fields from service-normalized span format.
 
-    The AgentCore evaluation service sends spans with gen_ai semantic convention
-    events (gen_ai.user.message, gen_ai.choice) instead of body with input/output.
-    This handles that format as a fallback when strands-evals mappers can't parse it.
+    Handles two service formats:
+    1. SESSION format with span_events[*].body (multi-turn conversations where the
+       service collapses all ADOT spans into one span with multiple span_events)
+    2. gen_ai semantic convention events (single-turn Strands spans)
     """
     import json as _json
 
+    # --- Multi-turn: extract from span_events[*].body ---
+    for span in session_spans:
+        span_events = span.get("span_events", [])
+        if len(span_events) >= 1:
+            turns: List[Dict[str, Any]] = []
+            last_input = None
+            last_output = None
+            for se in span_events:
+                body = se.get("body", {})
+                inp_msgs = (body.get("input") or {}).get("messages", [])
+                out_msgs = (body.get("output") or {}).get("messages", [])
+                user_text = _extract_message_text(inp_msgs) if inp_msgs else None
+                asst_text = _extract_message_text(out_msgs) if out_msgs else None
+                if user_text:
+                    turns.append({"role": "user", "content": user_text})
+                    last_input = user_text
+                if asst_text:
+                    turns.append({"role": "assistant", "content": asst_text})
+                    last_output = asst_text
+            if turns and last_input and last_output:
+                return SpanMapResult(
+                    input=last_input,
+                    actual_output=last_output,
+                    turns=turns if len(turns) > 2 else None,
+                )
+
+    # --- Single-turn: extract from gen_ai semantic convention events ---
     for span in session_spans:
         scope = span.get("scope", {}).get("name", "")
         events = span.get("events", [])

tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/test_error_handling.py

@@ -107,7 +107,7 @@ def test_02_unrecognized_scope(self):
         ]
         adapter = AutoEvalsAdapter(metric=_mock_scorer())
         result = adapter(_make_evaluator_input(spans=spans))
-        _assert_error_response(result, "FIELD_EXTRACTION_ERROR")
+        _assert_error_response(result, "MISSING_REQUIRED_FIELD")
 
     def test_03_spans_missing_body_input(self):
         spans = [

tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/test_adapter.py

@@ -396,3 +396,130 @@ def test_conversational_metric_single_turn_returns_error(self):
 
         assert result.errorCode == "FIELD_EXTRACTION_ERROR"
         assert "multi-turn" in result.errorMessage.lower() or "Multiple" in result.errorMessage
+
+
+class TestDeepEvalAdapterServiceNormalizedMultiTurn:
+    """Tests for conversational metrics with service-normalized SESSION format.
+
+    The AgentCore service collapses multi-turn ADOT docs into one span with
+    span_events[*].body. These tests verify the adapter correctly extracts
+    all turns and passes a ConversationalTestCase to the metric.
+    """
+
+    def _make_session_evaluator_input(self, num_turns=3):
+        """Build EvaluatorInput in service-normalized SESSION format."""
+        span_events = []
+        for i in range(num_turns):
+            span_events.append({
+                "body": {
+                    "input": {
+                        "messages": [
+                            {"role": "user", "content": {"content": [{"text": f"User turn {i+1}"}]}}
+                        ]
+                    },
+                    "output": {
+                        "messages": [
+                            {"role": "assistant", "content": {"message": [{"text": f"Bot turn {i+1}"}]}}
+                        ]
+                    },
+                }
+            })
+        spans = [
+            {
+                "traceId": "t-session",
+                "spanId": "s-session",
+                "source": "adot_cw",
+                "attributes": {"session.id": "multi-turn-session"},
+                "span_events": span_events,
+            }
+        ]
+        return EvaluatorInput(
+            evaluation_level="SESSION",
+            session_spans=spans,
+        )
+
+    def test_conversational_metric_receives_all_turns(self):
+        """Multi-turn metric gets ConversationalTestCase with correct turn count."""
+        from deepeval.metrics import BaseConversationalMetric
+        from deepeval.test_case import ConversationalTestCase
+
+        metric = MagicMock(spec=BaseConversationalMetric)
+        type(metric).__name__ = "GoalAccuracyMetric"
+        metric.threshold = 0.5
+        metric.score = 0.9
+        metric.reason = "Goal achieved"
+        del metric.success
+
+        captured_test_case = {}
+
+        def measure_side_effect(test_case):
+            captured_test_case["tc"] = test_case
+            metric.score = 0.9
+            metric.reason = "Goal achieved"
+
+        metric.measure = MagicMock(side_effect=measure_side_effect)
+        adapter = DeepEvalAdapter(metric=metric)
+
+        result = adapter(self._make_session_evaluator_input(num_turns=4))
+
+        assert result.value == 0.9
+        assert result.label == "Pass"
+        tc = captured_test_case["tc"]
+        assert isinstance(tc, ConversationalTestCase)
+        assert len(tc.turns) == 8  # 4 user + 4 assistant turns
+
+    def test_conversational_metric_turn_content_correct(self):
+        """Verify turn content is correctly extracted from nested message format."""
+        from deepeval.metrics import BaseConversationalMetric
+        from deepeval.test_case import ConversationalTestCase
+
+        metric = MagicMock(spec=BaseConversationalMetric)
+        type(metric).__name__ = "RoleAdherenceMetric"
+        metric.threshold = 0.5
+        metric.score = 1.0
+        metric.reason = "No violations"
+        del metric.success
+
+        captured_test_case = {}
+
+        def measure_side_effect(test_case):
+            captured_test_case["tc"] = test_case
+            metric.score = 1.0
+
+        metric.measure = MagicMock(side_effect=measure_side_effect)
+        adapter = DeepEvalAdapter(metric=metric)
+
+        result = adapter(self._make_session_evaluator_input(num_turns=2))
+
+        assert result.value == 1.0
+        tc = captured_test_case["tc"]
+        assert tc.turns[0].role == "user"
+        assert tc.turns[0].content == "User turn 1"
+        assert tc.turns[1].role == "assistant"
+        assert tc.turns[1].content == "Bot turn 1"
+        assert tc.turns[2].role == "user"
+        assert tc.turns[2].content == "User turn 2"
+        assert tc.turns[3].role == "assistant"
+        assert tc.turns[3].content == "Bot turn 2"
+
+    def test_five_turn_session_evaluation(self):
+        """Realistic 5-turn session evaluation (matches typical MACE migration)."""
+        from deepeval.metrics import BaseConversationalMetric
+
+        metric = MagicMock(spec=BaseConversationalMetric)
+        type(metric).__name__ = "ConversationCompletenessMetric"
+        metric.threshold = 0.5
+        metric.score = 0.75
+        metric.reason = "Mostly complete"
+        del metric.success
+
+        metric.measure = MagicMock(side_effect=lambda tc: None)
+        adapter = DeepEvalAdapter(metric=metric)
+
+        result = adapter(self._make_session_evaluator_input(num_turns=5))
+
+        assert result.value == 0.75
+        assert result.label == "Pass"
+        metric.measure.assert_called_once()
+        tc = metric.measure.call_args[0][0]
+        assert len(tc.turns) == 10  # 5 user + 5 assistant

tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/test_error_handling.py

@@ -185,7 +185,7 @@ def test_15_unrecognized_scope_deepeval(self):
         ]
         adapter = DeepEvalAdapter(metric=_mock_metric())
         result = adapter(_make_evaluator_input(spans=spans))
-        _assert_error_response(result, "FIELD_EXTRACTION_ERROR")
+        _assert_error_response(result, "MISSING_REQUIRED_FIELD")
 
     def test_16_spans_missing_body_input(self):
         spans = [

tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/test_span_mappers.py

@@ -112,3 +112,157 @@ def test_span_map_result_fields(self):
         assert result.tools_called == [{"name": "tool1", "input_parameters": {"a": 1}, "output": "result"}]
         assert result.expected_output is None
         assert result.system_prompt is None
+
+
+def _make_service_normalized_session_spans(num_turns=3):
+    """Build service-normalized SESSION format spans (span_events[*].body).
+
+    This is the format the AgentCore service sends to Lambda for SESSION-level
+    evaluators: one span with multiple span_events, each representing a turn.
+    """
+    span_events = []
+    for i in range(num_turns):
+        span_events.append({
+            "body": {
+                "input": {
+                    "messages": [
+                        {"role": "user", "content": {"content": [{"text": f"User message {i+1}"}]}}
+                    ]
+                },
+                "output": {
+                    "messages": [
+                        {"role": "assistant", "content": {"message": [{"text": f"Assistant response {i+1}"}]}}
+                    ]
+                },
+            }
+        })
+    return [
+        {
+            "traceId": "trace-multi",
+            "spanId": "span-multi",
+            "source": "adot_cw",
+            "attributes": {"session.id": "session-1"},
+            "span_events": span_events,
+        }
+    ]
+
+
+class TestServiceNormalizedMultiTurn:
+    """Tests for multi-turn extraction from service-normalized SESSION format."""
+
+    def test_extracts_all_turns_from_span_events(self):
+        spans = _make_service_normalized_session_spans(num_turns=3)
+        result = map_spans(spans)
+
+        assert result.turns is not None
+        assert len(result.turns) == 6  # 3 user + 3 assistant
+        assert result.turns[0] == {"role": "user", "content": "User message 1"}
+        assert result.turns[1] == {"role": "assistant", "content": "Assistant response 1"}
+        assert result.turns[4] == {"role": "user", "content": "User message 3"}
+        assert result.turns[5] == {"role": "assistant", "content": "Assistant response 3"}
+
+    def test_input_and_output_are_last_turn(self):
+        spans = _make_service_normalized_session_spans(num_turns=3)
+        result = map_spans(spans)
+
+        assert result.input == "User message 3"
+        assert result.actual_output == "Assistant response 3"
+
+    def test_single_span_event_returns_none_turns(self):
+        """Single span_event should NOT populate turns (not multi-turn)."""
+        spans = _make_service_normalized_session_spans(num_turns=1)
+        result = map_spans(spans)
+
+        # With only 1 turn (2 entries: user+assistant), turns should be None
+        assert result.turns is None
+        # But input/output should still be extracted
+        assert result.input == "User message 1"
+        assert result.actual_output == "Assistant response 1"
+
+    def test_handles_string_content_variant(self):
+        """Test spans where content is a plain string instead of list of dicts."""
+        spans = [
+            {
+                "traceId": "t1",
+                "spanId": "s1",
+                "attributes": {"session.id": "sess"},
+                "span_events": [
+                    {
+                        "body": {
+                            "input": {"messages": [{"role": "user", "content": "Hello plain"}]},
+                            "output": {"messages": [{"role": "assistant", "content": "Hi plain"}]},
+                        }
+                    },
+                    {
+                        "body": {
+                            "input": {"messages": [{"role": "user", "content": "Follow up"}]},
+                            "output": {"messages": [{"role": "assistant", "content": "Got it"}]},
+                        }
+                    },
+                ],
+            }
+        ]
+        result = map_spans(spans)
+
+        assert result.turns is not None
+        assert len(result.turns) == 4
+        assert result.turns[0] == {"role": "user", "content": "Hello plain"}
+        assert result.turns[3] == {"role": "assistant", "content": "Got it"}
+
+    def test_handles_nested_content_dict_variant(self):
+        """Test the {content: {content: [{text: ...}]}} nesting."""
+        spans = [
+            {
+                "traceId": "t1",
+                "spanId": "s1",
+                "attributes": {"session.id": "sess"},
+                "span_events": [
+                    {
+                        "body": {
+                            "input": {
+                                "messages": [
+                                    {"role": "user", "content": {"content": [{"text": "Turn 1 input"}]}}
+                                ]
+                            },
+                            "output": {
+                                "messages": [
+                                    {"role": "assistant", "content": {"message": [{"text": "Turn 1 output"}]}}
+                                ]
+                            },
+                        }
+                    },
+                    {
+                        "body": {
+                            "input": {
+                                "messages": [
+                                    {"role": "user", "content": {"content": [{"text": "Turn 2 input"}]}}
+                                ]
+                            },
+                            "output": {
+                                "messages": [
+                                    {"role": "assistant", "content": {"message": [{"text": "Turn 2 output"}]}}
+                                ]
+                            },
+                        }
+                    },
+                ],
+            }
+        ]
+        result = map_spans(spans)
+
+        assert result.turns is not None
+        assert len(result.turns) == 4
+        assert result.turns[0]["content"] == "Turn 1 input"
+        assert result.turns[1]["content"] == "Turn 1 output"
+        assert result.turns[2]["content"] == "Turn 2 input"
+        assert result.turns[3]["content"] == "Turn 2 output"
+
+    def test_five_turns_for_session_evaluation(self):
+        """Realistic test: 5-turn conversation as sent by the service."""
+        spans = _make_service_normalized_session_spans(num_turns=5)
+        result = map_spans(spans)
+
+        assert result.turns is not None
+        assert len(result.turns) == 10
+        assert result.input == "User message 5"
+        assert result.actual_output == "Assistant response 5"