From 0c1070433f387d15ea7052f0992e3934760b3f77 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 10:42:48 -0500 Subject: [PATCH] Clean up and commenting --- .../processors/conversation_flow/flow.py | 81 ++++++++++++++--- .../processors/conversation_flow/processor.py | 88 ++++++++++++++----- 2 files changed, 136 insertions(+), 33 deletions(-) diff --git a/src/pipecat/processors/conversation_flow/flow.py b/src/pipecat/processors/conversation_flow/flow.py index 51359ca8c..e0d3371e3 100644 --- a/src/pipecat/processors/conversation_flow/flow.py +++ b/src/pipecat/processors/conversation_flow/flow.py @@ -12,7 +12,13 @@ from loguru import logger @dataclass class NodeConfig: - """Configuration for a single node in the conversation flow""" + """Configuration for a single node in the conversation flow. + + 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 + """ message: dict functions: List[dict] @@ -20,15 +26,41 @@ class NodeConfig: class ConversationFlow: - """Manages the state and transitions of the conversation flow""" + """Manages state transitions in a conversation flow. + + This class handles the state machine logic for conversation flows, where each state + (node) has its own message, available functions, and optional actions. It manages + transitions between states based on function calls and handles both regular and + terminal functions. + + Attributes: + nodes: Dictionary mapping node IDs to their configurations + current_node: ID of the currently active node + """ def __init__(self, flow_config: dict): + """Initialize the conversation flow. + + Args: + flow_config: Dictionary containing the complete flow configuration, + must include 'initial_node' and 'nodes' keys + + Raises: + ValueError: If required configuration keys are missing + """ self.nodes: Dict[str, NodeConfig] = {} self.current_node: str = flow_config["initial_node"] self._load_config(flow_config) def _load_config(self, config: dict): - """Load and validate the flow configuration""" + """Load and validate the flow configuration. + + Args: + config: Dictionary containing the flow configuration + + Raises: + ValueError: If required configuration keys are missing + """ if "initial_node" not in config: raise ValueError("Flow config must specify 'initial_node'") if "nodes" not in config: @@ -42,38 +74,63 @@ class ConversationFlow: ) def get_current_message(self) -> dict: - """Get the message for the current node""" + """Get the message configuration for the current node. + + Returns: + Dictionary containing role and content for the current node's message + """ return self.nodes[self.current_node].message def get_current_functions(self) -> List[dict]: - """Get the available functions for the current node""" + """Get the available functions for the current node. + + Returns: + List of function definitions available in the current node + """ return self.nodes[self.current_node].functions def get_current_actions(self) -> Optional[List[dict]]: - """Get the actions for the current node""" + """Get the actions for the current node. + + Returns: + List of actions to execute when entering the node, or None if no actions + """ return self.nodes[self.current_node].actions def get_available_function_names(self) -> Set[str]: - """Get the names of available functions for the current node""" + """Get the names of available functions for the current node. + + Returns: + Set of function names that can be called from the current node + """ names = {f["function"]["name"] for f in self.nodes[self.current_node].functions} logger.debug(f"Available function names for node {self.current_node}: {names}") return names def transition(self, function_name: str) -> Optional[str]: - """Attempt to transition based on function call""" + """Attempt to transition based on a function call. + + Handles both regular transitions (where the function name matches a node) + and terminal functions (which execute but don't change nodes). + + Args: + function_name: Name of the function that was called + + Returns: + The name of the new node after transition, or None if transition failed. + For terminal functions, returns the current node name. + """ available_functions = self.get_available_function_names() - logger.info(f"Attempting transition with {function_name}") - logger.info(f"Current node: {self.current_node}") - logger.info(f"Available functions: {available_functions}") + logger.debug(f"Attempting transition from {self.current_node} to {function_name}") if function_name in available_functions: if function_name in self.nodes: + # Regular transition to a new node self.current_node = function_name logger.info(f"Transitioned to node: {self.current_node}") return self.current_node else: # Handle terminal function calls logger.info(f"Executed terminal function: {function_name}") - # Optionally, could transition to a "completion" node or stay in current node return self.current_node return None diff --git a/src/pipecat/processors/conversation_flow/processor.py b/src/pipecat/processors/conversation_flow/processor.py index ce3ee8158..8d12c6b0d 100644 --- a/src/pipecat/processors/conversation_flow/processor.py +++ b/src/pipecat/processors/conversation_flow/processor.py @@ -16,48 +16,85 @@ from .flow import ConversationFlow # processor.py class ConversationFlowProcessor(FrameProcessor): - """Processor that manages conversation flow based on function calls""" + """Processor that manages conversation flow based on function calls. + + This processor maintains conversation state and handles transitions between states + based on LLM function calls. Each state (node) has its own message, available + functions, and optional actions. The processor ensures the LLM context is updated + appropriately as the conversation progresses. + + The flow is defined by a configuration that specifies: + - Initial state + - Available states (nodes) + - Messages for each state + - Available functions for each state + - Optional actions for each state + """ def __init__(self, flow_config: dict): + """Initialize the conversation flow processor. + + Args: + flow_config: Dictionary containing the complete flow configuration, + including initial_node, nodes, and their configurations + """ super().__init__() self.flow = ConversationFlow(flow_config) self.initialized = False async def initialize(self, initial_messages: List[dict]): - """Initialize the conversation with starting messages and functions""" + """Initialize the conversation with starting messages and functions. + + This method sets up the initial context for the conversation, combining + any system-level messages with the initial node's message and functions. + + Args: + initial_messages: List of initial messages (typically system messages) + to include in the context + + Note: + This must be called before the processor can handle any frames. + """ if not self.initialized: messages = initial_messages + [self.flow.get_current_message()] - logger.info(f"Initializing with messages: {messages}") - logger.info(f"Initial tools: {self.flow.get_current_functions()}") - await self.push_frame(LLMMessagesUpdateFrame(messages=messages)) await self.push_frame(LLMSetToolsFrame(tools=self.flow.get_current_functions())) self.initialized = True - logger.info(f"Initialized conversation flow at node: {self.flow.current_node}") + logger.debug(f"Initialized conversation flow at node: {self.flow.current_node}") else: logger.warning("Attempted to initialize ConversationFlowProcessor multiple times") async def handle_transition(self, function_name: str): - """Handle state transition triggered by function call""" - logger.info(f"Handling transition for function: {function_name}") + """Handle state transition triggered by a function call. + + This method: + 1. Validates the function call against available functions + 2. Transitions to the new state if appropriate + 3. Executes any actions associated with the new state + 4. Updates the LLM context with new messages and available functions + + Args: + function_name: Name of the function that was called + """ + if not self.initialized: + raise RuntimeError( + "ConversationFlowProcessor must be initialized before handling transitions" + ) + available_functions = self.flow.get_available_function_names() - logger.info(f"Available functions: {available_functions}") if function_name in available_functions: new_node = self.flow.transition(function_name) if new_node: if self.flow.get_current_actions(): - logger.info(f"Executing actions for node {new_node}") await self._execute_actions(self.flow.get_current_actions()) current_message = self.flow.get_current_message() - logger.info(f"New node message: {current_message}") - logger.info(f"New node functions: {self.flow.get_current_functions()}") await self.push_frame(LLMMessagesAppendFrame(messages=[current_message])) await self.push_frame(LLMSetToolsFrame(tools=self.flow.get_current_functions())) - logger.info("Transition complete") + logger.debug(f"Transition to {new_node} complete") else: logger.warning( f"Received invalid function call '{function_name}' for node '{self.flow.current_node}'. " @@ -65,23 +102,32 @@ class ConversationFlowProcessor(FrameProcessor): ) async def _execute_actions(self, actions: Optional[List[dict]]) -> None: - """Execute actions specified for the current node""" + """Execute actions specified for the current node. + + Currently supports: + - tts.say: Sends a TTSSpeakFrame with the specified text + + Args: + actions: List of action configurations to execute + """ if not actions: return for action in actions: if action["type"] == "tts.say": - logger.info(f"Executing TTS action: {action['text']}") + logger.debug(f"Executing TTS action: {action['text']}") await self.push_frame(TTSSpeakFrame(text=action["text"])) else: logger.warning(f"Unknown action type: {action['type']}") async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: - """Process incoming frames and manage state transitions""" - if not self.initialized: - logger.warning("ConversationFlowProcessor received frames before initialization") - await self.push_frame(frame, direction) - return + """Pass frames through the processor. - # Pass all frames through + State transitions are handled via function calls rather than frames, + so this processor only needs to maintain the pipeline flow. + + Args: + frame: The frame to process + direction: Direction the frame is flowing through the pipeline + """ await self.push_frame(frame, direction)