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.

.github/workflows/format.yaml

@@ -38,4 +38,4 @@ jobs:
         id: ruff
         run: |
           source .venv/bin/activate
-          ruff format --diff
+          ruff format --config line-length=100 --diff --exclude "*_pb2.py"

.gitignore

@@ -4,7 +4,6 @@ __pycache__/
 *~
 venv
 .venv
-/.idea
 #*#
 
 # Distribution / packaging

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,238 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Added `GatedOpenAILLMContextAggregator`. This aggregator keeps the last
-  received OpenAI LLM context frame and it doesn't let it through until the
-  notifier is notified.
-
-- Added `WakeNotifierFilter`. This processor expects a list of frame types and
-  will execute a given callback predicate when a frame of any of those type is
-  being processed. If the callback returns true the notifier will be notified.
-
-- Added `NullFilter`. A null filter doesn't push any frames upstream or
-  downstream. This is usually used to disable one of the pipelines in
-  `ParallelPipeline`.
-
-- Added `EventNotifier`. This can be used as a very simple synchronization
-  feature between processors.
-
-- Added `TavusVideoService`. This is an integration for Tavus digital twins.
-  (see https://www.tavus.io/)
-
-- Added `DailyTransport.update_subscriptions()`. This allows you to have fine
-  grained control of what media subscriptions you want for each participant in a
-  room.
-
-### Changed
-
-- The following `DailyTransport` functions are now `async` which means they need
-  to be awaited: `start_dialout`, `stop_dialout`, `start_recording`,
-  `stop_recording`, `capture_participant_transcription` and
-  `capture_participant_video`.
-
-- Changed default output sample rate to 24000. This changes all TTS service to
-  output to 24000 and also the default output transport sample rate. This
-  improves audio quality at the cost of some extra bandwidth.
-
-### Fixed
-
-- Improved bot speaking detection for all TTS services by using actual bot
-  audio.
-
-- Fixed an issue that was generating constant bot started/stopped speaking
-  frames for HTTP TTS services.
-
-- Fixed an issue that was causing stuttering with AWS TTS service.
-
-- Fixed an issue with PlayHTTTSService, where the TTFB metrics were reporting
-  very small time values.
-
-### Other
-
-- Added a new foundational example 22-natural-conversation.py. This examples
-  shows how to achieve a more natural conversation detecting when the user ends
-  statement.
-
-## [0.0.47] - 2024-10-22
-
-### 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`
-
-### Changed
-
-- Updated `GladiaSTTService` to use the V2 API.
-
-- Changed `DailyTransport` transcription model to `nova-2-general`.
-
-### Fixed
-
-- Fixed an issue that would cause an import error when importing
-  `SileroVADAnalyzer` from the old package `pipecat.vad.silero`.
-
-- 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
-
-- Changed `DeepgramSTTService` model to `nova-2-general`.
-
-- 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.
 
 ```
@@ -312,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`
@@ -337,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`.
 
@@ -363,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.
@@ -381,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="", # leave empty. Note: token is _not_ your api key
-    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:_
@@ -178,7 +179,7 @@ You can use [use-package](https://github.com/jwiegley/use-package) to install [e
   :ensure t
   :hook ((python-mode . lazy-ruff-mode))
   :config
-  (setq lazy-ruff-format-command "ruff format")
+  (setq lazy-ruff-format-command "ruff format --config line-length=100")
   (setq lazy-ruff-only-format-block t)
   (setq lazy-ruff-only-format-region t)
   (setq lazy-ruff-only-format-buffer t))
@@ -197,13 +198,14 @@ You can use [use-package](https://github.com/jwiegley/use-package) to install [e
 ### Visual Studio Code
 
 Install the
-[Ruff](https://marketplace.visualstudio.com/items?itemName=charliermarsh.ruff) extension. Then edit the user settings (_Ctrl-Shift-P_ `Open User Settings (JSON)`) and set it as the default Python formatter, and enable formatting on save:
+[Ruff](https://marketplace.visualstudio.com/items?itemName=charliermarsh.ruff) extension. Then edit the user settings (_Ctrl-Shift-P_ `Open User Settings (JSON)`) and set it as the default Python formatter, enable formatting on save and configure `ruff` arguments:
 
 ```json
 "[python]": {
     "editor.defaultFormatter": "charliermarsh.ruff",
     "editor.formatOnSave": true
-}
+},
+"ruff.format.args": ["--config", "line-length=100"]
 ```
 
 ## Getting help

docs/CONTRIBUTING.md

@@ -1,165 +0,0 @@
-## Contributing to Pipecat
-
-We welcome contributions of all kinds! Your help is appreciated. Follow these steps to get involved:
-
-1. **Fork this repository**: Start by forking the Pipecat Documentation repository to your GitHub account.
-
-2. **Clone the repository**: Clone your forked repository to your local machine.
-   ```bash
-   git clone https://github.com/your-username/pipecat
-   ```
-3. **Create a branch**: For your contribution, create a new branch.
-   ```bash
-   git checkout -b your-branch-name
-   ```
-4. **Make your changes**: Edit or add files as necessary.
-5. **Test your changes**: Ensure that your changes look correct and follow the style set in the codebase.
-6. **Commit your changes**: Once you're satisfied with your changes, commit them with a meaningful message.
-
-```bash
-git commit -m "Description of your changes"
-```
-
-7. **Push your changes**: Push your branch to your forked repository.
-
-```bash
-git push origin your-branch-name
-```
-
-9. **Submit a Pull Request (PR)**: Open a PR from your forked repository to the main branch of this repo. 
-> Important: Describe the changes you've made clearly!
-
-Our maintainers will review your PR, and once everything is good, your contributions will be merged!
-
-
-# Contributor Covenant Code of Conduct
-
-## Our Pledge
-
-We as members, contributors, and leaders pledge to make participation in our
-community a harassment-free experience for everyone, regardless of age, body
-size, visible or invisible disability, ethnicity, sex characteristics, gender
-identity and expression, level of experience, education, socio-economic status,
-nationality, personal appearance, race, caste, color, religion, or sexual
-identity and orientation.
-
-We pledge to act and interact in ways that contribute to an open, welcoming,
-diverse, inclusive, and healthy community.
-
-## Our Standards
-
-Examples of behavior that contributes to a positive environment for our
-community include:
-
-* Demonstrating empathy and kindness toward other people
-* Being respectful of differing opinions, viewpoints, and experiences
-* Giving and gracefully accepting constructive feedback
-* Accepting responsibility and apologizing to those affected by our mistakes,
-  and learning from the experience
-* Focusing on what is best not just for us as individuals, but for the overall
-  community
-
-Examples of unacceptable behavior include:
-
-* The use of sexualized language or imagery, and sexual attention or advances of
-  any kind
-* Trolling, insulting or derogatory comments, and personal or political attacks
-* Public or private harassment
-* Publishing others' private information, such as a physical or email address,
-  without their explicit permission
-* Other conduct which could reasonably be considered inappropriate in a
-  professional setting
-
-## Enforcement Responsibilities
-
-Community leaders are responsible for clarifying and enforcing our standards of
-acceptable behavior and will take appropriate and fair corrective action in
-response to any behavior that they deem inappropriate, threatening, offensive,
-or harmful.
-
-Community leaders have the right and responsibility to remove, edit, or reject
-comments, commits, code, wiki edits, issues, and other contributions that are
-not aligned to this Code of Conduct, and will communicate reasons for moderation
-decisions when appropriate.
-
-## Scope
-
-This Code of Conduct applies within all community spaces, and also applies when
-an individual is officially representing the community in public spaces.
-Examples of representing our community include using an official email address,
-posting via an official social media account, or acting as an appointed
-representative at an online or offline event.
-
-## Enforcement
-
-Instances of abusive, harassing, or otherwise unacceptable behavior may be
-reported to the community leaders responsible for enforcement at pipecat-ai@daily.co.
-All complaints will be reviewed and investigated promptly and fairly.
-
-All community leaders are obligated to respect the privacy and security of the
-reporter of any incident.
-
-## Enforcement Guidelines
-
-Community leaders will follow these Community Impact Guidelines in determining
-the consequences for any action they deem in violation of this Code of Conduct:
-
-### 1. Correction
-
-**Community Impact**: Use of inappropriate language or other behavior deemed
-unprofessional or unwelcome in the community.
-
-**Consequence**: A private, written warning from community leaders, providing
-clarity around the nature of the violation and an explanation of why the
-behavior was inappropriate. A public apology may be requested.
-
-### 2. Warning
-
-**Community Impact**: A violation through a single incident or series of
-actions.
-
-**Consequence**: A warning with consequences for continued behavior. No
-interaction with the people involved, including unsolicited interaction with
-those enforcing the Code of Conduct, for a specified period of time. This
-includes avoiding interactions in community spaces as well as external channels
-like social media. Violating these terms may lead to a temporary or permanent
-ban.
-
-### 3. Temporary Ban
-
-**Community Impact**: A serious violation of community standards, including
-sustained inappropriate behavior.
-
-**Consequence**: A temporary ban from any sort of interaction or public
-communication with the community for a specified period of time. No public or
-private interaction with the people involved, including unsolicited interaction
-with those enforcing the Code of Conduct, is allowed during this period.
-Violating these terms may lead to a permanent ban.
-
-### 4. Permanent Ban
-
-**Community Impact**: Demonstrating a pattern of violation of community
-standards, including sustained inappropriate behavior, harassment of an
-individual, or aggression toward or disparagement of classes of individuals.
-
-**Consequence**: A permanent ban from any sort of public interaction within the
-community.
-
-## Attribution
-
-This Code of Conduct is adapted from the [Contributor Covenant][homepage],
-version 2.1, available at
-[https://www.contributor-covenant.org/version/2/1/code_of_conduct.html][v2.1].
-
-Community Impact Guidelines were inspired by
-[Mozilla's code of conduct enforcement ladder][Mozilla CoC].
-
-For answers to common questions about this code of conduct, see the FAQ at
-[https://www.contributor-covenant.org/faq][FAQ]. Translations are available at
-[https://www.contributor-covenant.org/translations][translations].
-
-[homepage]: https://www.contributor-covenant.org
-[v2.1]: https://www.contributor-covenant.org/version/2/1/code_of_conduct.html
-[Mozilla CoC]: https://github.com/mozilla/diversity
-[FAQ]: https://www.contributor-covenant.org/faq
-[translations]: https://www.contributor-covenant.org/translations

docs/ISSUE_TEMPLATE.md

@@ -1,22 +0,0 @@
-# Description
-Is this reporting a bug or feature request?
-
-
-If reporting a bug, please fill out the following:
-
-### Environment
-- pipecat-ai version:
-- python version:
-- OS:
-
-### Issue description
-Provide a clear description of the issue.
-
-### Repro steps
-List the steps to reproduce the issue.
-
-### Expected behavior
-
-### Actual behavior
-
-### Logs

docs/PULL_REQUEST_TEMPLATE.md

@@ -1 +0,0 @@
-#### Please describe the changes in your PR. If it is addressing an issue, please reference that as well.

dot-env.template

@@ -46,10 +46,5 @@ PLAY_HT_API_KEY=...
 # OpenAI
 OPENAI_API_KEY=...
 
-# OpenPipe
+#OpenPipe
 OPENPIPE_API_KEY=...
-
-# Tavus
-TAVUS_API_KEY=...
-TAVUS_REPLICA_ID=...
-TAVUS_PERSONA_ID=...

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,66 +0,0 @@
-# Chatbot with canonical-metrics
-
-This project implements a chatbot using a pipeline architecture that integrates audio processing, transcription, and a language model for conversational interactions. The chatbot operates within a daily communication environment, utilizing various services for text-to-speech and language model responses.
-
-## Features
-
-- **Audio Input and Output**: Captures microphone input and plays back audio responses.
-- **Voice Activity Detection**: Utilizes Silero VAD to manage audio input intelligently.
-- **Text-to-Speech**: Integrates ElevenLabs TTS service to convert text responses into audio.
-- **Language Model Interaction**: Uses OpenAI's GPT-4 model to generate responses based on user input.
-- **Transcription Services**: Captures and transcribes participant speech for analytics.
-- **Metrics Collection**: Sends audio data for analysis via Canonical Metrics Service.
-
-## Requirements
-
-- Python 3.10+
-- `python-dotenv`
-- Additional libraries from the `pipecat` package.
-
-## Setup
-
-1. Clone the repository.
-2. Install the required packages.
-3. Set up environment variables for API keys:
-   - `OPENAI_API_KEY`
-   - `ELEVENLABS_API_KEY`
-   - `CANONICAL_API_KEY`
-   - `CANONICAL_API_URL`
-4. Run the script.
-
-## Usage
-
-The chatbot introduces itself and engages in conversations, providing brief and creative responses. Designed for flexibility, it can support multiple languages with appropriate configuration.
-
-## Events
-
-- Participants joining or leaving the call are handled dynamically, adjusting the chatbot's behavior accordingly.
-
-
-ℹ️ 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):
-            await 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):
-            await 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,
         ]
     )
 
@@ -75,7 +78,7 @@ async def main(room_url: str, token: str):
 
     @transport.event_handler("on_first_participant_joined")
     async def on_first_participant_joined(transport, participant):
-        await transport.capture_participant_transcription(participant["id"])
+        transport.capture_participant_transcription(participant["id"])
         await task.queue_frames([LLMMessagesFrame(messages)])
 
     @transport.event_handler("on_participant_left")

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,
         ]
     )
 
@@ -81,7 +83,7 @@ async def main(room_url: str, token: str, callId: str, callDomain: str):
 
     @transport.event_handler("on_first_participant_joined")
     async def on_first_participant_joined(transport, participant):
-        await transport.capture_participant_transcription(participant["id"])
+        transport.capture_participant_transcription(participant["id"])
         await task.queue_frames([LLMMessagesFrame(messages)])
 
     @transport.event_handler("on_participant_left")

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,
         ]
     )
 
@@ -84,7 +87,7 @@ async def main(room_url: str, token: str, callId: str, sipUri: str):
 
     @transport.event_handler("on_first_participant_joined")
     async def on_first_participant_joined(transport, participant):
-        await transport.capture_participant_transcription(participant["id"])
+        transport.capture_participant_transcription(participant["id"])
         await task.queue_frames([LLMMessagesFrame(messages)])
 
     @transport.event_handler("on_participant_left")

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)
@@ -81,7 +78,7 @@ async def main():
             url=url,
             token=token,
             room_name=room_name,
-            params=LiveKitParams(audio_out_enabled=True),
+            params=LiveKitParams(audio_out_enabled=True, audio_out_sample_rate=16000),
         )
 
         tts = CartesiaTTSService(

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

@@ -5,30 +5,35 @@
 #
 
 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 Frame, LLMMessagesFrame, MetricsFrame
 from pipecat.metrics.metrics import (
-    LLMUsageMetricsData,
-    ProcessingMetricsData,
     TTFBMetricsData,
+    ProcessingMetricsData,
+    LLMUsageMetricsData,
     TTSUsageMetricsData,
 )
 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.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
+
+from loguru import logger
+
+from dotenv import load_dotenv
 
 load_dotenv(override=True)
 
@@ -87,30 +92,26 @@ 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,
             ]
         )
 
-        task = PipelineTask(
-            pipeline,
-            PipelineParams(enable_metrics=True, enable_usage_metrics=True),
-        )
+        task = PipelineTask(pipeline)
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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/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,8 +129,8 @@ 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", "")
-            await transport.capture_participant_transcription(participant["id"])
+            participant_name = participant["info"]["userName"] or ""
+            transport.capture_participant_transcription(participant["id"])
             await task.queue_frames([TextFrame(f"Hi there {participant_name}!")])
 
         runner = PipelineRunner()

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):
-            await transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

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
             ]
         )
 
@@ -87,7 +90,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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/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
             ]
         )
 
@@ -82,7 +87,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             await task.queue_frames([LLMMessagesFrame(messages)])
 

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
@@ -109,7 +109,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_transcription(participant["id"])
             lc.set_participant_id(participant["id"])
             # Kick off the conversation.
             # the `LLMMessagesFrame` will be picked up by the LangchainProcessor using

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
             ]
         )
 
@@ -85,7 +88,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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/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)
 
@@ -40,6 +44,7 @@ async def main():
             "Respond bot",
             DailyParams(
                 audio_out_enabled=True,
+                audio_out_sample_rate=16000,
                 transcription_enabled=True,
                 vad_enabled=True,
                 vad_analyzer=SileroVADAnalyzer(),
@@ -50,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")
@@ -62,33 +66,25 @@ 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):
-            await transport.capture_participant_transcription(participant["id"])
+            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/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
@@ -41,6 +44,7 @@ async def main():
             "Respond bot",
             DailyParams(
                 audio_out_enabled=True,
+                audio_out_sample_rate=16000,
                 vad_enabled=True,
                 vad_analyzer=SileroVADAnalyzer(),
                 vad_audio_passthrough=True,
@@ -70,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
             ]
         )
 
@@ -89,7 +93,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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/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
             ]
         )
 
@@ -74,7 +80,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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/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
             ]
         )
 
@@ -86,7 +88,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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/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
             ]
         )
 
@@ -81,7 +85,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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/07j-interruptible-gladia.py

@@ -5,24 +5,29 @@
 #
 
 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 EndFrame, LLMMessagesFrame
+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
+
+from loguru import logger
+
+from dotenv import load_dotenv
 
 load_dotenv(override=True)
 
@@ -64,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
             ]
         )
 
@@ -83,16 +88,11 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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)])
 
-        # 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())
-
         runner = PipelineRunner()
 
         await runner.run(task)

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
             ]
         )
 
@@ -77,7 +80,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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/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,36 +72,29 @@ 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):
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             await task.queue_frames([LLMMessagesFrame(messages)])
 

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)
 
@@ -40,6 +43,7 @@ async def main():
             "Respond bot",
             DailyParams(
                 audio_out_enabled=True,
+                audio_out_sample_rate=16000,
                 vad_enabled=True,
                 vad_analyzer=SileroVADAnalyzer(),
                 vad_audio_passthrough=True,
@@ -65,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
             ]
         )
 
@@ -84,7 +88,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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/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
             ]
         )
 
@@ -82,7 +86,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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/07o-interruptible-assemblyai.py

@@ -1,97 +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.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.transports.services.daily import DailyParams, DailyTransport
-
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                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")
-
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
-            },
-        ]
-
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        pipeline = Pipeline(
-            [
-                transport.input(),  # Transport user input
-                stt,  # STT
-                context_aggregator.user(),  # User responses
-                llm,  # LLM
-                tts,  # TTS
-                transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
-            ]
-        )
-
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/09-mirror.py

@@ -63,7 +63,6 @@ async def main():
             "Test",
             DailyParams(
                 audio_in_enabled=True,
-                audio_in_sample_rate=24000,
                 audio_out_enabled=True,
                 camera_out_enabled=True,
                 camera_out_is_live=True,
@@ -74,7 +73,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_video(participant["id"])
+            transport.capture_participant_video(participant["id"])
 
         pipeline = Pipeline([transport.input(), MirrorProcessor(), transport.output()])
 

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

@@ -65,7 +65,7 @@ async def main():
         tk_root.title("Local Mirror")
 
         daily_transport = DailyTransport(
-            room_url, token, "Test", DailyParams(audio_in_enabled=True, audio_in_sample_rate=24000)
+            room_url, token, "Test", DailyParams(audio_in_enabled=True)
         )
 
         tk_transport = TkLocalTransport(
@@ -81,7 +81,7 @@ async def main():
 
         @daily_transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_video(participant["id"])
+            transport.capture_participant_video(participant["id"])
 
         pipeline = Pipeline([daily_transport.input(), MirrorProcessor(), tk_transport.output()])
 

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
             ]
         )
 
@@ -82,7 +84,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_transcription(participant["id"])
             await tts.say("Hi! If you want to talk to me, just say 'Hey Robot'.")
 
         runner = PipelineRunner()

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,13 +131,13 @@ async def main():
                 tts,
                 out_sound,
                 transport.output(),
-                context_aggregator.assistant(),
+                tma_out,
             ]
         )
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_transcription(participant["id"])
             await tts.say("Hi, I'm listening!")
             await transport.send_audio(sounds["ding1.wav"])
 

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
 
@@ -84,8 +84,8 @@ async def main():
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             await tts.say("Hi there! Feel free to ask me what I see.")
-            await transport.capture_participant_video(participant["id"], framerate=0)
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_video(participant["id"], framerate=0)
+            transport.capture_participant_transcription(participant["id"])
             image_requester.set_participant_id(participant["id"])
 
         pipeline = Pipeline(

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
 
@@ -86,8 +86,8 @@ async def main():
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             await tts.say("Hi there! Feel free to ask me what I see.")
-            await transport.capture_participant_video(participant["id"], framerate=0)
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_video(participant["id"], framerate=0)
+            transport.capture_participant_transcription(participant["id"])
             image_requester.set_participant_id(participant["id"])
 
         pipeline = Pipeline(

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
 
@@ -83,8 +83,8 @@ async def main():
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             await tts.say("Hi there! Feel free to ask me what I see.")
-            await transport.capture_participant_video(participant["id"], framerate=0)
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_video(participant["id"], framerate=0)
+            transport.capture_participant_transcription(participant["id"])
             image_requester.set_participant_id(participant["id"])
 
         pipeline = Pipeline(

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
 
@@ -78,13 +78,16 @@ async def main():
         tts = CartesiaTTSService(
             api_key=os.getenv("CARTESIA_API_KEY"),
             voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+            params=CartesiaTTSService.InputParams(
+                sample_rate=16000,
+            ),
         )
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             await tts.say("Hi there! Feel free to ask me what I see.")
-            await transport.capture_participant_video(participant["id"], framerate=0)
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_video(participant["id"], framerate=0)
+            transport.capture_participant_transcription(participant["id"])
             image_requester.set_participant_id(participant["id"])
 
         pipeline = Pipeline(

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):
-            await transport.capture_participant_transcription(participant["id"])
+            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):
-            await 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"]
-            await transport.capture_participant_transcription(participant["id"])
-            await 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/14e-function-calling-gemini.py

@@ -1,173 +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 PipelineParams, PipelineTask
-from pipecat.services.cartesia import CartesiaTTSService
-from pipecat.services.google import GoogleLLMService
-from pipecat.services.openai import OpenAILLMContext
-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")
-
-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 = GoogleLLMService(model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY"))
-        llm.register_function("get_weather", get_weather)
-        llm.register_function("get_image", get_image)
-
-        tools = [
-            {
-                "function_declarations": [
-                    {
-                        "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"],
-                        },
-                    },
-                    {
-                        "name": "get_image",
-                        "description": "Get and image from the camera or video stream.",
-                        "parameters": {
-                            "type": "object",
-                            "properties": {
-                                "question": {
-                                    "type": "string",
-                                    "description": "The question to to use when running inference on the acquired image.",
-                                },
-                            },
-                            "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},
-            {"role": "user", "content": "Say hello."},
-        ]
-
-        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,
-            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):
-            global video_participant_id
-            video_participant_id = participant["id"]
-            await transport.capture_participant_transcription(participant["id"])
-            await transport.capture_participant_video(video_participant_id, framerate=0)
-            # 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/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
 
@@ -141,7 +141,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append(
                 {

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

@@ -9,8 +9,7 @@ import aiohttp
 import os
 import sys
 
-from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import LLMMessagesFrame, TTSUpdateSettingsFrame
+from pipecat.frames.frames import LLMMessagesFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -19,7 +18,9 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.filters.function_filter import FunctionFilter
 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
 
@@ -60,14 +61,16 @@ async def main():
             token,
             "Pipecat",
             DailyParams(
+                audio_in_enabled=True,
                 audio_out_enabled=True,
-                transcription_enabled=True,
                 vad_enabled=True,
                 vad_analyzer=SileroVADAnalyzer(),
                 vad_audio_passthrough=True,
             ),
         )
 
+        stt = WhisperSTTService(model=Model.LARGE)
+
         english_tts = CartesiaTTSService(
             api_key=os.getenv("CARTESIA_API_KEY"),
             voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
@@ -113,6 +116,7 @@ async def main():
         pipeline = Pipeline(
             [
                 transport.input(),  # Transport user input
+                stt,  # STT
                 context_aggregator.user(),  # User responses
                 llm,  # LLM
                 ParallelPipeline(  # TTS (bot will speak the chosen language)
@@ -128,7 +132,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             messages.append(
                 {

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
             ]
         )
 
@@ -92,7 +97,7 @@ async def main():
         # bot can "hear" and respond to them.
         @transport.event_handler("on_participant_joined")
         async def on_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_transcription(participant["id"])
 
         # When the first participant joins, the bot should introduce itself.
         @transport.event_handler("on_first_participant_joined")
@@ -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
             ]
         )
 
@@ -99,7 +102,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            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/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):
-            await transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            await task.queue_frames([context_aggregator.user().get_context_frame()])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

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
 
@@ -105,7 +105,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

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
 
@@ -160,8 +160,8 @@ If you need to use a tool, simply use the tool. Do not tell the user the tool yo
         async def on_first_participant_joined(transport, participant):
             global video_participant_id
             video_participant_id = participant["id"]
-            await transport.capture_participant_transcription(video_participant_id)
-            await transport.capture_participant_video(video_participant_id, framerate=0)
+            transport.capture_participant_transcription(video_participant_id)
+            transport.capture_participant_video(video_participant_id, framerate=0)
             # Kick off the conversation.
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

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

@@ -0,0 +1,137 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import aiohttp
+import os
+import sys
+import json
+
+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.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)
+
+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)
+
+        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=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": system_prompt},
+            {"role": "user", "content": "Wait for the user to say something."},
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                context_aggregator.user(),  # User speech to text
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses and tool context
+            ]
+        )
+
+        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.
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

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):
-            await transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            await task.queue_frames([context_aggregator.user().get_context_frame()])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

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):
-            await transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            await task.queue_frames([context_aggregator.user().get_context_frame()])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

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):
-            await transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            await task.queue_frames([context_aggregator.user().get_context_frame()])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/20d-persistent-context-gemini.py

@@ -1,290 +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.google import GoogleLLMService
-
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-video_participant_id = None
-
-
-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_image(function_name, tool_call_id, arguments, llm, context, result_callback):
-    question = arguments["question"]
-    await llm.request_image_frame(user_id=video_participant_id, text_content=question)
-
-
-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.get_messages_for_logging(), 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 (the instruction to save the context)
-            messages.pop()
-            json.dump(messages, file, indent=2)
-        await result_callback({"success": True})
-    except Exception as e:
-        logger.debug(f"error saving conversation: {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))
-        await result_callback(
-            {
-                "success": True,
-                "message": "The most recent conversation has been loaded. Awaiting further instructions.",
-            }
-        )
-    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.
-
-You have several tools you can use to help you.
-
-You can respond to questions about the weather using the get_weather tool.
-
-You can save the current conversation using the save_conversation tool. This tool allows you to save
-the current conversation to external storage. If the user asks you to save the conversation, use this
-save_conversation too.
-
-You can load a saved conversation using the load_conversation tool. This tool allows you to load a
-conversation from external storage. You can get a list of conversations that have been saved using the
-get_saved_conversation_filenames 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?
-        """,
-    },
-    # {"role": "user", "content": ""},
-    # {"role": "assistant", "content": []},
-    # {"role": "user", "content": "Tell me"},
-    # {"role": "user", "content": "a joke"},
-]
-tools = [
-    {
-        "function_declarations": [
-            {
-                "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"],
-                },
-            },
-            {
-                "name": "save_conversation",
-                "description": "Save the current conversation. Use this function to persist the current conversation to external storage.",
-                "parameters": {
-                    "type": "object",
-                    "properties": {
-                        "user_request_text": {
-                            "type": "string",
-                            "description": "The text of the user's request to save the conversation.",
-                        }
-                    },
-                    "required": ["user_request_text"],
-                },
-            },
-            {
-                "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": None,
-            },
-            {
-                "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"],
-                },
-            },
-            {
-                "name": "get_image",
-                "description": "Get and image from the camera or video stream.",
-                "parameters": {
-                    "type": "object",
-                    "properties": {
-                        "question": {
-                            "type": "string",
-                            "description": "The question to to use when running inference on the acquired image.",
-                        },
-                    },
-                    "required": ["question"],
-                },
-            },
-        ]
-    },
-]
-
-
-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 = GoogleLLMService(model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY"))
-
-        # 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)
-        llm.register_function("get_image", get_image)
-
-        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):
-            global video_participant_id
-            video_participant_id = participant["id"]
-            await transport.capture_participant_transcription(participant["id"])
-            await transport.capture_participant_video(video_participant_id, framerate=0)
-            # 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/21-tavus-layer.py

@@ -1,133 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import aiohttp
-import os
-import sys
-
-from typing import Any, Mapping
-
-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.cartesia import CartesiaTTSService
-from pipecat.services.openai import OpenAILLMService
-from pipecat.services.deepgram import DeepgramSTTService
-from pipecat.services.tavus import TavusVideoService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-from pipecat.audio.vad.silero import SileroVADAnalyzer
-
-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:
-        tavus = TavusVideoService(
-            api_key=os.getenv("TAVUS_API_KEY"),
-            replica_id=os.getenv("TAVUS_REPLICA_ID"),
-            persona_id=os.getenv("TAVUS_PERSONA_ID", "pipecat0"),
-            session=session,
-        )
-
-        # get persona, look up persona_name, set this as the bot name to ignore
-        persona_name = await tavus.get_persona_name()
-        room_url = await tavus.initialize()
-
-        transport = DailyTransport(
-            room_url=room_url,
-            token=None,
-            bot_name="Pipecat bot",
-            params=DailyParams(
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer(),
-                vad_audio_passthrough=True,
-            ),
-        )
-
-        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
-
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab",
-        )
-
-        llm = OpenAILLMService(model="gpt-4o-mini")
-
-        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.",
-            },
-        ]
-
-        tma_in = LLMUserResponseAggregator(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
-
-        pipeline = Pipeline(
-            [
-                transport.input(),  # Transport user input
-                stt,  # STT
-                tma_in,  # User responses
-                llm,  # LLM
-                tts,  # TTS
-                tavus,  # Tavus output layer
-                transport.output(),  # Transport bot output
-                tma_out,  # Assistant spoken responses
-            ]
-        )
-
-        task = PipelineTask(
-            pipeline,
-            PipelineParams(
-                allow_interruptions=True,
-                enable_metrics=True,
-                enable_usage_metrics=True,
-                report_only_initial_ttfb=True,
-            ),
-        )
-
-        @transport.event_handler("on_participant_joined")
-        async def on_participant_joined(
-            transport: DailyTransport, participant: Mapping[str, Any]
-        ) -> None:
-            # Ignore the Tavus replica's microphone
-            if participant.get("info", {}).get("userName", "") == persona_name:
-                logger.debug(f"Ignoring {participant['id']}'s microphone")
-                await transport.update_subscriptions(
-                    participant_settings={
-                        participant["id"]: {
-                            "media": {"microphone": "unsubscribed"},
-                        }
-                    }
-                )
-
-            if participant.get("info", {}).get("userName", "") != persona_name:
-                # 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/22-natural-conversation.py

@@ -1,168 +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.frames.frames import LLMMessagesFrame, TextFrame
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.parallel_pipeline import ParallelPipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.gated_openai_llm_context import GatedOpenAILLMContextAggregator
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
-from pipecat.processors.filters.null_filter import NullFilter
-from pipecat.processors.filters.wake_notifier_filter import WakeNotifierFilter
-from pipecat.processors.user_idle_processor import UserIdleProcessor
-from pipecat.services.cartesia import CartesiaTTSService
-from pipecat.services.deepgram import DeepgramSTTService
-from pipecat.services.openai import OpenAILLMService
-from pipecat.sync.event_notifier import EventNotifier
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-
-from 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, _) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            None,
-            "Respond bot",
-            DailyParams(
-                audio_out_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer(),
-                vad_audio_passthrough=True,
-            ),
-        )
-
-        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
-
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
-        )
-
-        # This is the LLM that will be used to detect if the user has finished a
-        # statement. This doesn't really need to be an LLM, we could use NLP
-        # libraries for that, but it was easier as an example because we
-        # leverage the context aggregators.
-        statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
-
-        statement_messages = [
-            {
-                "role": "system",
-                "content": "Determine if the user's statement is a complete sentence or question, ending in a natural pause or punctuation. Return 'YES' if it is complete and 'NO' if it seems to leave a thought unfinished.",
-            },
-        ]
-
-        statement_context = OpenAILLMContext(statement_messages)
-        statement_context_aggregator = statement_llm.create_context_aggregator(statement_context)
-
-        # This is the regular LLM.
-        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
-
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
-            },
-        ]
-
-        context = OpenAILLMContext(messages)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        # We have instructed the LLM to return 'YES' if it thinks the user
-        # completed a sentence. So, if it's 'YES' we will return true in this
-        # predicate which will wake up the notifier.
-        async def wake_check_filter(frame):
-            return frame.text == "YES"
-
-        # This is a notifier that we use to synchronize the two LLMs.
-        notifier = EventNotifier()
-
-        # This a filter that will wake up the notifier if the given predicate
-        # (wake_check_filter) returns true.
-        completness_check = WakeNotifierFilter(
-            notifier, types=(TextFrame,), filter=wake_check_filter
-        )
-
-        # This processor keeps the last context and will let it through once the
-        # notifier is woken up.
-        gated_context_aggregator = GatedOpenAILLMContextAggregator(notifier)
-
-        # Notify if the user hasn't said anything.
-        async def user_idle_notifier(frame):
-            await notifier.notify()
-
-        # Sometimes the LLM will fail detecting if a user has completed a
-        # sentence, this will wake up the notifier if that happens.
-        user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=3.0)
-
-        # The ParallePipeline input are the user transcripts. We have two
-        # contexts. The first one will be used to determine if the user finished
-        # a statement and if so the notifier will be woken up. The second
-        # context is simply the regular context but it's gated waiting for the
-        # notifier to be woken up.
-        pipeline = Pipeline(
-            [
-                transport.input(),  # Transport user input
-                stt,
-                ParallelPipeline(
-                    [
-                        statement_context_aggregator.user(),
-                        statement_llm,
-                        completness_check,
-                        NullFilter(),
-                    ],
-                    [context_aggregator.user(), gated_context_aggregator, llm],
-                ),
-                user_idle,
-                tts,  # TTS
-                transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
-            ]
-        )
-
-        task = PipelineTask(
-            pipeline,
-            PipelineParams(
-                allow_interruptions=True,
-                enable_metrics=True,
-                enable_usage_metrics=True,
-                report_only_initial_ttfb=True,
-            ),
-        )
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/99-anthropic-hackathon.py

@@ -1,298 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import base64
-import io
-import os
-import sys
-from collections import deque
-
-import aiohttp
-from dotenv import load_dotenv
-from loguru import logger
-from PIL import Image
-from runner import configure
-
-from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import (
-    BotInterruptionFrame,
-    Frame,
-    ImageRawFrame,
-    LLMFullResponseEndFrame,
-    LLMMessagesFrame,
-    TextFrame,
-    TranscriptionFrame,
-)
-from pipecat.pipeline.parallel_pipeline import ParallelPipeline
-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,
-    OpenAILLMContextFrame,
-)
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.processors.frameworks.rtvi import (
-    RTVIBotTranscriptionProcessor,
-    RTVIUserTranscriptionProcessor,
-)
-from pipecat.services.anthropic import AnthropicLLMContext, AnthropicLLMService
-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")
-
-MAX_FRAMES = 5
-FRAMES_PER_SECOND = 0.2
-
-
-video_participant_id = None
-anthropic_context = None
-recent_image_frames = deque(maxlen=MAX_FRAMES)
-most_recent_image_summary = ""
-
-
-class ImageFrameCatcher(FrameProcessor):
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        global recent_image_frames
-
-        await super().process_frame(frame, direction)
-        if isinstance(frame, ImageRawFrame):
-            recent_image_frames.append(frame)
-        else:
-            await self.push_frame(frame, direction)
-
-
-class TranscriptFrameCatcher(FrameProcessor):
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        await super().process_frame(frame, direction)
-        if isinstance(frame, TranscriptionFrame):
-            logger.debug(
-                f"TranscriptLogger: {frame}, num frames: {len(recent_image_frames)}, anthropic context: {anthropic_context}"
-            )
-            if anthropic_context:
-                add_message_with_images(
-                    anthropic_context, frame.text, frames=list(recent_image_frames)
-                )
-        await self.push_frame(frame, direction)
-
-
-class MessageFrameCatcher(FrameProcessor):
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        await super().process_frame(frame, direction)
-        if isinstance(frame, OpenAILLMContextFrame):
-            last_message = frame.context.messages[-1]
-
-            system_message = """
-Give me a concise summary of the images supplied.
-            """
-            frame = LLMMessagesFrame(
-                messages=[
-                    {
-                        "role": "system",
-                        "content": system_message,
-                    },
-                    last_message,
-                ],
-            )
-            await self.push_frame(frame, direction)
-            return
-
-
-class MessageFrameCatcher2(FrameProcessor):
-    def __init__(self):
-        super().__init__()
-        self.text_blob = ""
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        global most_recent_image_summary
-        await super().process_frame(frame, direction)
-        if isinstance(frame, TextFrame):
-            self.text_blob += f" {frame.text}"
-
-        if isinstance(frame, LLMFullResponseEndFrame):
-            logger.debug(f"MessageFrameCatcher2: {self.text_blob}")
-            most_recent_image_summary = self.text_blob
-            self.text_blob = ""
-
-        await self.push_frame(frame, direction)
-
-
-async def main():
-    global llm
-    global anthropic_context
-
-    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 = AnthropicLLMService(
-            api_key=os.getenv("ANTHROPIC_API_KEY"),
-            model="claude-3-5-sonnet-20240620",
-            enable_prompt_caching_beta=True,
-        )
-
-        vision_llm = AnthropicLLMService(
-            api_key=os.getenv("ANTHROPIC_API_KEY"),
-            model="claude-3-5-sonnet-20240620",
-            enable_prompt_caching_beta=True,
-        )
-
-        # todo: test with very short initial user message
-
-        system_prompt = """\
-You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions. Keep
-your answers brief unless explicitly asked for more information.
-
-Your response will be turned into speech so use only simple words and punctuation.
-        """
-
-        messages = [
-            {
-                "role": "system",
-                "content": [
-                    {
-                        "type": "text",
-                        "text": system_prompt,
-                    }
-                ],
-            },
-            {"role": "user", "content": "Start the conversation by saying 'hello'."},
-        ]
-
-        context = OpenAILLMContext(messages)
-        anthropic_context = AnthropicLLMContext.upgrade_to_anthropic(context)
-        context_aggregator = llm.create_context_aggregator(context)
-
-        rtvi_user_transcription = RTVIUserTranscriptionProcessor()
-        rtvi_bot_transcription = RTVIBotTranscriptionProcessor()
-
-        pipeline = Pipeline(
-            [
-                transport.input(),  # Transport user input
-                ImageFrameCatcher(),
-                TranscriptFrameCatcher(),
-                rtvi_user_transcription,
-                context_aggregator.user(),  # User speech to text
-                ParallelPipeline(
-                    [
-                        llm,  # LLM
-                        rtvi_bot_transcription,
-                        tts,  # TTS
-                        transport.output(),  # Transport bot output
-                        context_aggregator.assistant(),  # Assistant spoken responses and tool context
-                    ],
-                    [MessageFrameCatcher(), vision_llm, MessageFrameCatcher2()],
-                ),
-            ],
-        )
-
-        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):
-            global video_participant_id
-            video_participant_id = participant["id"]
-            await transport.capture_participant_transcription(video_participant_id)
-            await transport.capture_participant_video(
-                video_participant_id, framerate=FRAMES_PER_SECOND, video_source="screenVideo"
-            )
-            # Kick off the conversation.
-            await task.queue_frames([context_aggregator.user().get_context_frame()])
-
-        @transport.event_handler("on_app_message")
-        async def on_app_message(transport, message, sender):
-            logger.debug(f"Received app message: {message} - {context}")
-
-            if not recent_image_frames:
-                logger.debug("No image frames to send")
-                return
-
-            add_message_with_images(
-                anthropic_context, message["message"], frames=list(recent_image_frames)
-            )
-
-            interrupt_message = "STOP"
-
-            if interrupt_message == message["message"]:
-                logger.debug("Interrupting")
-                await task.queue_frames([BotInterruptionFrame()])
-            else:
-                await task.queue_frames([context_aggregator.user().get_context_frame()])
-
-        runner = PipelineRunner()
-        await runner.run(task)
-
-
-def add_message_with_images(c, message, frames=None):
-    if frames is None:
-        frames = list(recent_image_frames)
-
-    if not frames:
-        logger.debug("No image frames to send")
-        return
-
-    # Create content list starting with all images
-    content = []
-    for frame in frames:
-        buffer = io.BytesIO()
-        Image.frombytes(frame.format, frame.size, frame.image).save(buffer, format="JPEG")
-        encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
-
-        content.append(
-            {
-                "type": "image",
-                "source": {
-                    "type": "base64",
-                    "media_type": "image/jpeg",
-                    "data": encoded_image,
-                },
-            }
-        )
-
-    # Add text message at the end if provided
-    if message:
-        content.append({"type": "text", "text": message})
-
-    # Go through all messages and replace user messages containing images
-    if c.messages:
-        for i, msg in enumerate(c.messages):
-            if (
-                msg["role"] == "user"
-                and isinstance(msg["content"], list)
-                and len(msg["content"]) > 0
-            ):
-                if msg["content"][0].get("type") == "image":
-                    logger.debug(
-                        f"Replacing user message {i} containing images with summary: {most_recent_image_summary}"
-                    )
-                    c.messages[i] = {"role": "user", "content": most_recent_image_summary}
-
-    c.add_message({"role": "user", "content": content})
-
-
-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(),
             ]
         )
 
@@ -203,8 +202,8 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
-            await transport.capture_participant_video(participant["id"], framerate=0)
+            transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_video(participant["id"], framerate=0)
             ir.set_participant_id(participant["id"])
             await task.queue_frames([LLMMessagesFrame(messages)])
 

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
 
@@ -352,7 +352,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_transcription(participant["id"])
             print(f"Context is: {context}")
             await task.queue_frames([OpenAILLMContextFrame(context)])
 

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

@@ -17,10 +17,6 @@ from fastapi.responses import JSONResponse, RedirectResponse
 
 from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
 
-from dotenv import load_dotenv
-
-load_dotenv(override=True)
-
 MAX_BOTS_PER_ROOM = 1
 
 # Bot sub-process dict for status reporting and concurrency control
@@ -61,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 ------------- #
 
@@ -102,7 +108,7 @@ async def main(room_url, token=None):
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             logger.debug("Participant joined, storytime commence!")
-            await transport.capture_participant_transcription(participant["id"])
+            transport.capture_participant_transcription(participant["id"])
             await intro_task.queue_frames(
                 [
                     images["book1"],
@@ -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")