Merge pull request #211 from pipecat-ai/aleix/base-transport-async

various fixes and improvements

CHANGELOG.md

@@ -7,12 +7,44 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Allow passing `output_format` and `model_id` to `CartesiaTTSService` to change
+  audio sample format and the model to use.
+
+- Added `DailyRESTHelper` which helps you create Daily rooms and tokens in an
+  easy way.
+
+- `PipelineTask` now has a `has_finished()` method to indicate if the task has
+  completed. If a task is never ran `has_finished()` will return False.
+
+- `PipelineRunner` now supports SIGTERM. If received, the runner will be
+  canceled.
+
 ### Fixed
 
+- Fixed an issue where `BaseInputTransport` and `BaseOutputTransport` where
+  stopping push tasks before pushing `EndFrame` frames.
+
+- Fixed an error closing local audio transports.
+
 - Fixed an issue with Deepgram TTS that was introduced in the previous release.
 
+- Fixed `AnthropicLLMService` interruptions. If an interruption occurred, a
+  `user` message could be appended after the previous `user` message. Anthropic
+  does not allow that because it requires alternate `user` and `assistant`
+  messages.
+
 ### Performance
 
+- The `BaseInputTransport` does not pull audio frames from sub-classes any
+  more. Instead, sub-classes now push audio frames into a queue in the base
+  class. Also, `DailyInputTransport` now pushes audio frames every 20ms instead
+  of 10ms.
+
+- Remove redundant camera input thread from `DailyInputTransport`. This should
+  improve performance a little bit when processing participant videos.
+
 - Load Cartesia voice on startup.
 
 ## [0.0.25] - 2024-05-31

examples/foundational/05a-local-sync-speech-and-image.py

@@ -156,7 +156,7 @@ async def main():
             await runner.stop_when_done()
 
         async def run_tk():
-            while True:
+            while not task.has_finished():
                 tk_root.update()
                 tk_root.update_idletasks()
                 await asyncio.sleep(0.1)

examples/foundational/07b-interruptible-langchain.py

@@ -4,15 +4,11 @@
 # 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.frames.frames import LLMMessagesFrame
 from pipecat.pipeline.pipeline import Pipeline
@@ -25,20 +21,19 @@ from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
 
+from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_community.chat_message_histories import ChatMessageHistory
+from langchain_core.chat_history import BaseChatMessageHistory
+from langchain_core.runnables.history import RunnableWithMessageHistory
+from langchain_openai import ChatOpenAI
+
+from loguru import logger
+
+from runner import configure
+
+from dotenv import load_dotenv
 load_dotenv(override=True)
 
-try:
-    from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
-    from langchain_community.chat_message_histories import ChatMessageHistory
-    from langchain_core.chat_history import BaseChatMessageHistory
-    from langchain_core.runnables.history import RunnableWithMessageHistory
-    from langchain_openai import ChatOpenAI
-
-except ModuleNotFoundError as e:
-    logger.exception(
-        "In order to run this example you need to `pip install pipecat-ai[langchain] langchain-community langchain-openai. Also, be sure to set `OPENAI_API_KEY` in the environment variable."
-    )
-    raise Exception(f"Missing module: {e}")
 
 logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")

examples/foundational/07d-interruptible-cartesia.py

@@ -39,6 +39,7 @@ async def main(room_url: str, token):
             "Respond bot",
             DailyParams(
                 audio_out_enabled=True,
+                audio_out_sample_rate=44100,
                 transcription_enabled=True,
                 vad_enabled=True,
                 vad_analyzer=SileroVADAnalyzer()
@@ -47,7 +48,8 @@ async def main(room_url: str, token):
 
         tts = CartesiaTTSService(
             api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_name="Barbershop Man"
+            voice_name="British Lady",
+            output_format="pcm_44100"
         )
 
         llm = OpenAILLMService(

examples/foundational/09-mirror.py

@@ -30,6 +30,7 @@ async def main(room_url, token):
             audio_in_enabled=True,
             audio_out_enabled=True,
             camera_out_enabled=True,
+            camera_out_is_live=True,
             camera_out_width=1280,
             camera_out_height=720
         )

examples/foundational/09a-local-mirror.py

@@ -38,6 +38,7 @@ async def main(room_url, token):
         TransportParams(
             audio_out_enabled=True,
             camera_out_enabled=True,
+            camera_out_is_live=True,
             camera_out_width=1280,
             camera_out_height=720))
 
@@ -47,15 +48,15 @@ async def main(room_url, token):
 
     pipeline = Pipeline([daily_transport.input(), tk_transport.output()])
 
-    runner = PipelineRunner()
+    task = PipelineTask(pipeline)
 
     async def run_tk():
-        while runner.is_active():
+        while not task.has_finished():
             tk_root.update()
             tk_root.update_idletasks()
             await asyncio.sleep(0.1)
 
-    task = PipelineTask(pipeline)
+    runner = PipelineRunner()
 
     await asyncio.gather(runner.run(task), run_tk())
 

examples/foundational/13a-whisper-local.py

@@ -16,8 +16,6 @@ from pipecat.services.whisper import WhisperSTTService
 from pipecat.transports.base_transport import TransportParams
 from pipecat.transports.local.audio import LocalAudioTransport
 
-from runner import configure
-
 from loguru import logger
 
 from dotenv import load_dotenv
@@ -34,7 +32,7 @@ class TranscriptionLogger(FrameProcessor):
             print(f"Transcription: {frame.text}")
 
 
-async def main(room_url: str):
+async def main():
     transport = LocalAudioTransport(TransportParams(audio_in_enabled=True))
 
     stt = WhisperSTTService()
@@ -51,5 +49,4 @@ async def main(room_url: str):
 
 
 if __name__ == "__main__":
-    (url, token) = configure()
+    asyncio.run(main())
-    asyncio.run(main(url))

linux-py3.10-requirements.txt

@@ -5,7 +5,11 @@
 #    pip-compile --all-extras pyproject.toml
 #
 aiohttp==3.9.5
-    # via pipecat-ai (pyproject.toml)
+    # via
+    #   cartesia
+    #   langchain
+    #   langchain-community
+    #   pipecat-ai (pyproject.toml)
 aiosignal==1.3.1
     # via aiohttp
 annotated-types==0.7.0
@@ -18,10 +22,12 @@ anyio==4.4.0
     #   httpx
     #   openai
 async-timeout==4.0.3
-    # via aiohttp
+    # via
+    #   aiohttp
+    #   langchain
 attrs==23.2.0
     # via aiohttp
-av==12.0.0
+av==12.1.0
     # via faster-whisper
 azure-cognitiveservices-speech==1.37.0
     # via pipecat-ai (pyproject.toml)
@@ -29,11 +35,15 @@ blinker==1.8.2
     # via flask
 cachetools==5.3.3
     # via google-auth
-certifi==2024.2.2
+cartesia==0.1.1
+    # via pipecat-ai (pyproject.toml)
+certifi==2024.6.2
     # via
     #   httpcore
     #   httpx
     #   requests
+cffi==1.16.0
+    # via sounddevice
 charset-normalizer==3.3.2
     # via requests
 click==8.1.7
@@ -44,6 +54,8 @@ ctranslate2==4.2.1
     # via faster-whisper
 daily-python==0.9.1
     # via pipecat-ai (pyproject.toml)
+dataclasses-json==0.6.6
+    # via langchain-community
 distro==1.9.0
     # via
     #   anthropic
@@ -51,7 +63,9 @@ distro==1.9.0
 einops==0.8.0
     # via pipecat-ai (pyproject.toml)
 exceptiongroup==1.2.1
-    # via anyio
+    # via
+    #   anyio
+    #   pytest
 fal-client==0.4.0
     # via pipecat-ai (pyproject.toml)
 faster-whisper==1.0.2
@@ -75,7 +89,7 @@ frozenlist==1.4.1
     # via
     #   aiohttp
     #   aiosignal
-fsspec==2024.5.0
+fsspec==2024.6.0
     # via
     #   huggingface-hub
     #   torch
@@ -88,7 +102,7 @@ google-api-core[grpc]==2.19.0
     #   google-ai-generativelanguage
     #   google-api-python-client
     #   google-generativeai
-google-api-python-client==2.131.0
+google-api-python-client==2.132.0
     # via google-generativeai
 google-auth==2.29.0
     # via
@@ -101,11 +115,13 @@ google-auth-httplib2==0.2.0
     # via google-api-python-client
 google-generativeai==0.5.4
     # via pipecat-ai (pyproject.toml)
-googleapis-common-protos==1.63.0
+googleapis-common-protos==1.63.1
     # via
     #   google-api-core
     #   grpcio-status
-grpcio==1.64.0
+greenlet==3.0.3
+    # via sqlalchemy
+grpcio==1.64.1
     # via
     #   google-api-core
     #   grpcio-status
@@ -123,6 +139,7 @@ httplib2==0.22.0
 httpx==0.27.0
     # via
     #   anthropic
+    #   cartesia
     #   fal-client
     #   openai
 httpx-sse==0.4.0
@@ -141,29 +158,62 @@ idna==3.7
     #   httpx
     #   requests
     #   yarl
+iniconfig==2.0.0
+    # via pytest
 itsdangerous==2.2.0
     # via flask
 jinja2==3.1.4
     # via
     #   flask
     #   torch
+jsonpatch==1.33
+    # via langchain-core
+jsonpointer==2.4
+    # via jsonpatch
+langchain==0.2.1
+    # via
+    #   langchain-community
+    #   pipecat-ai (pyproject.toml)
+langchain-community==0.2.1
+    # via pipecat-ai (pyproject.toml)
+langchain-core==0.2.3
+    # via
+    #   langchain
+    #   langchain-community
+    #   langchain-openai
+    #   langchain-text-splitters
+langchain-openai==0.1.8
+    # via pipecat-ai (pyproject.toml)
+langchain-text-splitters==0.2.0
+    # via langchain
+langsmith==0.1.69
+    # via
+    #   langchain
+    #   langchain-community
+    #   langchain-core
 loguru==0.7.2
     # via pipecat-ai (pyproject.toml)
 markupsafe==2.1.5
     # via
     #   jinja2
     #   werkzeug
+marshmallow==3.21.2
+    # via dataclasses-json
 mpmath==1.3.0
     # via sympy
 multidict==6.0.5
     # via
     #   aiohttp
     #   yarl
+mypy-extensions==1.0.0
+    # via typing-inspect
 networkx==3.3
     # via torch
 numpy==1.26.4
     # via
     #   ctranslate2
+    #   langchain
+    #   langchain-community
     #   onnxruntime
     #   pipecat-ai (pyproject.toml)
     #   pyloudnorm
@@ -204,16 +254,25 @@ nvidia-nvtx-cu12==12.1.105
 onnxruntime==1.18.0
     # via faster-whisper
 openai==1.26.0
-    # via pipecat-ai (pyproject.toml)
+    # via
-packaging==24.0
+    #   langchain-openai
+    #   pipecat-ai (pyproject.toml)
+orjson==3.10.3
+    # via langsmith
+packaging==23.2
     # via
     #   huggingface-hub
+    #   langchain-core
+    #   marshmallow
     #   onnxruntime
+    #   pytest
     #   transformers
 pillow==10.3.0
     # via
     #   pipecat-ai (pyproject.toml)
     #   torchvision
+pluggy==1.5.0
+    # via pytest
 proto-plus==1.23.0
     # via
     #   google-ai-generativelanguage
@@ -237,12 +296,17 @@ pyasn1-modules==0.4.0
     # via google-auth
 pyaudio==0.2.14
     # via pipecat-ai (pyproject.toml)
-pydantic==2.7.2
+pycparser==2.22
+    # via cffi
+pydantic==2.7.3
     # via
     #   anthropic
     #   google-generativeai
+    #   langchain
+    #   langchain-core
+    #   langsmith
     #   openai
-pydantic-core==2.18.3
+pydantic-core==2.18.4
     # via pydantic
 pyht==0.0.28
     # via pipecat-ai (pyproject.toml)
@@ -250,21 +314,35 @@ pyloudnorm==0.1.1
     # via pipecat-ai (pyproject.toml)
 pyparsing==3.1.2
     # via httplib2
+pytest==8.2.2
+    # via pytest-asyncio
+pytest-asyncio==0.23.7
+    # via cartesia
 python-dotenv==1.0.1
     # via pipecat-ai (pyproject.toml)
 pyyaml==6.0.1
     # via
     #   ctranslate2
     #   huggingface-hub
+    #   langchain
+    #   langchain-community
+    #   langchain-core
     #   timm
     #   transformers
 regex==2024.5.15
-    # via transformers
+    # via
+    #   tiktoken
+    #   transformers
 requests==2.32.3
     # via
+    #   cartesia
     #   google-api-core
     #   huggingface-hub
+    #   langchain
+    #   langchain-community
+    #   langsmith
     #   pyht
+    #   tiktoken
     #   transformers
 rsa==4.9
     # via google-auth
@@ -280,10 +358,23 @@ sniffio==1.3.1
     #   anyio
     #   httpx
     #   openai
-sympy==1.12
+sounddevice==0.4.7
+    # via pipecat-ai (pyproject.toml)
+sqlalchemy==2.0.30
+    # via
+    #   langchain
+    #   langchain-community
+sympy==1.12.1
     # via
     #   onnxruntime
     #   torch
+tenacity==8.3.0
+    # via
+    #   langchain
+    #   langchain-community
+    #   langchain-core
+tiktoken==0.7.0
+    # via langchain-openai
 timm==0.9.16
     # via pipecat-ai (pyproject.toml)
 tokenizers==0.19.1
@@ -291,6 +382,8 @@ tokenizers==0.19.1
     #   anthropic
     #   faster-whisper
     #   transformers
+tomli==2.0.1
+    # via pytest
 torch==2.3.0
     # via
     #   pipecat-ai (pyproject.toml)
@@ -311,7 +404,7 @@ transformers==4.40.2
     # via pipecat-ai (pyproject.toml)
 triton==2.3.0
     # via torch
-typing-extensions==4.11.0
+typing-extensions==4.12.1
     # via
     #   anthropic
     #   anyio
@@ -321,13 +414,19 @@ typing-extensions==4.11.0
     #   pipecat-ai (pyproject.toml)
     #   pydantic
     #   pydantic-core
+    #   sqlalchemy
     #   torch
+    #   typing-inspect
+typing-inspect==0.9.0
+    # via dataclasses-json
 uritemplate==4.1.1
     # via google-api-python-client
 urllib3==2.2.1
     # via requests
 websockets==12.0
-    # via pipecat-ai (pyproject.toml)
+    # via
+    #   cartesia
+    #   pipecat-ai (pyproject.toml)
 werkzeug==3.0.3
     # via flask
 yarl==1.9.4

macos-py3.10-requirements.txt

@@ -7,6 +7,8 @@
 aiohttp==3.9.5
     # via
     #   cartesia
+    #   langchain
+    #   langchain-community
     #   pipecat-ai (pyproject.toml)
 aiosignal==1.3.1
     # via aiohttp
@@ -20,7 +22,9 @@ anyio==4.4.0
     #   httpx
     #   openai
 async-timeout==4.0.3
-    # via aiohttp
+    # via
+    #   aiohttp
+    #   langchain
 attrs==23.2.0
     # via aiohttp
 av==12.1.0
@@ -31,9 +35,9 @@ blinker==1.8.2
     # via flask
 cachetools==5.3.3
     # via google-auth
-cartesia==0.1.0
+cartesia==0.1.1
     # via pipecat-ai (pyproject.toml)
-certifi==2024.2.2
+certifi==2024.6.2
     # via
     #   httpcore
     #   httpx
@@ -50,6 +54,8 @@ ctranslate2==4.2.1
     # via faster-whisper
 daily-python==0.9.1
     # via pipecat-ai (pyproject.toml)
+dataclasses-json==0.6.6
+    # via langchain-community
 distro==1.9.0
     # via
     #   anthropic
@@ -82,7 +88,7 @@ frozenlist==1.4.1
     # via
     #   aiohttp
     #   aiosignal
-fsspec==2024.5.0
+fsspec==2024.6.0
     # via
     #   huggingface-hub
     #   torch
@@ -95,7 +101,7 @@ google-api-core[grpc]==2.19.0
     #   google-ai-generativelanguage
     #   google-api-python-client
     #   google-generativeai
-google-api-python-client==2.131.0
+google-api-python-client==2.132.0
     # via google-generativeai
 google-auth==2.29.0
     # via
@@ -108,11 +114,11 @@ google-auth-httplib2==0.2.0
     # via google-api-python-client
 google-generativeai==0.5.4
     # via pipecat-ai (pyproject.toml)
-googleapis-common-protos==1.63.0
+googleapis-common-protos==1.63.1
     # via
     #   google-api-core
     #   grpcio-status
-grpcio==1.64.0
+grpcio==1.64.1
     # via
     #   google-api-core
     #   grpcio-status
@@ -157,23 +163,54 @@ jinja2==3.1.4
     # via
     #   flask
     #   torch
+jsonpatch==1.33
+    # via langchain-core
+jsonpointer==2.4
+    # via jsonpatch
+langchain==0.2.2
+    # via
+    #   langchain-community
+    #   pipecat-ai (pyproject.toml)
+langchain-community==0.2.2
+    # via pipecat-ai (pyproject.toml)
+langchain-core==0.2.4
+    # via
+    #   langchain
+    #   langchain-community
+    #   langchain-openai
+    #   langchain-text-splitters
+langchain-openai==0.1.8
+    # via pipecat-ai (pyproject.toml)
+langchain-text-splitters==0.2.1
+    # via langchain
+langsmith==0.1.69
+    # via
+    #   langchain
+    #   langchain-community
+    #   langchain-core
 loguru==0.7.2
     # via pipecat-ai (pyproject.toml)
 markupsafe==2.1.5
     # via
     #   jinja2
     #   werkzeug
+marshmallow==3.21.2
+    # via dataclasses-json
 mpmath==1.3.0
     # via sympy
 multidict==6.0.5
     # via
     #   aiohttp
     #   yarl
+mypy-extensions==1.0.0
+    # via typing-inspect
 networkx==3.3
     # via torch
 numpy==1.26.4
     # via
     #   ctranslate2
+    #   langchain
+    #   langchain-community
     #   onnxruntime
     #   pipecat-ai (pyproject.toml)
     #   pyloudnorm
@@ -183,10 +220,16 @@ numpy==1.26.4
 onnxruntime==1.18.0
     # via faster-whisper
 openai==1.26.0
-    # via pipecat-ai (pyproject.toml)
+    # via
-packaging==24.0
+    #   langchain-openai
+    #   pipecat-ai (pyproject.toml)
+orjson==3.10.3
+    # via langsmith
+packaging==23.2
     # via
     #   huggingface-hub
+    #   langchain-core
+    #   marshmallow
     #   onnxruntime
     #   pytest
     #   transformers
@@ -221,12 +264,15 @@ pyaudio==0.2.14
     # via pipecat-ai (pyproject.toml)
 pycparser==2.22
     # via cffi
-pydantic==2.7.2
+pydantic==2.7.3
     # via
     #   anthropic
     #   google-generativeai
+    #   langchain
+    #   langchain-core
+    #   langsmith
     #   openai
-pydantic-core==2.18.3
+pydantic-core==2.18.4
     # via pydantic
 pyht==0.0.28
     # via pipecat-ai (pyproject.toml)
@@ -234,7 +280,7 @@ pyloudnorm==0.1.1
     # via pipecat-ai (pyproject.toml)
 pyparsing==3.1.2
     # via httplib2
-pytest==8.2.1
+pytest==8.2.2
     # via pytest-asyncio
 pytest-asyncio==0.23.7
     # via cartesia
@@ -244,16 +290,25 @@ pyyaml==6.0.1
     # via
     #   ctranslate2
     #   huggingface-hub
+    #   langchain
+    #   langchain-community
+    #   langchain-core
     #   timm
     #   transformers
 regex==2024.5.15
-    # via transformers
+    # via
+    #   tiktoken
+    #   transformers
 requests==2.32.3
     # via
     #   cartesia
     #   google-api-core
     #   huggingface-hub
+    #   langchain
+    #   langchain-community
+    #   langsmith
     #   pyht
+    #   tiktoken
     #   transformers
 rsa==4.9
     # via google-auth
@@ -271,10 +326,21 @@ sniffio==1.3.1
     #   openai
 sounddevice==0.4.7
     # via pipecat-ai (pyproject.toml)
+sqlalchemy==2.0.30
+    # via
+    #   langchain
+    #   langchain-community
 sympy==1.12.1
     # via
     #   onnxruntime
     #   torch
+tenacity==8.3.0
+    # via
+    #   langchain
+    #   langchain-community
+    #   langchain-core
+tiktoken==0.7.0
+    # via langchain-openai
 timm==0.9.16
     # via pipecat-ai (pyproject.toml)
 tokenizers==0.19.1
@@ -302,7 +368,7 @@ tqdm==4.66.4
     #   transformers
 transformers==4.40.2
     # via pipecat-ai (pyproject.toml)
-typing-extensions==4.11.0
+typing-extensions==4.12.1
     # via
     #   anthropic
     #   anyio
@@ -312,7 +378,11 @@ typing-extensions==4.11.0
     #   pipecat-ai (pyproject.toml)
     #   pydantic
     #   pydantic-core
+    #   sqlalchemy
     #   torch
+    #   typing-inspect
+typing-inspect==0.9.0
+    # via dataclasses-json
 uritemplate==4.1.1
     # via google-api-python-client
 urllib3==2.2.1

pyproject.toml

@@ -26,7 +26,7 @@ dependencies = [
     "Pillow~=10.3.0",
     "protobuf~=4.25.3",
     "pyloudnorm~=0.1.1",
-    "typing-extensions~=4.11.0",
+    "typing-extensions~=4.12.1",
 ]
 
 [project.urls]
@@ -42,7 +42,7 @@ examples = [ "python-dotenv~=1.0.0", "flask~=3.0.3", "flask_cors~=4.0.1" ]
 fal = [ "fal-client~=0.4.0" ]
 google = [ "google-generativeai~=0.5.3" ]
 fireworks = [ "openai~=1.26.0" ]
-langchain = [ "langchain~=0.2.1" ]
+langchain = [ "langchain~=0.2.1", "langchain-community~=0.2.1", "langchain-openai~=0.1.8" ]
 local = [ "pyaudio~=0.2.0" ]
 moondream = [ "einops~=0.8.0", "timm~=0.9.16", "transformers~=4.40.2" ]
 openai = [ "openai~=1.26.0" ]

src/pipecat/pipeline/runner.py

@@ -20,18 +20,15 @@ class PipelineRunner:
         self.name: str = name or f"{self.__class__.__name__}#{obj_count(self)}"
 
         self._tasks = {}
-        self._running = True
 
         if handle_sigint:
             self._setup_sigint()
 
     async def run(self, task: PipelineTask):
         logger.debug(f"Runner {self} started running {task}")
-        self._running = True
         self._tasks[task.name] = task
         await task.run()
         del self._tasks[task.name]
-        self._running = False
         logger.debug(f"Runner {self} finished running {task}")
 
     async def stop_when_done(self):
@@ -42,18 +39,19 @@ class PipelineRunner:
         logger.debug(f"Canceling runner {self}")
         await asyncio.gather(*[t.cancel() for t in self._tasks.values()])
 
-    def is_active(self):
-        return self._running
-
     def _setup_sigint(self):
         loop = asyncio.get_running_loop()
         loop.add_signal_handler(
             signal.SIGINT,
-            lambda *args: asyncio.create_task(self._sigint_handler())
+            lambda *args: asyncio.create_task(self._sig_handler())
+        )
+        loop.add_signal_handler(
+            signal.SIGTERM,
+            lambda *args: asyncio.create_task(self._sig_handler())
         )
 
-    async def _sigint_handler(self):
+    async def _sig_handler(self):
-        logger.warning(f"Ctrl-C detected. Canceling runner {self}")
+        logger.warning(f"Interruption detected. Canceling runner {self}")
         await self.cancel()
 
     def __str__(self):

src/pipecat/pipeline/task.py

@@ -43,6 +43,7 @@ class PipelineTask:
 
         self._pipeline = pipeline
         self._params = params
+        self._finished = False
 
         self._down_queue = asyncio.Queue()
         self._up_queue = asyncio.Queue()
@@ -50,6 +51,9 @@ class PipelineTask:
         self._source = Source(self._up_queue)
         self._source.link(pipeline)
 
+    def has_finished(self):
+        return self._finished
+
     async def stop_when_done(self):
         logger.debug(f"Task {self} scheduled to stop when done")
         await self.queue_frame(EndFrame())
@@ -67,6 +71,7 @@ class PipelineTask:
         self._process_up_task = asyncio.create_task(self._process_up_queue())
         self._process_down_task = asyncio.create_task(self._process_down_queue())
         await asyncio.gather(self._process_up_task, self._process_down_task)
+        self._finished = True
 
     async def queue_frame(self, frame: Frame):
         await self._down_queue.put(frame)

src/pipecat/services/anthropic.py

@@ -4,8 +4,6 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import os
-import asyncio
 import time
 import base64
 
@@ -80,8 +78,20 @@ class AnthropicLLMService(LLMService):
                     }]
                 })
             else:
-                # text frame
+                # Text frame. Anthropic needs the roles to alternate. This will
-                anthropic_messages.append({"role": role, "content": content})
+                # cause an issue with interruptions. So, if we detect we are the
+                # ones asking again it probably means we were interrupted.
+                if role == "user" and len(anthropic_messages) > 1:
+                    last_message = anthropic_messages[-1]
+                    if last_message["role"] == "user":
+                        anthropic_messages = anthropic_messages[:-1]
+                        content = last_message["content"]
+                        anthropic_messages.append(
+                            {"role": "user", "content": f"Sorry, I just asked you about [{content}] but now I would like to know [{text}]."})
+                    else:
+                        anthropic_messages.append({"role": role, "content": text})
+                else:
+                    anthropic_messages.append({"role": role, "content": text})
 
         return anthropic_messages
 
@@ -107,7 +117,7 @@ class AnthropicLLMService(LLMService):
                     await self.push_frame(LLMResponseEndFrame())
 
         except Exception as e:
-            logger.error(f"Exception: {e}")
+            logger.error(f"Anthrophic exception: {e}")
         finally:
             await self.push_frame(LLMFullResponseEndFrame())
 
@@ -125,22 +135,3 @@ class AnthropicLLMService(LLMService):
 
         if context:
             await self._process_context(context)
-
-    async def x_process_frame(self, frame: Frame, direction: FrameDirection):
-        if isinstance(frame, LLMMessagesFrame):
-            stream = await self.client.messages.create(
-                max_tokens=self.max_tokens,
-                messages=[
-                    {
-                        "role": "user",
-                        "content": "Hello, Claude",
-                    }
-                ],
-                model=self.model,
-                stream=True,
-            )
-            async for event in stream:
-                if event.type == "content_block_delta":
-                    await self.push_frame(TextFrame(event.delta.text))
-        else:
-            await self.push_frame(frame, direction)

src/pipecat/services/cartesia.py

@@ -21,11 +21,15 @@ class CartesiaTTSService(TTSService):
             *,
             api_key: str,
             voice_name: str,
+            model_id: str = "upbeat-moon",
+            output_format: str = "pcm_16000",
             **kwargs):
         super().__init__(**kwargs)
 
         self._api_key = api_key
         self._voice_name = voice_name
+        self._model_id = model_id
+        self._output_format = output_format
 
         try:
             self._client = AsyncCartesiaTTS(api_key=self._api_key)
@@ -36,18 +40,18 @@ class CartesiaTTSService(TTSService):
             logger.error(f"Cartesia initialization error: {e}")
 
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
-        logger.debug(f"Transcribing text: [{text}]")
+        logger.debug(f"Generating TTS: [{text}]")
 
         try:
             chunk_generator = await self._client.generate(
-                transcript=text, voice=self._voice, stream=True,
+                stream=True,
-                model_id="upbeat-moon", data_rtype='array', output_format='pcm_16000',
+                transcript=text,
-                # a chunk_time of 0.1 seems to be the default. there are small audio pops/gaps which
+                voice=self._voice,
-                # we need to debug
+                model_id=self._model_id,
-                chunk_time=0.1
+                output_format=self._output_format,
             )
 
             async for chunk in chunk_generator:
-                yield AudioRawFrame(chunk['audio'], 16000, 1)
+                yield AudioRawFrame(chunk["audio"], chunk["sampling_rate"], 1)
         except Exception as e:
-            logger.error(f"Cartesia error: {e}")
+            logger.error(f"Cartesia exception: {e}")

src/pipecat/services/deepgram.py

@@ -30,7 +30,8 @@ class DeepgramTTSService(TTSService):
         self._aiohttp_session = aiohttp_session
 
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
-        logger.info(f"Running Deepgram TTS for {text}")
+        logger.debug(f"Generating TTS: [{text}]")
+
         base_url = "https://api.deepgram.com/v1/speak"
         request_url = f"{base_url}?model={self._voice}&encoding=linear16&container=none&sample_rate=16000"
         headers = {"authorization": f"token {self._api_key}"}
@@ -48,4 +49,4 @@ class DeepgramTTSService(TTSService):
                     frame = AudioRawFrame(audio=data, sample_rate=16000, num_channels=1)
                     yield frame
         except Exception as e:
-            logger.error(f"Exception {e}")
+            logger.error(f"Deepgram exception: {e}")

src/pipecat/transports/base_input.py

@@ -36,11 +36,7 @@ class BaseInputTransport(FrameProcessor):
         self._running = False
         self._allow_interruptions = False
 
-        self._in_executor = ThreadPoolExecutor(max_workers=5)
+        self._executor = ThreadPoolExecutor(max_workers=5)
-
-        # Create audio input queue if needed.
-        if self._params.audio_in_enabled or self._params.vad_enabled:
-            self._audio_in_queue = queue.Queue()
 
         # Create push frame task. This is the task that will push frames in
         # order. We also guarantee that all frames are pushed in the same task.
@@ -57,9 +53,11 @@ class BaseInputTransport(FrameProcessor):
 
         self._running = True
 
+        # Create audio input queue and thread if needed.
         if self._params.audio_in_enabled or self._params.vad_enabled:
-            loop = self.get_event_loop()
+            self._audio_in_queue = queue.Queue()
-            self._audio_thread = loop.run_in_executor(self._in_executor, self._audio_thread_handler)
+            self._audio_thread = self._loop.run_in_executor(
+                self._executor, self._audio_thread_handler)
 
     async def stop(self):
         if not self._running:
@@ -77,8 +75,8 @@ class BaseInputTransport(FrameProcessor):
     def vad_analyzer(self) -> VADAnalyzer | None:
         return self._params.vad_analyzer
 
-    def read_next_audio_frame(self) -> AudioRawFrame | None:
+    def push_audio_frame(self, frame: AudioRawFrame):
-        pass
+        self._audio_in_queue.put_nowait(frame)
 
     #
     # Frame processor
@@ -89,16 +87,16 @@ class BaseInputTransport(FrameProcessor):
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, CancelFrame):
-            await self.stop()
             # We don't queue a CancelFrame since we want to stop ASAP.
             await self.push_frame(frame, direction)
+            await self.stop()
         elif isinstance(frame, StartFrame):
             self._allow_interruption = frame.allow_interruptions
             await self.start(frame)
             await self._internal_push_frame(frame, direction)
         elif isinstance(frame, EndFrame):
-            await self.stop()
             await self._internal_push_frame(frame, direction)
+            await self.stop()
         else:
             await self._internal_push_frame(frame, direction)
 
@@ -174,21 +172,22 @@ class BaseInputTransport(FrameProcessor):
         vad_state: VADState = VADState.QUIET
         while self._running:
             try:
-                frame = self.read_next_audio_frame()
+                frame: AudioRawFrame = self._audio_in_queue.get(timeout=1)
 
-                if frame:
+                audio_passthrough = True
-                    audio_passthrough = True
 
-                    # Check VAD and push event if necessary. We just care about
+                # Check VAD and push event if necessary. We just care about
-                    # changes from QUIET to SPEAKING and vice versa.
+                # changes from QUIET to SPEAKING and vice versa.
-                    if self._params.vad_enabled:
+                if self._params.vad_enabled:
-                        vad_state = self._handle_vad(frame.audio, vad_state)
+                    vad_state = self._handle_vad(frame.audio, vad_state)
-                        audio_passthrough = self._params.vad_audio_passthrough
+                    audio_passthrough = self._params.vad_audio_passthrough
 
                     # Push audio downstream if passthrough.
-                    if audio_passthrough:
+                if audio_passthrough:
-                        future = asyncio.run_coroutine_threadsafe(
+                    future = asyncio.run_coroutine_threadsafe(
-                            self._internal_push_frame(frame), self.get_event_loop())
+                        self._internal_push_frame(frame), self._loop)
-                        future.result()
+                    future.result()
+            except queue.Empty:
+                pass
             except BaseException as e:
                 logger.error(f"Error reading audio frames: {e}")

src/pipecat/transports/base_output.py

@@ -43,7 +43,7 @@ class BaseOutputTransport(FrameProcessor):
         self._running = False
         self._allow_interruptions = False
 
-        self._out_executor = ThreadPoolExecutor(max_workers=5)
+        self._executor = ThreadPoolExecutor(max_workers=5)
 
         # These are the images that we should send to the camera at our desired
         # framerate.
@@ -57,6 +57,10 @@ class BaseOutputTransport(FrameProcessor):
         self._stopped_event = asyncio.Event()
         self._is_interrupted = threading.Event()
 
+        # Create push frame task. This is the task that will push frames in
+        # order. We also guarantee that all frames are pushed in the same task.
+        self._create_push_task()
+
     async def start(self, frame: StartFrame):
         # Make sure we have the latest params. Note that this transport might
         # have been started on another task that might not need interruptions,
@@ -70,15 +74,12 @@ class BaseOutputTransport(FrameProcessor):
 
         loop = self.get_event_loop()
 
+        # Create queues and threads.
         if self._params.camera_out_enabled:
             self._camera_out_thread = loop.run_in_executor(
-                self._out_executor, self._camera_out_thread_handler)
+                self._executor, self._camera_out_thread_handler)
 
-        self._sink_thread = loop.run_in_executor(self._out_executor, self._sink_thread_handler)
+        self._sink_thread = loop.run_in_executor(self._executor, self._sink_thread_handler)
-
-        # Create push frame task. This is the task that will push frames in
-        # order. We also guarantee that all frames are pushed in the same task.
-        self._create_push_task()
 
     async def stop(self):
         if not self._running:
@@ -117,16 +118,16 @@ class BaseOutputTransport(FrameProcessor):
         #
         if isinstance(frame, StartFrame):
             await self.start(frame)
-            self._sink_queue.put(frame)
+            self._sink_queue.put_nowait(frame)
         # EndFrame is managed in the queue handler.
         elif isinstance(frame, CancelFrame):
+            await self.push_frame(frame, direction)
             await self.stop()
-            await self.push_frame(frame, direction)
         elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame):
-            await self._handle_interruptions(frame)
             await self.push_frame(frame, direction)
+            await self._handle_interruptions(frame)
         else:
-            self._sink_queue.put(frame)
+            self._sink_queue.put_nowait(frame)
 
         # If we are finishing, wait here until we have stopped, otherwise we might
         # close things too early upstream. We need this event because we don't
@@ -233,7 +234,7 @@ class BaseOutputTransport(FrameProcessor):
 
     def _set_camera_image(self, image: ImageRawFrame):
         if self._params.camera_out_is_live:
-            self._camera_out_queue.put(image)
+            self._camera_out_queue.put_nowait(image)
         else:
             self._camera_images = itertools.cycle([image])
 

src/pipecat/transports/local/audio.py

@@ -28,22 +28,17 @@ class LocalAudioInputTransport(BaseInputTransport):
     def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
         super().__init__(params)
 
+        sample_rate = self._params.audio_in_sample_rate
+        num_frames = int(sample_rate / 100) * 2  # 20ms of audio
+
         self._in_stream = py_audio.open(
             format=py_audio.get_format_from_width(2),
             channels=params.audio_in_channels,
             rate=params.audio_in_sample_rate,
-            frames_per_buffer=params.audio_in_sample_rate,
+            frames_per_buffer=num_frames,
+            stream_callback=self._audio_in_callback,
             input=True)
 
-    def read_next_audio_frame(self) -> AudioRawFrame | None:
-        sample_rate = self._params.audio_in_sample_rate
-        num_channels = self._params.audio_in_channels
-        num_frames = int(sample_rate / 100)  # 10ms of audio
-
-        audio = self._in_stream.read(num_frames, exception_on_overflow=False)
-
-        return AudioRawFrame(audio=audio, sample_rate=sample_rate, num_channels=num_channels)
-
     async def start(self, frame: StartFrame):
         await super().start(frame)
         self._in_stream.start_stream()
@@ -60,6 +55,17 @@ class LocalAudioInputTransport(BaseInputTransport):
 
         await super().cleanup()
 
+    def _audio_in_callback(self, in_data, frame_count, time_info, status):
+        if not self._running:
+            return (None, pyaudio.paAbort)
+
+        frame = AudioRawFrame(audio=in_data,
+                              sample_rate=self._params.audio_in_sample_rate,
+                              num_channels=self._params.audio_in_channels)
+        self.push_audio_frame(frame)
+
+        return (None, pyaudio.paContinue)
+
 
 class LocalAudioOutputTransport(BaseOutputTransport):
 
@@ -75,21 +81,9 @@ class LocalAudioOutputTransport(BaseOutputTransport):
     def write_raw_audio_frames(self, frames: bytes):
         self._out_stream.write(frames)
 
-    async def start(self, frame: StartFrame):
-        await super().start(frame)
-        self._out_stream.start_stream()
-
-    async def stop(self):
-        await super().stop()
-        self._out_stream.stop_stream()
-
     async def cleanup(self):
-        # This is not very pretty (taken from PyAudio docs).
-        while self._out_stream.is_active():
-            await asyncio.sleep(0.1)
-        self._out_stream.close()
-
         await super().cleanup()
+        self._out_stream.close()
 
 
 class LocalAudioTransport(BaseTransport):

src/pipecat/transports/local/tk.py

@@ -38,22 +38,17 @@ class TkInputTransport(BaseInputTransport):
     def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
         super().__init__(params)
 
+        sample_rate = self._params.audio_in_sample_rate
+        num_frames = int(sample_rate / 100) * 2  # 20ms of audio
+
         self._in_stream = py_audio.open(
             format=py_audio.get_format_from_width(2),
             channels=params.audio_in_channels,
             rate=params.audio_in_sample_rate,
-            frames_per_buffer=params.audio_in_sample_rate,
+            frames_per_buffer=num_frames,
+            stream_callback=self._audio_in_callback,
             input=True)
 
-    def read_next_audio_frame(self) -> AudioRawFrame | None:
-        sample_rate = self._params.audio_in_sample_rate
-        num_channels = self._params.audio_in_channels
-        num_frames = int(sample_rate / 100)  # 10ms of audio
-
-        audio = self._in_stream.read(num_frames, exception_on_overflow=False)
-
-        return AudioRawFrame(audio=audio, sample_rate=sample_rate, num_channels=num_channels)
-
     async def start(self, frame: StartFrame):
         await super().start(frame)
         self._in_stream.start_stream()
@@ -63,12 +58,22 @@ class TkInputTransport(BaseInputTransport):
         self._in_stream.stop_stream()
 
     async def cleanup(self):
+        await super().cleanup()
         # This is not very pretty (taken from PyAudio docs).
         while self._in_stream.is_active():
             await asyncio.sleep(0.1)
         self._in_stream.close()
 
-        await super().cleanup()
+    def _audio_in_callback(self, in_data, frame_count, time_info, status):
+        if not self._running:
+            return (None, pyaudio.paAbort)
+
+        frame = AudioRawFrame(audio=in_data,
+                              sample_rate=self._params.audio_in_sample_rate,
+                              num_channels=self._params.audio_in_channels)
+        self.push_audio_frame(frame)
+
+        return (None, pyaudio.paContinue)
 
 
 class TkOutputTransport(BaseOutputTransport):
@@ -95,21 +100,9 @@ class TkOutputTransport(BaseOutputTransport):
     def write_frame_to_camera(self, frame: ImageRawFrame):
         self.get_event_loop().call_soon(self._write_frame_to_tk, frame)
 
-    async def start(self, frame: StartFrame):
-        await super().start(frame)
-        self._out_stream.start_stream()
-
-    async def stop(self):
-        await super().stop()
-        self._out_stream.stop_stream()
-
     async def cleanup(self):
-        # This is not very pretty (taken from PyAudio docs).
-        while self._out_stream.is_active():
-            await asyncio.sleep(0.1)
-        self._out_stream.close()
-
         await super().cleanup()
+        self._out_stream.close()
 
     def _write_frame_to_tk(self, frame: ImageRawFrame):
         width = frame.size[0]

src/pipecat/transports/services/daily.py

@@ -37,7 +37,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.transports.base_input import BaseInputTransport
 from pipecat.transports.base_output import BaseOutputTransport
 from pipecat.transports.base_transport import BaseTransport, TransportParams
-from pipecat.vad.vad_analyzer import VADAnalyzer, VADParams, VADState
+from pipecat.vad.vad_analyzer import VADAnalyzer, VADParams
 
 from loguru import logger
 
@@ -193,7 +193,7 @@ class DailyTransportClient(EventHandler):
         num_channels = self._params.audio_in_channels
 
         if self._other_participant_has_joined:
-            num_frames = int(sample_rate / 100)  # 10ms of audio
+            num_frames = int(sample_rate / 100) * 2  # 20ms of audio
 
             audio = self._speaker.read_frames(num_frames)
 
@@ -472,7 +472,6 @@ class DailyInputTransport(BaseInputTransport):
         self._client = client
 
         self._video_renderers = {}
-        self._camera_in_queue = queue.Queue()
 
         self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer
         if params.vad_enabled and not params.vad_analyzer:
@@ -483,23 +482,26 @@ class DailyInputTransport(BaseInputTransport):
     async def start(self, frame: StartFrame):
         if self._running:
             return
+        # Parent start.
+        await super().start(frame)
         # Join the room.
         await self._client.join()
-        # This will set _running=True
+        # Create audio task. It reads audio frames from Daily and push them
-        await super().start(frame)
+        # internally for VAD processing.
-        # Create camera in thread (runs if _running is true).
+        if self._params.audio_in_enabled or self._params.vad_enabled:
-        self._camera_in_thread = self._loop.run_in_executor(
+            self._audio_in_thread = self._loop.run_in_executor(
-            self._in_executor, self._camera_in_thread_handler)
+                self._executor, self._audio_in_thread_handler)
 
     async def stop(self):
         if not self._running:
             return
+        # Parent stop. This will set _running to False.
+        await super().stop()
         # Leave the room.
         await self._client.leave()
-        # This will set _running=False
+        # Stop audio thread.
-        await super().stop()
+        if self._params.audio_in_enabled or self._params.vad_enabled:
-        # The thread will stop.
+            await self._audio_in_thread
-        await self._camera_in_thread
 
     async def cleanup(self):
         await super().cleanup()
@@ -508,9 +510,6 @@ class DailyInputTransport(BaseInputTransport):
     def vad_analyzer(self) -> VADAnalyzer | None:
         return self._vad_analyzer
 
-    def read_next_audio_frame(self) -> AudioRawFrame | None:
-        return self._client.read_next_audio_frame()
-
     #
     # FrameProcessor
     #
@@ -536,6 +535,16 @@ class DailyInputTransport(BaseInputTransport):
             self._internal_push_frame(frame), self.get_event_loop())
         future.result()
 
+    #
+    # Audio in
+    #
+
+    def _audio_in_thread_handler(self):
+        while self._running:
+            frame = self._client.read_next_audio_frame()
+            if frame:
+                self.push_audio_frame(frame)
+
     #
     # Camera in
     #
@@ -584,23 +593,12 @@ class DailyInputTransport(BaseInputTransport):
                 image=buffer,
                 size=size,
                 format=format)
-            self._camera_in_queue.put(frame)
+            future = asyncio.run_coroutine_threadsafe(
+                self._internal_push_frame(frame), self.get_event_loop())
+            future.result()
 
         self._video_renderers[participant_id]["timestamp"] = curr_time
 
-    def _camera_in_thread_handler(self):
-        while self._running:
-            try:
-                frame = self._camera_in_queue.get(timeout=1)
-                future = asyncio.run_coroutine_threadsafe(
-                    self._internal_push_frame(frame), self.get_event_loop())
-                future.result()
-                self._camera_in_queue.task_done()
-            except queue.Empty:
-                pass
-            except BaseException as e:
-                logger.error(f"Error capturing video: {e}")
-
 
 class DailyOutputTransport(BaseOutputTransport):
 
@@ -612,7 +610,7 @@ class DailyOutputTransport(BaseOutputTransport):
     async def start(self, frame: StartFrame):
         if self._running:
             return
-        # This will set _running=True
+        # Parent start.
         await super().start(frame)
         # Join the room.
         await self._client.join()
@@ -620,7 +618,7 @@ class DailyOutputTransport(BaseOutputTransport):
     async def stop(self):
         if not self._running:
             return
-        # This will set _running=False
+        # Parent stop. This will set _running to False.
         await super().stop()
         # Leave the room.
         await self._client.leave()