feat: add custom code-based evaluator decorator and typed models (#383)

* feat: add custom code-based evaluator decorator and typed models

Add @custom_code_based_evaluator() decorator that wraps typed Python
functions into Lambda handlers for Bedrock AgentCore evaluation.

- EvaluatorInput/EvaluatorOutput Pydantic models for typed Lambda I/O
- Decorator parses Lambda events, handles null evaluationTarget, and
  serializes results via model_dump()
- .unwrapped attribute for unit testing without Lambda overhead
- Context passthrough via signature inspection
- errorCode/errorMessage fields for evaluation failures

* fix: apply ruff formatting to decorator.py

Author: jariy17

src/bedrock_agentcore/evaluation/__init__.py

@@ -1,6 +1,11 @@
 """AgentCore Evaluation: EvaluationClient, OnDemandEvaluationDatasetRunner, and Strands integration."""
 
 from bedrock_agentcore.evaluation.client import EvaluationClient, ReferenceInputs
+from bedrock_agentcore.evaluation.custom_code_based_evaluators import (
+    EvaluatorInput,
+    EvaluatorOutput,
+    custom_code_based_evaluator,
+)
 from bedrock_agentcore.evaluation.runner.dataset_providers import (
     DatasetProvider,
     FileDatasetProvider,
@@ -47,11 +52,13 @@
     "EvaluationClient",
     "EvaluationResult",
     "EvaluationRunConfig",
-    "OnDemandEvaluationDatasetRunner",
     "EvaluatorConfig",
+    "EvaluatorInput",
+    "EvaluatorOutput",
     "EvaluatorResult",
     "FileDatasetProvider",
     "Input",
+    "OnDemandEvaluationDatasetRunner",
     "ReferenceInputs",
     "Scenario",
     "ScenarioExecutionResult",
@@ -62,6 +69,7 @@
     "Turn",
     "PredefinedScenario",
     "PredefinedScenarioExecutor",
+    "custom_code_based_evaluator",
     "convert_strands_to_adot",
     "create_strands_evaluator",
     "fetch_spans_from_cloudwatch",

src/bedrock_agentcore/evaluation/custom_code_based_evaluators/__init__.py

@@ -0,0 +1,10 @@
+"""Code-based evaluator support for AgentCore Evaluation."""
+
+from bedrock_agentcore.evaluation.custom_code_based_evaluators.decorator import custom_code_based_evaluator
+from bedrock_agentcore.evaluation.custom_code_based_evaluators.models import EvaluatorInput, EvaluatorOutput
+
+__all__ = [
+    "custom_code_based_evaluator",
+    "EvaluatorInput",
+    "EvaluatorOutput",
+]

src/bedrock_agentcore/evaluation/custom_code_based_evaluators/decorator.py

@@ -0,0 +1,56 @@
+"""Decorator that adapts a typed evaluator function into a Lambda handler."""
+
+import functools
+import logging
+
+from bedrock_agentcore.evaluation.custom_code_based_evaluators.models import EvaluatorInput, EvaluatorOutput
+
+logger = logging.getLogger(__name__)
+
+
+def custom_code_based_evaluator():
+    """Decorator that wraps a typed evaluator function as a Lambda handler.
+
+    The decorated function receives an ``EvaluatorInput`` and the Lambda
+    ``context``, and returns an ``EvaluatorOutput``. The decorator handles
+    parsing the raw Lambda event dict into ``EvaluatorInput`` and serializing
+    the ``EvaluatorOutput`` into the response contract expected by the
+    evaluation service.
+
+    Must be called with parentheses: ``@custom_code_based_evaluator()``.
+
+    Example::
+
+        @custom_code_based_evaluator()
+        def handler(input: EvaluatorInput, context) -> EvaluatorOutput:
+            return EvaluatorOutput(value=1.0, label="Pass")
+    """
+
+    def decorator(fn):
+        @functools.wraps(fn)
+        def lambda_handler(event, context=None):
+            logger.debug("Raw Lambda event: %s", event)
+
+            target = event.get("evaluationTarget") or {}
+            trace_ids = target.get("traceIds") or []
+            span_ids = target.get("spanIds") or []
+
+            evaluator_input = EvaluatorInput(
+                evaluation_level=event["evaluationLevel"],
+                session_spans=event["evaluationInput"]["sessionSpans"],
+                target_trace_id=trace_ids[0] if trace_ids else None,
+                target_span_id=span_ids[0] if span_ids else None,
+                schema_version=event.get("schemaVersion", "1.0"),
+            )
+
+            result = fn(evaluator_input, context)
+
+            if not isinstance(result, EvaluatorOutput):
+                raise TypeError(f"Evaluator must return an EvaluatorOutput, got {type(result).__name__}")
+
+            return result.model_dump()
+
+        lambda_handler.unwrapped = fn
+        return lambda_handler
+
+    return decorator

src/bedrock_agentcore/evaluation/custom_code_based_evaluators/models.py

@@ -0,0 +1,39 @@
+"""Typed models for code-based evaluator Lambda input and output."""
+
+from typing import Dict, List, Optional
+
+from pydantic import BaseModel
+
+
+class EvaluatorInput(BaseModel):
+    """Parsed input for a code-based evaluator Lambda function.
+
+    Attributes:
+        evaluation_level: The evaluation granularity - "SESSION", "TRACE", or "TOOL_CALL".
+        session_spans: Raw ADOT span dicts from the evaluation service.
+        target_trace_id: The target trace ID (set for TRACE level, None otherwise).
+        target_span_id: The target span ID (set for TOOL_CALL level, None otherwise).
+        schema_version: Schema version of the Lambda contract.
+    """
+
+    evaluation_level: str
+    session_spans: List[Dict]
+    target_trace_id: Optional[str] = None
+    target_span_id: Optional[str] = None
+    schema_version: str = "1.0"
+
+
+class EvaluatorOutput(BaseModel):
+    """Result returned by a code-based evaluator function.
+
+    Attributes:
+        value: Numerical score for the evaluation.
+        label: Categorical label (e.g. "Pass", "Fail"). Required.
+        explanation: Optional explanation of the evaluation result.
+    """
+
+    value: Optional[float] = None
+    label: str
+    explanation: Optional[str] = None
+    errorCode: Optional[str] = None
+    errorMessage: Optional[str] = None

tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/__init__.py

@@ -0,0 +1,177 @@
+"""Tests for the @custom_code_based_evaluator decorator."""
+
+import pytest
+
+from bedrock_agentcore.evaluation.custom_code_based_evaluators import (
+    EvaluatorInput,
+    EvaluatorOutput,
+    custom_code_based_evaluator,
+)
+
+
+def _make_event(level="TRACE", trace_ids=None, span_ids=None):
+    """Build a raw Lambda event dict."""
+    event = {
+        "schemaVersion": "1.0",
+        "evaluationLevel": level,
+        "evaluationInput": {
+            "sessionSpans": [
+                {"traceId": "abc123", "spanId": "span1", "name": "Agent", "attributes": {}},
+            ]
+        },
+        "evaluationTarget": {},
+    }
+    if trace_ids is not None:
+        event["evaluationTarget"]["traceIds"] = trace_ids
+    if span_ids is not None:
+        event["evaluationTarget"]["spanIds"] = span_ids
+    return event
+
+
+class TestDecoratorWithRawEvent:
+    def test_trace_level(self):
+        @custom_code_based_evaluator()
+        def handler(inp, context):
+            return EvaluatorOutput(value=1.0, label="Pass", explanation="Valid")
+
+        event = _make_event(level="TRACE", trace_ids=["abc123"])
+        result = handler(event)
+
+        assert result["value"] == 1.0
+        assert result["label"] == "Pass"
+        assert result["explanation"] == "Valid"
+
+    def test_session_level(self):
+        @custom_code_based_evaluator()
+        def handler(inp, context):
+            return EvaluatorOutput(value=0.5, label="Partial")
+
+        event = _make_event(level="SESSION")
+        result = handler(event)
+
+        assert result["value"] == 0.5
+        assert result["label"] == "Partial"
+        assert result["explanation"] is None
+
+    def test_tool_call_level(self):
+        @custom_code_based_evaluator()
+        def handler(inp, context):
+            return EvaluatorOutput(value=0.0, label="Fail")
+
+        event = _make_event(level="TOOL_CALL", span_ids=["span1"])
+        result = handler(event)
+
+        assert result["value"] == 0.0
+        assert result["label"] == "Fail"
+
+    def test_default_schema_version(self):
+        @custom_code_based_evaluator()
+        def handler(inp, context):
+            assert inp.schema_version == "1.0"
+            return EvaluatorOutput(value=1.0, label="Pass")
+
+        event = _make_event()
+        del event["schemaVersion"]
+        handler(event)
+
+
+class TestExceptionPropagation:
+    def test_exception_propagates(self):
+        @custom_code_based_evaluator()
+        def handler(inp, context):
+            raise RuntimeError("boom")
+
+        event = _make_event()
+        with pytest.raises(RuntimeError, match="boom"):
+            handler(event)
+
+
+class TestFunctoolsWraps:
+    def test_preserves_name(self):
+        @custom_code_based_evaluator()
+        def my_evaluator(inp, context):
+            return EvaluatorOutput(value=1.0, label="Pass")
+
+        assert my_evaluator.__name__ == "my_evaluator"
+
+    def test_preserves_module(self):
+        @custom_code_based_evaluator()
+        def my_evaluator(inp, context):
+            return EvaluatorOutput(value=1.0, label="Pass")
+
+        assert my_evaluator.__module__ == __name__
+
+
+class TestContextPassthrough:
+    def test_context_passed_to_function(self):
+        received_context = []
+
+        @custom_code_based_evaluator()
+        def handler(inp, context):
+            received_context.append(context)
+            return EvaluatorOutput(value=1.0, label="Pass")
+
+        mock_context = {"function_name": "my-lambda"}
+        event = _make_event()
+        handler(event, mock_context)
+
+        assert received_context == [mock_context]
+
+    def test_context_defaults_to_none(self):
+        received_context = []
+
+        @custom_code_based_evaluator()
+        def handler(inp, context):
+            received_context.append(context)
+            return EvaluatorOutput(value=1.0, label="Pass")
+
+        event = _make_event()
+        handler(event)
+
+        assert received_context == [None]
+
+
+class TestReturnTypeValidation:
+    def test_rejects_dict_return(self):
+        @custom_code_based_evaluator()
+        def handler(inp, context):
+            return {"value": 1.0, "label": "Pass"}
+
+        event = _make_event()
+        with pytest.raises(TypeError, match="Evaluator must return an EvaluatorOutput, got dict"):
+            handler(event)
+
+    def test_rejects_none_return(self):
+        @custom_code_based_evaluator()
+        def handler(inp, context):
+            return None
+
+        event = _make_event()
+        with pytest.raises(TypeError, match="Evaluator must return an EvaluatorOutput, got NoneType"):
+            handler(event)
+
+
+class TestUnwrapped:
+    def test_unwrapped_returns_evaluator_output(self):
+        @custom_code_based_evaluator()
+        def handler(inp, context):
+            return EvaluatorOutput(value=1.0, label="Pass")
+
+        inp = EvaluatorInput(
+            evaluation_level="TRACE",
+            session_spans=[],
+            target_trace_id="t1",
+            target_span_id=None,
+            schema_version="1.0",
+        )
+        result = handler.unwrapped(inp, None)
+        assert isinstance(result, EvaluatorOutput)
+        assert result.value == 1.0
+        assert result.label == "Pass"
+
+    def test_unwrapped_is_original_function(self):
+        def my_eval(inp, context):
+            return EvaluatorOutput(value=0.5, label="Partial")
+
+        handler = custom_code_based_evaluator()(my_eval)
+        assert handler.unwrapped is my_eval

tests/bedrock_agentcore/evaluation/custom_code_based_evaluators/test_decorator.py

@@ -0,0 +1,56 @@
+"""Tests for EvaluatorInput and EvaluatorOutput dataclasses."""
+
+from bedrock_agentcore.evaluation.custom_code_based_evaluators.models import EvaluatorInput, EvaluatorOutput
+
+
+class TestEvaluatorInput:
+    def test_all_fields(self):
+        inp = EvaluatorInput(
+            evaluation_level="TRACE",
+            session_spans=[{"traceId": "t1", "spanId": "s1"}],
+            target_trace_id="t1",
+            target_span_id=None,
+            schema_version="1.0",
+        )
+        assert inp.evaluation_level == "TRACE"
+        assert inp.session_spans == [{"traceId": "t1", "spanId": "s1"}]
+        assert inp.target_trace_id == "t1"
+        assert inp.target_span_id is None
+        assert inp.schema_version == "1.0"
+
+    def test_session_level_no_targets(self):
+        inp = EvaluatorInput(
+            evaluation_level="SESSION",
+            session_spans=[],
+            target_trace_id=None,
+            target_span_id=None,
+            schema_version="1.0",
+        )
+        assert inp.target_trace_id is None
+        assert inp.target_span_id is None
+
+
+class TestEvaluatorOutput:
+    def test_defaults(self):
+        out = EvaluatorOutput(label="Pass")
+        assert out.value is None
+        assert out.label == "Pass"
+        assert out.explanation is None
+
+    def test_all_fields(self):
+        out = EvaluatorOutput(value=0.85, label="Pass", explanation="Looks good")
+        assert out.value == 0.85
+        assert out.label == "Pass"
+        assert out.explanation == "Looks good"
+
+    def test_label_required(self):
+        import pytest
+        from pydantic import ValidationError
+
+        with pytest.raises(ValidationError):
+            EvaluatorOutput(value=1.0)
+
+    def test_label_only(self):
+        out = EvaluatorOutput(label="Fail")
+        assert out.label == "Fail"
+        assert out.value is None