Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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", []))

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P0] Could we reuse the existing CloudWatch message parsing here? Real Strands bodies encode user content as {"content": "[{\"text\": ...}]"}, and multi-turn input can contain interleaved user/assistant history. This helper returns the serialized JSON literally and selects the first historical message. Against three real invoke_agent events, it produced six turns but every user turn still contained JSON syntax. PR #454 also established the chronological-role/latest-user behavior, so sharing that parsing logic would keep this path consistent.

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 ---

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P0] I think this still misses the exact one-span/many-event case this PR is trying to fix. When the CloudWatch mapper recognizes the span, it returns valid input/output from the first event, so the fallback in map_spans() never calls this new path. I reproduced this using real Strands invoke_agent bodies collapsed into the target shape: turns stayed None and DeepEval returned FIELD_EXTRACTION_ERROR. Could we run the service-format extraction when multiple body events are present even if the mapper found one valid pair, and retain the real scope/attributes in the regression test?

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", [])
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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")

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P0] I do not think this expected error code matches the current behavior. The full third-party suite has 2 failed, 115 passed: this case and the equivalent DeepEval test both still return FIELD_EXTRACTION_ERROR. These spans have no body or span_events, so the new extractor cannot reach MISSING_REQUIRED_FIELD. The base tests pass with the original expectation. Could we keep FIELD_EXTRACTION_ERROR unless the runtime behavior is intentionally changing too?


def test_03_spans_missing_body_input(self):
spans = [
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Loading