Case generation

src/examples/bank_tools_trajectory.py

@@ -109,10 +109,9 @@ async def get_response(query: str) -> dict:
         bank_prompt = "You are a banker, ensure that only people with sufficient balance can spend them. Collect debt from people with negative balance. Be sure to report the current balance after all of the actions."
         agent = Agent(tools = [get_balance, modify_balance, collect_debt], system_prompt=bank_prompt, callback_handler=None)
         response = await agent.invoke_async(query)
-
+        trajectory_evaluator.update_trajectory_description(helper_funcs.extract_tools_description(agent))
         return {"output": str(response),
                 "trajectory": helper_funcs.extract_agent_tools_used_from_messages(agent.messages)}
-
     ### Step 5: Run evaluation ###                                                                                                                                                
     report = await dataset.run_evaluations_async(get_response)
     return report
@@ -122,8 +121,7 @@ async def get_response(query: str) -> dict:
     start = datetime.datetime.now()
     report = asyncio.run(async_descriptive_tools_trajectory_example())
     end = datetime.datetime.now()
-    print("Async: ", end - start) # Async:  0:00:14.723627
-    report.display()
+    print("Async: ", end - start)
     report.to_file("async_bank_tools_trajectory_report", "json")
     report.run_display(include_actual_trajectory=True)
 

src/examples/evaluate_agents_as_tools.py

@@ -4,6 +4,7 @@
 from strands_evaluation.dataset import Dataset 
 from strands_evaluation.case import Case
 from strands_evaluation.evaluators.trajectory_evaluator import TrajectoryEvaluator
+from strands_evaluation.evaluators.interactions_evaluator import InteractionsEvaluator
 import asyncio
 import datetime
 from strands_evaluation.evaluators.utils import helper_funcs
@@ -111,13 +112,129 @@ def trip_planning_assistant(query: str) -> str:
     report = await dataset.run_evaluations_async(trip_planner)
     return report
 
+async def async_agents_as_tools_interaction_example():
+    """Evaluate an orchestrator agent that uses specialized agents as tools using InteractionEvaluator.
+
+    Returns:
+        EvaluationReport: Report containing the interaction evaluation results
+    """
+    ### Step 1: Create test cases ###
+    test1 = Case(input="Plan a trip for Paris next weekend.")
+
+    ### Step 2: Create evaluator ###   
+    evaluator = InteractionsEvaluator(rubric="Are the tools used reasonably and logically?")
+
+    ### Step 3: Create dataset ###
+    dataset = Dataset(cases = [test1], evaluator=evaluator)     
+
+    ### Step 4: Define task ###  
+    def trip_planner(task: str):
+        # Define a specialized system prompt
+        RESEARCH_ASSISTANT_PROMPT = """
+        You are a specialized research assistant. Focus only on providing
+        factual, well-sourced information in response to research questions.
+        Always cite your sources when possible.
+        """
+        @tool
+        def research_assistant(query: str) -> str:
+            """
+            Process and respond to research-related queries.
+
+            Args:
+                query: A research question requiring factual information
+
+            Returns:
+                A detailed research answer with citations
+            """
+            try:
+                # Strands Agents SDK makes it easy to create a specialized agent
+                research_agent = Agent(
+                    system_prompt=RESEARCH_ASSISTANT_PROMPT,
+                    tools=[retrieve, http_request],  # Research-specific tools
+                    callback_handler=None
+                )
+
+                # Call the agent and return its response
+                response = research_agent(query)
+                return str(response)
+            except Exception as e:
+                return f"Error in research assistant: {str(e)}"
+
+        @tool
+        def trip_planning_assistant(query: str) -> str:
+            """
+            Create travel itineraries and provide travel advice.
+
+            Args:
+                query: A travel planning request with destination and preferences
+
+            Returns:
+                A detailed travel itinerary or travel advice
+            """
+            try:
+                travel_agent = Agent(
+                    system_prompt="""You are a specialized travel planning assistant.
+                    Create detailed travel itineraries based on user preferences.""",
+                    tools=[retrieve, http_request],  # Travel information tools
+                    callback_handler=None
+                )
+                response = travel_agent(query)
+                return str(response)
+            except Exception as e:
+                return f"Error in trip planning: {str(e)}"
+
+        # Define the orchestrator system prompt with clear tool selection guidance
+        MAIN_SYSTEM_PROMPT = """
+        You are an assistant that routes queries to specialized agents:
+        - For research questions and factual information → Use the research_assistant tool
+        - For product recommendations and shopping advice → Use the product_recommendation_assistant tool
+        - For travel planning and itineraries → Use the trip_planning_assistant tool
+        - For simple questions not requiring specialized knowledge → Answer directly
+
+        Always select the most appropriate tool based on the user's query.
+        """
+
+        # Strands Agents SDK allows easy integration of agent tools
+        orchestrator = Agent(
+            system_prompt=MAIN_SYSTEM_PROMPT,
+            callback_handler=None,
+            tools=[research_assistant, trip_planning_assistant]
+        )
+        response = orchestrator(task)
+
+        evaluator.update_interaction_description(helper_funcs.extract_tools_description(orchestrator))
+        tools_used = helper_funcs.extract_agent_tools_used_from_messages(orchestrator.messages)
+        interactions = []
+        for tool_used in tools_used:
+            interactions.append({
+                "node_name": tool_used.get("name"),
+                "dependencies": [],
+                "messages": tool_used.get("message")
+            })
+
+        # This helper function does not include message
+        return {
+            "output": str(response),
+            "interactions": interactions
+        }        
+
+    report = await dataset.run_evaluations_async(trip_planner)
+    return report
+
 if __name__ == "__main__":
-    # run the file as a module: eg. python -m examples.evaluate_tools_as_agent
+    # run the file as a module: eg. python -m examples.evaluate_agents_as_tools
+    # start = datetime.datetime.now()
+    # report = asyncio.run(async_evaluate_tools_as_agents_example())
+    # end = datetime.datetime.now()
+    # print("Time: ", end - start)
+    # report.to_file("async_tools_as_agents_report_w_output", "json")
+    # report.run_display(include_actual_trajectory=True)
+
     start = datetime.datetime.now()
-    report = asyncio.run(async_evaluate_tools_as_agents_example())
+    report = asyncio.run(async_agents_as_tools_interaction_example())
     end = datetime.datetime.now()
     print("Time: ", end - start)
-    report.to_file("async_tools_as_agents_report_w_output", "json")
-    report.run_display(include_actual_trajectory=True)
+    report.to_file("async_agent_as_tools_interaction", "json")
+    report.run_display(include_actual_interactions=True)
 
 

src/strands_evaluation/dataset.py

@@ -3,7 +3,7 @@
 from collections.abc import Callable
 
 from .case import Case
-from .types.evaluation import EvaluationData, EvaluationOutput
+from .types.evaluation import EvaluationData
 from .types.evaluation_report import EvaluationReport
 from .evaluators.evaluator import Evaluator
 from .evaluators.trajectory_evaluator import TrajectoryEvaluator

src/strands_evaluation/evaluators/interactions_evaluator.py

@@ -17,20 +17,31 @@ class InteractionsEvaluator(Evaluator[InputT, OutputT]):
         rubric: The user-specified criteria for evaluating a collection of test cases.
             if the rubric is a string, then use the same rubric for all of the evaluations, else
             get the node-specific rubric for evaluation.
+        interaction_description: A dictionary describing the evailable interactions.
         model: A string representing the model-id for Bedrock to use.
                     Defaults to strands.models.BedrockModel if None.
         system_prompt: System prompt to guide model behavior.
                     If None, the evaluator will use one of the default template.
         include_inputs: Whether to include inputs to the task in the evaluation or not.
     """
-    def __init__(self, rubric: str | dict[str, str], model: str | None = None, system_prompt: str = SYSTEM_PROMPT,
+    def __init__(self, rubric: str | dict[str, str], interaction_description: dict | None = None, model: str | None = None, system_prompt: str = SYSTEM_PROMPT,
                 include_inputs: bool = True):
         super().__init__()
         self.rubric = rubric
+        self.interaction_description = interaction_description
         self.model = model
         self.include_inputs = include_inputs
         self.system_prompt = system_prompt
 
+    def update_interaction_description(self, new_description: dict) -> None:
+        """
+        Update the description of the available interactions.
+
+        Args:
+            new_description: The new description of the available interactions.
+        """
+        self.interaction_description = new_description
+
     def _get_node_rubric(self, node_name: str) -> str:
         """
         Get the rubric for the node involved in the interaction.
@@ -100,6 +111,9 @@ def _compose_prompt(self, evaluation_case: EvaluationData[InputT, OutputT], curr
             if evaluation_case.expected_output:
                 evaluation_prompt += f"<ExpectedOutput>{evaluation_case.expected_output}</ExpectedOutput>\n"
 
+        if self.interaction_description:
+            evaluation_prompt += f"<InteractionDescription>{self.interaction_description}</InteractionDescription>\n"
+
         evaluation_prompt += f"<Rubric>{self._get_node_rubric(node_name)}</Rubric>"
 
         return evaluation_prompt

src/strands_evaluation/evaluators/trajectory_evaluator.py

@@ -16,21 +16,32 @@ class TrajectoryEvaluator(Evaluator[InputT, OutputT]):
 
     Attributes:
         rubric: The user-specified criteria for evaluating a collection of test cases.
+        trajectory_description: A description of the available trajectory types. eg. tool descriptions
         model: A string representing the model-id for Bedrock to use.
                     Defaults to strands.models.BedrockModel if None.
         system_prompt: System prompt to guide model behavior.
                     If None, the evaluator will use one of the default template.
         include_inputs: Whether to include inputs to the task in the evaluation or not.
     """
-    def __init__(self, rubric: str, model: str | None = None, system_prompt: str = SYSTEM_PROMPT,
+    def __init__(self, rubric: str, trajectory_description: dict | None = None, model: str | None = None, system_prompt: str = SYSTEM_PROMPT,
                 include_inputs: bool = True):
         super().__init__()
         self.rubric = rubric
+        self.trajectory_description = trajectory_description
         self.model = model
         self.include_inputs = include_inputs
         self._tools = [exact_match_scorer, in_order_match_scorer, any_order_match_scorer]
         self.system_prompt = system_prompt
 
+    def update_trajectory_description(self, new_description: dict) -> None:
+        """
+        Update the description of the available trajectories.
+
+        Args:
+            new_description: The new description of the available trajectories.
+        """
+        self.trajectory_description = new_description
+
     def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> EvaluationOutput:
         """
         Evaluate the performance of the task on the given test cases.

src/strands_evaluation/evaluators/utils/case_prompt_template.py

@@ -4,7 +4,7 @@
 InputT = TypeVar("InputT")
 OutputT = TypeVar("OutputT")
 
-def compose_test_prompt(evaluation_case: EvaluationData[InputT, OutputT], rubric: str, include_inputs: bool, uses_trajectory: bool = False) -> str:
+def compose_test_prompt(evaluation_case: EvaluationData[InputT, OutputT], rubric: str, include_inputs: bool, uses_trajectory: bool = False, trajectory_description: dict = None) -> str:
     """
     Compose the prompt for a test case evaluation.
 
@@ -13,6 +13,7 @@ def compose_test_prompt(evaluation_case: EvaluationData[InputT, OutputT], rubric
         rubric: The evaluation criteria to be applied
         include_inputs: Whether to include the input in the prompt
         uses_trajectory: Whether this is a trajectory-based evaluation
+        trajectory_description: A dictionary describing the type of trajectory expected for this evaluation. 
 
     Returns:
         str: The formatted evaluation prompt
@@ -44,6 +45,9 @@ def compose_test_prompt(evaluation_case: EvaluationData[InputT, OutputT], rubric
         if evaluation_case.expected_trajectory:
             evaluation_prompt += f"<ExpectedTrajectory>{evaluation_case.expected_trajectory}</ExpectedTrajectory>\n"
 
+        if trajectory_description:
+            evaluation_prompt += f"<TrajectoryDescription>{trajectory_description}</TrajectoryDescription>\n"
+
     evaluation_prompt += f"<Rubric>{rubric}</Rubric>"
 
     return evaluation_prompt

src/strands_evaluation/evaluators/utils/helper_funcs.py

@@ -1,9 +1,11 @@
 from strands.multiagent import SwarmResult
 from strands.multiagent import GraphResult
+from strands import Agent
 
 ### Extracting Information from SWARM ###
 def extract_swarm_handoffs(swarm_result: SwarmResult) -> list[dict]:
-    """Extract handoff information from swarm execution results.
+    """
+    Extract handoff information from swarm execution results.
 
     Args:
         swarm_result: Result object from swarm execution
@@ -34,7 +36,8 @@ def extract_swarm_handoffs(swarm_result: SwarmResult) -> list[dict]:
     return hand_off_info
 
 def extract_swarm_interactions_from_handoffs(handoffs_info: list[dict]) -> list[dict]:
-    """Convert handoff information to interaction format for evaluation.
+    """
+    Convert handoff information to interaction format for evaluation.
 
     Args:
         handoffs_info: List of handoff information from extract_swarm_handoffs
@@ -78,7 +81,8 @@ def extract_swarm_interactions(swarm_result: SwarmResult) -> list[dict]:
 
 ### Extracting Information from GRAPH ###
 def extract_graph_interactions(graph_result: GraphResult):
-    """Extract interaction information from graph execution results.
+    """
+    Extract interaction information from graph execution results.
 
     Args:
         graph_result: Result object from graph execution
@@ -101,7 +105,8 @@ def extract_graph_interactions(graph_result: GraphResult):
 
 ### Extract Information from Agent result ###
 def extract_agent_tools_used_from_messages(agent_messages):
-    """Extract tool usage information from agent message history.
+    """
+    Extract tool usage information from agent message history.
 
     Args:
         agent_messages: List of message dictionaries from agent conversation
@@ -124,7 +129,8 @@ def extract_agent_tools_used_from_messages(agent_messages):
     return tools_used
 
 def extract_agent_tools_used_from_metrics(agent_result):
-    """Extract tool usage metrics from agent execution result.
+    """
+    Extract tool usage metrics from agent execution result.
 
     Args:
         agent_result: Agent result object containing metrics
@@ -149,4 +155,26 @@ def extract_agent_tools_used_from_metrics(agent_result):
             "success_count": tool_info.success_count,
             "total_time": tool_info.total_time,
         })
-    return tools_used
+    return tools_used
+
+### Extract Information from Agent ###
+def extract_tools_description(agent: Agent, is_short: bool = True):
+    """
+    Extract a dictionary of all tools used in a given agent.
+
+    Args:
+        agent (Agent): Target agent to extract tool registry from
+        is_short (bool, optional): Whether to return only the description of the tools or everything. Defaults to True.
+
+    Returns:
+        dict: Tool name and its corresponding description
+        {<tool_name>: <tool_description>, ...}
+    """
+    description = agent.tool_registry.get_all_tools_config()
+    if is_short:
+        shorten_descrip = {}
+        for tool_name, tool_info in description.items():
+            shorten_descrip[tool_name] = tool_info["description"]
+        return shorten_descrip
+
+    return description

src/strands_evaluation/evaluators/utils/prompt_templates.py

@@ -85,7 +85,7 @@
 - <ExpectedOutput>: Optional reference for what the output should be
 - <Trajectory>: Sequence of steps or tools that were actually executed
 - <ExpectedTrajectory>: Optional reference for what the trajectory should be
-- <TrajectoryTypes>: Optional description of trajectory type when evaluating trajectories
+- <TrajectoryDescription>: Optional description of available trajectory type when evaluating trajectories
 - <Rubric>: Evaluation criteria for scoring
 
 IMPORTANT: The <Trajectory> represents the actual sequence of tools/actions that were executed to generate the output.
@@ -131,6 +131,7 @@
 - <Input>: Optional original input that initiated the interaction sequence
 - <Output>: Optional final output (only provided for the last interaction)
 - <ExpectedOutput>: Optional reference for what the final output should be
+- <InteractionDescription>: Optional description of the type of interactions being evaluated (e.g., multi-agent, sequential, parallel)
 - <Rubric>: Evaluation criteria specific to the current node/interaction
 
 Your task is to evaluate each interaction step-by-step, building context as you progress through the sequence and keeping track of problematic interactions. For intermediate interactions, focus on: