Clean up and commenting

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.debug(f"Attempting transition from {self.current_node} to {function_name}")
-        logger.info(f"Current node: {self.current_node}")
-        logger.info(f"Available functions: {available_functions}")
 
         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

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"""
+        """Handle state transition triggered by a function call.
-        logger.info(f"Handling transition for function: {function_name}")
+
+        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"""
+        """Pass frames through the processor.
-        if not self.initialized:
-            logger.warning("ConversationFlowProcessor received frames before initialization")
-            await self.push_frame(frame, direction)
-            return
 
-        # 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)