extract and improve prompt formatting methods

Author: ybdarrenwang

src/strands_evals/evaluators/conciseness_evaluator.py

@@ -7,7 +7,7 @@
 from typing_extensions import Union
 
 from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT
-from ..types.trace import EvaluationLevel, TraceLevelInput
+from ..types.trace import EvaluationLevel
 from .evaluator import Evaluator
 from .prompt_templates.conciseness import get_template
 
@@ -53,14 +53,14 @@ def __init__(
 
     def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:
         parsed_input = self._get_last_turn(evaluation_case)
-        prompt = self._format_prompt(parsed_input)
+        prompt = self._format_trace_level_prompt(parsed_input)
         evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
         result = evaluator_agent(prompt, structured_output_model=ConcisenessRating)
         return self._create_evaluation_output(result)
 
     async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:
         parsed_input = self._get_last_turn(evaluation_case)
-        prompt = self._format_prompt(parsed_input)
+        prompt = self._format_trace_level_prompt(parsed_input)
         evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
         result = await evaluator_agent.invoke_async(prompt, structured_output_model=ConcisenessRating)
         return self._create_evaluation_output(result)
@@ -76,25 +76,3 @@ def _create_evaluation_output(self, result) -> list[EvaluationOutput]:
                 label=rating.score,
             )
         ]
-
-    def _format_prompt(self, parsed_input: TraceLevelInput) -> str:
-        """Format evaluation prompt from parsed turn data."""
-        parts = []
-
-        if parsed_input.session_history:
-            history_lines = []
-            for msg in parsed_input.session_history:
-                if isinstance(msg, list):
-                    # Handle tool execution lists
-                    for tool_exec in msg:
-                        history_lines.append(f"Action: {tool_exec.tool_call.name}({tool_exec.tool_call.arguments})")
-                        history_lines.append(f"Tool: {tool_exec.tool_result.content}")
-                else:
-                    text = msg.content[0].text if msg.content and hasattr(msg.content[0], "text") else ""
-                    history_lines.append(f"{msg.role.value.capitalize()}: {text}")
-            history_str = "\n".join(history_lines)
-            parts.append(f"# Conversation History:\n{history_str}")
-
-        parts.append(f"# Assistant's Response:\n{parsed_input.agent_response.text}")
-
-        return "\n\n".join(parts)

src/strands_evals/evaluators/evaluator.py

@@ -13,6 +13,7 @@
     Session,
     TextContent,
     ToolConfig,
+    ToolLevelInput,
     TraceLevelInput,
     UserMessage,
 )
@@ -143,6 +144,58 @@ def _format_session_history(self, contexts: list[Context]) -> str:
             lines.append(f"Assistant: {ctx.agent_response.text}")
         return "\n".join(lines)
 
+    def _format_tool_level_prompt(self, tool_input: ToolLevelInput) -> str:
+        """Format evaluation prompt from tool-level input."""
+        parts = []
+
+        # Format available tools
+        if tool_input.available_tools:
+            parts.append(f"## Available tool-calls\n{self._format_tools(tool_input.available_tools)}")
+
+        # Format previous conversation history
+        if tool_input.session_history:
+            history_lines = []
+            for msg in tool_input.session_history:
+                if isinstance(msg, list):
+                    # Handle tool execution lists
+                    for tool_exec in msg:
+                        history_lines.append(f"Tool call: {tool_exec.tool_call.name}({tool_exec.tool_call.arguments})")
+                        history_lines.append(f"Tool result: {tool_exec.tool_result.content}")
+                else:
+                    text = msg.content[0].text if msg.content and hasattr(msg.content[0], "text") else ""
+                    history_lines.append(f"{msg.role.value.capitalize()}: {text}")
+            history_str = "\n".join(history_lines)
+            parts.append(f"## Previous conversation history\n{history_str}")
+
+        # Format target tool call to evaluate
+        tool_details = tool_input.tool_execution_details
+        tool_call_str = f"Tool call: {tool_details.tool_call.name}({tool_details.tool_call.arguments})"
+        parts.append(f"## Target tool-call to evaluate\n{tool_call_str}")
+
+        return "\n\n".join(parts)
+
+    def _format_trace_level_prompt(self, parsed_input: TraceLevelInput) -> str:
+        """Format evaluation prompt from parsed turn data."""
+        parts = []
+
+        if parsed_input.session_history:
+            history_lines = []
+            for msg in parsed_input.session_history:
+                if isinstance(msg, list):
+                    # Handle tool execution lists
+                    for tool_exec in msg:
+                        history_lines.append(f"Tool call: {tool_exec.tool_call.name}({tool_exec.tool_call.arguments})")
+                        history_lines.append(f"Tool result: {tool_exec.tool_result.content}")
+                else:
+                    text = msg.content[0].text if msg.content and hasattr(msg.content[0], "text") else ""
+                    history_lines.append(f"{msg.role.value.capitalize()}: {text}")
+            history_str = "\n".join(history_lines)
+            parts.append(f"# Conversation History:\n{history_str}")
+
+        parts.append(f"# Assistant's Response:\n{parsed_input.agent_response.text}")
+
+        return "\n\n".join(parts)
+
     def _has_text_content(self, msg: UserMessage | AssistantMessage) -> TypeGuard[UserMessage | AssistantMessage]:
         """Check if a message object has accessible text content.
 

src/strands_evals/evaluators/faithfulness_evaluator.py

@@ -7,7 +7,7 @@
 from typing_extensions import Union
 
 from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT
-from ..types.trace import EvaluationLevel, TraceLevelInput
+from ..types.trace import EvaluationLevel
 from .evaluator import Evaluator
 from .prompt_templates.faithfulness import get_template
 
@@ -55,7 +55,7 @@ def __init__(
 
     def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:
         parsed_input = self._get_last_turn(evaluation_case)
-        prompt = self._format_prompt(parsed_input)
+        prompt = self._format_trace_level_prompt(parsed_input)
         evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
         result = evaluator_agent(prompt, structured_output_model=FaithfulnessRating)
         rating = cast(FaithfulnessRating, result.structured_output)
@@ -71,7 +71,7 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva
 
     async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:
         parsed_input = self._get_last_turn(evaluation_case)
-        prompt = self._format_prompt(parsed_input)
+        prompt = self._format_trace_level_prompt(parsed_input)
         evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
         result = await evaluator_agent.invoke_async(prompt, structured_output_model=FaithfulnessRating)
         rating = cast(FaithfulnessRating, result.structured_output)
@@ -84,25 +84,3 @@ async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT])
                 label=rating.score,
             )
         ]
-
-    def _format_prompt(self, parsed_input: TraceLevelInput) -> str:
-        """Format evaluation prompt from parsed turn data."""
-        parts = []
-
-        if parsed_input.session_history:
-            history_lines = []
-            for msg in parsed_input.session_history:
-                if isinstance(msg, list):
-                    # Handle tool execution lists
-                    for tool_exec in msg:
-                        history_lines.append(f"Action: {tool_exec.tool_call.name}({tool_exec.tool_call.arguments})")
-                        history_lines.append(f"Tool: {tool_exec.tool_result.content}")
-                else:
-                    text = msg.content[0].text if msg.content and hasattr(msg.content[0], "text") else ""
-                    history_lines.append(f"{msg.role.value.capitalize()}: {text}")
-            history_str = "\n".join(history_lines)
-            parts.append(f"# Conversation History:\n{history_str}")
-
-        parts.append(f"# Assistant's Response:\n{parsed_input.agent_response.text}")
-
-        return "\n\n".join(parts)

src/strands_evals/evaluators/harmfulness_evaluator.py

@@ -5,7 +5,7 @@
 from strands import Agent
 
 from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT
-from ..types.trace import EvaluationLevel, TraceLevelInput
+from ..types.trace import EvaluationLevel
 from .evaluator import Evaluator
 from .prompt_templates.harmfulness import get_template
 
@@ -47,7 +47,7 @@ def __init__(
 
     def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:
         parsed_input = self._get_last_turn(evaluation_case)
-        prompt = self._format_prompt(parsed_input)
+        prompt = self._format_trace_level_prompt(parsed_input)
         evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
         result = evaluator_agent(prompt, structured_output_model=HarmfulnessRating)
         rating = cast(HarmfulnessRating, result.structured_output)
@@ -63,7 +63,7 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva
 
     async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:
         parsed_input = self._get_last_turn(evaluation_case)
-        prompt = self._format_prompt(parsed_input)
+        prompt = self._format_trace_level_prompt(parsed_input)
         evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
         result = await evaluator_agent.invoke_async(prompt, structured_output_model=HarmfulnessRating)
         rating = cast(HarmfulnessRating, result.structured_output)
@@ -76,25 +76,3 @@ async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT])
                 label=rating.score,
             )
         ]
-
-    def _format_prompt(self, parsed_input: TraceLevelInput) -> str:
-        """Format evaluation prompt from parsed turn data."""
-        parts = []
-
-        if parsed_input.session_history:
-            history_lines = []
-            for msg in parsed_input.session_history:
-                if isinstance(msg, list):
-                    # Handle tool execution lists
-                    for tool_exec in msg:
-                        history_lines.append(f"Action: {tool_exec.tool_call.name}({tool_exec.tool_call.arguments})")
-                        history_lines.append(f"Tool: {tool_exec.tool_result.content}")
-                else:
-                    text = msg.content[0].text if msg.content and hasattr(msg.content[0], "text") else ""
-                    history_lines.append(f"{msg.role.value.capitalize()}: {text}")
-            history_str = "\n".join(history_lines)
-            parts.append(f"# Conversation History:\n{history_str}")
-
-        parts.append(f"# Assistant's Response:\n{parsed_input.agent_response.text}")
-
-        return "\n\n".join(parts)

src/strands_evals/evaluators/helpfulness_evaluator.py

@@ -7,7 +7,7 @@
 from typing_extensions import Union
 
 from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT
-from ..types.trace import EvaluationLevel, TraceLevelInput
+from ..types.trace import EvaluationLevel
 from .evaluator import Evaluator
 from .prompt_templates.helpfulness import get_template
 
@@ -61,7 +61,7 @@ def __init__(
 
     def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:
         parsed_input = self._get_last_turn(evaluation_case)
-        prompt = self._format_prompt(parsed_input)
+        prompt = self._format_trace_level_prompt(parsed_input)
         evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
         result = evaluator_agent(prompt, structured_output_model=HelpfulnessRating)
         rating = cast(HelpfulnessRating, result.structured_output)
@@ -77,7 +77,7 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva
 
     async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:
         parsed_input = self._get_last_turn(evaluation_case)
-        prompt = self._format_prompt(parsed_input)
+        prompt = self._format_trace_level_prompt(parsed_input)
         evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
         result = await evaluator_agent.invoke_async(prompt, structured_output_model=HelpfulnessRating)
         rating = cast(HelpfulnessRating, result.structured_output)
@@ -90,25 +90,3 @@ async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT])
                 label=rating.score,
             )
         ]
-
-    def _format_prompt(self, parsed_input: TraceLevelInput) -> str:
-        """Format evaluation prompt from parsed turn data."""
-        parts = []
-
-        if parsed_input.session_history:
-            history_lines = []
-            for msg in parsed_input.session_history:
-                if isinstance(msg, list):
-                    # Handle tool execution lists
-                    for tool_exec in msg:
-                        history_lines.append(f"Action: {tool_exec.tool_call.name}({tool_exec.tool_call.arguments})")
-                        history_lines.append(f"Tool: {tool_exec.tool_result.content}")
-                else:
-                    text = msg.content[0].text if msg.content and hasattr(msg.content[0], "text") else ""
-                    history_lines.append(f"{msg.role.value.capitalize()}: {text}")
-            history_str = "\n".join(history_lines)
-            parts.append(f"# Conversation History:\n{history_str}")
-
-        parts.append(f"# Assistant's Response:\n{parsed_input.agent_response.text}")
-
-        return "\n\n".join(parts)

src/strands_evals/evaluators/response_relevance_evaluator.py

@@ -8,7 +8,7 @@
 from typing_extensions import Union
 
 from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT
-from ..types.trace import EvaluationLevel, TraceLevelInput
+from ..types.trace import EvaluationLevel
 from .evaluator import Evaluator
 from .prompt_templates.response_relevance import get_template
 
@@ -58,14 +58,14 @@ def __init__(
 
     def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:
         parsed_input = self._get_last_turn(evaluation_case)
-        prompt = self._format_prompt(parsed_input)
+        prompt = self._format_trace_level_prompt(parsed_input)
         evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
         result = evaluator_agent(prompt, structured_output_model=ResponseRelevanceRating)
         return self._create_evaluation_output(result)
 
     async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[EvaluationOutput]:
         parsed_input = self._get_last_turn(evaluation_case)
-        prompt = self._format_prompt(parsed_input)
+        prompt = self._format_trace_level_prompt(parsed_input)
         evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
         result = await evaluator_agent.invoke_async(prompt, structured_output_model=ResponseRelevanceRating)
         return self._create_evaluation_output(result)
@@ -81,25 +81,3 @@ def _create_evaluation_output(self, result: AgentResult) -> list[EvaluationOutpu
                 label=rating.score,
             )
         ]
-
-    def _format_prompt(self, parsed_input: TraceLevelInput) -> str:
-        """Format evaluation prompt from parsed turn data."""
-        parts = []
-
-        if parsed_input.session_history:
-            history_lines = []
-            for msg in parsed_input.session_history:
-                if isinstance(msg, list):
-                    # Handle tool execution lists
-                    for tool_exec in msg:
-                        history_lines.append(f"Action: {tool_exec.tool_call.name}({tool_exec.tool_call.arguments})")
-                        history_lines.append(f"Tool: {tool_exec.tool_result.content}")
-                else:
-                    text = msg.content[0].text if msg.content and hasattr(msg.content[0], "text") else ""
-                    history_lines.append(f"{msg.role.value.capitalize()}: {text}")
-            history_str = "\n".join(history_lines)
-            parts.append(f"# Conversation History:\n{history_str}")
-
-        parts.append(f"# Assistant's Response:\n{parsed_input.agent_response.text}")
-
-        return "\n\n".join(parts)

src/strands_evals/evaluators/tool_parameter_accuracy_evaluator.py

@@ -7,7 +7,7 @@
 from typing_extensions import Union
 
 from ..types.evaluation import EvaluationData, EvaluationOutput, InputT, OutputT
-from ..types.trace import EvaluationLevel, ToolLevelInput
+from ..types.trace import EvaluationLevel
 from .evaluator import Evaluator
 from .prompt_templates.tool_parameter_accuracy import get_template
 
@@ -52,7 +52,7 @@ def evaluate(self, evaluation_case: EvaluationData[InputT, OutputT]) -> list[Eva
         results = []
 
         for tool_input in tool_inputs:
-            prompt = self._format_prompt(tool_input)
+            prompt = self._format_tool_level_prompt(tool_input)
             evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
             result = evaluator_agent(prompt, structured_output_model=ToolParameterAccuracyRating)
             rating = cast(ToolParameterAccuracyRating, result.structured_output)
@@ -73,7 +73,7 @@ async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT])
         results = []
 
         for tool_input in tool_inputs:
-            prompt = self._format_prompt(tool_input)
+            prompt = self._format_tool_level_prompt(tool_input)
             evaluator_agent = Agent(model=self.model, system_prompt=self.system_prompt, callback_handler=None)
             result = await evaluator_agent.invoke_async(prompt, structured_output_model=ToolParameterAccuracyRating)
             rating = cast(ToolParameterAccuracyRating, result.structured_output)
@@ -88,33 +88,3 @@ async def evaluate_async(self, evaluation_case: EvaluationData[InputT, OutputT])
             )
 
         return results
-
-    def _format_prompt(self, tool_input: ToolLevelInput) -> str:
-        """Format evaluation prompt from tool-level input."""
-        parts = []
-
-        # Format available tools
-        if tool_input.available_tools:
-            parts.append(f"## Available tool-calls\n{self._format_tools(tool_input.available_tools)}")
-
-        # Format previous conversation history
-        if tool_input.session_history:
-            history_lines = []
-            for msg in tool_input.session_history:
-                if isinstance(msg, list):
-                    # Handle tool execution lists
-                    for tool_exec in msg:
-                        history_lines.append(f"Action: {tool_exec.tool_call.name}({tool_exec.tool_call.arguments})")
-                        history_lines.append(f"Tool: {tool_exec.tool_result.content}")
-                else:
-                    text = msg.content[0].text if msg.content and hasattr(msg.content[0], "text") else ""
-                    history_lines.append(f"{msg.role.value.capitalize()}: {text}")
-            history_str = "\n".join(history_lines)
-            parts.append(f"## Previous conversation history\n{history_str}")
-
-        # Format target tool call to evaluate
-        tool_details = tool_input.tool_execution_details
-        tool_call_str = f"Action: {tool_details.tool_call.name}({tool_details.tool_call.arguments})"
-        parts.append(f"## Target tool-call to evaluate\n{tool_call_str}")
-
-        return "\n\n".join(parts)