Merge pull request #133 from pipecat-ai/support-waiting-until-user-speaks-attempt-2

Support waiting until user speaks attempt 2

Author: kompfner

CHANGELOG.md

@@ -13,6 +13,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Added `respond_immediately` to `NodeConfig`. Setting it to `False` has the effect of making the
+  bot wait, after the node is activated, for the user to speak before responding.
+
 - Bumped the minimum required `pipecat-ai` version to 0.0.67 to align with AWS
   Bedrock additions in Pipecat. This also adds support for `FunctionCallParams`
   which were added in 0.0.66.

examples/dynamic/restaurant_reservation.py

@@ -26,6 +26,8 @@
 from pipecat_flows import FlowArgs, FlowManager, FlowResult, FlowsFunctionSchema, NodeConfig
 
 sys.path.append(str(Path(__file__).parent.parent))
+import argparse
+
 from runner import configure
 
 load_dotenv(override=True)
@@ -166,7 +168,7 @@ async def handle_end(_: Dict, result: FlowResult, flow_manager: FlowManager):
 
 
 # Node configurations
-def create_initial_node() -> NodeConfig:
+def create_initial_node(wait_for_user: bool) -> NodeConfig:
     """Create initial node for party size collection."""
     return {
         "role_messages": [
@@ -178,10 +180,11 @@ def create_initial_node() -> NodeConfig:
         "task_messages": [
             {
                 "role": "system",
-                "content": "Warmly greet the customer and ask how many people are in their party.",
+                "content": "Warmly greet the customer and ask how many people are in their party. This is your only job for now; if the customer asks for something else, politely remind them you can't do it.",
             }
         ],
         "functions": [party_size_schema],
+        "respond_immediately": not wait_for_user,
     }
 
 
@@ -245,7 +248,7 @@ def create_end_node() -> NodeConfig:
 
 
 # Main setup
-async def main():
+async def main(wait_for_user: bool):
     async with aiohttp.ClientSession() as session:
         (room_url, _) = await configure(session)
 
@@ -297,11 +300,19 @@ async def on_first_participant_joined(transport, participant):
             logger.debug("Initializing flow manager")
             await flow_manager.initialize()
             logger.debug("Setting initial node")
-            await flow_manager.set_node("initial", create_initial_node())
+            await flow_manager.set_node("initial", create_initial_node(wait_for_user))
 
         runner = PipelineRunner()
         await runner.run(task)
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    parser = argparse.ArgumentParser(description="Restaurant reservation bot")
+    parser.add_argument(
+        "--wait-for-user",
+        action="store_true",
+        help="If set, the bot will wait for the user to speak first",
+    )
+    args = parser.parse_args()
+
+    asyncio.run(main(args.wait_for_user))

examples/runner.py

@@ -8,7 +8,11 @@
 import os
 
 import aiohttp
-from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyMeetingTokenParams, DailyMeetingTokenProperties
+from pipecat.transports.services.helpers.daily_rest import (
+    DailyMeetingTokenParams,
+    DailyMeetingTokenProperties,
+    DailyRESTHelper,
+)
 
 
 async def configure(aiohttp_session: aiohttp.ClientSession):
@@ -58,11 +62,9 @@ async def configure_with_args(
     expiry_time: float = 60 * 60
 
     token = await daily_rest_helper.get_token(
-        url, 
-        expiry_time, 
-        params=DailyMeetingTokenParams(properties=DailyMeetingTokenProperties(
-            user_id="bot"
-        ))
+        url,
+        expiry_time,
+        params=DailyMeetingTokenParams(properties=DailyMeetingTokenProperties(user_id="bot")),
     )
 
     return (url, token, args)

src/pipecat_flows/actions.py

@@ -22,17 +22,18 @@
 """
 
 import asyncio
-from dataclasses import dataclass
 import inspect
+from dataclasses import dataclass
 from typing import Callable, Dict, List, Optional
 
 from loguru import logger
 from pipecat.frames.frames import (
+    BotStoppedSpeakingFrame,
+    ControlFrame,
     EndFrame,
     TTSSpeakFrame,
 )
 from pipecat.pipeline.task import PipelineTask
-from pipecat.frames.frames import ControlFrame
 
 from .exceptions import ActionError
 from .types import ActionConfig, FlowActionHandler
@@ -73,19 +74,23 @@ def __init__(self, task: PipelineTask, flow_manager: "FlowManager", tts=None):
         self._flow_manager = flow_manager
         self.tts = tts
         self.function_finished_event = asyncio.Event()
+        self._deferred_post_actions: List[ActionConfig] = []
 
         # Register built-in actions
         self._register_action("tts_say", self._handle_tts_action)
         self._register_action("end_conversation", self._handle_end_action)
         self._register_action("function", self._handle_function_action)
 
         # Wire up function actions
-        task.set_reached_downstream_filter((FunctionActionFrame,))
+        task.set_reached_downstream_filter((FunctionActionFrame, BotStoppedSpeakingFrame))
+
         @task.event_handler("on_frame_reached_downstream")
         async def on_frame_reached_downstream(task, frame):
             if isinstance(frame, FunctionActionFrame):
                 await frame.function(frame.action, flow_manager)
                 self.function_finished_event.set()
+            elif isinstance(frame, BotStoppedSpeakingFrame):
+                await self._execute_deferred_post_actions()
 
     def _register_action(self, action_type: str, handler: Callable) -> None:
         """Register a handler for a specific action type.
@@ -159,6 +164,25 @@ async def execute_actions(self, actions: Optional[List[ActionConfig]]) -> None:
             except Exception as e:
                 raise ActionError(f"Failed to execute action {action_type}: {str(e)}") from e
 
+    def schedule_deferred_post_actions(self, post_actions: List[ActionConfig]) -> None:
+        """Schedule "deferred" post-actions to be executed after next LLM completion.
+
+        Args:
+            post_actions: List of actions to execute
+        """
+        self._deferred_post_actions = post_actions
+
+    def clear_deferred_post_actions(self) -> None:
+        """Clear any scheduled deferred post-actions."""
+        self._deferred_post_actions = []
+
+    async def _execute_deferred_post_actions(self) -> None:
+        """Execute deferred post-actions."""
+        actions = self._deferred_post_actions
+        self._deferred_post_actions = []
+        if actions:
+            await self.execute_actions(actions)
+
     async def _handle_tts_action(self, action: dict) -> None:
         """Built-in handler for TTS actions.
 
@@ -209,7 +233,7 @@ async def _handle_function_action(self, action: dict) -> None:
         if not handler:
             logger.error("Function action missing 'handler' field")
             return
-        # the reason we're queueing a frame here is to ensure it happens after bot turn is over in 
+        # the reason we're queueing a frame here is to ensure it happens after bot turn is over in
         # post_actions
         await self.task.queue_frame(FunctionActionFrame(action=action, function=handler))
         await self.function_finished_event.wait()

src/pipecat_flows/manager.py

@@ -482,6 +482,9 @@ async def set_node(self, node_id: str, node_config: NodeConfig) -> None:
             self._validate_node_config(node_id, node_config)
             logger.debug(f"Setting node: {node_id}")
 
+            # Clear any deferred post-actions from previous node
+            self.action_manager.clear_deferred_post_actions()
+
             # Register action handlers from config
             for action_list in [
                 node_config.get("pre_actions", []),
@@ -558,19 +561,28 @@ async def register_function_schema(schema):
             self.current_functions = new_functions
 
             # Trigger completion with new context
-            if self._context_aggregator:
+            respond_immediately = node_config.get("respond_immediately", True)
+            if self._context_aggregator and respond_immediately:
                 await self.task.queue_frames([self._context_aggregator.user().get_context_frame()])
 
             # Execute post-actions if any
             if post_actions := node_config.get("post_actions"):
-                await self._execute_actions(post_actions=post_actions)
+                if respond_immediately:
+                    await self._execute_actions(post_actions=post_actions)
+                else:
+                    # Schedule post-actions for execution after first LLM response in this node
+                    print("[pk] Scheduling post-actions for execution after LLM response")
+                    self._schedule_deferred_post_actions(post_actions=post_actions)
 
             logger.debug(f"Successfully set node: {node_id}")
 
         except Exception as e:
             logger.error(f"Error setting node {node_id}: {str(e)}")
             raise FlowError(f"Failed to set node {node_id}: {str(e)}") from e
 
+    def _schedule_deferred_post_actions(self, post_actions: List[ActionConfig]) -> None:
+        self.action_manager.schedule_deferred_post_actions(post_actions=post_actions)
+
     async def _create_conversation_summary(
         self, summary_prompt: str, messages: List[dict]
     ) -> Optional[str]:

src/pipecat_flows/types.py

@@ -228,6 +228,7 @@ class NodeConfig(NodeConfigRequired, total=False):
         pre_actions: Actions to execute before LLM inference
         post_actions: Actions to execute after LLM inference
         context_strategy: Strategy for updating context during transitions
+        respond_immediately: Whether to run LLM inference as soon as the node is set (default: True)
 
     Example:
         {
@@ -254,6 +255,7 @@ class NodeConfig(NodeConfigRequired, total=False):
     pre_actions: List[ActionConfig]
     post_actions: List[ActionConfig]
     context_strategy: ContextStrategyConfig
+    respond_immediately: bool
 
 
 class FlowConfig(TypedDict):