logging

Make Pipecat Flows an independent package

CHANGELOG.md

@@ -12,12 +12,6 @@ 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.
-
 ### Changed
 
 - Expanded the transcriptions.language module to support a superset of

README.md

@@ -13,6 +13,7 @@ Pipecat is an open source Python framework for building voice and multimodal con
 - **Multimodal Apps**: Combine voice, video, images, and text
 - **Creative Tools**: [Story-telling experiences](https://storytelling-chatbot.fly.dev/) and social companions
 - **Business Solutions**: [Customer intake flows](https://www.youtube.com/watch?v=lDevgsp9vn0) and support bots
+- **Complex conversational flows**: [Refer to Pipecat Flows](https://github.com/pipecat-ai/pipecat-flows) to learn more
 
 ## See it in action
 
@@ -32,6 +33,8 @@ Pipecat is an open source Python framework for building voice and multimodal con
 - **Real-time Processing**: Frame-based pipeline architecture for fluid interactions
 - **Production Ready**: Enterprise-grade WebRTC and Websocket support
 
+💡 Looking to build structured conversations? Check out [Pipecat Flows](https://github.com/pipecat-ai/pipecat-flows) for managing complex conversational states and transitions.
+
 ## Getting started
 
 You can get started with Pipecat running on your local machine, then move your agent processes to the cloud when you’re ready. You can also add a 📞 telephone number, 🖼️ image output, 📺 video input, use different LLMs, and more.

examples/foundational/07-interruptible.py

@@ -10,11 +10,12 @@ import os
 import sys
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import LLMMessagesFrame
+from pipecat.frames.frames import BotSpeakingFrame, Frame, InputAudioRawFrame, LLMMessagesFrame, TTSAudioRawFrame, TextFrame, UserStoppedSpeakingFrame
 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.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -30,6 +31,22 @@ load_dotenv(override=True)
 logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
+class DebugProcessor(FrameProcessor):
+    def __init__(self, name, **kwargs):
+        self._name = name
+        super().__init__(**kwargs)
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if not (
+            isinstance(frame, InputAudioRawFrame)
+            or isinstance(frame, BotSpeakingFrame)
+            or isinstance(frame, TTSAudioRawFrame)
+            or isinstance(frame, TextFrame)
+        ):
+            logger.debug(f"--- {self._name}: {frame} {direction}")
+        await self.push_frame(frame, direction)
+
 
 async def main():
     async with aiohttp.ClientSession() as session:
@@ -63,11 +80,14 @@ async def main():
 
         context = OpenAILLMContext(messages)
         context_aggregator = llm.create_context_aggregator(context)
+        
+        dp = DebugProcessor("dp")
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 context_aggregator.user(),  # User responses
+                dp,
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output

examples/foundational/25-conversation-flow.py

@@ -1,266 +0,0 @@
-#
-# 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 pipecat_flows import FlowManager
-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.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")
-
-# 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": {
-        "start": {
-            "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. Start off by greeting them. Be friendly and casual; you're taking an order for food over the phone.",
-                }
-            ],
-            "functions": [
-                {
-                    "type": "function",
-                    "function": {
-                        "name": "choose_pizza",
-                        "description": "User wants to order pizza",
-                        "parameters": {"type": "object", "properties": {}},
-                    },
-                },
-                {
-                    "type": "function",
-                    "function": {
-                        "name": "choose_sushi",
-                        "description": "User wants to order sushi",
-                        "parameters": {"type": "object", "properties": {}},
-                    },
-                },
-            ],
-        },
-        "choose_pizza": {
-            "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.
-                
-                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": [
-                {
-                    "type": "function",
-                    "function": {
-                        "name": "select_pizza_size",
-                        "description": "Record the selected pizza size",
-                        "parameters": {
-                            "type": "object",
-                            "properties": {
-                                "size": {
-                                    "type": "string",
-                                    "enum": ["small", "medium", "large"],
-                                    "description": "Size of the pizza",
-                                }
-                            },
-                            "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..."}
-            ],
-        },
-        "choose_sushi": {
-            "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.
-                
-                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": [
-                {
-                    "type": "function",
-                    "function": {
-                        "name": "select_roll_count",
-                        "description": "Record the number of sushi rolls",
-                        "parameters": {
-                            "type": "object",
-                            "properties": {
-                                "count": {
-                                    "type": "integer",
-                                    "minimum": 1,
-                                    "maximum": 10,
-                                    "description": "Number of rolls to order",
-                                }
-                            },
-                            "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, let me help you with your sushi order..."}
-            ],
-        },
-        "end": {
-            "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"}],
-        },
-    },
-}
-
-
-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")
-
-        # Get initial tools from the first node
-        initial_tools = flow_config["nodes"]["start"]["functions"]
-
-        # Create initial context
-        messages = [
-            {
-                "role": "system",
-                "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, initial_tools)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        pipeline = Pipeline(
-            [
-                transport.input(),  # Transport user input
-                stt,  # STT
-                context_aggregator.user(),  # User responses
-                llm,  # LLM
-                tts,  # TTS
-                transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
-            ]
-        )
-
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
-
-        # Initialize flow manager
-        flow_manager = FlowManager(flow_config, task, tts)
-
-        # Register functions with LLM service
-        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_manager.initialize(messages)
-            # Kick off the conversation using the context aggregator
-            await task.queue_frames([context_aggregator.user().get_context_frame()])
-
-        runner = PipelineRunner()
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/race_bot.py

@@ -0,0 +1,191 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+import time
+
+import aiohttp
+from loguru import logger
+from runner import configure
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import (
+    BotSpeakingFrame,
+    EndFrame,
+    Frame,
+    InputAudioRawFrame,
+    StartInterruptionFrame,
+    StopInterruptionFrame,
+    TextFrame,
+    TranscriptionFrame,
+    TTSAudioRawFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
+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.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+class DebugProcessor(FrameProcessor):
+    def __init__(self, name, **kwargs):
+        self._name = name
+        super().__init__(**kwargs)
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if not (
+            isinstance(frame, InputAudioRawFrame)
+            or isinstance(frame, BotSpeakingFrame)
+            or isinstance(frame, UserStoppedSpeakingFrame)
+            or isinstance(frame, TTSAudioRawFrame)
+            or isinstance(frame, TextFrame)
+        ):
+            logger.debug(f"--- {self._name}: {frame} {direction}")
+        await self.push_frame(frame, direction)
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, _) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            None,
+            "AI Bot",
+            DailyParams(
+                audio_out_enabled=True,
+                transcription_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+            ),
+        )
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        )
+
+        llm = OpenAILLMService(api_key=os.environ["OPENAI_API_KEY"], model="gpt-4o")
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        dp = DebugProcessor("dp")
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        runner = PipelineRunner()
+
+        task = PipelineTask(
+            Pipeline(
+                [
+                    # transport.input(),
+                    context_aggregator.user(),
+                    llm,
+                    dp,
+                    tts,
+                    transport.output(),
+                    context_aggregator.assistant(),
+                ]
+            ),
+            PipelineParams(
+                allow_interruptions=True,
+            ),
+        )
+
+        # Register an event handler so we can play the audio when the
+        # participant joins.
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            participant_id = participant.get("info", {}).get("participantId", "")
+
+            # Create frames for 600 seconds
+            start_time = time.time()
+            while time.time() - start_time < 300:
+                elapsed_time = round(time.time() - start_time)
+                logger.info(f"Running for {elapsed_time} seconds")
+                await task.queue_frame(
+                    StartInterruptionFrame(),
+                )
+                await asyncio.sleep(1)
+
+                await task.queue_frame(
+                    UserStartedSpeakingFrame(),
+                )
+
+                await asyncio.sleep(1)
+
+                await task.queue_frame(
+                    TranscriptionFrame("Tell me more about your company.", participant_id, time.time()),
+                )
+
+                await asyncio.sleep(1)
+
+                await task.queue_frame(
+                    StopInterruptionFrame(),
+                )
+
+                await asyncio.sleep(1)
+
+                await task.queue_frame(
+                    UserStoppedSpeakingFrame(),
+                )
+
+                await asyncio.sleep(5)
+
+                await task.queue_frame(StartInterruptionFrame())
+                await asyncio.sleep(1)
+
+                await task.queue_frame(
+                    UserStartedSpeakingFrame(),
+                )
+
+                await asyncio.sleep(1)
+
+                await task.queue_frame(
+                    TranscriptionFrame("Give me a list of appointment dates.", participant_id, time.time()),
+                )
+
+                await asyncio.sleep(1)
+
+                await task.queue_frames(
+                    StopInterruptionFrame(),
+                )
+
+                await asyncio.sleep(1)
+                await task.queue_frame(
+                    UserStoppedSpeakingFrame(),
+                )
+                await asyncio.sleep(5)
+            await task.queue_frame(EndFrame())
+
+        # @transport.event_handler("on_first_participant_joined")
+        # async def on_first_participant_joined(transport, participant):
+        #     await transport.capture_participant_transcription(participant["id"])
+        #     # Kick off the conversation.
+        #     messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        #     await task.queue_frames([LLMMessagesFrame(messages)])
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -51,7 +51,6 @@ 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.3.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" ]

src/pipecat/services/cartesia.py

@@ -7,6 +7,7 @@
 import asyncio
 import base64
 import json
+import random
 import uuid
 from typing import AsyncGenerator, List, Optional, Union
 
@@ -222,6 +223,10 @@ class CartesiaTTSService(WordTTSService):
     async def _receive_task_handler(self):
         try:
             async for message in self._get_websocket():
+                # Randomly cancel the asyncio task 1% of the time
+                if random.random() < 0.01:
+                    logger.info(f"Cancelling task for {self} due to random chance")
+                    asyncio.current_task().cancel()
                 msg = json.loads(message)
                 if not msg or msg["context_id"] != self._context_id:
                     continue
@@ -256,6 +261,7 @@ class CartesiaTTSService(WordTTSService):
                     logger.error(f"Cartesia error, unknown message type: {msg}")
         except asyncio.CancelledError:
             pass
+            # await self.push_error(ErrorFrame(f"{self} cancelled", True))
         except Exception as e:
             logger.error(f"{self} exception: {e}")
 

src/pipecat/transports/base_input.py

@@ -71,6 +71,7 @@ class BaseInputTransport(FrameProcessor):
         return self._params.vad_analyzer
 
     async def push_audio_frame(self, frame: InputAudioRawFrame):
+        logger.info(f"Pushing audio qsize: {self._audio_in_queue.qsize()}")
         if self._params.audio_in_enabled or self._params.vad_enabled:
             await self._audio_in_queue.put(frame)
 
@@ -167,6 +168,7 @@ class BaseInputTransport(FrameProcessor):
         return vad_state
 
     async def _audio_task_handler(self):
+        logger.info("_audio_task_handler started")
         vad_state: VADState = VADState.QUIET
         while True:
             try:

src/pipecat/transports/network/websocket_server.py

@@ -106,6 +106,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
                 continue
 
             if isinstance(frame, AudioRawFrame):
+                logger.info("websocket_server")
                 await self.push_audio_frame(
                     InputAudioRawFrame(
                         audio=frame.audio,