feat: add new speaker handoff pattern that is more scalable and generalizable

Author: philipph-askui

src/askui/agent_base.py

@@ -23,12 +23,10 @@
     LocateSettings,
 )
 from askui.models.shared.tools import Tool, ToolCollection
-from askui.prompts.act_prompts import create_default_prompt
-from askui.prompts.caching import CACHE_USE_PROMPT
+from askui.prompts.act_prompts import CACHE_USE_PROMPT, create_default_prompt
 from askui.tools.agent_os import AgentOs
 from askui.tools.android.agent_os import AndroidAgentOs
 from askui.tools.caching_tools import (
-    ExecuteCachedTrajectory,
     InspectCacheMetadata,
     RetrieveCachedTestExecutions,
     VerifyCacheExecution,
@@ -231,7 +229,7 @@ def act(
 
         _caching_settings: CachingSettings = caching_settings or self.caching_settings
 
-        tools, cached_execution_tool, cache_manager = self._patch_act_with_cache(
+        tools, cache_manager = self._patch_act_with_cache(
             _caching_settings, _act_settings, tools, goal_str
         )
         _tools = self._build_tools(tools)
@@ -264,11 +262,7 @@ def _patch_act_with_cache(
         settings: ActSettings,
         tools: list[Tool] | ToolCollection | None,
         goal: str,
-    ) -> tuple[
-        list[Tool] | ToolCollection,
-        ExecuteCachedTrajectory | None,
-        CacheManager | None,
-    ]:
+    ) -> tuple[list[Tool] | ToolCollection, CacheManager | None]:
         """Patch act settings and tools with caching functionality.
 
         Args:
@@ -278,10 +272,9 @@ def _patch_act_with_cache(
             goal: The goal string for cache recording
 
         Returns:
-            A tuple of (modified_tools, cached_execution_tool, cache_manager)
+            A tuple of (modified_tools, cache_manager)
         """
         caching_tools: list[Tool] = []
-        cached_execution_tool: ExecuteCachedTrajectory | None = None
         cache_manager: CacheManager | None = None
 
         # Setup execute mode: add caching tools and modify system prompt
@@ -290,12 +283,11 @@ def _patch_act_with_cache(
             cache_executor = CacheExecutor(caching_settings.execution_settings)
             self._conversation.speakers.add_speaker(cache_executor)
 
-            # Add caching tools
-            cached_execution_tool = ExecuteCachedTrajectory()
+            # Add caching tools (switch_speaker tool is added automatically
+            # by Conversation._setup_speaker_handoff)
             caching_tools.extend(
                 [
                     RetrieveCachedTestExecutions(caching_settings.cache_dir),
-                    cached_execution_tool,
                     VerifyCacheExecution(),
                     InspectCacheMetadata(),
                 ]
@@ -328,7 +320,7 @@ def _patch_act_with_cache(
                 vlm_provider=self._vlm_provider,
             )
 
-        return tools, cached_execution_tool, cache_manager
+        return tools, cache_manager
 
     @overload
     def get(

src/askui/models/shared/conversation.py

@@ -10,8 +10,6 @@
 from askui.model_providers.vlm_provider import VlmProvider
 from askui.models.shared.agent_message_param import (
     MessageParam,
-    ToolResultBlockParam,
-    ToolUseBlockParam,
     UsageParam,
 )
 from askui.models.shared.settings import ActSettings
@@ -23,6 +21,7 @@
 )
 from askui.reporting import NULL_REPORTER, Reporter
 from askui.speaker.speaker import SpeakerResult, Speakers
+from askui.tools.switch_speaker_tool import SwitchSpeakerTool
 
 if TYPE_CHECKING:
     from askui.models.shared.conversation_callback import ConversationCallback
@@ -103,9 +102,6 @@ def __init__(
         self._reporters: list[Reporter] = []
         self._step_index: int = 0
 
-        # Cache execution context (for communication between tools and CacheExecutor)
-        self.cache_execution_context: dict[str, Any] = {}
-
         # Track if cache execution was used (to prevent recording during playback)
         self._executed_from_cache: bool = False
 
@@ -162,7 +158,6 @@ def _setup_control_loop(
     ) -> None:
         # Reset state
         self.accumulated_usage = UsageParam()
-        self.cache_execution_context = {}
         self._executed_from_cache = False
         self.speakers.reset_state()
 
@@ -171,6 +166,9 @@ def _setup_control_loop(
         self.tools = tools or ToolCollection()
         self._reporters = reporters or []
 
+        # Auto-populate speaker descriptions and switch_speaker tool
+        self._setup_speaker_handoff()
+
         # Initialize truncation strategy
         self._truncation_strategy = (
             self._truncation_strategy_factory.create_truncation_strategy(
@@ -199,6 +197,51 @@ def _conclude_control_loop(self) -> None:
         # Report final usage
         self._reporter.add_usage_summary(self.accumulated_usage.model_dump())
 
+    def _setup_speaker_handoff(self) -> None:
+        """Set up speaker handoff infrastructure.
+
+        If there are speakers with descriptions (handoff targets), this method:
+        1. Appends an ``<AVAILABLE_SPEAKERS>`` section to ``system_capabilities``
+        2. Adds a ``SwitchSpeakerTool`` to the tool collection
+        """
+        speaker_descriptions = self._build_speaker_descriptions()
+        if not speaker_descriptions:
+            return
+
+        # Append speaker descriptions to system_capabilities
+        if self.settings.messages.system is not None:
+            has_capabilities = self.settings.messages.system.system_capabilities
+            separator = "\n\n" if has_capabilities else ""
+            self.settings.messages.system.system_capabilities += (
+                f"{separator}<AVAILABLE_SPEAKERS>\n"
+                "The following specialized speakers are available in this "
+                "conversation. Use the switch_speaker tool to hand off to "
+                "them when appropriate.\n\n"
+                f"{speaker_descriptions}\n"
+                "</AVAILABLE_SPEAKERS>"
+            )
+
+        # Create switch_speaker tool with valid speaker names
+        handoff_speakers = [
+            speaker.get_name() for speaker in self.speakers if speaker.get_description()
+        ]
+        switch_tool = SwitchSpeakerTool(speaker_names=handoff_speakers)
+        self.tools.append_tool(switch_tool)
+
+    def _build_speaker_descriptions(self) -> str:
+        """Build formatted speaker descriptions for the system prompt.
+
+        Returns:
+            Formatted string with speaker names and descriptions,
+            or empty string if no speakers have descriptions.
+        """
+        descriptions: list[str] = []
+        for speaker in self.speakers:
+            description = speaker.get_description()
+            if description:
+                descriptions.append(f"### {speaker.get_name()}\n{description}")
+        return "\n\n".join(descriptions)
+
     @tracer.start_as_current_span("step")
     def _execute_step(self) -> bool:
         """Execute one step of the conversation loop with speakers.
@@ -238,14 +281,13 @@ def _execute_step(self) -> bool:
             tool_result_message = self._execute_tools_if_present(last_message)
             if tool_result_message:
                 self._add_message(tool_result_message)
-
-                # Handle side effects of tool execution (e.g., speaker switches)
-                self._handle_tool_results(last_message, tool_result_message)
-
                 continue_loop = True  # we always continue after a tool was called
 
         # 4. Check if conversation should continue and switch speaker if necessary
-        continue_loop = continue_loop or self._handle_result_status(result)
+        # Note: _handle_result_status must always be called (not short-circuited)
+        # because it has side effects (e.g., triggering speaker switches).
+        status_continue = self._handle_result_status(result)
+        continue_loop = continue_loop or status_continue
 
         # 5. Collect Statistics
         if result.usage:
@@ -295,67 +337,6 @@ def _execute_tools_if_present(self, message: MessageParam) -> MessageParam | Non
         # Return tool results as a user message
         return MessageParam(content=tool_results, role="user")
 
-    @tracer.start_as_current_span("handle_tool_result")
-    def _handle_tool_results(
-        self,
-        assistant_message: MessageParam,
-        tool_result_message: MessageParam,
-    ) -> None:
-        """Handle side effects of tool execution.
-
-        Extracts tool use blocks and tool results from messages, then checks
-        if specific tools require speaker switches or other actions.
-
-        Currently handles:
-        - ExecuteCachedTrajectory: Switches to CacheExecutor if successful
-
-        Args:
-            assistant_message: The assistant message containing tool use blocks
-            tool_result_message: The user message containing tool results
-        """
-        # Extract tool use blocks from assistant message
-        if isinstance(assistant_message.content, str):
-            return
-
-        tool_use_blocks: list[ToolUseBlockParam] = [
-            block for block in assistant_message.content if block.type == "tool_use"
-        ]
-
-        if isinstance(tool_result_message.content, str):
-            return
-
-        tool_results: list[ToolResultBlockParam] = tool_result_message.content  # type: ignore[assignment]
-
-        # Handle side effects for each tool
-        for tool_use_block, tool_result in zip(
-            tool_use_blocks, tool_results, strict=False
-        ):
-            # Check if ExecuteCachedTrajectory was called successfully
-            if (
-                tool_use_block.name.startswith("execute_cached_executions_tool")
-                and not tool_result.is_error
-            ):
-                # Extract parameters from tool call (input is dict at runtime)
-                trajectory_file: str = tool_use_block.input["trajectory_file"]  # type: ignore[index]
-                start_from_step_index: int = tool_use_block.input.get(  # type: ignore[attr-defined]
-                    "start_from_step_index", 0
-                )
-                parameter_values: dict[str, str] = tool_use_block.input.get(  # type: ignore[attr-defined]
-                    "parameter_values", {}
-                )
-
-                # Prepare cache execution context for CacheExecutor
-                # CacheExecutor will validate and load the cache file
-                self.cache_execution_context = {
-                    "trajectory_file": trajectory_file,
-                    "start_from_step_index": start_from_step_index,
-                    "parameter_values": parameter_values,
-                    "toolbox": self.tools,
-                    "reporter": self._reporter,
-                }
-                self._executed_from_cache = True
-                self.switch_speaker("CacheExecutor")
-
     def _add_message(self, message: MessageParam) -> None:
         """Add message to conversation history.
 
@@ -392,17 +373,26 @@ def _handle_result_status(self, result: SpeakerResult) -> bool:
             return False
         if result.status == "switch_speaker":
             if result.next_speaker:
-                self.switch_speaker(result.next_speaker)
+                self.switch_speaker(
+                    result.next_speaker,
+                    speaker_context=result.speaker_context,
+                )
             return True
         # status == "continue"
         return True
 
     @tracer.start_as_current_span("switch_speaker")
-    def switch_speaker(self, speaker_name: str) -> None:
-        """Switch to a different speaker.
+    def switch_speaker(
+        self,
+        speaker_name: str,
+        speaker_context: dict[str, Any] | None = None,
+    ) -> None:
+        """Switch to a different speaker, optionally passing activation context.
 
         Args:
-            speaker_name: Name of the speaker to switch to
+            speaker_name: Name of the speaker to switch to.
+            speaker_context: Optional activation context to pass to the
+                target speaker via ``on_activate()``.
         """
         old_speaker = self.current_speaker
         self.current_speaker = self.speakers[speaker_name]
@@ -411,6 +401,8 @@ def switch_speaker(self, speaker_name: str) -> None:
             old_speaker.get_name(),
             self.current_speaker.get_name(),
         )
+        if speaker_context is not None:
+            self.current_speaker.on_activate(speaker_context)
 
     def get_messages(self) -> list[MessageParam]:
         """Get current message history from truncation strategy.

src/askui/prompts/act_prompts.py

@@ -388,27 +388,29 @@
     "CRITICAL: Before taking ANY action, you MUST first call the"
     " retrieve_available_trajectories_tool to check for cached trajectories. If the"
     " name of an available cached trajectory matches the one specified by the user,"
-    " you MUST execute it using the execute_cached_executions_tool before calling any"
-    " other tools!\n"
+    " you MUST switch to the CacheExecutor speaker using the switch_speaker tool"
+    " before calling any other tools!\n"
     "\n"
     "WORKFLOW:\n"
     "1. ALWAYS start by calling retrieve_available_trajectories_tool\n"
-    "2. If a matching cached trajectory exists, execute it immediately using"
-    " the execute_cached_executions_tool"
+    "2. If a matching cached trajectory exists, switch to CacheExecutor using"
+    " the switch_speaker tool with speaker_context containing the trajectory details\n"
     "3. Only proceed with manual execution if no matching trajectory is available\n"
     "\n"
     "EXECUTING TRAJECTORIES:\n"
-    "- Use execute_cached_executions_tool to run cached trajectories\n"
+    "- Use switch_speaker(speaker_name='CacheExecutor', speaker_context={"
+    "'trajectory_file': '<path>', 'parameter_values': {...}}) to start execution\n"
     "- Trajectories contain complete sequences of mouse movements, clicks, and typing"
     " from successful executions\n"
     "- You'll see all screenshots and results in message history\n"
     "- Verify results after execution completes\n"
     "\n"
     "DYNAMIC PARAMETERS:\n"
     "- Trajectories may require parameters like {{current_date}} or {{user_name}}\n"
-    "- Provide values via parameter_values as a dictionary\n"
-    "- Example: execute_cached_executions_tool(trajectory_file='test.json',"
-    " parameter_values={'current_date': '2025-12-11'})\n"
+    "- Provide values via parameter_values in the speaker_context\n"
+    "- Example: switch_speaker(speaker_name='CacheExecutor', speaker_context={"
+    "'trajectory_file': 'test.json', 'parameter_values': {"
+    "'current_date': '2025-12-11'}})\n"
     "- Missing required parameters will cause execution failure with an error message."
     " In that case try again with providing the correct parameters\n"
     "\n"
@@ -417,13 +419,13 @@
     "- Trajectory pauses at non-cacheable steps, returning NEEDS_AGENT status with"
     " current step index\n"
     "- Execute the non-cacheable step manually\n"
-    "- Resume using execute_cached_executions_tool with start_from_step_index"
-    " parameter\n"
+    "- Resume by switching to CacheExecutor again with start_from_step_index"
+    " in the speaker_context\n"
     "\n"
     "CONTINUING TRAJECTORIES:\n"
-    "- Resume after non-cacheable steps:"
-    " execute_cached_executions_tool(trajectory_file='test.json',"
-    " start_from_step_index=5, parameter_values={...})\n"
+    "- Resume after non-cacheable steps: switch_speaker(speaker_name='CacheExecutor',"
+    " speaker_context={'trajectory_file': 'test.json',"
+    " 'start_from_step_index': 5, 'parameter_values': {...}})\n"
     "\n"
     "FAILURE HANDLING:\n"
     "- On failure, you'll see the error and failed step index\n"

src/askui/prompts/caching.py

@@ -34,30 +34,3 @@
 }
 
 If no parameters are found, return an empty parameters array."""
-
-
-CACHE_USE_PROMPT = (
-    "<TRAJECTORY USE>\n"
-    "    You can use precomputed trajectories to make the execution of the "
-    "task more robust and faster!\n"
-    "    To do so, first use the RetrieveCachedTestExecutions tool to check "
-    "which trajectories are available for you.\n"
-    "    The details what each trajectory that is available for you does are "
-    "at the end of this prompt.\n"
-    "    A trajectory contains all necessary mouse movements, clicks, and "
-    "typing actions from a previously successful execution.\n"
-    "    If there is a trajectory available for a step you need to take, "
-    "always use it!\n"
-    "    You can execute a trajectory with the ExecuteCachedExecution tool.\n"
-    "    After a trajectory was executed, make sure to verify the results! "
-    "While it works most of the time, occasionally, the execution can be "
-    "(partly) incorrect. So make sure to verify if everything is filled out "
-    "as expected, and make corrections where necessary!\n"
-    "    </TRAJECTORY USE>\n"
-    "    <TRAJECTORY DETAILS>\n"
-    "    There are several trajectories available to you.\n"
-    "    Their filename is a unique testID.\n"
-    "    If executed using the ExecuteCachedExecution tool, a trajectory will "
-    "automatically execute all necessary steps for the test with that id.\n"
-    "    </TRAJECTORY DETAILS>\n"
-)