Add a conversation flow processor

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())

src/pipecat/processors/conversation_flow/__init__.py

@@ -0,0 +1,3 @@
+from .processor import ConversationFlowProcessor
+
+__all__ = ["ConversationFlowProcessor"]

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

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}"
+                )