From 5301f44b3b1618d23ffca5e4af39675f0680f827 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 13:51:43 -0500 Subject: [PATCH] Add pre- and post-actions --- examples/foundational/25-conversation-flow.py | 8 +- src/pipecat/flows/manager.py | 103 ++++++++++-------- src/pipecat/flows/state.py | 35 ++++-- 3 files changed, 89 insertions(+), 57 deletions(-) diff --git a/examples/foundational/25-conversation-flow.py b/examples/foundational/25-conversation-flow.py index 6806db332..84434d0a2 100644 --- a/examples/foundational/25-conversation-flow.py +++ b/examples/foundational/25-conversation-flow.py @@ -81,12 +81,12 @@ flow_config = { }, } ], - "actions": [{"type": "tts_say", "text": "Let me help you order a pizza..."}], + "pre_actions": [{"type": "tts_say", "text": "Ok, let me pull up our pizza menu..."}], }, "choose_sushi": { "message": { "role": "system", - "content": "The user has chosen sushi. Immediately ask them: 'How many sushi rolls would you like to order?' If they answer provide to the question of how many rolls, use the select_roll_count function.", + "content": "The user has chosen sushi. Immediately say: 'How many sushi rolls would you like to order?' If they answer provide to the question of how many rolls, use the select_roll_count function.", }, "functions": [ { @@ -109,7 +109,7 @@ flow_config = { }, } ], - "actions": [{"type": "tts_say", "text": "Ok, one moment..."}], + "pre_actions": [{"type": "tts_say", "text": "Ok, one moment..."}], }, }, } @@ -164,7 +164,7 @@ async def main(): task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) # Initialize flow manager - flow_manager = FlowManager(flow_config, task) + flow_manager = FlowManager(flow_config, task, tts) # Register functions with LLM service await flow_manager.register_functions(llm) diff --git a/src/pipecat/flows/manager.py b/src/pipecat/flows/manager.py index df335dc1a..2a94c99d5 100644 --- a/src/pipecat/flows/manager.py +++ b/src/pipecat/flows/manager.py @@ -26,7 +26,8 @@ class FlowManager: represents a state in the conversation. Each node has: - A message for the LLM - Available functions that can be called - - Optional actions to execute when entering the node + - Optional pre-actions to execute before LLM inference + - Optional post-actions to execute after LLM inference The flow is defined by a configuration that specifies: - Initial node @@ -34,7 +35,7 @@ class FlowManager: - Transitions between nodes via function calls """ - def __init__(self, flow_config: dict, task): + def __init__(self, flow_config: dict, task, tts=None): """Initialize the flow manager. Args: @@ -45,6 +46,7 @@ class FlowManager: self.flow = FlowState(flow_config) self.initialized = False self.task = task + self.tts = tts self.action_handlers: Dict[str, Callable] = {} # Register built-in actions @@ -96,46 +98,6 @@ class FlowManager: function_name = function["function"]["name"] llm_service.register_function(function_name, handle_function_call) - async def handle_transition(self, function_name: str): - """Handle node transition triggered by a function call. - - This method: - 1. Validates the function call against available functions - 2. Transitions to the new node if appropriate - 3. Executes any actions associated with the new node - 4. Updates the LLM context with new messages and available functions - - Args: - function_name: Name of the function that was called - - Raises: - RuntimeError: If handle_transition is called before initialization - """ - if not self.initialized: - raise RuntimeError("FlowManager must be initialized before handling transitions") - - available_functions = self.flow.get_available_function_names() - - if function_name in available_functions: - new_node = self.flow.transition(function_name) - if new_node: - if self.flow.get_current_actions(): - await self._execute_actions(self.flow.get_current_actions()) - - current_message = self.flow.get_current_message() - - await self.task.queue_frame(LLMMessagesAppendFrame(messages=[current_message])) - await self.task.queue_frame( - LLMSetToolsFrame(tools=self.flow.get_current_functions()) - ) - - logger.debug(f"Transition to node {new_node} complete") - else: - logger.warning( - f"Received invalid function call '{function_name}' for node '{self.flow.current_node}'. " - f"Available functions are: {available_functions}" - ) - def register_action(self, action_type: str, handler: Callable): """Register a handler for a specific action type. @@ -175,5 +137,58 @@ class FlowManager: logger.warning(f"No handler registered for action type: {action_type}") async def _handle_tts_action(self, action: dict): - """Built-in handler for tts_say actions""" - await self.task.queue_frame(TTSSpeakFrame(text=action["text"])) + """Built-in handler for TTS actions""" + if self.tts: + # Direct call to TTS service to speak text + await self.tts.say(action["text"]) + else: + # Fall back to queued TTS if no direct service available + await self.task.queue_frame(TTSSpeakFrame(text=action["text"])) + + async def handle_transition(self, function_name: str): + """Handle node transition triggered by a function call. + + This method: + 1. Validates the function call against available functions + 2. Transitions to the new node if appropriate + 3. Executes any pre-actions before updating the LLM context + 4. Updates the LLM context with new messages and available functions + 5. Executes any post-actions after updating the LLM context + + Args: + function_name: Name of the function that was called + + Raises: + RuntimeError: If handle_transition is called before initialization + """ + if not self.initialized: + raise RuntimeError("FlowManager must be initialized before handling transitions") + + available_functions = self.flow.get_available_function_names() + + if function_name in available_functions: + new_node = self.flow.transition(function_name) + if new_node: + # Execute pre-actions before updating LLM context + if self.flow.get_current_pre_actions(): + logger.debug(f"Executing pre-actions for node {new_node}") + await self._execute_actions(self.flow.get_current_pre_actions()) + + # Update LLM context and tools + current_message = self.flow.get_current_message() + await self.task.queue_frame(LLMMessagesAppendFrame(messages=[current_message])) + await self.task.queue_frame( + LLMSetToolsFrame(tools=self.flow.get_current_functions()) + ) + + # Execute post-actions after updating LLM context + if self.flow.get_current_post_actions(): + logger.debug(f"Executing post-actions for node {new_node}") + await self._execute_actions(self.flow.get_current_post_actions()) + + logger.debug(f"Transition to node {new_node} complete") + else: + logger.warning( + f"Received invalid function call '{function_name}' for node '{self.flow.current_node}'. " + f"Available functions are: {available_functions}" + ) diff --git a/src/pipecat/flows/state.py b/src/pipecat/flows/state.py index e22b8bd5a..5c7464b8b 100644 --- a/src/pipecat/flows/state.py +++ b/src/pipecat/flows/state.py @@ -20,12 +20,14 @@ class NodeConfig: Attributes: message: Dict containing role and content for the LLM at this node functions: List of available function definitions for this node - actions: Optional list of actions to execute when entering this node + pre_actions: Optional list of actions to execute before LLM inference + post_actions: Optional list of actions to execute after LLM inference """ message: dict functions: List[dict] - actions: Optional[List[dict]] = None + pre_actions: Optional[List[dict]] = None + post_actions: Optional[List[dict]] = None class FlowState: @@ -33,8 +35,8 @@ class FlowState: This class handles the state machine logic for conversation flows, where each node represents a distinct state with its own message, available functions, and optional - actions. It manages transitions between nodes based on function calls and handles - both regular and terminal functions. + pre- and post-actions. It manages transitions between nodes based on function calls + and handles both regular and terminal functions. Attributes: nodes: Dictionary mapping node IDs to their configurations @@ -73,7 +75,8 @@ class FlowState: self.nodes[node_id] = NodeConfig( message=node_config["message"], functions=node_config["functions"], - actions=node_config.get("actions"), + pre_actions=node_config.get("pre_actions"), + post_actions=node_config.get("post_actions"), ) def get_current_message(self) -> dict: @@ -92,13 +95,27 @@ class FlowState: """ return self.nodes[self.current_node].functions - def get_current_actions(self) -> Optional[List[dict]]: - """Get the actions for the current node. + def get_current_pre_actions(self) -> Optional[List[dict]]: + """Get the pre-actions for the current node. + + Pre-actions are executed before updating the LLM context when + transitioning to this node. Returns: - List of actions to execute when entering the node, or None if no actions + List of pre-actions to execute, or None if no pre-actions """ - return self.nodes[self.current_node].actions + return self.nodes[self.current_node].pre_actions + + def get_current_post_actions(self) -> Optional[List[dict]]: + """Get the post-actions for the current node. + + Post-actions are executed after updating the LLM context when + transitioning to this node. + + Returns: + List of post-actions to execute, or None if no post-actions + """ + return self.nodes[self.current_node].post_actions def get_available_function_names(self) -> Set[str]: """Get the names of available functions for the current node.