From 03e92954c7d6b020cc999df9d1138cf5fc9894b1 Mon Sep 17 00:00:00 2001 From: DEBJYOTI PAUL Date: Thu, 23 Jul 2026 08:32:37 -0700 Subject: [PATCH] feat(experiment): add opt-in model routing for evaluators Adds ModelRouter and RoutingRule, plus an optional model_router parameter on Experiment, so evaluation cost/accuracy can be matched to evaluator complexity: fast models for structural/rubric checks, strong models for nuanced judgment. Semantics: - Opt-in: no model_router means behavior is unchanged. - Explicit wins: an evaluator constructed with model=... is never re-routed. - First matching RoutingRule wins; rules match on evaluator type (class or name) and an optional per-case condition (e.g., trace length) enabling complexity-based routing. - Routing returns a shallow copy of the evaluator with the routed model so shared instances are never mutated across concurrent workers; evaluators without a model attribute (deterministic) pass through untouched. - A raising condition logs a warning and skips the rule instead of failing the evaluation. model_router is intentionally excluded from Experiment.to_dict()/from_dict() since routing conditions are arbitrary callables. --- src/strands_evals/__init__.py | 3 + src/strands_evals/experiment.py | 7 + src/strands_evals/model_router.py | 151 +++++++++++++++++++ tests/strands_evals/test_model_router.py | 180 +++++++++++++++++++++++ 4 files changed, 341 insertions(+) create mode 100644 src/strands_evals/model_router.py create mode 100644 tests/strands_evals/test_model_router.py diff --git a/src/strands_evals/__init__.py b/src/strands_evals/__init__.py index 5dbaee36..c2e7755a 100644 --- a/src/strands_evals/__init__.py +++ b/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", diff --git a/src/strands_evals/experiment.py b/src/strands_evals/experiment.py index de42b984..091c806b 100644 --- a/src/strands_evals/experiment.py +++ b/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,6 +126,7 @@ 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()] @@ -130,6 +134,7 @@ def __init__( 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), diff --git a/src/strands_evals/model_router.py b/src/strands_evals/model_router.py new file mode 100644 index 00000000..55ad56f1 --- /dev/null +++ b/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 diff --git a/tests/strands_evals/test_model_router.py b/tests/strands_evals/test_model_router.py new file mode 100644 index 00000000..fb80ae99 --- /dev/null +++ b/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