diff --git a/src/strands_evals/evaluators/__init__.py b/src/strands_evals/evaluators/__init__.py index c163ec8e..890051e3 100644 --- a/src/strands_evals/evaluators/__init__.py +++ b/src/strands_evals/evaluators/__init__.py @@ -1,7 +1,7 @@ from .coherence_evaluator import CoherenceEvaluator from .conciseness_evaluator import ConcisenessEvaluator from .correctness_evaluator import CorrectnessEvaluator -from .deterministic import Contains, Equals, StartsWith, StateEquals, ToolCalled +from .deterministic import Contains, Equals, InclusiveLanguage, StartsWith, StateEquals, ToolCalled from .evaluator import Evaluator from .faithfulness_evaluator import FaithfulnessEvaluator from .goal_success_rate_evaluator import GoalSuccessRateEvaluator @@ -47,6 +47,7 @@ "InstructionFollowingEvaluator", "Contains", "Equals", + "InclusiveLanguage", "StartsWith", "StateEquals", "ToolCalled", diff --git a/src/strands_evals/evaluators/deterministic/__init__.py b/src/strands_evals/evaluators/deterministic/__init__.py index 66cba320..950ef1cb 100644 --- a/src/strands_evals/evaluators/deterministic/__init__.py +++ b/src/strands_evals/evaluators/deterministic/__init__.py @@ -1,10 +1,12 @@ from .environment_state import StateEquals +from .inclusive_language import InclusiveLanguage from .output import Contains, Equals, StartsWith from .trajectory import ToolCalled __all__ = [ "Contains", "Equals", + "InclusiveLanguage", "StartsWith", "StateEquals", "ToolCalled", diff --git a/src/strands_evals/evaluators/deterministic/inclusive_language.py b/src/strands_evals/evaluators/deterministic/inclusive_language.py new file mode 100644 index 00000000..bd7cd48d --- /dev/null +++ b/src/strands_evals/evaluators/deterministic/inclusive_language.py @@ -0,0 +1,77 @@ +import re +from typing import ClassVar + +from ...types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT +from ..evaluator import Evaluator + + +class InclusiveLanguage(Evaluator[InputT, OutputT]): + """Scans actual_output for non-inclusive terms against a configurable term list. + + Performs a case-insensitive word-boundary regex scan of actual_output against a + mapping of banned terms to suggested replacements. Returns a passing score when + no banned terms are found, and a failing score with details when any are detected. + + Note on false positives: The default term list includes bare tokens like 'master' + and 'slave' which can produce false positives in legitimate contexts (e.g., + "master's degree", "master volume"). Word-boundary matching reduces this risk + (e.g., "masterful" will not match), but standalone uses like "master branch" will + still trigger. If your domain frequently uses these words in non-exclusionary + contexts, consider providing a custom terms dict that omits them or uses more + specific compound patterns like 'master/slave' instead. + """ + + DEFAULT_TERMS: ClassVar[dict[str, str]] = { + "blacklist": "denylist", + "whitelist": "allowlist", + "master": "primary", + "slave": "replica", + "blackday": "blocked day", + "whiteday": "clear day", + } + + def __init__(self, terms: dict[str, str] | None = None, name: str | None = None): + """Initialize the inclusive language evaluator. + + Args: + terms: Mapping of banned terms to suggested replacements. + If None, uses a copy of the built-in DEFAULT_TERMS. + name: Optional instance name for identification in reports. + """ + super().__init__(name=name) + self.terms = terms if terms is not None else dict(self.DEFAULT_TERMS) + self._compiled_patterns: list[tuple[re.Pattern[str], str, str]] = [ + (re.compile(rf"\b{re.escape(term)}\b"), term, suggestion) for term, suggestion in self.terms.items() + ] + + def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: + """Evaluate actual_output for non-inclusive terminology. + + Args: + evaluation_case: The evaluation data containing the output to scan. + + Returns: + A list with a single EvaluationOutput. Score is 1.0 (pass) if no + banned terms are found, 0.0 (fail) otherwise. + """ + text = str(evaluation_case.actual_output).lower() + found: list[tuple[str, str]] = [] + for pattern, term, suggestion in self._compiled_patterns: + if pattern.search(text): + found.append((term, suggestion)) + + if not found: + return [EvaluationOutput(score=1.0, test_pass=True, reason="no non-inclusive terms found")] + + details = ", ".join(f"'{t}' -> '{s}'" for t, s in found) + return [ + EvaluationOutput( + score=0.0, + test_pass=False, + reason=f"found {len(found)} non-inclusive term(s): {details}", + ) + ] + + async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]: + """Async version of evaluate. Delegates to the synchronous implementation.""" + return self.evaluate(evaluation_case) diff --git a/tests/strands_evals/evaluators/deterministic/test_inclusive_language.py b/tests/strands_evals/evaluators/deterministic/test_inclusive_language.py new file mode 100644 index 00000000..d426ff3b --- /dev/null +++ b/tests/strands_evals/evaluators/deterministic/test_inclusive_language.py @@ -0,0 +1,240 @@ +import pytest + +from strands_evals.evaluators.deterministic.inclusive_language import InclusiveLanguage +from strands_evals.types import EvaluationData +from strands_evals.types.evaluation import EvaluationOutput + + +class TestInclusiveLanguageDefaults: + def test_passes_when_no_banned_terms(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="Use the denylist to block bad actors") + results = evaluator.evaluate(data) + assert results == [EvaluationOutput(score=1.0, test_pass=True, reason="no non-inclusive terms found")] + + def test_fails_when_blacklist_found(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="Add the IP to the blacklist") + results = evaluator.evaluate(data) + assert results == [ + EvaluationOutput( + score=0.0, + test_pass=False, + reason="found 1 non-inclusive term(s): 'blacklist' -> 'denylist'", + ) + ] + + def test_fails_when_whitelist_found(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="Add it to the whitelist") + results = evaluator.evaluate(data) + assert results == [ + EvaluationOutput( + score=0.0, + test_pass=False, + reason="found 1 non-inclusive term(s): 'whitelist' -> 'allowlist'", + ) + ] + + def test_fails_when_master_found(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="Push to the master branch") + results = evaluator.evaluate(data) + assert results == [ + EvaluationOutput( + score=0.0, + test_pass=False, + reason="found 1 non-inclusive term(s): 'master' -> 'primary'", + ) + ] + + def test_fails_when_slave_found(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="Configure the slave node") + results = evaluator.evaluate(data) + assert results == [ + EvaluationOutput( + score=0.0, + test_pass=False, + reason="found 1 non-inclusive term(s): 'slave' -> 'replica'", + ) + ] + + def test_reports_multiple_terms(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="The master sends to the slave") + results = evaluator.evaluate(data) + assert results == [ + EvaluationOutput( + score=0.0, + test_pass=False, + reason="found 2 non-inclusive term(s): 'master' -> 'primary', 'slave' -> 'replica'", + ) + ] + + +class TestInclusiveLanguageCaseSensitivity: + def test_case_insensitive_uppercase(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="Add to BLACKLIST") + results = evaluator.evaluate(data) + assert results[0].test_pass is False + + def test_case_insensitive_mixed_case(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="The Whitelist is updated") + results = evaluator.evaluate(data) + assert results[0].test_pass is False + + +class TestInclusiveLanguageWordBoundary: + def test_no_false_positive_on_masterful(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="That was a masterful performance") + results = evaluator.evaluate(data) + assert results == [EvaluationOutput(score=1.0, test_pass=True, reason="no non-inclusive terms found")] + + def test_no_false_positive_on_mastering(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="She is mastering the skill") + results = evaluator.evaluate(data) + assert results == [EvaluationOutput(score=1.0, test_pass=True, reason="no non-inclusive terms found")] + + def test_no_false_positive_on_slavery(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="The history of slavery is complex") + results = evaluator.evaluate(data) + assert results == [EvaluationOutput(score=1.0, test_pass=True, reason="no non-inclusive terms found")] + + def test_matches_term_at_start_of_string(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="master is the default branch") + results = evaluator.evaluate(data) + assert results[0].test_pass is False + + def test_matches_term_at_end_of_string(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="push to master") + results = evaluator.evaluate(data) + assert results[0].test_pass is False + + def test_matches_term_with_punctuation_boundary(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="check the blacklist, then proceed") + results = evaluator.evaluate(data) + assert results[0].test_pass is False + + +class TestInclusiveLanguageCustomTerms: + def test_custom_terms_override_defaults(self): + custom = {"legacy": "historical"} + evaluator = InclusiveLanguage(terms=custom) + # Default term should not trigger + data = EvaluationData(input="q", actual_output="Push to the master branch") + results = evaluator.evaluate(data) + assert results == [EvaluationOutput(score=1.0, test_pass=True, reason="no non-inclusive terms found")] + + def test_custom_terms_detected(self): + custom = {"legacy": "historical", "deprecated": "removed"} + evaluator = InclusiveLanguage(terms=custom) + data = EvaluationData(input="q", actual_output="This is a legacy system") + results = evaluator.evaluate(data) + assert results == [ + EvaluationOutput( + score=0.0, + test_pass=False, + reason="found 1 non-inclusive term(s): 'legacy' -> 'historical'", + ) + ] + + def test_empty_terms_always_passes(self): + evaluator = InclusiveLanguage(terms={}) + data = EvaluationData(input="q", actual_output="blacklist whitelist master slave") + results = evaluator.evaluate(data) + assert results == [EvaluationOutput(score=1.0, test_pass=True, reason="no non-inclusive terms found")] + + +class TestInclusiveLanguageSharedState: + def test_default_terms_not_shared_between_instances(self): + evaluator1 = InclusiveLanguage() + evaluator2 = InclusiveLanguage() + evaluator1.terms["newterm"] = "replacement" + assert "newterm" not in evaluator2.terms + assert "newterm" not in InclusiveLanguage.DEFAULT_TERMS + + +class TestInclusiveLanguageEdgeCases: + def test_none_actual_output_passes(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output=None) + results = evaluator.evaluate(data) + assert results == [EvaluationOutput(score=1.0, test_pass=True, reason="no non-inclusive terms found")] + + def test_empty_string_passes(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="") + results = evaluator.evaluate(data) + assert results == [EvaluationOutput(score=1.0, test_pass=True, reason="no non-inclusive terms found")] + + def test_numeric_output_coerced(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output=42) + results = evaluator.evaluate(data) + assert results == [EvaluationOutput(score=1.0, test_pass=True, reason="no non-inclusive terms found")] + + def test_reason_on_pass(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="clean text") + results = evaluator.evaluate(data) + assert results[0].reason == "no non-inclusive terms found" + + def test_reason_on_fail_format(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="Add to blacklist") + results = evaluator.evaluate(data) + assert results[0].reason == "found 1 non-inclusive term(s): 'blacklist' -> 'denylist'" + + +class TestInclusiveLanguageAsync: + @pytest.mark.asyncio + async def test_evaluate_async_delegates_to_evaluate(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="Use the denylist") + results = await evaluator.evaluate_async(data) + assert results == [EvaluationOutput(score=1.0, test_pass=True, reason="no non-inclusive terms found")] + + @pytest.mark.asyncio + async def test_evaluate_async_detects_terms(self): + evaluator = InclusiveLanguage() + data = EvaluationData(input="q", actual_output="Add to blacklist") + results = await evaluator.evaluate_async(data) + assert results == [ + EvaluationOutput( + score=0.0, + test_pass=False, + reason="found 1 non-inclusive term(s): 'blacklist' -> 'denylist'", + ) + ] + + +class TestInclusiveLanguageSerialization: + def test_to_dict_default_terms(self): + evaluator = InclusiveLanguage() + d = evaluator.to_dict() + assert d["evaluator_type"] == "InclusiveLanguage" + assert d["terms"] == InclusiveLanguage.DEFAULT_TERMS + + def test_to_dict_custom_terms(self): + custom = {"foo": "bar"} + evaluator = InclusiveLanguage(terms=custom) + d = evaluator.to_dict() + assert d["evaluator_type"] == "InclusiveLanguage" + assert d["terms"] == {"foo": "bar"} + + def test_name_parameter(self): + evaluator = InclusiveLanguage(name="my_scanner") + assert evaluator.get_name() == "my_scanner" + + def test_default_name(self): + evaluator = InclusiveLanguage() + assert evaluator.get_name() == "InclusiveLanguage"