feat: add ToolEfficiencyEvaluator for session-level tool usage analysis

src/strands_evals/evaluators/__init__.py

@@ -18,6 +18,7 @@
 from .refusal_evaluator import RefusalEvaluator
 from .response_relevance_evaluator import ResponseRelevanceEvaluator
 from .stereotyping_evaluator import StereotypingEvaluator
+from .tool_efficiency_evaluator import ToolEfficiencyEvaluator
 from .tool_parameter_accuracy_evaluator import ToolParameterAccuracyEvaluator
 from .tool_selection_accuracy_evaluator import ToolSelectionAccuracyEvaluator
 from .trajectory_evaluator import TrajectoryEvaluator
@@ -40,6 +41,7 @@
     "ResponseRelevanceEvaluator",
     "ToolSelectionAccuracyEvaluator",
     "ToolParameterAccuracyEvaluator",
+    "ToolEfficiencyEvaluator",
     "ConcisenessEvaluator",
     "CoherenceEvaluator",
     "RefusalEvaluator",

src/strands_evals/evaluators/prompt_templates/tool_efficiency/__init__.py

@@ -0,0 +1,11 @@
+from . import tool_efficiency_v0
+
+VERSIONS = {
+    "v0": tool_efficiency_v0,
+}
+
+DEFAULT_VERSION = "v0"
+
+
+def get_template(version: str = DEFAULT_VERSION):
+    return VERSIONS[version]

src/strands_evals/evaluators/prompt_templates/tool_efficiency/tool_efficiency_v0.py

@@ -0,0 +1,27 @@
+SYSTEM_PROMPT = """You are an objective judge evaluating the efficiency of tool usage in an AI assistant's conversation. You will analyze the complete conversation trajectory and classify each tool call into one of four categories.
+
+## Classification Categories
+
+- necessary: The tool call's result directly contributed to the final response given to the user.
+- redundant: The same tool was called with the same or equivalent parameters earlier in the conversation, making this call unnecessary.
+- errored: The tool call failed due to malformed input or incorrect parameters, requiring a subsequent retry.
+- unnecessary: The tool call succeeded but its result was never referenced or used in the final response to the user.
+
+## Evaluation Process
+
+1. Read all tool calls in chronological order.
+2. Read the final assistant response that concludes the conversation.
+3. For each tool call, check if a prior call already produced the same information (redundant).
+4. For each tool call that resulted in an error, check if it was due to bad input that was later corrected (errored).
+5. For each successful tool call, check if the result appears in or contributed to the final response (unnecessary if not).
+6. Everything remaining is necessary.
+
+## Guidelines
+
+- A tool call is necessary if removing it would change or degrade the final response.
+- A tool call is redundant only if a previous call already produced equivalent information that was available in context.
+- A tool call is errored only if it failed and a subsequent call with corrected parameters succeeded.
+- A tool call is unnecessary only if it succeeded but its output had no bearing on the final response.
+- When in doubt between necessary and unnecessary, consider whether the information retrieved could have influenced the assistant's reasoning even if not quoted verbatim.
+
+Classify every tool call and provide an overall efficiency assessment."""

src/strands_evals/evaluators/tool_efficiency_evaluator.py

@@ -0,0 +1,116 @@
+from enum import Enum
+from typing import cast
+
+from pydantic import BaseModel, Field
+from strands import Agent
+from strands.models.model import Model
+
+from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT
+from ..types.trace import EvaluationLevel, SessionLevelInput
+from .evaluator import Evaluator
+from .prompt_templates.tool_efficiency import get_template
+
+
+class ToolCallCategory(str, Enum):
+    """Classification categories for individual tool calls."""
+
+    NECESSARY = "necessary"
+    REDUNDANT = "redundant"
+    ERRORED = "errored"
+    UNNECESSARY = "unnecessary"
+
+
+class ToolCallClassification(BaseModel):
+    """Classification result for a single tool call."""
+
+    tool_name: str
+    call_index: int = Field(description="0-based position in the trajectory")
+    category: ToolCallCategory
+    reasoning: str = Field(description="One sentence explaining the classification")
+
+
+class ToolEfficiencyRating(BaseModel):
+    """Structured output for the tool efficiency evaluation."""
+
+    classifications: list[ToolCallClassification]
+    reasoning: str = Field(description="Overall assessment of tool usage efficiency")
+
+
+class ToolEfficiencyEvaluator(Evaluator[InputT, OutputT]):
+    """Evaluates whether all tool calls in a trajectory were necessary.
+
+    Operates at SESSION_LEVEL. Reads the full trajectory and asks an LLM judge
+    to classify each tool call as NECESSARY, REDUNDANT, ERRORED, or UNNECESSARY.
+
+    Score is calculated as necessary_count / total_count (1.0 if no tool calls).
+    The per-call breakdown is stored in EvaluationOutput.label as a JSON string.
+    """
+
+    evaluation_level = EvaluationLevel.SESSION_LEVEL
+
+    def __init__(
+        self,
+        version: str = "v0",
+        model: Model | str | None = None,
+        system_prompt: str | None = None,
+        max_tool_result_length: int = 2000,
+        pass_threshold: float = 0.5,
+        name: str | None = None,
+    ):
+        super().__init__(name=name)
+        self.system_prompt = system_prompt if system_prompt is not None else get_template(version).SYSTEM_PROMPT
+        self.version = version
+        self.model = model
+        self.max_tool_result_length = max_tool_result_length
+        self.pass_threshold = pass_threshold
+
+    def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:
+        session_input: SessionLevelInput = self._parse_trajectory(evaluation_case)
+        prompt = self._format_prompt(session_input)
+
+        evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
+        result = evaluator_agent(prompt, structured_output_model=ToolEfficiencyRating)
+        rating = cast(ToolEfficiencyRating, result.structured_output)
+
+        total_count = len(rating.classifications)
+        necessary_count = sum(1 for c in rating.classifications if c.category == ToolCallCategory.NECESSARY)
+        score = necessary_count / total_count if total_count > 0 else 1.0
+
+        return [
+            EvaluationOutput(
+                score=score,
+                test_pass=score >= self.pass_threshold,
+                reason=rating.reasoning,
+                label=rating.model_dump_json(),
+            )
+        ]
+
+    def _format_prompt(self, session_input: SessionLevelInput) -> str:
+        """Format evaluation prompt from session-level input."""
+        parts = []
+
+        if session_input.available_tools:
+            parts.append(f"# Available tools\n{self._format_tools(session_input.available_tools)}")
+
+        if session_input.session_history:
+            parts.append(f"# Conversation record\n{self._format_session_history_with_truncation(session_input)}")
+
+        return "\n\n".join(parts)
+
+    def _format_session_history_with_truncation(self, session_input: SessionLevelInput) -> str:
+        """Format session history, truncating long tool results."""
+        lines = []
+        for ctx in session_input.session_history:
+            lines.append(f"User: {ctx.user_prompt.text}")
+            if ctx.tool_execution_history:
+                for tool_exec in ctx.tool_execution_history:
+                    lines.append(f"Action: {tool_exec.tool_call.name}({tool_exec.tool_call.arguments})")
+                    result_content = tool_exec.tool_result.content
+                    if len(result_content) > self.max_tool_result_length:
+                        result_content = result_content[: self.max_tool_result_length] + "... [truncated]"
+                    if tool_exec.tool_result.error:
+                        lines.append(f"Tool Error: {tool_exec.tool_result.error}")
+                    else:
+                        lines.append(f"Tool: {result_content}")
+            lines.append(f"Assistant: {ctx.agent_response.text}")
+        return "\n".join(lines)