Gen AI Evaluation Toolkit on AWS

v2.0.0

Copyright © 2026 Amazon Web Services, Inc. and/or its affiliates. All rights reserved. Amazon’s trademarks and trade dress may not be used in connection with any product or service that is not Amazon’s, in any manner that is likely to cause confusion among customers, or in any manner that disparages or discredits Amazon. All other trademarks not owned by Amazon are the property of their respective owners, who may or may not be affiliated with, connected to, or sponsored by Amazon.

Table of contents

• Evaluator SDK reference
  • Evaluator
  • Models
  • Scorer

Evaluator SDK reference

The Evaluator SDK provides the framework for building custom evaluators that integrate with the Gen AI Evaluation Toolkit on AWS.

Evaluator

evaluator_sdk.evaluator.Evaluator

Simple evaluator that processes evaluation jobs.

Uses the generated DefaultApi client for API calls.

evaluate(testcases: list[InlineTestCase | dict[str, Any]] | None = None, dataset_id: str | None = None, dataset_version: str | None = None, scorers: dict[str, dict[str, Any] | None] | None = None, app: dict[str, Any] | None = None, experiment_name: str | None = None, return_results: bool = True, app_invoke_name: str | None = None, annotation_filter: dict[str, Any] | None = None) -> EvaluationResponse

Run a full evaluation: app_invoke + scoring.

Local execution (default). app_invoke must be set and at least one scorer registered. Use app to select which app_invoke function to call and pass runtime parameters::

evaluator.evaluate(
    testcases=...,
    app={"name": "my_fn", "params": {"agent_arn": "arn:..."}},
    scorers={"ragas": {"scores": {"faithfulness": None}}},
)

Remote execution. Add an evaluatorName to a task's config to run it on a deployed evaluator worker instead of in-process — for the app (app invocation) and/or individual scorers. A task with evaluatorName runs remotely (its name need not be registered locally); without one it runs locally and must be registered. For a fully-remote evaluation, route BOTH the app and every scorer::

evaluator.evaluate(
    testcases=...,
    app={"evaluatorName": "my-worker", "name": "my_fn", "params": {...}},
    scorers={"ragas": {"evaluatorName": "builtin",
                       "scores": {"faithfulness": None}}},
)

The call blocks until the remote job finishes and returns the results the workers wrote. Mixing local and remote tasks in one call is not yet supported (raises EvaluatorError); route everything one way or the other. See :meth:score for scoring pre-existing outputs remotely.

Use annotation_filter with dataset_id to evaluate only test cases matching the annotation condition::

evaluator.evaluate(
    dataset_id="ds-123",
    annotation_filter={
        "condition": {"stringEnumCondition": {
            "name": "SENTIMENT", "value": "NEGATIVE", "comparator": "=",
        }}
    },
    scorers={"llm_as_judge": None},
)

score(testcases: list[dict[str, Any]] | None = None, source_report_id: str | None = None, scorers: dict[str, dict[str, Any] | None] | None = None, experiment_name: str | None = None, return_results: bool = True) -> EvaluationResponse

Score pre-existing outputs — no app_invoke.

Provide either testcases (dicts with input and output keys) or source_report_id (an existing report whose results will be copied and scored).

Local scoring (default) runs registered scorers in-process::

evaluator.score(testcases=[...], scorers={"exact_match": None})

Remote scoring. Add an evaluatorName to a scorer's config to run it on a deployed evaluator worker. This is how remote-only evaluators such as agent-as-judge are driven through the SDK — the scorer need not be registered locally::

evaluator.score(
    testcases=[{"input": {...}, "output": {...}}],
    scorers={"agent_as_judge": {"evaluatorName": "agent-as-judge",
                                "scores": {"correctness": None}}},
)

# or score results already in a report:
evaluator.score(
    source_report_id="report-123",
    scorers={"agent_as_judge": {"evaluatorName": "agent-as-judge",
                                "scores": {"correctness": None}}},
)

The call blocks until the remote job completes and returns the scored results. Scorers may route to multiple different evaluators (they fan out in parallel). At least one scorer must be registered locally or carry an evaluatorName.

invoke(testcases: list[InlineTestCase | dict[str, Any]] | None = None, dataset_id: str | None = None, dataset_version: str | None = None, app: dict[str, Any] | None = None, experiment_name: str | None = None, return_results: bool = True, app_invoke_name: str | None = None, annotation_filter: dict[str, Any] | None = None) -> EvaluationResponse

Run app_invoke only — no scoring.

Local execution (default) requires app_invoke to be set; use app to select which function to call and pass runtime parameters::

evaluator.invoke(
    testcases=...,
    app={"name": "my_fn", "params": {"agent_arn": "arn:..."}},
)

Remote execution. Add an evaluatorName to the app config to run app invocation on a deployed evaluator worker instead of in-process (name then selects the worker's app-invoke function)::

evaluator.invoke(
    testcases=...,
    app={"evaluatorName": "my-worker", "name": "my_fn",
         "params": {...}},
)

The call blocks until the remote job completes and returns the results.

Use annotation_filter with dataset_id to invoke only on test cases matching the annotation condition::

evaluator.invoke(
    dataset_id="ds-123",
    annotation_filter={"condition": {
        "stringEnumCondition": {
            "name": "TIER", "value": "HIGH", "comparator": "=",
        }
    }},
)

evaluator_handler(event: dict[str, Any], context: Any) -> HandlerResult

Lambda handler for remotely-triggered evaluation jobs.

Receives an EventBridge event with job details in event["detail"]. The job and report must already exist (created by the submission API).

Supports three job types: - FULL: app_invoke + scoring (requires app_invoke and scorers) - APP_INVOKE: app_invoke only (requires app_invoke) - SCORE: scoring only (requires scorers)

Models

Public types, dataclasses, and type aliases for the Evaluator SDK.

AppInvokeFn = Callable[..., Any] module-attribute

ExpectedType = TypeVar('ExpectedType') module-attribute

InputType = TypeVar('InputType') module-attribute

MetadataType = TypeVar('MetadataType') module-attribute

OutputType = TypeVar('OutputType') module-attribute

ScorerFn = Callable[..., Any] module-attribute

TestCase = InlineTestCase module-attribute

EvaluationResponse dataclass

Response from evaluate() — contains results + identifiers.

job_id: str instance-attribute

report_id: str instance-attribute

results: list[EvaluationResult] | None instance-attribute

score_summary: dict[str, dict[str, float | int]] property

Per-scorer summary: count, errors, and mean value (excluding errors).

total_errors: int property

total_results: int property

total_with_output: int property

print_results() -> None

Print each result to stdout.

print_summary() -> None

Print a summary of the evaluation.

EvaluationResult dataclass

Result of evaluating a single test case.

context: dict[str, Any] | None = None class-attribute instance-attribute

error: str | None = None class-attribute instance-attribute

expected: dict[str, Any] | None = None class-attribute instance-attribute

input_data: dict[str, Any] instance-attribute

metadata: dict[str, Any] | None = None class-attribute instance-attribute

output: dict[str, Any] | None = None class-attribute instance-attribute

scores: dict[str, ScoreResult] = field(default_factory=dict) class-attribute instance-attribute

EvaluatorError

Base exception for evaluator errors.

ScoreResult dataclass

Result of a single score.

Attributes:

Name	Type	Description
value	float | None	The numeric score (typically 0-1).
error	str | None	Error message if scoring failed.
reason	str | None	Human-readable explanation of the score.
context	dict[str, Any] | None	Optional free-form structured detail (e.g. a per-requirement breakdown). Use it to attach machine-readable metadata that consumers can parse alongside value/reason. It is persisted and returned by the API (surfaced as ScoreOutput.context). Leave it None if the scorer produces no structured detail.

context: dict[str, Any] | None = None class-attribute instance-attribute

error: str | None = None class-attribute instance-attribute

reason: str | None = None class-attribute instance-attribute

value: float | None = None class-attribute instance-attribute

TestResult dataclass

The single object handed to scorers — a test case that has been run.

Carries the full test-case core (input, expected, context, metadata) plus the execution artifacts (output, app_metrics, error). Scorers receive this as the sole test_result data argument; scoring configuration (params, score_name, score_params) is passed separately.

app_metrics: dict[str, Any] | None = None class-attribute instance-attribute

context: dict[str, Any] | None = None class-attribute instance-attribute

error: str | None = None class-attribute instance-attribute

expected: dict[str, Any] | None = None class-attribute instance-attribute

input: dict[str, Any] | None = None class-attribute instance-attribute

metadata: dict[str, Any] | None = None class-attribute instance-attribute

output: dict[str, Any] | None = None class-attribute instance-attribute

Scorer

Scorer ABC, score decorator, and scorer helpers.

Scorer

Base class for plugin scorers that produce multiple scores.

Use @scorer to declare individual scoring methods::

class MyScorer(Scorer):
    name = "quality"  # optional, defaults to class name

    @scorer
    def length(self, test_result):
        return 1.0 if len(test_result.output.get("text", "")) > 0 else 0.0

    @scorer
    def prefix(self, test_result):
        return 1.0 if test_result.output.get("text", "").startswith("Response:") else 0.0

Each method can return: - float or ScoreResult — single score, named after the method - dict[str, float | ScoreResult] — multiple dynamic scores

Alternatively, override score() directly for full control.

score(test_result: TestResult, scores: dict[str, dict[str, Any] | None] | None = None, params: dict[str, Any] | None = None) -> dict[str, float | ScoreResult]

Dispatch to @scorer methods. Override for full control.

scorer(fn: Any = None, *, dynamic: bool = False) -> Any

Mark a method as a score on a Scorer subclass.

::

@scorer
def length(self, test_result):
    return 1.0 if len(test_result.output.get("text", "")) > 0 else 0.0

Use dynamic=True to mark a fallback handler for score names that don't have a dedicated @scorer method. Dynamic handlers receive score_name and score_params in addition to the usual kwargs::

@scorer(dynamic=True)
def handle(self, score_name, score_params, test_result, params):
    ...

Methods receive kwargs filtered by signature (test_result, params, score_name, score_params). Return a float, ScoreResult, or dict for dynamic multi-score results.