From 0b9742da9ecd1d0a2ba8a51bf5425793a826f365 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 15 Nov 2024 08:37:44 -0500 Subject: [PATCH] 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}" + )