evals/key_points_evaluator.py

"""
Custom evaluator for checking if specific key points are present in the response.
"""
from typing import Any, List
from strands_evals.evaluators import Evaluator
from strands_evals.types.evaluation import EvaluationData, EvaluationOutput
from strands_evals.types.trace import EvaluationLevel
from typing_extensions import TypeVar

InputT = TypeVar("InputT")
OutputT = TypeVar("OutputT")

class KeyPointsEvaluator(Evaluator[InputT, OutputT]):
    """Evaluates output by checking for presence of expected key points (keywords/phrases)."""
    
    evaluation_level = EvaluationLevel.TRACE_LEVEL

    def __init__(self, version: str = "v1"):
        super().__init__()
        self.version = version

    def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> List[EvaluationOutput]:
        """Synchronous evaluation."""
        return self._do_evaluation(evaluation_case)

    async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> List[EvaluationOutput]:
        """Asynchronous evaluation."""
        return self._do_evaluation(evaluation_case)

    def _do_evaluation(self, evaluation_case: EvaluationData[InputT, OutputT]) -> List[EvaluationOutput]:
        """
        Check if expected key points are present in the actual output.
        Expects 'expected_key_points' list in case metadata.
        """
        # Get actual output
        actual_output = str(evaluation_case.actual_output)
        
        # Get expectations from case metadata (which is attached to evaluation_case)
        # Note: The SDK passes the whole Case object or relevant parts.
        # However, EvaluationData typically has input/output. 
        # Metadata is likely accessible if evaluation_case is constructed from a Case.
        # But SDK EvaluationData doesn't strictly carry metadata field in all versions.
        # We rely on how Experiment constructs it.
        
        # EXPERIMENTAL: The SDK's Experiment loop constructs EvaluationData.
        # If it doesn't pass metadata, we need to inspect the source 'case'. 
        # But Evaluator.evaluate receives EvaluationData, not Case.
        # Wait, Strands SDK 1.22 might have metadata on EvaluationData?
        # Let's check the type definition if needed. 
        # For now, assuming we can access it or we need a workaround.
        
        # Workaround: For this custom evaluator to work with Experiment, 
        # the Experiment must pass metadata. 
        
        # Actually, looking at the Experiment source (which we can't see right now but inferred),
        # it might be easier to pass expected_output as the key points string?
        # Dataset loader sets: expected_key_points in metadata.
        
        # Let's try to access metadata if it exists on EvaluationData, 
        # Otherwise fall back to a safe default.
        
        key_points = []
        if hasattr(evaluation_case, 'metadata') and evaluation_case.metadata:
            key_points = evaluation_case.metadata.get("expected_key_points", [])
        
        # Calculate score
        if not key_points:
            return [EvaluationOutput(
                score=1.0, 
                test_pass=True, 
                reason="No key points defined for this case.",
                label="N/A"
            )]

        hits = 0
        misses = []
        
        for point in key_points:
            point_lower = point.lower()
            output_lower = actual_output.lower()
            
            if point_lower in output_lower:
                hits += 1
            # partial match check (heuristic from run_full_suite)
            elif any(word in output_lower for word in point_lower.split() if len(word) > 4):
                 hits += 0.5
                 misses.append(f"{point} (Partial)")
            else:
                 misses.append(point)

        score = min(1.0, hits / len(key_points))
        
        reason = f"Matched {hits}/{len(key_points)} key points."
        if misses:
            reason += f" Missed: {', '.join(misses[:3])}..."
            
        return [EvaluationOutput(
            score=score,
            test_pass=score >= 0.7, # 70% threshold
            reason=reason,
            label=f"{int(score*100)}%"
        )]