testing pushing a frame from function call start hook

In this commit we make all frame processors asynchronous, that is, they have an
internal queue and they push frames using a task from that queue.

CHANGELOG.md

@@ -1,6 +1,6 @@
 # Changelog
 
-All notable changes to **Pipecat** will be documented in this file.
+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).
@@ -9,171 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Added `AssemblyAISTTService` and corresponding foundational examples
-  `07o-interruptible-assemblyai.py` and `13d-assemblyai-transcription.py`.
-
-- Added a foundational example for Gladia transcription:
-  `13c-gladia-transcription.py`
-
-### Fixed
-
-- Fixed `enable_usage_metrics` to control LLM/TTS usage metrics separately
-  from `enable_metrics`.
-
-## [0.0.46] - 2024-10-19
-
-### Added
-
-- Added `audio_passthrough` parameter to `STTService`. If enabled it allows
-  audio frames to be pushed downstream in case other processors need them.
-
-- Added input parameter options for `PlayHTTTSService` and
-  `PlayHTHttpTTSService`.
-
-### Changed
-
-- Moved `SileroVAD` audio processor to `processors.audio.vad`.
-
-- Module `utils.audio` is now `audio.utils`. A new `resample_audio` function has
-  been added.
-
-- `PlayHTTTSService` now uses PlayHT websockets instead of HTTP requests.
-
-- The previous `PlayHTTTSService` HTTP implementation is now
-  `PlayHTHttpTTSService`.
-
-- `PlayHTTTSService` and `PlayHTHttpTTSService` now use a `voice_engine` of
-  `PlayHT3.0-mini`, which allows for multi-lingual support.
-
-- Renamed `OpenAILLMServiceRealtimeBeta` to `OpenAIRealtimeBetaLLMService` to
-  match other services.
-
-### Deprecated
-
-- `LLMUserResponseAggregator` and `LLMAssistantResponseAggregator` are
-  mostly deprecated, use `OpenAILLMContext` instead.
-
-- The `vad` package is now deprecated and `audio.vad` should be used
-  instead. The `avd` package will get removed in a future release.
-
-### Fixed
-
-- Fixed an issue that would cause an error if no VAD analyzer was passed to
-  `LiveKitTransport` params.
-
-- Fixed `SileroVAD` processor to support interruptions properly.
-
-### Other
-
-- Added `examples/foundational/07-interruptible-vad.py`. This is the same as
-  `07-interruptible.py` but using the `SileroVAD` processor instead of passing
-  the `VADAnalyzer` in the transport.
-
-## [0.0.45] - 2024-10-16
-
-### Changed
-
-- Metrics messages have moved out from the transport's base output into RTVI.
-
-## [0.0.44] - 2024-10-15
-
-### Added
-
-- Added support for OpenAI Realtime API with the new
-  `OpenAILLMServiceRealtimeBeta` processor.
-  (see https://platform.openai.com/docs/guides/realtime/overview)
-
-- Added `RTVIBotTranscriptionProcessor` which will send the RTVI
-  `bot-transcription` protocol message. These are TTS text aggregated (into
-  sentences) messages.
-
-- Added new input params to the `MarkdownTextFilter` utility. You can set
-  `filter_code` to filter code from text and `filter_tables` to filter tables
-  from text.
-
-- Added `CanonicalMetricsService`. This processor uses the new
-  `AudioBufferProcessor` to capture conversation audio and later send it to
-  Canonical AI.
-  (see https://canonical.chat/)
-
-- Added `AudioBufferProcessor`. This processor can be used to buffer mixed user and
-  bot audio. This can later be saved into an audio file or processed by some
-  audio analyzer.
-
-- Added `on_first_participant_joined` event to `LiveKitTransport`.
-
-### Changed
-
-- LLM text responses are now logged properly as unicode characters.
-
-- `UserStartedSpeakingFrame`, `UserStoppedSpeakingFrame`,
-  `BotStartedSpeakingFrame`, `BotStoppedSpeakingFrame`, `BotSpeakingFrame` and
-  `UserImageRequestFrame` are now based from `SystemFrame`
-
-### Fixed
-
-- Merge `RTVIBotLLMProcessor`/`RTVIBotLLMTextProcessor` and
-  `RTVIBotTTSProcessor`/`RTVIBotTTSTextProcessor` to avoid out of order issues.
-
-- Fixed an issue in RTVI protocol that could cause a `bot-llm-stopped` or
-  `bot-tts-stopped` message to be sent before a `bot-llm-text` or `bot-tts-text`
-  message.
-
-- Fixed `DeepgramSTTService` constructor settings not being merged with default
-  ones.
-
-- Fixed an issue in Daily transport that would cause tasks to be hanging if
-  urgent transport messages were being sent from a transport event handler.
-
-- Fixed an issue in `BaseOutputTransport` that would cause `EndFrame` to be
-  pushed downed too early and call `FrameProcessor.cleanup()` before letting the
-  transport stop properly.
-
-## [0.0.43] - 2024-10-10
-
-### Added
-
-- Added a new util called `MarkdownTextFilter` which is a subclass of a new
-  base class called `BaseTextFilter`. This is a configurable utility which
-  is intended to filter text received by TTS services.
-
-- Added new `RTVIUserLLMTextProcessor`. This processor will send an RTVI
-  `user-llm-text` message with the user content's that was sent to the LLM.
-
-### Changed
-
-- `TransportMessageFrame` doesn't have an `urgent` field anymore, instead
-  there's now a `TransportMessageUrgentFrame` which is a `SystemFrame` and
-  therefore skip all internal queuing.
-
-- For TTS services, convert inputted languages to match each service's language
-  format
-
-### Fixed
-
-- Fixed an issue where changing a language with the Deepgram STT service
-  wouldn't apply the change. This was fixed by disconnecting and reconnecting
-  when the language changes.
-
-## [0.0.42] - 2024-10-02
-
-### Added
-
-- `SentryMetrics` has been added to report frame processor metrics to
-  Sentry. This is now possible because `FrameProcessorMetrics` can now be passed
-  to `FrameProcessor`.
-
-- Added Google TTS service and corresponding foundational example
-  `07n-interruptible-google.py`
+- Added Google TTS service and corresponding foundational example `07n-interruptible-google.py`
 
 - Added AWS Polly TTS support and `07m-interruptible-aws.py` as an example.
 
 - Added InputParams to Azure TTS service.
 
-- Added `LivekitTransport` (audio-only for now).
-
-- RTVI 0.2.0 is now supported.
-
 - All `FrameProcessors` can now register event handlers.
 
 ```
@@ -245,12 +86,8 @@ async def on_connected(processor):
 
 ### Changed
 
-- Context frames are now pushed downstream from assistant context aggregators.
-
-- Removed Silero VAD torch dependency.
-
-- Updated individual update settings frame classes into a single
-  `ServiceUpdateSettingsFrame` class.
+- Updated individual update settings frame classes into a single UpdateSettingsFrame
+  class for STT, LLM, and TTS.
 
 - We now distinguish between input and output audio and image frames. We
   introduce `InputAudioRawFrame`, `OutputAudioRawFrame`, `InputImageRawFrame`
@@ -270,9 +107,9 @@ async def on_connected(processor):
   pipelines is synchronous (e.g. an HTTP-based service that waits for the
   response).
 
-- `StartFrame` is back a system frame to make sure it's processed immediately by
-  all processors. `EndFrame` stays a control frame since it needs to be ordered
-  allowing the frames in the pipeline to be processed.
+- `StartFrame` is back a system frame so we make sure it's processed immediately
+  by all processors. `EndFrame` stays a control frame since it needs to be
+  ordered allowing the frames in the pipeline to be processed.
 
 - Updated `MoondreamService` revision to `2024-08-26`.
 
@@ -296,11 +133,6 @@ async def on_connected(processor):
 
 ### Fixed
 
-- Fixed OpenAI multiple function calls.
-
-- Fixed a Cartesia TTS issue that would cause audio to be truncated in some
-  cases.
-
 - Fixed a `BaseOutputTransport` issue that would stop audio and video rendering
   tasks (after receiving and `EndFrame`) before the internal queue was emptied,
   causing the pipeline to finish prematurely.
@@ -314,10 +146,6 @@ async def on_connected(processor):
 - `obj_id()` and `obj_count()` now use `itertools.count` avoiding the need of
   `threading.Lock`.
 
-### Other
-
-- Pipecat now uses Ruff as its formatter (https://github.com/astral-sh/ruff).
-
 ## [0.0.41] - 2024-08-22
 
 ### Added

README.md

@@ -38,7 +38,7 @@ 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:
 
-- **AI services**: `anthropic`, `assemblyai`, `aws`, `azure`, `deepgram`, `gladia`, `google`, `fal`, `lmnt`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper`, `xtts`
+- **AI services**: `anthropic`, `aws`, `azure`, `deepgram`, `gladia`, `google`, `fal`, `lmnt`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper`, `xtts`
 - **Transports**: `local`, `websocket`, `daily`
 
 ## Code examples
@@ -51,7 +51,10 @@ Your project may or may not need these, so they're made available as optional re
 Here is a very basic Pipecat bot that greets a user when they join a real-time session. We'll use [Daily](https://daily.co) for real-time media transport, and [Cartesia](https://cartesia.ai/) for text-to-speech.
 
 ```python
+#app.py
+
 import asyncio
+import aiohttp
 
 from pipecat.frames.frames import EndFrame, TextFrame
 from pipecat.pipeline.pipeline import Pipeline
@@ -61,43 +64,39 @@ from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
 async def main():
-  # Use Daily as a real-time media transport (WebRTC)
-  transport = DailyTransport(
-    room_url=...,
-    token=...,
-    bot_name="Bot Name",
-    params=DailyParams(audio_out_enabled=True))
+  async with aiohttp.ClientSession() as session:
+    # Use Daily as a real-time media transport (WebRTC)
+    transport = DailyTransport(
+      room_url=...,
+      token=...,
+      bot_name="Bot Name",
+      params=DailyParams(audio_out_enabled=True))
 
-  # Use Cartesia for Text-to-Speech
-  tts = CartesiaTTSService(
-    api_key=...,
-    voice_id=...
-  )
+    # Use Cartesia for Text-to-Speech
+    tts = CartesiaTTSService(
+        api_key=...,
+        voice_id=...
+      )
 
-  # Simple pipeline that will process text to speech and output the result
-  pipeline = Pipeline([tts, transport.output()])
+    # Simple pipeline that will process text to speech and output the result
+    pipeline = Pipeline([tts, transport.output()])
 
-  # Create Pipecat processor that can run one or more pipelines tasks
-  runner = PipelineRunner()
+    # Create Pipecat processor that can run one or more pipelines tasks
+    runner = PipelineRunner()
 
-  # Assign the task callable to run the pipeline
-  task = PipelineTask(pipeline)
+    # Assign the task callable to run the pipeline
+    task = PipelineTask(pipeline)
 
-  # Register an event handler to play audio when a
-  # participant joins the transport WebRTC session
-  @transport.event_handler("on_first_participant_joined")
-  async def on_first_participant_joined(transport, participant):
-    participant_name = participant.get("info", {}).get("userName", "")
-    # Queue a TextFrame that will get spoken by the TTS service (Cartesia)
-    await task.queue_frame(TextFrame(f"Hello there, {participant_name}!"))
+    # Register an event handler to play audio when a
+    # participant joins the transport WebRTC session
+    @transport.event_handler("on_participant_joined")
+    async def on_new_participant_joined(transport, participant):
+      participant_name = participant["info"]["userName"] or ''
+      # Queue a TextFrame that will get spoken by the TTS service (Cartesia)
+      await task.queue_frames([TextFrame(f"Hello there, {participant_name}!"), EndFrame()])
 
-  # Register an event handler to exit the application when the user leaves.
-  @transport.event_handler("on_participant_left")
-  async def on_participant_left(transport, participant, reason):
-    await task.queue_frame(EndFrame())
-
-  # Run the pipeline task
-  await runner.run(task)
+    # Run the pipeline task
+    await runner.run(task)
 
 if __name__ == "__main__":
   asyncio.run(main())
@@ -129,6 +128,8 @@ Pipecat makes use of WebRTC VAD by default when using a WebRTC transport layer.
 pip install pipecat-ai[silero]
 ```
 
+The first time your run your bot with Silero, startup may take a while whilst it downloads and caches the model in the background. You can check the progress of this in the console.
+
 ## 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:_

examples/canonical-metrics/.gitignore

@@ -1,161 +0,0 @@
-# Byte-compiled / optimized / DLL files
-__pycache__/
-*.py[cod]
-*$py.class
-recordings/
-# C extensions
-*.so
-
-# Distribution / packaging
-.Python
-build/
-develop-eggs/
-dist/
-downloads/
-eggs/
-.eggs/
-lib/
-lib64/
-parts/
-sdist/
-var/
-wheels/
-share/python-wheels/
-*.egg-info/
-.installed.cfg
-*.egg
-MANIFEST
-
-# PyInstaller
-#  Usually these files are written by a python script from a template
-#  before PyInstaller builds the exe, so as to inject date/other infos into it.
-*.manifest
-*.spec
-
-# Installer logs
-pip-log.txt
-pip-delete-this-directory.txt
-
-# Unit test / coverage reports
-htmlcov/
-.tox/
-.nox/
-.coverage
-.coverage.*
-.cache
-nosetests.xml
-coverage.xml
-*.cover
-*.py,cover
-.hypothesis/
-.pytest_cache/
-cover/
-
-# Translations
-*.mo
-*.pot
-
-# Django stuff:
-*.log
-local_settings.py
-db.sqlite3
-db.sqlite3-journal
-
-# Flask stuff:
-instance/
-.webassets-cache
-
-# Scrapy stuff:
-.scrapy
-
-# Sphinx documentation
-docs/_build/
-
-# PyBuilder
-.pybuilder/
-target/
-
-# Jupyter Notebook
-.ipynb_checkpoints
-
-# IPython
-profile_default/
-ipython_config.py
-
-# pyenv
-#   For a library or package, you might want to ignore these files since the code is
-#   intended to run in multiple environments; otherwise, check them in:
-# .python-version
-
-# pipenv
-#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
-#   However, in case of collaboration, if having platform-specific dependencies or dependencies
-#   having no cross-platform support, pipenv may install dependencies that don't work, or not
-#   install all needed dependencies.
-#Pipfile.lock
-
-# poetry
-#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
-#   This is especially recommended for binary packages to ensure reproducibility, and is more
-#   commonly ignored for libraries.
-#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
-#poetry.lock
-
-# pdm
-#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
-#pdm.lock
-#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
-#   in version control.
-#   https://pdm.fming.dev/#use-with-ide
-.pdm.toml
-
-# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
-__pypackages__/
-
-# Celery stuff
-celerybeat-schedule
-celerybeat.pid
-
-# SageMath parsed files
-*.sage.py
-
-# Environments
-.env
-.venv
-env/
-venv/
-ENV/
-env.bak/
-venv.bak/
-
-# Spyder project settings
-.spyderproject
-.spyproject
-
-# Rope project settings
-.ropeproject
-
-# mkdocs documentation
-/site
-
-# mypy
-.mypy_cache/
-.dmypy.json
-dmypy.json
-
-# Pyre type checker
-.pyre/
-
-# pytype static type analyzer
-.pytype/
-
-# Cython debug symbols
-cython_debug/
-
-# PyCharm
-#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
-#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
-#  and can be added to the global gitignore or merged into this file.  For a more nuclear
-#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
-runpod.toml

examples/canonical-metrics/Dockerfile

@@ -1,10 +0,0 @@
-FROM python:3.10-bullseye
-RUN mkdir /app
-COPY *.py /app/
-COPY requirements.txt /app/
-WORKDIR /app
-RUN pip3 install -r requirements.txt
-
-EXPOSE 7860
-
-CMD ["python3", "server.py"]

examples/canonical-metrics/README.md

@@ -1,37 +0,0 @@
-# Simple Chatbot
-
-<img src="image.png" width="420px">
-
-This app connects you to a chatbot powered by GPT-4, complete with animations generated by Stable Video Diffusion.
-
-See a video of it in action: https://x.com/kwindla/status/1778628911817183509
-
-And a quick video walkthrough of the code: https://www.loom.com/share/13df1967161f4d24ade054e7f8753416
-
-ℹ️ The first time, things might take extra time to get started since VAD (Voice Activity Detection) model needs to be downloaded.
-
-## Get started
-
-```python
-python3 -m venv venv
-source venv/bin/activate
-pip install -r requirements.txt
-
-cp env.example .env # and add your credentials
-
-```
-
-## Run the server
-
-```bash
-python server.py
-```
-
-Then, visit `http://localhost:7860/` in your browser to start a chatbot session.
-
-## Build and test the Docker image
-
-```
-docker build -t chatbot .
-docker run --env-file .env -p 7860:7860 chatbot
-```

examples/canonical-metrics/bot.py

@@ -1,146 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import os
-import sys
-import uuid
-
-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 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.processors.audio.audio_buffer_processor import AudioBufferProcessor
-from pipecat.services.canonical import CanonicalMetricsService
-from pipecat.services.elevenlabs import ElevenLabsTTSService
-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, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Chatbot",
-            DailyParams(
-                audio_out_enabled=True,
-                audio_in_enabled=True,
-                camera_out_enabled=False,
-                vad_enabled=True,
-                vad_audio_passthrough=True,
-                vad_analyzer=SileroVADAnalyzer(),
-                transcription_enabled=True,
-                #
-                # Spanish
-                #
-                # transcription_settings=DailyTranscriptionSettings(
-                #     language="es",
-                #     tier="nova",
-                #     model="2-general"
-                # )
-            ),
-        )
-
-        tts = ElevenLabsTTSService(
-            api_key=os.getenv("ELEVENLABS_API_KEY"),
-            #
-            # English
-            #
-            voice_id="cgSgspJ2msm6clMCkdW9",
-            aiohttp_session=session,
-            #
-            # Spanish
-            #
-            # model="eleven_multilingual_v2",
-            # voice_id="gD1IexrzCvsXPHUuT0s3",
-        )
-
-        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
-
-        messages = [
-            {
-                "role": "system",
-                #
-                # English
-                #
-                "content": "You are Chatbot, a friendly, helpful robot. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by introducing yourself. Keep all your responses to 12 words or fewer.",
-                #
-                # Spanish
-                #
-                # "content": "Eres Chatbot, un amigable y útil robot. Tu objetivo es demostrar tus capacidades de una manera breve. Tus respuestas se convertiran a audio así que nunca no debes incluir caracteres especiales. Contesta a lo que el usuario pregunte de una manera creativa, útil y breve. Empieza por presentarte a ti mismo.",
-            },
-        ]
-
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        """
-        CanonicalMetrics uses AudioBufferProcessor under the hood to buffer the audio. On
-        call completion, CanonicalMetrics will send the audio buffer to Canonical for
-        analysis. Visit https://voice.canonical.chat to learn more.
-        """
-        audio_buffer_processor = AudioBufferProcessor()
-        canonical = CanonicalMetricsService(
-            audio_buffer_processor=audio_buffer_processor,
-            aiohttp_session=session,
-            api_key=os.getenv("CANONICAL_API_KEY"),
-            api_url=os.getenv("CANONICAL_API_URL"),
-            call_id=str(uuid.uuid4()),
-            assistant="pipecat-chatbot",
-            assistant_speaks_first=True,
-        )
-        pipeline = Pipeline(
-            [
-                transport.input(),  # microphone
-                context_aggregator.user(),
-                llm,
-                tts,
-                transport.output(),
-                audio_buffer_processor,  # captures audio into a buffer
-                canonical,  # uploads audio buffer to Canonical AI for metrics
-                context_aggregator.assistant(),
-            ]
-        )
-
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        @transport.event_handler("on_participant_left")
-        async def on_participant_left(transport, participant, reason):
-            print(f"Participant left: {participant}")
-            await task.queue_frame(EndFrame())
-
-        @transport.event_handler("on_call_state_updated")
-        async def on_call_state_updated(transport, state):
-            if state == "left":
-                await task.queue_frame(EndFrame())
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/canonical-metrics/env.example

@@ -1,6 +0,0 @@
-DAILY_SAMPLE_ROOM_URL=https://yourdomain.daily.co/yourroom # (for joining the bot to the same room repeatedly for local dev)
-DAILY_API_KEY=7df...
-OPENAI_API_KEY=sk-PL...
-ELEVENLABS_API_KEY=aeb...
-CANONICAL_API_KEY=can...
-CANONICAL_API_URL=

examples/canonical-metrics/requirements.txt

@@ -1,5 +0,0 @@
-python-dotenv
-fastapi[all]
-uvicorn
-pipecat-ai[daily,openai,silero,elevenlabs,canonical]
-

examples/canonical-metrics/runner.py

@@ -1,56 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import argparse
-import os
-
-import aiohttp
-
-from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
-
-
-async def configure(aiohttp_session: aiohttp.ClientSession):
-    parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
-    parser.add_argument(
-        "-u", "--url", type=str, required=False, help="URL of the Daily room to join"
-    )
-    parser.add_argument(
-        "-k",
-        "--apikey",
-        type=str,
-        required=False,
-        help="Daily API Key (needed to create an owner token for the room)",
-    )
-
-    args, unknown = parser.parse_known_args()
-
-    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
-    key = args.apikey or os.getenv("DAILY_API_KEY")
-
-    if not url:
-        raise Exception(
-            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
-        )
-
-    if not key:
-        raise Exception(
-            "No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
-        )
-
-    daily_rest_helper = DailyRESTHelper(
-        daily_api_key=key,
-        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
-        aiohttp_session=aiohttp_session,
-    )
-
-    # Create a meeting token for the given room with an expiration 1 hour in
-    # the future.
-    expiry_time: float = 60 * 60
-
-    token = await daily_rest_helper.get_token(url, expiry_time)
-
-    return (url, token)
-    return (url, token)

examples/canonical-metrics/server.py

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

examples/chatbot-audio-recording/.gitignore

@@ -1,161 +0,0 @@
-# Byte-compiled / optimized / DLL files
-__pycache__/
-*.py[cod]
-*$py.class
-
-# C extensions
-*.so
-
-# Distribution / packaging
-.Python
-build/
-develop-eggs/
-dist/
-downloads/
-eggs/
-.eggs/
-lib/
-lib64/
-parts/
-sdist/
-var/
-wheels/
-share/python-wheels/
-*.egg-info/
-.installed.cfg
-*.egg
-MANIFEST
-
-# PyInstaller
-#  Usually these files are written by a python script from a template
-#  before PyInstaller builds the exe, so as to inject date/other infos into it.
-*.manifest
-*.spec
-
-# Installer logs
-pip-log.txt
-pip-delete-this-directory.txt
-
-# Unit test / coverage reports
-htmlcov/
-.tox/
-.nox/
-.coverage
-.coverage.*
-.cache
-nosetests.xml
-coverage.xml
-*.cover
-*.py,cover
-.hypothesis/
-.pytest_cache/
-cover/
-
-# Translations
-*.mo
-*.pot
-
-# Django stuff:
-*.log
-local_settings.py
-db.sqlite3
-db.sqlite3-journal
-
-# Flask stuff:
-instance/
-.webassets-cache
-
-# Scrapy stuff:
-.scrapy
-
-# Sphinx documentation
-docs/_build/
-
-# PyBuilder
-.pybuilder/
-target/
-
-# Jupyter Notebook
-.ipynb_checkpoints
-
-# IPython
-profile_default/
-ipython_config.py
-
-# pyenv
-#   For a library or package, you might want to ignore these files since the code is
-#   intended to run in multiple environments; otherwise, check them in:
-# .python-version
-
-# pipenv
-#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
-#   However, in case of collaboration, if having platform-specific dependencies or dependencies
-#   having no cross-platform support, pipenv may install dependencies that don't work, or not
-#   install all needed dependencies.
-#Pipfile.lock
-
-# poetry
-#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
-#   This is especially recommended for binary packages to ensure reproducibility, and is more
-#   commonly ignored for libraries.
-#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
-#poetry.lock
-
-# pdm
-#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
-#pdm.lock
-#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
-#   in version control.
-#   https://pdm.fming.dev/#use-with-ide
-.pdm.toml
-
-# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
-__pypackages__/
-
-# Celery stuff
-celerybeat-schedule
-celerybeat.pid
-
-# SageMath parsed files
-*.sage.py
-
-# Environments
-.env
-.venv
-env/
-venv/
-ENV/
-env.bak/
-venv.bak/
-
-# Spyder project settings
-.spyderproject
-.spyproject
-
-# Rope project settings
-.ropeproject
-
-# mkdocs documentation
-/site
-
-# mypy
-.mypy_cache/
-.dmypy.json
-dmypy.json
-
-# Pyre type checker
-.pyre/
-
-# pytype static type analyzer
-.pytype/
-
-# Cython debug symbols
-cython_debug/
-
-# PyCharm
-#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
-#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
-#  and can be added to the global gitignore or merged into this file.  For a more nuclear
-#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
-runpod.toml

examples/chatbot-audio-recording/Dockerfile

@@ -1,15 +0,0 @@
-FROM python:3.10-bullseye
-
-RUN mkdir /app
-RUN mkdir /app/assets
-RUN mkdir /app/utils
-COPY *.py /app/
-COPY requirements.txt /app/
-
-
-WORKDIR /app
-RUN pip3 install -r requirements.txt
-
-EXPOSE 7860
-
-CMD ["python3", "server.py"]

examples/chatbot-audio-recording/README.md

@@ -1,37 +0,0 @@
-# Simple Chatbot
-
-<img src="image.png" width="420px">
-
-This app connects you to a chatbot powered by GPT-4, complete with animations generated by Stable Video Diffusion.
-
-See a video of it in action: https://x.com/kwindla/status/1778628911817183509
-
-And a quick video walkthrough of the code: https://www.loom.com/share/13df1967161f4d24ade054e7f8753416
-
-ℹ️ The first time, things might take extra time to get started since VAD (Voice Activity Detection) model needs to be downloaded.
-
-## Get started
-
-```python
-python3 -m venv venv
-source venv/bin/activate
-pip install -r requirements.txt
-
-cp env.example .env # and add your credentials
-
-```
-
-## Run the server
-
-```bash
-python server.py
-```
-
-Then, visit `http://localhost:7860/` in your browser to start a chatbot session.
-
-## Build and test the Docker image
-
-```
-docker build -t chatbot .
-docker run --env-file .env -p 7860:7860 chatbot
-```

examples/chatbot-audio-recording/bot.py

@@ -1,141 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import os
-import sys
-
-import aiohttp
-import datetime
-import wave
-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 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.processors.audio.audio_buffer_processor import AudioBufferProcessor
-from pipecat.services.elevenlabs import ElevenLabsTTSService
-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 save_audio(audiobuffer):
-    if audiobuffer.has_audio():
-        merged_audio = audiobuffer.merge_audio_buffers()
-        filename = f"conversation_recording{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav"
-        with wave.open(filename, "wb") as wf:
-            wf.setnchannels(2)
-            wf.setsampwidth(2)
-            wf.setframerate(audiobuffer._sample_rate)
-            wf.writeframes(merged_audio)
-        print(f"Merged audio saved to {filename}")
-    else:
-        print("No audio data to save")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Chatbot",
-            DailyParams(
-                audio_out_enabled=True,
-                audio_in_enabled=True,
-                camera_out_enabled=False,
-                vad_enabled=True,
-                vad_audio_passthrough=True,
-                vad_analyzer=SileroVADAnalyzer(),
-                transcription_enabled=True,
-                #
-                # Spanish
-                #
-                # transcription_settings=DailyTranscriptionSettings(
-                #     language="es",
-                #     tier="nova",
-                #     model="2-general"
-                # )
-            ),
-        )
-
-        tts = ElevenLabsTTSService(
-            api_key=os.getenv("ELEVENLABS_API_KEY"),
-            #
-            # English
-            #
-            voice_id="cgSgspJ2msm6clMCkdW9",
-            aiohttp_session=session,
-            #
-            # Spanish
-            #
-            # model="eleven_multilingual_v2",
-            # voice_id="gD1IexrzCvsXPHUuT0s3",
-        )
-
-        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
-
-        messages = [
-            {
-                "role": "system",
-                #
-                # English
-                #
-                "content": "You are Chatbot, a friendly, helpful robot. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by introducing yourself. Keep all your response to 12 words or fewer.",
-                #
-                # Spanish
-                #
-                # "content": "Eres Chatbot, un amigable y útil robot. Tu objetivo es demostrar tus capacidades de una manera breve. Tus respuestas se convertiran a audio así que nunca no debes incluir caracteres especiales. Contesta a lo que el usuario pregunte de una manera creativa, útil y breve. Empieza por presentarte a ti mismo.",
-            },
-        ]
-
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        audiobuffer = AudioBufferProcessor()
-        pipeline = Pipeline(
-            [
-                transport.input(),  # microphone
-                context_aggregator.user(),
-                llm,
-                tts,
-                transport.output(),
-                audiobuffer,  # used to buffer the audio in the pipeline
-                context_aggregator.assistant(),
-            ]
-        )
-
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        @transport.event_handler("on_participant_left")
-        async def on_participant_left(transport, participant, reason):
-            print(f"Participant left: {participant}")
-            await task.queue_frame(EndFrame())
-            await save_audio(audiobuffer)
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/chatbot-audio-recording/env.example

@@ -1,4 +0,0 @@
-DAILY_SAMPLE_ROOM_URL=https://yourdomain.daily.co/yourroom # (for joining the bot to the same room repeatedly for local dev)
-DAILY_API_KEY=7df...
-OPENAI_API_KEY=sk-PL...
-ELEVENLABS_API_KEY=aeb...

examples/chatbot-audio-recording/requirements.txt

@@ -1,4 +0,0 @@
-python-dotenv
-fastapi[all]
-uvicorn
-pipecat-ai[daily,openai,silero,elevenlabs]

examples/chatbot-audio-recording/runner.py

@@ -1,56 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import argparse
-import os
-
-import aiohttp
-
-from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
-
-
-async def configure(aiohttp_session: aiohttp.ClientSession):
-    parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
-    parser.add_argument(
-        "-u", "--url", type=str, required=False, help="URL of the Daily room to join"
-    )
-    parser.add_argument(
-        "-k",
-        "--apikey",
-        type=str,
-        required=False,
-        help="Daily API Key (needed to create an owner token for the room)",
-    )
-
-    args, unknown = parser.parse_known_args()
-
-    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
-    key = args.apikey or os.getenv("DAILY_API_KEY")
-
-    if not url:
-        raise Exception(
-            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
-        )
-
-    if not key:
-        raise Exception(
-            "No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
-        )
-
-    daily_rest_helper = DailyRESTHelper(
-        daily_api_key=key,
-        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
-        aiohttp_session=aiohttp_session,
-    )
-
-    # Create a meeting token for the given room with an expiration 1 hour in
-    # the future.
-    expiry_time: float = 60 * 60
-
-    token = await daily_rest_helper.get_token(url, expiry_time)
-
-    return (url, token)
-    return (url, token)

examples/chatbot-audio-recording/server.py

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

examples/deployment/flyio-example/README.md

@@ -34,6 +34,6 @@ Note: you can do this manually via the fly.io dashboard under the "secrets" sub-
 
 Send a post request to your running fly.io instance:
 
-`curl --location --request POST 'https://YOUR_FLY_APP_NAME/'`
+`curl --location --request POST 'https://YOUR_FLY_APP_NAME/start_bot'`
 
 This request will wait until the machine enters into a `starting` state, before returning the a room URL and token to join.

examples/deployment/flyio-example/bot.py

@@ -3,15 +3,18 @@ import os
 import sys
 import argparse
 
-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.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.frames.frames import LLMMessagesFrame, EndFrame
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai import OpenAILLMService
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from loguru import logger
 
@@ -57,17 +60,17 @@ async def main(room_url: str, token: str):
         },
     ]
 
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    tma_in = LLMUserResponseAggregator(messages)
+    tma_out = LLMAssistantResponseAggregator(messages)
 
     pipeline = Pipeline(
         [
             transport.input(),
-            context_aggregator.user(),
+            tma_in,
             llm,
             tts,
             transport.output(),
-            context_aggregator.assistant(),
+            tma_out,
         ]
     )
 

examples/deployment/flyio-example/bot_runner.py

@@ -124,7 +124,7 @@ async def spawn_fly_machine(room_url: str, token: str):
     print(f"Machine joined room: {room_url}")
 
 
-@app.post("/")
+@app.post("/start_bot")
 async def start_bot(request: Request) -> JSONResponse:
     try:
         data = await request.json()

examples/dialin-chatbot/bot_daily.py

@@ -3,16 +3,18 @@ import os
 import sys
 import argparse
 
-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.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.frames.frames import LLMMessagesFrame, EndFrame
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport, DailyDialinSettings
-
+from pipecat.vad.silero import SileroVADAnalyzer
 from loguru import logger
 
 from dotenv import load_dotenv
@@ -63,17 +65,17 @@ async def main(room_url: str, token: str, callId: str, callDomain: str):
         },
     ]
 
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    tma_in = LLMUserResponseAggregator(messages)
+    tma_out = LLMAssistantResponseAggregator(messages)
 
     pipeline = Pipeline(
         [
             transport.input(),
-            context_aggregator.user(),
+            tma_in,
             llm,
             tts,
             transport.output(),
-            context_aggregator.assistant(),
+            tma_out,
         ]
     )
 

examples/dialin-chatbot/bot_runner.py

@@ -108,9 +108,11 @@ async def _create_daily_room(room_url, callId, callDomain=None, vendor="daily"):
     # Spawn a new agent, and join the user session
     # Note: this is mostly for demonstration purposes (refer to 'deployment' in docs)
     if vendor == "daily":
-        bot_proc = f"python3 -m bot_daily -u {room.url} -t {token} -i {callId} -d {callDomain}"
+        bot_proc = f"python3 - m bot_daily - u {room.url} - t {token} - i {
+            callId} - d {callDomain}"
     else:
-        bot_proc = f"python3 -m bot_twilio -u {room.url} -t {token} -i {callId} -s {room.config.sip_endpoint}"
+        bot_proc = f"python3 - m bot_twilio - u {room.url} - t {
+            token} - i {callId} - s {room.config.sip_endpoint}"
 
     try:
         subprocess.Popen(

examples/dialin-chatbot/bot_twilio.py

@@ -3,15 +3,18 @@ import os
 import sys
 import argparse
 
-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.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.frames.frames import LLMMessagesFrame, EndFrame
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from twilio.rest import Client
 
@@ -66,17 +69,17 @@ async def main(room_url: str, token: str, callId: str, sipUri: str):
         },
     ]
 
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    tma_in = LLMUserResponseAggregator(messages)
+    tma_out = LLMAssistantResponseAggregator(messages)
 
     pipeline = Pipeline(
         [
             transport.input(),
-            context_aggregator.user(),
+            tma_in,
             llm,
             tts,
             transport.output(),
-            context_aggregator.assistant(),
+            tma_out,
         ]
     )
 

examples/foundational/01-say-one-thing.py

@@ -47,15 +47,10 @@ async def main():
 
         # Register an event handler so we can play the audio when the
         # participant joins.
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            participant_name = participant.get("info", {}).get("userName", "")
-            await task.queue_frame(TextFrame(f"Hello there, {participant_name}!"))
-
-        # Register an event handler to exit the application when the user leaves.
-        @transport.event_handler("on_participant_left")
-        async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+        @transport.event_handler("on_participant_joined")
+        async def on_new_participant_joined(transport, participant):
+            participant_name = participant["info"]["userName"] or ""
+            await task.queue_frames([TextFrame(f"Hello there, {participant_name}!"), EndFrame()])
 
         await runner.run(task)
 

examples/foundational/01b-livekit-audio.py

@@ -4,6 +4,9 @@ import os
 import sys
 
 import aiohttp
+from dotenv import load_dotenv
+from livekit import api  # pip install livekit-api
+from loguru import logger
 
 from pipecat.frames.frames import TextFrame
 from pipecat.pipeline.pipeline import Pipeline
@@ -12,12 +15,6 @@ from pipecat.pipeline.task import PipelineTask
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.transports.services.livekit import LiveKitParams, LiveKitTransport
 
-from livekit import api
-
-from loguru import logger
-
-from dotenv import load_dotenv
-
 load_dotenv(override=True)
 
 logger.remove(0)

examples/foundational/02-llm-say-one-thing.py

@@ -57,11 +57,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await task.queue_frame(LLMMessagesFrame(messages))
-
-        @transport.event_handler("on_participant_left")
-        async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.queue_frames([LLMMessagesFrame(messages), EndFrame()])
 
         await runner.run(task)
 

examples/foundational/03-still-frame.py

@@ -9,7 +9,7 @@ import aiohttp
 import os
 import sys
 
-from pipecat.frames.frames import EndFrame, TextFrame
+from pipecat.frames.frames import TextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
@@ -51,11 +51,11 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await task.queue_frame(TextFrame("a cat in the style of picasso"))
-
-        @transport.event_handler("on_participant_left")
-        async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            # Note that we do not put an EndFrame() item in the pipeline for this demo.
+            # This means that the bot will stay in the channel until it times out.
+            # An EndFrame() in the pipeline would cause the transport to shut
+            # down.
+            await task.queue_frames([TextFrame("a cat in the style of picasso")])
 
         await runner.run(task)
 

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

@@ -82,7 +82,6 @@ async def main():
                         self.frame = OutputAudioRawFrame(
                             bytes(self.audio), frame.sample_rate, frame.num_channels
                         )
-                    await self.push_frame(frame, direction)
 
             class ImageGrabber(FrameProcessor):
                 def __init__(self):
@@ -94,7 +93,6 @@ async def main():
 
                     if isinstance(frame, URLImageRawFrame):
                         self.frame = frame
-                    await self.push_frame(frame, direction)
 
             llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
 

examples/foundational/06-listen-and-respond.py

@@ -9,7 +9,6 @@ import aiohttp
 import os
 import sys
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import Frame, LLMMessagesFrame, MetricsFrame
 from pipecat.metrics.metrics import (
     TTFBMetricsData,
@@ -19,12 +18,16 @@ from pipecat.metrics.metrics import (
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -89,19 +92,18 @@ async def main():
                 "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)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),
-                context_aggregator.user(),
+                tma_in,
                 llm,
                 tts,
                 ml,
                 transport.output(),
-                context_aggregator.assistant(),
+                tma_out,
             ]
         )
 

examples/foundational/06a-image-sync.py

@@ -11,16 +11,19 @@ import sys
 
 from PIL import Image
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import Frame, OutputImageRawFrame, SystemFrame, TextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.cartesia import CartesiaHttpTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from pipecat.transports.services.daily import DailyParams
 from runner import configure
@@ -102,8 +105,8 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         image_sync_aggregator = ImageSyncAggregator(
             os.path.join(os.path.dirname(__file__), "assets", "speaking.png"),
@@ -114,11 +117,11 @@ async def main():
             [
                 transport.input(),
                 image_sync_aggregator,
-                context_aggregator.user(),
+                tma_in,
                 llm,
                 tts,
                 transport.output(),
-                context_aggregator.assistant(),
+                tma_out,
             ]
         )
 
@@ -126,7 +129,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            participant_name = participant.get("info", {}).get("userName", "")
+            participant_name = participant["info"]["userName"] or ""
             transport.capture_participant_transcription(participant["id"])
             await task.queue_frames([TextFrame(f"Hi there {participant_name}!")])
 

examples/foundational/07-interruptible-vad.py

@@ -1,103 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import os
-import sys
-
-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.audio.vad.silero import SileroVAD
-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
-
-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")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_in_enabled=True,
-                audio_out_enabled=True,
-                transcription_enabled=True,
-            ),
-        )
-
-        vad = SileroVAD()
-
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
-
-        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(),
-                vad,
-                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):
-            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())

examples/foundational/07-interruptible.py

@@ -9,15 +9,18 @@ import aiohttp
 import os
 import sys
 
-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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -61,17 +64,17 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/07a-interruptible-anthropic.py

@@ -5,23 +5,28 @@
 #
 
 import asyncio
+import aiohttp
 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.anthropic import AnthropicLLMService
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.anthropic import AnthropicLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
 
 load_dotenv(override=True)
 
@@ -64,17 +69,17 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

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

@@ -10,7 +10,6 @@ import sys
 
 import aiohttp
 
-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
@@ -22,6 +21,7 @@ from pipecat.processors.aggregators.llm_response import (
 from pipecat.processors.frameworks.langchain import LangchainProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
 from langchain_community.chat_message_histories import ChatMessageHistory

examples/foundational/07c-interruptible-deepgram.py

@@ -13,15 +13,18 @@ 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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.deepgram import DeepgramSTTService, DeepgramTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 load_dotenv(override=True)
 
@@ -58,18 +61,18 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 stt,  # STT
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 
@@ -77,6 +80,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"])
             # Kick off the conversation.
             messages.append({"role": "system", "content": "Please introduce yourself to the user."})
             await task.queue_frames([LLMMessagesFrame(messages)])

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

@@ -11,17 +11,20 @@ import sys
 import aiohttp
 from dotenv import load_dotenv
 from loguru import logger
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 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.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 load_dotenv(override=True)
 
@@ -59,17 +62,17 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/07e-interruptible-playht.py

@@ -4,25 +4,29 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+import aiohttp
 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.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.playht import PlayHTTTSService
-from pipecat.transcriptions.language import Language
+from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
 
 load_dotenv(override=True)
 
@@ -51,7 +55,6 @@ async def main():
             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",
-            params=PlayHTTTSService.InputParams(language=Language.EN),
         )
 
         llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
@@ -63,29 +66,21 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 
-        task = PipelineTask(
-            pipeline,
-            PipelineParams(
-                allow_interruptions=True,
-                enable_metrics=True,
-                enable_usage_metrics=True,
-                report_only_initial_ttfb=True,
-            ),
-        )
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):

examples/foundational/07f-interruptible-azure.py

@@ -9,14 +9,17 @@ import asyncio
 import os
 import sys
 
-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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.azure import AzureLLMService, AzureSTTService, AzureTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 
 from runner import configure
@@ -71,18 +74,18 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 stt,  # STT
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/07g-interruptible-openai-tts.py

@@ -4,23 +4,29 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+import aiohttp
 import asyncio
 import os
 import sys
 
-import aiohttp
-from dotenv import load_dotenv
-from loguru import logger
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-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.services.openai import OpenAILLMService, OpenAITTSService
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
+from pipecat.services.openai import OpenAITTSService
+from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
 
 load_dotenv(override=True)
 
@@ -56,17 +62,17 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/07h-interruptible-openpipe.py

@@ -9,15 +9,18 @@ import aiohttp
 import os
 import sys
 
-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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openpipe import OpenPipeLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -67,18 +70,17 @@ async def main():
                 "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)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/07i-interruptible-xtts.py

@@ -9,15 +9,19 @@ import aiohttp
 import os
 import sys
 
-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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
+from pipecat.services.deepgram import DeepgramSTTService, DeepgramTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.services.xtts import XTTSService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -63,17 +67,17 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/07j-interruptible-gladia.py

@@ -9,16 +9,19 @@ import aiohttp
 import os
 import sys
 
-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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.gladia import GladiaSTTService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -66,18 +69,18 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 stt,  # STT
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/07k-interruptible-lmnt.py

@@ -9,15 +9,18 @@ import asyncio
 import os
 import sys
 
-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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.lmnt import LmntTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -59,17 +62,17 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                context_aggregator.user(),  # User respones
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/07l-interruptible-together.py

@@ -5,23 +5,28 @@
 #
 
 import asyncio
+import aiohttp
 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.services.ai_services import OpenAILLMContext
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.together import TogetherLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
 
 load_dotenv(override=True)
 
@@ -52,7 +57,7 @@ async def main():
 
         llm = TogetherLLMService(
             api_key=os.getenv("TOGETHER_API_KEY"),
-            model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
+            model=os.getenv("TOGETHER_MODEL"),
             params=TogetherLLMService.InputParams(
                 temperature=1.0,
                 top_p=0.9,
@@ -67,32 +72,25 @@ async def main():
         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 in plain language. Respond to what the user said in a creative and helpful way.",
+                "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)
-        user_aggregator = context_aggregator.user()
-        assistant_aggregator = context_aggregator.assistant()
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                user_aggregator,  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                assistant_aggregator,  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 
-        task = PipelineTask(
-            pipeline,
-            PipelineParams(
-                allow_interruptions=True, enable_metrics=True, enable_usage_metrics=True
-            ),
-        )
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):

examples/foundational/07m-interruptible-aws.py

@@ -13,16 +13,19 @@ 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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.aws import AWSTTSService
 from pipecat.services.deepgram import DeepgramSTTService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 load_dotenv(override=True)
 
@@ -66,18 +69,18 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 stt,  # STT
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/07n-interruptible-google.py

@@ -13,16 +13,19 @@ 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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.deepgram import DeepgramSTTService
 from pipecat.services.google import GoogleTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 load_dotenv(override=True)
 
@@ -50,6 +53,7 @@ async def main():
         stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
         tts = GoogleTTSService(
+            credentials=os.getenv("GOOGLE_CREDENTIALS"),
             voice_id="en-US-Neural2-J",
             params=GoogleTTSService.InputParams(language="en-US", rate="1.05"),
         )
@@ -63,18 +67,18 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 stt,  # STT
-                context_aggregator.user(),  # User respones
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/10-wake-phrase.py

@@ -9,15 +9,18 @@ import aiohttp
 import os
 import sys
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.processors.filters.wake_check_filter import WakeCheckFilter
 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.wake_check_filter import WakeCheckFilter
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -62,19 +65,18 @@ async def main():
         ]
 
         hey_robot_filter = WakeCheckFilter(["hey robot", "hey, robot"])
-
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
                 hey_robot_filter,  # Filter out speech not directed at the robot
-                context_aggregator.user(),  # User responses
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/11-sound-effects.py

@@ -10,7 +10,6 @@ import os
 import sys
 import wave
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import (
     Frame,
     LLMFullResponseEndFrame,
@@ -20,12 +19,16 @@ from pipecat.frames.frames import (
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_response import (
+    LLMUserResponseAggregator,
+    LLMAssistantResponseAggregator,
+)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.processors.logger import FrameLogger
 from pipecat.services.cartesia import CartesiaHttpTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -110,8 +113,8 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
         out_sound = OutboundSoundEffectWrapper()
         in_sound = InboundSoundEffectWrapper()
         fl = FrameLogger("LLM Out")
@@ -120,7 +123,7 @@ async def main():
         pipeline = Pipeline(
             [
                 transport.input(),
-                context_aggregator.user(),
+                tma_in,
                 in_sound,
                 fl2,
                 llm,
@@ -128,7 +131,7 @@ async def main():
                 tts,
                 out_sound,
                 transport.output(),
-                context_aggregator.assistant(),
+                tma_out,
             ]
         )
 

examples/foundational/12-describe-video.py

@@ -9,7 +9,6 @@ import aiohttp
 import os
 import sys
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import Frame, TextFrame, UserImageRequestFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -20,6 +19,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.moondream import MoondreamService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 

examples/foundational/12a-describe-video-gemini-flash.py

@@ -9,7 +9,6 @@ import aiohttp
 import os
 import sys
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import Frame, TextFrame, UserImageRequestFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -20,6 +19,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.google import GoogleLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 

examples/foundational/12b-describe-video-gpt-4o.py

@@ -9,7 +9,6 @@ import aiohttp
 import os
 import sys
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import Frame, TextFrame, UserImageRequestFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -20,6 +19,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 

examples/foundational/12c-describe-video-anthropic.py

@@ -9,7 +9,6 @@ import aiohttp
 import os
 import sys
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import Frame, TextFrame, UserImageRequestFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -20,6 +19,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.anthropic import AnthropicLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 

examples/foundational/13b-deepgram-transcription.py

@@ -14,7 +14,7 @@ from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.deepgram import DeepgramSTTService, LiveOptions, Language
+from pipecat.services.deepgram import DeepgramSTTService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
 from runner import configure
@@ -45,10 +45,7 @@ async def main():
             room_url, None, "Transcription bot", DailyParams(audio_in_enabled=True)
         )
 
-        stt = DeepgramSTTService(
-            api_key=os.getenv("DEEPGRAM_API_KEY"),
-            # live_options=LiveOptions(language=Language.FR),
-        )
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
         tl = TranscriptionLogger()
 

examples/foundational/13c-gladia-transcription.py

@@ -1,63 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import os
-import sys
-
-import aiohttp
-from dotenv import load_dotenv
-from loguru import logger
-from runner import configure
-
-from pipecat.frames.frames import Frame, TranscriptionFrame
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.gladia import GladiaSTTService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-class TranscriptionLogger(FrameProcessor):
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        await super().process_frame(frame, direction)
-
-        if isinstance(frame, TranscriptionFrame):
-            print(f"Transcription: {frame.text}")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, _) = await configure(session)
-
-        transport = DailyTransport(
-            room_url, None, "Transcription bot", DailyParams(audio_in_enabled=True)
-        )
-
-        stt = GladiaSTTService(
-            api_key=os.getenv("GLADIA_API_KEY"),
-            # live_options=LiveOptions(language=Language.FR),
-        )
-
-        tl = TranscriptionLogger()
-
-        pipeline = Pipeline([transport.input(), stt, tl])
-
-        task = PipelineTask(pipeline)
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/13d-assemblyai-transcription.py

@@ -1,62 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import os
-import sys
-
-import aiohttp
-from dotenv import load_dotenv
-from loguru import logger
-from runner import configure
-
-from pipecat.frames.frames import Frame, TranscriptionFrame
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.assemblyai import AssemblyAISTTService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-class TranscriptionLogger(FrameProcessor):
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        await super().process_frame(frame, direction)
-
-        if isinstance(frame, TranscriptionFrame):
-            print(f"Transcription: {frame.text}")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, _) = await configure(session)
-
-        transport = DailyTransport(
-            room_url, None, "Transcription bot", DailyParams(audio_in_enabled=True)
-        )
-
-        stt = AssemblyAISTTService(
-            api_key=os.getenv("ASSEMBLYAI_API_KEY"),
-        )
-
-        tl = TranscriptionLogger()
-
-        pipeline = Pipeline([transport.input(), stt, tl])
-
-        task = PipelineTask(pipeline)
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/14-function-calling.py

@@ -5,25 +5,24 @@
 #
 
 import asyncio
-import aiohttp
 import os
 import sys
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+from openai.types.chat import ChatCompletionToolParam
+from runner import configure
+
+from pipecat.frames.frames import TextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.task import PipelineTask
+from pipecat.processors.logger import FrameLogger
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMContext, OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-from openai.types.chat import ChatCompletionToolParam
-
-from runner import configure
-
-from loguru import logger
-
-from dotenv import load_dotenv
+from pipecat.vad.silero import SileroVADAnalyzer
 
 load_dotenv(override=True)
 
@@ -36,7 +35,7 @@ async def start_fetch_weather(function_name, llm, context):
     # can interrupt itself and/or cause audio overlapping glitches.
     # possible question for Aleix and Chad about what the right way
     # to trigger speech is, now, with the new queues/async/sync refactors.
-    # await llm.push_frame(TextFrame("Let me check on that."))
+    await llm.push_frame(TextFrame("Let me check on that.  "))
     logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")
 
 
@@ -70,6 +69,9 @@ async def main():
         # sent to the same callback with an additional function_name parameter.
         llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
 
+        fl_in = FrameLogger("Inner")
+        fl_out = FrameLogger("Outer")
+
         tools = [
             ChatCompletionToolParam(
                 type="function",
@@ -106,30 +108,24 @@ async def main():
 
         pipeline = Pipeline(
             [
+                # fl_in,
                 transport.input(),
                 context_aggregator.user(),
                 llm,
+                # fl_out,
                 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,
-            ),
-        )
+        task = PipelineTask(pipeline)
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
-            await task.queue_frames([context_aggregator.user().get_context_frame()])
+            await tts.say("Hi! Ask me about the weather in San Francisco.")
 
         runner = PipelineRunner()
 

examples/foundational/14c-function-calling-together.py

@@ -1,136 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import os
-import sys
-
-from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineTask
-from pipecat.services.cartesia import CartesiaTTSService
-from pipecat.services.openai import OpenAILLMContext
-from pipecat.services.together import TogetherLLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-from openai.types.chat import ChatCompletionToolParam
-
-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")
-
-
-async def start_fetch_weather(function_name, llm, context):
-    # note: we can't push a frame to the LLM here. the bot
-    # can interrupt itself and/or cause audio overlapping glitches.
-    # possible question for Aleix and Chad about what the right way
-    # to trigger speech is, now, with the new queues/async/sync refactors.
-    # await llm.push_frame(TextFrame("Let me check on that."))
-    logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")
-
-
-async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
-    await result_callback({"conditions": "nice", "temperature": "75"})
-
-
-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 = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
-
-        llm = TogetherLLMService(
-            api_key=os.getenv("TOGETHER_API_KEY"),
-            model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
-        )
-        # Register a function_name of None to get all functions
-        # sent to the same callback with an additional function_name parameter.
-        llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
-
-        tools = [
-            ChatCompletionToolParam(
-                type="function",
-                function={
-                    "name": "get_current_weather",
-                    "description": "Get the current weather",
-                    "parameters": {
-                        "type": "object",
-                        "properties": {
-                            "location": {
-                                "type": "string",
-                                "description": "The city and state, e.g. San Francisco, CA",
-                            },
-                            "format": {
-                                "type": "string",
-                                "enum": ["celsius", "fahrenheit"],
-                                "description": "The temperature unit to use. Infer this from the users location.",
-                            },
-                        },
-                        "required": ["location", "format"],
-                    },
-                },
-            )
-        ]
-        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, tools)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        pipeline = Pipeline(
-            [
-                transport.input(),
-                context_aggregator.user(),
-                llm,
-                tts,
-                transport.output(),
-                context_aggregator.assistant(),
-            ]
-        )
-
-        task = PipelineTask(pipeline)
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            # await tts.say("Hi! Ask me about the weather in San Francisco.")
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/14d-function-calling-video.py

@@ -1,167 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import os
-import sys
-
-from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineTask
-from pipecat.services.cartesia import CartesiaTTSService
-from pipecat.services.openai import OpenAILLMContext, OpenAILLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-from openai.types.chat import ChatCompletionToolParam
-
-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")
-
-video_participant_id = None
-
-
-async def get_weather(function_name, tool_call_id, arguments, llm, context, result_callback):
-    location = arguments["location"]
-    await result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
-
-
-async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
-    logger.debug(f"!!! IN get_image {video_participant_id}, {arguments}")
-    question = arguments["question"]
-    await llm.request_image_frame(user_id=video_participant_id, text_content=question)
-
-
-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 = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
-
-        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
-        llm.register_function("get_weather", get_weather)
-        llm.register_function("get_image", get_image)
-
-        tools = [
-            ChatCompletionToolParam(
-                type="function",
-                function={
-                    "name": "get_weather",
-                    "description": "Get the current weather",
-                    "parameters": {
-                        "type": "object",
-                        "properties": {
-                            "location": {
-                                "type": "string",
-                                "description": "The city and state, e.g. San Francisco, CA",
-                            },
-                            "format": {
-                                "type": "string",
-                                "enum": ["celsius", "fahrenheit"],
-                                "description": "The temperature unit to use. Infer this from the users location.",
-                            },
-                        },
-                        "required": ["location", "format"],
-                    },
-                },
-            ),
-            ChatCompletionToolParam(
-                type="function",
-                function={
-                    "name": "get_image",
-                    "description": "Get an image from the video stream.",
-                    "parameters": {
-                        "type": "object",
-                        "properties": {
-                            "question": {
-                                "type": "string",
-                                "description": "The question to ask the AI to generate an image of",
-                            },
-                        },
-                        "required": ["question"],
-                    },
-                },
-            ),
-        ]
-
-        system_prompt = """\
-You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions.
-
-Your response will be turned into speech so use only simple words and punctuation.
-
-You have access to two tools: get_weather and get_image.
-
-You can respond to questions about the weather using the get_weather tool.
-
-You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \
-indicate you should use the get_image tool are:
-  - What do you see?
-  - What's in the video?
-  - Can you describe the video?
-  - Tell me about what you see.
-  - Tell me something interesting about what you see.
-  - What's happening in the video?
-"""
-        messages = [
-            {"role": "system", "content": system_prompt},
-        ]
-
-        context = OpenAILLMContext(messages, tools)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        pipeline = Pipeline(
-            [
-                transport.input(),
-                context_aggregator.user(),
-                llm,
-                tts,
-                transport.output(),
-                context_aggregator.assistant(),
-            ]
-        )
-
-        task = PipelineTask(pipeline)
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            global video_participant_id
-            video_participant_id = participant["id"]
-            transport.capture_participant_transcription(participant["id"])
-            transport.capture_participant_video(video_participant_id, framerate=0)
-            # Kick off the conversation.
-            await tts.say("Hi! Ask me about the weather in San Francisco.")
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/15-switch-voices.py

@@ -9,7 +9,6 @@ import asyncio
 import os
 import sys
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import LLMMessagesFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
@@ -20,6 +19,7 @@ from pipecat.processors.filters.function_filter import FunctionFilter
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from openai.types.chat import ChatCompletionToolParam
 

examples/foundational/15a-switch-languages.py

@@ -9,7 +9,6 @@ import aiohttp
 import os
 import sys
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import LLMMessagesFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
@@ -21,6 +20,7 @@ from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.services.whisper import Model, WhisperSTTService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from openai.types.chat import ChatCompletionToolParam
 

examples/foundational/16-gpu-container-local-bot.py

@@ -5,20 +5,18 @@
 #
 
 import asyncio
+import aiohttp
 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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.deepgram import DeepgramTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import (
@@ -26,6 +24,13 @@ from pipecat.transports.services.daily import (
     DailyTransport,
     DailyTransportMessageFrame,
 )
+from pipecat.vad.silero import SileroVADAnalyzer
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
 
 load_dotenv(override=True)
 
@@ -72,17 +77,17 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                context_aggregator.user(),
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),
+                tma_out,  # Assistant spoken responses
             ]
         )
 
@@ -120,7 +125,7 @@ async def main():
                         )
                     )
                     # And push to the pipeline for the Daily transport.output to send
-                    await task.queue_frame(
+                    await tma_in.push_frame(
                         DailyTransportMessageFrame(
                             message={"latency-pong-pipeline-delivery": {"ts": ts}},
                             participant_id=sender,

examples/foundational/17-detect-user-idle.py

@@ -9,16 +9,19 @@ import aiohttp
 import os
 import sys
 
-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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.processors.user_idle_processor import UserIdleProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -62,8 +65,8 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         async def user_idle_callback(user_idle: UserIdleProcessor):
             messages.append(
@@ -80,11 +83,11 @@ async def main():
             [
                 transport.input(),  # Transport user input
                 user_idle,  # Idle user check-in
-                context_aggregator.user(),
+                tma_in,  # User responses
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),
+                tma_out,  # Assistant spoken responses
             ]
         )
 

examples/foundational/19-openai-realtime-beta.py

@@ -1,179 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import os
-import sys
-from datetime import datetime
-
-import aiohttp
-from dotenv import load_dotenv
-from loguru import logger
-from runner import configure
-
-from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.audio.vad.vad_analyzer import VADParams
-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_realtime_beta import (
-    InputAudioTranscription,
-    OpenAIRealtimeBetaLLMService,
-    SessionProperties,
-    TurnDetection,
-)
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
-    temperature = 75 if args["format"] == "fahrenheit" else 24
-    await result_callback(
-        {
-            "conditions": "nice",
-            "temperature": temperature,
-            "format": args["format"],
-            "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
-        }
-    )
-
-
-tools = [
-    {
-        "type": "function",
-        "name": "get_current_weather",
-        "description": "Get the current weather",
-        "parameters": {
-            "type": "object",
-            "properties": {
-                "location": {
-                    "type": "string",
-                    "description": "The city and state, e.g. San Francisco, CA",
-                },
-                "format": {
-                    "type": "string",
-                    "enum": ["celsius", "fahrenheit"],
-                    "description": "The temperature unit to use. Infer this from the users location.",
-                },
-            },
-            "required": ["location", "format"],
-        },
-    }
-]
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_in_enabled=True,
-                audio_in_sample_rate=24000,
-                audio_out_enabled=True,
-                audio_out_sample_rate=24000,
-                transcription_enabled=False,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)),
-                vad_audio_passthrough=True,
-            ),
-        )
-
-        session_properties = SessionProperties(
-            input_audio_transcription=InputAudioTranscription(),
-            # Set openai TurnDetection parameters. Not setting this at all will turn it
-            # on by default
-            turn_detection=TurnDetection(silence_duration_ms=1000),
-            # Or set to False to disable openai turn detection and use transport VAD
-            # turn_detection=False,
-            # tools=tools,
-            instructions="""Your knowledge cutoff is 2023-10. You are a helpful and friendly AI.
-
-Act like a human, but remember that you aren't a human and that you can't do human
-things in the real world. Your voice and personality should be warm and engaging, with a lively and
-playful tone.
-
-If interacting in a non-English language, start by using the standard accent or dialect familiar to
-the user. Talk quickly. You should always call a function if you can. Do not refer to these rules,
-even if you're asked about them.
--
-You are participating in a voice conversation. Keep your responses concise, short, and to the point
-unless specifically asked to elaborate on a topic.
-
-Remember, your responses should be short. Just one or two sentences, usually.""",
-        )
-
-        llm = OpenAIRealtimeBetaLLMService(
-            api_key=os.getenv("OPENAI_API_KEY"),
-            session_properties=session_properties,
-            start_audio_paused=False,
-        )
-
-        # you can either register a single function for all function calls, or specific functions
-        # llm.register_function(None, fetch_weather_from_api)
-        llm.register_function("get_current_weather", fetch_weather_from_api)
-
-        # Create a standard OpenAI LLM context object using the normal messages format. The
-        # OpenAIRealtimeBetaLLMService will convert this internally to messages that the
-        # openai WebSocket API can understand.
-        context = OpenAILLMContext(
-            [{"role": "user", "content": "Say hello!"}],
-            # [{"role": "user", "content": [{"type": "text", "text": "Say hello!"}]}],
-            #     [
-            #         {
-            #             "role": "user",
-            #             "content": [
-            #                 {"type": "text", "text": "Say"},
-            #                 {"type": "text", "text": "yo what's up!"},
-            #             ],
-            #         }
-            #     ],
-            tools,
-        )
-
-        context_aggregator = llm.create_context_aggregator(context)
-
-        pipeline = Pipeline(
-            [
-                transport.input(),  # Transport user input
-                context_aggregator.user(),
-                llm,  # LLM
-                context_aggregator.assistant(),
-                transport.output(),  # Transport bot output
-            ]
-        )
-
-        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):
-            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())

examples/foundational/19a-tools-anthropic.py

@@ -9,7 +9,6 @@ import aiohttp
 import os
 import sys
 
-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
@@ -17,6 +16,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.anthropic import AnthropicLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 

examples/foundational/19b-tools-video-anthropic.py

@@ -9,7 +9,6 @@ import aiohttp
 import os
 import sys
 
-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
@@ -17,6 +16,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.anthropic import AnthropicLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 

examples/foundational/19c-tools-togetherai.py

@@ -5,24 +5,26 @@
 #
 
 import asyncio
+import aiohttp
 import os
 import sys
+import json
 
-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.assemblyai import AssemblyAISTTService
 from pipecat.services.cartesia import CartesiaTTSService
-from pipecat.services.openai import OpenAILLMService
+from pipecat.services.together import TogetherLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
 
 load_dotenv(override=True)
 
@@ -30,6 +32,14 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
+async def get_current_weather(
+    function_name, tool_call_id, arguments, llm, context, result_callback
+):
+    logger.debug("IN get_current_weather")
+    location = arguments["location"]
+    await result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
+
+
 async def main():
     async with aiohttp.ClientSession() as session:
         (room_url, token) = await configure(session)
@@ -40,28 +50,60 @@ async def main():
             "Respond bot",
             DailyParams(
                 audio_out_enabled=True,
+                transcription_enabled=True,
                 vad_enabled=True,
                 vad_analyzer=SileroVADAnalyzer(),
-                vad_audio_passthrough=True,
             ),
         )
 
-        stt = AssemblyAISTTService(
-            api_key=os.getenv("ASSEMBLYAI_API_KEY"),
-        )
-
         tts = CartesiaTTSService(
             api_key=os.getenv("CARTESIA_API_KEY"),
             voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
         )
 
-        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+        llm = TogetherLLMService(
+            api_key=os.getenv("TOGETHER_API_KEY"),
+            model=os.getenv("TOGETHER_MODEL"),
+        )
+        llm.register_function("get_current_weather", get_current_weather)
+
+        weatherTool = {
+            "name": "get_current_weather",
+            "description": "Get the current weather in a given location",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "location": {
+                        "type": "string",
+                        "description": "The city and state, e.g. San Francisco, CA",
+                    },
+                },
+                "required": ["location"],
+            },
+        }
+
+        system_prompt = f"""\
+You have access to the following functions:
+
+Use the function '{weatherTool["name"]}' to '{weatherTool["description"]}':
+{json.dumps(weatherTool)}
+
+If you choose to call a function ONLY reply in the following format with no prefix or suffix:
+
+<function=example_function_name>{{\"example_name\": \"example_value\"}}</function>
+
+Reminder:
+- Function calls MUST follow the specified format, start with <function= and end with </function>
+- Required parameters MUST be specified
+- Only call one function at a time
+- Put the entire function call reply on one line
+- If there is no function call available, answer the question like normal with your current knowledge and do not tell the user about function calls
+
+"""
 
         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.",
-            },
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": "Wait for the user to say something."},
         ]
 
         context = OpenAILLMContext(messages)
@@ -70,22 +112,20 @@ async def main():
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
-                stt,  # STT
-                context_aggregator.user(),  # User responses
+                context_aggregator.user(),  # User speech to text
                 llm,  # LLM
                 tts,  # TTS
                 transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
+                context_aggregator.assistant(),  # Assistant spoken responses and tool context
             ]
         )
 
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True))
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             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()

examples/foundational/20a-persistent-context-openai.py

@@ -1,236 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import glob
-import json
-import os
-import sys
-from datetime import datetime
-
-import aiohttp
-from dotenv import load_dotenv
-from loguru import logger
-from runner import configure
-
-from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.audio.vad.vad_analyzer import VADParams
-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.cartesia import CartesiaTTSService
-
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-BASE_FILENAME = "/tmp/pipecat_conversation_"
-tts = None
-
-
-async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
-    temperature = 75 if args["format"] == "fahrenheit" else 24
-    await result_callback(
-        {
-            "conditions": "nice",
-            "temperature": temperature,
-            "format": args["format"],
-            "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
-        }
-    )
-
-
-async def get_saved_conversation_filenames(
-    function_name, tool_call_id, args, llm, context, result_callback
-):
-    # Construct the full pattern including the BASE_FILENAME
-    full_pattern = f"{BASE_FILENAME}*.json"
-
-    # Use glob to find all matching files
-    matching_files = glob.glob(full_pattern)
-    logger.debug(f"matching files: {matching_files}")
-
-    await result_callback({"filenames": matching_files})
-
-
-async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback):
-    timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
-    filename = f"{BASE_FILENAME}{timestamp}.json"
-    logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}")
-    try:
-        with open(filename, "w") as file:
-            messages = context.get_messages_for_persistent_storage()
-            # remove the last message, which is the instruction we just gave to save the conversation
-            messages.pop()
-            json.dump(messages, file, indent=2)
-        await result_callback({"success": True})
-    except Exception as e:
-        await result_callback({"success": False, "error": str(e)})
-
-
-async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback):
-    global tts
-    filename = args["filename"]
-    logger.debug(f"loading conversation from {filename}")
-    try:
-        with open(filename, "r") as file:
-            context.set_messages(json.load(file))
-            logger.debug(
-                f"loaded conversation from {filename}\n{json.dumps(context.messages, indent=4)}"
-            )
-        await tts.say("Ok, I've loaded that conversation.")
-    except Exception as e:
-        await result_callback({"success": False, "error": str(e)})
-
-
-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.",
-    },
-]
-tools = [
-    {
-        "type": "function",
-        "function": {
-            "name": "get_current_weather",
-            "description": "Get the current weather",
-            "parameters": {
-                "type": "object",
-                "properties": {
-                    "location": {
-                        "type": "string",
-                        "description": "The city and state, e.g. San Francisco, CA",
-                    },
-                    "format": {
-                        "type": "string",
-                        "enum": ["celsius", "fahrenheit"],
-                        "description": "The temperature unit to use. Infer this from the users location.",
-                    },
-                },
-                "required": ["location", "format"],
-            },
-        },
-    },
-    {
-        "type": "function",
-        "function": {
-            "name": "save_conversation",
-            "description": "Save the current conversatione. Use this function to persist the current conversation to external storage.",
-            "parameters": {
-                "type": "object",
-                "properties": {},
-                "required": [],
-            },
-        },
-    },
-    {
-        "type": "function",
-        "function": {
-            "name": "get_saved_conversation_filenames",
-            "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.",
-            "parameters": {
-                "type": "object",
-                "properties": {},
-                "required": [],
-            },
-        },
-    },
-    {
-        "type": "function",
-        "function": {
-            "name": "load_conversation",
-            "description": "Load a conversation history. Use this function to load a conversation history into the current session.",
-            "parameters": {
-                "type": "object",
-                "properties": {
-                    "filename": {
-                        "type": "string",
-                        "description": "The filename of the conversation history to load.",
-                    }
-                },
-                "required": ["filename"],
-            },
-        },
-    },
-]
-
-
-async def main():
-    global tts
-    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(params=VADParams(stop_secs=0.8)),
-            ),
-        )
-
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
-
-        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
-
-        # you can either register a single function for all function calls, or specific functions
-        # llm.register_function(None, fetch_weather_from_api)
-        llm.register_function("get_current_weather", fetch_weather_from_api)
-        llm.register_function("save_conversation", save_conversation)
-        llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames)
-        llm.register_function("load_conversation", load_conversation)
-
-        context = OpenAILLMContext(messages, tools)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        pipeline = Pipeline(
-            [
-                transport.input(),  # Transport user input
-                context_aggregator.user(),
-                llm,  # LLM
-                tts,
-                context_aggregator.assistant(),
-                transport.output(),  # Transport bot output
-            ]
-        )
-
-        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):
-            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())

examples/foundational/20b-persistent-context-openai-realtime.py

@@ -1,262 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import glob
-import json
-import os
-import sys
-from datetime import datetime
-
-import aiohttp
-from dotenv import load_dotenv
-from loguru import logger
-from runner import configure
-
-from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.audio.vad.vad_analyzer import VADParams
-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_realtime_beta import (
-    InputAudioTranscription,
-    OpenAIRealtimeBetaLLMService,
-    SessionProperties,
-    TurnDetection,
-)
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-BASE_FILENAME = "/tmp/pipecat_conversation_"
-
-
-async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
-    temperature = 75 if args["format"] == "fahrenheit" else 24
-    await result_callback(
-        {
-            "conditions": "nice",
-            "temperature": temperature,
-            "format": args["format"],
-            "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
-        }
-    )
-
-
-async def get_saved_conversation_filenames(
-    function_name, tool_call_id, args, llm, context, result_callback
-):
-    # Construct the full pattern including the BASE_FILENAME
-    full_pattern = f"{BASE_FILENAME}*.json"
-
-    # Use glob to find all matching files
-    matching_files = glob.glob(full_pattern)
-    logger.debug(f"matching files: {matching_files}")
-
-    await result_callback({"filenames": matching_files})
-
-
-# async def get_saved_conversation_filenames(
-#     function_name, tool_call_id, args, llm, context, result_callback
-# ):
-#     pattern = re.compile(re.escape(BASE_FILENAME) + "\\d{8}_\\d{6}\\.json$")
-#     matching_files = []
-
-#     for filename in os.listdir("."):
-#         if pattern.match(filename):
-#             matching_files.append(filename)
-
-#     await result_callback({"filenames": matching_files})
-
-
-async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback):
-    timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
-    filename = f"{BASE_FILENAME}{timestamp}.json"
-    logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}")
-    try:
-        with open(filename, "w") as file:
-            messages = context.get_messages_for_persistent_storage()
-            # remove the last message, which is the instruction we just gave to save the conversation
-            messages.pop()
-            json.dump(messages, file, indent=2)
-        await result_callback({"success": True})
-    except Exception as e:
-        await result_callback({"success": False, "error": str(e)})
-
-
-async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback):
-    async def _reset():
-        filename = args["filename"]
-        logger.debug(f"loading conversation from {filename}")
-        try:
-            with open(filename, "r") as file:
-                context.set_messages(json.load(file))
-                await llm.reset_conversation()
-                await llm._create_response()
-        except Exception as e:
-            await result_callback({"success": False, "error": str(e)})
-
-    asyncio.create_task(_reset())
-
-
-tools = [
-    {
-        "type": "function",
-        "name": "get_current_weather",
-        "description": "Get the current weather",
-        "parameters": {
-            "type": "object",
-            "properties": {
-                "location": {
-                    "type": "string",
-                    "description": "The city and state, e.g. San Francisco, CA",
-                },
-                "format": {
-                    "type": "string",
-                    "enum": ["celsius", "fahrenheit"],
-                    "description": "The temperature unit to use. Infer this from the users location.",
-                },
-            },
-            "required": ["location", "format"],
-        },
-    },
-    {
-        "type": "function",
-        "name": "save_conversation",
-        "description": "Save the current conversatione. Use this function to persist the current conversation to external storage.",
-        "parameters": {
-            "type": "object",
-            "properties": {},
-            "required": [],
-        },
-    },
-    {
-        "type": "function",
-        "name": "get_saved_conversation_filenames",
-        "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.",
-        "parameters": {
-            "type": "object",
-            "properties": {},
-            "required": [],
-        },
-    },
-    {
-        "type": "function",
-        "name": "load_conversation",
-        "description": "Load a conversation history. Use this function to load a conversation history into the current session.",
-        "parameters": {
-            "type": "object",
-            "properties": {
-                "filename": {
-                    "type": "string",
-                    "description": "The filename of the conversation history to load.",
-                }
-            },
-            "required": ["filename"],
-        },
-    },
-]
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_in_enabled=True,
-                audio_in_sample_rate=24000,
-                audio_out_enabled=True,
-                audio_out_sample_rate=24000,
-                transcription_enabled=False,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)),
-                vad_audio_passthrough=True,
-            ),
-        )
-
-        session_properties = SessionProperties(
-            input_audio_transcription=InputAudioTranscription(),
-            # Set openai TurnDetection parameters. Not setting this at all will turn it
-            # on by default
-            turn_detection=TurnDetection(silence_duration_ms=1000),
-            # Or set to False to disable openai turn detection and use transport VAD
-            # turn_detection=False,
-            # tools=tools,
-            instructions="""Your knowledge cutoff is 2023-10. You are a helpful and friendly AI.
-
-Act like a human, but remember that you aren't a human and that you can't do human
-things in the real world. Your voice and personality should be warm and engaging, with a lively and
-playful tone.
-
-If interacting in a non-English language, start by using the standard accent or dialect familiar to
-the user. Talk quickly. You should always call a function if you can. Do not refer to these rules,
-even if you're asked about them.
--
-You are participating in a voice conversation. Keep your responses concise, short, and to the point
-unless specifically asked to elaborate on a topic.
-
-Remember, your responses should be short. Just one or two sentences, usually.""",
-        )
-
-        llm = OpenAIRealtimeBetaLLMService(
-            api_key=os.getenv("OPENAI_API_KEY"),
-            session_properties=session_properties,
-            start_audio_paused=False,
-        )
-
-        # you can either register a single function for all function calls, or specific functions
-        # llm.register_function(None, fetch_weather_from_api)
-        llm.register_function("get_current_weather", fetch_weather_from_api)
-        llm.register_function("save_conversation", save_conversation)
-        llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames)
-        llm.register_function("load_conversation", load_conversation)
-
-        context = OpenAILLMContext([], tools)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        pipeline = Pipeline(
-            [
-                transport.input(),  # Transport user input
-                context_aggregator.user(),
-                llm,  # LLM
-                context_aggregator.assistant(),
-                transport.output(),  # Transport bot output
-            ]
-        )
-
-        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):
-            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())

examples/foundational/20c-persistent-context-anthropic.py

@@ -1,232 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import glob
-import json
-import os
-import sys
-from datetime import datetime
-
-import aiohttp
-from dotenv import load_dotenv
-from loguru import logger
-from runner import configure
-
-from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.audio.vad.vad_analyzer import VADParams
-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.anthropic import AnthropicLLMService
-
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-BASE_FILENAME = "/tmp/pipecat_conversation_"
-tts = None
-
-
-async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
-    temperature = 75 if args["format"] == "fahrenheit" else 24
-    await result_callback(
-        {
-            "conditions": "nice",
-            "temperature": temperature,
-            "format": args["format"],
-            "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
-        }
-    )
-
-
-async def get_saved_conversation_filenames(
-    function_name, tool_call_id, args, llm, context, result_callback
-):
-    # Construct the full pattern including the BASE_FILENAME
-    full_pattern = f"{BASE_FILENAME}*.json"
-
-    # Use glob to find all matching files
-    matching_files = glob.glob(full_pattern)
-    logger.debug(f"matching files: {matching_files}")
-
-    await result_callback({"filenames": matching_files})
-
-
-async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback):
-    timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
-    filename = f"{BASE_FILENAME}{timestamp}.json"
-    logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}")
-    try:
-        with open(filename, "w") as file:
-            # todo: extract 'system' into the first message in the list
-            messages = context.get_messages_for_persistent_storage()
-            # remove the last message, which is the instruction we just gave to save the conversation
-            messages.pop()
-            json.dump(messages, file, indent=2)
-        await result_callback({"success": True})
-    except Exception as e:
-        await result_callback({"success": False, "error": str(e)})
-
-
-async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback):
-    global tts
-    filename = args["filename"]
-    logger.debug(f"loading conversation from {filename}")
-    try:
-        with open(filename, "r") as file:
-            context.set_messages(json.load(file))
-            logger.debug(
-                f"loaded conversation from {filename}\n{json.dumps(context.messages, indent=4)}"
-            )
-        await tts.say("Ok, I've loaded that conversation.")
-    except Exception as e:
-        await result_callback({"success": False, "error": str(e)})
-
-
-# Test message munging ...
-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.",
-    },
-    {"role": "user", "content": ""},
-    {"role": "assistant", "content": []},
-    {"role": "user", "content": "Tell me"},
-    {"role": "user", "content": "a joke"},
-]
-tools = [
-    {
-        "name": "get_current_weather",
-        "description": "Get the current weather",
-        "input_schema": {
-            "type": "object",
-            "properties": {
-                "location": {
-                    "type": "string",
-                    "description": "The city and state, e.g. San Francisco, CA",
-                },
-                "format": {
-                    "type": "string",
-                    "enum": ["celsius", "fahrenheit"],
-                    "description": "The temperature unit to use. Infer this from the users location.",
-                },
-            },
-            "required": ["location", "format"],
-        },
-    },
-    {
-        "name": "save_conversation",
-        "description": "Save the current conversation. Use this function to persist the current conversation to external storage.",
-        "input_schema": {
-            "type": "object",
-            "properties": {},
-            "required": [],
-        },
-    },
-    {
-        "name": "get_saved_conversation_filenames",
-        "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.",
-        "input_schema": {
-            "type": "object",
-            "properties": {},
-            "required": [],
-        },
-    },
-    {
-        "name": "load_conversation",
-        "description": "Load a conversation history. Use this function to load a conversation history into the current session.",
-        "input_schema": {
-            "type": "object",
-            "properties": {
-                "filename": {
-                    "type": "string",
-                    "description": "The filename of the conversation history to load.",
-                }
-            },
-            "required": ["filename"],
-        },
-    },
-]
-
-
-async def main():
-    global tts
-    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(params=VADParams(stop_secs=0.8)),
-            ),
-        )
-
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
-
-        llm = AnthropicLLMService(
-            api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-sonnet-20240620"
-        )
-
-        # you can either register a single function for all function calls, or specific functions
-        # llm.register_function(None, fetch_weather_from_api)
-        llm.register_function("get_current_weather", fetch_weather_from_api)
-        llm.register_function("save_conversation", save_conversation)
-        llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames)
-        llm.register_function("load_conversation", load_conversation)
-
-        context = OpenAILLMContext(messages, tools)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        pipeline = Pipeline(
-            [
-                transport.input(),  # Transport user input
-                context_aggregator.user(),
-                llm,  # LLM
-                tts,
-                context_aggregator.assistant(),
-                transport.output(),  # Transport bot output
-            ]
-        )
-
-        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):
-            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())

examples/moondream-chatbot/README.md

@@ -24,7 +24,7 @@ cp env.example .env # and add your credentials
 python server.py
 ```
 
-Then, visit `http://localhost:7860/` in your browser to start a chatbot
+Then, visit `http://localhost:7860/start` in your browser to start a chatbot
 session.
 
 ## Build and test the Docker image
@@ -41,4 +41,4 @@ docker build -t moonbot -f Dockerfile.intel .
 docker run --env-file .env -p 7860:7860 --device /dev/dri moonbot
 ```
 
-You can try to visit `http://localhost:7860/` again.
+You can try to visit `http://localhost:7860/start` again.

examples/moondream-chatbot/bot.py

@@ -11,7 +11,6 @@ import sys
 
 from PIL import Image
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import (
     ImageRawFrame,
     OutputImageRawFrame,
@@ -24,11 +23,12 @@ from pipecat.frames.frames import (
     UserImageRawFrame,
     UserImageRequestFrame,
 )
+
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_response import LLMUserResponseAggregator
 from pipecat.processors.aggregators.sentence import SentenceAggregator
 from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -36,6 +36,7 @@ from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.moondream import MoondreamService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -182,19 +183,17 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        ura = LLMUserResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),
-                context_aggregator.user(),
+                ura,
                 llm,
                 ParallelPipeline([sa, ir, va, moondream], [tf, imgf]),
                 tts,
                 ta,
                 transport.output(),
-                context_aggregator.assistant(),
             ]
         )
 

examples/moondream-chatbot/env.example

@@ -1,4 +1,4 @@
 DAILY_SAMPLE_ROOM_URL=https://yourdomain.daily.co/yourroom # (for joining the bot to the same room repeatedly for local dev)
 DAILY_API_KEY=7df...
 OPENAI_API_KEY=sk-PL...
-CARTESIA_API_KEY=your_cartesia_api_key_here
+ELEVENLABS_API_KEY=aeb...

examples/moondream-chatbot/server.py

@@ -57,7 +57,7 @@ app.add_middleware(
 )
 
 
-@app.get("/")
+@app.get("/start")
 async def start_agent(request: Request):
     print(f"!!! Creating room")
     room = await daily_helpers["rest"].create_room(DailyRoomParams())

examples/patient-intake/README.md

@@ -1,39 +1,12 @@
-# Patient-intake chatbot
+# Simple Chatbot
 
 <img src="image.png" width="420px">
 
-This project implements an AI-powered chatbot designed to streamline the medical intake process for Tri-County Health Services. The chatbot, named Jessica, interacts with patients to collect essential information before their doctor's visit, enhancing efficiency and improving the patient experience.
+This app connects you to a chatbot powered by GPT-4, complete with animations generated by Stable Video Diffusion.
 
-## Features
+See a video of it in action: https://x.com/kwindla/status/1778628911817183509
 
-Identity Verification: Confirms patient identity by verifying their date of birth.
-Prescription Information: Collects details about current medications and dosages.
-Allergy Documentation: Records patient allergies.
-Medical Conditions: Gathers information about existing medical conditions.
-Reason for Visit: Asks patients about the purpose of their current doctor's visit.
-
-## Technical Stack
-
-Language: Python
-AI Model: OpenAI's GPT-4
-Text-to-Speech: Cartesia TTS Service
-Audio Processing: Silero VAD (Voice Activity Detection)
-Real-time Communication: Daily.co API
-
-## Key Components
-
-IntakeProcessor: Manages the conversation flow and information gathering process.
-DailyTransport: Handles real-time audio communication.
-CartesiaTTSService: Converts text responses to speech.
-OpenAILLMService: Processes natural language and generates appropriate responses.
-Pipeline: Orchestrates the flow of information between different components.
-
-How It Works
-
-The chatbot introduces itself and verifies the patient's identity.
-It systematically collects information about prescriptions, allergies, medical conditions, and the reason for the visit.
-The conversation is guided by a series of function calls that transition between different stages of the intake process.
-All collected information is logged for later use by medical professionals.
+And a quick video walkthrough of the code: https://www.loom.com/share/13df1967161f4d24ade054e7f8753416
 
 ℹ️ The first time, things might take extra time to get started since VAD (Voice Activity Detection) model needs to be downloaded.
 
@@ -54,7 +27,7 @@ cp env.example .env # and add your credentials
 python server.py
 ```
 
-Then, visit `http://localhost:7860/` in your browser to start a chatbot session.
+Then, visit `http://localhost:7860/start` in your browser to start a chatbot session.
 
 ## Build and test the Docker image
 

examples/patient-intake/bot.py

@@ -10,7 +10,6 @@ import os
 import sys
 import wave
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import OutputAudioRawFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -20,6 +19,7 @@ from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 

examples/patient-intake/env.example

@@ -1,4 +1,4 @@
 DAILY_SAMPLE_ROOM_URL=https://yourdomain.daily.co/yourroom # (for joining the bot to the same room repeatedly for local dev)
 DAILY_API_KEY=7df...
 OPENAI_API_KEY=sk-PL...
-CARTESIA_API_KEY=your_cartesia_api_key_here
+ELEVENLABS_API_KEY=aeb...

examples/patient-intake/server.py

@@ -57,7 +57,7 @@ app.add_middleware(
 )
 
 
-@app.get("/")
+@app.get("/start")
 async def start_agent(request: Request):
     print(f"!!! Creating room")
     room = await daily_helpers["rest"].create_room(DailyRoomParams())
@@ -122,13 +122,13 @@ if __name__ == "__main__":
     default_host = os.getenv("HOST", "0.0.0.0")
     default_port = int(os.getenv("FAST_API_PORT", "7860"))
 
-    parser = argparse.ArgumentParser(description="Daily patient-intake FastAPI server")
+    parser = argparse.ArgumentParser(description="Daily Storyteller FastAPI server")
     parser.add_argument("--host", type=str, default=default_host, help="Host address")
     parser.add_argument("--port", type=int, default=default_port, help="Port number")
     parser.add_argument("--reload", action="store_true", help="Reload code on change")
 
     config = parser.parse_args()
-    print(f"to join a test room, visit http://localhost:{config.port}/")
+    print(f"to join a test room, visit http://localhost:{config.port}/start")
     uvicorn.run(
         "server:app",
         host=config.host,

examples/simple-chatbot/README.md

@@ -27,7 +27,7 @@ cp env.example .env # and add your credentials
 python server.py
 ```
 
-Then, visit `http://localhost:7860/` in your browser to start a chatbot session.
+Then, visit `http://localhost:7860/start` in your browser to start a chatbot session.
 
 ## Build and test the Docker image
 

examples/simple-chatbot/bot.py

@@ -11,10 +11,13 @@ import sys
 
 from PIL import Image
 
-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.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.frames.frames import (
     OutputImageRawFrame,
     SpriteFrame,
@@ -23,11 +26,11 @@ from pipecat.frames.frames import (
     TTSAudioRawFrame,
     TTSStoppedFrame,
 )
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -140,20 +143,20 @@ async def main():
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        user_response = LLMUserResponseAggregator()
+        assistant_response = LLMAssistantResponseAggregator()
 
         ta = TalkingAnimation()
 
         pipeline = Pipeline(
             [
                 transport.input(),
-                context_aggregator.user(),
+                user_response,
                 llm,
                 tts,
                 ta,
                 transport.output(),
-                context_aggregator.assistant(),
+                assistant_response,
             ]
         )
 

examples/simple-chatbot/server.py

@@ -57,7 +57,7 @@ app.add_middleware(
 )
 
 
-@app.get("/")
+@app.get("/start")
 async def start_agent(request: Request):
     print(f"!!! Creating room")
     room = await daily_helpers["rest"].create_room(DailyRoomParams())

examples/storytelling-chatbot/frontend/components/App.tsx

@@ -27,7 +27,7 @@ export default function Call() {
 
     // Create a new room for the story session
     try {
-      const response = await fetch("/", {
+      const response = await fetch("/start_bot", {
         method: "POST",
         headers: {
           "Content-Type": "application/json",

examples/storytelling-chatbot/frontend/package.json

@@ -11,28 +11,28 @@
   "dependencies": {
     "@daily-co/daily-js": "^0.62.0",
     "@daily-co/daily-react": "^0.18.0",
-    "@radix-ui/react-select": "^2.1.2",
+    "@radix-ui/react-select": "^2.0.0",
     "@radix-ui/react-slot": "^1.0.2",
-    "@tabler/icons-react": "^3.19.0",
+    "@tabler/icons-react": "^3.1.0",
     "class-variance-authority": "^0.7.0",
-    "clsx": "^2.1.1",
-    "framer-motion": "^11.9.0",
-    "next": "^14.2.15",
-    "react": "^18.3.1",
-    "react-dom": "^18.3.1",
+    "clsx": "^2.1.0",
+    "framer-motion": "^11.0.27",
+    "next": "14.1.4",
+    "react": "^18",
+    "react-dom": "^18",
     "recoil": "^0.7.7",
-    "tailwind-merge": "^2.5.2",
+    "tailwind-merge": "^2.2.2",
     "tailwindcss-animate": "^1.0.7"
   },
   "devDependencies": {
-    "@types/node": "^20.16.10",
-    "@types/react": "^18.3.11",
-    "@types/react-dom": "^18.3.0",
-    "autoprefixer": "^10.4.20",
-    "eslint": "^8.57.1",
+    "@types/node": "^20",
+    "@types/react": "^18",
+    "@types/react-dom": "^18",
+    "autoprefixer": "^10.0.1",
+    "eslint": "^8",
     "eslint-config-next": "14.1.4",
-    "postcss": "^8.4.47",
-    "tailwindcss": "^3.4.13",
-    "typescript": "^5.6.2"
+    "postcss": "^8",
+    "tailwindcss": "^3.4.3",
+    "typescript": "^5"
   }
 }

examples/storytelling-chatbot/src/bot.py

@@ -1,20 +1,18 @@
 import argparse
 import asyncio
+import aiohttp
 import os
 import sys
 
-import aiohttp
-from dotenv import load_dotenv
-from loguru import logger
-from processors import StoryImageProcessor, StoryProcessor
-from prompts import CUE_USER_TURN, LLM_BASE_PROMPT, LLM_INTRO_PROMPT
-from utils.helpers import load_images, load_sounds
 
-from pipecat.frames.frames import EndFrame, LLMMessagesFrame, StopTaskFrame
+from pipecat.frames.frames import LLMMessagesFrame, StopTaskFrame, EndFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.fal import FalImageGenService
 from pipecat.services.openai import OpenAILLMService
@@ -24,6 +22,14 @@ from pipecat.transports.services.daily import (
     DailyTransportMessageFrame,
 )
 
+from processors import StoryProcessor, StoryImageProcessor
+from prompts import LLM_BASE_PROMPT, LLM_INTRO_PROMPT, CUE_USER_TURN
+from utils.helpers import load_sounds, load_images
+
+from loguru import logger
+
+from dotenv import load_dotenv
+
 load_dotenv(override=True)
 
 logger.remove(0)
@@ -79,8 +85,8 @@ async def main(room_url, token=None):
         story_pages = []
 
         # We need aggregators to keep track of user and LLM responses
-        context = OpenAILLMContext(message_history)
-        context_aggregator = llm_service.create_context_aggregator(context)
+        llm_responses = LLMAssistantResponseAggregator(message_history)
+        user_responses = LLMUserResponseAggregator(message_history)
 
         # -------------- Processors ------------- #
 
@@ -123,13 +129,13 @@ async def main(room_url, token=None):
         main_pipeline = Pipeline(
             [
                 transport.input(),
-                context_aggregator.user(),
+                user_responses,
                 llm_service,
                 story_processor,
                 image_processor,
                 tts_service,
                 transport.output(),
-                context_aggregator.assistant(),
+                llm_responses,
             ]
         )
 
@@ -137,7 +143,7 @@ async def main(room_url, token=None):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await intro_task.queue_frame(EndFrame())
+            intro_task.queue_frame(EndFrame())
             await main_task.queue_frame(EndFrame())
 
         @transport.event_handler("on_call_state_updated")

examples/storytelling-chatbot/src/bot_runner.py

@@ -69,7 +69,7 @@ STATIC_DIR = "frontend/out"
 app.mount("/static", StaticFiles(directory=STATIC_DIR, html=True), name="static")
 
 
-@app.post("/")
+@app.post("/start_bot")
 async def start_bot(request: Request) -> JSONResponse:
     if os.getenv("ENV", "dev") == "production":
         # Only allow requests from the specified domain

examples/studypal/studypal.py

@@ -8,15 +8,18 @@ from bs4 import BeautifulSoup
 from pypdf import PdfReader
 import tiktoken
 
-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.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from runner import configure
 
@@ -147,17 +150,17 @@ Your task is to help the user understand and learn from this article in 2 senten
             },
         ]
 
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
 
         pipeline = Pipeline(
             [
                 transport.input(),
-                context_aggregator.user(),
+                tma_in,
                 llm,
                 tts,
                 transport.output(),
-                context_aggregator.assistant(),
+                tma_out,
             ]
         )
 

examples/translation-chatbot/README.md

@@ -23,7 +23,7 @@ cp env.example .env # and add your credentials
 python server.py
 ```
 
-Then, visit `http://localhost:7860/` in your browser to start a translatorbot session.
+Then, visit `http://localhost:7860/start` in your browser to start a translatorbot session.
 
 ## Build and test the Docker image
 

examples/translation-chatbot/requirements.txt

@@ -1,4 +1,3 @@
 python-dotenv
 fastapi[all]
 pipecat-ai[daily,openai,azure]
-aiohttp

examples/translation-chatbot/server.py

@@ -57,7 +57,7 @@ app.add_middleware(
 )
 
 
-@app.get("/")
+@app.get("/start")
 async def start_agent(request: Request):
     print(f"!!! Creating room")
     room = await daily_helpers["rest"].create_room(DailyRoomParams())

examples/twilio-chatbot/README.md

@@ -53,7 +53,7 @@ This project is a FastAPI-based chatbot that integrates with Twilio to handle We
     ```
 
 2. **Update the Twilio Webhook**:
-    Copy the ngrok URL and update your Twilio phone number webhook URL to `http://<ngrok_url>/`.
+    Copy the ngrok URL and update your Twilio phone number webhook URL to `http://<ngrok_url>/start_call`.
 
 3. **Update streams.xml**:
     Copy the ngrok URL and update templates/streams.xml with `wss://<ngrok_url>/ws`.

examples/twilio-chatbot/bot.py

@@ -1,12 +1,14 @@
 import os
 import sys
 
-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.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.services.deepgram import DeepgramSTTService
@@ -14,6 +16,7 @@ from pipecat.transports.network.fastapi_websocket import (
     FastAPIWebsocketTransport,
     FastAPIWebsocketParams,
 )
+from pipecat.vad.silero import SileroVADAnalyzer
 from pipecat.serializers.twilio import TwilioFrameSerializer
 
 from loguru import logger
@@ -55,18 +58,18 @@ async def run_bot(websocket_client, stream_sid):
         },
     ]
 
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    tma_in = LLMUserResponseAggregator(messages)
+    tma_out = LLMAssistantResponseAggregator(messages)
 
     pipeline = Pipeline(
         [
             transport.input(),  # Websocket input from client
             stt,  # Speech-To-Text
-            context_aggregator.user(),
+            tma_in,  # User responses
             llm,  # LLM
             tts,  # Text-To-Speech
             transport.output(),  # Websocket output to client
-            context_aggregator.assistant(),
+            tma_out,  # LLM responses
         ]
     )
 

examples/twilio-chatbot/env.example

@@ -1,3 +1,4 @@
 OPENAI_API_KEY=
 DEEPGRAM_API_KEY=
-CARTESIA_API_KEY=
+ELEVENLABS_API_KEY=
+ELEVENLABS_VOICE_ID=

examples/twilio-chatbot/server.py

@@ -19,7 +19,7 @@ app.add_middleware(
 )
 
 
-@app.post("/")
+@app.post("/start_call")
 async def start_call():
     print("POST TwiML")
     return HTMLResponse(content=open("templates/streams.xml").read(), media_type="application/xml")

examples/websocket-server/Dockerfile

@@ -1,15 +0,0 @@
-FROM python:3.10-bullseye
-
-RUN mkdir /app
-
-COPY *.py /app/
-COPY requirements.txt /app/
-COPY .env /app/
-
-WORKDIR /app
-
-RUN pip3 install -r requirements.txt
-
-EXPOSE 7860
-
-CMD ["python3", "bot.py"]

examples/websocket-server/README.md

@@ -8,7 +8,6 @@ This is an example that shows how to use `WebsocketServerTransport` to communica
 python3 -m venv venv
 source venv/bin/activate
 pip install -r requirements.txt
-cp env.example .env # and add your credentials
 ```
 
 ## Run the bot

examples/websocket-server/bot.py

@@ -8,12 +8,14 @@ import asyncio
 import os
 import sys
 
-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 PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator,
+    LLMUserResponseAggregator,
+)
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.deepgram import DeepgramSTTService
 from pipecat.services.openai import OpenAILLMService
@@ -21,6 +23,7 @@ from pipecat.transports.network.websocket_server import (
     WebsocketServerParams,
     WebsocketServerTransport,
 )
+from pipecat.vad.silero import SileroVADAnalyzer
 
 from loguru import logger
 
@@ -59,18 +62,18 @@ async def main():
         },
     ]
 
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    tma_in = LLMUserResponseAggregator(messages)
+    tma_out = LLMAssistantResponseAggregator(messages)
 
     pipeline = Pipeline(
         [
             transport.input(),  # Websocket input from client
             stt,  # Speech-To-Text
-            context_aggregator.user(),
+            tma_in,  # User responses
             llm,  # LLM
             tts,  # Text-To-Speech
             transport.output(),  # Websocket output to client
-            context_aggregator.assistant(),
+            tma_out,  # LLM responses
         ]
     )
 

examples/websocket-server/env.example

@@ -1,8 +0,0 @@
-# OpenAI API Key
-OPENAI_API_KEY=your_openai_api_key_here
-
-# Deepgram API Key
-DEEPGRAM_API_KEY=your_deepgram_api_key_here
-
-# Cartesia API Key
-CARTESIA_API_KEY=your_cartesia_api_key_here

examples/websocket-server/requirements.txt

@@ -1,2 +1,2 @@
 python-dotenv
-pipecat-ai[cartesia,openai,silero,websocket,deepgram]
+pipecat-ai[cartesia,openai,silero,websocket,whisper]