Fix missing await in simple chatbot example

Merge pull request #716 from pipecat-ai/mb/mute-stt-service
Add STTMuteFilter to un/mute the STT
2024-11-15 21:24:07 -05:00 · 2024-11-14 19:55:00 -05:00 · 2024-11-14 19:47:37 -05:00 · 2024-11-14 17:11:51 -05:00 · 2024-11-14 17:05:08 -05:00 · 2024-11-14 13:47:08 -05:00
34 changed files with 2679 additions and 630 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,21 @@ 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).

+## Unreleased
+
+### Added
+
+- Added `RimeHttpTTSService` and the `07q-interruptible-rime.py` foundational
+  example.
+- Added `STTMuteFilter`, a general-purpose processor that combines STT
+  muting and interruption control. When active, it prevents both transcription
+  and interruptions during bot speech. The processor supports multiple
+  strategies: `FIRST_SPEECH` (mute only during bot's first
+  speech), `ALWAYS` (mute during all bot speech), or `CUSTOM` (using provided
+  callback).
+- Added `STTMuteFrame`, a control frame that enables/disables speech
+  transcription in STT services.
+
 ## [0.0.48] - 2024-11-10 "Antonio release"

 ### Added
--- a/docs/CONTRIBUTING.md
+++ b/docs/CONTRIBUTING.md
--- a/README.md
+++ b/README.md
@@ -1,14 +1,20 @@
-<div align="center">
+<h1><div align="center">
  <img alt="pipecat" width="300px" height="auto" src="https://raw.githubusercontent.com/pipecat-ai/pipecat/main/pipecat.png">
-</div>
-
-# Pipecat
+</div></h1>

 [![PyPI](https://img.shields.io/pypi/v/pipecat-ai)](https://pypi.org/project/pipecat-ai) [![Discord](https://img.shields.io/discord/1239284677165056021)](https://discord.gg/pipecat) <a href="https://app.commanddash.io/agent/github_pipecat-ai_pipecat"><img src="https://img.shields.io/badge/AI-Code%20Agent-EB9FDA"></a>

-`pipecat` is a framework for building voice (and multimodal) conversational agents. Things like personal coaches, meeting assistants, [story-telling toys for kids](https://storytelling-chatbot.fly.dev/), customer support bots, [intake flows](https://www.youtube.com/watch?v=lDevgsp9vn0), and snarky social companions.
+Pipecat is an open source Python framework for building voice and multimodal conversational agents. It handles the complex orchestration of AI services, network transport, audio processing, and multimodal interactions, letting you focus on creating engaging experiences.

-Take a look at some example apps:
+## What you can build
+
+- **Voice Assistants**: [Natural, real-time conversations with AI](https://demo.dailybots.ai/)
+- **Interactive Agents**: Personal coaches and meeting assistants
+- **Multimodal Apps**: Combine voice, video, images, and text
+- **Creative Tools**: [Story-telling experiences](https://storytelling-chatbot.fly.dev/) and social companions
+- **Business Solutions**: [Customer intake flows](https://www.youtube.com/watch?v=lDevgsp9vn0) and support bots
+
+## See it in action

 <p float="left">
    <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;
@@ -18,33 +24,52 @@ Take a look at some example apps:
    <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
+## Key features
+
+- **Voice-first Design**: Built-in speech recognition, TTS, and conversation handling
+- **Flexible Integration**: Works with popular AI services (OpenAI, ElevenLabs, etc.)
+- **Pipeline Architecture**: Build complex apps from simple, reusable components
+- **Real-time Processing**: Frame-based pipeline architecture for fluid interactions
+- **Production Ready**: Enterprise-grade WebRTC and Websocket support
+
+## Getting started

 You can get started with Pipecat running on your local machine, then move your agent processes to the cloud when you’re ready. You can also add a 📞 telephone number, 🖼️ image output, 📺 video input, use different LLMs, and more.

 ```shell
-# install the module
+# Install the module
 pip install pipecat-ai

-# set up an .env file with API keys
+# Set up your environment
 cp dot-env.template .env
 ```

-By default, in order to minimize dependencies, only the basic framework functionality is available. Some third-party AI services require additional dependencies that you can install with:
+To keep things lightweight, only the core framework is included by default. If you need support for third-party AI services, you can add the necessary dependencies with:

 ```shell
 pip install "pipecat-ai[option,...]"
 ```

-Your project may or may not need these, so they're made available as optional requirements. Here is a list:
+Available options include:

- **AI services**: `anthropic`, `assemblyai`, `aws`, `azure`, `deepgram`, `gladia`, `google`, `fal`, `lmnt`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper`, `xtts`
- **Transports**: `local`, `websocket`, `daily`
+| Category            | Services                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          | Install Command Example               |
+| ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------- |
+| Speech-to-Text      | [AssemblyAI](https://docs.pipecat.ai/api-reference/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/api-reference/services/stt/azure), [Deepgram](https://docs.pipecat.ai/api-reference/services/stt/deepgram), [Gladia](https://docs.pipecat.ai/api-reference/services/stt/gladia), [Whisper](https://docs.pipecat.ai/api-reference/services/stt/whisper)                                                                                                                                                                                                                                                                                                                                                                                                               | `pip install "pipecat-ai[deepgram]"`  |
+| LLMs                | [Anthropic](https://docs.pipecat.ai/api-reference/services/llm/anthropic), [Azure](https://docs.pipecat.ai/api-reference/services/llm/azure), [Fireworks AI](https://docs.pipecat.ai/api-reference/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/api-reference/services/llm/gemini), [Ollama](https://docs.pipecat.ai/api-reference/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/api-reference/services/llm/openai), [Together AI](https://docs.pipecat.ai/api-reference/services/llm/together)                                                                                                                                                                                                                                                             | `pip install "pipecat-ai[openai]"`    |
+| Text-to-Speech      | [AWS](https://docs.pipecat.ai/api-reference/services/tts/aws), [Azure](https://docs.pipecat.ai/api-reference/services/tts/azure), [Cartesia](https://docs.pipecat.ai/api-reference/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/api-reference/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/api-reference/services/tts/elevenlabs), [Google](https://docs.pipecat.ai/api-reference/services/tts/google), [LMNT](https://docs.pipecat.ai/api-reference/services/tts/lmnt), [OpenAI](https://docs.pipecat.ai/api-reference/services/tts/openai), [PlayHT](https://docs.pipecat.ai/api-reference/services/tts/playht), [Rime](https://docs.pipecat.ai/api-reference/services/tts/rime), [XTTS](https://docs.pipecat.ai/api-reference/services/tts/xtts) | `pip install "pipecat-ai[cartesia]"`  |
+| Speech-to-Speech    | [OpenAI Realtime](https://docs.pipecat.ai/api-reference/services/s2s/openai)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      | `pip install "pipecat-ai[openai]"`    |
+| Transport           | [Daily (WebRTC)](https://docs.pipecat.ai/api-reference/services/transport/daily), WebSocket, Local                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                | `pip install "pipecat-ai[daily]"`     |
+| Video               | [Tavus](https://docs.pipecat.ai/api-reference/services/video/tavus)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               | `pip install "pipecat-ai[tavus]"`     |
+| Vision & Image      | [Moondream](https://docs.pipecat.ai/api-reference/services/vision/moondream), [fal](https://docs.pipecat.ai/api-reference/services/image-generation/fal)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          | `pip install "pipecat-ai[moondream]"` |
+| Audio Processing    | [Silero VAD](https://docs.pipecat.ai/api-reference/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/api-reference/utilities/audio/krisp-filter), [Noisereduce](https://docs.pipecat.ai/api-reference/utilities/audio/noisereduce-filter)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     | `pip install "pipecat-ai[silero]"`    |
+| Analytics & Metrics | [Canonical AI](https://docs.pipecat.ai/api-reference/services/analytics/canonical), [Sentry](https://docs.pipecat.ai/api-reference/services/analytics/sentry)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     | `pip install "pipecat-ai[canonical]"` |
+
+📚 [View full services documentation →](https://docs.pipecat.ai/api-reference/services/supported-services)

 ## Code examples

- [foundational](https://github.com/pipecat-ai/pipecat/tree/main/examples/foundational) — small snippets that build on each other, introducing one or two concepts at a time
- [example apps](https://github.com/pipecat-ai/pipecat/tree/main/examples/) — complete applications that you can use as starting points for development
+- [Foundational](https://github.com/pipecat-ai/pipecat/tree/main/examples/foundational) — small snippets that build on each other, introducing one or two concepts at a time
+- [Example apps](https://github.com/pipecat-ai/pipecat/tree/main/examples/) — complete applications that you can use as starting points for development

 ## A simple voice agent running locally

@@ -109,7 +134,7 @@ Run it with:
 python app.py
 ```

-Daily provides a prebuilt WebRTC user interface. Whilst the app is running, you can visit at `https://<yourdomain>.daily.co/<room_url>` and listen to the bot say hello!
+Daily provides a prebuilt WebRTC user interface. While the app is running, you can visit at `https://<yourdomain>.daily.co/<room_url>` and listen to the bot say hello!

 ## WebRTC for production use

@@ -119,34 +144,6 @@ One way to get up and running quickly with WebRTC is to sign up for a Daily deve

 Sign up [here](https://dashboard.daily.co/u/signup) and [create a room](https://docs.daily.co/reference/rest-api/rooms) in the developer Dashboard.

-## What is VAD?
-
-Voice Activity Detection &mdash; very important for knowing when a user has finished speaking to your bot. If you are not using press-to-talk, and want Pipecat to detect when the user has finished talking, VAD is an essential component for a natural feeling conversation.
-
-Pipecat makes use of WebRTC VAD by default when using a WebRTC transport layer. Optionally, you can use Silero VAD for improved accuracy at the cost of higher CPU usage.
-
-```shell
-pip install pipecat-ai[silero]
-```
-
-## Running the Krisp Audio Filter
-
-To use the Krisp Filter in this project, you’ll need access to the **Krisp C++ SDK**.
-
-### Step 1: Obtain Access to the Krisp SDK
-1. **Create a Krisp Account**: If you don’t already have an account, [sign up at Krisp](https://krisp.ai/) to access the SDK.
-2. **Download the SDK**: Once you have an account, follow the instructions on the Krisp platform to download the [Krisp's desktop SDKs](https://sdk.krisp.ai/sdk/desktop).
-3. **Export the path to you krisp SDK**:
-`export KRISP_SDK_PATH=/PATH/TO/KRISP/SDK`
-
-### Step 2: Install the `pipecat-krisp` Module
-Once the environment variable `KRISP_SDK_PATH` is exported, activate your Python virtual environment and install it with `pip`:
-
-```shell
-source venv/bin/activate
-pip install pipecat-ai[krisp]
-```
-
 ## Hacking on the framework itself

 _Note that you may need to set up a virtual environment before following the instructions below. For instance, you might need to run the following from the root of the repo:_
@@ -224,8 +221,23 @@ Install the
 }
 ```

+## Contributing
+
+We welcome contributions from the community! Whether you're fixing bugs, improving documentation, or adding new features, here's how you can help:
+
+- **Found a bug?** Open an [issue](https://github.com/pipecat-ai/pipecat/issues)
+- **Have a feature idea?** Start a [discussion](https://discord.gg/pipecat)
+- **Want to contribute code?** Check our [CONTRIBUTING.md](CONTRIBUTING.md) guide
+- **Documentation improvements?** [Docs](https://github.com/pipecat-ai/docs) PRs are always welcome
+
+Before submitting a pull request, please check existing issues and PRs to avoid duplicates.
+
+We aim to review all contributions promptly and provide constructive feedback to help get your changes merged.
+
 ## Getting help

 ➡️ [Join our Discord](https://discord.gg/pipecat)

+➡️ [Read the docs](https://docs.pipecat.ai)
+
 ➡️ [Reach us on X](https://x.com/pipecat_ai)
--- a/examples/deployment/modal-example/.gitignore
+++ b/examples/deployment/modal-example/.gitignore
@@ -0,0 +1,91 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+dist/
+*.egg-info/
+*.egg
+.installed.cfg
+.eggs/
+downloads/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+MANIFEST
+
+# Virtual Environments
+venv/
+env/
+.env
+.venv/
+ENV/
+env.bak/
+venv.bak/
+
+# IDE
+.idea/
+.vscode/
+.spyderproject
+.spyproject
+.ropeproject
+
+# Testing and Coverage
+.coverage
+.coverage.*
+htmlcov/
+.pytest_cache/
+.tox/
+.nox/
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+cover/
+
+# Logs and Databases
+*.log
+*.db
+db.sqlite3
+db.sqlite3-journal
+pip-log.txt
+
+# System Files
+.DS_Store
+Thumbs.db
+desktop.ini
+*.swp
+*.swo
+*.bak
+*.tmp
+*~
+
+# Build and Documentation
+docs/_build/
+.pybuilder/
+target/
+instance/
+.webassets-cache
+.pdm.toml
+.pdm-python
+.pdm-build/
+__pypackages__/
+
+# Other
+*.mo
+*.pot
+*.sage.py
+.mypy_cache/
+.dmypy.json
+dmypy.json
+.pyre/
+.pytype/
+cython_debug/
+.ipynb_checkpoints
--- a/examples/deployment/modal-example/README.md
+++ b/examples/deployment/modal-example/README.md
@@ -0,0 +1,37 @@
+# Deploying Pipecat to Modal.com
+
+Barebones deployment example for [modal.com](https://www.modal.com)
+
+1. Install dependencies
+
+```bash
+python -m venv venv
+source venv/bin/active # or OS equivalent
+pip install -r requirements.txt
+```
+
+2. Setup .env
+
+```bash
+cp env.example .env
+```
+
+Alternatively, you can configure your Modal app to use [secrets](https://modal.com/docs/guide/secrets)
+
+3. Test the app locally
+
+```bash
+modal serve app.py
+```
+
+4. Deploy to production
+
+```bash
+modal deploy app.py
+```
+
+## Configuration options
+
+This app sets some sensible defaults for reducing cold starts, such as `minkeep_warm=1`, which will keep at least 1 warm instance ready for your bot function.
+
+It has been configured to only allow a concurrency of 1 (`max_inputs=1`) as each user will require their own running function.
--- a/examples/deployment/modal-example/init.py
+++ b/examples/deployment/modal-example/init.py
--- a/examples/deployment/modal-example/app.py
+++ b/examples/deployment/modal-example/app.py
@@ -0,0 +1,75 @@
+import os
+
+import aiohttp
+import modal
+from fastapi import HTTPException
+from fastapi.responses import JSONResponse
+from loguru import logger
+
+from bot import _voice_bot_process
+
+MAX_SESSION_TIME = 15 * 60  # 15 minutes
+
+app = modal.App("pipecat-modal")
+
+
+image = modal.Image.debian_slim(python_version="3.12").pip_install_from_requirements(
+    "requirements.txt"
+)
+
+
+@app.function(
+    image=image,
+    cpu=1.0,
+    secrets=[modal.Secret.from_dotenv()],
+    keep_warm=1,
+    enable_memory_snapshot=True,
+    max_inputs=1,  # Do not reuse instances across requests
+    retries=0,
+)
+def launch_bot_process(room_url: str, token: str):
+    _voice_bot_process(room_url, token)
+
+
+@app.function(
+    image=image,
+    secrets=[modal.Secret.from_dotenv()],
+)
+@modal.web_endpoint(method="POST")
+async def start():
+    from pipecat.transports.services.helpers.daily_rest import (
+        DailyRESTHelper,
+        DailyRoomParams,
+    )
+
+    logger.info("Request received")
+
+    async with aiohttp.ClientSession() as session:
+        daily_rest_helper = DailyRESTHelper(
+            daily_api_key=os.getenv("DAILY_API_KEY", ""),
+            daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+            aiohttp_session=session,
+        )
+
+        # Create new Daily room
+        room = await daily_rest_helper.create_room(DailyRoomParams())
+        if not room.url:
+            raise HTTPException(
+                status_code=500,
+                detail="Unable to create room",
+            )
+        logger.info(f"Created room: {room.url}")
+
+        # Create bot token for room
+        token = await daily_rest_helper.get_token(room.url, MAX_SESSION_TIME)
+        if not token:
+            raise HTTPException(status_code=500, detail=f"Failed to get token for room: {room.url}")
+
+        logger.info(f"Bot token created: {token}")
+
+        # Spawn a new bot process
+        launch_bot_process.spawn(room_url=room.url, token=token)
+
+        # Return room URL to the user to join
+        # Note: in production, you would want to return a token to the user
+        return JSONResponse(content={"room_url": room.url, token: token})
--- a/examples/deployment/modal-example/bot.py
+++ b/examples/deployment/modal-example/bot.py
@@ -0,0 +1,90 @@
+import asyncio
+import os
+import sys
+
+from dotenv import load_dotenv
+from loguru import logger
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main(room_url: str, token: str):
+    from pipecat.audio.vad.silero import SileroVADAnalyzer
+    from pipecat.frames.frames import EndFrame, LLMMessagesFrame
+    from pipecat.pipeline.pipeline import Pipeline
+    from pipecat.pipeline.runner import PipelineRunner
+    from pipecat.pipeline.task import PipelineParams, PipelineTask
+    from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+    from pipecat.services.cartesia import CartesiaTTSService
+    from pipecat.services.openai import OpenAILLMService
+    from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+    transport = DailyTransport(
+        room_url,
+        token,
+        "bot",
+        DailyParams(
+            audio_out_enabled=True,
+            transcription_enabled=True,
+            vad_enabled=True,
+            vad_analyzer=SileroVADAnalyzer(),
+        ),
+    )
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY", ""), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22"
+    )
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            context_aggregator.user(),
+            llm,
+            tts,
+            transport.output(),
+            context_aggregator.assistant(),
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        PipelineParams(
+            allow_interruptions=True,
+            enable_metrics=True,
+            enable_usage_metrics=True,
+            report_only_initial_ttfb=True,
+        ),
+    )
+
+    @transport.event_handler("on_first_participant_joined")
+    async def on_first_participant_joined(transport, participant):
+        await transport.capture_participant_transcription(participant["id"])
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([LLMMessagesFrame(messages)])
+
+    @transport.event_handler("on_participant_left")
+    async def on_participant_left(transport, participant, reason):
+        await task.queue_frame(EndFrame())
+
+    runner = PipelineRunner()
+
+    await runner.run(task)
+
+
+def _voice_bot_process(room_url: str, token: str):
+    asyncio.run(main(room_url, token))
--- a/examples/deployment/modal-example/env.example
+++ b/examples/deployment/modal-example/env.example
@@ -0,0 +1,3 @@
+DAILY_API_KEY=
+OPENAI_API_KEY=
+CARTESIA_API_KEY=
--- a/examples/deployment/modal-example/requirements.txt
+++ b/examples/deployment/modal-example/requirements.txt
@@ -0,0 +1,5 @@
+python-dotenv==1.0.1
+modal==0.65.48
+pipecat-ai[daily,silero,cartesia,openai]==0.0.48
+fastapi==0.115.4
+aiohttp==3.10.10
--- a/examples/foundational/07e-interruptible-playht.py
+++ b/examples/foundational/07e-interruptible-playht.py
@@ -49,7 +49,7 @@ async def main():
        tts = PlayHTTTSService(
            user_id=os.getenv("PLAYHT_USER_ID"),
            api_key=os.getenv("PLAYHT_API_KEY"),
-            voice_url="s3://voice-cloning-zero-shot/801a663f-efd0-4254-98d0-5c175514c3e8/jennifer/manifest.json",
+            voice_url="s3://voice-cloning-zero-shot/d9ff78ba-d016-47f6-b0ef-dd630f59414e/female-cs/manifest.json",
            params=PlayHTTTSService.InputParams(language=Language.EN),
        )

--- a/examples/foundational/07p-interruptible-google-audio-in.py
+++ b/examples/foundational/07p-interruptible-google-audio-in.py
@@ -0,0 +1,274 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import aiohttp
+import asyncio
+import os
+import sys
+
+import google.ai.generativelanguage as glm
+
+from dataclasses import dataclass
+from dotenv import load_dotenv
+from loguru import logger
+from runner import configure
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.google import GoogleLLMService
+from pipecat.processors.frame_processor import FrameProcessor
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.frames.frames import (
+    LLMFullResponseStartFrame,
+    LLMFullResponseEndFrame,
+    InputAudioRawFrame,
+    Frame,
+    StartInterruptionFrame,
+    TextFrame,
+    TranscriptionFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+marker = "|----|"
+system_message = f"""
+You are a helpful LLM in a WebRTC call. Your goals are to be helpful and brief in your responses.
+
+You are expert at transcribing audio to text. You will receive a mixture of audio and text input. When
+asked to transcribe what the user said, output an exact, word-for-word transcription.
+
+Your output will be converted to audio so don't include special characters in your answers.
+
+Each time you answer, you should respond in three parts.
+
+1. Transcribe exactly what the user said.
+2. Output the separator field '{marker}'.
+3. Respond to the user's input in a helpful, creative way using only simple text and punctuation.
+
+Example:
+
+User: How many ounces are in a pound?
+
+You: How many ounces are in a pound?
+{marker}
+There are 16 ounces in a pound.
+"""
+
+
+@dataclass
+class MagicDemoTranscriptionFrame(Frame):
+    text: str
+
+
+class UserAudioCollector(FrameProcessor):
+    def __init__(self, context, user_context_aggregator):
+        super().__init__()
+        self._context = context
+        self._user_context_aggregator = user_context_aggregator
+        self._audio_frames = []
+        self._start_secs = 0.2  # this should match VAD start_secs (hardcoding for now)
+        self._user_speaking = False
+
+    async def process_frame(self, frame, direction):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, TranscriptionFrame):
+            # We could gracefully handle both audio input and text/transcription input ...
+            # but let's leave that as an exercise to the reader. :-)
+            return
+        if isinstance(frame, UserStartedSpeakingFrame):
+            self._user_speaking = True
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            self._user_speaking = False
+            self._context.add_audio_frames_message(audio_frames=self._audio_frames)
+            await self._user_context_aggregator.push_frame(
+                self._user_context_aggregator.get_context_frame()
+            )
+        elif isinstance(frame, InputAudioRawFrame):
+            if self._user_speaking:
+                self._audio_frames.append(frame)
+            else:
+                # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest
+                # frames as necessary. Assume all audio frames have the same duration.
+                self._audio_frames.append(frame)
+                frame_duration = len(frame.audio) / 16 * frame.num_channels / frame.sample_rate
+                buffer_duration = frame_duration * len(self._audio_frames)
+                while buffer_duration > self._start_secs:
+                    self._audio_frames.pop(0)
+                    buffer_duration -= frame_duration
+
+        await self.push_frame(frame, direction)
+
+
+class TranscriptExtractor(FrameProcessor):
+    def __init__(self, context):
+        super().__init__()
+        self._context = context
+        self._accumulator = ""
+        self._processing_llm_response = False
+        self._accumulating_transcript = False
+
+    def reset(self):
+        self._accumulator = ""
+        self._processing_llm_response = False
+        self._accumulating_transcript = False
+
+    async def process_frame(self, frame, direction):
+        await super().process_frame(frame, direction)
+        if isinstance(frame, LLMFullResponseStartFrame):
+            self._processing_llm_response = True
+            self._accumulating_transcript = True
+        elif isinstance(frame, TextFrame) and self._processing_llm_response:
+            if self._accumulating_transcript:
+                text = frame.text
+                split_index = text.find(marker)
+                if split_index < 0:
+                    self._accumulator += frame.text
+                    # do not push this frame
+                    return
+                else:
+                    self._accumulating_transcript = False
+                    self._accumulator += text[:split_index]
+                    frame.text = text[split_index + len(marker) :]
+            await self.push_frame(frame)
+            return
+        elif isinstance(frame, LLMFullResponseEndFrame):
+            await self.push_frame(MagicDemoTranscriptionFrame(text=self._accumulator.strip()))
+            self.reset()
+
+        await self.push_frame(frame, direction)
+
+
+class TanscriptionContextFixup(FrameProcessor):
+    def __init__(self, context):
+        super().__init__()
+        self._context = context
+        self._transcript = "THIS IS A TRANSCRIPT"
+
+    def swap_user_audio(self):
+        if not self._transcript:
+            return
+        message = self._context.messages[-2]
+        last_part = message.parts[-1]
+        if (
+            message.role == "user"
+            and last_part.inline_data
+            and last_part.inline_data.mime_type == "audio/wav"
+        ):
+            self._context.messages[-2] = glm.Content(
+                role="user", parts=[glm.Part(text=self._transcript)]
+            )
+
+    def add_transcript_back_to_inference_output(self):
+        if not self._transcript:
+            return
+        message = self._context.messages[-1]
+        last_part = message.parts[-1]
+        if message.role == "model" and last_part.text:
+            self._context.messages[-1].parts[-1].text += f"\n\n{marker}\n{self._transcript}\n"
+
+    async def process_frame(self, frame, direction):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, MagicDemoTranscriptionFrame):
+            self._transcript = frame.text
+        elif isinstance(frame, LLMFullResponseEndFrame) or isinstance(
+            frame, StartInterruptionFrame
+        ):
+            self.swap_user_audio()
+            self.add_transcript_back_to_inference_output()
+            self._transcript = ""
+
+        await self.push_frame(frame, direction)
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, token) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                # No transcription at all. just audio input to Gemini!
+                # transcription_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                vad_audio_passthrough=True,
+            ),
+        )
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        )
+
+        llm = GoogleLLMService(model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY"))
+
+        messages = [
+            {
+                "role": "system",
+                "content": system_message,
+            },
+            {
+                "role": "user",
+                "content": "Start by saying hello.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+        audio_collector = UserAudioCollector(context, context_aggregator.user())
+        pull_transcript_out_of_llm_output = TranscriptExtractor(context)
+        fixup_context_messages = TanscriptionContextFixup(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                audio_collector,
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                pull_transcript_out_of_llm_output,
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+                fixup_context_messages,
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+            ),
+        )
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            await transport.capture_participant_transcription(participant["id"])
+            # Kick off the conversation.
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
--- a/examples/foundational/07q-interruptible-rime.py
+++ b/examples/foundational/07q-interruptible-rime.py
@@ -0,0 +1,100 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+from runner import configure
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMMessagesFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.openai import OpenAILLMService
+from pipecat.services.rime import RimeHttpTTSService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, token) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                transcription_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+            ),
+        )
+
+        tts = RimeHttpTTSService(
+            api_key=os.getenv("RIME_API_KEY", ""),
+            voice_id="rex",
+            params=RimeHttpTTSService.InputParams(reduce_latency=True),
+        )
+
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+                report_only_initial_ttfb=True,
+            ),
+        )
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            await transport.capture_participant_transcription(participant["id"])
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
--- a/examples/foundational/22b-natural-conversation-proposal.py
+++ b/examples/foundational/22b-natural-conversation-proposal.py
@@ -0,0 +1,339 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import aiohttp
+import asyncio
+import os
+import sys
+import time
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMMessagesFrame, TextFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.parallel_pipeline import ParallelPipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+)
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.deepgram import DeepgramSTTService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.sync.event_notifier import EventNotifier
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.processors.frame_processor import FrameProcessor, FrameDirection
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    Frame,
+    StartFrame,
+    StartInterruptionFrame,
+    StopInterruptionFrame,
+    SystemFrame,
+    TranscriptionFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
+from pipecat.sync.base_notifier import BaseNotifier
+from pipecat.processors.filters.function_filter import FunctionFilter
+from pipecat.processors.user_idle_processor import UserIdleProcessor
+
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+classifier_statement = "Determine if the user's statement ends with a complete thought and you should respond. The user text is transcribed speech. It may contain multiple fragments concatentated together. You are trying to determine only the completeness of the last user statement. The previous assistant statement is provided only for context. Categorize the text as either complete with the user now expecting a response, or incomplete. Return 'YES' if text is likely complete and the user is expecting a response. Return 'NO' if the text seems to be a partial expression or unfinished thought."
+
+
+class StatementJudgeContextFilter(FrameProcessor):
+    def __init__(self, notifier: BaseNotifier, **kwargs):
+        super().__init__(**kwargs)
+        self._notifier = notifier
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        # We must not block system frames.
+        if isinstance(frame, SystemFrame):
+            await self.push_frame(frame, direction)
+            return
+
+        # Just treat an LLMMessagesFrame as complete, no matter what.
+        if isinstance(frame, LLMMessagesFrame):
+            await self._notifier.notify()
+            return
+
+        # Otherwise, we only want to handle OpenAILLMContextFrames, and only want to push a simple
+        # messages frame that contains a system prompt and the most recent user messages,
+        # concatenated.
+        if isinstance(frame, OpenAILLMContextFrame):
+            logger.debug(f"Context Frame: {frame}")
+            # Take text content from the most recent user messages.
+            messages = frame.context.messages
+            user_text_messages = []
+            last_assistant_message = None
+            for message in reversed(messages):
+                if message["role"] != "user":
+                    if message["role"] == "assistant":
+                        last_assistant_message = message
+                    break
+                if isinstance(message["content"], str):
+                    user_text_messages.append(message["content"])
+                elif isinstance(message["content"], list):
+                    for content in message["content"]:
+                        if content["type"] == "text":
+                            user_text_messages.insert(0, content["text"])
+            # If we have any user text content, push an LLMMessagesFrame
+            if user_text_messages:
+                logger.debug(f"User text messages: {user_text_messages}")
+                user_message = " ".join(reversed(user_text_messages))
+                logger.debug(f"User message: {user_message}")
+                messages = [
+                    {
+                        "role": "system",
+                        "content": classifier_statement,
+                    }
+                ]
+                if last_assistant_message:
+                    messages.append(last_assistant_message)
+                messages.append({"role": "user", "content": user_message})
+                await self.push_frame(LLMMessagesFrame(messages))
+
+
+class CompletenessCheck(FrameProcessor):
+    def __init__(self, notifier: BaseNotifier):
+        super().__init__()
+        self._notifier = notifier
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if isinstance(frame, TextFrame) and frame.text == "YES":
+            logger.debug("Completeness check YES")
+            await self.push_frame(UserStoppedSpeakingFrame())
+            await self._notifier.notify()
+        elif isinstance(frame, TextFrame) and frame.text == "NO":
+            logger.debug("Completeness check NO")
+
+
+class OutputGate(FrameProcessor):
+    def __init__(self, notifier: BaseNotifier, **kwargs):
+        super().__init__(**kwargs)
+        self._gate_open = False
+        self._frames_buffer = []
+        self._notifier = notifier
+
+    def close_gate(self):
+        self._gate_open = False
+
+    def open_gate(self):
+        self._gate_open = True
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        # We must not block system frames.
+        if isinstance(frame, SystemFrame):
+            if isinstance(frame, StartFrame):
+                await self._start()
+            if isinstance(frame, (EndFrame, CancelFrame)):
+                await self._stop()
+            if isinstance(frame, StartInterruptionFrame):
+                self._frames_buffer = []
+                self.close_gate()
+            await self.push_frame(frame, direction)
+            return
+
+        # Ignore frames that are not following the direction of this gate.
+        if direction != FrameDirection.DOWNSTREAM:
+            await self.push_frame(frame, direction)
+            return
+
+        if self._gate_open:
+            await self.push_frame(frame, direction)
+            return
+
+        self._frames_buffer.append((frame, direction))
+
+    async def _start(self):
+        self._frames_buffer = []
+        self._gate_task = self.get_event_loop().create_task(self._gate_task_handler())
+
+    async def _stop(self):
+        self._gate_task.cancel()
+        await self._gate_task
+
+    async def _gate_task_handler(self):
+        while True:
+            try:
+                await self._notifier.wait()
+                self.open_gate()
+                for frame, direction in self._frames_buffer:
+                    await self.push_frame(frame, direction)
+                self._frames_buffer = []
+            except asyncio.CancelledError:
+                break
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, _) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            None,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                vad_audio_passthrough=True,
+            ),
+        )
+
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        )
+
+        # This is the LLM that will be used to detect if the user has finished a
+        # statement. This doesn't really need to be an LLM, we could use NLP
+        # libraries for that, but we have the machinery to use an LLM, so we might as well!
+        statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+
+        # This is the regular LLM.
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        # We have instructed the LLM to return 'YES' if it thinks the user
+        # completed a sentence. So, if it's 'YES' we will return true in this
+        # predicate which will wake up the notifier.
+        async def wake_check_filter(frame):
+            logger.debug(f"Completeness check frame: {frame}")
+            return frame.text == "YES"
+
+        # This is a notifier that we use to synchronize the two LLMs.
+        notifier = EventNotifier()
+
+        # This turns the LLM context into an inference request to classify the user's speech
+        # as complete or incomplete.
+        statement_judge_context_filter = StatementJudgeContextFilter(notifier=notifier)
+
+        # This sends a UserStoppedSpeakingFrame and triggers the notifier event
+        completeness_check = CompletenessCheck(notifier=notifier)
+
+        # # Notify if the user hasn't said anything.
+        async def user_idle_notifier(frame):
+            await notifier.notify()
+
+        # Sometimes the LLM will fail detecting if a user has completed a
+        # sentence, this will wake up the notifier if that happens.
+        user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
+
+        bot_output_gate = OutputGate(notifier=notifier)
+
+        async def block_user_stopped_speaking(frame):
+            return not isinstance(frame, UserStoppedSpeakingFrame)
+
+        async def pass_only_llm_trigger_frames(frame):
+            return (
+                isinstance(frame, OpenAILLMContextFrame)
+                or isinstance(frame, LLMMessagesFrame)
+                or isinstance(frame, StartInterruptionFrame)
+                or isinstance(frame, StopInterruptionFrame)
+            )
+
+        pipeline = Pipeline(
+            [
+                transport.input(),
+                stt,
+                context_aggregator.user(),
+                ParallelPipeline(
+                    [
+                        # Pass everything except UserStoppedSpeaking to the elements after
+                        # this ParallelPipeline
+                        FunctionFilter(filter=block_user_stopped_speaking),
+                    ],
+                    [
+                        # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
+                        # LLMMessagesFrame to the statement classifier LLM. The only frame this
+                        # sub-pipeline will output is a UserStoppedSpeakingFrame.
+                        statement_judge_context_filter,
+                        statement_llm,
+                        completeness_check,
+                    ],
+                    [
+                        # Block everything except OpenAILLMContextFrame and LLMMessagesFrame
+                        FunctionFilter(filter=pass_only_llm_trigger_frames),
+                        llm,
+                        bot_output_gate,  # Buffer all llm/tts output until notified.
+                    ],
+                ),
+                tts,
+                user_idle,
+                transport.output(),
+                context_aggregator.assistant(),
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+                report_only_initial_ttfb=True,
+            ),
+        )
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            await transport.capture_participant_transcription(participant["id"])
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        @transport.event_handler("on_app_message")
+        async def on_app_message(transport, message, sender):
+            logger.debug(f"Received app message: {message} - {sender}")
+            if "message" not in message:
+                return
+
+            await task.queue_frames(
+                [
+                    UserStartedSpeakingFrame(),
+                    TranscriptionFrame(
+                        user_id=sender, timestamp=time.time(), text=message["message"]
+                    ),
+                    UserStoppedSpeakingFrame(),
+                ]
+            )
+
+        runner = PipelineRunner()
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
--- a/examples/foundational/22c-natural-conversation-anthropic.py
+++ b/examples/foundational/22c-natural-conversation-anthropic.py
@@ -0,0 +1,433 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import aiohttp
+import asyncio
+import os
+import sys
+import time
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMMessagesFrame, TextFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.parallel_pipeline import ParallelPipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+)
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.deepgram import DeepgramSTTService
+from pipecat.services.anthropic import AnthropicLLMService
+from pipecat.sync.event_notifier import EventNotifier
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.processors.frame_processor import FrameProcessor, FrameDirection
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    Frame,
+    StartFrame,
+    StartInterruptionFrame,
+    StopInterruptionFrame,
+    SystemFrame,
+    TranscriptionFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
+from pipecat.sync.base_notifier import BaseNotifier
+from pipecat.processors.filters.function_filter import FunctionFilter
+from pipecat.processors.user_idle_processor import UserIdleProcessor
+
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+classifier_statement = """Determine if the user's statement ends with a complete thought and you should respond.
+
+The user text is transcribed speech. You are trying to determine if:
+
+1. the user has finished talking and expects a response from you, or
+2. this statement is incomplete and the user will continue talking
+
+A previous assistant response is provided for additional context. But you are only evaluating the user text. 
+
+The user text may contain multiple fragments concatentated together. There may be repeated words or mistakes in the transcription. There may be grammatical errors. There may be extra punctuation. Ignore all of that. Interpret the transcribed text as text that would have been spoken. Then consider only whether the user has finished speaking and is expecting a response.
+
+Categorize the last user statement as either complete with the user now expecting a response, or incomplete.
+
+Return 'YES' if text is likely complete and the user is expecting a response. Return 'NO' if the text seems to be a partial expression or unfinished thought.
+
+If you are not sure, respond with your best guess. If the user is expecting a response, respond with YES. If the user is not expecting a response, respond with NO. Always output either YES or NO and no other text.
+
+Respond only YES or NO
+
+Examples:
+
+User: What's the capital of
+Assistant: NO
+
+User: What's the captial of France?
+Assistant: YES
+
+User: Tell me a story about
+Assistant: NO
+
+User: Tell me a story about a dragon
+Assistant YES
+
+User: Is there a
+Assistant: NO
+
+User: Is there a large
+Assistant: NO
+
+User: Is there a large lake near Chicago?
+Assistant: YES
+
+User: When is the longest day of the year?
+Assistant: YES
+
+User: When when is the longest day of the year
+Assistant: YES
+
+User: When when is the
+ASSISTANT: NO
+
+User: What is the um I u
+Assistant: NO
+
+User: What is the um i u largest city in the world
+Assistant: YES
+
+User: How much does a how much does an adult elephant weigh?
+Assistant: YES
+
+User: How much does a how much does
+Assistant: NO
+
+User: What can you tell me All the
+Assistant: NO
+
+User: What can you tell me All the prime numbers less than 100
+Assistant: YES
+
+User: What's the what's the length of the Amazon River?
+Assistant: YES
+
+User: What's what's the length of the Amazon River?
+Assistant: YES
+
+User: What's what's the length of the Amazon River
+Assistant: YES
+
+User: What's what's the best way to get a coffee stain out of a white shirt
+Assistant: YES
+"""
+
+conversational_system_message = """You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.
+
+Please be very concise in your responses. Unless you are explicitly asked to do otherwise, give me the shortest complete answer possible without unnecessary elaboration. Generally you should answer with a single sentence.
+"""
+
+
+class StatementJudgeContextFilter(FrameProcessor):
+    def __init__(self, notifier: BaseNotifier, **kwargs):
+        super().__init__(**kwargs)
+        self._notifier = notifier
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        # We must not block system frames.
+        if isinstance(frame, SystemFrame):
+            await self.push_frame(frame, direction)
+            return
+
+        # Just treat an LLMMessagesFrame as complete, no matter what.
+        if isinstance(frame, LLMMessagesFrame):
+            await self._notifier.notify()
+            return
+
+        # Otherwise, we only want to handle OpenAILLMContextFrames, and only want to push a simple
+        # messages frame that contains a system prompt and the most recent user messages,
+        # concatenated.
+        if isinstance(frame, OpenAILLMContextFrame):
+            # Take text content from the most recent user messages.
+            messages = frame.context.messages
+            user_text_messages = []
+            last_assistant_message = None
+            for message in reversed(messages):
+                if message["role"] != "user":
+                    if message["role"] == "assistant":
+                        last_assistant_message = message
+                    break
+                if isinstance(message["content"], str):
+                    user_text_messages.append(message["content"])
+                elif isinstance(message["content"], list):
+                    for content in message["content"]:
+                        if content["type"] == "text":
+                            user_text_messages.insert(0, content["text"])
+            # If we have any user text content, push an LLMMessagesFrame
+            if user_text_messages:
+                user_message = " ".join(reversed(user_text_messages))
+                logger.debug(f"!!! {user_message}")
+                messages = [
+                    {
+                        "role": "system",
+                        "content": classifier_statement,
+                    }
+                ]
+                if last_assistant_message:
+                    messages.append(last_assistant_message)
+                messages.append({"role": "user", "content": user_message})
+                await self.push_frame(LLMMessagesFrame(messages))
+
+
+class CompletenessCheck(FrameProcessor):
+    def __init__(self, notifier: BaseNotifier):
+        super().__init__()
+        self._notifier = notifier
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, TextFrame) and frame.text == "YES":
+            logger.debug("!!! Completeness check YES")
+            await self.push_frame(UserStoppedSpeakingFrame())
+            await self._notifier.notify()
+        elif isinstance(frame, TextFrame) and frame.text == "NO":
+            logger.debug("!!! Completeness check NO")
+
+
+class OutputGate(FrameProcessor):
+    def __init__(self, notifier: BaseNotifier, **kwargs):
+        super().__init__(**kwargs)
+        self._gate_open = False
+        self._frames_buffer = []
+        self._notifier = notifier
+
+    def close_gate(self):
+        self._gate_open = False
+
+    def open_gate(self):
+        self._gate_open = True
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        # We must not block system frames.
+        if isinstance(frame, SystemFrame):
+            if isinstance(frame, StartFrame):
+                await self._start()
+            if isinstance(frame, (EndFrame, CancelFrame)):
+                await self._stop()
+            if isinstance(frame, StartInterruptionFrame):
+                self._frames_buffer = []
+                self.close_gate()
+            await self.push_frame(frame, direction)
+            return
+
+        # Ignore frames that are not following the direction of this gate.
+        if direction != FrameDirection.DOWNSTREAM:
+            await self.push_frame(frame, direction)
+            return
+
+        if self._gate_open:
+            await self.push_frame(frame, direction)
+            return
+
+        self._frames_buffer.append((frame, direction))
+
+    async def _start(self):
+        self._frames_buffer = []
+        self._gate_task = self.get_event_loop().create_task(self._gate_task_handler())
+
+    async def _stop(self):
+        self._gate_task.cancel()
+        await self._gate_task
+
+    async def _gate_task_handler(self):
+        while True:
+            try:
+                await self._notifier.wait()
+                self.open_gate()
+                for frame, direction in self._frames_buffer:
+                    await self.push_frame(frame, direction)
+                self._frames_buffer = []
+            except asyncio.CancelledError:
+                break
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, _) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            None,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                vad_audio_passthrough=True,
+            ),
+        )
+
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        )
+
+        # This is the LLM that will be used to detect if the user has finished a
+        # statement. This doesn't really need to be an LLM, we could use NLP
+        # libraries for that, but we have the machinery to use an LLM, so we might as well!
+        statement_llm = AnthropicLLMService(
+            api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-haiku-20241022", name="Haiku"
+        )
+
+        # This is the regular LLM.
+        llm = AnthropicLLMService(
+            api_key=os.getenv("ANTHROPIC_API_KEY"),
+            model="claude-3-5-sonnet-20241022",
+            name="Sonnet",
+            params=AnthropicLLMService.InputParams(enable_prompt_caching_beta=True),
+        )
+
+        messages = [
+            {
+                "role": "system",
+                "content": conversational_system_message,
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        # We have instructed the LLM to return 'YES' if it thinks the user
+        # completed a sentence. So, if it's 'YES' we will return true in this
+        # predicate which will wake up the notifier.
+        async def wake_check_filter(frame):
+            return frame.text == "YES"
+
+        # This is a notifier that we use to synchronize the two LLMs.
+        notifier = EventNotifier()
+
+        # This turns the LLM context into an inference request to classify the user's speech
+        # as complete or incomplete.
+        statement_judge_context_filter = StatementJudgeContextFilter(notifier=notifier)
+
+        # This sends a UserStoppedSpeakingFrame and triggers the notifier event
+        completeness_check = CompletenessCheck(notifier=notifier)
+
+        # # Notify if the user hasn't said anything.
+        async def user_idle_notifier(frame):
+            await notifier.notify()
+
+        # Sometimes the LLM will fail detecting if a user has completed a
+        # sentence, this will wake up the notifier if that happens.
+        user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
+
+        bot_output_gate = OutputGate(notifier=notifier)
+
+        async def block_user_stopped_speaking(frame):
+            return not isinstance(frame, UserStoppedSpeakingFrame)
+
+        async def pass_only_llm_trigger_frames(frame):
+            return (
+                isinstance(frame, OpenAILLMContextFrame)
+                or isinstance(frame, LLMMessagesFrame)
+                or isinstance(frame, StartInterruptionFrame)
+                or isinstance(frame, StopInterruptionFrame)
+            )
+
+        pipeline = Pipeline(
+            [
+                transport.input(),
+                stt,
+                context_aggregator.user(),
+                ParallelPipeline(
+                    [
+                        # Pass everything except UserStoppedSpeaking to the elements after
+                        # this ParallelPipeline
+                        FunctionFilter(filter=block_user_stopped_speaking),
+                    ],
+                    [
+                        # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
+                        # LLMMessagesFrame to the statement classifier LLM. The only frame this
+                        # sub-pipeline will output is a UserStoppedSpeakingFrame.
+                        statement_judge_context_filter,
+                        statement_llm,
+                        completeness_check,
+                    ],
+                    [
+                        # Block everything except OpenAILLMContextFrame and LLMMessagesFrame
+                        FunctionFilter(filter=pass_only_llm_trigger_frames),
+                        llm,
+                        bot_output_gate,  # Buffer all llm/tts output until notified.
+                    ],
+                ),
+                tts,
+                user_idle,
+                transport.output(),
+                context_aggregator.assistant(),
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+            ),
+        )
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            await transport.capture_participant_transcription(participant["id"])
+            # Kick off the conversation.
+            messages.append(
+                {
+                    "role": "user",
+                    "content": "Start by just saying \"Hello I'm ready.\" Don't say anything else.",
+                }
+            )
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        @transport.event_handler("on_app_message")
+        async def on_app_message(transport, message, sender):
+            logger.debug(f"Received app message: {message} - {sender}")
+            if "message" not in message:
+                return
+
+            await task.queue_frames(
+                [
+                    UserStartedSpeakingFrame(),
+                    TranscriptionFrame(
+                        user_id=sender, timestamp=time.time(), text=message["message"]
+                    ),
+                    UserStoppedSpeakingFrame(),
+                ]
+            )
+
+        runner = PipelineRunner()
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
--- a/examples/foundational/22d-natural-conversation-gemini-audio.py
+++ b/examples/foundational/22d-natural-conversation-gemini-audio.py
@@ -0,0 +1,355 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import aiohttp
+import asyncio
+import os
+import sys
+import time
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMMessagesFrame, TextFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.parallel_pipeline import ParallelPipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.services.deepgram import DeepgramSTTService
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+)
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.google import GoogleLLMService, GoogleLLMContext
+from pipecat.sync.event_notifier import EventNotifier
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.processors.frame_processor import FrameProcessor, FrameDirection
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    Frame,
+    InputAudioRawFrame,
+    StartFrame,
+    StartInterruptionFrame,
+    StopInterruptionFrame,
+    SystemFrame,
+    TranscriptionFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
+from pipecat.sync.base_notifier import BaseNotifier
+from pipecat.processors.filters.function_filter import FunctionFilter
+from pipecat.processors.user_idle_processor import UserIdleProcessor
+
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+classifier_statement = """You are an audio language classifier model. You are receiving audio from a user in a WebRTC call. Your job is to decide whether the user has finished speaking or not.
+
+Categorize the input you receive as either:
+
+1.  a complete thought, statement, or question, or
+2.  an incomplete thought, statement, or question
+
+Output 'YES' if the input is likely to be a completed thought, statement, or question.
+
+Output 'NO' if the input indicates that the user is still speaking and does not yet expect a response yet.
+
+If you are unsure, output 'YES'.
+"""
+
+conversational_system_message = """You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.
+
+Please be very concise in your responses. Unless you are explicitly asked to do otherwise, give me the shortest complete answer possible without unnecessary elaboration. Generally you should answer with a single sentence.
+"""
+
+
+class StatementJudgeAudioContextAccumulator(FrameProcessor):
+    def __init__(self, *, notifier: BaseNotifier, **kwargs):
+        super().__init__(**kwargs)
+        self._notifier = notifier
+        self._audio_frames = []
+        self._audio_frames = []
+        self._start_secs = 0.2  # this should match VAD start_secs (hardcoding for now)
+        self._user_speaking = False
+
+    async def reset(self):
+        self._audio_frames = []
+        self._user_speaking = False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        # ignore context frame
+        if isinstance(frame, OpenAILLMContextFrame):
+            return
+
+        if isinstance(frame, TranscriptionFrame):
+            # We could gracefully handle both audio input and text/transcription input ...
+            # but let's leave that as an exercise to the reader. :-)
+            return
+        if isinstance(frame, UserStartedSpeakingFrame):
+            self._user_speaking = True
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            self._user_speaking = False
+            context = GoogleLLMContext()
+            context.set_messages([{"role": "system", "content": classifier_statement}])
+            context.add_audio_frames_message(audio_frames=self._audio_frames)
+            await self.push_frame(OpenAILLMContextFrame(context=context))
+        elif isinstance(frame, InputAudioRawFrame):
+            if self._user_speaking:
+                self._audio_frames.append(frame)
+            else:
+                # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest
+                # frames as necessary. Assume all audio frames have the same duration.
+                self._audio_frames.append(frame)
+                frame_duration = len(frame.audio) / 16 * frame.num_channels / frame.sample_rate
+                buffer_duration = frame_duration * len(self._audio_frames)
+                while buffer_duration > self._start_secs:
+                    self._audio_frames.pop(0)
+                    buffer_duration -= frame_duration
+
+        await self.push_frame(frame, direction)
+
+
+class CompletenessCheck(FrameProcessor):
+    def __init__(
+        self, notifier: BaseNotifier, audio_accumulator: StatementJudgeAudioContextAccumulator
+    ):
+        super().__init__()
+        self._notifier = notifier
+        self._audio_accumulator = audio_accumulator
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, TextFrame) and frame.text.startswith("YES"):
+            logger.debug("Completeness check YES")
+            await self.push_frame(UserStoppedSpeakingFrame())
+            await self._audio_accumulator.reset()
+            await self._notifier.notify()
+        elif isinstance(frame, TextFrame):
+            if frame.text.strip():
+                logger.debug(f"Completeness check NO - '{frame.text}'")
+
+
+class OutputGate(FrameProcessor):
+    def __init__(self, notifier: BaseNotifier, **kwargs):
+        super().__init__(**kwargs)
+        self._gate_open = False
+        self._frames_buffer = []
+        self._notifier = notifier
+
+    def close_gate(self):
+        self._gate_open = False
+
+    def open_gate(self):
+        self._gate_open = True
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        # We must not block system frames.
+        if isinstance(frame, SystemFrame):
+            if isinstance(frame, StartFrame):
+                await self._start()
+            if isinstance(frame, (EndFrame, CancelFrame)):
+                await self._stop()
+            if isinstance(frame, StartInterruptionFrame):
+                self._frames_buffer = []
+                self.close_gate()
+            await self.push_frame(frame, direction)
+            return
+
+        # Ignore frames that are not following the direction of this gate.
+        if direction != FrameDirection.DOWNSTREAM:
+            await self.push_frame(frame, direction)
+            return
+
+        if self._gate_open:
+            await self.push_frame(frame, direction)
+            return
+
+        self._frames_buffer.append((frame, direction))
+
+    async def _start(self):
+        self._frames_buffer = []
+        self._gate_task = self.get_event_loop().create_task(self._gate_task_handler())
+
+    async def _stop(self):
+        self._gate_task.cancel()
+        await self._gate_task
+
+    async def _gate_task_handler(self):
+        while True:
+            try:
+                await self._notifier.wait()
+                self.open_gate()
+                for frame, direction in self._frames_buffer:
+                    await self.push_frame(frame, direction)
+                self._frames_buffer = []
+            except asyncio.CancelledError:
+                break
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, _) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            None,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                vad_audio_passthrough=True,
+            ),
+        )
+
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        )
+
+        # This is the LLM that will be used to detect if the user has finished a
+        # statement. This doesn't really need to be an LLM, we could use NLP
+        # libraries for that, but we have the machinery to use an LLM, so we might as well!
+        statement_llm = GoogleLLMService(
+            model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY")
+        )
+
+        # This is the regular LLM.
+        llm = GoogleLLMService(model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY"))
+
+        messages = [
+            {
+                "role": "system",
+                "content": conversational_system_message,
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        # We have instructed the LLM to return 'YES' if it thinks the user
+        # completed a sentence. So, if it's 'YES' we will return true in this
+        # predicate which will wake up the notifier.
+        async def wake_check_filter(frame):
+            return frame.text == "YES"
+
+        # This is a notifier that we use to synchronize the two LLMs.
+        notifier = EventNotifier()
+
+        # This turns the LLM context into an inference request to classify the user's speech
+        # as complete or incomplete.
+        statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier)
+
+        # This sends a UserStoppedSpeakingFrame and triggers the notifier event
+        completeness_check = CompletenessCheck(
+            notifier=notifier, audio_accumulator=statement_judge_context_filter
+        )
+
+        # # Notify if the user hasn't said anything.
+        async def user_idle_notifier(frame):
+            await notifier.notify()
+
+        # Sometimes the LLM will fail detecting if a user has completed a
+        # sentence, this will wake up the notifier if that happens.
+        user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
+
+        bot_output_gate = OutputGate(notifier=notifier)
+
+        async def block_user_stopped_speaking(frame):
+            return not isinstance(frame, UserStoppedSpeakingFrame)
+
+        async def pass_only_llm_trigger_frames(frame):
+            return (
+                isinstance(frame, OpenAILLMContextFrame)
+                or isinstance(frame, LLMMessagesFrame)
+                or isinstance(frame, StartInterruptionFrame)
+                or isinstance(frame, StopInterruptionFrame)
+            )
+
+        pipeline = Pipeline(
+            [
+                transport.input(),
+                ParallelPipeline(
+                    [
+                        # Pass everything except UserStoppedSpeaking to the elements after
+                        # this ParallelPipeline
+                        FunctionFilter(filter=block_user_stopped_speaking),
+                    ],
+                    [
+                        statement_judge_context_filter,
+                        statement_llm,
+                        completeness_check,
+                    ],
+                    [
+                        stt,
+                        context_aggregator.user(),
+                        # Block everything except OpenAILLMContextFrame and LLMMessagesFrame
+                        FunctionFilter(filter=pass_only_llm_trigger_frames),
+                        llm,
+                        bot_output_gate,  # Buffer all llm/tts output until notified.
+                    ],
+                ),
+                tts,
+                user_idle,
+                transport.output(),
+                context_aggregator.assistant(),
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+            ),
+        )
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            await transport.capture_participant_transcription(participant["id"])
+            # Kick off the conversation.
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        @transport.event_handler("on_app_message")
+        async def on_app_message(transport, message, sender):
+            logger.debug(f"Received app message: {message} - {sender}")
+            if "message" not in message:
+                return
+
+            await task.queue_frames(
+                [
+                    UserStartedSpeakingFrame(),
+                    TranscriptionFrame(
+                        user_id=sender, timestamp=time.time(), text=message["message"]
+                    ),
+                    UserStoppedSpeakingFrame(),
+                ]
+            )
+
+        runner = PipelineRunner()
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
--- a/examples/foundational/24-stt-mute-filter.py
+++ b/examples/foundational/24-stt-mute-filter.py
@@ -0,0 +1,98 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+from runner import configure
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import (
+    LLMMessagesFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.filters.stt_mute_filter import STTMuteConfig, STTMuteFilter, STTMuteStrategy
+from pipecat.services.deepgram import DeepgramSTTService, DeepgramTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, _) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            None,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                vad_audio_passthrough=True,
+            ),
+        )
+
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+        # Configure the mute processor to mute only during first speech
+        stt_mute_processor = STTMuteFilter(
+            stt_service=stt, config=STTMuteConfig(strategy=STTMuteStrategy.FIRST_SPEECH)
+        )
+
+        tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en")
+
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                stt_mute_processor,  # Add the mute processor before STT
+                stt,  # STT
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
+
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
--- a/examples/simple-chatbot/bot.py
+++ b/examples/simple-chatbot/bot.py
@@ -162,7 +162,7 @@ async def main():

        @transport.event_handler("on_first_participant_joined")
        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
+            await transport.capture_participant_transcription(participant["id"])
            await task.queue_frames([LLMMessagesFrame(messages)])

        runner = PipelineRunner()
--- a/src/pipecat/frames/frames.py
+++ b/src/pipecat/frames/frames.py
@@ -570,6 +570,13 @@ class TTSUpdateSettingsFrame(ServiceUpdateSettingsFrame):
    pass


+@dataclass
+class STTMuteFrame(ControlFrame):
+    """Control frame to mute/unmute the STT service."""
+
+    mute: bool
+
+
@dataclass
 class STTUpdateSettingsFrame(ServiceUpdateSettingsFrame):
    pass
--- a/src/pipecat/processors/aggregators/openai_llm_context.py
+++ b/src/pipecat/processors/aggregators/openai_llm_context.py
@@ -15,6 +15,7 @@ from loguru import logger
 from PIL import Image

 from pipecat.frames.frames import (
+    AudioRawFrame,
    Frame,
    FunctionCallInProgressFrame,
    FunctionCallResultFrame,
@@ -174,6 +175,10 @@ class OpenAILLMContext:
            content.append({"type": "text", "text": text})
        self.add_message({"role": "user", "content": content})

+    def add_audio_frames_message(self, *, audio_frames: list[AudioRawFrame], text: str = None):
+        # todo: implement for OpenAI models and others
+        pass
+
    async def call_function(
        self,
        f: Callable[
@@ -213,6 +218,29 @@ class OpenAILLMContext:

        await f(function_name, tool_call_id, arguments, llm, self, function_call_result_callback)

+    def create_wav_header(self, sample_rate, num_channels, bits_per_sample, data_size):
+        # RIFF chunk descriptor
+        header = bytearray()
+        header.extend(b"RIFF")  # ChunkID
+        header.extend((data_size + 36).to_bytes(4, "little"))  # ChunkSize: total size - 8
+        header.extend(b"WAVE")  # Format
+        # "fmt " sub-chunk
+        header.extend(b"fmt ")  # Subchunk1ID
+        header.extend((16).to_bytes(4, "little"))  # Subchunk1Size (16 for PCM)
+        header.extend((1).to_bytes(2, "little"))  # AudioFormat (1 for PCM)
+        header.extend(num_channels.to_bytes(2, "little"))  # NumChannels
+        header.extend(sample_rate.to_bytes(4, "little"))  # SampleRate
+        # Calculate byte rate and block align
+        byte_rate = sample_rate * num_channels * (bits_per_sample // 8)
+        block_align = num_channels * (bits_per_sample // 8)
+        header.extend(byte_rate.to_bytes(4, "little"))  # ByteRate
+        header.extend(block_align.to_bytes(2, "little"))  # BlockAlign
+        header.extend(bits_per_sample.to_bytes(2, "little"))  # BitsPerSample
+        # "data" sub-chunk
+        header.extend(b"data")  # Subchunk2ID
+        header.extend(data_size.to_bytes(4, "little"))  # Subchunk2Size
+        return header
+

@dataclass
 class OpenAILLMContextFrame(Frame):
--- a/src/pipecat/processors/filters/function_filter.py
+++ b/src/pipecat/processors/filters/function_filter.py
@@ -11,19 +11,27 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor


 class FunctionFilter(FrameProcessor):
-    def __init__(self, filter: Callable[[Frame], Awaitable[bool]]):
+    def __init__(
+        self,
+        filter: Callable[[Frame], Awaitable[bool]],
+        direction: FrameDirection = FrameDirection.DOWNSTREAM,
+    ):
        super().__init__()
        self._filter = filter
+        self._direction = direction

    #
    # Frame processor
    #

-    def _should_passthrough_frame(self, frame):
-        return isinstance(frame, SystemFrame)
+    # Ignore system frames and frames that are not following the direction of this gate
+    def _should_passthrough_frame(self, frame, direction):
+        return isinstance(frame, SystemFrame) or direction != self._direction

    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        passthrough = self._should_passthrough_frame(frame)
+        await super().process_frame(frame, direction)
+
+        passthrough = self._should_passthrough_frame(frame, direction)
        allowed = await self._filter(frame)
        if passthrough or allowed:
            await self.push_frame(frame, direction)
--- a/src/pipecat/processors/filters/stt_mute_filter.py
+++ b/src/pipecat/processors/filters/stt_mute_filter.py
@@ -0,0 +1,111 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Awaitable, Callable, Optional
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
+    Frame,
+    StartInterruptionFrame,
+    StopInterruptionFrame,
+    STTMuteFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.ai_services import STTService
+
+
+class STTMuteStrategy(Enum):
+    FIRST_SPEECH = "first_speech"  # Mute only during first bot speech
+    ALWAYS = "always"  # Mute during all bot speech
+    CUSTOM = "custom"  # Allow custom logic via callback
+
+
+@dataclass
+class STTMuteConfig:
+    """Configuration for STTMuteFilter"""
+
+    strategy: STTMuteStrategy
+    # Optional callback for custom muting logic
+    should_mute_callback: Optional[Callable[["STTMuteFilter"], Awaitable[bool]]] = None
+
+
+class STTMuteFilter(FrameProcessor):
+    """A general-purpose processor that handles STT muting and interruption control.
+
+    This processor combines the concepts of STT muting and interruption control,
+    treating them as a single coordinated feature. When STT is muted, interruptions
+    are automatically disabled.
+    """
+
+    def __init__(self, stt_service: STTService, config: STTMuteConfig, **kwargs):
+        super().__init__(**kwargs)
+        self._stt_service = stt_service
+        self._config = config
+        self._first_speech_handled = False
+        self._bot_is_speaking = False
+
+    @property
+    def is_muted(self) -> bool:
+        """Returns whether STT is currently muted."""
+        return self._stt_service.is_muted
+
+    async def _handle_mute_state(self, should_mute: bool):
+        """Handles both STT muting and interruption control."""
+        if should_mute != self.is_muted:
+            logger.debug(f"STT {'muting' if should_mute else 'unmuting'}")
+            await self.push_frame(STTMuteFrame(mute=should_mute))
+
+    async def _should_mute(self) -> bool:
+        """Determines if STT should be muted based on current state and strategy."""
+        if not self._bot_is_speaking:
+            return False
+
+        if self._config.strategy == STTMuteStrategy.ALWAYS:
+            return True
+        elif (
+            self._config.strategy == STTMuteStrategy.FIRST_SPEECH and not self._first_speech_handled
+        ):
+            self._first_speech_handled = True
+            return True
+        elif self._config.strategy == STTMuteStrategy.CUSTOM and self._config.should_mute_callback:
+            return await self._config.should_mute_callback(self)
+
+        return False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        # Handle bot speaking state changes
+        if isinstance(frame, BotStartedSpeakingFrame):
+            self._bot_is_speaking = True
+            await self._handle_mute_state(await self._should_mute())
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            self._bot_is_speaking = False
+            await self._handle_mute_state(await self._should_mute())
+
+        # Handle frame propagation
+        if isinstance(
+            frame,
+            (
+                StartInterruptionFrame,
+                StopInterruptionFrame,
+                UserStartedSpeakingFrame,
+                UserStoppedSpeakingFrame,
+            ),
+        ):
+            # Only pass VAD-related frames when not muted
+            if not self.is_muted:
+                await self.push_frame(frame, direction)
+            else:
+                logger.debug(f"{frame.__class__.__name__} suppressed - STT currently muted")
+        else:
+            # Pass all other frames through
+            await self.push_frame(frame, direction)
--- a/src/pipecat/services/ai_services.py
+++ b/src/pipecat/services/ai_services.py
@@ -22,6 +22,7 @@ from pipecat.frames.frames import (
    LLMFullResponseEndFrame,
    StartFrame,
    StartInterruptionFrame,
+    STTMuteFrame,
    STTUpdateSettingsFrame,
    TextFrame,
    TTSAudioRawFrame,
@@ -454,6 +455,12 @@ class STTService(AIService):
        super().__init__(**kwargs)
        self._audio_passthrough = audio_passthrough
        self._settings: Dict[str, Any] = {}
+        self._muted: bool = False
+
+    @property
+    def is_muted(self) -> bool:
+        """Returns whether the STT service is currently muted."""
+        return self._muted

    @abstractmethod
    async def set_model(self, model: str):
@@ -482,7 +489,8 @@ class STTService(AIService):
                logger.warning(f"Unknown setting for STT service: {key}")

    async def process_audio_frame(self, frame: AudioRawFrame):
-        await self.process_generator(self.run_stt(frame.audio))
+        if not self._muted:
+            await self.process_generator(self.run_stt(frame.audio))

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        """Processes a frame of audio data, either buffering or transcribing it."""
@@ -497,6 +505,9 @@ class STTService(AIService):
                await self.push_frame(frame, direction)
        elif isinstance(frame, STTUpdateSettingsFrame):
            await self._update_settings(frame.settings)
+        elif isinstance(frame, STTMuteFrame):
+            self._muted = frame.mute
+            logger.debug(f"STT service {'muted' if frame.mute else 'unmuted'}")
        else:
            await self.push_frame(frame, direction)

--- a/src/pipecat/services/aws.py
+++ b/src/pipecat/services/aws.py
@@ -5,7 +5,6 @@
 #

 import asyncio
-
 from typing import AsyncGenerator, Optional

 from loguru import logger
@@ -33,6 +32,40 @@ except ModuleNotFoundError as e:
    raise Exception(f"Missing module: {e}")


+def language_to_aws_language(language: Language) -> str | None:
+    language_map = {
+        Language.CA: "ca-ES",
+        Language.ZH: "cmn-CN",
+        Language.DA: "da-DK",
+        Language.NL: "nl-NL",
+        Language.NL_BE: "nl-BE",
+        Language.EN: "en-US",
+        Language.EN_US: "en-US",
+        Language.EN_AU: "en-AU",
+        Language.EN_GB: "en-GB",
+        Language.EN_NZ: "en-NZ",
+        Language.EN_IN: "en-IN",
+        Language.FI: "fi-FI",
+        Language.FR: "fr-FR",
+        Language.FR_CA: "fr-CA",
+        Language.DE: "de-DE",
+        Language.HI: "hi-IN",
+        Language.IT: "it-IT",
+        Language.JA: "ja-JP",
+        Language.KO: "ko-KR",
+        Language.NO: "nb-NO",
+        Language.PL: "pl-PL",
+        Language.PT: "pt-PT",
+        Language.PT_BR: "pt-BR",
+        Language.RO: "ro-RO",
+        Language.RU: "ru-RU",
+        Language.ES: "es-ES",
+        Language.SV: "sv-SE",
+        Language.TR: "tr-TR",
+    }
+    return language_map.get(language)
+
+
 class AWSTTSService(TTSService):
    class InputParams(BaseModel):
        engine: Optional[str] = None
@@ -65,7 +98,7 @@ class AWSTTSService(TTSService):
            "engine": params.engine,
            "language": self.language_to_service_language(params.language)
            if params.language
-            else Language.EN,
+            else "en-US",
            "pitch": params.pitch,
            "rate": params.rate,
            "volume": params.volume,
@@ -77,62 +110,7 @@ class AWSTTSService(TTSService):
        return True

    def language_to_service_language(self, language: Language) -> str | None:
-        match language:
-            case Language.CA:
-                return "ca-ES"
-            case Language.ZH:
-                return "cmn-CN"
-            case Language.DA:
-                return "da-DK"
-            case Language.NL:
-                return "nl-NL"
-            case Language.NL_BE:
-                return "nl-BE"
-            case Language.EN | Language.EN_US:
-                return "en-US"
-            case Language.EN_AU:
-                return "en-AU"
-            case Language.EN_GB:
-                return "en-GB"
-            case Language.EN_NZ:
-                return "en-NZ"
-            case Language.EN_IN:
-                return "en-IN"
-            case Language.FI:
-                return "fi-FI"
-            case Language.FR:
-                return "fr-FR"
-            case Language.FR_CA:
-                return "fr-CA"
-            case Language.DE:
-                return "de-DE"
-            case Language.HI:
-                return "hi-IN"
-            case Language.IT:
-                return "it-IT"
-            case Language.JA:
-                return "ja-JP"
-            case Language.KO:
-                return "ko-KR"
-            case Language.NO:
-                return "nb-NO"
-            case Language.PL:
-                return "pl-PL"
-            case Language.PT:
-                return "pt-PT"
-            case Language.PT_BR:
-                return "pt-BR"
-            case Language.RO:
-                return "ro-RO"
-            case Language.RU:
-                return "ru-RU"
-            case Language.ES:
-                return "es-ES"
-            case Language.SV:
-                return "sv-SE"
-            case Language.TR:
-                return "tr-TR"
-        return None
+        return language_to_aws_language(language)

    def _construct_ssml(self, text: str) -> str:
        ssml = "<speak>"
--- a/src/pipecat/services/azure.py
+++ b/src/pipecat/services/azure.py
@@ -61,6 +61,67 @@ except ModuleNotFoundError as e:
    raise Exception(f"Missing module: {e}")


+def language_to_azure_language(language: Language) -> str | None:
+    language_map = {
+        Language.BG: "bg-BG",
+        Language.CA: "ca-ES",
+        Language.ZH: "zh-CN",
+        Language.ZH_TW: "zh-TW",
+        Language.CS: "cs-CZ",
+        Language.DA: "da-DK",
+        Language.NL: "nl-NL",
+        Language.EN: "en-US",
+        Language.EN_US: "en-US",
+        Language.EN_AU: "en-AU",
+        Language.EN_GB: "en-GB",
+        Language.EN_NZ: "en-NZ",
+        Language.EN_IN: "en-IN",
+        Language.ET: "et-EE",
+        Language.FI: "fi-FI",
+        Language.NL_BE: "nl-BE",
+        Language.FR: "fr-FR",
+        Language.FR_CA: "fr-CA",
+        Language.DE: "de-DE",
+        Language.DE_CH: "de-CH",
+        Language.EL: "el-GR",
+        Language.HI: "hi-IN",
+        Language.HU: "hu-HU",
+        Language.ID: "id-ID",
+        Language.IT: "it-IT",
+        Language.JA: "ja-JP",
+        Language.KO: "ko-KR",
+        Language.LV: "lv-LV",
+        Language.LT: "lt-LT",
+        Language.MS: "ms-MY",
+        Language.NO: "nb-NO",
+        Language.PL: "pl-PL",
+        Language.PT: "pt-PT",
+        Language.PT_BR: "pt-BR",
+        Language.RO: "ro-RO",
+        Language.RU: "ru-RU",
+        Language.SK: "sk-SK",
+        Language.ES: "es-ES",
+        Language.SV: "sv-SE",
+        Language.TH: "th-TH",
+        Language.TR: "tr-TR",
+        Language.UK: "uk-UA",
+        Language.VI: "vi-VN",
+    }
+    return language_map.get(language)
+
+
+def sample_rate_to_output_format(sample_rate: int) -> SpeechSynthesisOutputFormat:
+    sample_rate_map = {
+        8000: SpeechSynthesisOutputFormat.Raw8Khz16BitMonoPcm,
+        16000: SpeechSynthesisOutputFormat.Raw16Khz16BitMonoPcm,
+        22050: SpeechSynthesisOutputFormat.Raw22050Hz16BitMonoPcm,
+        24000: SpeechSynthesisOutputFormat.Raw24Khz16BitMonoPcm,
+        44100: SpeechSynthesisOutputFormat.Raw44100Hz16BitMonoPcm,
+        48000: SpeechSynthesisOutputFormat.Raw48Khz16BitMonoPcm,
+    }
+    return sample_rate_map.get(sample_rate, SpeechSynthesisOutputFormat.Raw24Khz16BitMonoPcm)
+
+
 class AzureLLMService(BaseOpenAILLMService):
    def __init__(
        self, *, api_key: str, endpoint: str, model: str, api_version: str = "2023-12-01-preview"
@@ -89,23 +150,6 @@ class AzureLLMService(BaseOpenAILLMService):
        return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)


-def sample_rate_to_output_format(sample_rate: int) -> SpeechSynthesisOutputFormat:
-    match sample_rate:
-        case 8000:
-            return SpeechSynthesisOutputFormat.Raw8Khz16BitMonoPcm
-        case 16000:
-            return SpeechSynthesisOutputFormat.Raw16Khz16BitMonoPcm
-        case 22050:
-            return SpeechSynthesisOutputFormat.Raw22050Hz16BitMonoPcm
-        case 24000:
-            return SpeechSynthesisOutputFormat.Raw24Khz16BitMonoPcm
-        case 44100:
-            return SpeechSynthesisOutputFormat.Raw44100Hz16BitMonoPcm
-        case 48000:
-            return SpeechSynthesisOutputFormat.Raw48Khz16BitMonoPcm
-    return SpeechSynthesisOutputFormat.Raw16Khz16BitMonoPcm
-
-
 class AzureBaseTTSService(TTSService):
    class InputParams(BaseModel):
        emphasis: Optional[str] = None
@@ -134,7 +178,7 @@ class AzureBaseTTSService(TTSService):
            "emphasis": params.emphasis,
            "language": self.language_to_service_language(params.language)
            if params.language
-            else Language.EN_US,
+            else "en-US",
            "pitch": params.pitch,
            "rate": params.rate,
            "role": params.role,
@@ -152,92 +196,7 @@ class AzureBaseTTSService(TTSService):
        return True

    def language_to_service_language(self, language: Language) -> str | None:
-        match language:
-            case Language.BG:
-                return "bg-BG"
-            case Language.CA:
-                return "ca-ES"
-            case Language.ZH:
-                return "zh-CN"
-            case Language.ZH_TW:
-                return "zh-TW"
-            case Language.CS:
-                return "cs-CZ"
-            case Language.DA:
-                return "da-DK"
-            case Language.NL:
-                return "nl-NL"
-            case Language.EN | Language.EN_US:
-                return "en-US"
-            case Language.EN_AU:
-                return "en-AU"
-            case Language.EN_GB:
-                return "en-GB"
-            case Language.EN_NZ:
-                return "en-NZ"
-            case Language.EN_IN:
-                return "en-IN"
-            case Language.ET:
-                return "et-EE"
-            case Language.FI:
-                return "fi-FI"
-            case Language.NL_BE:
-                return "nl-BE"
-            case Language.FR:
-                return "fr-FR"
-            case Language.FR_CA:
-                return "fr-CA"
-            case Language.DE:
-                return "de-DE"
-            case Language.DE_CH:
-                return "de-CH"
-            case Language.EL:
-                return "el-GR"
-            case Language.HI:
-                return "hi-IN"
-            case Language.HU:
-                return "hu-HU"
-            case Language.ID:
-                return "id-ID"
-            case Language.IT:
-                return "it-IT"
-            case Language.JA:
-                return "ja-JP"
-            case Language.KO:
-                return "ko-KR"
-            case Language.LV:
-                return "lv-LV"
-            case Language.LT:
-                return "lt-LT"
-            case Language.MS:
-                return "ms-MY"
-            case Language.NO:
-                return "nb-NO"
-            case Language.PL:
-                return "pl-PL"
-            case Language.PT:
-                return "pt-PT"
-            case Language.PT_BR:
-                return "pt-BR"
-            case Language.RO:
-                return "ro-RO"
-            case Language.RU:
-                return "ru-RU"
-            case Language.SK:
-                return "sk-SK"
-            case Language.ES:
-                return "es-ES"
-            case Language.SV:
-                return "sv-SE"
-            case Language.TH:
-                return "th-TH"
-            case Language.TR:
-                return "tr-TR"
-            case Language.UK:
-                return "uk-UA"
-            case Language.VI:
-                return "vi-VN"
-        return None
+        return language_to_azure_language(language)

    def _construct_ssml(self, text: str) -> str:
        language = self._settings["language"]
--- a/src/pipecat/services/cartesia.py
+++ b/src/pipecat/services/cartesia.py
@@ -44,29 +44,24 @@ except ModuleNotFoundError as e:


 def language_to_cartesia_language(language: Language) -> str | None:
-    match language:
-        case Language.DE:
-            return "de"
-        case (
-            Language.EN
-            | Language.EN_US
-            | Language.EN_GB
-            | Language.EN_AU
-            | Language.EN_NZ
-            | Language.EN_IN
-        ):
-            return "en"
-        case Language.ES:
-            return "es"
-        case Language.FR | Language.FR_CA:
-            return "fr"
-        case Language.JA:
-            return "ja"
-        case Language.PT | Language.PT_BR:
-            return "pt"
-        case Language.ZH | Language.ZH_TW:
-            return "zh"
-    return None
+    language_map = {
+        Language.DE: "de",
+        Language.EN: "en",
+        Language.EN_US: "en",
+        Language.EN_GB: "en",
+        Language.EN_AU: "en",
+        Language.EN_NZ: "en",
+        Language.EN_IN: "en",
+        Language.ES: "es",
+        Language.FR: "fr",
+        Language.FR_CA: "fr",
+        Language.JA: "ja",
+        Language.PT: "pt",
+        Language.PT_BR: "pt",
+        Language.ZH: "zh",
+        Language.ZH_TW: "zh",
+    }
+    return language_map.get(language)


 class CartesiaTTSService(WordTTSService):
@@ -117,7 +112,7 @@ class CartesiaTTSService(WordTTSService):
            },
            "language": self.language_to_service_language(params.language)
            if params.language
-            else Language.EN,
+            else "en",
            "speed": params.speed,
            "emotion": params.emotion,
        }
@@ -331,7 +326,7 @@ class CartesiaHttpTTSService(TTSService):
            },
            "language": self.language_to_service_language(params.language)
            if params.language
-            else Language.EN,
+            else "en",
            "speed": params.speed,
            "emotion": params.emotion,
        }
--- a/src/pipecat/services/elevenlabs.py
+++ b/src/pipecat/services/elevenlabs.py
@@ -42,6 +42,48 @@ except ModuleNotFoundError as e:
 ElevenLabsOutputFormat = Literal["pcm_16000", "pcm_22050", "pcm_24000", "pcm_44100"]


+def language_to_elevenlabs_language(language: Language) -> str | None:
+    language_map = {
+        Language.BG: "bg",
+        Language.ZH: "zh",
+        Language.CS: "cs",
+        Language.DA: "da",
+        Language.NL: "nl",
+        Language.EN: "en",
+        Language.EN_US: "en",
+        Language.EN_AU: "en",
+        Language.EN_GB: "en",
+        Language.EN_NZ: "en",
+        Language.EN_IN: "en",
+        Language.FI: "fi",
+        Language.FR: "fr",
+        Language.FR_CA: "fr",
+        Language.DE: "de",
+        Language.DE_CH: "de",
+        Language.EL: "el",
+        Language.HI: "hi",
+        Language.HU: "hu",
+        Language.ID: "id",
+        Language.IT: "it",
+        Language.JA: "ja",
+        Language.KO: "ko",
+        Language.MS: "ms",
+        Language.NO: "no",
+        Language.PL: "pl",
+        Language.PT: "pt-PT",
+        Language.PT_BR: "pt-BR",
+        Language.RO: "ro",
+        Language.RU: "ru",
+        Language.SK: "sk",
+        Language.ES: "es",
+        Language.SV: "sv",
+        Language.TR: "tr",
+        Language.UK: "uk",
+        Language.VI: "vi",
+    }
+    return language_map.get(language)
+
+
 def sample_rate_from_output_format(output_format: str) -> int:
    match output_format:
        case "pcm_16000":
@@ -135,7 +177,7 @@ class ElevenLabsTTSService(WordTTSService):
            "sample_rate": sample_rate_from_output_format(output_format),
            "language": self.language_to_service_language(params.language)
            if params.language
-            else Language.EN,
+            else "en",
            "output_format": output_format,
            "optimize_streaming_latency": params.optimize_streaming_latency,
            "stability": params.stability,
@@ -158,73 +200,7 @@ class ElevenLabsTTSService(WordTTSService):
        return True

    def language_to_service_language(self, language: Language) -> str | None:
-        match language:
-            case Language.BG:
-                return "bg"
-            case Language.ZH:
-                return "zh"
-            case Language.CS:
-                return "cs"
-            case Language.DA:
-                return "da"
-            case Language.NL:
-                return "nl"
-            case (
-                Language.EN
-                | Language.EN_US
-                | Language.EN_AU
-                | Language.EN_GB
-                | Language.EN_NZ
-                | Language.EN_IN
-            ):
-                return "en"
-            case Language.FI:
-                return "fi"
-            case Language.FR | Language.FR_CA:
-                return "fr"
-            case Language.DE | Language.DE_CH:
-                return "de"
-            case Language.EL:
-                return "el"
-            case Language.HI:
-                return "hi"
-            case Language.HU:
-                return "hu"
-            case Language.ID:
-                return "id"
-            case Language.IT:
-                return "it"
-            case Language.JA:
-                return "ja"
-            case Language.KO:
-                return "ko"
-            case Language.MS:
-                return "ms"
-            case Language.NO:
-                return "no"
-            case Language.PL:
-                return "pl"
-            case Language.PT:
-                return "pt-PT"
-            case Language.PT_BR:
-                return "pt-BR"
-            case Language.RO:
-                return "ro"
-            case Language.RU:
-                return "ru"
-            case Language.SK:
-                return "sk"
-            case Language.ES:
-                return "es"
-            case Language.SV:
-                return "sv"
-            case Language.TR:
-                return "tr"
-            case Language.UK:
-                return "uk"
-            case Language.VI:
-                return "vi"
-        return None
+        return language_to_elevenlabs_language(language)

    def _set_voice_settings(self):
        voice_settings = {}
--- a/src/pipecat/services/gladia.py
+++ b/src/pipecat/services/gladia.py
@@ -34,6 +34,53 @@ except ModuleNotFoundError as e:
    raise Exception(f"Missing module: {e}")


+def language_to_gladia_language(language: Language) -> str | None:
+    language_map = {
+        Language.BG: "bg",
+        Language.CA: "ca",
+        Language.ZH: "zh",
+        Language.CS: "cs",
+        Language.DA: "da",
+        Language.NL: "nl",
+        Language.EN: "en",
+        Language.EN_US: "en",
+        Language.EN_AU: "en",
+        Language.EN_GB: "en",
+        Language.EN_NZ: "en",
+        Language.EN_IN: "en",
+        Language.ET: "et",
+        Language.FI: "fi",
+        Language.FR: "fr",
+        Language.FR_CA: "fr",
+        Language.DE: "de",
+        Language.DE_CH: "de",
+        Language.EL: "el",
+        Language.HI: "hi",
+        Language.HU: "hu",
+        Language.ID: "id",
+        Language.IT: "it",
+        Language.JA: "ja",
+        Language.KO: "ko",
+        Language.LV: "lv",
+        Language.LT: "lt",
+        Language.MS: "ms",
+        Language.NO: "no",
+        Language.PL: "pl",
+        Language.PT: "pt",
+        Language.PT_BR: "pt",
+        Language.RO: "ro",
+        Language.RU: "ru",
+        Language.SK: "sk",
+        Language.ES: "es",
+        Language.SV: "sv",
+        Language.TH: "th",
+        Language.TR: "tr",
+        Language.UK: "uk",
+        Language.VI: "vi",
+    }
+    return language_map.get(language)
+
+
 class GladiaSTTService(STTService):
    class InputParams(BaseModel):
        sample_rate: Optional[int] = 16000
@@ -79,50 +126,7 @@ class GladiaSTTService(STTService):
        self._confidence = confidence

    def language_to_service_language(self, language: Language) -> str | None:
-        language_map = {
-            Language.BG: "bg",
-            Language.CA: "ca",
-            Language.ZH: "zh",
-            Language.CS: "cs",
-            Language.DA: "da",
-            Language.NL: "nl",
-            Language.EN: "en",
-            Language.EN_US: "en",
-            Language.EN_AU: "en",
-            Language.EN_GB: "en",
-            Language.EN_NZ: "en",
-            Language.EN_IN: "en",
-            Language.ET: "et",
-            Language.FI: "fi",
-            Language.FR: "fr",
-            Language.FR_CA: "fr",
-            Language.DE: "de",
-            Language.DE_CH: "de",
-            Language.EL: "el",
-            Language.HI: "hi",
-            Language.HU: "hu",
-            Language.ID: "id",
-            Language.IT: "it",
-            Language.JA: "ja",
-            Language.KO: "ko",
-            Language.LV: "lv",
-            Language.LT: "lt",
-            Language.MS: "ms",
-            Language.NO: "no",
-            Language.PL: "pl",
-            Language.PT: "pt",
-            Language.PT_BR: "pt",
-            Language.RO: "ro",
-            Language.RU: "ru",
-            Language.SK: "sk",
-            Language.ES: "es",
-            Language.SV: "sv",
-            Language.TH: "th",
-            Language.TR: "tr",
-            Language.UK: "uk",
-            Language.VI: "vi",
-        }
-        return language_map.get(language)
+        return language_to_gladia_language(language)

    async def start(self, frame: StartFrame):
        await super().start(frame)
--- a/src/pipecat/services/google.py
+++ b/src/pipecat/services/google.py
@@ -16,6 +16,7 @@ from PIL import Image
 from pydantic import BaseModel, Field

 from pipecat.frames.frames import (
+    AudioRawFrame,
    ErrorFrame,
    Frame,
    LLMFullResponseEndFrame,
@@ -55,6 +56,53 @@ except ModuleNotFoundError as e:
    raise Exception(f"Missing module: {e}")


+def language_to_google_language(language: Language) -> str | None:
+    language_map = {
+        Language.BG: "bg-BG",
+        Language.CA: "ca-ES",
+        Language.ZH: "cmn-CN",
+        Language.ZH_TW: "cmn-TW",
+        Language.CS: "cs-CZ",
+        Language.DA: "da-DK",
+        Language.NL: "nl-NL",
+        Language.EN: "en-US",
+        Language.EN_US: "en-US",
+        Language.EN_AU: "en-AU",
+        Language.EN_GB: "en-GB",
+        Language.EN_IN: "en-IN",
+        Language.ET: "et-EE",
+        Language.FI: "fi-FI",
+        Language.NL_BE: "nl-BE",
+        Language.FR: "fr-FR",
+        Language.FR_CA: "fr-CA",
+        Language.DE: "de-DE",
+        Language.EL: "el-GR",
+        Language.HI: "hi-IN",
+        Language.HU: "hu-HU",
+        Language.ID: "id-ID",
+        Language.IT: "it-IT",
+        Language.JA: "ja-JP",
+        Language.KO: "ko-KR",
+        Language.LV: "lv-LV",
+        Language.LT: "lt-LT",
+        Language.MS: "ms-MY",
+        Language.NO: "nb-NO",
+        Language.PL: "pl-PL",
+        Language.PT: "pt-PT",
+        Language.PT_BR: "pt-BR",
+        Language.RO: "ro-RO",
+        Language.RU: "ru-RU",
+        Language.SK: "sk-SK",
+        Language.ES: "es-ES",
+        Language.SV: "sv-SE",
+        Language.TH: "th-TH",
+        Language.TR: "tr-TR",
+        Language.UK: "uk-UA",
+        Language.VI: "vi-VN",
+    }
+    return language_map.get(language)
+
+
 class GoogleUserContextAggregator(OpenAIUserContextAggregator):
    async def _push_aggregation(self):
        if len(self._aggregation) > 0:
@@ -184,11 +232,53 @@ class GoogleLLMContext(OpenAILLMContext):
            msgs.append(obj)
        return msgs

+    def add_image_frame_message(
+        self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
+    ):
+        buffer = io.BytesIO()
+        Image.frombytes(format, size, image).save(buffer, format="JPEG")
+
+        parts = []
+        if text:
+            parts.append(glm.Part(text=text))
+        parts.append(
+            glm.Part(inline_data=glm.Blob(mime_type="image/jpeg", data=buffer.getvalue())),
+        )
+        self.add_message(glm.Content(role="user", parts=parts))
+
+    def add_audio_frames_message(self, *, audio_frames: list[AudioRawFrame], text: str = None):
+        if not audio_frames:
+            return
+
+        sample_rate = audio_frames[0].sample_rate
+        num_channels = audio_frames[0].num_channels
+
+        parts = []
+        data = b"".join(frame.audio for frame in audio_frames)
+        if text:
+            parts.append(glm.Part(text=text))
+        parts.append(
+            glm.Part(
+                inline_data=glm.Blob(
+                    mime_type="audio/wav",
+                    data=(
+                        bytes(
+                            self.create_wav_header(sample_rate, num_channels, 16, len(data)) + data
+                        )
+                    ),
+                )
+            ),
+        )
+        self.add_message(glm.Content(role="user", parts=parts))
+        # message = {"mime_type": "audio/mp3", "data": bytes(data + create_wav_header(sample_rate, num_channels, 16, len(data)))}
+        # self.add_message(message)
+
    def from_standard_message(self, message):
        role = message["role"]
        content = message.get("content", [])
        if role == "system":
-            role = "user"
+            self.system_message = content
+            return None
        elif role == "assistant":
            role = "model"

@@ -232,20 +322,6 @@ class GoogleLLMContext(OpenAILLMContext):
        message = glm.Content(role=role, parts=parts)
        return message

-    def add_image_frame_message(
-        self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
-    ):
-        buffer = io.BytesIO()
-        Image.frombytes(format, size, image).save(buffer, format="JPEG")
-
-        parts = []
-        if text:
-            parts.append(glm.Part(text=text))
-        parts.append(
-            glm.Part(inline_data=glm.Blob(mime_type="image/jpeg", data=buffer.getvalue())),
-        )
-        self.add_message(glm.Content(role="user", parts=parts))
-
    def to_standard_messages(self, obj) -> list:
        msg = {"role": obj.role, "content": []}
        if msg["role"] == "model":
@@ -289,9 +365,20 @@ class GoogleLLMContext(OpenAILLMContext):
        return [msg]

    def _restructure_from_openai_messages(self):
+        self.system_message = None
        # first, map across self._messages calling self.from_standard_message(m) to modify messages in place
        try:
-            self._messages[:] = [self.from_standard_message(m) for m in self._messages]
+            self._messages[:] = [
+                msg
+                for msg in (self.from_standard_message(m) for m in self._messages)
+                if msg is not None
+            ]
+            # We might have been given a messages list with only a system message. If so, let's put that back in
+            # the messages list as a user message.
+            if self.system_message and not self._messages:
+                self.add_message(
+                    glm.Content(role="user", parts=[glm.Part(text=self.system_message)])
+                )
        except Exception as e:
            logger.error(f"Error mapping messages: {e}")
        # iterate over messages and remove any messages that have an empty content list
@@ -319,11 +406,14 @@ class GoogleLLMService(LLMService):
        api_key: str,
        model: str = "gemini-1.5-flash-latest",
        params: InputParams = InputParams(),
+        system_instruction: Optional[str] = None,
        **kwargs,
    ):
        super().__init__(**kwargs)
        gai.configure(api_key=api_key)
-        self._create_client(model)
+        self.set_model_name(model)
+        self._system_instruction = system_instruction
+        self._create_client()
        self._settings = {
            "max_tokens": params.max_tokens,
            "temperature": params.temperature,
@@ -335,34 +425,10 @@ class GoogleLLMService(LLMService):
    def can_generate_metrics(self) -> bool:
        return True

-    def _create_client(self, model: str):
-        self.set_model_name(model)
-        self._client = gai.GenerativeModel(model)
-
-    def _get_messages_from_openai_context(self, context: OpenAILLMContext) -> List[glm.Content]:
-        openai_messages = context.get_messages()
-        google_messages = []
-
-        for message in openai_messages:
-            role = message["role"]
-            content = message["content"]
-            if role == "system":
-                role = "user"
-            elif role == "assistant":
-                role = "model"
-
-            parts = [glm.Part(text=content)]
-            if "mime_type" in message:
-                parts.append(
-                    glm.Part(
-                        inline_data=glm.Blob(
-                            mime_type=message["mime_type"], data=message["data"].getvalue()
-                        )
-                    )
-                )
-            google_messages.append({"role": role, "parts": parts})
-
-        return google_messages
+    def _create_client(self):
+        self._client = gai.GenerativeModel(
+            self._model_name, system_instruction=self._system_instruction
+        )

    async def _async_generator_wrapper(self, sync_generator):
        for item in sync_generator:
@@ -374,10 +440,11 @@ class GoogleLLMService(LLMService):
        try:
            logger.debug(f"Generating chat: {context.get_messages_for_logging()}")

-            # todo: move this into the new context code structure, convert from openai context one time
-            # todo: add system instructions
-            # messages = self._get_messages_from_openai_context(context)
            messages = context.messages
+            if self._system_instruction != context.system_message:
+                logger.debug(f"System instruction changed: {context.system_message}")
+                self._system_instruction = context.system_message
+                self._create_client()

            # Filter out None values and create GenerationConfig
            generation_params = {
@@ -394,24 +461,21 @@ class GoogleLLMService(LLMService):
            generation_config = GenerationConfig(**generation_params) if generation_params else None

            await self.start_ttfb_metrics()
-
            tools = context.tools if context.tools else []
            response = self._client.generate_content(
                contents=messages, tools=tools, stream=True, generation_config=generation_config
            )
-
-            tokens = LLMTokenUsage(
-                prompt_tokens=response.usage_metadata.prompt_token_count,
-                completion_tokens=response.usage_metadata.candidates_token_count,
-                total_tokens=response.usage_metadata.total_token_count,
-            )
-
-            await self.start_llm_usage_metrics(tokens)
-
            await self.stop_ttfb_metrics()

+            prompt_tokens = response.usage_metadata.prompt_token_count
+            completion_tokens = response.usage_metadata.candidates_token_count
+            total_tokens = response.usage_metadata.total_token_count
+
            async for chunk in self._async_generator_wrapper(response):
-                # todo: usage
+                if chunk.usage_metadata:
+                    prompt_tokens += response.usage_metadata.prompt_token_count
+                    completion_tokens += response.usage_metadata.candidates_token_count
+                    total_tokens += response.usage_metadata.total_token_count
                try:
                    for c in chunk.parts:
                        if c.text:
@@ -436,6 +500,13 @@ class GoogleLLMService(LLMService):
        except Exception as e:
            logger.exception(f"{self} exception: {e}")
        finally:
+            await self.start_llm_usage_metrics(
+                LLMTokenUsage(
+                    prompt_tokens=prompt_tokens,
+                    completion_tokens=completion_tokens,
+                    total_tokens=total_tokens,
+                )
+            )
            await self.push_frame(LLMFullResponseEndFrame())

    async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -499,7 +570,7 @@ class GoogleTTSService(TTSService):
            "emphasis": params.emphasis,
            "language": self.language_to_service_language(params.language)
            if params.language
-            else Language.EN,
+            else "en-US",
            "gender": params.gender,
            "google_style": params.google_style,
        }
@@ -530,88 +601,7 @@ class GoogleTTSService(TTSService):
        return True

    def language_to_service_language(self, language: Language) -> str | None:
-        match language:
-            case Language.BG:
-                return "bg-BG"
-            case Language.CA:
-                return "ca-ES"
-            case Language.ZH:
-                return "cmn-CN"
-            case Language.ZH_TW:
-                return "cmn-TW"
-            case Language.CS:
-                return "cs-CZ"
-            case Language.DA:
-                return "da-DK"
-            case Language.NL:
-                return "nl-NL"
-            case Language.EN | Language.EN_US:
-                return "en-US"
-            case Language.EN_AU:
-                return "en-AU"
-            case Language.EN_GB:
-                return "en-GB"
-            case Language.EN_IN:
-                return "en-IN"
-            case Language.ET:
-                return "et-EE"
-            case Language.FI:
-                return "fi-FI"
-            case Language.NL_BE:
-                return "nl-BE"
-            case Language.FR:
-                return "fr-FR"
-            case Language.FR_CA:
-                return "fr-CA"
-            case Language.DE:
-                return "de-DE"
-            case Language.EL:
-                return "el-GR"
-            case Language.HI:
-                return "hi-IN"
-            case Language.HU:
-                return "hu-HU"
-            case Language.ID:
-                return "id-ID"
-            case Language.IT:
-                return "it-IT"
-            case Language.JA:
-                return "ja-JP"
-            case Language.KO:
-                return "ko-KR"
-            case Language.LV:
-                return "lv-LV"
-            case Language.LT:
-                return "lt-LT"
-            case Language.MS:
-                return "ms-MY"
-            case Language.NO:
-                return "nb-NO"
-            case Language.PL:
-                return "pl-PL"
-            case Language.PT:
-                return "pt-PT"
-            case Language.PT_BR:
-                return "pt-BR"
-            case Language.RO:
-                return "ro-RO"
-            case Language.RU:
-                return "ru-RU"
-            case Language.SK:
-                return "sk-SK"
-            case Language.ES:
-                return "es-ES"
-            case Language.SV:
-                return "sv-SE"
-            case Language.TH:
-                return "th-TH"
-            case Language.TR:
-                return "tr-TR"
-            case Language.UK:
-                return "uk-UA"
-            case Language.VI:
-                return "vi-VN"
-        return None
+        return language_to_google_language(language)

    def _construct_ssml(self, text: str) -> str:
        ssml = "<speak>"
--- a/src/pipecat/services/lmnt.py
+++ b/src/pipecat/services/lmnt.py
@@ -35,6 +35,27 @@ except ModuleNotFoundError as e:
    raise Exception(f"Missing module: {e}")


+def language_to_lmnt_language(language: Language) -> str | None:
+    language_map = {
+        Language.DE: "de",
+        Language.EN: "en",
+        Language.EN_US: "en",
+        Language.EN_AU: "en",
+        Language.EN_GB: "en",
+        Language.EN_NZ: "en",
+        Language.EN_IN: "en",
+        Language.ES: "es",
+        Language.FR: "fr",
+        Language.FR_CA: "fr",
+        Language.PT: "pt",
+        Language.PT_BR: "pt",
+        Language.ZH: "zh",
+        Language.ZH_TW: "zh",
+        Language.KO: "ko",
+    }
+    return language_map.get(language)
+
+
 class LmntTTSService(TTSService):
    def __init__(
        self,
@@ -72,29 +93,7 @@ class LmntTTSService(TTSService):
        return True

    def language_to_service_language(self, language: Language) -> str | None:
-        match language:
-            case Language.DE:
-                return "de"
-            case (
-                Language.EN
-                | Language.EN_US
-                | Language.EN_AU
-                | Language.EN_GB
-                | Language.EN_NZ
-                | Language.EN_IN
-            ):
-                return "en"
-            case Language.ES:
-                return "es"
-            case Language.FR | Language.FR_CA:
-                return "fr"
-            case Language.PT | Language.PT_BR:
-                return "pt"
-            case Language.ZH | Language.ZH_TW:
-                return "zh"
-            case Language.KO:
-                return "ko"
-        return None
+        return language_to_lmnt_language(language)

    async def start(self, frame: StartFrame):
        await super().start(frame)
--- a/src/pipecat/services/openai_realtime_beta/openai.py
+++ b/src/pipecat/services/openai_realtime_beta/openai.py
@@ -441,8 +441,6 @@ class OpenAIRealtimeBetaLLMService(LLMService):

    async def _handle_evt_speech_started(self, evt):
        await self._truncate_current_audio_response()
-        # todo: might need to guard sending these when we fully support using either openai
-        # turn detection of Pipecat turn detection
        await self._start_interruption()  # cancels this processor task
        await self.push_frame(StartInterruptionFrame())  # cancels downstream tasks
        await self.push_frame(UserStartedSpeakingFrame())
--- a/src/pipecat/services/playht.py
+++ b/src/pipecat/services/playht.py
@@ -47,63 +47,40 @@ except ModuleNotFoundError as e:


 def language_to_playht_language(language: Language) -> str | None:
-    match language:
-        case Language.BG:
-            return "BULGARIAN"
-        case Language.CA:
-            return "CATALAN"
-        case Language.CS:
-            return "CZECH"
-        case Language.DA:
-            return "DANISH"
-        case Language.DE:
-            return "GERMAN"
-        case (
-            Language.EN
-            | Language.EN_US
-            | Language.EN_GB
-            | Language.EN_AU
-            | Language.EN_NZ
-            | Language.EN_IN
-        ):
-            return "ENGLISH"
-        case Language.ES:
-            return "SPANISH"
-        case Language.FR | Language.FR_CA:
-            return "FRENCH"
-        case Language.EL:
-            return "GREEK"
-        case Language.HI:
-            return "HINDI"
-        case Language.HU:
-            return "HUNGARIAN"
-        case Language.ID:
-            return "INDONESIAN"
-        case Language.IT:
-            return "ITALIAN"
-        case Language.JA:
-            return "JAPANESE"
-        case Language.KO:
-            return "KOREAN"
-        case Language.MS:
-            return "MALAY"
-        case Language.NL:
-            return "DUTCH"
-        case Language.PL:
-            return "POLISH"
-        case Language.PT | Language.PT_BR:
-            return "PORTUGUESE"
-        case Language.RU:
-            return "RUSSIAN"
-        case Language.SV:
-            return "SWEDISH"
-        case Language.TH:
-            return "THAI"
-        case Language.TR:
-            return "TURKISH"
-        case Language.UK:
-            return "UKRAINIAN"
-    return None
+    language_map = {
+        Language.BG: "bulgarian",
+        Language.CA: "catalan",
+        Language.CS: "czech",
+        Language.DA: "danish",
+        Language.DE: "german",
+        Language.EN: "english",
+        Language.EN_US: "english",
+        Language.EN_GB: "english",
+        Language.EN_AU: "english",
+        Language.EN_NZ: "english",
+        Language.EN_IN: "english",
+        Language.ES: "spanish",
+        Language.FR: "french",
+        Language.FR_CA: "french",
+        Language.EL: "greek",
+        Language.HI: "hindi",
+        Language.HU: "hungarian",
+        Language.ID: "indonesian",
+        Language.IT: "italian",
+        Language.JA: "japanese",
+        Language.KO: "korean",
+        Language.MS: "malay",
+        Language.NL: "dutch",
+        Language.PL: "polish",
+        Language.PT: "portuguese",
+        Language.PT_BR: "portuguese",
+        Language.RU: "russian",
+        Language.SV: "swedish",
+        Language.TH: "thai",
+        Language.TR: "turkish",
+        Language.UK: "ukrainian",
+    }
+    return language_map.get(language)


 class PlayHTTTSService(TTSService):
@@ -118,7 +95,7 @@ class PlayHTTTSService(TTSService):
        api_key: str,
        user_id: str,
        voice_url: str,
-        voice_engine: str = "PlayHT3.0-mini",
+        voice_engine: str = "Play3.0-mini",
        sample_rate: int = 24000,
        output_format: str = "wav",
        params: InputParams = InputParams(),
@@ -140,7 +117,7 @@ class PlayHTTTSService(TTSService):
            "sample_rate": sample_rate,
            "language": self.language_to_service_language(params.language)
            if params.language
-            else Language.EN,
+            else "english",
            "output_format": output_format,
            "voice_engine": voice_engine,
            "speed": params.speed,
@@ -153,8 +130,7 @@ class PlayHTTTSService(TTSService):
        return True

    def language_to_service_language(self, language: Language) -> str | None:
-        # Keep your existing language mapping logic here
-        pass
+        return language_to_playht_language(language)

    async def start(self, frame: StartFrame):
        await super().start(frame)
@@ -234,17 +210,11 @@ class PlayHTTTSService(TTSService):

    async def _receive_task_handler(self):
        try:
-            header_size = 78  # Size of the WAV header + extra bytes we want to skip
-            header_received = False
            async for message in self._get_websocket():
                if isinstance(message, bytes):
-                    chunk_size = len(message)
-
-                    # Skip the WAV header
-                    if not header_received and chunk_size == header_size:
-                        header_received = True
+                    # Skip the WAV header message
+                    if message.startswith(b"RIFF"):
                        continue
-
                    await self.stop_ttfb_metrics()
                    frame = TTSAudioRawFrame(message, self._settings["sample_rate"], 1)
                    await self.push_frame(frame)
@@ -254,7 +224,6 @@ class PlayHTTTSService(TTSService):
                        msg = json.loads(message)
                        if "request_id" in msg and msg["request_id"] == self._request_id:
                            await self.push_frame(TTSStoppedFrame())
-                            header_received = False  # Reset for the next audio stream
                            self._request_id = None
                        elif "error" in msg:
                            logger.error(f"{self} error: {msg}")
@@ -334,7 +303,7 @@ class PlayHTHttpTTSService(TTSService):
        api_key: str,
        user_id: str,
        voice_url: str,
-        voice_engine: str = "PlayHT3.0-mini",
+        voice_engine: str = "Play3.0-mini",
        sample_rate: int = 24000,
        params: InputParams = InputParams(),
        **kwargs,
@@ -352,7 +321,7 @@ class PlayHTHttpTTSService(TTSService):
            "sample_rate": sample_rate,
            "language": self.language_to_service_language(params.language)
            if params.language
-            else Language.EN,
+            else "english",
            "format": Format.FORMAT_WAV,
            "voice_engine": voice_engine,
            "speed": params.speed,
--- a/src/pipecat/services/rime.py
+++ b/src/pipecat/services/rime.py
@@ -0,0 +1,101 @@
+from typing import AsyncGenerator, Optional
+
+import aiohttp
+from loguru import logger
+from pydantic import BaseModel
+
+from pipecat.frames.frames import (
+    ErrorFrame,
+    Frame,
+    TTSAudioRawFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
+)
+from pipecat.services.ai_services import TTSService
+
+
+class RimeHttpTTSService(TTSService):
+    class InputParams(BaseModel):
+        pause_between_brackets: Optional[bool] = False
+        phonemize_between_brackets: Optional[bool] = False
+        inline_speed_alpha: Optional[str] = None
+        speed_alpha: Optional[float] = 1.0
+        reduce_latency: Optional[bool] = False
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        voice_id: str = "eva",
+        model: str = "mist",
+        sample_rate: int = 24000,
+        params: InputParams = InputParams(),
+        **kwargs,
+    ):
+        super().__init__(sample_rate=sample_rate, **kwargs)
+
+        self._api_key = api_key
+        self._base_url = "https://users.rime.ai/v1/rime-tts"
+        self._settings = {
+            "speaker": voice_id,
+            "modelId": model,
+            "samplingRate": sample_rate,
+            "speedAlpha": params.speed_alpha,
+            "reduceLatency": params.reduce_latency,
+            "pauseBetweenBrackets": params.pause_between_brackets,
+            "phonemizeBetweenBrackets": params.phonemize_between_brackets,
+        }
+
+        if params.inline_speed_alpha:
+            self._settings["inlineSpeedAlpha"] = params.inline_speed_alpha
+
+    def can_generate_metrics(self) -> bool:
+        return True
+
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        logger.debug(f"Generating TTS: [{text}]")
+
+        headers = {
+            "Accept": "audio/pcm",
+            "Authorization": f"Bearer {self._api_key}",
+            "Content-Type": "application/json",
+        }
+
+        payload = self._settings.copy()
+        payload["text"] = text
+
+        try:
+            await self.start_ttfb_metrics()
+            await self.start_tts_usage_metrics(text)
+
+            yield TTSStartedFrame()
+
+            async with aiohttp.ClientSession() as session:
+                async with session.post(self._base_url, json=payload, headers=headers) as response:
+                    if response.status != 200:
+                        error_message = f"Rime TTS error: HTTP {response.status}"
+                        logger.error(error_message)
+                        yield ErrorFrame(error=error_message)
+                        return
+
+                    # Process the streaming response
+                    chunk_size = 8192
+                    first_chunk = True
+
+                    async for chunk in response.content.iter_chunked(chunk_size):
+                        if first_chunk:
+                            await self.stop_ttfb_metrics()
+                            first_chunk = False
+
+                        if chunk:
+                            frame = TTSAudioRawFrame(chunk, self._settings["samplingRate"], 1)
+                            yield frame
+
+            yield TTSStoppedFrame()
+
+        except Exception as e:
+            logger.exception(f"Error generating TTS: {e}")
+            yield ErrorFrame(error=f"Rime TTS error: {str(e)}")
+
+        finally:
+            yield TTSStoppedFrame()
--- a/src/pipecat/services/xtts.py
+++ b/src/pipecat/services/xtts.py
@@ -31,6 +31,34 @@ from loguru import logger
 # https://github.com/coqui-ai/xtts-streaming-server


+def language_to_xtts_language(language: Language) -> str | None:
+    language_map = {
+        Language.CS: "cs",
+        Language.DE: "de",
+        Language.EN: "en",
+        Language.EN_US: "en",
+        Language.EN_AU: "en",
+        Language.EN_GB: "en",
+        Language.EN_NZ: "en",
+        Language.EN_IN: "en",
+        Language.ES: "es",
+        Language.FR: "fr",
+        Language.HI: "hi",
+        Language.HU: "hu",
+        Language.IT: "it",
+        Language.JA: "ja",
+        Language.KO: "ko",
+        Language.NL: "nl",
+        Language.PL: "pl",
+        Language.PT: "pt",
+        Language.PT_BR: "pt",
+        Language.RU: "ru",
+        Language.TR: "tr",
+        Language.ZH: "zh-cn",
+    }
+    return language_map.get(language)
+
+
 class XTTSService(TTSService):
    def __init__(
        self,
@@ -56,47 +84,7 @@ class XTTSService(TTSService):
        return True

    def language_to_service_language(self, language: Language) -> str | None:
-        match language:
-            case Language.CS:
-                return "cs"
-            case Language.DE:
-                return "de"
-            case (
-                Language.EN
-                | Language.EN_US
-                | Language.EN_AU
-                | Language.EN_GB
-                | Language.EN_NZ
-                | Language.EN_IN
-            ):
-                return "en"
-            case Language.ES:
-                return "es"
-            case Language.FR:
-                return "fr"
-            case Language.HI:
-                return "hi"
-            case Language.HU:
-                return "hu"
-            case Language.IT:
-                return "it"
-            case Language.JA:
-                return "ja"
-            case Language.KO:
-                return "ko"
-            case Language.NL:
-                return "nl"
-            case Language.PL:
-                return "pl"
-            case Language.PT | Language.PT_BR:
-                return "pt"
-            case Language.RU:
-                return "ru"
-            case Language.TR:
-                return "tr"
-            case Language.ZH:
-                return "zh-cn"
-        return None
+        return language_to_xtts_language(language)

    async def start(self, frame: StartFrame):
        await super().start(frame)
Author	SHA1	Message	Date
Mark Backman	4249abc849	Fix missing await in simple chatbot example	2024-11-15 21:24:07 -05:00
Mark Backman	d0bca67666	Merge pull request #716 from pipecat-ai/mb/mute-stt-service Add STTMuteFilter to un/mute the STT	2024-11-14 19:55:00 -05:00
Mark Backman	966974bfc6	Change STTMuteProcessor to STTMuteFilter	2024-11-14 19:47:37 -05:00
Mark Backman	f807f233bd	Suppress UserStartedSpeakingFrame and UserStoppedSpeakingFrame when muted	2024-11-14 17:11:51 -05:00
Mark Backman	33108f5798	Code review feedback	2024-11-14 17:05:08 -05:00
Mark Backman	52de825af8	Update CHANGELOG	2024-11-14 13:47:08 -05:00
Mark Backman	5fe679039c	Add STTMuteProcessor to un/mute the STT	2024-11-14 13:35:02 -05:00
Kwindla Hultman Kramer	534f710f5d	Merge pull request #688 from pipecat-ai/khk/natural-conversation More work on llm-as-judge phrase endpointing	2024-11-14 09:15:16 -08:00
Mark Backman	53a11744a8	Merge pull request #712 from pipecat-ai/aleix/some-languages-tweaks some languages tweaks	2024-11-14 09:33:26 -05:00
Mark Backman	72412cc0c4	Code review feedback	2024-11-14 09:31:04 -05:00
Mark Backman	b77ac07bc6	Merge pull request #715 from pipecat-ai/mb/update-readme-2 Add visual divider below Pipecat README image	2024-11-14 08:54:25 -05:00
Mark Backman	eb6926e0ce	Add visual divider below Pipecat README image	2024-11-14 08:51:07 -05:00
Mark Backman	3b2c9de944	Merge pull request #713 from pipecat-ai/mb/update-readme Update README	2024-11-14 08:45:28 -05:00
Mark Backman	27ff868e5a	Move CONTRIBUTING to top directory	2024-11-14 08:43:03 -05:00
Mark Backman	57ef525a8e	Update README	2024-11-14 08:43:03 -05:00
Aleix Conchillo Flaqué	d1db54d5fe	examples(playht): use a 2.0 engine	2024-11-13 17:19:23 +01:00
Aleix Conchillo Flaqué	4f88fc0eb8	services(tts): initialize language to the proper language code	2024-11-13 17:19:23 +01:00
Aleix Conchillo Flaqué	37d1f4c4e1	services(tts): some language to service language cleanup	2024-11-13 17:19:23 +01:00
Aleix Conchillo Flaqué	ef9e86d997	services(playht): make sure we only skip wav header no matter the size	2024-11-13 17:19:23 +01:00
Aleix Conchillo Flaqué	2d2ef5a417	services(playht): voice engine is Play3.0-mini	2024-11-13 17:19:23 +01:00
Aleix Conchillo Flaqué	c1fff00586	services(playht): fix language codes	2024-11-13 17:19:23 +01:00
Mark Backman	0af2196f50	Merge pull request #708 from pipecat-ai/mb/add-rime-ai Add RimeTTSService	2024-11-12 18:29:53 -05:00
Mark Backman	cd42320788	Update changelog	2024-11-12 18:28:04 -05:00
Mark Backman	70fce52499	Merge pull request #710 from pipecat-ai/mb/update-readme-krisp Update Krisp README instructions	2024-11-12 11:15:25 -05:00
Mark Backman	70b60c0593	Update Krisp README instructions	2024-11-12 10:26:12 -05:00
Jon Taylor	2d8aa03f31	Merge pull request #706 from pipecat-ai/jpt/modal-example barebones modal.com deployment example	2024-11-12 11:41:00 +00:00
Kwindla Hultman Kramer	581ff26704	Merge pull request #707 from pipecat-ai/khk/clean-up tiny PR to remove old comment lines	2024-11-11 21:14:16 -08:00
Kwindla Hultman Kramer	335178ff06	some gemini audio input examples	2024-11-11 21:04:50 -08:00
Kwindla Hultman Kramer	ee53535f41	gemini audio-in with no transcription	2024-11-11 21:04:50 -08:00
Kwindla Hultman Kramer	91ac40307e	small fix and more prompt examples	2024-11-11 21:04:50 -08:00
Kwindla Hultman Kramer	b6c2c1f730	anthropic natural conversation example using claude haiku	2024-11-11 21:04:50 -08:00
Kwindla Hultman Kramer	b56c789ae4	fixes for proposed judge pipeline	2024-11-11 21:04:50 -08:00
Kwindla Hultman Kramer	bd435d9e62	missing commit	2024-11-11 21:04:50 -08:00
Kwindla Hultman Kramer	55a81df84f	contributing to llm-as-judge phrase endpointing work	2024-11-11 21:04:50 -08:00
Kwindla Hultman Kramer	87434460f5	temp hacking	2024-11-11 21:04:50 -08:00
Mark Backman	958ec42e8d	Add Rime.ai TTS service	2024-11-11 21:58:09 -05:00
Jon Taylor	d1fff60d1d	barebones modal.com deployment example	2024-11-11 22:30:07 +00:00
Kwindla Hultman Kramer	1438e5654a	remove old comment	2024-11-10 16:08:10 -08:00