feat(experiment): add opt-in model routing for evaluators

src/strands_evals/__init__.py

@@ -4,6 +4,7 @@
 from .evaluation_data_store import EvaluationDataStore
 from .experiment import Experiment
 from .local_file_task_result_store import LocalFileTaskResultStore
+from .model_router import ModelRouter, RoutingRule
 from .simulation import ActorSimulator, UserSimulator
 from .telemetry import StrandsEvalsTelemetry, get_tracer
 from .types.detector import DiagnosisConfig
@@ -13,6 +14,8 @@
     "DiagnosisConfig",
     "Experiment",
     "Case",
+    "ModelRouter",
+    "RoutingRule",
     "LocalFileTaskResultStore",
     "EvaluationDataStore",
     "EvaluationReport",

src/strands_evals/experiment.py

@@ -42,6 +42,7 @@
 from .evaluators.tool_parameter_accuracy_evaluator import ToolParameterAccuracyEvaluator
 from .evaluators.tool_selection_accuracy_evaluator import ToolSelectionAccuracyEvaluator
 from .evaluators.trajectory_evaluator import TrajectoryEvaluator
+from .model_router import ModelRouter
 from .telemetry import get_tracer, serialize
 from .telemetry._cloudwatch_logger import _send_to_cloudwatch
 from .types.detector import DiagnosisConfig
@@ -93,6 +94,8 @@ class Experiment(Generic[InputT, OutputT]):
     Attributes:
         cases: A list of test cases in the experiment.
         evaluators: The list of evaluators to be used on the test cases.
+        model_router: Optional ModelRouter that selects a model per (evaluator, case)
+            pair. Only applies to evaluators constructed without an explicit model.
 
     Example:
         experiment = Experiment[str, str](
@@ -123,13 +126,15 @@ def __init__(
         cases: list[Case[InputT, OutputT]] | None = None,
         evaluators: list[Evaluator[InputT, OutputT]] | None = None,
         diagnosis_config: DiagnosisConfig | None = None,
+        model_router: ModelRouter | None = None,
     ):
         self._cases = cases or []
         self._evaluators = evaluators or [Evaluator()]
         self._tracer = get_tracer()
 
         self._config_id = os.environ.get("EVALUATION_RESULTS_LOG_GROUP", "default-strands-evals")
         self._diagnosis_config = diagnosis_config
+        self._model_router = model_router
 
     @property
     def cases(self) -> list[Case[InputT, OutputT]]:
@@ -338,6 +343,8 @@ async def _run_evaluator(
         Returns:
             A dict with evaluator_name, test_pass, score, reason, and detailed_results.
         """
+        if self._model_router is not None:
+            evaluator = self._model_router.route(evaluator, evaluation_context)
 
         @retry(
             retry=retry_if_exception(is_throttling_error),

src/strands_evals/model_router.py

@@ -0,0 +1,151 @@
+"""Model routing for evaluators.
+
+Lets an Experiment match model tier to evaluation complexity: cheap/fast models
+for structural or rubric-style checks, stronger models for nuanced judgment.
+
+Routing is opt-in and advisory:
+- An evaluator constructed with an explicit ``model=`` always keeps that model.
+- Evaluators whose ``model`` is None (the default) are routed through the
+  router's rules; the first matching rule wins.
+- If no rule matches, the router's ``default_model`` (if any) is used;
+  otherwise the evaluator keeps its default model resolution.
+
+Example::
+
+    from strands_evals import Experiment, ModelRouter, RoutingRule
+
+    router = ModelRouter(rules=[
+        # Structural checks -> fast model
+        RoutingRule(
+            evaluator_types=["ToolSelectionAccuracyEvaluator", "ToolParameterAccuracyEvaluator"],
+            model="us.anthropic.claude-haiku-4-5-20251001-v1:0",
+        ),
+        # Complex judgment on long traces -> strongest model
+        RoutingRule(
+            evaluator_types=["GoalSuccessRateEvaluator"],
+            model="us.anthropic.claude-opus-4-1-20250805-v1:0",
+            condition=lambda case: len(str(case.actual_trajectory or "")) > 10_000,
+        ),
+    ])
+
+    experiment = Experiment(cases=cases, evaluators=evaluators, model_router=router)
+"""
+
+import copy
+import logging
+from typing import TYPE_CHECKING, Callable
+
+from strands.models.model import Model
+
+if TYPE_CHECKING:
+    from .evaluators.evaluator import Evaluator
+    from .types.evaluation import EvaluationData
+
+logger = logging.getLogger(__name__)
+
+
+class RoutingRule:
+    """A rule mapping evaluator types (and optionally case properties) to a model.
+
+    Attributes:
+        evaluator_types: Evaluator classes or class names this rule applies to.
+        model: The model (Model instance or Bedrock model-id string) to use when
+            this rule matches.
+        condition: Optional predicate over the evaluation data. When provided,
+            the rule only matches cases for which it returns True. This enables
+            complexity-based routing (e.g., trace length, span count).
+    """
+
+    def __init__(
+        self,
+        evaluator_types: list[type | str],
+        model: Model | str,
+        condition: Callable[["EvaluationData"], bool] | None = None,
+    ):
+        self._type_names = {t if isinstance(t, str) else t.__name__ for t in evaluator_types}
+        self.model = model
+        self.condition = condition
+
+    def matches(self, evaluator: "Evaluator", evaluation_data: "EvaluationData") -> bool:
+        """Check whether this rule applies to the given evaluator and case.
+
+        Args:
+            evaluator: The evaluator about to run.
+            evaluation_data: The evaluation context for the current case.
+
+        Returns:
+            True if the evaluator's type is covered by this rule and the
+            condition (if any) holds for the case.
+        """
+        if evaluator.get_type_name() not in self._type_names:
+            return False
+        if self.condition is not None:
+            try:
+                return bool(self.condition(evaluation_data))
+            except Exception as e:
+                logger.warning(
+                    "rule_types=<%s>, error=<%s> | routing condition raised, skipping rule", self._type_names, e
+                )
+                return False
+        return True
+
+
+class ModelRouter:
+    """Selects a model for each (evaluator, case) pair using an ordered rule list.
+
+    Attributes:
+        rules: Ordered list of RoutingRule; the first matching rule wins.
+        default_model: Model used when no rule matches. None means "leave the
+            evaluator's own model resolution untouched".
+    """
+
+    def __init__(self, rules: list[RoutingRule] | None = None, default_model: Model | str | None = None):
+        self.rules = rules or []
+        self.default_model = default_model
+
+    def select_model(self, evaluator: "Evaluator", evaluation_data: "EvaluationData") -> Model | str | None:
+        """Select a model for an evaluator run, or None to keep the evaluator's default.
+
+        Args:
+            evaluator: The evaluator about to run.
+            evaluation_data: The evaluation context for the current case.
+
+        Returns:
+            The routed model, the router default, or None when routing does not apply.
+        """
+        for rule in self.rules:
+            if rule.matches(evaluator, evaluation_data):
+                return rule.model
+        return self.default_model
+
+    def route(self, evaluator: "Evaluator", evaluation_data: "EvaluationData") -> "Evaluator":
+        """Return the evaluator to run, applying model routing when appropriate.
+
+        An evaluator with an explicit model (its ``model`` attribute is set) is
+        returned unchanged — user configuration always wins. Evaluators without
+        a ``model`` attribute (e.g., deterministic evaluators) are also returned
+        unchanged. Otherwise a shallow copy with the routed model is returned so
+        the shared evaluator instance is never mutated across concurrent workers.
+
+        Args:
+            evaluator: The evaluator about to run.
+            evaluation_data: The evaluation context for the current case.
+
+        Returns:
+            Either the original evaluator or a shallow copy carrying the routed model.
+        """
+        if getattr(evaluator, "model", None) is not None or not hasattr(evaluator, "model"):
+            return evaluator
+
+        model = self.select_model(evaluator, evaluation_data)
+        if model is None:
+            return evaluator
+
+        routed = copy.copy(evaluator)
+        routed.model = model
+        logger.debug(
+            "evaluator=<%s>, model=<%s> | routed evaluator to model",
+            evaluator.get_name(),
+            model if isinstance(model, str) else type(model).__name__,
+        )
+        return routed

tests/strands_evals/test_model_router.py

@@ -0,0 +1,180 @@
+from unittest.mock import Mock, patch
+
+import pytest
+
+from strands_evals import Case, Experiment, ModelRouter, RoutingRule
+from strands_evals.evaluators import OutputEvaluator, TrajectoryEvaluator
+from strands_evals.evaluators.deterministic import Contains
+from strands_evals.types import EvaluationData, EvaluationOutput
+
+FAST_MODEL = "us.anthropic.claude-haiku-4-5-20251001-v1:0"
+STRONG_MODEL = "us.anthropic.claude-opus-4-1-20250805-v1:0"
+
+
+@pytest.fixture
+def evaluation_data():
+    return EvaluationData(input="What is 2+2?", actual_output="4", expected_output="4", name="math_test")
+
+
+class TestRoutingRule:
+    def test_matches_by_class_name_string(self, evaluation_data):
+        rule = RoutingRule(evaluator_types=["OutputEvaluator"], model=FAST_MODEL)
+        evaluator = OutputEvaluator(rubric="Test rubric")
+
+        assert rule.matches(evaluator, evaluation_data) is True
+
+    def test_matches_by_class_object(self, evaluation_data):
+        rule = RoutingRule(evaluator_types=[OutputEvaluator], model=FAST_MODEL)
+        evaluator = OutputEvaluator(rubric="Test rubric")
+
+        assert rule.matches(evaluator, evaluation_data) is True
+
+    def test_does_not_match_other_evaluator_types(self, evaluation_data):
+        rule = RoutingRule(evaluator_types=[TrajectoryEvaluator], model=FAST_MODEL)
+        evaluator = OutputEvaluator(rubric="Test rubric")
+
+        assert rule.matches(evaluator, evaluation_data) is False
+
+    def test_condition_gates_matching(self, evaluation_data):
+        rule = RoutingRule(
+            evaluator_types=[OutputEvaluator],
+            model=STRONG_MODEL,
+            condition=lambda case: len(str(case.actual_output)) > 100,
+        )
+        evaluator = OutputEvaluator(rubric="Test rubric")
+
+        assert rule.matches(evaluator, evaluation_data) is False
+
+        long_output = EvaluationData(input="q", actual_output="x" * 200)
+        assert rule.matches(evaluator, long_output) is True
+
+    def test_condition_exception_skips_rule(self, evaluation_data):
+        def broken_condition(case):
+            raise RuntimeError("boom")
+
+        rule = RoutingRule(evaluator_types=[OutputEvaluator], model=FAST_MODEL, condition=broken_condition)
+        evaluator = OutputEvaluator(rubric="Test rubric")
+
+        assert rule.matches(evaluator, evaluation_data) is False
+
+
+class TestModelRouter:
+    def test_first_matching_rule_wins(self, evaluation_data):
+        router = ModelRouter(
+            rules=[
+                RoutingRule(evaluator_types=[OutputEvaluator], model=FAST_MODEL),
+                RoutingRule(evaluator_types=[OutputEvaluator], model=STRONG_MODEL),
+            ]
+        )
+        evaluator = OutputEvaluator(rubric="Test rubric")
+
+        assert router.select_model(evaluator, evaluation_data) == FAST_MODEL
+
+    def test_default_model_when_no_rule_matches(self, evaluation_data):
+        router = ModelRouter(
+            rules=[RoutingRule(evaluator_types=[TrajectoryEvaluator], model=FAST_MODEL)],
+            default_model=STRONG_MODEL,
+        )
+        evaluator = OutputEvaluator(rubric="Test rubric")
+
+        assert router.select_model(evaluator, evaluation_data) == STRONG_MODEL
+
+    def test_no_rule_and_no_default_returns_none(self, evaluation_data):
+        router = ModelRouter(rules=[RoutingRule(evaluator_types=[TrajectoryEvaluator], model=FAST_MODEL)])
+        evaluator = OutputEvaluator(rubric="Test rubric")
+
+        assert router.select_model(evaluator, evaluation_data) is None
+
+    def test_route_returns_copy_with_routed_model(self, evaluation_data):
+        router = ModelRouter(rules=[RoutingRule(evaluator_types=[OutputEvaluator], model=FAST_MODEL)])
+        evaluator = OutputEvaluator(rubric="Test rubric")
+
+        routed = router.route(evaluator, evaluation_data)
+
+        assert routed is not evaluator
+        assert routed.model == FAST_MODEL
+        # Original instance is never mutated
+        assert evaluator.model is None
+
+    def test_route_preserves_explicit_model(self, evaluation_data):
+        """An evaluator constructed with an explicit model must never be re-routed."""
+        router = ModelRouter(rules=[RoutingRule(evaluator_types=[OutputEvaluator], model=FAST_MODEL)])
+        evaluator = OutputEvaluator(rubric="Test rubric", model=STRONG_MODEL)
+
+        routed = router.route(evaluator, evaluation_data)
+
+        assert routed is evaluator
+        assert routed.model == STRONG_MODEL
+
+    def test_route_leaves_evaluator_without_model_attribute_unchanged(self, evaluation_data):
+        """Deterministic evaluators have no model attribute and must pass through."""
+        router = ModelRouter(
+            rules=[RoutingRule(evaluator_types=["Contains"], model=FAST_MODEL)], default_model=FAST_MODEL
+        )
+        evaluator = Contains(value="4")
+
+        routed = router.route(evaluator, evaluation_data)
+
+        assert routed is evaluator
+
+    def test_route_without_match_returns_original(self, evaluation_data):
+        router = ModelRouter(rules=[RoutingRule(evaluator_types=[TrajectoryEvaluator], model=FAST_MODEL)])
+        evaluator = OutputEvaluator(rubric="Test rubric")
+
+        routed = router.route(evaluator, evaluation_data)
+
+        assert routed is evaluator
+        assert routed.model is None
+
+
+class TestExperimentIntegration:
+    @patch("strands_evals.evaluators.output_evaluator.Agent")
+    def test_experiment_routes_evaluator_model(self, mock_agent_class):
+        mock_agent = Mock()
+        mock_result = Mock()
+        mock_result.structured_output = EvaluationOutput(score=1.0, test_pass=True, reason="ok")
+        mock_agent.return_value = mock_result
+
+        async def mock_invoke_async(*args, **kwargs):
+            return mock_result
+
+        mock_agent.invoke_async = mock_invoke_async
+        mock_agent_class.return_value = mock_agent
+
+        router = ModelRouter(rules=[RoutingRule(evaluator_types=[OutputEvaluator], model=FAST_MODEL)])
+        evaluator = OutputEvaluator(rubric="Output must be correct.")
+        experiment = Experiment(
+            cases=[Case(name="math", input="What is 2+2?", expected_output="4")],
+            evaluators=[evaluator],
+            model_router=router,
+        )
+
+        report = experiment.run_evaluations(lambda case: "4")
+
+        # The judge agent was constructed with the routed model
+        assert mock_agent_class.call_args[1]["model"] == FAST_MODEL
+        assert report.test_passes == [True]
+        # The user's evaluator instance is untouched
+        assert evaluator.model is None
+
+    @patch("strands_evals.evaluators.output_evaluator.Agent")
+    def test_experiment_without_router_keeps_default_model(self, mock_agent_class):
+        mock_agent = Mock()
+        mock_result = Mock()
+        mock_result.structured_output = EvaluationOutput(score=1.0, test_pass=True, reason="ok")
+        mock_agent.return_value = mock_result
+
+        async def mock_invoke_async(*args, **kwargs):
+            return mock_result
+
+        mock_agent.invoke_async = mock_invoke_async
+        mock_agent_class.return_value = mock_agent
+
+        experiment = Experiment(
+            cases=[Case(name="math", input="What is 2+2?", expected_output="4")],
+            evaluators=[OutputEvaluator(rubric="Output must be correct.")],
+        )
+
+        experiment.run_evaluations(lambda case: "4")
+
+        assert mock_agent_class.call_args[1]["model"] is None