From 0b9742da9ecd1d0a2ba8a51bf5425793a826f365 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 08:37:44 -0500 Subject: [PATCH 01/13] Add a conversation flow processor --- examples/foundational/25-conversation-flow.py | 186 ++++++++++++++++++ .../processors/conversation_flow/__init__.py | 3 + .../processors/conversation_flow/flow.py | 69 +++++++ .../processors/conversation_flow/processor.py | 83 ++++++++ 4 files changed, 341 insertions(+) create mode 100644 examples/foundational/25-conversation-flow.py create mode 100644 src/pipecat/processors/conversation_flow/__init__.py create mode 100644 src/pipecat/processors/conversation_flow/flow.py create mode 100644 src/pipecat/processors/conversation_flow/processor.py diff --git a/examples/foundational/25-conversation-flow.py b/examples/foundational/25-conversation-flow.py new file mode 100644 index 000000000..78c28a9d7 --- /dev/null +++ b/examples/foundational/25-conversation-flow.py @@ -0,0 +1,186 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os +import sys + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +from pipecat.audio.vad.silero import SileroVADAnalyzer +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 + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +# Define our conversation flow +flow_config = { + "initial_node": "start", + "nodes": { + "start": { + "message": { + "role": "assistant", + "content": "You are starting a conversation. Ask the user if they'd like to hear a joke or get weather information.", + }, + "functions": [ + { + "name": "tell_joke", + "description": "User wants to hear a joke", + "parameters": {"type": "object", "properties": {}}, + }, + { + "name": "get_weather", + "description": "User wants weather information", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The location to get weather for", + } + }, + "required": ["location"], + }, + }, + ], + }, + "tell_joke": { + "message": { + "role": "assistant", + "content": "Tell a funny, clean joke and then ask if they'd like to hear another joke or get weather information.", + }, + "functions": [ + { + "name": "tell_joke", + "description": "User wants another joke", + "parameters": {"type": "object", "properties": {}}, + }, + { + "name": "get_weather", + "description": "User wants weather information", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The location to get weather for", + } + }, + "required": ["location"], + }, + }, + ], + "actions": [{"type": "tts.say", "text": "Let me think of a good one..."}], + }, + "get_weather": { + "message": { + "role": "assistant", + "content": "Provide the weather information and ask if they'd like to hear a joke or check another location's weather.", + }, + "functions": [ + { + "name": "tell_joke", + "description": "User wants to hear a joke", + "parameters": {"type": "object", "properties": {}}, + }, + { + "name": "get_weather", + "description": "User wants weather for another location", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The location to get weather for", + } + }, + "required": ["location"], + }, + }, + ], + "actions": [ + {"type": "tts.say", "text": "Let me check that weather information for you..."} + ], + }, + }, +} + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, _) = await configure(session) + + transport = DailyTransport( + room_url, + None, + "Respond bot", + DailyParams( + audio_out_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, + ), + ) + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4") + + # Initialize conversation flow processor + flow_processor = ConversationFlowProcessor(flow_config) + + # Create initial context + messages = [ + { + "role": "system", + "content": "You are a helpful assistant in a WebRTC call. Your responses will be converted to audio so avoid special characters. Always use the available functions to progress the conversation.", + } + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + 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 + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + + @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) + # Kick off the conversation + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/pipecat/processors/conversation_flow/__init__.py b/src/pipecat/processors/conversation_flow/__init__.py new file mode 100644 index 000000000..d8dab4228 --- /dev/null +++ b/src/pipecat/processors/conversation_flow/__init__.py @@ -0,0 +1,3 @@ +from .processor import ConversationFlowProcessor + +__all__ = ["ConversationFlowProcessor"] diff --git a/src/pipecat/processors/conversation_flow/flow.py b/src/pipecat/processors/conversation_flow/flow.py new file mode 100644 index 000000000..5f41378e2 --- /dev/null +++ b/src/pipecat/processors/conversation_flow/flow.py @@ -0,0 +1,69 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from dataclasses import dataclass +from typing import Dict, List, Optional, Set + +from loguru import logger + + +@dataclass +class NodeConfig: + """Configuration for a single node in the conversation flow""" + + message: dict + functions: List[dict] + actions: Optional[List[dict]] = None + + +class ConversationFlow: + """Manages the state and transitions of the conversation flow""" + + def __init__(self, flow_config: dict): + 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""" + if "initial_node" not in config: + raise ValueError("Flow config must specify 'initial_node'") + if "nodes" not in config: + raise ValueError("Flow config must specify 'nodes'") + + for node_id, node_config in config["nodes"].items(): + self.nodes[node_id] = NodeConfig( + message=node_config["message"], + functions=node_config["functions"], + actions=node_config.get("actions"), + ) + + def get_current_message(self) -> dict: + """Get the message for the current node""" + return self.nodes[self.current_node].message + + def get_current_functions(self) -> List[dict]: + """Get the available functions for 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""" + 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""" + return {f["name"] for f in self.nodes[self.current_node].functions} + + def transition(self, function_name: str) -> Optional[str]: + """Attempt to transition based on function call""" + available_functions = self.get_available_function_names() + if function_name in available_functions: + if function_name in self.nodes: + self.current_node = function_name + return self.current_node + else: + logger.warning(f"Function {function_name} is available but no matching node exists") + return None diff --git a/src/pipecat/processors/conversation_flow/processor.py b/src/pipecat/processors/conversation_flow/processor.py new file mode 100644 index 000000000..01a90218c --- /dev/null +++ b/src/pipecat/processors/conversation_flow/processor.py @@ -0,0 +1,83 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from typing import List, Optional + +from loguru import logger + +from pipecat.frames.frames import ( + Frame, + FunctionCallResultFrame, + LLMMessagesAppendFrame, + LLMMessagesUpdateFrame, + LLMSetToolsFrame, + TTSSpeakFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + +from .flow import ConversationFlow + + +class ConversationFlowProcessor(FrameProcessor): + """Processor that manages conversation flow based on function calls""" + + def __init__(self, flow_config: dict): + 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""" + if not self.initialized: + # Combine initial messages with the first node's message + # TODO: Not sure if this is needed + messages = initial_messages + [self.flow.get_current_message()] + + await self.push_frame(LLMMessagesUpdateFrame(messages=initial_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}") + else: + logger.warning("Attempted to initialize ConversationFlowProcessor multiple times") + + async def _execute_actions(self, actions: Optional[List[dict]]) -> None: + """Execute actions specified for the current node""" + if not actions: + return + + for action in actions: + if action["type"] == "tts.say": + 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") + return + + if isinstance(frame, FunctionCallResultFrame): + available_functions = self.flow.get_available_function_names() + if frame.function_name in available_functions: + new_node = self.flow.transition(frame.function_name) + if new_node: + # Execute any entry actions for the new node + await self._execute_actions(self.flow.get_current_actions()) + + # Update the LLM context with the new node's message + await self.push_frame( + LLMMessagesAppendFrame(messages=[self.flow.get_current_message()]) + ) + + # Update available functions for this node + await self.push_frame(LLMSetToolsFrame(tools=self.flow.get_current_functions())) + + logger.info(f"Transitioned to node: {new_node}") + else: + logger.warning( + f"Received function call '{frame.function_name}' not in available functions: {available_functions}" + ) From ece2c08cde36ca47e17c3bef5f09e74121accba5 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 09:31:47 -0500 Subject: [PATCH 02/13] debugging --- examples/foundational/25-conversation-flow.py | 136 ++++++++++-------- .../processors/conversation_flow/flow.py | 14 +- .../processors/conversation_flow/processor.py | 66 +++++---- 3 files changed, 122 insertions(+), 94 deletions(-) diff --git a/examples/foundational/25-conversation-flow.py b/examples/foundational/25-conversation-flow.py index 78c28a9d7..52aefdd0b 100644 --- a/examples/foundational/25-conversation-flow.py +++ b/examples/foundational/25-conversation-flow.py @@ -34,88 +34,82 @@ flow_config = { "nodes": { "start": { "message": { - "role": "assistant", - "content": "You are starting a conversation. Ask the user if they'd like to hear a joke or get weather information.", + "role": "system", + "content": "You are an order-taking assistant. You must ALWAYS use one of the available functions to progress the conversation. For this step, ask the user if they want pizza or sushi, and wait for them to use a function to choose.", }, "functions": [ { - "name": "tell_joke", - "description": "User wants to hear a joke", - "parameters": {"type": "object", "properties": {}}, + "type": "function", + "function": { + "name": "choose_pizza", + "description": "User wants to order pizza", + "parameters": {"type": "object", "properties": {}}, + }, }, { - "name": "get_weather", - "description": "User wants weather information", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "The location to get weather for", - } - }, - "required": ["location"], + "type": "function", + "function": { + "name": "choose_sushi", + "description": "User wants to order sushi", + "parameters": {"type": "object", "properties": {}}, }, }, ], }, - "tell_joke": { + "choose_pizza": { "message": { - "role": "assistant", - "content": "Tell a funny, clean joke and then ask if they'd like to hear another joke or get weather information.", + "role": "system", + "content": "The user has chosen pizza. You must now ask them to select a size using the select_pizza_size function. Do not proceed until they use this function. Do not assume any selections have been made.", }, "functions": [ { - "name": "tell_joke", - "description": "User wants another joke", - "parameters": {"type": "object", "properties": {}}, - }, - { - "name": "get_weather", - "description": "User wants weather information", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "The location to get weather for", - } + "type": "function", + "function": { + "name": "select_pizza_size", + "description": "Select pizza size", + "parameters": { + "type": "object", + "properties": { + "size": { + "type": "string", + "enum": ["small", "medium", "large"], + "description": "Size of the pizza", + } + }, + "required": ["size"], }, - "required": ["location"], }, - }, + } ], - "actions": [{"type": "tts.say", "text": "Let me think of a good one..."}], + "actions": [{"type": "tts.say", "text": "Let me help you order a pizza..."}], }, - "get_weather": { + "choose_sushi": { "message": { - "role": "assistant", - "content": "Provide the weather information and ask if they'd like to hear a joke or check another location's weather.", + "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.", }, "functions": [ { - "name": "tell_joke", - "description": "User wants to hear a joke", - "parameters": {"type": "object", "properties": {}}, - }, - { - "name": "get_weather", - "description": "User wants weather for another location", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "The location to get weather for", - } + "type": "function", + "function": { + "name": "select_roll_count", + "description": "Select number of sushi rolls", + "parameters": { + "type": "object", + "properties": { + "count": { + "type": "integer", + "minimum": 1, + "maximum": 10, + "description": "Number of rolls to order", + } + }, + "required": ["count"], }, - "required": ["location"], }, - }, - ], - "actions": [ - {"type": "tts.say", "text": "Let me check that weather information for you..."} + } ], + "actions": [{"type": "tts.say", "text": "Ok, one moment..."}], }, }, } @@ -144,15 +138,35 @@ async def main(): # Initialize conversation flow processor flow_processor = ConversationFlowProcessor(flow_config) + # Get initial tools from the first node + initial_tools = flow_config["nodes"]["start"]["functions"] + # Create initial context messages = [ { "role": "system", - "content": "You are a helpful assistant in a WebRTC call. Your responses will be converted to audio so avoid special characters. Always use the available functions to progress the conversation.", + "content": "You are an order-taking assistant. You must ALWAYS use the available functions to progress the conversation. Never assume an order is complete without the proper function calls. Your responses will be converted to audio so avoid special characters.", } ] - context = OpenAILLMContext(messages) + # 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) pipeline = Pipeline( @@ -175,7 +189,7 @@ async def main(): await transport.capture_participant_transcription(participant["id"]) # Initialize the flow processor await flow_processor.initialize(messages) - # Kick off the conversation + # Kick off the conversation using the context aggregator await task.queue_frames([context_aggregator.user().get_context_frame()]) runner = PipelineRunner() diff --git a/src/pipecat/processors/conversation_flow/flow.py b/src/pipecat/processors/conversation_flow/flow.py index 5f41378e2..51359ca8c 100644 --- a/src/pipecat/processors/conversation_flow/flow.py +++ b/src/pipecat/processors/conversation_flow/flow.py @@ -55,15 +55,25 @@ class ConversationFlow: def get_available_function_names(self) -> Set[str]: """Get the names of available functions for the current node""" - return {f["name"] for f in self.nodes[self.current_node].functions} + 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""" 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}") + if function_name in available_functions: if function_name in self.nodes: self.current_node = function_name + logger.info(f"Transitioned to node: {self.current_node}") return self.current_node else: - logger.warning(f"Function {function_name} is available but no matching node exists") + # 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 01a90218c..ce3ee8158 100644 --- a/src/pipecat/processors/conversation_flow/processor.py +++ b/src/pipecat/processors/conversation_flow/processor.py @@ -1,16 +1,9 @@ -# -# Copyright (c) 2024, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - from typing import List, Optional from loguru import logger from pipecat.frames.frames import ( Frame, - FunctionCallResultFrame, LLMMessagesAppendFrame, LLMMessagesUpdateFrame, LLMSetToolsFrame, @@ -21,6 +14,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from .flow import ConversationFlow +# processor.py class ConversationFlowProcessor(FrameProcessor): """Processor that manages conversation flow based on function calls""" @@ -32,17 +26,44 @@ class ConversationFlowProcessor(FrameProcessor): async def initialize(self, initial_messages: List[dict]): """Initialize the conversation with starting messages and functions""" if not self.initialized: - # Combine initial messages with the first node's message - # TODO: Not sure if this is needed 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=initial_messages)) + 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}") 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}") + 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") + else: + logger.warning( + f"Received invalid function call '{function_name}' for node '{self.flow.current_node}'. " + f"Available functions are: {available_functions}" + ) + async def _execute_actions(self, actions: Optional[List[dict]]) -> None: """Execute actions specified for the current node""" if not actions: @@ -50,6 +71,7 @@ class ConversationFlowProcessor(FrameProcessor): for action in actions: if action["type"] == "tts.say": + logger.info(f"Executing TTS action: {action['text']}") await self.push_frame(TTSSpeakFrame(text=action["text"])) else: logger.warning(f"Unknown action type: {action['type']}") @@ -58,26 +80,8 @@ class ConversationFlowProcessor(FrameProcessor): """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 - if isinstance(frame, FunctionCallResultFrame): - available_functions = self.flow.get_available_function_names() - if frame.function_name in available_functions: - new_node = self.flow.transition(frame.function_name) - if new_node: - # Execute any entry actions for the new node - await self._execute_actions(self.flow.get_current_actions()) - - # Update the LLM context with the new node's message - await self.push_frame( - LLMMessagesAppendFrame(messages=[self.flow.get_current_message()]) - ) - - # Update available functions for this node - await self.push_frame(LLMSetToolsFrame(tools=self.flow.get_current_functions())) - - logger.info(f"Transitioned to node: {new_node}") - else: - logger.warning( - f"Received function call '{frame.function_name}' not in available functions: {available_functions}" - ) + # Pass all frames through + await self.push_frame(frame, direction) From 0c1070433f387d15ea7052f0992e3934760b3f77 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 10:42:48 -0500 Subject: [PATCH 03/13] 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) From 1b74560f9d3e2a923a85d0a0c382ced29061317e Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 10:54:18 -0500 Subject: [PATCH 04/13] Move function registration into the ConversationFlowProcessor class --- examples/foundational/25-conversation-flow.py | 9 ++- .../processors/conversation_flow/processor.py | 61 ++++++++++++------- 2 files changed, 45 insertions(+), 25 deletions(-) diff --git a/examples/foundational/25-conversation-flow.py b/examples/foundational/25-conversation-flow.py index 52aefdd0b..ee8828676 100644 --- a/examples/foundational/25-conversation-flow.py +++ b/examples/foundational/25-conversation-flow.py @@ -135,9 +135,6 @@ async def main(): tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en") llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4") - # Initialize conversation flow processor - flow_processor = ConversationFlowProcessor(flow_config) - # Get initial tools from the first node initial_tools = flow_config["nodes"]["start"]["functions"] @@ -184,6 +181,12 @@ async def main(): task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + # Initialize conversation flow processor + flow_processor = ConversationFlowProcessor(flow_config, task) + + # Register functions with LLM service + await flow_processor.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"]) diff --git a/src/pipecat/processors/conversation_flow/processor.py b/src/pipecat/processors/conversation_flow/processor.py index 8d12c6b0d..3f61ff9a7 100644 --- a/src/pipecat/processors/conversation_flow/processor.py +++ b/src/pipecat/processors/conversation_flow/processor.py @@ -1,21 +1,24 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + from typing import List, Optional from loguru import logger from pipecat.frames.frames import ( - Frame, LLMMessagesAppendFrame, LLMMessagesUpdateFrame, LLMSetToolsFrame, TTSSpeakFrame, ) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from .flow import ConversationFlow -# processor.py -class ConversationFlowProcessor(FrameProcessor): +class ConversationFlowProcessor: """Processor that manages conversation flow based on function calls. This processor maintains conversation state and handles transitions between states @@ -31,7 +34,7 @@ class ConversationFlowProcessor(FrameProcessor): - Optional actions for each state """ - def __init__(self, flow_config: dict): + def __init__(self, flow_config: dict, task, **kwargs): """Initialize the conversation flow processor. Args: @@ -41,6 +44,7 @@ class ConversationFlowProcessor(FrameProcessor): super().__init__() self.flow = ConversationFlow(flow_config) self.initialized = False + self.task = task async def initialize(self, initial_messages: List[dict]): """Initialize the conversation with starting messages and functions. @@ -57,13 +61,36 @@ class ConversationFlowProcessor(FrameProcessor): """ if not self.initialized: messages = initial_messages + [self.flow.get_current_message()] - await self.push_frame(LLMMessagesUpdateFrame(messages=messages)) - await self.push_frame(LLMSetToolsFrame(tools=self.flow.get_current_functions())) + 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}") else: logger.warning("Attempted to initialize ConversationFlowProcessor 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. + + Args: + llm_service: The LLM service to register functions with + """ + + async def handle_function_call( + function_name, tool_call_id, arguments, llm, context, result_callback + ): + await self.handle_transition(function_name) + await result_callback("Acknowledged") + + # Register all functions from all nodes + for node in self.flow.nodes.values(): + for function in node.functions: + function_name = function["function"]["name"] + 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. @@ -91,8 +118,10 @@ class ConversationFlowProcessor(FrameProcessor): current_message = self.flow.get_current_message() - await self.push_frame(LLMMessagesAppendFrame(messages=[current_message])) - await self.push_frame(LLMSetToolsFrame(tools=self.flow.get_current_functions())) + 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 {new_node} complete") else: @@ -116,18 +145,6 @@ class ConversationFlowProcessor(FrameProcessor): for action in actions: if action["type"] == "tts.say": logger.debug(f"Executing TTS action: {action['text']}") - await self.push_frame(TTSSpeakFrame(text=action["text"])) + await self.task.queue_frame(TTSSpeakFrame(text=action["text"])) else: logger.warning(f"Unknown action type: {action['type']}") - - async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: - """Pass frames through the processor. - - 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) From 4e0ecdd673ecaa712fae9a4857b431197c51a34f Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 13:19:05 -0500 Subject: [PATCH 05/13] 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"] From 686165b95aeaacd75525b71ad0f4006c08b8560b Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 13:35:45 -0500 Subject: [PATCH 06/13] Add ability to register actions --- examples/foundational/25-conversation-flow.py | 4 +- src/pipecat/flows/manager.py | 45 +++++++++++++++---- 2 files changed, 38 insertions(+), 11 deletions(-) diff --git a/examples/foundational/25-conversation-flow.py b/examples/foundational/25-conversation-flow.py index df39c9992..6806db332 100644 --- a/examples/foundational/25-conversation-flow.py +++ b/examples/foundational/25-conversation-flow.py @@ -81,7 +81,7 @@ flow_config = { }, } ], - "actions": [{"type": "tts.say", "text": "Let me help you order a pizza..."}], + "actions": [{"type": "tts_say", "text": "Let me help you order a pizza..."}], }, "choose_sushi": { "message": { @@ -109,7 +109,7 @@ flow_config = { }, } ], - "actions": [{"type": "tts.say", "text": "Ok, one moment..."}], + "actions": [{"type": "tts_say", "text": "Ok, one moment..."}], }, }, } diff --git a/src/pipecat/flows/manager.py b/src/pipecat/flows/manager.py index 0e0d78c77..df335dc1a 100644 --- a/src/pipecat/flows/manager.py +++ b/src/pipecat/flows/manager.py @@ -4,7 +4,8 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from typing import List, Optional +from asyncio import iscoroutinefunction +from typing import Callable, Dict, List, Optional from loguru import logger @@ -41,10 +42,13 @@ class FlowManager: including initial_node and node configurations task: PipelineTask instance used to queue frames into the pipeline """ - super().__init__() self.flow = FlowState(flow_config) self.initialized = False self.task = task + self.action_handlers: Dict[str, Callable] = {} + + # Register built-in actions + self.register_action("tts_say", self._handle_tts_action) async def initialize(self, initial_messages: List[dict]): """Initialize the flow with starting messages and functions. @@ -132,21 +136,44 @@ class FlowManager: f"Available functions are: {available_functions}" ) + def register_action(self, action_type: str, handler: Callable): + """Register a handler for a specific action type. + + Args: + action_type: String identifier for the action (e.g., "tts_say") + handler: Async or sync function that handles the action. + Should accept action configuration as parameter. + """ + if not callable(handler): + raise ValueError("Action handler must be callable") + self.action_handlers[action_type] = handler + async def _execute_actions(self, actions: Optional[List[dict]]) -> None: """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 + + Note: + Each action must have a 'type' field matching a registered handler """ if not actions: return for action in actions: - if action["type"] == "tts.say": - logger.debug(f"Executing TTS action: {action['text']}") - await self.task.queue_frame(TTSSpeakFrame(text=action["text"])) + action_type = action["type"] + if action_type in self.action_handlers: + handler = self.action_handlers[action_type] + try: + if iscoroutinefunction(handler): + await handler(action) + else: + handler(action) + except Exception as e: + logger.warning(f"Error executing action {action_type}: {e}") else: - logger.warning(f"Unknown action type: {action['type']}") + 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"])) From 5301f44b3b1618d23ffca5e4af39675f0680f827 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 13:51:43 -0500 Subject: [PATCH 07/13] 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. From b7308dca5ddc8c0a826697012ea62594de401555 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 15:44:25 -0500 Subject: [PATCH 08/13] Fix issue where actions would execute on terminating nodes --- src/pipecat/flows/manager.py | 8 ++++++-- src/pipecat/flows/state.py | 7 ++++++- 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/src/pipecat/flows/manager.py b/src/pipecat/flows/manager.py index 2a94c99d5..83e1f858e 100644 --- a/src/pipecat/flows/manager.py +++ b/src/pipecat/flows/manager.py @@ -165,12 +165,16 @@ class FlowManager: raise RuntimeError("FlowManager must be initialized before handling transitions") available_functions = self.flow.get_available_function_names() + current_node = self.flow.get_current_node() if function_name in available_functions: new_node = self.flow.transition(function_name) if new_node: + # Only execute actions if we actually changed nodes + is_new_node = new_node != current_node + # Execute pre-actions before updating LLM context - if self.flow.get_current_pre_actions(): + if is_new_node and 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()) @@ -182,7 +186,7 @@ class FlowManager: ) # Execute post-actions after updating LLM context - if self.flow.get_current_post_actions(): + if is_new_node and 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()) diff --git a/src/pipecat/flows/state.py b/src/pipecat/flows/state.py index 5c7464b8b..21b9939d4 100644 --- a/src/pipecat/flows/state.py +++ b/src/pipecat/flows/state.py @@ -146,11 +146,16 @@ class FlowState: if function_name in available_functions: if function_name in self.nodes: # Regular transition to a new node + previous_node = self.current_node self.current_node = function_name - logger.info(f"Transitioned to node: {self.current_node}") + logger.info(f"Transitioned from {previous_node} to node: {self.current_node}") return self.current_node else: # 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 + + def get_current_node(self) -> str: + """Get the current node ID.""" + return self.current_node From 7a55d2d7db1fd2a2a9090ef6ab5b2870a0753df0 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 16:11:35 -0500 Subject: [PATCH 09/13] Add end session handler and update example --- examples/foundational/25-conversation-flow.py | 85 +++++++++++++++++-- src/pipecat/flows/manager.py | 29 ++++++- 2 files changed, 103 insertions(+), 11 deletions(-) diff --git a/examples/foundational/25-conversation-flow.py b/examples/foundational/25-conversation-flow.py index 84434d0a2..e9e22233e 100644 --- a/examples/foundational/25-conversation-flow.py +++ b/examples/foundational/25-conversation-flow.py @@ -28,7 +28,35 @@ load_dotenv(override=True) logger.remove(0) logger.add(sys.stderr, level="DEBUG") -# Define our conversation flow +# Flow Configuration +# +# This configuration defines a simple food ordering system with the following states: +# +# 1. start +# - Initial state where user chooses between pizza or sushi +# - Functions: choose_pizza, choose_sushi +# - Transitions to: choose_pizza or choose_sushi +# +# 2. choose_pizza +# - Handles pizza size selection and order confirmation +# - Functions: +# * select_pizza_size (terminal function, can be called multiple times) +# * end (transitions to end node after order confirmation) +# - Pre-action: Immediate TTS acknowledgment +# +# 3. choose_sushi +# - Handles sushi roll count selection and order confirmation +# - Functions: +# * select_roll_count (terminal function, can be called multiple times) +# * end (transitions to end node after order confirmation) +# - Pre-action: Immediate TTS acknowledgment +# +# 4. end +# - Final state that closes the conversation +# - No functions available +# - Pre-action: Farewell message +# - Post-action: Ends conversation + flow_config = { "initial_node": "start", "nodes": { @@ -59,14 +87,19 @@ flow_config = { "choose_pizza": { "message": { "role": "system", - "content": "The user has chosen pizza. You must now ask them to select a size using the select_pizza_size function. Do not proceed until they use this function. Do not assume any selections have been made.", + "content": """You are handling a pizza order. Use the available functions: + - Use select_pizza_size when the user specifies a size (can be used multiple times if they change their mind) + - Use the end function ONLY when the user confirms they are done with their order + + After each size selection, confirm the selection and ask if they want to change it or complete their order. + Only use the end function after the user confirms they are satisfied with their order.""", }, "functions": [ { "type": "function", "function": { "name": "select_pizza_size", - "description": "Select pizza size", + "description": "Record the selected pizza size", "parameters": { "type": "object", "properties": { @@ -79,21 +112,36 @@ flow_config = { "required": ["size"], }, }, - } + }, + { + "type": "function", + "function": { + "name": "end", + "description": "Complete the order (use only after user confirms)", + "parameters": {"type": "object", "properties": {}}, + }, + }, + ], + "pre_actions": [ + {"type": "tts_say", "text": "Ok, let me help you with your pizza order..."} ], - "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 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.", + "content": """You are handling a sushi order. Use the available functions: + - Use select_roll_count when the user specifies how many rolls (can be used multiple times if they change their mind) + - Use the end function ONLY when the user confirms they are done with their order + + After each roll count selection, confirm the count and ask if they want to change it or complete their order. + Only use the end function after the user confirms they are satisfied with their order.""", }, "functions": [ { "type": "function", "function": { "name": "select_roll_count", - "description": "Select number of sushi rolls", + "description": "Record the number of sushi rolls", "parameters": { "type": "object", "properties": { @@ -107,9 +155,28 @@ flow_config = { "required": ["count"], }, }, - } + }, + { + "type": "function", + "function": { + "name": "end", + "description": "Complete the order (use only after user confirms)", + "parameters": {"type": "object", "properties": {}}, + }, + }, ], - "pre_actions": [{"type": "tts_say", "text": "Ok, one moment..."}], + "pre_actions": [ + {"type": "tts_say", "text": "Ok, let me help you with your sushi order..."} + ], + }, + "end": { + "message": { + "role": "system", + "content": "The order is complete. Thank the user and end the conversation.", + }, + "functions": [], + "pre_actions": [{"type": "tts_say", "text": "Thank you for your order! Goodbye!"}], + "post_actions": [{"type": "end_conversation"}], }, }, } diff --git a/src/pipecat/flows/manager.py b/src/pipecat/flows/manager.py index 83e1f858e..a923628ac 100644 --- a/src/pipecat/flows/manager.py +++ b/src/pipecat/flows/manager.py @@ -10,6 +10,7 @@ from typing import Callable, Dict, List, Optional from loguru import logger from pipecat.frames.frames import ( + EndFrame, LLMMessagesAppendFrame, LLMMessagesUpdateFrame, LLMSetToolsFrame, @@ -51,6 +52,7 @@ class FlowManager: # Register built-in actions self.register_action("tts_say", self._handle_tts_action) + self.register_action("end_conversation", self._handle_end_action) async def initialize(self, initial_messages: List[dict]): """Initialize the flow with starting messages and functions. @@ -137,14 +139,37 @@ 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 actions""" + """Built-in handler for TTS actions that speak immediately. + + This handler attempts to use the TTS service directly to speak the text + immediately, bypassing the pipeline queue. If no TTS service is available, + it falls back to queueing the text through the pipeline. + + Args: + action: Dictionary containing the action configuration. + Must include a 'text' key with the text to speak. + """ if self.tts: - # Direct call to TTS service to speak text + # Direct call to TTS service to speak text immediately 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_end_action(self, action: dict): + """Built-in handler for ending the conversation. + + This handler queues an EndFrame to terminate the conversation. If the action + includes a 'text' key, it will queue that text to be spoken before ending. + + Args: + action: Dictionary containing the action configuration. + Optional 'text' key for a goodbye message. + """ + if action.get("text"): # Optional goodbye message + await self.task.queue_frame(TTSSpeakFrame(text=action["text"])) + await self.task.queue_frame(EndFrame()) + async def handle_transition(self, function_name: str): """Handle node transition triggered by a function call. From bd020320cd2d240c050ebf4387910d6e975c7543 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 17:37:33 -0500 Subject: [PATCH 10/13] Support a list of messages --- examples/foundational/25-conversation-flow.py | 40 +++++++++++-------- src/pipecat/flows/manager.py | 8 ++-- src/pipecat/flows/state.py | 18 +++++---- 3 files changed, 38 insertions(+), 28 deletions(-) diff --git a/examples/foundational/25-conversation-flow.py b/examples/foundational/25-conversation-flow.py index e9e22233e..4ecedb992 100644 --- a/examples/foundational/25-conversation-flow.py +++ b/examples/foundational/25-conversation-flow.py @@ -61,10 +61,12 @@ flow_config = { "initial_node": "start", "nodes": { "start": { - "message": { - "role": "system", - "content": "You are an order-taking assistant. You must ALWAYS use one of the available functions to progress the conversation. For this step, ask the user if they want pizza or sushi, and wait for them to use a function to choose.", - }, + "messages": [ + { + "role": "system", + "content": "You are an order-taking assistant. You must ALWAYS use one of the available functions to progress the conversation. For this step, ask the user if they want pizza or sushi, and wait for them to use a function to choose.", + } + ], "functions": [ { "type": "function", @@ -85,15 +87,17 @@ flow_config = { ], }, "choose_pizza": { - "message": { - "role": "system", - "content": """You are handling a pizza order. Use the available functions: + "messages": [ + { + "role": "system", + "content": """You are handling a pizza order. Use the available functions: - Use select_pizza_size when the user specifies a size (can be used multiple times if they change their mind) - Use the end function ONLY when the user confirms they are done with their order After each size selection, confirm the selection and ask if they want to change it or complete their order. Only use the end function after the user confirms they are satisfied with their order.""", - }, + } + ], "functions": [ { "type": "function", @@ -127,15 +131,17 @@ flow_config = { ], }, "choose_sushi": { - "message": { - "role": "system", - "content": """You are handling a sushi order. Use the available functions: + "messages": [ + { + "role": "system", + "content": """You are handling a sushi order. Use the available functions: - Use select_roll_count when the user specifies how many rolls (can be used multiple times if they change their mind) - Use the end function ONLY when the user confirms they are done with their order After each roll count selection, confirm the count and ask if they want to change it or complete their order. Only use the end function after the user confirms they are satisfied with their order.""", - }, + } + ], "functions": [ { "type": "function", @@ -170,10 +176,12 @@ flow_config = { ], }, "end": { - "message": { - "role": "system", - "content": "The order is complete. Thank the user and end the conversation.", - }, + "messages": [ + { + "role": "system", + "content": "The order is complete. Thank the user and end the conversation.", + } + ], "functions": [], "pre_actions": [{"type": "tts_say", "text": "Thank you for your order! Goodbye!"}], "post_actions": [{"type": "end_conversation"}], diff --git a/src/pipecat/flows/manager.py b/src/pipecat/flows/manager.py index a923628ac..d19541f73 100644 --- a/src/pipecat/flows/manager.py +++ b/src/pipecat/flows/manager.py @@ -25,7 +25,7 @@ class FlowManager: 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 + - Messages for the LLM (system/user/assistant messages) - Available functions that can be called - Optional pre-actions to execute before LLM inference - Optional post-actions to execute after LLM inference @@ -65,7 +65,7 @@ class FlowManager: to include in the context """ if not self.initialized: - messages = initial_messages + [self.flow.get_current_message()] + messages = initial_messages + self.flow.get_current_messages() await self.task.queue_frame(LLMMessagesUpdateFrame(messages=messages)) await self.task.queue_frame(LLMSetToolsFrame(tools=self.flow.get_current_functions())) self.initialized = True @@ -204,8 +204,8 @@ class FlowManager: 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])) + current_messages = self.flow.get_current_messages() + await self.task.queue_frame(LLMMessagesAppendFrame(messages=current_messages)) await self.task.queue_frame( LLMSetToolsFrame(tools=self.flow.get_current_functions()) ) diff --git a/src/pipecat/flows/state.py b/src/pipecat/flows/state.py index 21b9939d4..d1e284589 100644 --- a/src/pipecat/flows/state.py +++ b/src/pipecat/flows/state.py @@ -18,13 +18,15 @@ class NodeConfig: information needed for that particular point in the conversation. Attributes: - message: Dict containing role and content for the LLM at this node + messages: List of message dicts to be added to LLM context at this node. + Each message should have 'role' (system/user/assistant) and 'content'. + Messages are added in order, allowing for complex prompt building. functions: List of available function definitions for 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 + messages: List[dict] functions: List[dict] pre_actions: Optional[List[dict]] = None post_actions: Optional[List[dict]] = None @@ -34,7 +36,7 @@ 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 node - represents a distinct state with its own message, available functions, and optional + represents a distinct state with its own messages, available functions, and optional pre- and post-actions. It manages transitions between nodes based on function calls and handles both regular and terminal functions. @@ -73,19 +75,19 @@ class FlowState: for node_id, node_config in config["nodes"].items(): self.nodes[node_id] = NodeConfig( - message=node_config["message"], + messages=node_config["messages"], functions=node_config["functions"], pre_actions=node_config.get("pre_actions"), post_actions=node_config.get("post_actions"), ) - def get_current_message(self) -> dict: - """Get the message configuration for the current node. + def get_current_messages(self) -> List[dict]: + """Get the messages for the current node. Returns: - Dictionary containing role and content for the current node's message + List of message dictionaries for the current node """ - return self.nodes[self.current_node].message + return self.nodes[self.current_node].messages def get_current_functions(self) -> List[dict]: """Get the available functions for the current node. From b3dfeb61c448d0e1ea3cf7e48d8d8c6b26c3c13b Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 17:48:16 -0500 Subject: [PATCH 11/13] Add CHANGELOG entry --- CHANGELOG.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5df4a88bd..46417a0ce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `STTMuteFrame`, a control frame that enables/disables speech transcription in STT services. +- Added conversation flow management system through `FlowState` and `FlowManager` classes. + This system enables developers to create structured, multi-turn conversations using + a node-based state machine. Each node can contain: + - Multiple LLM context messages (system/user/assistant) + - Available functions for that state + - Pre- and post-actions for state transitions + - Support for both terminal functions (stay in same node) and transitional functions + - Built-in handlers for immediate TTS feedback and conversation end +- Added `NodeConfig` dataclass for defining conversation states, supporting: + - Multiple messages per node for complex prompt building + - Function definitions for available actions + - Optional pre- and post-action hooks + - Clear separation between node configuration and state management +- Added foundational example `25-conversation-flow.py` showing how to use the new + conversation flow functionality. ## [0.0.48] - 2024-11-10 "Antonio release" From 97659ca3f0cea51085083534826eda164de5f332 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 18 Nov 2024 21:29:35 -0500 Subject: [PATCH 12/13] Use the new pipecat-ai-flows module --- examples/foundational/25-conversation-flow.py | 14 +- pyproject.toml | 1 + src/pipecat/flows/__init__.py | 10 - src/pipecat/flows/manager.py | 223 ------------------ src/pipecat/flows/state.py | 163 ------------- 5 files changed, 11 insertions(+), 400 deletions(-) delete mode 100644 src/pipecat/flows/__init__.py delete mode 100644 src/pipecat/flows/manager.py delete mode 100644 src/pipecat/flows/state.py diff --git a/examples/foundational/25-conversation-flow.py b/examples/foundational/25-conversation-flow.py index 4ecedb992..2b43deb98 100644 --- a/examples/foundational/25-conversation-flow.py +++ b/examples/foundational/25-conversation-flow.py @@ -11,10 +11,10 @@ import sys import aiohttp from dotenv import load_dotenv from loguru import logger +from pipecat_flows import FlowManager 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 @@ -64,7 +64,7 @@ flow_config = { "messages": [ { "role": "system", - "content": "You are an order-taking assistant. You must ALWAYS use one of the available functions to progress the conversation. For this step, ask the user if they want pizza or sushi, and wait for them to use a function to choose.", + "content": "You are an order-taking assistant. You must ALWAYS use one of the available functions to progress the conversation. For this step, ask the user if they want pizza or sushi, and wait for them to use a function to choose. Start off by greeting them. Be friendly and casual; you're taking an order for food over the phone.", } ], "functions": [ @@ -95,7 +95,10 @@ flow_config = { - Use the end function ONLY when the user confirms they are done with their order After each size selection, confirm the selection and ask if they want to change it or complete their order. - Only use the end function after the user confirms they are satisfied with their order.""", + Only use the end function after the user confirms they are satisfied with their order. + + Start off by acknowledging the user's choice. Once they've chosen a size, ask if they'd like anything else. + Remember to be friendly and casual.""", } ], "functions": [ @@ -139,7 +142,10 @@ flow_config = { - Use the end function ONLY when the user confirms they are done with their order After each roll count selection, confirm the count and ask if they want to change it or complete their order. - Only use the end function after the user confirms they are satisfied with their order.""", + Only use the end function after the user confirms they are satisfied with their order. + + Start off by acknowledging the user's choice. Once they've chosen a size, ask if they'd like anything else. + Remember to be friendly and casual.""", } ], "functions": [ diff --git a/pyproject.toml b/pyproject.toml index bd160540d..e47dc75ba 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,6 +51,7 @@ gladia = [ "websockets~=13.1" ] google = [ "google-generativeai~=0.8.3", "google-cloud-texttospeech~=2.17.2" ] gstreamer = [ "pygobject~=3.48.2" ] fireworks = [ "openai~=1.37.2" ] +flows = [ "pipecat-ai-flows~=0.0.1" ] krisp = [ "pipecat-ai-krisp~=0.2.0" ] langchain = [ "langchain~=0.2.14", "langchain-community~=0.2.12", "langchain-openai~=0.1.20" ] livekit = [ "livekit~=0.17.5", "livekit-api~=0.7.1", "tenacity~=8.5.0" ] diff --git a/src/pipecat/flows/__init__.py b/src/pipecat/flows/__init__.py deleted file mode 100644 index 7cd99d0d1..000000000 --- a/src/pipecat/flows/__init__.py +++ /dev/null @@ -1,10 +0,0 @@ -# -# 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/flows/manager.py b/src/pipecat/flows/manager.py deleted file mode 100644 index d19541f73..000000000 --- a/src/pipecat/flows/manager.py +++ /dev/null @@ -1,223 +0,0 @@ -# -# Copyright (c) 2024, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -from asyncio import iscoroutinefunction -from typing import Callable, Dict, List, Optional - -from loguru import logger - -from pipecat.frames.frames import ( - EndFrame, - LLMMessagesAppendFrame, - LLMMessagesUpdateFrame, - LLMSetToolsFrame, - TTSSpeakFrame, -) - -from .state import FlowState - - -class FlowManager: - """Manages conversation flows in a Pipecat pipeline. - - This manager handles the progression through a flow defined by nodes, where each node - represents a state in the conversation. Each node has: - - Messages for the LLM (system/user/assistant messages) - - Available functions that can be called - - 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 - - Available nodes and their configurations - - Transitions between nodes via function calls - """ - - def __init__(self, flow_config: dict, task, tts=None): - """Initialize the flow manager. - - Args: - flow_config: Dictionary containing the complete flow configuration, - including initial_node and node configurations - task: PipelineTask instance used to queue frames into the pipeline - """ - self.flow = FlowState(flow_config) - self.initialized = False - self.task = task - self.tts = tts - self.action_handlers: Dict[str, Callable] = {} - - # Register built-in actions - self.register_action("tts_say", self._handle_tts_action) - self.register_action("end_conversation", self._handle_end_action) - - async def initialize(self, initial_messages: List[dict]): - """Initialize the flow with starting messages 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 - """ - if not self.initialized: - messages = initial_messages + self.flow.get_current_messages() - 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 flow at node: {self.flow.current_node}") - else: - 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 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 - """ - - async def handle_function_call( - function_name, tool_call_id, arguments, llm, context, result_callback - ): - await self.handle_transition(function_name) - await result_callback("Acknowledged") - - # Register all functions from all nodes - for node in self.flow.nodes.values(): - for function in node.functions: - function_name = function["function"]["name"] - llm_service.register_function(function_name, handle_function_call) - - def register_action(self, action_type: str, handler: Callable): - """Register a handler for a specific action type. - - Args: - action_type: String identifier for the action (e.g., "tts_say") - handler: Async or sync function that handles the action. - Should accept action configuration as parameter. - """ - if not callable(handler): - raise ValueError("Action handler must be callable") - self.action_handlers[action_type] = handler - - async def _execute_actions(self, actions: Optional[List[dict]]) -> None: - """Execute actions specified for the current node. - - Args: - actions: List of action configurations to execute - - Note: - Each action must have a 'type' field matching a registered handler - """ - if not actions: - return - - for action in actions: - action_type = action["type"] - if action_type in self.action_handlers: - handler = self.action_handlers[action_type] - try: - if iscoroutinefunction(handler): - await handler(action) - else: - handler(action) - except Exception as e: - logger.warning(f"Error executing action {action_type}: {e}") - else: - logger.warning(f"No handler registered for action type: {action_type}") - - async def _handle_tts_action(self, action: dict): - """Built-in handler for TTS actions that speak immediately. - - This handler attempts to use the TTS service directly to speak the text - immediately, bypassing the pipeline queue. If no TTS service is available, - it falls back to queueing the text through the pipeline. - - Args: - action: Dictionary containing the action configuration. - Must include a 'text' key with the text to speak. - """ - if self.tts: - # Direct call to TTS service to speak text immediately - 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_end_action(self, action: dict): - """Built-in handler for ending the conversation. - - This handler queues an EndFrame to terminate the conversation. If the action - includes a 'text' key, it will queue that text to be spoken before ending. - - Args: - action: Dictionary containing the action configuration. - Optional 'text' key for a goodbye message. - """ - if action.get("text"): # Optional goodbye message - await self.task.queue_frame(TTSSpeakFrame(text=action["text"])) - await self.task.queue_frame(EndFrame()) - - 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() - current_node = self.flow.get_current_node() - - if function_name in available_functions: - new_node = self.flow.transition(function_name) - if new_node: - # Only execute actions if we actually changed nodes - is_new_node = new_node != current_node - - # Execute pre-actions before updating LLM context - if is_new_node and 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_messages = self.flow.get_current_messages() - await self.task.queue_frame(LLMMessagesAppendFrame(messages=current_messages)) - await self.task.queue_frame( - LLMSetToolsFrame(tools=self.flow.get_current_functions()) - ) - - # Execute post-actions after updating LLM context - if is_new_node and 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 deleted file mode 100644 index d1e284589..000000000 --- a/src/pipecat/flows/state.py +++ /dev/null @@ -1,163 +0,0 @@ -# -# Copyright (c) 2024, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -from dataclasses import dataclass -from typing import Dict, List, Optional, Set - -from loguru import logger - - -@dataclass -class NodeConfig: - """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: - messages: List of message dicts to be added to LLM context at this node. - Each message should have 'role' (system/user/assistant) and 'content'. - Messages are added in order, allowing for complex prompt building. - functions: List of available function definitions for this node - pre_actions: Optional list of actions to execute before LLM inference - post_actions: Optional list of actions to execute after LLM inference - """ - - messages: List[dict] - functions: List[dict] - pre_actions: Optional[List[dict]] = None - post_actions: Optional[List[dict]] = None - - -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 node - represents a distinct state with its own messages, available functions, and optional - 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 - 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. - - 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: - raise ValueError("Flow config must specify 'nodes'") - - for node_id, node_config in config["nodes"].items(): - self.nodes[node_id] = NodeConfig( - messages=node_config["messages"], - functions=node_config["functions"], - pre_actions=node_config.get("pre_actions"), - post_actions=node_config.get("post_actions"), - ) - - def get_current_messages(self) -> List[dict]: - """Get the messages for the current node. - - Returns: - List of message dictionaries for the current node - """ - return self.nodes[self.current_node].messages - - def get_current_functions(self) -> List[dict]: - """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_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 pre-actions to execute, or None if no pre-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. - - 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 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). - - Args: - function_name: Name of the function that was called - - Returns: - 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}") - - if function_name in available_functions: - if function_name in self.nodes: - # Regular transition to a new node - previous_node = self.current_node - self.current_node = function_name - logger.info(f"Transitioned from {previous_node} to node: {self.current_node}") - return self.current_node - else: - # 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 - - def get_current_node(self) -> str: - """Get the current node ID.""" - return self.current_node From e4c2f6d4c2e64520c5e418e6e0eae6f77ecfb7f2 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 18 Nov 2024 21:32:53 -0500 Subject: [PATCH 13/13] Update changelog --- CHANGELOG.md | 21 ++++++--------------- 1 file changed, 6 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 46417a0ce..56dd710fa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added a new RTVI message called `disconnect-bot`, which when handled pushes an `EndFrame` to trigger the pipeline to stop. +- Added support for the new Pipecat Flows package (pipecat-ai-flows). Learn + more at: https://github.com/pipecat-ai/pipecat-flows. + +- Added foundational example `25-conversation-flow.py` showing how to use + Pipecat Flows. + ## [0.0.49] - 2024-11-17 ### Added @@ -34,21 +40,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `STTMuteFrame`, a control frame that enables/disables speech transcription in STT services. -- Added conversation flow management system through `FlowState` and `FlowManager` classes. - This system enables developers to create structured, multi-turn conversations using - a node-based state machine. Each node can contain: - - Multiple LLM context messages (system/user/assistant) - - Available functions for that state - - Pre- and post-actions for state transitions - - Support for both terminal functions (stay in same node) and transitional functions - - Built-in handlers for immediate TTS feedback and conversation end -- Added `NodeConfig` dataclass for defining conversation states, supporting: - - Multiple messages per node for complex prompt building - - Function definitions for available actions - - Optional pre- and post-action hooks - - Clear separation between node configuration and state management -- Added foundational example `25-conversation-flow.py` showing how to use the new - conversation flow functionality. ## [0.0.48] - 2024-11-10 "Antonio release"