moved expiry time to create_room

add CHANGELOG.md

CHANGELOG.md

@@ -0,0 +1,232 @@
+# Changelog
+
+All notable changes to **pipecat** will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [0.0.15] - 2024-05-15
+
+### Fixed
+
+- Quick hot fix for receiving `DailyTransportMessage`.
+
+## [0.0.14] - 2024-05-15
+
+### Added
+
+- Added `DailyTransport` event `on_participant_left`.
+
+- Added support for receiving `DailyTransportMessage`.
+
+### Fixed
+
+- Images are now resized to the size of the output camera. This was causing
+  images not being displayed.
+
+- Fixed an issue in `DailyTransport` that would not allow the input processor to
+  shutdown if no participant ever joined the room.
+
+- Fixed base transports start and stop. In some situation processors would halt
+  or not shutdown properly.
+
+## [0.0.13] - 2024-05-14
+
+### Changed
+
+- `MoondreamService` argument `model_id` is now `model`.
+
+- `VADAnalyzer` arguments have been renamed for more clarity.
+
+### Fixed
+
+- Fixed an issue with `DailyInputTransport` and `DailyOutputTransport` that
+  could cause some threads to not start properly.
+
+- Fixed `STTService`. Add `max_silence_secs` and `max_buffer_secs` to handle
+  better what's being passed to the STT service. Also add exponential smoothing
+  to the RMS.
+
+- Fixed `WhisperSTTService`. Add `no_speech_prob` to avoid garbage output text.
+
+## [0.0.12] - 2024-05-14
+
+### Added
+
+- Added `DailyTranscriptionSettings` to be able to specify transcription
+  settings much easier (e.g. language).
+
+### Other
+
+- Updated `simple-chatbot` with Spanish.
+
+- Add missing dependencies in some of the examples.
+
+## [0.0.11] - 2024-05-13
+
+### Added
+
+- Allow stopping pipeline tasks with new `StopTaskFrame`.
+
+### Changed
+
+- TTS, STT and image generation service now use `AsyncGenerator`.
+
+### Fixed
+
+- `DailyTransport`: allow registering for participant transcriptions even if
+  input transport is not initialized yet.
+
+### Other
+
+- Updated `storytelling-chatbot`.
+
+## [0.0.10] - 2024-05-13
+
+### Added
+
+- Added Intel GPU support to `MoondreamService`.
+
+- Added support for sending transport messages (e.g. to communicate with an app
+  at the other end of the transport).
+
+- Added `FrameProcessor.push_error()` to easily send an `ErrorFrame` upstream.
+
+### Fixed
+
+- Fixed Azure services (TTS and image generation).
+
+### Other
+
+- Updated `simple-chatbot`, `moondream-chatbot` and `translation-chatbot`
+  examples.
+
+## [0.0.9] - 2024-05-12
+
+### Changed
+
+Many things have changed in this version. Many of the main ideas such as frames,
+processors, services and transports are still there but some things have changed
+a bit.
+
+- `Frame`s describe the basic units for processing. For example, text, image or
+  audio frames. Or control frames to indicate a user has started or stopped
+  speaking.
+
+- `FrameProcessor`s process frames (e.g. they convert a `TextFrame` to an
+  `ImageRawFrame`) and push new frames downstream or upstream to their linked
+  peers.
+
+- `FrameProcessor`s can be linked together. The easiest wait is to use the
+  `Pipeline` which is a container for processors. Linking processors allow
+  frames to travel upstream or downstream easily.
+
+- `Transport`s are a way to send or receive frames. There can be local
+  transports (e.g. local audio or native apps), network transports
+  (e.g. websocket) or service transports (e.g. https://daily.co).
+
+- `Pipeline`s are just a processor container for other processors.
+
+- A `PipelineTask` know how to run a pipeline.
+
+- A `PipelineRunner` can run one or more tasks and it is also used, for example,
+  to capture Ctrl-C from the user.
+
+## [0.0.8] - 2024-04-11
+
+### Added
+
+- Added `FireworksLLMService`.
+
+- Added `InterimTranscriptionFrame` and enable interim results in
+  `DailyTransport` transcriptions.
+
+### Changed
+
+- `FalImageGenService` now uses new `fal_client` package.
+
+### Fixed
+
+- `FalImageGenService`: use `asyncio.to_thread` to not block main loop when
+  generating images.
+
+- Allow `TranscriptionFrame` after an end frame (transcriptions can be delayed
+  and received after `UserStoppedSpeakingFrame`).
+
+## [0.0.7] - 2024-04-10
+
+### Added
+
+- Add `use_cpu` argument to `MoondreamService`.
+
+## [0.0.6] - 2024-04-10
+
+### Added
+
+- Added `FalImageGenService.InputParams`.
+
+- Added `URLImageFrame` and `UserImageFrame`.
+
+- Added `UserImageRequestFrame` and allow requesting an image from a participant.
+
+- Added base `VisionService` and `MoondreamService`
+
+### Changed
+
+- Don't pass `image_size` to `ImageGenService`, images should have their own size.
+
+- `ImageFrame` now receives a tuple`(width,height)` to specify the size.
+
+- `on_first_other_participant_joined` now gets a participant argument.
+
+### Fixed
+
+- Check if camera, speaker and microphone are enabled before writing to them.
+
+### Performance
+
+- `DailyTransport` only subscribe to desired participant video track.
+
+## [0.0.5] - 2024-04-06
+
+### Changed
+
+- Use `camera_bitrate` and `camera_framerate`.
+
+- Increase `camera_framerate` to 30 by default.
+
+### Fixed
+
+- Fixed `LocalTransport.read_audio_frames`.
+
+## [0.0.4] - 2024-04-04
+
+### Added
+
+- Added project optional dependencies `[silero,openai,...]`.
+
+### Changed
+
+- Moved thransports to its own directory.
+
+- Use `OPENAI_API_KEY` instead of `OPENAI_CHATGPT_API_KEY`.
+
+### Fixed
+
+- Don't write to microphone/speaker if not enabled.
+
+### Other
+
+- Added live translation example.
+
+- Fix foundational examples.
+
+## [0.0.3] - 2024-03-13
+
+### Other
+
+- Added `storybot` and `chatbot` examples.
+
+## [0.0.2] - 2024-03-12
+
+Initial public release.

CHANGELOG.md.template

@@ -0,0 +1,62 @@
+# Changelog
+
+All notable changes to the **<project name>** SDK will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+Please make sure to add your changes to the appropriate categories:
+
+## [Unreleased]
+
+### Added
+
+<!-- for new functionality -->
+
+- n/a
+
+### Changed
+
+<!-- for changed functionality -->
+
+- n/a
+
+### Deprecated
+
+<!-- for soon-to-be removed functionality -->
+
+- n/a
+
+### Removed
+
+<!-- for removed functionality -->
+
+- n/a
+
+### Fixed
+
+<!-- for fixed bugs -->
+
+- n/a
+
+### Performance
+
+<!-- for performance-relevant changes -->
+
+- n/a
+
+### Security
+
+<!-- for security-relevant changes -->
+
+- n/a
+
+### Other
+
+<!-- for everything else -->
+
+- n/a
+
+## [0.1.0] - YYYY-MM-DD
+
+Initial release.

README.md

@@ -1,5 +1,5 @@
 <div align="center">
- <img alt="pipecat" width="300px" height="auto" src="pipecat.png">
+ <img alt="pipecat" width="300px" height="auto" src="https://raw.githubusercontent.com/pipecat-ai/pipecat/main/pipecat.png">
 </div>
 
 # Pipecat
@@ -12,11 +12,11 @@
 Take a look at some example apps:
 
 <p float="left">
-    <a href="https://github.com/pipecat-ai/pipecat/tree/main/examples/simple-chatbot"><img src="examples/simple-chatbot/image.png" width="280" /></a>&nbsp;
-    <a href="https://github.com/pipecat-ai/pipecat/tree/main/examples/storytelling-chatbot"><img src="examples/storytelling-chatbot/image.png" width="280" /></a>
+    <a href="https://github.com/pipecat-ai/pipecat/tree/main/examples/simple-chatbot"><img src="https://raw.githubusercontent.com/pipecat-ai/pipecat/main/examples/simple-chatbot/image.png" width="280" /></a>&nbsp;
+    <a href="https://github.com/pipecat-ai/pipecat/tree/main/examples/storytelling-chatbot"><img src="https://raw.githubusercontent.com/pipecat-ai/pipecat/main/examples/storytelling-chatbot/image.png" width="280" /></a>
     <br/>
-    <a href="https://github.com/pipecat-ai/pipecat/tree/main/examples/translation-chatbot"><img src="examples/translation-chatbot/image.png" width="280" /></a>&nbsp;
-    <a href="https://github.com/pipecat-ai/pipecat/tree/main/examples/moondream-chatbot"><img src="examples/moondream-chatbot/image.png" width="280" /></a>
+    <a href="https://github.com/pipecat-ai/pipecat/tree/main/examples/translation-chatbot"><img src="https://raw.githubusercontent.com/pipecat-ai/pipecat/main/examples/translation-chatbot/image.png" width="280" /></a>&nbsp;
+    <a href="https://github.com/pipecat-ai/pipecat/tree/main/examples/moondream-chatbot"><img src="https://raw.githubusercontent.com/pipecat-ai/pipecat/main/examples/moondream-chatbot/image.png" width="280" /></a>
 </p>
 
 ## Getting started with voice agents
@@ -218,4 +218,4 @@ Install the
 
 ➡️ [Join our Discord](https://discord.gg/pipecat)
 
-➡️ [Reach us on Twitter](https://x.com/pipecat_ai)
+➡️ [Reach us on X](https://x.com/pipecat_ai)

examples/foundational/09-mirror.py

@@ -7,12 +7,9 @@
 import asyncio
 import sys
 
-from pipecat.frames.frames import AudioRawFrame, ImageRawFrame
-from pipecat.processors.filter import Filter
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.transports.services.daily import DailyTransport, DailyParams
 
 from runner import configure
@@ -42,13 +39,7 @@ async def main(room_url, token):
     async def on_first_participant_joined(transport, participant):
         transport.capture_participant_video(participant["id"])
 
-    # The ParallelPipeline is not really necessary here but it shows how you
-    # would process audio and video concurrently in parallel pipelines.
-    pipeline = Pipeline([transport.input(),
-                         ParallelPipeline(
-                             [Filter([AudioRawFrame])],
-                             [Filter([ImageRawFrame])]),
-                         transport.output()])
+    pipeline = Pipeline([transport.input(), transport.output()])
 
     runner = PipelineRunner()
 

examples/simple-chatbot/bot.py

@@ -2,6 +2,7 @@ import asyncio
 import aiohttp
 import os
 import sys
+import argparse
 
 from PIL import Image
 
@@ -20,11 +21,9 @@ from pipecat.frames.frames import (
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.transports.services.daily import DailyParams, DailyTranscriptionSettings, DailyTransport
 from pipecat.vad.silero import SileroVAD
 
-from runner import configure
-
 from loguru import logger
 
 from dotenv import load_dotenv
@@ -43,7 +42,8 @@ for i in range(1, 26):
     # Get the filename without the extension to use as the dictionary key
     # Open the image and convert it to bytes
     with Image.open(full_path) as img:
-        sprites.append(ImageRawFrame(image=img.tobytes(), size=img.size, format=img.format))
+        sprites.append(ImageRawFrame(image=img.tobytes(),
+                       size=img.size, format=img.format))
 
 flipped = sprites[::-1]
 sprites.extend(flipped)
@@ -87,7 +87,15 @@ async def main(room_url: str, token):
                 camera_out_enabled=True,
                 camera_out_width=1024,
                 camera_out_height=576,
-                transcription_enabled=True
+                transcription_enabled=True,
+                #
+                # Spanish
+                #
+                # transcription_settings=DailyTranscriptionSettings(
+                #     language="es",
+                #     tier="nova",
+                #     model="2-general"
+                # )
             )
         )
 
@@ -96,7 +104,16 @@ async def main(room_url: str, token):
         tts = ElevenLabsTTSService(
             aiohttp_session=session,
             api_key=os.getenv("ELEVENLABS_API_KEY"),
+            #
+            # English
+            #
             voice_id="pNInz6obpgDQGcFmaJgB",
+
+            #
+            # Spanish
+            #
+            # model="eleven_multilingual_v2",
+            # voice_id="gD1IexrzCvsXPHUuT0s3",
         )
 
         llm = OpenAILLMService(
@@ -106,7 +123,15 @@ async def main(room_url: str, token):
         messages = [
             {
                 "role": "system",
+                #
+                # English
+                #
                 "content": "You are Chatbot, a friendly, helpful robot. 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, but keep your responses brief. Start by introducing yourself.",
+
+                #
+                # Spanish
+                #
+                # "content": "Eres Chatbot, un amigable y útil robot. Tu objetivo es demostrar tus capacidades de una manera breve. Tus respuestas se convertiran a audio así que nunca no debes incluir caracteres especiales. Contesta a lo que el usuario pregunte de una manera creativa, útil y breve. Empieza por presentarte a ti mismo.",
             },
         ]
 
@@ -131,5 +156,9 @@ async def main(room_url: str, token):
 
 
 if __name__ == "__main__":
-    (url, token) = configure()
-    asyncio.run(main(url, token))
+    parser = argparse.ArgumentParser(description="Daily Storyteller Bot")
+    parser.add_argument("-u", type=str, help="Room URL")
+    parser.add_argument("-t", type=str, help="Token")
+    config = parser.parse_args()
+
+    asyncio.run(main(config.u, config.t))

examples/simple-chatbot/bot_runner.py

@@ -0,0 +1,169 @@
+from daily_helpers import create_room, get_token, check_room_url
+import os
+import argparse
+import subprocess
+import atexit
+from pathlib import Path
+from typing import Optional
+
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from fastapi.responses import FileResponse, JSONResponse
+
+from dotenv import load_dotenv
+load_dotenv(override=True)
+
+# Bot sub-process dict for status reporting and concurrency control
+bot_procs = {}
+
+
+def cleanup():
+    # Clean up function, just to be extra safe
+    for proc in bot_procs.values():
+        proc[0].terminate()
+        proc[0].wait()
+
+
+atexit.register(cleanup)
+
+# ------------ Configuration ------------ #
+
+MAX_SESSION_TIME = + 5 * 60  # 5 minutes
+BOT_CAN_IDLE = True
+SERVE_STATIC = True
+STATIC_DIR = "../web-ui/dist"
+STATIC_ROUTE = "/static"
+STATIC_INDEX = "index.html"
+USE_OPEN_MIC = True  # Can the user freely talk, or do they need to wait their turn?
+
+# ----------------- API ----------------- #
+
+app = FastAPI()
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"]
+)
+
+# Optionally serve client static files
+if SERVE_STATIC:
+    app.mount(STATIC_ROUTE, StaticFiles(
+        directory=STATIC_DIR, html=True), name="static")
+
+    @app.get("/{path_name:path}", response_class=FileResponse)
+    async def catch_all(path_name: Optional[str] = ""):
+        if path_name == "":
+            return FileResponse(f"{STATIC_DIR}/{STATIC_INDEX}")
+
+        file_path = Path(STATIC_DIR) / (path_name or "")
+
+        if file_path.is_file():
+            return file_path
+
+        html_file_path = file_path.with_suffix(".html")
+        if html_file_path.is_file():
+            return FileResponse(html_file_path)
+
+        raise HTTPException(
+            status_code=404, detail="Page not found")
+
+
+@app.post("/start_bot")
+async def start_bot(request: Request) -> JSONResponse:
+    try:
+        data = await request.json()
+        # Is this a webhook creation request?
+        if "test" in data:
+            return JSONResponse({"test": True})
+    except Exception:
+        pass
+
+    # Use specified room URL, or create a new one if not specified
+    room_url = os.getenv("DAILY_SAMPLE_ROOM_URL", None)
+
+    if not room_url:
+        try:
+            room_url = create_room(MAX_SESSION_TIME)
+        except Exception:
+            raise HTTPException(
+                status_code=500,
+                detail="Unable to provision room")
+    else:
+        # Check passed room URL exists
+        try:
+            check_room_url(room_url)
+        except Exception:
+            raise HTTPException(
+                status_code=500, detail=f"Room not found: {room_url}")
+
+    # Give the agent a token to join the session
+    token = get_token(room_url)
+
+    if not room_url or not token:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to get token for room: {room_url}")
+
+    # Spawn a new agent, and join the user session
+    # Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
+
+    # @TODO: Spawn a new fly machine here...
+    try:
+        proc = subprocess.Popen(
+            [
+                f"python3 -m bot -u {room_url} -t {token}"
+            ],
+            shell=True,
+            bufsize=1,
+            cwd=os.path.dirname(os.path.abspath(__file__))
+        )
+        bot_procs[proc.pid] = (proc, room_url)
+    except Exception as e:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to start subprocess: {e}")
+
+     # Grab a token for the user to join with
+    user_token = get_token(room_url)
+
+    return JSONResponse({
+        "bot_id": proc.pid,
+        "room_url": room_url,
+        "token": user_token,
+        "config": {"open_mic": USE_OPEN_MIC}})
+
+
+# ----------------- Main ----------------- #
+
+if __name__ == "__main__":
+    # Check environment variables
+    required_env_vars = ['OPENAI_API_KEY', 'DAILY_API_KEY',
+                         'ELEVENLABS_VOICE_ID', 'ELEVENLABS_API_KEY']
+    for env_var in required_env_vars:
+        if env_var not in os.environ:
+            raise Exception(f"Missing environment variable: {env_var}.")
+
+    parser = argparse.ArgumentParser(description="Pipecat Bot Runner")
+    parser.add_argument("--host", type=str,
+                        default=os.getenv("HOST", "localhost"), help="Host address")
+    parser.add_argument("--port", type=int,
+                        default=os.getenv("PORT", 7860), help="Port number")
+    parser.add_argument("--reload", action="store_true",
+                        default=True, help="Reload code on change")
+
+    config = parser.parse_args()
+
+    try:
+        import uvicorn
+
+        uvicorn.run(
+            "bot_runner:app",
+            host=config.host,
+            port=config.port,
+            reload=config.reload
+        )
+
+    except KeyboardInterrupt:
+        print("Pipecat runner shutting down...")

examples/simple-chatbot/daily_helpers.py

@@ -1,4 +1,5 @@
 
+from re import X
 import urllib.parse
 import os
 import time
@@ -9,11 +10,11 @@ from dotenv import load_dotenv
 load_dotenv()
 
 
-daily_api_path = os.getenv("DAILY_API_URL") or "api.daily.co/v1"
+daily_api_path = os.getenv("DAILY_API_URL", "api.daily.co/v1")
 daily_api_key = os.getenv("DAILY_API_KEY")
 
 
-def create_room() -> tuple[str, str]:
+def create_room(expiry_time=5 * 60) -> tuple[str, str]:
     """
     Helper function to create a Daily room.
     # See: https://docs.daily.co/reference/rest-api/rooms
@@ -25,7 +26,7 @@ def create_room() -> tuple[str, str]:
         Exception: If the request to create the room fails or if the response does not contain the room URL or room name.
     """
     room_props = {
-        "exp": time.time() + 60 * 60,  # 1 hour
+        "exp": time.time() * expiry_time,
         "enable_chat": True,
         "enable_emoji_reactions": True,
         "eject_at_room_exp": True,
@@ -50,6 +51,31 @@ def create_room() -> tuple[str, str]:
     return room_url, room_name
 
 
+def check_room_url(room_url: str) -> bool:
+    """
+    Checks if a room exists in Daily.
+    # See: https://docs.daily.co/reference/rest-api/rooms/get-room-config
+
+    Args:
+        room_name (str): The url of the room to check for
+
+    Returns:
+        bool: True if 200 OK, Exception otherwise.
+    """
+
+    room_name = get_name_from_url(room_url)
+
+    res: requests.Response = requests.get(
+        f"https://{daily_api_path}/rooms/{room_name}",
+        headers={"Authorization": f"Bearer {daily_api_key}"}
+    )
+
+    if res.status_code != 200:
+        raise Exception(f"Room not found: {room_name}")
+
+    return True
+
+
 def get_name_from_url(room_url: str) -> str:
     """
     Extracts the name from a given room URL.

examples/simple-chatbot/requirements.txt

@@ -1,5 +1,7 @@
-python-dotenv
-requests
-fastapi[all]
+pipecat-ai[daily,openai,fal]
+fastapi
 uvicorn
-pipecat-ai[daily,openai]
+requests
+python-dotenv
+loguru
+requests

examples/simple-chatbot/runner.py

@@ -1,58 +0,0 @@
-import argparse
-import os
-import time
-import urllib
-import requests
-
-
-def configure():
-    parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
-    parser.add_argument(
-        "-u",
-        "--url",
-        type=str,
-        required=False,
-        help="URL of the Daily room to join")
-    parser.add_argument(
-        "-k",
-        "--apikey",
-        type=str,
-        required=False,
-        help="Daily API Key (needed to create an owner token for the room)",
-    )
-
-    args, unknown = parser.parse_known_args()
-
-    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
-    key = args.apikey or os.getenv("DAILY_API_KEY")
-
-    if not url:
-        raise Exception(
-            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL.")
-
-    if not key:
-        raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.")
-
-    # Create a meeting token for the given room with an expiration 1 hour in
-    # the future.
-    room_name: str = urllib.parse.urlparse(url).path[1:]
-    expiration: float = time.time() + 60 * 60
-
-    res: requests.Response = requests.post(
-        f"https://api.daily.co/v1/meeting-tokens",
-        headers={
-            "Authorization": f"Bearer {key}"},
-        json={
-            "properties": {
-                "room_name": room_name,
-                "is_owner": True,
-                "exp": expiration}},
-    )
-
-    if res.status_code != 200:
-        raise Exception(
-            f"Failed to create meeting token: {res.status_code} {res.text}")
-
-    token: str = res.json()["token"]
-
-    return (url, token)

examples/simple-chatbot/server.py

@@ -1,124 +0,0 @@
-import os
-import argparse
-import subprocess
-import atexit
-
-from fastapi import FastAPI, Request, HTTPException
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import JSONResponse, RedirectResponse
-
-from utils.daily_helpers import create_room as _create_room, get_token
-
-MAX_BOTS_PER_ROOM = 1
-
-# Bot sub-process dict for status reporting and concurrency control
-bot_procs = {}
-
-
-def cleanup():
-    # Clean up function, just to be extra safe
-    for proc in bot_procs.values():
-        proc.terminate()
-        proc.wait()
-
-
-atexit.register(cleanup)
-
-
-app = FastAPI()
-
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-
-@app.get("/start")
-async def start_agent(request: Request):
-    print(f"!!! Creating room")
-    room_url, room_name = _create_room()
-    print(f"!!! Room URL: {room_url}")
-    # Ensure the room property is present
-    if not room_url:
-        raise HTTPException(
-            status_code=500,
-            detail="Missing 'room' property in request data. Cannot start agent without a target room!")
-
-    # Check if there is already an existing process running in this room
-    num_bots_in_room = sum(
-        1 for proc in bot_procs.values() if proc[1] == room_url and proc[0].poll() is None)
-    if num_bots_in_room >= MAX_BOTS_PER_ROOM:
-        raise HTTPException(
-            status_code=500, detail=f"Max bot limited reach for room: {room_url}")
-
-    # Get the token for the room
-    token = get_token(room_url)
-
-    if not token:
-        raise HTTPException(
-            status_code=500, detail=f"Failed to get token for room: {room_url}")
-
-    # Spawn a new agent, and join the user session
-    # Note: this is mostly for demonstration purposes (refer to 'deployment' in README)
-    try:
-        proc = subprocess.Popen(
-            [
-                f"python3 -m bot -u {room_url} -t {token}"
-            ],
-            shell=True,
-            bufsize=1,
-            cwd=os.path.dirname(os.path.abspath(__file__))
-        )
-        bot_procs[proc.pid] = (proc, room_url)
-    except Exception as e:
-        raise HTTPException(
-            status_code=500, detail=f"Failed to start subprocess: {e}")
-
-    return RedirectResponse(room_url)
-
-
-@app.get("/status/{pid}")
-def get_status(pid: int):
-    # Look up the subprocess
-    proc = bot_procs.get(pid)
-
-    # If the subprocess doesn't exist, return an error
-    if not proc:
-        raise HTTPException(
-            status_code=404, detail=f"Bot with process id: {pid} not found")
-
-    # Check the status of the subprocess
-    if proc[0].poll() is None:
-        status = "running"
-    else:
-        status = "finished"
-
-    return JSONResponse({"bot_id": pid, "status": status})
-
-
-if __name__ == "__main__":
-    import uvicorn
-
-    default_host = os.getenv("HOST", "0.0.0.0")
-    default_port = int(os.getenv("FAST_API_PORT", "7860"))
-
-    parser = argparse.ArgumentParser(
-        description="Daily Storyteller FastAPI server")
-    parser.add_argument("--host", type=str,
-                        default=default_host, help="Host address")
-    parser.add_argument("--port", type=int,
-                        default=default_port, help="Port number")
-    parser.add_argument("--reload", action="store_true",
-                        help="Reload code on change")
-
-    config = parser.parse_args()
-
-    uvicorn.run(
-        "server:app",
-        host=config.host,
-        port=config.port,
-        reload=config.reload,
-    )

examples/storytelling-chatbot/env.example

@@ -1,6 +1,5 @@
-ELEVENLABS_API_KEY=
-ELEVENLABS_VOICE_ID=
-FAL_KEY=
-DAILY_API_URL=api.daily.co/v1
-DAILY_API_KEY=
-OPENAI_API_KEY=
+DAILY_API_KEY=7df...
+ELEVENLABS_API_KEY=aeb...
+ELEVENLABS_VOICE_ID=7S...
+FAL_KEY=8c...
+OPENAI_API_KEY=sk-PL...

examples/storytelling-chatbot/requirements.txt

@@ -1,5 +1,6 @@
-dailyai[daily,openai,fal]==0.0.8
+async_timeout
 fastapi
 uvicorn
 requests
-python-dotenv
+python-dotenv
+pipecat-ai[daily,openai,fal]

examples/storytelling-chatbot/src/bot.py

@@ -1,37 +1,31 @@
+import argparse
 import asyncio
 import aiohttp
-import logging
 import os
-import argparse
+import sys
 
-from dailyai.pipeline.pipeline import Pipeline
-from dailyai.pipeline.frames import (
-    AudioFrame,
-    ImageFrame,
-    EndPipeFrame,
-    LLMMessagesFrame,
-    SendAppMessageFrame
-)
-from dailyai.pipeline.aggregators import (
-    LLMUserResponseAggregator,
-    LLMAssistantResponseAggregator,
-)
-from dailyai.transports.daily_transport import DailyTransport
-from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
-from dailyai.services.open_ai_services import OpenAILLMService
-from dailyai.services.fal_ai_services import FalImageGenService
+
+from pipecat.frames.frames import LLMMessagesFrame, StopTaskFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineTask
+from pipecat.processors.aggregators.llm_response import LLMAssistantResponseAggregator, LLMUserResponseAggregator
+from pipecat.services.elevenlabs import ElevenLabsTTSService
+from pipecat.services.fal import FalImageGenService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport, DailyTransportMessageFrame
 
 from processors import StoryProcessor, StoryImageProcessor
 from prompts import LLM_BASE_PROMPT, LLM_INTRO_PROMPT, CUE_USER_TURN
 from utils.helpers import load_sounds, load_images
 
+from loguru import logger
+
 from dotenv import load_dotenv
 load_dotenv(override=True)
 
-logging.basicConfig(format=f"[STORYBOT] %(levelno)s %(asctime)s %(message)s")
-logger = logging.getLogger("dailyai")
-logger.setLevel(logging.INFO)
-
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
 
 sounds = load_sounds(["listening.wav"])
 images = load_images(["book1.png", "book2.png"])
@@ -46,16 +40,14 @@ async def main(room_url, token=None):
             room_url,
             token,
             "Storytelling Bot",
-            duration_minutes=5,
-            start_transcription=True,
-            mic_enabled=True,
-            mic_sample_rate=16000,
-            vad_enabled=True,
-            camera_framerate=30,
-            camera_bitrate=680000,
-            camera_enabled=True,
-            camera_width=768,
-            camera_height=768,
+            DailyParams(
+                audio_out_enabled=True,
+                camera_out_enabled=True,
+                camera_out_width=768,
+                camera_out_height=768,
+                transcription_enabled=True,
+                vad_enabled=True,
+            )
         )
 
         logger.debug("Transport created for room:" + room_url)
@@ -103,68 +95,54 @@ async def main(room_url, token=None):
 
         # -------------- Story Loop ------------- #
 
+        runner = PipelineRunner()
+
+        # The intro pipeline is used to start
+        # the story (as per LLM_INTRO_PROMPT)
+        intro_pipeline = Pipeline([llm_service, tts_service, transport.output()])
+
+        intro_task = PipelineTask(intro_pipeline)
+
         logger.debug("Waiting for participant...")
 
-        start_storytime_event = asyncio.Event()
-
-        @transport.event_handler("on_first_other_participant_joined")
-        async def on_first_other_participant_joined(transport, participant):
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
             logger.debug("Participant joined, storytime commence!")
-            start_storytime_event.set()
-
-        # The storytime coroutine will wait for the start_storytime_event
-        # to be set before starting the storytime pipeline
-        async def storytime():
-            await start_storytime_event.wait()
-
-            # The intro pipeline is used to start
-            # the story (as per LLM_INTRO_PROMPT)
-            intro_pipeline = Pipeline(processors=[
-                llm_service,
-                tts_service,
-            ], sink=transport.send_queue)
-
-            await intro_pipeline.queue_frames(
+            transport.capture_participant_transcription(participant["id"])
+            await intro_task.queue_frames(
                 [
-                    ImageFrame(images['book1'], (768, 768)),
+                    images['book1'],
                     LLMMessagesFrame([LLM_INTRO_PROMPT]),
-                    SendAppMessageFrame(CUE_USER_TURN, None),
-                    AudioFrame(sounds["listening"]),
-                    ImageFrame(images['book2'], (768, 768)),
-                    EndPipeFrame(),
+                    DailyTransportMessageFrame(CUE_USER_TURN),
+                    sounds["listening"],
+                    images['book2'],
+                    StopTaskFrame()
                 ]
             )
 
-            # We start the pipeline as soon as the user joins
-            await intro_pipeline.run_pipeline()
+        # We run the intro pipeline. This will start the transport. The intro
+        # task will exit after StopTaskFrame is processed.
+        await runner.run(intro_task)
 
-            # The main story pipeline is used to continue the
-            # story based on user input
-            pipeline = Pipeline(processors=[
-                user_responses,
-                llm_service,
-                story_processor,
-                image_processor,
-                tts_service,
-                llm_responses,
-            ])
+        # The main story pipeline is used to continue the story based on user
+        # input.
+        main_pipeline = Pipeline([
+            transport.input(),
+            user_responses,
+            llm_service,
+            story_processor,
+            image_processor,
+            tts_service,
+            llm_responses,
+            transport.output()
+        ])
 
-            await transport.run_pipeline(pipeline)
-
-        transport.transcription_settings["extra"]["endpointing"] = True
-        transport.transcription_settings["extra"]["punctuate"] = True
-
-        try:
-            await asyncio.gather(transport.run(), storytime())
-        except (asyncio.CancelledError, KeyboardInterrupt):
-            transport.stop()
-
-        logger.debug("Pipeline finished. Exiting.")
+        main_task = PipelineTask(main_pipeline)
 
+        await runner.run(main_task)
 
 if __name__ == "__main__":
-    parser = argparse.ArgumentParser(
-        description="Daily Storyteller Bot")
+    parser = argparse.ArgumentParser(description="Daily Storyteller Bot")
     parser.add_argument("-u", type=str, help="Room URL")
     parser.add_argument("-t", type=str, help="Token")
     config = parser.parse_args()

examples/storytelling-chatbot/src/processors.py

@@ -1,19 +1,13 @@
-from typing import AsyncGenerator
 import re
 
-from dailyai.pipeline.frames import TextFrame, Frame, AudioFrame
-from dailyai.pipeline.frame_processor import FrameProcessor
-from dailyai.pipeline.frames import (
-    Frame,
-    TextFrame,
-    SendAppMessageFrame,
-    LLMResponseEndFrame,
-    UserStoppedSpeakingFrame,
-)
+from async_timeout import timeout
+
+from pipecat.frames.frames import Frame, LLMResponseEndFrame, TextFrame, UserStoppedSpeakingFrame
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.transports.services.daily import DailyTransportMessageFrame
 
 from utils.helpers import load_sounds
 from prompts import IMAGE_GEN_PROMPT, CUE_USER_TURN, CUE_ASSISTANT_TURN
-import asyncio
 
 sounds = load_sounds(["talking.wav", "listening.wav", "ding.wav"])
 
@@ -42,7 +36,7 @@ class StoryImageProcessor(FrameProcessor):
     Processor for image prompt frames that will be sent to the FAL service.
 
     This processor is responsible for consuming frames of type `StoryImageFrame`.
-    It processes the by passing it to the FAL service
+    It processes them by passing it to the FAL service.
     The processed frames are then yielded back.
 
     Attributes:
@@ -50,25 +44,26 @@ class StoryImageProcessor(FrameProcessor):
     """
 
     def __init__(self, fal_service):
+        super().__init__()
         self._fal_service = fal_service
 
-    async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, StoryImageFrame):
             try:
-                async with asyncio.timeout(7):
-                    async for i in self._fal_service.process_frame(TextFrame(IMAGE_GEN_PROMPT % frame.text)):
-                        yield i
+                async with timeout(7):
+                    async for i in self._fal_service.run_image_gen(IMAGE_GEN_PROMPT % frame.text):
+                        await self.push_frame(i)
             except TimeoutError:
                 pass
             pass
         else:
-            yield frame
+            await self.push_frame(frame)
 
 
 class StoryProcessor(FrameProcessor):
     """
     Primary frame processor. It takes the frames generated by the LLM
-    and processes them into image prompts and story pages (sentences.)
+    and processes them into image prompts and story pages (sentences).
     For a clearer picture of how this works, reference prompts.py
 
     Attributes:
@@ -81,15 +76,16 @@ class StoryProcessor(FrameProcessor):
     """
 
     def __init__(self, messages, story):
+        super().__init__()
         self._messages = messages
         self._text = ""
         self._story = story
 
-    async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, UserStoppedSpeakingFrame):
             # Send an app message to the UI
-            yield SendAppMessageFrame(CUE_ASSISTANT_TURN, None)
-            yield AudioFrame(sounds["talking"])
+            await self.push_frame(DailyTransportMessageFrame(CUE_ASSISTANT_TURN))
+            await self.push_frame(sounds["talking"])
 
         elif isinstance(frame, TextFrame):
             # We want to look for sentence breaks in the text
@@ -111,7 +107,7 @@ class StoryProcessor(FrameProcessor):
                 # Remove the image prompt from the text
                 self._text = re.sub(r"<.*?>", '', self._text, count=1)
                 # Process the image prompt frame
-                yield StoryImageFrame(image_prompt)
+                await self.push_frame(StoryImageFrame(image_prompt))
 
             # STORY PAGE
             # Looking for: [break] in the LLM response
@@ -126,9 +122,9 @@ class StoryProcessor(FrameProcessor):
                 if len(self._text) > 2:
                     # Append the sentence to the story
                     self._story.append(self._text)
-                    yield StoryPageFrame(self._text)
+                    await self.push_frame(StoryPageFrame(self._text))
                     # Assert that it's the LLMs turn, until we're finished
-                    yield SendAppMessageFrame(CUE_ASSISTANT_TURN, None)
+                    await self.push_frame(DailyTransportMessageFrame(CUE_ASSISTANT_TURN))
                 # Clear the buffer
                 self._text = ""
 
@@ -136,13 +132,13 @@ class StoryProcessor(FrameProcessor):
         # Driven by the prompt, the LLM should have asked the user for input
         elif isinstance(frame, LLMResponseEndFrame):
             # We use a different frame type, as to avoid image generation ingest
-            yield StoryPromptFrame(self._text)
+            await self.push_frame(StoryPromptFrame(self._text))
             self._text = ""
-            yield frame
+            await self.push_frame(frame)
             # Send an app message to the UI
-            yield SendAppMessageFrame(CUE_USER_TURN, None)
-            yield AudioFrame(sounds["listening"])
+            await self.push_frame(DailyTransportMessageFrame(CUE_USER_TURN))
+            await self.push_frame(sounds["listening"])
 
         # Anything that is not a TextFrame pass through
         else:
-            yield frame
+            await self.push_frame(frame)

examples/storytelling-chatbot/src/prompts.py

@@ -3,7 +3,7 @@ LLM_INTRO_PROMPT = {
     "content": "You are a creative storyteller who loves to tell whimsical, fantastical stories. \
         Your goal is to craft an engaging and fun story. \
         Start by asking the user what kind of story they'd like to hear. Don't provide any examples. \
-        Keep your reponse to only a few sentences."
+        Keep your response to only a few sentences."
 }
 
 

examples/storytelling-chatbot/src/server.py

@@ -8,7 +8,7 @@ from typing import Optional
 from fastapi import FastAPI, Request, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles
-from fastapi.responses import FileResponse, JSONResponse, RedirectResponse
+from fastapi.responses import FileResponse, JSONResponse
 
 from utils.daily_helpers import create_room as _create_room, get_token, get_name_from_url
 

examples/storytelling-chatbot/src/utils/daily_helpers.py

@@ -9,7 +9,7 @@ from dotenv import load_dotenv
 load_dotenv()
 
 
-daily_api_path = os.getenv("DAILY_API_URL")
+daily_api_path = os.getenv("DAILY_API_URL") or "api.daily.co/v1"
 daily_api_key = os.getenv("DAILY_API_KEY")
 
 

examples/storytelling-chatbot/src/utils/helpers.py

@@ -2,6 +2,8 @@ import os
 import wave
 from PIL import Image
 
+from pipecat.frames.frames import AudioRawFrame, ImageRawFrame
+
 script_dir = os.path.dirname(__file__)
 
 
@@ -14,7 +16,7 @@ def load_images(image_files):
         filename = os.path.splitext(os.path.basename(full_path))[0]
         # Open the image and convert it to bytes
         with Image.open(full_path) as img:
-            images[filename] = img.tobytes()
+            images[filename] = ImageRawFrame(image=img.tobytes(), size=img.size, format=img.format)
     return images
 
 
@@ -28,6 +30,8 @@ def load_sounds(sound_files):
         filename = os.path.splitext(os.path.basename(full_path))[0]
         # Open the sound and convert it to bytes
         with wave.open(full_path) as audio_file:
-            sounds[filename] = audio_file.readframes(-1)
+            sounds[filename] = AudioRawFrame(audio=audio_file.readframes(-1),
+                                             sample_rate=audio_file.getframerate(),
+                                             num_channels=audio_file.getnchannels())
 
     return sounds

examples/web-ui/.eslintrc.cjs

@@ -0,0 +1,18 @@
+module.exports = {
+  root: true,
+  env: { browser: true, es2020: true },
+  extends: [
+    'eslint:recommended',
+    'plugin:@typescript-eslint/recommended',
+    'plugin:react-hooks/recommended',
+  ],
+  ignorePatterns: ['dist', '.eslintrc.cjs'],
+  parser: '@typescript-eslint/parser',
+  plugins: ['react-refresh'],
+  rules: {
+    'react-refresh/only-export-components': [
+      'warn',
+      { allowConstantExport: true },
+    ],
+  },
+}

examples/web-ui/.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

examples/web-ui/README.md

@@ -0,0 +1,30 @@
+# React + TypeScript + Vite
+
+This template provides a minimal setup to get React working in Vite with HMR and some ESLint rules.
+
+Currently, two official plugins are available:
+
+- [@vitejs/plugin-react](https://github.com/vitejs/vite-plugin-react/blob/main/packages/plugin-react/README.md) uses [Babel](https://babeljs.io/) for Fast Refresh
+- [@vitejs/plugin-react-swc](https://github.com/vitejs/vite-plugin-react-swc) uses [SWC](https://swc.rs/) for Fast Refresh
+
+## Expanding the ESLint configuration
+
+If you are developing a production application, we recommend updating the configuration to enable type aware lint rules:
+
+- Configure the top-level `parserOptions` property like this:
+
+```js
+export default {
+  // other rules...
+  parserOptions: {
+    ecmaVersion: 'latest',
+    sourceType: 'module',
+    project: ['./tsconfig.json', './tsconfig.node.json'],
+    tsconfigRootDir: __dirname,
+  },
+}
+```
+
+- Replace `plugin:@typescript-eslint/recommended` to `plugin:@typescript-eslint/recommended-type-checked` or `plugin:@typescript-eslint/strict-type-checked`
+- Optionally add `plugin:@typescript-eslint/stylistic-type-checked`
+- Install [eslint-plugin-react](https://github.com/jsx-eslint/eslint-plugin-react) and add `plugin:react/recommended` & `plugin:react/jsx-runtime` to the `extends` list

examples/web-ui/env.example

@@ -0,0 +1 @@
+VITE_SERVER_URL=... #optional: if serving frontend independetely from backend (otherwise relative.)

examples/web-ui/index.html

@@ -0,0 +1,17 @@
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="UTF-8" />
+    <link rel="icon" type="image/svg+xml" href="/favicon.ico" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <link rel="preconnect" href="https://fonts.googleapis.com">
+    <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
+    <link href="https://fonts.googleapis.com/css2?family=Inter:wght@100..900&family=Space+Mono:wght@400;700&display=swap" rel="stylesheet">
+
+    <title>Pipecat Demo</title>
+  </head>
+  <body>
+    <div id="root"></div>
+    <script type="module" src="/src/main.tsx"></script>
+  </body>
+</html>

examples/web-ui/package.json

@@ -0,0 +1,36 @@
+{
+  "name": "pipecatdemo",
+  "private": true,
+  "version": "0.0.0",
+  "type": "module",
+  "scripts": {
+    "dev": "vite",
+    "build": "tsc && vite build",
+    "lint": "eslint . --ext ts,tsx --report-unused-disable-directives --max-warnings 0",
+    "preview": "vite preview"
+  },
+  "dependencies": {
+    "@daily-co/daily-js": "^0.64.0",
+    "@daily-co/daily-react": "^0.19.0",
+    "class-variance-authority": "^0.7.0",
+    "lucide-react": "^0.378.0",
+    "react": "^18.2.0",
+    "react-dom": "^18.2.0",
+    "recoil": "^0.7.7"
+  },
+  "devDependencies": {
+    "@types/react": "^18.2.66",
+    "@types/react-dom": "^18.2.22",
+    "@typescript-eslint/eslint-plugin": "^7.2.0",
+    "@typescript-eslint/parser": "^7.2.0",
+    "@vitejs/plugin-react": "^4.2.1",
+    "eslint": "^8.57.0",
+    "eslint-plugin-react-hooks": "^4.6.0",
+    "eslint-plugin-react-refresh": "^0.4.6",
+    "i": "^0.3.7",
+    "npm": "^10.8.0",
+    "typescript": "^5.2.2",
+    "vite": "^5.2.0",
+    "vite-plugin-webfont-dl": "^3.9.4"
+  }
+}

examples/web-ui/src/App.tsx

@@ -0,0 +1,141 @@
+import { useState } from "react";
+import { useDaily } from "@daily-co/daily-react";
+
+import { Alert } from "./components/alert";
+import { Button } from "./components/button";
+import { ArrowRight, Loader2 } from "lucide-react";
+import { DeviceSelect } from "./components/DeviceSelect";
+import Session from "./components/Session";
+
+type State =
+  | "idle"
+  | "configuring"
+  | "requesting_agent"
+  | "connecting"
+  | "connected"
+  | "started"
+  | "finished"
+  | "error";
+
+export default function App() {
+  // Use Daily as our agent transport
+  const daily = useDaily();
+
+  const [state, setState] = useState<State>("idle");
+  const [error, setError] = useState<string | null>(null);
+  const [config, setConfig] = useState<{ open_mic?: boolean }>({});
+
+  async function start() {
+    if (!daily) return;
+
+    setState("requesting_agent");
+
+    const serverUrl =
+      import.meta.env.VITE_SERVER_URL || import.meta.env.BASE_URL;
+
+    // Request a bot to join your session
+    let data;
+
+    try {
+      const res = await fetch(`${serverUrl}start_bot`, {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+        },
+      });
+
+      data = await res.json();
+      setConfig(data.config || {});
+
+      if (!res.ok) {
+        setError(data.detail);
+        setState("error");
+        return;
+      }
+    } catch (e) {
+      setError(
+        `Unable to connect to the server at '${serverUrl}' - is it running?`
+      );
+      setState("error");
+      return;
+    }
+
+    setState("connecting");
+
+    await daily.join({
+      url: data.room_url,
+      token: data.token,
+      videoSource: false,
+      startAudioOff: true,
+    });
+
+    setState("connected");
+  }
+
+  async function leave() {
+    await daily?.leave();
+    setState("idle");
+  }
+
+  if (state === "error") {
+    return (
+      <Alert intent="danger" title="An error occurred">
+        {error}
+      </Alert>
+    );
+  }
+
+  if (state === "connected") {
+    return <Session onLeave={() => leave()} openMic={config?.open_mic} />;
+  }
+
+  const status_text = {
+    configuring: "Start",
+    requesting_agent: "Requesting agent...",
+    connecting: "Connecting to agent...",
+  };
+
+  if (state !== "idle") {
+    return (
+      <div className="card card-appear">
+        <div className="card-inner">
+          <h1 className="card-header">Configure your devices</h1>
+          <p className="card-text">
+            Please configure your microphone and speakers below
+          </p>
+          <DeviceSelect />
+          <Button
+            key="start"
+            onClick={() => start()}
+            disabled={state !== "configuring"}
+          >
+            {state !== "configuring" && <Loader2 className="animate-spin" />}
+            {status_text[state as keyof typeof status_text]}
+          </Button>
+        </div>
+      </div>
+    );
+  }
+
+  return (
+    <div className="card card-appear">
+      <div className="card-inner">
+        <h1 className="card-header">Pipecat Simple Chatbot</h1>
+        <p className="card-text">
+          Please ensure you microphone and speakers are connected and ready to
+          go
+        </p>
+        {import.meta.env.DEV && !import.meta.env.VITE_SERVER_URL && (
+          <div>
+            Warning: you have not set a server URL for local development. Please
+            set <code>VITE_SERVER_URL</code> in{" "}
+            <code>.env.development.local</code>
+          </div>
+        )}
+        <Button key="next" onClick={() => setState("configuring")}>
+          Next <ArrowRight />
+        </Button>
+      </div>
+    </div>
+  );
+}

examples/web-ui/src/assets/logo.svg

@@ -0,0 +1,7 @@
+<svg width="332" height="192" viewBox="0 0 332 192" fill="none" xmlns="http://www.w3.org/2000/svg">
+<path d="M45.7718 0.769635C50.4477 -0.990844 55.7252 0.330188 59.0204 4.08595L101.936 53H230.064L272.98 4.08595C276.275 0.330188 281.552 -0.990844 286.228 0.769635C290.904 2.53011 294 7.00367 294 12V120H332V144H270V43.8728L244.52 72.9141C242.242 75.5111 238.955 77 235.5 77H96.5C93.0452 77 89.7581 75.5111 87.4796 72.9141L62 43.8728V144H0V120H38V12C38 7.00367 41.0958 2.53011 45.7718 0.769635Z" fill="black"/>
+<path d="M270 168H332V192H270V168Z" fill="black"/>
+<path d="M0 168H62V192H0V168Z" fill="black"/>
+<path d="M128 128C128 136.837 120.837 144 112 144C103.163 144 96 136.837 96 128C96 119.164 103.163 112 112 112C120.837 112 128 119.164 128 128Z" fill="black"/>
+<path d="M236 128C236 136.837 228.837 144 220 144C211.163 144 204 136.837 204 128C204 119.164 211.163 112 220 112C228.837 112 236 119.164 236 128Z" fill="black"/>
+</svg>

examples/web-ui/src/components/AudioIndicator/index.tsx

@@ -0,0 +1,31 @@
+import {
+  useAudioLevel,
+  useAudioTrack,
+  useLocalSessionId,
+} from "@daily-co/daily-react";
+import { useCallback, useRef } from "react";
+
+import styles from "./styles.module.css";
+
+export const AudioIndicatorBar: React.FC = () => {
+  const localSessionId = useLocalSessionId();
+  const audioTrack = useAudioTrack(localSessionId);
+
+  const volRef = useRef<HTMLDivElement>(null);
+
+  useAudioLevel(
+    audioTrack?.persistentTrack,
+    useCallback((volume) => {
+      if (volRef.current)
+        volRef.current.style.width = Math.max(2, volume * 100) + "%";
+    }, [])
+  );
+
+  return (
+    <div className={styles.bar}>
+      <div ref={volRef} />
+    </div>
+  );
+};
+
+export default AudioIndicatorBar;

examples/web-ui/src/components/AudioIndicator/styles.module.css

@@ -0,0 +1,15 @@
+.bar {
+  background: var(--color-gray-200);
+  height: 8px;
+  width: 100%;
+  border-radius: 999px;
+  overflow: hidden;
+
+  > div {
+    background: var(--color-green-500);
+    height: 8px;
+    width: 0px;
+    border-radius: 999px;
+    transition: width 0.1s ease;
+  }
+}

examples/web-ui/src/components/DeviceSelect/index.tsx

@@ -0,0 +1,127 @@
+"use client";
+
+import { useEffect } from "react";
+import { DailyMeetingState } from "@daily-co/daily-js";
+import { useDaily, useDevices } from "@daily-co/daily-react";
+import { Mic, Speaker } from "lucide-react";
+import { AudioIndicatorBar } from "../AudioIndicator";
+
+import styles from "./styles.module.css";
+import { Alert } from "../alert";
+
+export function DeviceSelect() {
+  const daily = useDaily();
+  const {
+    currentMic,
+    hasMicError,
+    micState,
+    microphones,
+    setMicrophone,
+    currentSpeaker,
+    speakers,
+    setSpeaker,
+  } = useDevices();
+
+  const handleMicrophoneChange = (value: string) => {
+    setMicrophone(value);
+  };
+
+  const handleSpeakerChange = (value: string) => {
+    setSpeaker(value);
+  };
+
+  useEffect(() => {
+    if (microphones.length > 0 || !daily || daily.isDestroyed()) return;
+    const meetingState = daily.meetingState();
+    const meetingStatesBeforeJoin: DailyMeetingState[] = [
+      "new",
+      "loading",
+      "loaded",
+    ];
+    if (meetingStatesBeforeJoin.includes(meetingState)) {
+      daily.startCamera({ startVideoOff: true, startAudioOff: false });
+    }
+  }, [daily, microphones]);
+
+  return (
+    <div className={styles.deviceSelect}>
+      {hasMicError && (
+        <Alert intent="danger" title="Device error">
+          {micState === "blocked" ? (
+            <>
+              Please check your browser and system permissions. Make sure that
+              this app is allowed to access your microphone and refresh the
+              page.
+            </>
+          ) : micState === "in-use" ? (
+            <>
+              Your microphone is being used by another app. Please close any
+              other apps using your microphone and restart this app.
+            </>
+          ) : micState === "not-found" ? (
+            <>
+              No microphone seems to be connected. Please connect a microphone.
+            </>
+          ) : micState === "not-supported" ? (
+            <>
+              This app is not supported on your device. Please update your
+              software or use a different device.
+            </>
+          ) : (
+            <>
+              There seems to be an issue accessing your microphone. Try
+              restarting the app or consult a system administrator.
+            </>
+          )}
+        </Alert>
+      )}
+
+      <section className={styles.field}>
+        <label className={styles.label}>Microphone:</label>
+        <div className={styles.selectContainer}>
+          <Mic size={24} />
+          <select
+            onChange={(e) => handleMicrophoneChange(e.target.value)}
+            defaultValue={currentMic?.device.deviceId}
+            className={styles.deviceSelectField}
+          >
+            {microphones.length === 0 ? (
+              <option value="">Loading devices...</option>
+            ) : (
+              microphones.map((m) => (
+                <option key={m.device.deviceId} value={m.device.deviceId}>
+                  {m.device.label}
+                </option>
+              ))
+            )}
+          </select>
+        </div>
+        <AudioIndicatorBar />
+      </section>
+
+      <section className={styles.field}>
+        <label className={styles.label}>Speakers:</label>
+        <div className={styles.selectContainer}>
+          <Speaker size={24} />
+          <select
+            onChange={(e) => handleSpeakerChange(e.target.value)}
+            defaultValue={currentSpeaker?.device.deviceId}
+            className={styles.deviceSelectField}
+          >
+            {speakers.length === 0 ? (
+              <option value="">Loading devices...</option>
+            ) : (
+              speakers.map((m) => (
+                <option key={m.device.deviceId} value={m.device.deviceId}>
+                  {m.device.label}
+                </option>
+              ))
+            )}
+          </select>
+        </div>
+      </section>
+    </div>
+  );
+}
+
+export default DeviceSelect;

examples/web-ui/src/components/DeviceSelect/styles.module.css

@@ -0,0 +1,39 @@
+.deviceSelect {
+  display: flex;
+  flex-flow: column wrap;
+  width: 100%;
+  gap: 1rem;
+}
+
+.field {
+  display: flex;
+  flex-flow: column wrap;
+  align-items: flex-start;
+  gap: 0.5rem;
+}
+
+.label {
+  font-weight: 600;
+  font-size: 0.875rem;
+}
+
+.selectContainer {
+  position: relative;
+  width: 100%;
+
+  > svg {
+    position: absolute;
+    size: 24px;
+    top: 0;
+    bottom: 0;
+    margin: auto 0;
+    left: 0.75rem;
+    color: var(--color-gray-400);
+  }
+}
+
+.deviceSelectField {
+  width: 100%;
+  padding-left: 2.875rem;
+  margin-top: auto;
+}

examples/web-ui/src/components/Session/agent.tsx

@@ -0,0 +1,22 @@
+import React from "react";
+import Status from "./status";
+import styles from "./styles.module.css";
+
+import { useParticipantIds } from "@daily-co/daily-react";
+
+export const Agent: React.FC = () => {
+  const participantIds = useParticipantIds({ filter: "remote" });
+
+  const status = participantIds.length > 0 ? "connected" : "connecting";
+  return (
+    <div className={styles.agent}>
+      <div className={styles.agentWindow}></div>
+      <footer className={styles.agentFooter}>
+        <Status>User status</Status>
+        <Status variant={status}>Agent status</Status>
+      </footer>
+    </div>
+  );
+};
+
+export default Agent;

examples/web-ui/src/components/Session/index.tsx

@@ -0,0 +1,88 @@
+import React, { useEffect, useRef, useState } from "react";
+import { LogOut, Settings } from "lucide-react";
+import { DailyAudio, useAppMessage, useDaily } from "@daily-co/daily-react";
+
+import DeviceSelect from "../DeviceSelect";
+import Agent from "./agent";
+import { Button } from "../button";
+import UserMicBubble from "../UserMicBubble";
+
+import styles from "./styles.module.css";
+
+interface SessionProps {
+  onLeave: () => void;
+  openMic?: boolean;
+}
+
+export const Session: React.FC<SessionProps> = ({
+  onLeave,
+  openMic = false,
+}) => {
+  const daily = useDaily();
+  const [showDevices, setShowDevices] = useState(false);
+  const modalRef = useRef<HTMLDialogElement>(null);
+  const [talkState, setTalkState] = useState<"user" | "assistant" | "open">(
+    openMic ? "open" : "assistant"
+  );
+
+  useAppMessage({
+    onAppMessage: (e) => {
+      if (!daily || !e.data?.cue) return;
+
+      // Determine the UI state from the cue sent by the bot
+      if (e.data?.cue === "user_turn") {
+        // Delay enabling local mic input to avoid feedback from LLM
+        setTimeout(() => daily.setLocalAudio(true), 500);
+        setTalkState("user");
+      } else {
+        daily.setLocalAudio(false);
+        setTalkState("assistant");
+      }
+    },
+  });
+
+  useEffect(() => {
+    const current = modalRef.current;
+    // Backdrop doesn't currently work with dialog open, so we use setModal instead
+    if (current && showDevices) {
+      current.inert = true;
+      current.showModal();
+      current.inert = false;
+    }
+    return () => current?.close();
+  }, [showDevices]);
+
+  return (
+    <>
+      <dialog ref={modalRef}>
+        <h2>Configure devices</h2>
+        <DeviceSelect />
+        <Button onClick={() => setShowDevices(false)}>Close</Button>
+      </dialog>
+
+      <div className={styles.agentContainer}>
+        <Agent />
+        <UserMicBubble openMic={openMic} active={talkState !== "assistant"} />
+        <DailyAudio />
+      </div>
+
+      <footer className={styles.footer}>
+        <div className={styles.controls}>
+          <Button
+            variant="ghost"
+            size="icon"
+            onClick={() => setShowDevices(true)}
+          >
+            <Settings />
+          </Button>
+          <Button onClick={() => onLeave()}>
+            <LogOut size={16} />
+            End
+          </Button>
+        </div>
+      </footer>
+    </>
+  );
+};
+
+export default Session;

examples/web-ui/src/components/Session/status.tsx

@@ -0,0 +1,42 @@
+import { VariantProps, cva } from "class-variance-authority";
+import styles from "./styles.module.css";
+
+const statusVariants = cva(styles.statusIndicator, {
+  variants: {
+    variant: {
+      default: styles.statusDefault,
+      connecting: styles.statusOrange,
+      connected: styles.statusGreen,
+    },
+  },
+  defaultVariants: {
+    variant: "default",
+  },
+});
+
+export interface StatusProps
+  extends React.HTMLAttributes<HTMLElement>,
+    VariantProps<typeof statusVariants> {}
+
+const status_text = {
+  default: "Idle",
+  connecting: "Connecting",
+  connected: "Connected",
+};
+
+export const Status: React.FC<StatusProps> = ({
+  children,
+  variant = "default",
+}) => {
+  return (
+    <div className={styles.status}>
+      <span>{children}</span>
+      <div className={statusVariants({ variant })}>
+        <span />
+        {status_text[variant as keyof typeof status_text]}
+      </div>
+    </div>
+  );
+};
+
+export default Status;

examples/web-ui/src/components/Session/styles.module.css

@@ -0,0 +1,119 @@
+.footer {
+  display: flex;
+  flex-flow: row nowrap;
+  margin-top: auto;
+  align-self: flex-end;
+}
+
+.controls {
+  display: flex;
+  flex-flow: row nowrap;
+  justify-content: space-between;
+  gap: 0.75rem;
+}
+
+.agentContainer {
+  flex: 1;
+  display: flex;
+  flex-flow: column wrap;
+  align-items: center;
+  justify-content: center;
+}
+
+.agent {
+  margin-top: auto;
+  min-width: 420px;
+  padding: 0.75rem;
+  border-radius: var(--borderRadius-lg);
+  border-radius: 24px;
+  border: 1px solid #e2e8f0;
+  background: #fff;
+  box-shadow: 0px 360px 101px 0px rgba(0, 0, 0, 0),
+    0px 231px 92px 0px rgba(0, 0, 0, 0), 0px 130px 78px 0px rgba(0, 0, 0, 0.02),
+    0px 58px 58px 0px rgba(0, 0, 0, 0.03), 0px 14px 32px 0px rgba(0, 0, 0, 0.03);
+  background-image: linear-gradient(90deg, white, white),
+    linear-gradient(0deg, var(--color-gray-300), var(--color-gray-200));
+  background-clip: padding-box, border-box;
+  background-origin: border-box;
+  border: 1px solid transparent;
+}
+
+.agentWindow {
+  min-width: 400px;
+  aspect-ratio: 1;
+  background: var(--color-gray-200);
+  border-radius: var(--borderRadius-md);
+}
+
+.agentFooter {
+  margin: 1.5rem 0 0.75rem 0;
+  display: flex;
+  flex-flow: row nowrap;
+  justify-content: space-between;
+  width: 100%;
+
+  > :first-child {
+    border-right: 1px solid var(--color-gray-200);
+  }
+}
+
+.status {
+  flex: 1;
+  line-height: 1;
+  display: flex;
+  flex-flow: column wrap;
+  align-items: center;
+  gap: 0.5rem;
+
+  > span {
+    font-family: var(--font-mono);
+    font-size: 11px;
+    text-transform: uppercase;
+    line-height: 1;
+    letter-spacing: 0.7px;
+  }
+}
+
+.statusIndicator {
+  display: inline-flex;
+  align-items: center;
+  gap: 0.5rem;
+  font-size: 0.75rem;
+  font-weight: 700;
+  padding: 0.5rem 0.75rem;
+  border-radius: var(--borderRadius-xs);
+
+  > span {
+    display: block;
+    width: 9px;
+    height: 9px;
+    border-radius: 9px;
+    background: red;
+  }
+}
+
+.statusDefault {
+  color: var(--color-gray-500);
+  background: var(--color-gray-100);
+
+  > span {
+    background: var(--color-gray-400);
+  }
+}
+.statusOrange {
+  color: var(--color-orange-800);
+  background: var(--color-orange-100);
+
+  > span {
+    background: var(--color-orange-400);
+  }
+}
+
+.statusGreen {
+  color: var(--color-green-800);
+  background: var(--color-green-100);
+
+  > span {
+    background: var(--color-green-400);
+  }
+}

examples/web-ui/src/components/UserMicBubble/index.tsx

@@ -0,0 +1,71 @@
+import React, { useCallback, useRef } from "react";
+
+import {
+  useAudioLevel,
+  useAudioTrack,
+  useLocalSessionId,
+  //useAppMessage,
+} from "@daily-co/daily-react";
+//import { DailyEventObjectAppMessage } from "@daily-co/daily-js";
+import { Mic, MicOff } from "lucide-react";
+//import { TypewriterEffect } from "../ui/typewriter";
+import styles from "./styles.module.css";
+
+const AudioIndicatorBubble: React.FC = () => {
+  const localSessionId = useLocalSessionId();
+  const audioTrack = useAudioTrack(localSessionId);
+  const volRef = useRef<HTMLDivElement>(null);
+
+  useAudioLevel(
+    audioTrack?.persistentTrack,
+    useCallback((volume) => {
+      // this volume number will be between 0 and 1
+      // give it a minimum scale of 0.15 to not completely disappear 👻
+      if (volRef.current) {
+        const v = volume * 1.75;
+        volRef.current.style.transform = `scale(${Math.max(0.1, v)})`;
+      }
+    }, [])
+  );
+
+  // Your audio track's audio volume visualized in a small circle,
+  // whose size changes depending on the volume level
+  return <div ref={volRef} className={styles.volume} />;
+};
+
+interface Props {
+  active: boolean;
+  openMic: boolean;
+}
+
+export default function UserMicBubble({ active, openMic = false }: Props) {
+  /*
+  const [transcription, setTranscription] = useState<string[]>([]);
+  useAppMessage({
+    onAppMessage: (e: DailyEventObjectAppMessage<any>) => {
+      if (e.fromId && e.fromId === "transcription") {
+        if (e.data.user_id === "" && e.data.is_final) {
+          //setTranscription((t) => [...t, ...e.data.text.split(" ")]);
+        }
+      }
+    },
+  });
+
+  useEffect(() => {
+    if (active) return;
+    const t = setTimeout(() => setTranscription([]), 4000);
+    return () => clearTimeout(t);
+  }, [active]);*/
+
+  const cx = openMic ? styles.micIconOpen : active && styles.micIconActive;
+
+  return (
+    <div className={`${styles.bubbleContainer}`}>
+      <div className={`${styles.micIcon} ${cx}`}>
+        {!openMic && !active ? <MicOff size={42} /> : <Mic size={42} />}
+        {(openMic || active) && <AudioIndicatorBubble />}
+      </div>
+      <footer className={styles.transcript}></footer>
+    </div>
+  );
+}

examples/web-ui/src/components/UserMicBubble/styles.module.css

@@ -0,0 +1,103 @@
+.bubbleContainer {
+  color: #ffffff;
+  position: relative;
+  z-index: 20;
+  display: flex;
+  flex-direction: column;
+  margin: auto;
+}
+
+.micIcon {
+  position: relative;
+  box-sizing: border-box;
+  -moz-box-sizing: border-box;
+  -webkit-box-sizing: border-box;
+  width: 120px;
+  height: 120px;
+  border-radius: 120px;
+  display: flex;
+  justify-content: center;
+  align-items: center;
+  margin: 0 auto;
+  z-index: 20;
+  transition: all 0.5s ease;
+  border: 6px solid color-mix(in srgb, var(--color-gray-300), transparent 70%);
+  outline: 6px solid color-mix(in srgb, var(--color-gray-300), transparent 70%);
+  background-color: var(--color-gray-500);
+  background-image: radial-gradient(
+    var(--color-gray-300),
+    var(--color-gray-400)
+  );
+}
+
+.micIcon svg {
+  position: relative;
+  z-index: 20;
+  opacity: 0.3;
+  transition: opacity 0.5s ease;
+}
+
+@keyframes pulse {
+  0% {
+    outline-width: 6px;
+    @apply outline-teal-500/10;
+  }
+  50% {
+    outline-width: 24px;
+    @apply outline-teal-500/50;
+  }
+  100% {
+    outline-width: 6px;
+    @apply outline-teal-500/10;
+  }
+}
+
+.micIconOpen {
+  background-color: var(--color-gray-500);
+  background-image: radial-gradient(
+    var(--color-gray-500),
+    var(--color-gray-600)
+  );
+  border: 6px solid color-mix(in srgb, var(--color-gray-200), transparent 60%);
+  outline: 6px solid color-mix(in srgb, var(--color-gray-400), transparent 70%);
+}
+
+.micIconActive {
+  background-color: var(--color-green-500);
+  background-image: radial-gradient(
+    var(--color-green-500),
+    var(--color-green-600)
+  );
+  border: 6px solid color-mix(in srgb, var(--color-green-200), transparent 60%);
+  outline: 6px solid color-mix(in srgb, var(--color-green-400), transparent 70%);
+  animation: pulse 2s infinite ease-in-out;
+}
+
+.micIconOpen svg,
+.micIconActive svg {
+  opacity: 1;
+}
+
+.transcript {
+  flex: 0;
+  align-self: center;
+  opacity: 0.25;
+  transition: opacity 1s ease;
+  transition-delay: 2.5s;
+}
+
+.active .transcript {
+  opacity: 1;
+}
+
+.volume {
+  position: absolute;
+  overflow: hidden;
+  inset: 0px;
+  z-index: 0;
+  border-radius: 999px;
+  transition: all 0.1s ease;
+  transform: scale(0);
+  opacity: 0.4;
+  background-color: var(--color-green-200);
+}

examples/web-ui/src/components/alert.tsx

@@ -0,0 +1,27 @@
+import React from "react";
+import { cva, VariantProps } from "class-variance-authority";
+
+const alertVariants = cva("alert", {
+  variants: {
+    intent: {
+      info: "alert-info",
+      danger: "alert-danger",
+    },
+  },
+  defaultVariants: {
+    intent: "info",
+  },
+});
+
+export interface AlertProps
+  extends React.HTMLAttributes<HTMLElement>,
+    VariantProps<typeof alertVariants> {}
+
+export const Alert: React.FC<AlertProps> = ({ children, intent, title }) => {
+  return (
+    <div className={alertVariants({ intent })}>
+      <span>{title}</span>
+      {children}
+    </div>
+  );
+};

examples/web-ui/src/components/button.tsx

@@ -0,0 +1,27 @@
+import React from "react";
+import { cva, VariantProps } from "class-variance-authority";
+
+const buttonVariants = cva("button", {
+  variants: {
+    variant: {
+      primary: "button-primary",
+      ghost: "button-ghost",
+    },
+    size: {
+      base: "",
+      icon: "button-icon",
+    },
+  },
+  defaultVariants: {
+    variant: "primary",
+    size: "base",
+  },
+});
+
+export interface ButtonProps
+  extends React.ButtonHTMLAttributes<HTMLButtonElement>,
+    VariantProps<typeof buttonVariants> {}
+
+export const Button: React.FC<ButtonProps> = ({ variant, size, ...props }) => {
+  return <button className={buttonVariants({ variant, size })} {...props} />;
+};

examples/web-ui/src/components/header.tsx

@@ -0,0 +1,22 @@
+import Logo from "./logo";
+
+function Header() {
+  return (
+    <header className="header">
+      <span className="logo-button">
+        <Logo />
+      </span>
+
+      <nav>
+        <a href="https://git.new/ai" target="_blank">
+          GitHub
+        </a>
+        <a href="https://discord.gg/pipecat" target="_blank">
+          Discord
+        </a>
+      </nav>
+    </header>
+  );
+}
+
+export default Header;

examples/web-ui/src/components/logo.tsx

@@ -0,0 +1,46 @@
+import React from "react";
+
+const Logo: React.FC = () => {
+  return (
+    <svg
+      width="332"
+      height="192"
+      viewBox="0 0 332 192"
+      fill="none"
+      xmlns="http://www.w3.org/2000/svg"
+      className="logo"
+    >
+      <path
+        d="M45.7718 0.769635C50.4477 -0.990844 55.7252 0.330188 59.0204 4.08595L101.936 53H230.064L272.98 4.08595C276.275 0.330188 281.552 -0.990844 286.228 0.769635C290.904 2.53011 294 7.00367 294 12V120H332V144H270V43.8728L244.52 72.9141C242.242 75.5111 238.955 77 235.5 77H96.5C93.0452 77 89.7581 75.5111 87.4796 72.9141L62 43.8728V144H0V120H38V12C38 7.00367 41.0958 2.53011 45.7718 0.769635Z"
+        fill="black"
+      />
+      <path d="M270 168H332V192H270V168Z" fill="black" />
+      <path d="M0 168H62V192H0V168Z" fill="black" />
+      <g id="eyes_1">
+        <path
+          d="M128 128C128 136.837 120.837 144 112 144C103.163 144 96 136.837 96 128C96 119.164 103.163 112 112 112C120.837 112 128 119.164 128 128Z"
+          fill="black"
+        />
+        <path
+          d="M236 128C236 136.837 228.837 144 220 144C211.163 144 204 136.837 204 128C204 119.164 211.163 112 220 112C228.837 112 236 119.164 236 128Z"
+          fill="black"
+        />
+      </g>
+
+      <g id="eyes_2" visibility="hidden">
+        <path
+          d="M128 128C128 136.837 120.837 144 112 144C103.163 144 96 136.837 96 128C96 119.163 103.163 112 112 112C120.837 112 128 119.163 128 128Z"
+          stroke="black"
+          strokeWidth="12"
+        />
+        <path
+          d="M236 128C236 136.837 228.837 144 220 144C211.163 144 204 136.837 204 128C204 119.163 211.163 112 220 112C228.837 112 236 119.163 236 128Z"
+          stroke="black"
+          strokeWidth="12"
+        />
+      </g>
+    </svg>
+  );
+};
+
+export default Logo;

examples/web-ui/src/global.css

@@ -0,0 +1,387 @@
+:root {
+  --color-gray-50: #f8fafc;
+  --color-gray-100: #f1f5f9;
+  --color-gray-200: #e2e8f0;
+  --color-gray-300: #cbd5e1;
+  --color-gray-400: #94a3b8;
+  --color-gray-500: #64748b;
+  --color-gray-600: #475569;
+  --color-gray-700: #334155;
+  --color-gray-800: #1e293b;
+  --color-gray-900: #0f172a;
+  --color-gray-950: #020617;
+
+  --color-orange-50: #fff7ed;
+  --color-orange-100: #ffedd5;
+  --color-orange-200: #fed7aa;
+  --color-orange-300: #fdba74;
+  --color-orange-400: #fb923c;
+  --color-orange-500: #f97316;
+  --color-orange-600: #ea580c;
+  --color-orange-700: #c2410c;
+  --color-orange-800: #9a3412;
+  --color-orange-900: #7c2d12;
+  --color-orange-950: #431407;
+
+  --color-green-50: #f0fdf4;
+  --color-green-100: #dcfce7;
+  --color-green-200: #bbf7d0;
+  --color-green-300: #86efac;
+  --color-green-400: #4ade80;
+  --color-green-500: #22c55e;
+  --color-green-600: #16a34a;
+  --color-green-700: #15803d;
+  --color-green-800: #166534;
+  --color-green-900: #14532d;
+  --color-green-950: #052e16;
+
+  --borderRadius-xs: 6px;
+  --borderRadius-sm: 9px;
+  --borderRadius-md: 12px;
+  --borderRadius-lg: 24px;
+  --input-height: 48px;
+
+  --font-sans: "Inter", system-ui, Avenir, Helvetica, Arial, sans-serif;
+  --font-mono: "Space Mono", monospace;
+
+  --font-size-base: 1rem;
+  --font-size-sm: 0.875rem;
+  --font-size-lg: 1.125rem;
+  --font-size-xl: 1.25rem;
+}
+
+* {
+  font-family: var(--font-sans);
+}
+
+#root {
+  line-height: 1.5;
+  font-weight: 400;
+
+  font-synthesis: none;
+  text-rendering: optimizeLegibility;
+  -webkit-font-smoothing: antialiased;
+  -moz-osx-font-smoothing: grayscale;
+
+  margin: 0 auto;
+  padding: 1rem;
+  text-align: center;
+  display: flex;
+  flex-direction: column;
+  flex: 1;
+}
+
+* {
+  outline-color: black;
+  outline-offset: 4px;
+}
+
+body {
+  margin: 0;
+  min-width: 320px;
+  min-height: 100vh;
+  display: flex;
+  color: var(--color-gray-950);
+  background-color: var(--color-gray-50);
+}
+
+a {
+  color: var(--color-gray-950);
+  text-decoration: inherit;
+}
+a:hover {
+  text-decoration-line: underline;
+}
+
+main {
+  display: flex;
+  flex-direction: column;
+  align-items: center;
+  justify-content: center;
+  flex: 1;
+}
+
+h1 {
+  font-size: 3.2em;
+  line-height: 1.1;
+}
+
+.header {
+  display: flex;
+  align-self: flex-start;
+  align-items: center;
+  gap: 2rem;
+}
+
+.header .logo-button {
+  box-sizing: border-box;
+  border: 1px solid var(--color-gray-200);
+  padding: 0.5rem;
+  display: flex;
+  place-items: center;
+  border-radius: var(--borderRadius-md);
+  transition: all 0.3s ease;
+  background: white;
+  box-shadow: 0px 7px 2px 0px rgba(0, 0, 0, 0),
+    0px 5px 2px 0px rgba(0, 0, 0, 0.01), 0px 3px 2px 0px rgba(0, 0, 0, 0.03),
+    0px 1px 1px 0px rgba(0, 0, 0, 0.04), 0px 0px 1px 0px rgba(0, 0, 0, 0.05);
+}
+
+.header .logo {
+  width: 42px;
+  height: auto;
+  aspect-ratio: 1;
+}
+
+.header nav {
+  pointer-events: none;
+  flex-flow: row nowrap;
+  align-items: center;
+  gap: 2rem;
+  display: none;
+
+  a {
+    font-size: 1.125rem;
+    line-height: 1.75rem;
+    text-underline-offset: 4px;
+    text-decoration: underline solid transparent;
+    transition: text-decoration 0.5s ease;
+    &:hover {
+      text-decoration: underline solid var(--color-gray-400);
+    }
+  }
+}
+
+.header:hover {
+  .logo {
+    animation: wiggle 0.2s 1;
+  }
+  #eyes_2 {
+    visibility: visible;
+  }
+  #eyes_1 {
+    visibility: hidden;
+  }
+
+  nav {
+    display: flex;
+    pointer-events: all;
+    display: flex;
+    animation: fadeIn 0.5s ease;
+  }
+
+  .logo-button {
+    border-color: var(--color-gray-200);
+    box-shadow: 0px 100px 28px 0px rgba(0, 0, 0, 0),
+      0px 64px 26px 0px rgba(0, 0, 0, 0.01),
+      0px 36px 22px 0px rgba(0, 0, 0, 0.03),
+      0px 16px 16px 0px rgba(0, 0, 0, 0.04), 0px 4px 9px 0px rgba(0, 0, 0, 0.05);
+  }
+}
+
+@keyframes fadeIn {
+  from {
+    opacity: 0;
+  }
+  to {
+    opacity: 1;
+  }
+}
+
+@keyframes wiggle {
+  0% {
+    transform: translateX(0);
+  }
+  25% {
+    transform: translateX(-5px);
+  }
+  50% {
+    transform: translateX(5px);
+  }
+  75% {
+    transform: translateX(-5px);
+  }
+  100% {
+    transform: translateX(0);
+  }
+}
+
+.button {
+  border-radius: var(--borderRadius-md);
+  border: 1px solid transparent;
+  padding: 0 1.5rem;
+  height: var(--input-height);
+  font-size: 1rem;
+  font-weight: 600;
+  font-family: inherit;
+  background-color: var(--color-gray-950);
+  color: white;
+  cursor: pointer;
+  display: flex;
+  flex-direction: row;
+  gap: 0.5rem;
+  align-items: center;
+}
+.button > svg {
+  width: 20px;
+  height: 20px;
+}
+.button:hover:not(:disabled) {
+  background-color: var(--color-gray-600);
+}
+.button:disabled {
+  cursor: not-allowed;
+  background-color: var(--color-gray-300);
+  color: var(--color-gray-100);
+}
+
+.button-ghost {
+  border: 1px solid var(--color-gray-300);
+  background-color: white;
+  color: var(--color-gray-950);
+}
+.button-ghost:hover:not(:disabled) {
+  background-color: white;
+  border-color: var(--color-gray-600);
+}
+
+.button-icon {
+  padding: 1rem;
+}
+
+.alert {
+  border: 1px solid black;
+  border-radius: var(--borderRadius-md);
+  padding: 1rem;
+  display: flex;
+  flex-flow: column wrap;
+  gap: 2;
+  font-size: 0.875rem;
+
+  span {
+    font-size: 1rem;
+    font-weight: bold;
+  }
+}
+
+.alert-danger {
+  border-color: red;
+  color: red;
+}
+
+@keyframes spin {
+  0% {
+    transform: rotate(0deg);
+  }
+  100% {
+    transform: rotate(360deg);
+  }
+}
+
+.animate-spin {
+  animation: spin 1s linear infinite;
+}
+
+.controls {
+}
+
+.card {
+  padding: 12px;
+  border-radius: var(--borderRadius-lg);
+  background: #fff;
+  box-shadow: 0px 360px 101px 0px rgba(0, 0, 0, 0),
+    0px 231px 92px 0px rgba(0, 0, 0, 0), 0px 130px 78px 0px rgba(0, 0, 0, 0.02),
+    0px 58px 58px 0px rgba(0, 0, 0, 0.03), 0px 14px 32px 0px rgba(0, 0, 0, 0.03);
+
+  background-image: linear-gradient(90deg, white, white),
+    linear-gradient(0deg, var(--color-gray-300), var(--color-gray-200));
+  background-clip: padding-box, border-box;
+  background-origin: border-box;
+  border: 1px solid transparent;
+}
+
+.card-appear {
+  animation: appear 0.5s ease-out forwards;
+}
+
+.card-inner {
+  padding: 20px;
+  max-width: 420px;
+  display: flex;
+  gap: 24px;
+  flex-direction: column;
+  justify-content: center;
+  place-items: center;
+}
+
+.card-header {
+  font-size: 1.5rem;
+  font-weight: 600;
+  margin: 0px;
+}
+
+.card-text {
+  font-size: 1.125rem;
+  color: var(--color-gray-500);
+  text-wrap: pretty;
+  margin: 0px;
+}
+
+@keyframes appear {
+  from {
+    opacity: 0;
+    transform: translateY(1rem);
+  }
+  to {
+    opacity: 1;
+    transform: translateY(0);
+  }
+}
+
+select {
+  appearance: none;
+  background-image: url("data:image/svg+xml;base64,PHN2ZyB3aWR0aD0iMTYiIGhlaWdodD0iMTYiIHZpZXdCb3g9IjAgMCAxNiAxNiIgZmlsbD0ibm9uZSIgeG1sbnM9Imh0dHA6Ly93d3cudzMub3JnLzIwMDAvc3ZnIj4KPHBhdGggZD0iTTQgNkw4IDEwTDEyIDYiIHN0cm9rZT0iIzY0NzQ4QiIgc3Ryb2tlLWxpbmVjYXA9InJvdW5kIiBzdHJva2UtbGluZWpvaW49InJvdW5kIi8+Cjwvc3ZnPgo=");
+  background-repeat: no-repeat;
+  background-position: right 1rem top 50%;
+  background-size: 16px auto;
+  border-radius: var(--borderRadius-md);
+  border: 1px solid var(--color-gray-300);
+  padding: 0 1rem;
+  height: var(--input-height);
+  font-size: var(--font-size-sm);
+  white-space: nowrap;
+  text-overflow: ellipsis;
+  padding-right: 3rem;
+}
+
+dialog {
+  max-width: 420px;
+  padding: 2rem;
+  border-radius: var(--borderRadius-lg);
+  background: #fff;
+  box-shadow: 0px 360px 101px 0px rgba(0, 0, 0, 0),
+    0px 231px 92px 0px rgba(0, 0, 0, 0), 0px 130px 78px 0px rgba(0, 0, 0, 0.02),
+    0px 58px 58px 0px rgba(0, 0, 0, 0.03), 0px 14px 32px 0px rgba(0, 0, 0, 0.03);
+
+  background-image: linear-gradient(90deg, white, white),
+    linear-gradient(0deg, var(--color-gray-300), var(--color-gray-200));
+  background-clip: padding-box, border-box;
+  background-origin: border-box;
+  border: 1px solid transparent;
+  animation: appear 0.5s ease-out forwards;
+
+  > h2 {
+    font-size: var(--font-size-xl);
+    margin: 0;
+    margin-bottom: 2rem;
+  }
+
+  > button {
+    margin: 2rem auto 0 auto;
+  }
+}
+
+dialog::backdrop {
+  background-color: rgba(255, 255, 255, 0.5);
+  backdrop-filter: blur(5px);
+}

examples/web-ui/src/main.tsx

@@ -0,0 +1,18 @@
+import React from "react";
+import ReactDOM from "react-dom/client";
+import App from "./App.tsx";
+import Header from "./components/header.tsx";
+import { DailyProvider } from "@daily-co/daily-react";
+
+import "./global.css";
+
+ReactDOM.createRoot(document.getElementById("root")!).render(
+  <React.StrictMode>
+    <Header />
+    <main>
+      <DailyProvider>
+        <App />
+      </DailyProvider>
+    </main>
+  </React.StrictMode>
+);

examples/web-ui/src/vite-env.d.ts

@@ -0,0 +1 @@
+/// <reference types="vite/client" />

examples/web-ui/tsconfig.json

@@ -0,0 +1,25 @@
+{
+  "compilerOptions": {
+    "target": "ES2020",
+    "useDefineForClassFields": true,
+    "lib": ["ES2020", "DOM", "DOM.Iterable"],
+    "module": "ESNext",
+    "skipLibCheck": true,
+
+    /* Bundler mode */
+    "moduleResolution": "bundler",
+    "allowImportingTsExtensions": true,
+    "resolveJsonModule": true,
+    "isolatedModules": true,
+    "noEmit": true,
+    "jsx": "react-jsx",
+
+    /* Linting */
+    "strict": true,
+    "noUnusedLocals": true,
+    "noUnusedParameters": true,
+    "noFallthroughCasesInSwitch": true
+  },
+  "include": ["src"],
+  "references": [{ "path": "./tsconfig.node.json" }]
+}

examples/web-ui/tsconfig.node.json

@@ -0,0 +1,11 @@
+{
+  "compilerOptions": {
+    "composite": true,
+    "skipLibCheck": true,
+    "module": "ESNext",
+    "moduleResolution": "bundler",
+    "allowSyntheticDefaultImports": true,
+    "strict": true
+  },
+  "include": ["vite.config.ts"]
+}

examples/web-ui/vite.config.ts

@@ -0,0 +1,7 @@
+import { defineConfig } from "vite";
+import react from "@vitejs/plugin-react";
+import webfontDownload from "vite-plugin-webfont-dl";
+
+export default defineConfig({
+  plugins: [react(), webfontDownload()],
+});

src/pipecat/frames/frames.py

@@ -55,7 +55,7 @@ class ImageRawFrame(DataFrame):
     """
     image: bytes
     size: Tuple[int, int]
-    format: str
+    format: str | None
 
     def __str__(self):
         return f"{self.name}(size: {self.size}, format: {self.format})"
@@ -205,6 +205,17 @@ class ErrorFrame(SystemFrame):
     def __str__(self):
         return f"{self.name}(error: {self.error})"
 
+
+@dataclass
+class StopTaskFrame(SystemFrame):
+    """Indicates that a pipeline task should be stopped. This should inform the
+    pipeline processors that they should stop pushing frames but that they
+    should be kept in a running state.
+
+    """
+    pass
+
+
 #
 # Control frames
 #

src/pipecat/pipeline/runner.py

@@ -18,7 +18,6 @@ class PipelineRunner:
     def __init__(self, name: str | None = None, handle_sigint: bool = True):
         self.id: int = obj_id()
         self.name: str = name or f"{self.__class__.__name__}#{obj_count(self)}"
-        self._loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
 
         self._tasks = {}
         self._running = True
@@ -47,7 +46,8 @@ class PipelineRunner:
         return self._running
 
     def _setup_sigint(self):
-        self._loop.add_signal_handler(
+        loop = asyncio.get_running_loop()
+        loop.add_signal_handler(
             signal.SIGINT,
             lambda *args: asyncio.create_task(self._sigint_handler())
         )

src/pipecat/pipeline/task.py

@@ -8,7 +8,7 @@ import asyncio
 
 from typing import AsyncIterable, Iterable
 
-from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame, Frame, StartFrame
+from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame, Frame, StartFrame, StopTaskFrame
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.utils.utils import obj_count, obj_id
 
@@ -53,6 +53,8 @@ class PipelineTask:
 
     async def run(self):
         await asyncio.gather(self._process_task_queue(), self._process_up_queue())
+        await self._source.cleanup()
+        await self._pipeline.cleanup()
 
     async def queue_frame(self, frame: Frame):
         await self._task_queue.put(frame)
@@ -74,7 +76,9 @@ class PipelineTask:
             frame = await self._task_queue.get()
             await self._source.process_frame(frame, FrameDirection.DOWNSTREAM)
             self._task_queue.task_done()
-            running = not (isinstance(frame, CancelFrame) or isinstance(frame, EndFrame))
+            running = not (isinstance(frame, StopTaskFrame) or
+                           isinstance(frame, CancelFrame) or
+                           isinstance(frame, EndFrame))
         # We just enqueue None to terminate the task.
         await self._up_queue.put(None)
 

src/pipecat/processors/aggregators/openai_llm_context.py

@@ -6,23 +6,13 @@
 
 from dataclasses import dataclass
 
-from typing import AsyncGenerator, Callable, List
+from typing import List
 
-from pipecat.frames.frames import (
-    Frame,
-    LLMResponseEndFrame,
-    LLMResponseStartFrame,
-    TextFrame,
-    TranscriptionFrame,
-    UserStartedSpeakingFrame,
-    UserStoppedSpeakingFrame,
-)
-from pipecat.processors.frame_processor import FrameProcessor
+from pipecat.frames.frames import Frame
 
 from openai._types import NOT_GIVEN, NotGiven
 
 from openai.types.chat import (
-    ChatCompletionRole,
     ChatCompletionToolParam,
     ChatCompletionToolChoiceOptionParam,
     ChatCompletionMessageParam
@@ -42,7 +32,7 @@ class OpenAILLMContext:
         self.tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice
         self.tools: List[ChatCompletionToolParam] | NotGiven = tools
 
-    @ staticmethod
+    @staticmethod
     def from_messages(messages: List[dict]) -> "OpenAILLMContext":
         context = OpenAILLMContext()
         for message in messages:
@@ -71,100 +61,6 @@ class OpenAILLMContext:
         self.tools = tools
 
 
-class OpenAIContextAggregator(FrameProcessor):
-
-    def __init__(
-        self,
-        context: OpenAILLMContext,
-        aggregator: Callable[[Frame, str | None], str | None],
-        role: ChatCompletionRole,
-        start_frame: type,
-        end_frame: type,
-        accumulator_frame: type,
-        pass_through=True,
-    ):
-        if not (
-            issubclass(start_frame, Frame)
-            and issubclass(end_frame, Frame)
-            and issubclass(accumulator_frame, Frame)
-        ):
-            raise TypeError(
-                "start_frame, end_frame and accumulator_frame must be instances of Frame"
-            )
-
-        self._context: OpenAILLMContext = context
-        self._aggregator: Callable[[Frame, str | None], None] = aggregator
-        self._role: ChatCompletionRole = role
-        self._start_frame = start_frame
-        self._end_frame = end_frame
-        self._accumulator_frame = accumulator_frame
-        self._pass_through = pass_through
-
-        self._aggregating = False
-        self._aggregation = None
-
-    async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
-        if isinstance(frame, self._start_frame):
-            self._aggregating = True
-        elif isinstance(frame, self._end_frame):
-            self._aggregating = False
-            if self._aggregation:
-                self._context.add_message(
-                    {
-                        "role": self._role,
-                        "content": self._aggregation,
-                        "name": self._role,
-                    }  # type: ignore
-                )
-            self._aggregation = None
-            yield OpenAILLMContextFrame(self._context)
-        elif isinstance(frame, self._accumulator_frame) and self._aggregating:
-            self._aggregation = self._aggregator(frame, self._aggregation)
-            if self._pass_through:
-                yield frame
-        else:
-            yield frame
-
-    def string_aggregator(
-            self,
-            frame: Frame,
-            aggregation: str | None) -> str | None:
-        if not isinstance(frame, TextFrame):
-            raise TypeError(
-                "Frame must be a TextFrame instance to be aggregated by a string aggregator."
-            )
-        if not aggregation:
-            aggregation = ""
-        return " ".join([aggregation, frame.text])
-
-
-class OpenAIUserContextAggregator(OpenAIContextAggregator):
-    def __init__(self, context: OpenAILLMContext):
-        super().__init__(
-            context=context,
-            aggregator=self.string_aggregator,
-            role="user",
-            start_frame=UserStartedSpeakingFrame,
-            end_frame=UserStoppedSpeakingFrame,
-            accumulator_frame=TranscriptionFrame,
-            pass_through=False,
-        )
-
-
-class OpenAIAssistantContextAggregator(OpenAIContextAggregator):
-
-    def __init__(self, context: OpenAILLMContext):
-        super().__init__(
-            context,
-            aggregator=self.string_aggregator,
-            role="assistant",
-            start_frame=LLMResponseStartFrame,
-            end_frame=LLMResponseEndFrame,
-            accumulator_frame=TextFrame,
-            pass_through=True,
-        )
-
-
 @dataclass
 class OpenAILLMContextFrame(Frame):
     """Like an LLMMessagesFrame, but with extra context specific to the OpenAI

src/pipecat/services/ai_services.py

@@ -10,11 +10,13 @@ import math
 import wave
 
 from abc import abstractmethod
-from typing import BinaryIO
+from typing import AsyncGenerator
 
 from pipecat.frames.frames import (
     AudioRawFrame,
+    CancelFrame,
     EndFrame,
+    ErrorFrame,
     Frame,
     TextFrame,
     VisionImageRawFrame,
@@ -26,6 +28,13 @@ class AIService(FrameProcessor):
     def __init__(self):
         super().__init__()
 
+    async def process_generator(self, generator: AsyncGenerator[Frame, None]):
+        async for f in generator:
+            if isinstance(f, ErrorFrame):
+                await self.push_error(f)
+            else:
+                await self.push_frame(f)
+
 
 class LLMService(AIService):
     """This class is a no-op but serves as a base class for LLM services."""
@@ -42,7 +51,7 @@ class TTSService(AIService):
 
     # Converts the text to audio.
     @abstractmethod
-    async def run_tts(self, text: str):
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         pass
 
     async def say(self, text: str):
@@ -59,14 +68,14 @@ class TTSService(AIService):
                 self._current_sentence = ""
 
         if text:
-            await self.run_tts(text)
+            await self.process_generator(self.run_tts(text))
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, TextFrame):
             await self._process_text_frame(frame)
         elif isinstance(frame, EndFrame):
             if self._current_sentence:
-                await self.run_tts(self._current_sentence)
+                await self.process_generator(self.run_tts(self._current_sentence))
             await self.push_frame(frame)
         else:
             await self.push_frame(frame, direction)
@@ -76,64 +85,80 @@ class STTService(AIService):
     """STTService is a base class for speech-to-text services."""
 
     def __init__(self,
-                 min_rms: int = 400,
-                 max_silence_frames: int = 3,
+                 min_rms: int = 75,
+                 max_silence_secs: float = 0.3,
+                 max_buffer_secs: float = 1.5,
                  sample_rate: int = 16000,
                  num_channels: int = 1):
         super().__init__()
         self._min_rms = min_rms
-        self._max_silence_frames = max_silence_frames
+        self._max_silence_secs = max_silence_secs
+        self._max_buffer_secs = max_buffer_secs
         self._sample_rate = sample_rate
         self._num_channels = num_channels
-        self._current_silence_frames = 0
         (self._content, self._wave) = self._new_wave()
+        self._silence_num_frames = 0
+        # Exponential smoothing
+        self._smoothing_factor = 0.08
+        self._prev_rms = 1 - self._smoothing_factor
 
     @abstractmethod
-    async def run_stt(self, audio: BinaryIO):
+    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
         """Returns transcript as a string"""
         pass
 
     def _new_wave(self):
-        content = io.BufferedRandom(io.BytesIO())
+        content = io.BytesIO()
         ww = wave.open(content, "wb")
         ww.setsampwidth(2)
         ww.setnchannels(self._num_channels)
         ww.setframerate(self._sample_rate)
         return (content, ww)
 
-    def _get_volume(self, audio: bytes) -> float:
+    def _exp_smoothing(self, value: float, prev_value: float, factor: float) -> float:
+        return prev_value + factor * (value - prev_value)
+
+    def _get_smoothed_volume(self, audio: bytes, prev_rms: float, factor: float) -> float:
         # https://docs.python.org/3/library/array.html
         audio_array = array.array('h', audio)
         squares = [sample**2 for sample in audio_array]
         mean = sum(squares) / len(audio_array)
         rms = math.sqrt(mean)
-        return rms
+        return self._exp_smoothing(rms, prev_rms, factor)
+
+    async def _append_audio(self, frame: AudioRawFrame):
+        # Try to filter out empty background noise
+        # (Very rudimentary approach, can be improved)
+        rms = self._get_smoothed_volume(frame.audio, self._prev_rms, self._smoothing_factor)
+        if rms >= self._min_rms:
+            # If volume is high enough, write new data to wave file
+            self._wave.writeframes(frame.audio)
+            self._silence_num_frames = 0
+        else:
+            self._silence_num_frames += frame.num_frames
+        self._prev_rms = rms
+
+        # If buffer is not empty and we have enough data or there's been a long
+        # silence, transcribe the audio gathered so far.
+        silence_secs = self._silence_num_frames / self._sample_rate
+        buffer_secs = self._wave.getnframes() / self._sample_rate
+        if self._content.tell() > 0 and (
+                buffer_secs > self._max_buffer_secs or silence_secs > self._max_silence_secs):
+            self._silence_num_frames = 0
+            self._wave.close()
+            self._content.seek(0)
+            await self.process_generator(self.run_stt(self._content.read()))
+            (self._content, self._wave) = self._new_wave()
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Processes a frame of audio data, either buffering or transcribing it."""
-        if not isinstance(frame, AudioRawFrame):
-            await self.push_frame(frame, direction)
-            return
-
-        audio = frame.audio
-
-        # Try to filter out empty background noise
-        # (Very rudimentary approach, can be improved)
-        rms = self._get_volume(audio)
-        if rms >= self._min_rms:
-            # If volume is high enough, write new data to wave file
-            self._wave.writeframes(audio)
-
-        # If buffer is not empty and we detect a 3-frame pause in speech,
-        # transcribe the audio gathered so far.
-        if self._content.tell() > 0 and self._current_silence_frames > self._max_silence_frames:
-            self._current_silence_frames = 0
+        if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame):
             self._wave.close()
-            self._content.seek(0)
-            await self.run_stt(self._content)
-            (self._content, self._wave) = self._new_wave()
-        # If we get this far, this is a frame of silence
-        self._current_silence_frames += 1
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, AudioRawFrame):
+            await self._append_audio(frame)
+        else:
+            await self.push_frame(frame, direction)
 
 
 class ImageGenService(AIService):
@@ -142,13 +167,13 @@ class ImageGenService(AIService):
         super().__init__()
 
     # Renders the image. Returns an Image object.
-    @abstractmethod
-    async def run_image_gen(self, prompt: str):
+    @ abstractmethod
+    async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
         pass
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, TextFrame):
-            await self.run_image_gen(frame.text)
+            await self.process_generator(self.run_image_gen(frame.text))
         else:
             await self.push_frame(frame, direction)
 
@@ -160,12 +185,12 @@ class VisionService(AIService):
         super().__init__()
         self._describe_text = None
 
-    @abstractmethod
-    async def run_vision(self, frame: VisionImageRawFrame):
+    @ abstractmethod
+    async def run_vision(self, frame: VisionImageRawFrame) -> AsyncGenerator[Frame, None]:
         pass
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, VisionImageRawFrame):
-            await self.run_vision(frame)
+            await self.process_generator(self.run_vision(frame))
         else:
             await self.push_frame(frame, direction)

src/pipecat/services/azure.py

@@ -9,10 +9,11 @@ import asyncio
 import io
 
 from PIL import Image
+from typing import AsyncGenerator
 
 from openai import AsyncAzureOpenAI
 
-from pipecat.frames.frames import AudioRawFrame, ErrorFrame, URLImageRawFrame
+from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, URLImageRawFrame
 from pipecat.services.ai_services import TTSService, ImageGenService
 from pipecat.services.openai import BaseOpenAILLMService
 
@@ -34,8 +35,8 @@ except ModuleNotFoundError as e:
 
 
 class AzureTTSService(TTSService):
-    def __init__(self, *, api_key, region, voice="en-US-SaraNeural"):
-        super().__init__()
+    def __init__(self, *, api_key: str, region: str, voice="en-US-SaraNeural", **kwargs):
+        super().__init__(**kwargs)
 
         self.speech_config = SpeechConfig(subscription=api_key, region=region)
         self.speech_synthesizer = SpeechSynthesizer(
@@ -43,7 +44,7 @@ class AzureTTSService(TTSService):
         )
         self._voice = voice
 
-    async def run_tts(self, text: str):
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Transcribing text: {text}")
 
         ssml = (
@@ -60,7 +61,7 @@ class AzureTTSService(TTSService):
 
         if result.reason == ResultReason.SynthesizingAudioCompleted:
             # Azure always sends a 44-byte header. Strip it off.
-            await self.push_frame(AudioRawFrame(audio=result.audio_data[44:], sample_rate=16000, num_channels=1))
+            yield AudioRawFrame(audio=result.audio_data[44:], sample_rate=16000, num_channels=1)
         elif result.reason == ResultReason.Canceled:
             cancellation_details = result.cancellation_details
             logger.warning(f"Speech synthesis canceled: {cancellation_details.reason}")
@@ -110,7 +111,7 @@ class AzureImageGenServiceREST(ImageGenService):
         self._aiohttp_session = aiohttp_session
         self._image_size = image_size
 
-    async def run_image_gen(self, prompt: str):
+    async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
         url = f"{self._azure_endpoint}openai/images/generations:submit?api-version={self._api_version}"
 
         headers = {
@@ -136,7 +137,7 @@ class AzureImageGenServiceREST(ImageGenService):
                 attempts_left -= 1
                 if attempts_left == 0:
                     logger.error("Image generation timed out")
-                    await self.push_error(ErrorFrame("Image generation timed out"))
+                    yield ErrorFrame("Image generation timed out")
                     return
 
                 await asyncio.sleep(1)
@@ -149,7 +150,7 @@ class AzureImageGenServiceREST(ImageGenService):
             image_url = json_response["result"]["data"][0]["url"] if json_response else None
             if not image_url:
                 logger.error("Image generation failed")
-                await self.push_error(ErrorFrame("Image generation failed"))
+                yield ErrorFrame("Image generation failed")
                 return
 
             # Load the image from the url
@@ -161,4 +162,4 @@ class AzureImageGenServiceREST(ImageGenService):
                     image=image.tobytes(),
                     size=image.size,
                     format=image.format)
-                await self.push_frame(frame)
+                yield frame

src/pipecat/services/deepgram.py

@@ -4,7 +4,11 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from pipecat.frames.frames import AudioRawFrame
+import aiohttp
+
+from typing import AsyncGenerator
+
+from pipecat.frames.frames import AudioRawFrame, Frame
 from pipecat.services.ai_services import TTSService
 
 from loguru import logger
@@ -15,16 +19,17 @@ class DeepgramTTSService(TTSService):
     def __init__(
             self,
             *,
-            aiohttp_session,
-            api_key,
-            voice="alpha-asteria-en-v2"):
-        super().__init__()
+            aiohttp_session: aiohttp.ClientSession,
+            api_key: str,
+            voice: str = "alpha-asteria-en-v2",
+            **kwargs):
+        super().__init__(**kwargs)
 
         self._voice = voice
         self._api_key = api_key
         self._aiohttp_session = aiohttp_session
 
-    async def run_tts(self, text: str):
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.info(f"Running Deepgram TTS for {text}")
         base_url = "https://api.beta.deepgram.com/v1/speak"
         request_url = f"{base_url}?model={self._voice}&encoding=linear16&container=none&sample_rate=16000"
@@ -33,4 +38,4 @@ class DeepgramTTSService(TTSService):
         async with self._aiohttp_session.post(request_url, headers=headers, json=body) as r:
             async for data in r.content:
                 frame = AudioRawFrame(audio=data, sample_rate=16000, num_channels=1)
-                await self.push_frame(frame)
+                yield frame

src/pipecat/services/elevenlabs.py

@@ -6,7 +6,9 @@
 
 import aiohttp
 
-from pipecat.frames.frames import AudioRawFrame, TTSStartedFrame, TTSStoppedFrame
+from typing import AsyncGenerator
+
+from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, TTSStartedFrame, TTSStoppedFrame
 from pipecat.services.ai_services import TTSService
 
 from loguru import logger
@@ -15,21 +17,21 @@ from loguru import logger
 class ElevenLabsTTSService(TTSService):
 
     def __init__(
-        self,
-        *,
-        aiohttp_session: aiohttp.ClientSession,
-        api_key: str,
-        voice_id: str,
-        model: str = "eleven_turbo_v2",
-    ):
-        super().__init__()
+            self,
+            *,
+            aiohttp_session: aiohttp.ClientSession,
+            api_key: str,
+            voice_id: str,
+            model: str = "eleven_turbo_v2",
+            **kwargs):
+        super().__init__(**kwargs)
 
         self._api_key = api_key
         self._voice_id = voice_id
         self._aiohttp_session = aiohttp_session
         self._model = model
 
-    async def run_tts(self, text: str):
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Transcribing text: {text}")
 
         url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream"
@@ -48,11 +50,12 @@ class ElevenLabsTTSService(TTSService):
         async with self._aiohttp_session.post(url, json=payload, headers=headers, params=querystring) as r:
             if r.status != 200:
                 logger.error(f"Audio fetch status code: {r.status}, error: {r.text}")
+                yield ErrorFrame(f"Audio fetch status code: {r.status}, error: {r.text}")
                 return
 
-            await self.push_frame(TTSStartedFrame())
+            yield TTSStartedFrame()
             async for chunk in r.content:
                 if len(chunk) > 0:
                     frame = AudioRawFrame(chunk, 16000, 1)
-                    await self.push_frame(frame)
-            await self.push_frame(TTSStoppedFrame())
+                    yield frame
+            yield TTSStoppedFrame()

src/pipecat/services/fal.py

@@ -9,11 +9,10 @@ import io
 import os
 
 from PIL import Image
-from numpy import result_type
 from pydantic import BaseModel
-from typing import Optional, Union, Dict
+from typing import AsyncGenerator, Optional, Union, Dict
 
-from pipecat.frames.frames import URLImageRawFrame
+from pipecat.frames.frames import ErrorFrame, Frame, URLImageRawFrame
 from pipecat.services.ai_services import ImageGenService
 
 from loguru import logger
@@ -52,7 +51,7 @@ class FalImageGenService(ImageGenService):
         if key:
             os.environ["FAL_KEY"] = key
 
-    async def run_image_gen(self, prompt: str):
+    async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating image from prompt: {prompt}")
 
         response = await fal_client.run_async(
@@ -64,6 +63,7 @@ class FalImageGenService(ImageGenService):
 
         if not image_url:
             logger.error("Image generation failed")
+            yield ErrorFrame("Image generation failed")
             return
 
         logger.debug(f"Image generated at: {image_url}")
@@ -80,4 +80,4 @@ class FalImageGenService(ImageGenService):
                 image=image.tobytes(),
                 size=image.size,
                 format=image.format)
-            await self.push_frame(frame)
+            yield frame

src/pipecat/services/moondream.py

@@ -6,11 +6,13 @@
 
 import asyncio
 
-from pipecat.frames.frames import TextFrame, VisionImageRawFrame
-from pipecat.services.ai_services import VisionService
-
 from PIL import Image
 
+from typing import AsyncGenerator
+
+from pipecat.frames.frames import ErrorFrame, Frame, TextFrame, VisionImageRawFrame
+from pipecat.services.ai_services import VisionService
+
 from loguru import logger
 
 try:
@@ -44,7 +46,7 @@ def detect_device():
 class MoondreamService(VisionService):
     def __init__(
         self,
-        model_id="vikhyatk/moondream2",
+        model="vikhyatk/moondream2",
         revision="2024-04-02",
         use_cpu=False
     ):
@@ -56,26 +58,27 @@ class MoondreamService(VisionService):
             device = torch.device("cpu")
             dtype = torch.float32
 
-        self._tokenizer = AutoTokenizer.from_pretrained(model_id, revision=revision)
+        self._tokenizer = AutoTokenizer.from_pretrained(model, revision=revision)
 
         logger.debug("Loading Moondream model...")
 
         self._model = AutoModelForCausalLM.from_pretrained(
-            model_id, trust_remote_code=True, revision=revision
+            model, trust_remote_code=True, revision=revision
         ).to(device=device, dtype=dtype)
         self._model.eval()
 
         logger.debug("Loaded Moondream model")
 
-    async def run_vision(self, frame: VisionImageRawFrame):
+    async def run_vision(self, frame: VisionImageRawFrame) -> AsyncGenerator[Frame, None]:
         if not self._model:
             logger.error("Moondream model not available")
+            yield ErrorFrame("Moondream model not available")
             return
 
         logger.debug(f"Analyzing image: {frame}")
 
         def get_image_description(frame: VisionImageRawFrame):
-            image = Image.frombytes(frame.format, (frame.size[0], frame.size[1]), frame.image)
+            image = Image.frombytes(frame.format, frame.size, frame.image)
             image_embeds = self._model.encode_image(image)
             description = self._model.answer_question(
                 image_embeds=image_embeds,
@@ -85,4 +88,4 @@ class MoondreamService(VisionService):
 
         description = await asyncio.to_thread(get_image_description, frame)
 
-        await self.push_frame(TextFrame(text=description))
+        yield TextFrame(text=description)

src/pipecat/services/openai.py

@@ -8,11 +8,13 @@ import io
 import json
 import time
 import aiohttp
+
 from PIL import Image
 
-from typing import List, Literal
+from typing import AsyncGenerator, List, Literal
 
 from pipecat.frames.frames import (
+    ErrorFrame,
     Frame,
     LLMMessagesFrame,
     LLMResponseEndFrame,
@@ -174,7 +176,7 @@ class OpenAIImageGenService(ImageGenService):
         self._client = AsyncOpenAI(api_key=api_key)
         self._aiohttp_session = aiohttp_session
 
-    async def run_image_gen(self, prompt: str):
+    async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating image from prompt: {prompt}")
 
         image = await self._client.images.generate(
@@ -187,11 +189,13 @@ class OpenAIImageGenService(ImageGenService):
         image_url = image.data[0].url
 
         if not image_url:
-            logger.error(f"no image provided in response: {image}")
+            logger.error(f"No image provided in response: {image}")
+            yield ErrorFrame("Image generation failed")
+            return
 
         # Load the image from the url
         async with self._aiohttp_session.get(image_url) as response:
             image_stream = io.BytesIO(await response.content.read())
             image = Image.open(image_stream)
             frame = URLImageRawFrame(image_url, image.tobytes(), image.size, image.format)
-            await self.push_frame(frame)
+            yield frame

src/pipecat/services/playht.py

@@ -7,7 +7,9 @@
 import io
 import struct
 
-from pipecat.frames.frames import AudioRawFrame
+from typing import AsyncGenerator
+
+from pipecat.frames.frames import AudioRawFrame, Frame
 from pipecat.services.ai_services import TTSService
 
 from loguru import logger
@@ -25,8 +27,8 @@ except ModuleNotFoundError as e:
 
 class PlayHTAIService(TTSService):
 
-    def __init__(self, *, api_key, user_id, voice_url):
-        super().__init__()
+    def __init__(self, *, api_key: str, user_id: str, voice_url: str, **kwargs):
+        super().__init__(**kwargs)
 
         self._user_id = user_id
         self._speech_key = api_key
@@ -44,7 +46,7 @@ class PlayHTAIService(TTSService):
     def __del__(self):
         self._client.close()
 
-    async def run_tts(self, text: str):
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         b = bytearray()
         in_header = True
         for chunk in self._client.tts(text, self._options):
@@ -69,4 +71,4 @@ class PlayHTAIService(TTSService):
             else:
                 if len(chunk):
                     frame = AudioRawFrame(chunk, 16000, 1)
-                    await self.push_frame(frame)
+                    yield frame

src/pipecat/services/whisper.py

@@ -10,9 +10,11 @@ import asyncio
 import time
 
 from enum import Enum
-from typing import BinaryIO
+from typing_extensions import AsyncGenerator
 
-from pipecat.frames.frames import TranscriptionFrame
+import numpy as np
+
+from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
 from pipecat.services.ai_services import STTService
 
 from loguru import logger
@@ -39,14 +41,18 @@ class Model(Enum):
 class WhisperSTTService(STTService):
     """Class to transcribe audio with a locally-downloaded Whisper model"""
 
-    def __init__(self, model_name: Model = Model.DISTIL_MEDIUM_EN,
+    def __init__(self,
+                 model: Model = Model.DISTIL_MEDIUM_EN,
                  device: str = "auto",
-                 compute_type: str = "default"):
+                 compute_type: str = "default",
+                 no_speech_prob: float = 0.1,
+                 **kwargs):
 
-        super().__init__()
+        super().__init__(**kwargs)
         self._device: str = device
         self._compute_type = compute_type
-        self._model_name: Model = model_name
+        self._model_name: Model = model
+        self._no_speech_prob = no_speech_prob
         self._model: WhisperModel | None = None
         self._load()
 
@@ -60,15 +66,21 @@ class WhisperSTTService(STTService):
             compute_type=self._compute_type)
         logger.debug("Loaded Whisper model")
 
-    async def run_stt(self, audio: BinaryIO):
+    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
         """Transcribes given audio using Whisper"""
         if not self._model:
+            yield ErrorFrame("Whisper model not available")
             logger.error("Whisper model not available")
             return
 
-        segments, _ = await asyncio.to_thread(self._model.transcribe, audio)
+        # Divide by 32768 because we have signed 16-bit data.
+        audio_float = np.frombuffer(audio, dtype=np.int16).astype(np.float32) / 32768.0
+
+        segments, _ = await asyncio.to_thread(self._model.transcribe, audio_float)
         text: str = ""
         for segment in segments:
-            text += f"{segment.text} "
+            if segment.no_speech_prob < self._no_speech_prob:
+                text += f"{segment.text} "
 
-        await self.push_frame(TranscriptionFrame(text, "", int(time.time_ns() / 1000000)))
+        if text:
+            yield TranscriptionFrame(text, "", int(time.time_ns() / 1000000))

src/pipecat/transports/base_input.py

@@ -6,7 +6,6 @@
 
 import asyncio
 import queue
-import threading
 
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.frames.frames import (
@@ -30,26 +29,34 @@ class BaseInputTransport(FrameProcessor):
 
         self._params = params
 
-        self._running = True
+        self._running = False
 
         # Start media threads.
         if self._params.audio_in_enabled or self._params.vad_enabled:
             self._audio_in_queue = queue.Queue()
-            self._audio_in_thread = threading.Thread(target=self._audio_in_thread_handler)
-            self._audio_out_thread = threading.Thread(target=self._audio_out_thread_handler)
-
-        self._stopped_event = asyncio.Event()
 
     async def start(self):
+        if self._running:
+            return
+
+        self._running = True
+
         if self._params.audio_in_enabled or self._params.vad_enabled:
-            self._audio_in_thread.start()
-            self._audio_out_thread.start()
+            loop = self.get_event_loop()
+            self._audio_in_thread = loop.run_in_executor(None, self._audio_in_thread_handler)
+            self._audio_out_thread = loop.run_in_executor(None, self._audio_out_thread_handler)
 
     async def stop(self):
+        if not self._running:
+            return
+
         # This will exit all threads.
         self._running = False
 
-        self._stopped_event.set()
+        # Wait for the threads to finish.
+        if self._params.audio_in_enabled or self._params.vad_enabled:
+            await self._audio_in_thread
+            await self._audio_out_thread
 
     def vad_analyze(self, audio_frames: bytes) -> VADState:
         pass
@@ -62,25 +69,18 @@ class BaseInputTransport(FrameProcessor):
     #
 
     async def cleanup(self):
-        if self._params.audio_in_enabled or self._params.vad_enabled:
-            self._audio_in_thread.join()
-            self._audio_out_thread.join()
+        pass
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, StartFrame):
-            await self.push_frame(frame, direction)
             await self.start()
-        elif isinstance(frame, CancelFrame) or isinstance(frame, EndFrame):
             await self.push_frame(frame, direction)
+        elif isinstance(frame, CancelFrame) or isinstance(frame, EndFrame):
             await self.stop()
+            await self.push_frame(frame, direction)
         else:
             await self.push_frame(frame, direction)
 
-        # If we are finishing, wait here until we have stopped, otherwise we
-        # might close things too early upstream.
-        if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame):
-            await self._stopped_event.wait()
-
     #
     # Audio input
     #

src/pipecat/transports/base_output.py

@@ -7,10 +7,11 @@
 
 import asyncio
 import itertools
+from multiprocessing.context import _force_start_method
 import queue
-import threading
 import time
 
+from PIL import Image
 from typing import List
 
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -35,27 +36,36 @@ class BaseOutputTransport(FrameProcessor):
 
         self._params = params
 
-        self._running = True
+        self._running = False
 
         # These are the images that we should send to the camera at our desired
         # framerate.
         self._camera_images = None
 
-        # Start media threads.
+        # Create media threads queues.
         if self._params.camera_out_enabled:
             self._camera_out_queue = queue.Queue()
-            self._camera_out_thread = threading.Thread(target=self._camera_out_thread_handler)
-            self._camera_out_thread.start()
-
         self._sink_queue = queue.Queue()
-        self._sink_thread = threading.Thread(target=self._sink_thread_handler)
 
         self._stopped_event = asyncio.Event()
 
     async def start(self):
-        self._sink_thread.start()
+        if self._running:
+            return
+
+        self._running = True
+
+        loop = self.get_event_loop()
+
+        if self._params.camera_out_enabled:
+            self._camera_out_thread = loop.run_in_executor(None, self._camera_out_thread_handler)
+
+        self._sink_thread = loop.run_in_executor(None, self._sink_thread_handler)
 
     async def stop(self):
+        if not self._running:
+            return
+
         # This will exit all threads.
         self._running = False
 
@@ -75,26 +85,28 @@ class BaseOutputTransport(FrameProcessor):
     #
 
     async def cleanup(self):
+        # Wait on the threads to finish.
         if self._params.camera_out_enabled:
-            self._camera_out_thread.join()
+            await self._camera_out_thread
 
-        self._sink_thread.join()
+        await self._sink_thread
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, StartFrame):
-            await self.push_frame(frame, direction)
             await self.start()
+            await self.push_frame(frame, direction)
         # 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 self._frame_managed_by_sink(frame):
             self._sink_queue.put(frame)
         else:
             await self.push_frame(frame, direction)
 
         # If we are finishing, wait here until we have stopped, otherwise we might
-        # close things too early upstream.
+        # close things too early upstream. We need this event because we don't
+        # know when the internal threads will finish.
         if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame):
             await self._stopped_event.wait()
 
@@ -103,7 +115,6 @@ class BaseOutputTransport(FrameProcessor):
                 or isinstance(frame, ImageRawFrame)
                 or isinstance(frame, SpriteFrame)
                 or isinstance(frame, TransportMessageFrame)
-                or isinstance(frame, CancelFrame)
                 or isinstance(frame, EndFrame))
 
     def _sink_thread_handler(self):
@@ -113,7 +124,7 @@ class BaseOutputTransport(FrameProcessor):
         while self._running:
             try:
                 frame = self._sink_queue.get(timeout=1)
-                if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame):
+                if isinstance(frame, EndFrame):
                     # Send all remaining audio before stopping (multiple of 10ms of audio).
                     self._send_audio_truncated(buffer, bytes_size_10ms)
                     future = asyncio.run_coroutine_threadsafe(self.stop(), self.get_event_loop())
@@ -140,15 +151,17 @@ class BaseOutputTransport(FrameProcessor):
     async def send_image(self, frame: ImageRawFrame | SpriteFrame):
         await self.process_frame(frame, FrameDirection.DOWNSTREAM)
 
-    def _draw_image(self, image: ImageRawFrame):
+    def _draw_image(self, frame: ImageRawFrame):
         desired_size = (self._params.camera_out_width, self._params.camera_out_height)
 
-        if image.size != desired_size:
+        if frame.size != desired_size:
+            image = Image.frombytes(frame.format, frame.size, frame.image)
+            resized_image = image.resize(desired_size)
             logger.warning(
-                f"{image} does not have the expected size {desired_size}, ignoring")
-            return
+                f"{frame} does not have the expected size {desired_size}, resizing")
+            frame = ImageRawFrame(resized_image.tobytes(), resized_image.size, resized_image.format)
 
-        self.write_frame_to_camera(image)
+        self.write_frame_to_camera(frame)
 
     def _set_camera_image(self, image: ImageRawFrame):
         if self._params.camera_out_is_live:

src/pipecat/transports/local/audio.py

@@ -37,6 +37,14 @@ class LocalAudioInputTransport(BaseInputTransport):
     def read_raw_audio_frames(self, frame_count: int) -> bytes:
         return self._in_stream.read(frame_count, exception_on_overflow=False)
 
+    async def start(self):
+        await super().start()
+        self._in_stream.start_stream()
+
+    async def stop(self):
+        await super().stop()
+        self._in_stream.stop_stream()
+
     async def cleanup(self):
         # This is not very pretty (taken from PyAudio docs).
         while self._in_stream.is_active():
@@ -60,6 +68,14 @@ class LocalAudioOutputTransport(BaseOutputTransport):
     def write_raw_audio_frames(self, frames: bytes):
         self._out_stream.write(frames)
 
+    async def start(self):
+        await super().start()
+        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():

src/pipecat/transports/local/tk.py

@@ -48,6 +48,14 @@ class TkInputTransport(BaseInputTransport):
     def read_raw_audio_frames(self, frame_count: int) -> bytes:
         return self._in_stream.read(frame_count, exception_on_overflow=False)
 
+    async def start(self):
+        await super().start()
+        self._in_stream.start_stream()
+
+    async def stop(self):
+        await super().stop()
+        self._in_stream.stop_stream()
+
     async def cleanup(self):
         # This is not very pretty (taken from PyAudio docs).
         while self._in_stream.is_active():
@@ -79,7 +87,17 @@ class TkOutputTransport(BaseOutputTransport):
         self._out_stream.write(frames)
 
     def write_frame_to_camera(self, frame: ImageRawFrame):
-        asyncio.run_coroutine_threadsafe(self._write_frame_to_tk(frame), self.get_event_loop())
+        future = asyncio.run_coroutine_threadsafe(
+            self._write_frame_to_tk(frame), self.get_event_loop())
+        future.result()
+
+    async def start(self):
+        await super().start()
+        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).

src/pipecat/transports/services/daily.py

@@ -7,7 +7,6 @@
 import asyncio
 import inspect
 import queue
-import threading
 import time
 import types
 
@@ -80,32 +79,36 @@ class WebRTCVADAnalyzer(VADAnalyzer):
         return confidence
 
 
+class DailyTranscriptionSettings(BaseModel):
+    language: str = "en"
+    tier: str = "nova"
+    model: str = "2-conversationalai"
+    profanity_filter: bool = True
+    redact: bool = False
+    endpointing: bool = True
+    punctuate: bool = True
+    includeRawResponse: bool = True
+    extra: Mapping[str, Any] = {
+        "interim_results": True
+    }
+
+
 class DailyParams(TransportParams):
     transcription_enabled: bool = False
-    transcription_settings: Mapping[str, Any] = {
-        "language": "en",
-        "tier": "nova",
-        "model": "2-conversationalai",
-        "profanity_filter": True,
-        "redact": False,
-        "endpointing": True,
-        "punctuate": True,
-        "includeRawResponse": True,
-        "extra": {
-            "interim_results": True,
-        }
-    }
+    transcription_settings: DailyTranscriptionSettings = DailyTranscriptionSettings()
 
 
 class DailyCallbacks(BaseModel):
     on_joined: Callable[[Mapping[str, Any]], None]
     on_left: Callable[[], None]
     on_participant_joined: Callable[[Mapping[str, Any]], None]
+    on_participant_left: Callable[[Mapping[str, Any], str], None]
     on_first_participant_joined: Callable[[Mapping[str, Any]], None]
+    on_app_message: Callable[[Any, str], None]
     on_error: Callable[[str], None]
 
 
-class DailySession(EventHandler):
+class DailyTransportClient(EventHandler):
 
     _daily_initialized: bool = False
 
@@ -163,7 +166,7 @@ class DailySession(EventHandler):
                 sample_rate=self._params.audio_in_sample_rate,
                 num_channels=self._params.audio_in_channels)
 
-    @ property
+    @property
     def participant_id(self) -> str:
         return self._participant_id
 
@@ -180,7 +183,14 @@ class DailySession(EventHandler):
         self._client.send_app_message(frame.message, frame.participant_id)
 
     def read_raw_audio_frames(self, frame_count: int) -> bytes:
-        return self._speaker.read_frames(frame_count)
+        if self._other_participant_has_joined:
+            return self._speaker.read_frames(frame_count)
+        else:
+            # If no one has ever joined the meeting `read_frames()` would block,
+            # instead we just wait a bit. daily-python should probably return
+            # silence instead.
+            time.sleep(0.01)
+            return b''
 
     def write_raw_audio_frames(self, frames: bytes):
         self._mic.write_frames(frames)
@@ -265,7 +275,8 @@ class DailySession(EventHandler):
                 if self._token and self._params.transcription_enabled:
                     logger.info(
                         f"Enabling transcription with settings {self._params.transcription_settings}")
-                    self._client.start_transcription(self._params.transcription_settings)
+                    self._client.start_transcription(
+                        self._params.transcription_settings.model_dump())
 
                 self._callbacks.on_joined(data["participants"]["local"])
             else:
@@ -366,6 +377,12 @@ class DailySession(EventHandler):
 
         self._callbacks.on_participant_joined(participant)
 
+    def on_participant_left(self, participant, reason):
+        id = participant["id"]
+        logger.info(f"Participant left {id}")
+
+        self._callbacks.on_participant_left(participant, reason)
+
     def on_transcription_message(self, message: Mapping[str, Any]):
         participant_id = ""
         if "participantId" in message:
@@ -384,6 +401,10 @@ class DailySession(EventHandler):
     def on_transcription_stopped(self, stopped_by, stopped_by_error):
         logger.debug("Transcription stopped")
 
+    def on_app_message(self, message: Any, sender: str):
+        self._callbacks.on_app_message(message, sender)
+
+    #
     # Daily (CallClient callbacks)
     #
 
@@ -403,36 +424,44 @@ class DailySession(EventHandler):
 
 class DailyInputTransport(BaseInputTransport):
 
-    def __init__(self, session: DailySession, params: DailyParams):
+    def __init__(self, client: DailyTransportClient, params: DailyParams):
         super().__init__(params)
 
-        self._session = session
+        self._client = client
 
         self._video_renderers = {}
         self._camera_in_queue = queue.Queue()
-        self._camera_in_thread = threading.Thread(target=self._camera_in_thread_handler)
-        self._camera_in_thread.start()
 
     async def start(self):
-        await self._session.join()
+        if self._running:
+            return
+        # Join the room.
+        await self._client.join()
+        # This will set _running=True
         await super().start()
+        # Create camera in thread (runs if _running is true).
+        loop = asyncio.get_running_loop()
+        self._camera_in_thread = loop.run_in_executor(None, self._camera_in_thread_handler)
 
     async def stop(self):
-        await self._session.leave()
+        if not self._running:
+            return
+        # Leave the room.
+        await self._client.leave()
+        # This will set _running=False
         await super().stop()
+        # The thread will stop.
+        await self._camera_in_thread
 
     async def cleanup(self):
-        self._camera_in_thread.join()
-
-        await self._session.cleanup()
-
         await super().cleanup()
+        await self._client.cleanup()
 
     def vad_analyze(self, audio_frames: bytes) -> VADState:
-        return self._session.vad_analyze(audio_frames)
+        return self._client.vad_analyze(audio_frames)
 
     def read_raw_audio_frames(self, frame_count: int) -> bytes:
-        return self._session.read_raw_audio_frames(frame_count)
+        return self._client.read_raw_audio_frames(frame_count)
 
     #
     # FrameProcessor
@@ -445,23 +474,15 @@ class DailyInputTransport(BaseInputTransport):
         await super().process_frame(frame, direction)
 
     #
-    # Transcription
+    # Frames
     #
 
-    def capture_participant_transcription(self, participant_id: str):
-        self._session.capture_participant_transcription(
-            participant_id,
-            self._on_transcription_message
-        )
+    def push_transcription_frame(self, frame: TranscriptionFrame | InterimTranscriptionFrame):
+        future = asyncio.run_coroutine_threadsafe(self.push_frame(frame), self.get_event_loop())
+        future.result()
 
-    def _on_transcription_message(self, participant_id, message):
-        text = message["text"]
-        timestamp = message["timestamp"]
-        is_final = message["rawResponse"]["is_final"]
-        if is_final:
-            frame = TranscriptionFrame(text, participant_id, timestamp)
-        else:
-            frame = InterimTranscriptionFrame(text, participant_id, timestamp)
+    def push_app_message(self, message: Any, sender: str):
+        frame = DailyTransportMessageFrame(message=message, participant_id=sender)
         future = asyncio.run_coroutine_threadsafe(self.push_frame(frame), self.get_event_loop())
         future.result()
 
@@ -481,7 +502,7 @@ class DailyInputTransport(BaseInputTransport):
             "render_next_frame": False,
         }
 
-        self._session.capture_participant_video(
+        self._client.capture_participant_video(
             participant_id,
             self._on_participant_video_frame,
             framerate,
@@ -532,28 +553,36 @@ class DailyInputTransport(BaseInputTransport):
 
 class DailyOutputTransport(BaseOutputTransport):
 
-    def __init__(self, session: DailySession, params: DailyParams):
+    def __init__(self, client: DailyTransportClient, params: DailyParams):
         super().__init__(params)
 
-        self._session = session
+        self._client = client
 
     async def start(self):
-        await self._session.join()
+        if self._running:
+            return
+        # This will set _running=True
         await super().start()
+        # Join the room.
+        await self._client.join()
 
     async def stop(self):
-        await self._session.leave()
+        if not self._running:
+            return
+        # This will set _running=False
         await super().stop()
+        # Leave the room.
+        await self._client.leave()
 
     async def cleanup(self):
-        await self._session.cleanup()
         await super().cleanup()
+        await self._client.cleanup()
 
     def write_raw_audio_frames(self, frames: bytes):
-        self._session.write_raw_audio_frames(frames)
+        self._client.write_raw_audio_frames(frames)
 
     def write_frame_to_camera(self, frame: ImageRawFrame):
-        self._session.write_frame_to_camera(frame)
+        self._client.write_frame_to_camera(frame)
 
 
 class DailyTransport(BaseTransport):
@@ -564,11 +593,13 @@ class DailyTransport(BaseTransport):
             on_left=self._on_left,
             on_first_participant_joined=self._on_first_participant_joined,
             on_participant_joined=self._on_participant_joined,
+            on_participant_left=self._on_participant_left,
+            on_app_message=self._on_app_message,
             on_error=self._on_error,
         )
         self._params = params
 
-        self._session = DailySession(room_url, token, bot_name, params, callbacks)
+        self._client = DailyTransportClient(room_url, token, bot_name, params, callbacks)
         self._input: DailyInputTransport | None = None
         self._output: DailyOutputTransport | None = None
         self._loop = asyncio.get_running_loop()
@@ -580,6 +611,7 @@ class DailyTransport(BaseTransport):
         self._register_event_handler("on_joined")
         self._register_event_handler("on_left")
         self._register_event_handler("on_participant_joined")
+        self._register_event_handler("on_participant_left")
         self._register_event_handler("on_first_participant_joined")
 
     #
@@ -588,12 +620,12 @@ class DailyTransport(BaseTransport):
 
     def input(self) -> FrameProcessor:
         if not self._input:
-            self._input = DailyInputTransport(self._session, self._params)
+            self._input = DailyInputTransport(self._client, self._params)
         return self._input
 
     def output(self) -> FrameProcessor:
         if not self._output:
-            self._output = DailyOutputTransport(self._session, self._params)
+            self._output = DailyOutputTransport(self._client, self._params)
         return self._output
 
     #
@@ -602,7 +634,7 @@ class DailyTransport(BaseTransport):
 
     @property
     def participant_id(self) -> str:
-        return self._session.participant_id
+        return self._client.participant_id
 
     async def send_image(self, frame: ImageRawFrame | SpriteFrame):
         if self._output:
@@ -613,8 +645,10 @@ class DailyTransport(BaseTransport):
             await self._output.process_frame(frame, FrameDirection.DOWNSTREAM)
 
     def capture_participant_transcription(self, participant_id: str):
-        if self._input:
-            self._input.capture_participant_transcription(participant_id)
+        self._client.capture_participant_transcription(
+            participant_id,
+            self._on_transcription_message
+        )
 
     def capture_participant_video(
             self,
@@ -634,15 +668,34 @@ class DailyTransport(BaseTransport):
 
     def _on_error(self, error):
         # TODO(aleix): Report error to input/output transports. The one managing
-        # the session should report the error.
+        # the client should report the error.
         pass
 
     def _on_participant_joined(self, participant):
         self.on_participant_joined(participant)
 
+    def _on_participant_left(self, participant, reason):
+        self.on_participant_left(participant, reason)
+
     def _on_first_participant_joined(self, participant):
         self.on_first_participant_joined(participant)
 
+    def _on_app_message(self, message: Any, sender: str):
+        if self._input:
+            self._input.push_app_message(message, sender)
+
+    def _on_transcription_message(self, participant_id, message):
+        text = message["text"]
+        timestamp = message["timestamp"]
+        is_final = message["rawResponse"]["is_final"]
+        if is_final:
+            frame = TranscriptionFrame(text, participant_id, timestamp)
+        else:
+            frame = InterimTranscriptionFrame(text, participant_id, timestamp)
+
+        if self._input:
+            self._input.push_transcription_frame(frame)
+
     #
     # Decorators (event handlers)
     #
@@ -656,6 +709,9 @@ class DailyTransport(BaseTransport):
     def on_participant_joined(self, participant):
         pass
 
+    def on_participant_left(self, participant, reason):
+        pass
+
     def on_first_participant_joined(self, participant):
         pass
 
@@ -698,46 +754,8 @@ class DailyTransport(BaseTransport):
             logger.error(f"Exception in event handler {event_name}: {e}")
             raise e
 
-    #     def send_app_message(self, message: Any, participant_id: str | None):
-    #         self.client.send_app_message(message, participant_id)
-
-    #     def process_interrupt_handler(self, signum, frame):
-    #         self._post_run()
-    #         if callable(self.original_sigint_handler):
-    #             self.original_sigint_handler(signum, frame)
-
-    #     def _post_run(self):
-    #         self.client.leave()
-    #         self.client.release()
-
-    #     def on_first_other_participant_joined(self, participant):
-    #         pass
-
-    #     def call_joined(self, join_data, client_error):
-    #         # self._logger.info(f"Call_joined: {join_data}, {client_error}")
-    #         pass
-
     #     def dialout(self, number):
     #         self.client.start_dialout({"phoneNumber": number})
 
     #     def start_recording(self):
     #         self.client.start_recording()
-
-    #     def on_error(self, error):
-    #         self._logger.error(f"on_error: {error}")
-
-    #     def on_participant_joined(self, participant):
-    #         if not self._other_participant_has_joined and participant["id"] != self._my_participant_id:
-    #             self._other_participant_has_joined = True
-    #             self.on_first_other_participant_joined(participant)
-
-    #     def on_participant_left(self, participant, reason):
-    #         if len(self.client.participants()) < self._min_others_count + 1:
-    #             self._stop_threads.set()
-
-    #     def on_app_message(self, message: Any, sender: str):
-    #         if self._loop:
-    #             frame = ReceivedAppMessageFrame(message, sender)
-    #             asyncio.run_coroutine_threadsafe(
-    #                 self.receive_queue.put(frame), self._loop
-    #             )

src/pipecat/vad/vad_analyzer.py

@@ -19,22 +19,22 @@ class VADAnalyzer:
 
     def __init__(
             self,
-            sample_rate,
-            num_channels,
-            vad_confidence=0.5,
-            vad_start_s=0.2,
-            vad_stop_s=0.8):
+            sample_rate: int,
+            num_channels: int,
+            vad_confidence: float = 0.5,
+            vad_start_secs: float = 0.2,
+            vad_stop_secs: float = 0.8):
         self._sample_rate = sample_rate
         self._vad_confidence = vad_confidence
-        self._vad_start_s = vad_start_s
-        self._vad_stop_s = vad_stop_s
+        self._vad_start_secs = vad_start_secs
+        self._vad_stop_secs = vad_stop_secs
         self._vad_frames = self.num_frames_required()
         self._vad_frames_num_bytes = self._vad_frames * num_channels * 2
 
-        vad_frame_s = self._vad_frames / self._sample_rate
+        vad_frames_per_sec = self._vad_frames / self._sample_rate
 
-        self._vad_start_frames = round(self._vad_start_s / vad_frame_s)
-        self._vad_stop_frames = round(self._vad_stop_s / vad_frame_s)
+        self._vad_start_frames = round(self._vad_start_secs / vad_frames_per_sec)
+        self._vad_stop_frames = round(self._vad_stop_secs / vad_frames_per_sec)
         self._vad_starting_count = 0
         self._vad_stopping_count = 0
         self._vad_state: VADState = VADState.QUIET