From 4e0ecdd673ecaa712fae9a4857b431197c51a34f Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 13:19:05 -0500 Subject: [PATCH] Class name updates and remove FrameProcessor base class --- examples/foundational/25-conversation-flow.py | 28 ++------ src/pipecat/flows/__init__.py | 10 +++ .../processor.py => flows/manager.py} | 70 ++++++++++--------- .../flow.py => flows/state.py} | 25 ++++--- .../processors/conversation_flow/__init__.py | 3 - 5 files changed, 65 insertions(+), 71 deletions(-) create mode 100644 src/pipecat/flows/__init__.py rename src/pipecat/{processors/conversation_flow/processor.py => flows/manager.py} (66%) rename src/pipecat/{processors/conversation_flow/flow.py => flows/state.py} (83%) delete mode 100644 src/pipecat/processors/conversation_flow/__init__.py diff --git a/examples/foundational/25-conversation-flow.py b/examples/foundational/25-conversation-flow.py index ee8828676..df39c9992 100644 --- a/examples/foundational/25-conversation-flow.py +++ b/examples/foundational/25-conversation-flow.py @@ -14,11 +14,11 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.flows.manager import FlowManager from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.processors.conversation_flow import ConversationFlowProcessor from pipecat.services.deepgram import DeepgramSTTService, DeepgramTTSService from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -146,23 +146,6 @@ async def main(): } ] - # Register function handlers - async def handle_function_call( - function_name, tool_call_id, arguments, llm, context, result_callback - ): - logger.info(f"Function called: {function_name} with arguments: {arguments}") - # Handle the state transition - await flow_processor.handle_transition(function_name) - # Send the acknowledgment - await result_callback("Acknowledged") - logger.info(f"Function call result sent: {function_name}") - - # Register functions from all nodes - for node in flow_config["nodes"].values(): - for function in node["functions"]: - function_name = function["function"]["name"] - llm.register_function(function_name, handle_function_call) - context = OpenAILLMContext(messages, initial_tools) context_aggregator = llm.create_context_aggregator(context) @@ -171,7 +154,6 @@ async def main(): transport.input(), # Transport user input stt, # STT context_aggregator.user(), # User responses - flow_processor, # Conversation flow management llm, # LLM tts, # TTS transport.output(), # Transport bot output @@ -181,17 +163,17 @@ async def main(): task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) - # Initialize conversation flow processor - flow_processor = ConversationFlowProcessor(flow_config, task) + # Initialize flow manager + flow_manager = FlowManager(flow_config, task) # Register functions with LLM service - await flow_processor.register_functions(llm) + await flow_manager.register_functions(llm) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): await transport.capture_participant_transcription(participant["id"]) # Initialize the flow processor - await flow_processor.initialize(messages) + await flow_manager.initialize(messages) # Kick off the conversation using the context aggregator await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/src/pipecat/flows/__init__.py b/src/pipecat/flows/__init__.py new file mode 100644 index 000000000..7cd99d0d1 --- /dev/null +++ b/src/pipecat/flows/__init__.py @@ -0,0 +1,10 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from .manager import FlowManager +from .state import FlowState, NodeConfig + +__all__ = ["FlowState", "FlowManager", "NodeConfig"] diff --git a/src/pipecat/processors/conversation_flow/processor.py b/src/pipecat/flows/manager.py similarity index 66% rename from src/pipecat/processors/conversation_flow/processor.py rename to src/pipecat/flows/manager.py index 3f61ff9a7..0e0d78c77 100644 --- a/src/pipecat/processors/conversation_flow/processor.py +++ b/src/pipecat/flows/manager.py @@ -15,65 +15,66 @@ from pipecat.frames.frames import ( TTSSpeakFrame, ) -from .flow import ConversationFlow +from .state import FlowState -class ConversationFlowProcessor: - """Processor that manages conversation flow based on function calls. +class FlowManager: + """Manages conversation flows in a Pipecat pipeline. - 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. + This manager handles the progression through a flow defined by nodes, where each node + 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 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 + - Initial node + - Available nodes and their configurations + - Transitions between nodes via function calls """ - def __init__(self, flow_config: dict, task, **kwargs): - """Initialize the conversation flow processor. + def __init__(self, flow_config: dict, task): + """Initialize the flow manager. Args: flow_config: Dictionary containing the complete flow configuration, - including initial_node, nodes, and their configurations + including initial_node and node configurations + task: PipelineTask instance used to queue frames into the pipeline """ super().__init__() - self.flow = ConversationFlow(flow_config) + self.flow = FlowState(flow_config) self.initialized = False self.task = task async def initialize(self, initial_messages: List[dict]): - """Initialize the conversation with starting messages and functions. + """Initialize the flow 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. + This method sets up the initial context, 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()] await self.task.queue_frame(LLMMessagesUpdateFrame(messages=messages)) await self.task.queue_frame(LLMSetToolsFrame(tools=self.flow.get_current_functions())) self.initialized = True - logger.debug(f"Initialized conversation flow at node: {self.flow.current_node}") + logger.debug(f"Initialized flow at node: {self.flow.current_node}") else: - logger.warning("Attempted to initialize ConversationFlowProcessor multiple times") + logger.warning("Attempted to initialize FlowManager multiple times") async def register_functions(self, llm_service): """Register all functions from the flow configuration with the LLM service. - This method sets up function handlers for all functions defined in the flow - configuration. When a function is called, it will automatically trigger the - appropriate state transition. + This method sets up function handlers for all functions defined across all nodes. + When a function is called, it will automatically trigger the appropriate node + transition. + + Note: This registers handlers for all possible functions, but the LLM's access + to functions is controlled separately through LLMSetToolsFrame. The LLM will + only see the functions available in the current node. Args: llm_service: The LLM service to register functions with @@ -92,21 +93,22 @@ class ConversationFlowProcessor: llm_service.register_function(function_name, handle_function_call) async def handle_transition(self, function_name: str): - """Handle state transition triggered by a function call. + """Handle node 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 + 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( - "ConversationFlowProcessor must be initialized before handling transitions" - ) + raise RuntimeError("FlowManager must be initialized before handling transitions") available_functions = self.flow.get_available_function_names() @@ -123,7 +125,7 @@ class ConversationFlowProcessor: LLMSetToolsFrame(tools=self.flow.get_current_functions()) ) - logger.debug(f"Transition to {new_node} complete") + 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}'. " diff --git a/src/pipecat/processors/conversation_flow/flow.py b/src/pipecat/flows/state.py similarity index 83% rename from src/pipecat/processors/conversation_flow/flow.py rename to src/pipecat/flows/state.py index e0d3371e3..e22b8bd5a 100644 --- a/src/pipecat/processors/conversation_flow/flow.py +++ b/src/pipecat/flows/state.py @@ -12,7 +12,10 @@ from loguru import logger @dataclass class NodeConfig: - """Configuration for a single node in the conversation flow. + """Configuration for a single node in the flow. + + A node represents a state in the conversation flow, containing all the + information needed for that particular point in the conversation. Attributes: message: Dict containing role and content for the LLM at this node @@ -25,13 +28,13 @@ class NodeConfig: actions: Optional[List[dict]] = None -class ConversationFlow: - """Manages state transitions in a conversation flow. +class FlowState: + """Manages the state and transitions between nodes 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. + 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. Attributes: nodes: Dictionary mapping node IDs to their configurations @@ -108,7 +111,7 @@ class ConversationFlow: return names def transition(self, function_name: str) -> Optional[str]: - """Attempt to transition based on a function call. + """Attempt to transition to a new node 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). @@ -117,8 +120,8 @@ class ConversationFlow: 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. + The ID of the new node after transition, or None if transition failed. + For terminal functions, returns the current node ID. """ available_functions = self.get_available_function_names() logger.debug(f"Attempting transition from {self.current_node} to {function_name}") @@ -130,7 +133,7 @@ class ConversationFlow: logger.info(f"Transitioned to node: {self.current_node}") return self.current_node else: - # Handle terminal function calls + # Handle terminal function calls (functions that don't lead to new nodes) logger.info(f"Executed terminal function: {function_name}") return self.current_node return None diff --git a/src/pipecat/processors/conversation_flow/__init__.py b/src/pipecat/processors/conversation_flow/__init__.py deleted file mode 100644 index d8dab4228..000000000 --- a/src/pipecat/processors/conversation_flow/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .processor import ConversationFlowProcessor - -__all__ = ["ConversationFlowProcessor"]