diff --git a/.gitignore b/.gitignore index 01fe8e22..161403e7 100644 --- a/.gitignore +++ b/.gitignore @@ -229,3 +229,4 @@ local_settings.py Dockerfile CLAUDE.md .omc/ +.deepeval/ diff --git a/pyproject.toml b/pyproject.toml index 61520a5b..b1fc5e90 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -173,3 +173,9 @@ simulation = [ datasets = [ "requests>=2.31.0", ] +deepeval = [ + "deepeval>=2.0.0", +] +autoevals = [ + "autoevals>=0.0.50", +] diff --git a/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/__init__.py b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/__init__.py new file mode 100644 index 00000000..06ba3d0a --- /dev/null +++ b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/__init__.py @@ -0,0 +1,5 @@ +"""Third-party evaluation adapters for AgentCore code-based evaluators.""" + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.base import BaseAdapter + +__all__ = ["BaseAdapter"] diff --git a/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/__init__.py b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/__init__.py new file mode 100644 index 00000000..40e25fc1 --- /dev/null +++ b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/__init__.py @@ -0,0 +1,5 @@ +"""Autoevals adapter for AgentCore code-based evaluators.""" + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.autoevals.adapter import AutoevalsAdapter + +__all__ = ["AutoevalsAdapter"] diff --git a/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/adapter.py b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/adapter.py new file mode 100644 index 00000000..00b9a337 --- /dev/null +++ b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/adapter.py @@ -0,0 +1,86 @@ +"""Autoevals adapter for AgentCore code-based evaluators.""" + +import logging +from typing import Any, Callable, Dict, Optional + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.models import EvaluatorInput, EvaluatorOutput +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.base import BaseAdapter + +logger = logging.getLogger(__name__) + + +class AutoevalsAdapter(BaseAdapter): + """Adapter that runs an Autoevals scorer against AgentCore evaluation events. + + Example (default span mapping):: + + from autoevals import Factuality + from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.autoevals import AutoevalsAdapter + + scorer = Factuality() + adapter = AutoevalsAdapter(scorer=scorer) + + Example (customer mapper returning eval kwargs):: + + adapter = AutoevalsAdapter( + scorer=Factuality(), + customer_mapper=lambda ev: { + "input": ev.session_spans[0]["attributes"]["question"], + "output": ev.session_spans[0]["attributes"]["answer"], + "expected": "the expected answer", + }, + ) + """ + + def __init__( + self, + scorer: Any, + customer_mapper: Optional[Callable[[EvaluatorInput], Dict[str, Any]]] = None, + threshold: float = 0.5, + ): + """Initialize the adapter. + + Args: + scorer: An Autoevals scorer instance (e.g. Factuality(), ClosedQA()). + customer_mapper: Optional callable that receives the EvaluatorInput and + returns a dict of kwargs for scorer.eval(). Bypasses default span + mapping when provided. Expected keys: input, output, expected (optional). + threshold: Score threshold for Pass/Fail determination. Defaults to 0.5. + """ + self.scorer = scorer + self.customer_mapper = customer_mapper + self.threshold = threshold + + def _run(self, evaluator_input: EvaluatorInput) -> EvaluatorOutput: + """Run the Autoevals scorer pipeline.""" + if self.customer_mapper is not None: + kwargs = self.customer_mapper(evaluator_input) + else: + result = self._default_extract(evaluator_input) + if not result.input or not result.actual_output: + missing = [] + if not result.input: + missing.append("input") + if not result.actual_output: + missing.append("actual_output") + scorer_name = type(self.scorer).__name__ + return EvaluatorOutput( + label="Error", + errorCode="MISSING_REQUIRED_FIELD", + errorMessage=f"Field(s) {missing} required by {scorer_name} but not found in evaluation event. " + f"Provide a customer_mapper or ensure spans contain the necessary data.", + ) + kwargs: Dict[str, Any] = { + "input": result.input, + "output": result.actual_output, + } + if result.expected_output: + kwargs["expected"] = result.expected_output + + eval_result = self.scorer.eval(**kwargs) + + score = eval_result.score + label = "Pass" if score is not None and score >= self.threshold else "Fail" + explanation = getattr(eval_result, "metadata", {}).get("rationale", "") if hasattr(eval_result, "metadata") else "" + + return EvaluatorOutput(value=score, label=label, explanation=explanation) diff --git a/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/base.py b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/base.py new file mode 100644 index 00000000..93157f05 --- /dev/null +++ b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/base.py @@ -0,0 +1,77 @@ +"""Base adapter for third-party evaluation framework integrations.""" + +import abc +import logging +from typing import Any, Dict, List, Optional + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.models import EvaluatorInput, EvaluatorOutput +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.span_mappers import ( + SpanMapResult, + map_spans, +) + +logger = logging.getLogger(__name__) + + +class BaseAdapter(abc.ABC): + """Base adapter for third-party evaluation framework integrations. + + Accepts an EvaluatorInput (from the code_based_evaluators flow), + extracts fields from spans using the built-in mapper layer, runs the + evaluation via execute(), and returns an EvaluatorOutput. + + Never raises unhandled exceptions — always returns a valid EvaluatorOutput. + """ + + def __call__(self, evaluator_input: EvaluatorInput, context: Any = None) -> EvaluatorOutput: + """Handle an evaluation invocation. + + Args: + evaluator_input: Parsed EvaluatorInput from the code-based evaluator flow. + context: Lambda context object (unused). + + Returns: + EvaluatorOutput with score, label, and explanation or error fields. + """ + try: + return self._run(evaluator_input) + except ValueError as e: + logger.error("Field extraction failed: %s", e) + return EvaluatorOutput( + label="Error", + errorCode="FIELD_EXTRACTION_ERROR", + errorMessage=str(e), + ) + except Exception as e: + logger.error("Execution failed: %s", e, exc_info=True) + return EvaluatorOutput( + label="Error", + errorCode="METRIC_ERROR", + errorMessage=f"{type(self).__name__} failed: {e}", + ) + + @abc.abstractmethod + def _run(self, evaluator_input: EvaluatorInput) -> EvaluatorOutput: + """Run the full evaluation pipeline. Subclasses implement this.""" + + def _default_extract(self, evaluator_input: EvaluatorInput) -> SpanMapResult: + """Extract fields using the built-in span mapper layer.""" + spans = self._filter_spans_by_target(evaluator_input) + return map_spans(spans, evaluator_input.reference_inputs) + + def _filter_spans_by_target(self, evaluator_input: EvaluatorInput) -> List[Dict]: + """Filter session spans based on evaluationLevel and evaluationTarget. + + Per the AgentCore code-based evaluator contract: + - TRACE: only spans matching target_trace_id + - TOOL_CALL: only the span matching target_span_id + - SESSION: all spans (no filtering) + """ + spans = evaluator_input.session_spans + + if evaluator_input.evaluation_level == "TRACE" and evaluator_input.target_trace_id: + spans = [s for s in spans if s.get("traceId") == evaluator_input.target_trace_id] + elif evaluator_input.evaluation_level == "TOOL_CALL" and evaluator_input.target_span_id: + spans = [s for s in spans if s.get("spanId") == evaluator_input.target_span_id] + + return spans diff --git a/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/__init__.py b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/__init__.py new file mode 100644 index 00000000..99cf10d5 --- /dev/null +++ b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/__init__.py @@ -0,0 +1,5 @@ +"""DeepEval adapter for AgentCore code-based evaluators.""" + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.deepeval.adapter import DeepEvalAdapter + +__all__ = ["DeepEvalAdapter"] diff --git a/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/adapter.py b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/adapter.py new file mode 100644 index 00000000..3e859206 --- /dev/null +++ b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/adapter.py @@ -0,0 +1,101 @@ +"""DeepEval adapter for AgentCore code-based evaluators.""" + +import logging +from typing import Any, Callable, Optional + +from deepeval.metrics import BaseMetric +from deepeval.test_case import LLMTestCase + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.models import EvaluatorInput, EvaluatorOutput +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.base import BaseAdapter + +logger = logging.getLogger(__name__) + + +class DeepEvalAdapter(BaseAdapter): + """Adapter that runs a DeepEval metric against AgentCore evaluation events. + + Example (default span mapping):: + + from deepeval.metrics import AnswerRelevancyMetric + from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.deepeval import DeepEvalAdapter + + metric = AnswerRelevancyMetric(threshold=0.7) + adapter = DeepEvalAdapter(metric=metric) + + Example (customer mapper returning LLMTestCase):: + + adapter = DeepEvalAdapter( + metric=AnswerRelevancyMetric(threshold=0.7), + customer_mapper=lambda ev: LLMTestCase( + input=ev.session_spans[0]["attributes"]["user_query"], + actual_output=ev.session_spans[0]["attributes"]["response"], + ), + ) + """ + + def __init__( + self, + metric: BaseMetric, + customer_mapper: Optional[Callable[[EvaluatorInput], LLMTestCase]] = None, + model: Optional[Any] = None, + ): + """Initialize the adapter. + + Args: + metric: A DeepEval BaseMetric instance (e.g. AnswerRelevancyMetric). + customer_mapper: Optional callable that receives the EvaluatorInput and + returns a LLMTestCase. Bypasses default span mapping when provided. + model: Optional model override for the metric's LLM. + """ + self.metric = metric + self.customer_mapper = customer_mapper + if model is not None: + self.metric.model = model + + def _run(self, evaluator_input: EvaluatorInput) -> EvaluatorOutput: + """Run the DeepEval metric pipeline.""" + if self.customer_mapper is not None: + test_case = self.customer_mapper(evaluator_input) + else: + result = self._default_extract(evaluator_input) + if not result.input or not result.actual_output: + missing = [] + if not result.input: + missing.append("input") + if not result.actual_output: + missing.append("actual_output") + metric_name = type(self.metric).__name__ + return EvaluatorOutput( + label="Error", + errorCode="MISSING_REQUIRED_FIELD", + errorMessage=f"Field(s) {missing} required by {metric_name} but not found in evaluation event. " + f"Provide a customer_mapper or ensure spans contain the necessary data.", + ) + test_case = LLMTestCase( + input=result.input, + actual_output=result.actual_output, + expected_output=result.expected_output, + context=result.context, + retrieval_context=result.retrieval_context, + ) + + try: + self.metric.measure(test_case) + except Exception as e: + if type(e).__name__ == "MissingTestCaseParamsError": + return EvaluatorOutput( + label="Error", + errorCode="MISSING_REQUIRED_FIELD", + errorMessage=f"{type(self.metric).__name__} requires fields not extracted from spans: {e}. " + f"Provide a customer_mapper to supply custom fields from your trace data.", + ) + raise + + score = self.metric.score + reason = getattr(self.metric, "reason", None) or "" + threshold = getattr(self.metric, "threshold", 0.5) + success = getattr(self.metric, "success", score is not None and score >= threshold) + label = "Pass" if success else "Fail" + + return EvaluatorOutput(value=score, label=label, explanation=reason) diff --git a/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/__init__.py b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/__init__.py new file mode 100644 index 00000000..d7964b92 --- /dev/null +++ b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/__init__.py @@ -0,0 +1,13 @@ +"""Span mappers for extracting evaluation fields from Agent SDK trace formats.""" + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.span_mappers.base import ( + map_spans, +) +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.span_mappers.common import ( + SpanMapResult, +) +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.span_mappers.strands import ( + StrandsSpanMapper, +) + +__all__ = ["SpanMapResult", "map_spans", "StrandsSpanMapper"] diff --git a/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/base.py b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/base.py new file mode 100644 index 00000000..8e1c90d7 --- /dev/null +++ b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/base.py @@ -0,0 +1,51 @@ +"""Span mapping orchestration.""" + +import logging +from typing import Any, Dict, List, Optional + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.span_mappers.common import ( + SpanMapResult, +) +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.span_mappers.strands import ( + StrandsSpanMapper, +) + +logger = logging.getLogger(__name__) + +_strands_mapper = StrandsSpanMapper() + + +def map_spans( + session_spans: List[Dict[str, Any]], + reference_inputs: Optional[List[Any]] = None, +) -> SpanMapResult: + """Map session spans to evaluation fields. + + Currently supports Strands Agent SDK spans (scope.name == "strands.telemetry.tracer"). + Additional framework support can be added when real span data is available. + + Args: + session_spans: Raw ADOT span dicts from the evaluation service. + reference_inputs: Optional ReferenceInput list for expected_output. + + Returns: + SpanMapResult with extracted fields. + + Raises: + ValueError: If no mapper can extract data from the spans. + """ + result = _strands_mapper.map(session_spans) + if result is not None: + if reference_inputs: + ref = reference_inputs[0] + expected = getattr(ref, "expected_response_text", None) + if expected: + result.expected_output = expected + return result + + raise ValueError( + "Could not extract evaluation fields from spans. " + "No Strands agent span (scope.name=='strands.telemetry.tracer' with " + "gen_ai.operation.name=='invoke_agent') found. " + "Provide a customer_mapper for custom or unsupported span formats." + ) diff --git a/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/common.py b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/common.py new file mode 100644 index 00000000..3700bd46 --- /dev/null +++ b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/common.py @@ -0,0 +1,100 @@ +"""Common span mapping utilities shared across format-specific mappers.""" + +import json +import logging +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + + +@dataclass +class SpanMapResult: + """Extraction result from spans — only what metrics consume.""" + + input: Optional[str] = None + actual_output: Optional[str] = None + retrieval_context: Optional[List[str]] = None + context: Optional[List[str]] = None + system_prompt: Optional[str] = None + expected_output: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + """Convert to dict, omitting None values.""" + result: Dict[str, Any] = {} + if self.input is not None: + result["input"] = self.input + if self.actual_output is not None: + result["actual_output"] = self.actual_output + if self.retrieval_context is not None: + result["retrieval_context"] = self.retrieval_context + if self.context is not None: + result["context"] = self.context + if self.system_prompt is not None: + result["system_prompt"] = self.system_prompt + if self.expected_output is not None: + result["expected_output"] = self.expected_output + return result + + +def _try_parse_text_blocks(val: str) -> Optional[str]: + """Try to parse a JSON-encoded list of text blocks. + + Strands encodes content as: '[{"text": "Hello"}, {"text": "world"}]' + Returns joined text if parseable, None otherwise. + """ + try: + parsed = json.loads(val) + except (json.JSONDecodeError, TypeError): + return None + if not isinstance(parsed, list): + return None + parts = [] + for item in parsed: + if isinstance(item, dict) and "text" in item: + parts.append(item["text"]) + return "\n".join(parts) if parts else None + + +def _get_message_content(message: Any) -> str: + """Extract text content from a message object. + + Handles Strands format variations: + - {"content": "plain string"} + - {"content": {"content": '[{"text": "..."}]'}} (user messages) + - {"content": {"message": "...", "finish_reason": "..."}} (assistant messages) + - {"content": '[{"text": "..."}, {"toolUse": {...}}]'} (assistant with tool calls) + """ + if isinstance(message, str): + return _try_parse_text_blocks(message) or message + if isinstance(message, dict): + for key in ("content", "message"): + if key in message: + val = message[key] + if isinstance(val, str): + return _try_parse_text_blocks(val) or val + if isinstance(val, dict): + return _get_message_content(val) + if isinstance(val, list): + parts = [] + for item in val: + if isinstance(item, str): + parts.append(item) + elif isinstance(item, dict) and "text" in item: + parts.append(item["text"]) + if parts: + return "\n".join(parts) + return str(val) + return "" + + +def _parse_span_event_body(body: Any) -> Dict[str, Any]: + """Parse the body of a span event, handling both dict and JSON string.""" + if isinstance(body, str): + try: + return json.loads(body) + except (json.JSONDecodeError, TypeError): + return {} + if isinstance(body, dict): + return body + return {} diff --git a/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/strands.py b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/strands.py new file mode 100644 index 00000000..0ab41764 --- /dev/null +++ b/src/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/strands.py @@ -0,0 +1,158 @@ +"""Strands Agent SDK span mapper.""" + +import logging +from typing import Any, Dict, List, Optional + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.span_mappers.common import ( + SpanMapResult, + _get_message_content, + _parse_span_event_body, + _try_parse_text_blocks, +) + +logger = logging.getLogger(__name__) + +SCOPE_STRANDS = "strands.telemetry.tracer" + + +class StrandsSpanMapper: + """Maps Strands agent spans to evaluation fields. + + Extracts only what metrics consume: input, actual_output, retrieval_context, + system_prompt. Supports two trace formats: + - Inline events (unified ADOT): data in span.events[] + - Span body (CloudWatch ADOT): data in span.span_events[].body + """ + + def map(self, session_spans: List[Dict[str, Any]]) -> Optional[SpanMapResult]: + """Map session spans to evaluation fields. + + Args: + session_spans: Raw ADOT span dicts (already filtered by evaluationTarget). + + Returns: + SpanMapResult with extracted fields, or None if no Strands agent span found. + """ + agent_span = self._find_invoke_agent_span(session_spans) + if agent_span is None: + return None + + if self._has_inline_events(agent_span): + return self._extract_from_inline_events(agent_span) + + return self._extract_from_span_body(agent_span) + + def _find_invoke_agent_span(self, session_spans: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]: + """Find the invoke_agent span with strands.telemetry.tracer scope.""" + for span in session_spans: + scope = span.get("scope", {}) + if not isinstance(scope, dict): + continue + if scope.get("name") != SCOPE_STRANDS: + continue + attributes = span.get("attributes", {}) + if attributes.get("gen_ai.operation.name") == "invoke_agent": + return span + return None + + def _has_inline_events(self, span: Dict[str, Any]) -> bool: + """Check if span uses inline events format (unified ADOT).""" + events = span.get("events", []) + return any( + e.get("name") in ("gen_ai.user.message", "gen_ai.choice") + for e in events + ) + + def _extract_from_inline_events(self, agent_span: Dict[str, Any]) -> Optional[SpanMapResult]: + """Extract fields from inline events[] (unified ADOT format). + + - input: first gen_ai.user.message → attributes.content (JSON text blocks) + - actual_output: last gen_ai.choice → attributes.message (plain string) + - system_prompt: from span attributes + """ + events = agent_span.get("events", []) + + input_text = None + actual_output = None + + for event in events: + name = event.get("name") + attrs = event.get("attributes", {}) + + if name == "gen_ai.user.message" and input_text is None: + content_str = attrs.get("content", "") + input_text = _try_parse_text_blocks(content_str) or content_str + + elif name == "gen_ai.choice": + actual_output = attrs.get("message", "") + + if not input_text and not actual_output: + return None + + system_prompt = agent_span.get("attributes", {}).get("system_prompt") + + return SpanMapResult( + input=input_text, + actual_output=actual_output, + system_prompt=system_prompt, + ) + + def _extract_from_span_body(self, agent_span: Dict[str, Any]) -> Optional[SpanMapResult]: + """Extract fields from span_events[].body (CloudWatch ADOT format). + + Multi-turn: one span_event per turn. + - input: first user message across all turns + - actual_output: last assistant message across all turns + - retrieval_context: all tool outputs + - system_prompt: from system-role message or span attributes + """ + input_text = None + actual_output = None + tool_outputs: List[str] = [] + system_prompt: Optional[str] = None + + for event in agent_span.get("span_events", []): + body = _parse_span_event_body(event.get("body")) + if not body: + continue + + input_data = body.get("input", {}) + if isinstance(input_data, dict): + for msg in input_data.get("messages", []): + if not isinstance(msg, dict): + continue + role = msg.get("role", "") + if role == "system" and system_prompt is None: + content = _get_message_content(msg) + if content: + system_prompt = content + elif role == "user" and input_text is None: + content = _get_message_content(msg) + if content: + input_text = content + + output_data = body.get("output", {}) + if isinstance(output_data, dict): + for msg in output_data.get("messages", []): + if not isinstance(msg, dict): + continue + role = msg.get("role", "") + content = _get_message_content(msg) + if role == "assistant" and content: + actual_output = content + elif role == "tool" and content: + tool_outputs.append(content) + + if not input_text and not actual_output: + return None + + if system_prompt is None: + system_prompt = agent_span.get("attributes", {}).get("system_prompt") + + return SpanMapResult( + input=input_text, + actual_output=actual_output, + retrieval_context=tool_outputs if tool_outputs else None, + context=tool_outputs if tool_outputs else None, + system_prompt=system_prompt, + ) diff --git a/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/__init__.py b/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/__init__.py b/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/test_adapter.py b/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/test_adapter.py new file mode 100644 index 00000000..910c59aa --- /dev/null +++ b/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/autoevals/test_adapter.py @@ -0,0 +1,203 @@ +"""Tests for AutoevalsAdapter.""" + +from unittest.mock import MagicMock + +import pytest + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.models import EvaluatorInput, EvaluatorOutput +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.autoevals.adapter import AutoevalsAdapter + + +def _make_evaluator_input(spans=None): + """Build an EvaluatorInput with agent-level spans.""" + if spans is None: + spans = [ + { + "traceId": "t1", + "spanId": "s1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "span_events": [ + { + "body": { + "input": {"messages": [{"role": "user", "content": "What is AI?"}]}, + "output": {"messages": [{"role": "assistant", "content": "AI is artificial intelligence."}]}, + } + } + ], + } + ] + return EvaluatorInput( + evaluation_level="TRACE", + session_spans=spans, + target_trace_id="t1", + ) + + +def _mock_scorer(score=0.9, rationale="Good answer"): + """Create a mock Autoevals scorer.""" + scorer = MagicMock() + type(scorer).__name__ = "MockScorer" + + result = MagicMock() + result.score = score + result.metadata = {"rationale": rationale} + + scorer.eval = MagicMock(return_value=result) + return scorer + + +class TestAutoevalsAdapterSuccess: + def test_returns_pass_when_score_above_threshold(self): + scorer = _mock_scorer(score=0.8) + adapter = AutoevalsAdapter(scorer=scorer) + + result = adapter(_make_evaluator_input()) + + assert isinstance(result, EvaluatorOutput) + assert result.value == 0.8 + assert result.label == "Pass" + assert result.explanation == "Good answer" + + def test_returns_fail_when_score_below_threshold(self): + scorer = _mock_scorer(score=0.3) + adapter = AutoevalsAdapter(scorer=scorer) + + result = adapter(_make_evaluator_input()) + + assert result.value == 0.3 + assert result.label == "Fail" + + def test_custom_threshold(self): + scorer = _mock_scorer(score=0.6) + adapter = AutoevalsAdapter(scorer=scorer, threshold=0.7) + + result = adapter(_make_evaluator_input()) + + assert result.label == "Fail" + + def test_custom_threshold_pass(self): + scorer = _mock_scorer(score=0.8) + adapter = AutoevalsAdapter(scorer=scorer, threshold=0.7) + + result = adapter(_make_evaluator_input()) + + assert result.label == "Pass" + + def test_default_threshold_is_half(self): + scorer = _mock_scorer() + adapter = AutoevalsAdapter(scorer=scorer) + + assert adapter.threshold == 0.5 + + def test_scorer_eval_called_with_input_and_output(self): + scorer = _mock_scorer() + adapter = AutoevalsAdapter(scorer=scorer) + + adapter(_make_evaluator_input()) + + scorer.eval.assert_called_once() + call_kwargs = scorer.eval.call_args[1] + assert call_kwargs["input"] == "What is AI?" + assert call_kwargs["output"] == "AI is artificial intelligence." + + def test_custom_customer_mapper(self): + scorer = _mock_scorer() + adapter = AutoevalsAdapter( + scorer=scorer, + customer_mapper=lambda ev: { + "input": "custom input", + "output": "custom output", + }, + ) + + result = adapter(_make_evaluator_input()) + + call_kwargs = scorer.eval.call_args[1] + assert call_kwargs["input"] == "custom input" + assert call_kwargs["output"] == "custom output" + + +class TestAutoevalsAdapterErrors: + def test_no_agent_spans_returns_error(self): + spans = [ + { + "traceId": "t1", + "spanId": "s1", + "attributes": {"gen_ai.operation.name": "chat"}, + "span_events": [], + } + ] + scorer = _mock_scorer() + adapter = AutoevalsAdapter(scorer=scorer) + + result = adapter(_make_evaluator_input(spans=spans)) + + assert result.errorCode == "FIELD_EXTRACTION_ERROR" + + def test_missing_input_returns_error(self): + spans = [ + { + "traceId": "t1", + "spanId": "s1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "span_events": [ + { + "body": { + "output": {"messages": [{"role": "assistant", "content": "answer"}]}, + } + } + ], + } + ] + scorer = _mock_scorer() + adapter = AutoevalsAdapter(scorer=scorer) + + result = adapter(_make_evaluator_input(spans=spans)) + + assert result.errorCode == "MISSING_REQUIRED_FIELD" + assert "input" in result.errorMessage + + def test_scorer_exception_returns_error(self): + scorer = _mock_scorer() + scorer.eval = MagicMock(side_effect=RuntimeError("API error")) + adapter = AutoevalsAdapter(scorer=scorer) + + result = adapter(_make_evaluator_input()) + + assert result.errorCode == "METRIC_ERROR" + assert "API error" in result.errorMessage + + def test_never_raises(self): + scorer = _mock_scorer() + scorer.eval = MagicMock(side_effect=Exception("unexpected")) + adapter = AutoevalsAdapter(scorer=scorer) + + result = adapter(_make_evaluator_input()) + + assert isinstance(result, EvaluatorOutput) + assert result.errorCode is not None + + +class TestAutoevalsAdapterEdgeCases: + def test_score_none_returns_fail(self): + scorer = _mock_scorer(score=None) + adapter = AutoevalsAdapter(scorer=scorer) + + result = adapter(_make_evaluator_input()) + + assert result.label == "Fail" + + def test_no_metadata_returns_empty_explanation(self): + scorer = MagicMock() + type(scorer).__name__ = "MockScorer" + result_obj = MagicMock(spec=[]) + result_obj.score = 0.9 + scorer.eval = MagicMock(return_value=result_obj) + + adapter = AutoevalsAdapter(scorer=scorer) + + result = adapter(_make_evaluator_input()) + + assert result.explanation == "" diff --git a/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/__init__.py b/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/test_adapter.py b/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/test_adapter.py new file mode 100644 index 00000000..3c4766f4 --- /dev/null +++ b/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/deepeval/test_adapter.py @@ -0,0 +1,277 @@ +"""Tests for DeepEvalAdapter.""" + +from unittest.mock import MagicMock + +import pytest + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.models import EvaluatorInput, EvaluatorOutput +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.deepeval.adapter import DeepEvalAdapter + + +def _make_evaluator_input(spans=None): + """Build an EvaluatorInput with agent-level spans.""" + if spans is None: + spans = [ + { + "traceId": "t1", + "spanId": "s1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "span_events": [ + { + "body": { + "input": {"messages": [{"role": "user", "content": "What is AI?"}]}, + "output": {"messages": [{"role": "assistant", "content": "AI is artificial intelligence."}]}, + } + } + ], + } + ] + return EvaluatorInput( + evaluation_level="TRACE", + session_spans=spans, + target_trace_id="t1", + ) + + +def _mock_metric(score=0.85, reason="Looks good", threshold=0.7, name="MockMetric"): + """Create a mock metric that returns a fixed score on measure().""" + metric = MagicMock() + type(metric).__name__ = name + metric.threshold = threshold + metric.score = score + metric.reason = reason + del metric.success + + def measure_side_effect(test_case): + metric.score = score + metric.reason = reason + + metric.measure = MagicMock(side_effect=measure_side_effect) + return metric + + +class TestDeepEvalAdapterSuccess: + def test_returns_pass_when_score_above_threshold(self): + metric = _mock_metric(score=0.9, threshold=0.7) + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input()) + + assert isinstance(result, EvaluatorOutput) + assert result.value == 0.9 + assert result.label == "Pass" + assert result.explanation == "Looks good" + + def test_returns_fail_when_score_below_threshold(self): + metric = _mock_metric(score=0.3, threshold=0.7) + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input()) + + assert result.value == 0.3 + assert result.label == "Fail" + + def test_returns_pass_at_exact_threshold(self): + metric = _mock_metric(score=0.7, threshold=0.7) + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input()) + + assert result.label == "Pass" + + def test_metric_measure_called_with_test_case(self): + metric = _mock_metric() + adapter = DeepEvalAdapter(metric=metric) + + adapter(_make_evaluator_input()) + + metric.measure.assert_called_once() + test_case = metric.measure.call_args[0][0] + assert test_case.input == "What is AI?" + assert test_case.actual_output == "AI is artificial intelligence." + + def test_custom_customer_mapper(self): + from deepeval.test_case import LLMTestCase + + metric = _mock_metric() + adapter = DeepEvalAdapter( + metric=metric, + customer_mapper=lambda ev: LLMTestCase( + input="mapped input", + actual_output="mapped output", + ), + ) + + result = adapter(_make_evaluator_input()) + + assert result.value == 0.85 + test_case = metric.measure.call_args[0][0] + assert test_case.input == "mapped input" + assert test_case.actual_output == "mapped output" + + def test_reference_inputs_populates_expected_output(self): + metric = _mock_metric() + adapter = DeepEvalAdapter(metric=metric) + + evaluator_input = EvaluatorInput( + evaluation_level="TRACE", + session_spans=[ + { + "traceId": "t1", + "spanId": "s1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "span_events": [ + { + "body": { + "input": {"messages": [{"role": "user", "content": "What is AI?"}]}, + "output": {"messages": [{"role": "assistant", "content": "AI is artificial intelligence."}]}, + } + } + ], + } + ], + target_trace_id="t1", + reference_inputs=[{"expectedResponse": {"text": "AI stands for artificial intelligence."}}], + ) + + result = adapter(evaluator_input) + + test_case = metric.measure.call_args[0][0] + assert test_case.expected_output == "AI stands for artificial intelligence." + + def test_model_override_sets_metric_model(self): + metric = _mock_metric() + DeepEvalAdapter(metric=metric, model="bedrock/anthropic.claude-3") + + assert metric.model == "bedrock/anthropic.claude-3" + + def test_label_uses_metric_success_true(self): + metric = _mock_metric(score=0.3, threshold=0.7) + metric.success = True + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input()) + + assert result.value == 0.3 + assert result.label == "Pass" + + def test_label_uses_metric_success_false(self): + metric = _mock_metric(score=0.9, threshold=0.7) + metric.success = False + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input()) + + assert result.value == 0.9 + assert result.label == "Fail" + + +class TestDeepEvalAdapterErrors: + def test_no_agent_spans_returns_error(self): + spans = [ + { + "traceId": "t1", + "spanId": "s1", + "attributes": {"gen_ai.operation.name": "chat"}, + "span_events": [], + } + ] + metric = _mock_metric() + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input(spans=spans)) + + assert isinstance(result, EvaluatorOutput) + assert result.errorCode == "FIELD_EXTRACTION_ERROR" + assert result.label == "Error" + + def test_missing_input_returns_error(self): + spans = [ + { + "traceId": "t1", + "spanId": "s1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "span_events": [ + { + "body": { + "output": {"messages": [{"role": "assistant", "content": "answer"}]}, + } + } + ], + } + ] + metric = _mock_metric() + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input(spans=spans)) + + assert result.errorCode == "MISSING_REQUIRED_FIELD" + assert "input" in result.errorMessage + assert "customer_mapper" in result.errorMessage + metric.measure.assert_not_called() + + def test_metric_measure_exception_returns_error(self): + metric = _mock_metric() + metric.measure = MagicMock(side_effect=RuntimeError("LLM timeout")) + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input()) + + assert result.errorCode == "METRIC_ERROR" + assert "LLM timeout" in result.errorMessage + + def test_missing_params_error_caught(self): + metric = _mock_metric() + + MissingTestCaseParamsError = type("MissingTestCaseParamsError", (Exception,), {}) + metric.measure = MagicMock( + side_effect=MissingTestCaseParamsError("retrieval_context is required") + ) + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input()) + + assert result.errorCode == "MISSING_REQUIRED_FIELD" + assert "retrieval_context" in result.errorMessage + assert "customer_mapper" in result.errorMessage + + def test_never_raises(self): + metric = _mock_metric() + metric.measure = MagicMock(side_effect=Exception("unexpected")) + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input()) + + assert isinstance(result, EvaluatorOutput) + assert result.errorCode is not None + + +class TestDeepEvalAdapterEdgeCases: + def test_metric_with_no_reason(self): + metric = _mock_metric(score=0.8, reason=None) + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input()) + + assert result.explanation == "" + + def test_metric_score_zero(self): + metric = _mock_metric(score=0.0, threshold=0.5) + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input()) + + assert result.value == 0.0 + assert result.label == "Fail" + + def test_default_threshold_when_missing(self): + metric = _mock_metric(score=0.6) + del metric.threshold + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_evaluator_input()) + + assert result.label == "Pass" diff --git a/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/__init__.py b/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/test_span_mappers.py b/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/test_span_mappers.py new file mode 100644 index 00000000..864e2274 --- /dev/null +++ b/tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/third_party/span_mappers/test_span_mappers.py @@ -0,0 +1,427 @@ +"""Tests for span mappers.""" + +import json + +import pytest + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.models import ReferenceInput +from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.span_mappers import ( + SpanMapResult, + map_spans, +) + + +def _make_strands_agent_span(span_events, span_id="span1", trace_id="abc123"): + """Build a Strands invoke_agent span with given span_events.""" + return { + "traceId": trace_id, + "spanId": span_id, + "parentSpanId": "parent1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": { + "gen_ai.operation.name": "invoke_agent", + "gen_ai.agent.name": "test_agent", + "gen_ai.system": "strands-agents", + }, + "span_events": span_events, + } + + +def _make_span_event(input_messages=None, output_messages=None): + """Build a span_event body matching real Strands format.""" + body = {} + if input_messages is not None: + body["input"] = {"messages": input_messages} + if output_messages is not None: + body["output"] = {"messages": output_messages} + return {"event_name": "strands.telemetry.tracer", "body": body} + + +def _make_non_strands_span(operation_name="chat"): + """Build a span from a different scope (e.g., botocore).""" + return { + "traceId": "abc123", + "spanId": "other1", + "scope": {"name": "opentelemetry.instrumentation.botocore.bedrock-runtime", "version": "0.54b1"}, + "attributes": {"gen_ai.operation.name": operation_name}, + "span_events": [], + } + + +class TestMapSpansSuccess: + def test_extracts_input_and_output_plain_strings(self): + spans = [ + _make_strands_agent_span([ + _make_span_event( + input_messages=[{"role": "user", "content": "What is AI?"}], + output_messages=[{"role": "assistant", "content": "Artificial intelligence."}], + ) + ]) + ] + + result = map_spans(spans) + + assert result.input == "What is AI?" + assert result.actual_output == "Artificial intelligence." + + def test_extracts_from_real_strands_format(self): + """Test with content format matching real parser_output.json.""" + spans = [ + _make_strands_agent_span([ + _make_span_event( + input_messages=[ + {"role": "system", "content": "You are a travel assistant."}, + {"role": "user", "content": {"content": '[{"text": "What is the weather in Tokyo?"}]'}}, + ], + output_messages=[ + { + "role": "assistant", + "content": {"message": "The weather in Tokyo is sunny.", "finish_reason": "end_turn"}, + } + ], + ) + ]) + ] + + result = map_spans(spans) + + assert result.input == "What is the weather in Tokyo?" + assert result.actual_output == "The weather in Tokyo is sunny." + + def test_multi_turn_uses_first_user_last_assistant(self): + """Multiple span_events (one per turn) — first user input, last assistant output.""" + spans = [ + _make_strands_agent_span([ + _make_span_event( + input_messages=[{"role": "user", "content": {"content": '[{"text": "Plan a trip to Japan"}]'}}], + output_messages=[{"role": "assistant", "content": {"message": "Sure! Let me check flights."}}], + ), + _make_span_event( + input_messages=[{"role": "user", "content": {"content": '[{"text": "What about hotels?"}]'}}], + output_messages=[{"role": "assistant", "content": {"message": "Here are some hotel options."}}], + ), + _make_span_event( + input_messages=[{"role": "user", "content": {"content": '[{"text": "Thanks!"}]'}}], + output_messages=[{"role": "assistant", "content": {"message": "You are welcome! Have a great trip."}}], + ), + ]) + ] + + result = map_spans(spans) + + assert result.input == "Plan a trip to Japan" + assert result.actual_output == "You are welcome! Have a great trip." + + def test_extracts_tool_messages_as_retrieval_context(self): + spans = [ + _make_strands_agent_span([ + _make_span_event( + input_messages=[{"role": "user", "content": "query"}], + output_messages=[ + {"role": "tool", "content": "doc chunk 1"}, + {"role": "tool", "content": "doc chunk 2"}, + {"role": "assistant", "content": "answer"}, + ], + ) + ]) + ] + + result = map_spans(spans) + + assert result.retrieval_context == ["doc chunk 1", "doc chunk 2"] + assert result.context == ["doc chunk 1", "doc chunk 2"] + assert result.actual_output == "answer" + + def test_ignores_non_strands_spans(self): + """Only processes spans with strands.telemetry.tracer scope.""" + spans = [ + _make_non_strands_span(), + _make_strands_agent_span([ + _make_span_event( + input_messages=[{"role": "user", "content": "hello"}], + output_messages=[{"role": "assistant", "content": "hi"}], + ) + ]), + ] + + result = map_spans(spans) + + assert result.input == "hello" + assert result.actual_output == "hi" + + def test_skips_system_messages(self): + spans = [ + _make_strands_agent_span([ + _make_span_event( + input_messages=[ + {"role": "system", "content": "You are helpful."}, + {"role": "user", "content": "Hi"}, + ], + output_messages=[{"role": "assistant", "content": "Hello!"}], + ) + ]) + ] + + result = map_spans(spans) + + assert result.input == "Hi" + assert result.actual_output == "Hello!" + + def test_expected_output_from_reference_inputs(self): + spans = [ + _make_strands_agent_span([ + _make_span_event( + input_messages=[{"role": "user", "content": "q"}], + output_messages=[{"role": "assistant", "content": "a"}], + ) + ]) + ] + refs = [ReferenceInput(expectedResponse={"text": "expected answer"})] + + result = map_spans(spans, reference_inputs=refs) + + assert result.expected_output == "expected answer" + + def test_body_as_json_string(self): + body = { + "input": {"messages": [{"role": "user", "content": "hello"}]}, + "output": {"messages": [{"role": "assistant", "content": "hi"}]}, + } + span = { + "traceId": "t1", + "spanId": "s1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "span_events": [{"event_name": "strands.telemetry.tracer", "body": json.dumps(body)}], + } + + result = map_spans([span]) + + assert result.input == "hello" + assert result.actual_output == "hi" + + def test_to_dict_omits_none(self): + result = SpanMapResult(input="q", actual_output="a") + d = result.to_dict() + + assert d == {"input": "q", "actual_output": "a"} + assert "retrieval_context" not in d + + +class TestMapSpansErrors: + def test_no_strands_scope_raises(self): + spans = [_make_non_strands_span()] + + with pytest.raises(ValueError, match="Could not extract evaluation fields"): + map_spans(spans) + + def test_empty_spans_raises(self): + with pytest.raises(ValueError, match="Could not extract evaluation fields"): + map_spans([]) + + def test_strands_scope_but_no_invoke_agent_raises(self): + spans = [ + { + "traceId": "t1", + "spanId": "s1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": {"gen_ai.operation.name": "chat"}, + "span_events": [ + { + "body": { + "input": {"messages": [{"role": "user", "content": "q"}]}, + "output": {"messages": [{"role": "assistant", "content": "a"}]}, + } + } + ], + } + ] + + with pytest.raises(ValueError, match="Could not extract evaluation fields"): + map_spans(spans) + + def test_invoke_agent_with_empty_events_raises(self): + spans = [_make_strands_agent_span(span_events=[])] + + with pytest.raises(ValueError, match="Could not extract evaluation fields"): + map_spans(spans) + + +class TestMapSpansInlineEvents: + """Tests for unified ADOT format (inline events[]).""" + + def test_extracts_from_inline_events(self): + """Real format from in_memory_spans test data.""" + span = { + "traceId": "4ab9fca604243bbd9454c0a969732697", + "spanId": "966a414a17031f25", + "scope": {"name": "strands.telemetry.tracer", "version": None}, + "attributes": { + "gen_ai.operation.name": "invoke_agent", + "gen_ai.agent.name": "TravelAgent", + }, + "events": [ + { + "name": "gen_ai.user.message", + "attributes": {"content": '[{"text": "Hey, how can you help me"}]'}, + }, + { + "name": "gen_ai.choice", + "attributes": { + "message": "Hello! I'm a travel planning assistant.", + "finish_reason": "end_turn", + }, + }, + ], + "span_events": [], + } + + result = map_spans([span]) + + assert result.input == "Hey, how can you help me" + assert result.actual_output == "Hello! I'm a travel planning assistant." + + def test_inline_events_preferred_over_span_body(self): + """If both events[] and span_events[] exist, inline events win.""" + span = { + "traceId": "t1", + "spanId": "s1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "events": [ + {"name": "gen_ai.user.message", "attributes": {"content": '[{"text": "inline input"}]'}}, + {"name": "gen_ai.choice", "attributes": {"message": "inline output"}}, + ], + "span_events": [ + { + "body": { + "input": {"messages": [{"role": "user", "content": "body input"}]}, + "output": {"messages": [{"role": "assistant", "content": "body output"}]}, + } + } + ], + } + + result = map_spans([span]) + + assert result.input == "inline input" + assert result.actual_output == "inline output" + + def test_falls_back_to_span_body_when_no_inline_events(self): + """Empty events[] -> uses span_events[].body.""" + span = { + "traceId": "t1", + "spanId": "s1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "events": [], + "span_events": [ + { + "body": { + "input": {"messages": [{"role": "user", "content": "body input"}]}, + "output": {"messages": [{"role": "assistant", "content": "body output"}]}, + } + } + ], + } + + result = map_spans([span]) + + assert result.input == "body input" + assert result.actual_output == "body output" + + def test_inline_events_only_response(self): + """Only gen_ai.choice, no gen_ai.user.message -> still returns output.""" + span = { + "traceId": "t1", + "spanId": "s1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "events": [ + {"name": "gen_ai.choice", "attributes": {"message": "response only"}}, + ], + "span_events": [], + } + + result = map_spans([span]) + + assert result.input is None + assert result.actual_output == "response only" + + def test_inline_events_multi_turn(self): + """Multi-turn inline events: first user input, last agent output.""" + span = { + "traceId": "t1", + "spanId": "s1", + "scope": {"name": "strands.telemetry.tracer", "version": ""}, + "attributes": { + "gen_ai.operation.name": "invoke_agent", + "system_prompt": "You are a helpful assistant.", + }, + "events": [ + {"name": "gen_ai.user.message", "attributes": {"content": '[{"text": "Hello"}]'}}, + {"name": "gen_ai.choice", "attributes": {"message": "Hi there!"}}, + {"name": "gen_ai.user.message", "attributes": {"content": '[{"text": "What is 2+2?"}]'}}, + {"name": "gen_ai.choice", "attributes": {"message": "The answer is 4."}}, + ], + "span_events": [], + } + + result = map_spans([span]) + + assert result.input == "Hello" + assert result.actual_output == "The answer is 4." + assert result.system_prompt == "You are a helpful assistant." + + def test_span_body_multi_turn(self): + """Multi-turn span body: first user input, last assistant output, tool outputs as retrieval_context.""" + spans = [ + _make_strands_agent_span([ + _make_span_event( + input_messages=[ + {"role": "system", "content": "You are a travel agent."}, + {"role": "user", "content": "Plan a trip"}, + ], + output_messages=[{"role": "assistant", "content": "Sure! Where to?"}], + ), + _make_span_event( + input_messages=[{"role": "user", "content": "Tokyo"}], + output_messages=[ + {"role": "tool", "content": "Flight info: $500"}, + {"role": "assistant", "content": "Found flights to Tokyo."}, + ], + ), + ]) + ] + + result = map_spans(spans) + + assert result.input == "Plan a trip" + assert result.actual_output == "Found flights to Tokyo." + assert result.system_prompt == "You are a travel agent." + assert result.retrieval_context == ["Flight info: $500"] + + def test_to_dict_includes_system_prompt(self): + """to_dict() includes system_prompt when present.""" + spans = [ + _make_strands_agent_span([ + _make_span_event( + input_messages=[ + {"role": "system", "content": "System prompt here."}, + {"role": "user", "content": "Hi"}, + ], + output_messages=[ + {"role": "tool", "content": "tool result"}, + {"role": "assistant", "content": "Hello!"}, + ], + ), + ]) + ] + + result = map_spans(spans) + d = result.to_dict() + + assert d["input"] == "Hi" + assert d["actual_output"] == "Hello!" + assert d["system_prompt"] == "System prompt here." + assert d["retrieval_context"] == ["tool result"] diff --git a/tests_integ/evaluation/test_third_party_adapters.py b/tests_integ/evaluation/test_third_party_adapters.py new file mode 100644 index 00000000..a9f0eac6 --- /dev/null +++ b/tests_integ/evaluation/test_third_party_adapters.py @@ -0,0 +1,171 @@ +"""Integration tests for third-party evaluation adapters. + +These tests require `deepeval` and `autoevals` packages to be installed. +They verify the full adapter flow from EvaluatorInput through span parsing +to metric execution, using real library metrics (not mocks). + +SETUP: + pip install deepeval autoevals + +RUN: + pytest tests_integ/evaluation/test_third_party_adapters.py -v +""" + +import pytest + +from bedrock_agentcore.evaluation.custom_code_based_evaluators.models import EvaluatorInput, EvaluatorOutput + + +def _make_agent_evaluator_input( + user_prompt="What is the capital of France?", + agent_response="The capital of France is Paris.", + tool_messages=None, +): + """Build an EvaluatorInput with agent-level spans.""" + output_messages = [] + if tool_messages: + for msg in tool_messages: + output_messages.append({"role": "tool", "content": msg}) + output_messages.append({"role": "assistant", "content": agent_response}) + + spans = [ + { + "traceId": "integ-trace-1", + "spanId": "integ-span-1", + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "span_events": [ + { + "body": { + "input": {"messages": [{"role": "user", "content": user_prompt}]}, + "output": {"messages": output_messages}, + } + } + ], + } + ] + return EvaluatorInput( + evaluation_level="TRACE", + session_spans=spans, + target_trace_id="integ-trace-1", + ) + + +class TestDeepEvalAdapterIntegration: + """Integration tests for DeepEvalAdapter with real DeepEval metrics.""" + + @pytest.fixture(autouse=True) + def check_deepeval(self): + """Skip if deepeval is not installed.""" + pytest.importorskip("deepeval") + + def test_bias_metric_passes(self): + from deepeval.metrics import BiasMetric + + from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.deepeval import DeepEvalAdapter + + metric = BiasMetric(threshold=0.5) + adapter = DeepEvalAdapter(metric=metric) + + result = adapter(_make_agent_evaluator_input()) + + assert isinstance(result, EvaluatorOutput) + assert result.value is not None + assert result.label in ("Pass", "Fail") + + def test_missing_retrieval_context_returns_error(self): + from deepeval.metrics import FaithfulnessMetric + + from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.deepeval import DeepEvalAdapter + + metric = FaithfulnessMetric(threshold=0.7) + adapter = DeepEvalAdapter(metric=metric) + + result = adapter( + _make_agent_evaluator_input( + user_prompt="Is the sky blue?", + agent_response="Yes, the sky is blue.", + ) + ) + + assert isinstance(result, EvaluatorOutput) + assert result.errorCode == "MISSING_REQUIRED_FIELD" or result.value is not None + + def test_with_field_mapper(self): + from deepeval.metrics import BiasMetric + + from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.deepeval import DeepEvalAdapter + + metric = BiasMetric(threshold=0.5) + adapter = DeepEvalAdapter( + metric=metric, + field_mapper=lambda ev: { + "input": "Is Python a good language?", + "actual_output": "Python is a versatile programming language used widely.", + }, + ) + + result = adapter(_make_agent_evaluator_input()) + + assert isinstance(result, EvaluatorOutput) + assert result.value is not None + + +class TestAutoevalsAdapterIntegration: + """Integration tests for AutoevalsAdapter with real Autoevals scorers.""" + + @pytest.fixture(autouse=True) + def check_autoevals(self): + """Skip if autoevals is not installed.""" + pytest.importorskip("autoevals") + + def test_factuality_scorer(self): + from autoevals import Factuality + + from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.autoevals import AutoevalsAdapter + + scorer = Factuality() + adapter = AutoevalsAdapter(scorer=scorer) + + evaluator_input = _make_agent_evaluator_input() + evaluator_input.session_spans[0]["span_events"][0]["body"]["output"]["messages"] = [ + {"role": "assistant", "content": "The capital of France is Paris."} + ] + + result = adapter(evaluator_input) + + assert isinstance(result, EvaluatorOutput) + assert result.value is not None + assert result.label in ("Pass", "Fail") + + def test_custom_threshold(self): + from autoevals import Factuality + + from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.autoevals import AutoevalsAdapter + + scorer = Factuality() + adapter = AutoevalsAdapter(scorer=scorer, threshold=0.9) + + result = adapter(_make_agent_evaluator_input()) + + assert isinstance(result, EvaluatorOutput) + assert result.value is not None + + def test_with_field_mapper(self): + from autoevals import Factuality + + from bedrock_agentcore.evaluation.custom_code_based_evaluators.third_party.autoevals import AutoevalsAdapter + + scorer = Factuality() + adapter = AutoevalsAdapter( + scorer=scorer, + field_mapper=lambda ev: { + "input": "What is 2+2?", + "actual_output": "4", + "expected_output": "4", + }, + ) + + result = adapter(_make_agent_evaluator_input()) + + assert isinstance(result, EvaluatorOutput) + assert result.value is not None