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)