Compare commits

..

1 Commits

Author SHA1 Message Date
Brian Hill
23be3d8b57 [ENG-7781] Enable docker build without git 2024-08-12 08:47:36 -04:00
95 changed files with 2373 additions and 4290 deletions

View File

@@ -25,7 +25,14 @@ jobs:
id: setup_python
uses: actions/setup-python@v4
with:
python-version: "3.10"
python-version: '3.10'
- name: Cache virtual environment
uses: actions/cache@v3
with:
# We are hashing requirements-dev.txt and requirements-extra.txt which
# contain all dependencies needed to run the tests and examples.
key: venv-${{ runner.os }}-${{ steps.setup_python.outputs.python-version}}-${{ hashFiles('linux-py3.10-requirements.txt') }}-${{ hashFiles('dev-requirements.txt') }}
path: .venv
- name: Install system packages
run: sudo apt-get install -y portaudio19-dev
- name: Setup virtual environment
@@ -35,7 +42,7 @@ jobs:
run: |
source .venv/bin/activate
python -m pip install --upgrade pip
pip install -r dev-requirements.txt
pip install -r linux-py3.10-requirements.txt -r dev-requirements.txt
- name: Test with pytest
run: |
source .venv/bin/activate

View File

@@ -9,97 +9,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- A clock can now be specified to `PipelineTask` (defaults to
`SystemClock`). This clock will be passed to each frame processor via the
`StartFrame`.
- Added pipeline clocks. A pipeline clock is used by the output transport to
know when a frame needs to be presented. For that, all frames now have an
optional `pts` field (prensentation timestamp). There's currently just one
clock implementation `SystemClock` and the `pts` field is currently only used
for `TextFrame`s (audio and image frames will be next).
- `DailyTransport` now supports setting the audio bitrate to improve audio
quality through the `DailyParams.audio_out_bitrate` parameter. The new
default is 96kbps.
- `DailyTransport` now uses the number of audio output channels (1 or 2) to set
mono or stereo audio when needed.
- Interruptions support has been added to `TwilioFrameSerializer` when using
`FastAPIWebsocketTransport`.
- Added new `LmntTTSService` text-to-speech service.
(see https://www.lmnt.com/)
- Added `TTSModelUpdateFrame`, `TTSLanguageUpdateFrame`, `STTModelUpdateFrame`,
and `STTLanguageUpdateFrame` frames to allow you to switch models, language
and voices in TTS and STT services.
- Added new `transcriptions.Language` enum.
### Changed
- `CartesiaTTSService` and `ElevenLabsTTSService` now add presentation
timestamps to their text output. This allows the output transport to push the
text frames downstream at almost the same time the words are spoken. We say
"almost" because currently the audio frames don't have presentation timestamp
but they should be played at roughly the same time.
- `DailyTransport.on_joined` event now returns the full session data instead of
just the participant.
- `CartesiaTTSService` is now a subclass of `TTSService`.
- `DeepgramSTTService` is now a subclass of `STTService`.
- `WhisperSTTService` is now a subclass of `SegmentedSTTService`. A
`SegmentedSTTService` is a `STTService` where the provided audio is given in a
big chunk (i.e. from when the user starts speaking until the user stops
speaking) instead of a continous stream.
### Fixed
- `StartFrame` should be the first frame every processor receives to avoid
situations where things are not initialized (because initialization happens on
`StartFrame`) and other frames come in resulting in undesired behavior.
### Performance
- `obj_id()` and `obj_count()` now use `itertools.count` avoiding the need of
`threading.Lock`.
## [0.0.41] - 2024-08-22
### Added
- Added `LivekitFrameSerializer` audio frame serializer.
### Fixed
- Fix `FastAPIWebsocketOutputTransport` variable name clash with subclass.
- Fix an `AnthropicLLMService` issue with empty arguments in function calling.
### Other
- Fixed `studypal` example errors.
## [0.0.40] - 2024-08-20
### Added
- VAD parameters can now be dynamicallt updated using the
`VADParamsUpdateFrame`.
- `ErrorFrame` has now a `fatal` field to indicate the bot should exit if a
fatal error is pushed upstream (false by default). A new `FatalErrorFrame`
that sets this flag to true has been added.
- `AnthropicLLMService` now supports function calling and initial support for
prompt caching.
(see https://www.anthropic.com/news/prompt-caching)
- `ElevenLabsTTSService` can now specify ElevenLabs input parameters such as
`output_format`.
@@ -108,13 +17,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Added new `on_participant_updated` event to `DailyTransport`.
- Added `DailyRESTHelper.delete_room_by_name()` and
`DailyRESTHelper.delete_room_by_url()`.
- Added `DailyRESTHelper.delete_room_by_name()`.
- Added LLM and TTS usage metrics. Those are enabled when
`PipelineParams.enable_usage_metrics` is True.
- Added LLM and TTS usage metrics. Those will be enabled by when
`enable_usage_metrics` is True.
- `AudioRawFrame`s are now pushed downstream from the base output
- `AudioRawFrame`s are not pushed downstream from the base output
transport. This allows capturing the exact words the bot says by adding an STT
service at the end of the pipeline.
@@ -133,13 +41,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed
- Support RTVI message protocol 0.1. This includes new messages, support for
messages responses, support for actions, configuration, webhooks and a bunch
of new cool stuff.
(see https://docs.rtvi.ai/)
- `SileroVAD` dependency is now imported via pip's `silero-vad` package.
- `ElevenLabsTTSService` now uses `eleven_turbo_v2_5` model by default.
- `BotSpeakingFrame` is now a control frame.
@@ -151,13 +52,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
- `TTSStartFrame` and `TTSStopFrame` are now sent when TTS really starts and
stops. This allows for knowing when the bot starts and stops speaking even
with asynchronous services (like Cartesia).
- Fixed `AzureSTTService` transcription frame timestamps.
- Fixed an issue with `DailyRESTHelper.create_room()` expirations which would
- Fixed and issue with `DailyRESTHelper.create_room()` expirations which would
cause this function to stop working after the initial expiration elapsed.
- Improved `EndFrame` and `CancelFrame` handling. `EndFrame` should end things
@@ -178,14 +73,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Other
- Added `studypal` example (from to the Cartesia folks!).
- Most examples now use Cartesia.
- Added examples `foundational/19a-tools-anthropic.py`,
`foundational/19b-tools-video-anthropic.py` and
`foundational/19a-tools-togetherai.py`.
- Added examples `foundational/18-gstreamer-filesrc.py` and
`foundational/18a-gstreamer-videotestsrc.py` that show how to use
`GStreamerPipelineSource`

View File

@@ -4,7 +4,8 @@
# Pipecat
[![PyPI](https://img.shields.io/pypi/v/pipecat-ai)](https://pypi.org/project/pipecat-ai) [![Discord](https://img.shields.io/discord/1239284677165056021)](https://discord.gg/pipecat) <a href="https://app.commanddash.io/agent/github_pipecat-ai_pipecat"><img src="https://img.shields.io/badge/AI-Code%20Agent-EB9FDA"></a>
[![PyPI](https://img.shields.io/pypi/v/pipecat-ai)](https://pypi.org/project/pipecat-ai) [![Discord](https://img.shields.io/discord/1239284677165056021
)](https://discord.gg/pipecat)
`pipecat` is a framework for building voice (and multimodal) conversational agents. Things like personal coaches, meeting assistants, [story-telling toys for kids](https://storytelling-chatbot.fly.dev/), customer support bots, [intake flows](https://www.youtube.com/watch?v=lDevgsp9vn0), and snarky social companions.
@@ -38,7 +39,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`, `azure`, `deepgram`, `gladia`, `google`, `fal`, `lmnt`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper`, `xtts`
- **AI services**: `anthropic`, `azure`, `deepgram`, `gladia`, `google`, `fal`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper`, `xtts`
- **Transports**: `local`, `websocket`, `daily`
## Code examples
@@ -48,7 +49,7 @@ Your project may or may not need these, so they're made available as optional re
## A simple voice agent running locally
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.
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 [ElevenLabs](https://elevenlabs.io/) for text-to-speech.
```python
#app.py
@@ -60,7 +61,7 @@ from pipecat.frames.frames import EndFrame, TextFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask
from pipecat.pipeline.runner import PipelineRunner
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.transports.services.daily import DailyParams, DailyTransport
async def main():
@@ -72,10 +73,11 @@ async def main():
bot_name="Bot Name",
params=DailyParams(audio_out_enabled=True))
# Use Cartesia for Text-to-Speech
tts = CartesiaTTSService(
api_key=...,
voice_id=...
# Use Eleven Labs for Text-to-Speech
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=...,
voice_id=...,
)
# Simple pipeline that will process text to speech and output the result
@@ -92,7 +94,7 @@ async def main():
@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)
# Queue a TextFrame that will get spoken by the TTS service (Eleven Labs)
await task.queue_frames([TextFrame(f"Hello there, {participant_name}!"), EndFrame()])
# Run the pipeline task
@@ -144,20 +146,20 @@ source venv/bin/activate
From the root of this repo, run the following:
```shell
pip install -r dev-requirements.txt
pip install -r dev-requirements.txt -r {env}-requirements.txt
python -m build
```
This builds the package. To use the package locally (e.g. to run sample files), run
This builds the package. To use the package locally (eg to run sample files), run
```shell
pip install --editable ".[option,...]"
pip install --editable .
```
If you want to use this package from another directory, you can run:
```shell
pip install "path_to_this_repo[option,...]"
pip install path_to_this_repo
```
### Running tests

View File

@@ -1,8 +1,8 @@
autopep8~=2.3.1
autopep8~=2.1.0
build~=1.2.1
grpcio-tools~=1.62.2
pip-tools~=7.4.1
pyright~=1.1.376
pytest~=8.3.2
setuptools~=72.2.0
pyright~=1.1.367
pytest~=8.2.0
setuptools~=71.1.0
setuptools_scm~=8.1.0

View File

@@ -30,10 +30,6 @@ FIREWORKS_API_KEY=...
# Gladia
GLADIA_API_KEY=...
# LMNT
LMNT_API_KEY=...
LMNT_VOICE_ID=...
# PlayHT
PLAY_HT_USER_ID=...
PLAY_HT_API_KEY=...

View File

@@ -41,7 +41,6 @@ Next, follow the steps in the README for each demo.
| [Patient intake](patient-intake) | A chatbot that can call functions in response to user input. | Deepgram, ElevenLabs, OpenAI, Daily, Daily Prebuilt UI |
| [Dialin Chatbot](dialin-chatbot) | A chatbot that connects to an incoming phone call from Daily or Twilio. | Deepgram, ElevenLabs, OpenAI, Daily, Twilio |
| [Twilio Chatbot](twilio-chatbot) | A chatbot that connects to an incoming phone call from Twilio. | Deepgram, ElevenLabs, OpenAI, Daily, Twilio |
| [studypal](studypal) | A chatbot to have a conversation about any article on the web | |
> [!IMPORTANT]
> These example projects use Daily as a WebRTC transport and can be joined using their hosted Prebuilt UI.

View File

@@ -1,4 +1,5 @@
import asyncio
import aiohttp
import os
import sys
import argparse
@@ -26,69 +27,71 @@ daily_api_url = os.getenv("DAILY_API_URL", "https://api.daily.co/v1")
async def main(room_url: str, token: str):
transport = DailyTransport(
room_url,
token,
"Chatbot",
DailyParams(
api_url=daily_api_url,
api_key=daily_api_key,
audio_in_enabled=True,
audio_out_enabled=True,
camera_out_enabled=False,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
transcription_enabled=True,
async with aiohttp.ClientSession() as session:
transport = DailyTransport(
room_url,
token,
"Chatbot",
DailyParams(
api_url=daily_api_url,
api_key=daily_api_key,
audio_in_enabled=True,
audio_out_enabled=True,
camera_out_enabled=False,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
transcription_enabled=True,
)
)
)
tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY", ""),
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
)
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY", ""),
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o")
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o")
messages = [
{
"role": "system",
"content": "You are Chatbot, a friendly, helpful robot. Your output will be converted to audio so don't include special characters other than '!' or '?' in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by saying hello.",
},
]
messages = [
{
"role": "system",
"content": "You are Chatbot, a friendly, helpful robot. Your output will be converted to audio so don't include special characters other than '!' or '?' in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by saying hello.",
},
]
tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages)
tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages)
pipeline = Pipeline([
transport.input(),
tma_in,
llm,
tts,
transport.output(),
tma_out,
])
pipeline = Pipeline([
transport.input(),
tma_in,
llm,
tts,
transport.output(),
tma_out,
])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
await task.queue_frames([LLMMessagesFrame(messages)])
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
await task.queue_frames([LLMMessagesFrame(messages)])
@transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason):
await task.queue_frame(EndFrame())
@transport.event_handler("on_call_state_updated")
async def on_call_state_updated(transport, state):
if state == "left":
@transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason):
await task.queue_frame(EndFrame())
runner = PipelineRunner()
@transport.event_handler("on_call_state_updated")
async def on_call_state_updated(transport, state):
if state == "left":
await task.queue_frame(EndFrame())
await runner.run(task)
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":

View File

@@ -1,4 +1,5 @@
import asyncio
import aiohttp
import os
import sys
import argparse
@@ -28,74 +29,75 @@ daily_api_url = os.getenv("DAILY_API_URL", "https://api.daily.co/v1")
async def main(room_url: str, token: str, callId: str, callDomain: str):
# diallin_settings are only needed if Daily's SIP URI is used
# If you are handling this via Twilio, Telnyx, set this to None
# and handle call-forwarding when on_dialin_ready fires.
diallin_settings = DailyDialinSettings(
call_id=callId,
call_domain=callDomain
)
transport = DailyTransport(
room_url,
token,
"Chatbot",
DailyParams(
api_url=daily_api_url,
api_key=daily_api_key,
dialin_settings=diallin_settings,
audio_in_enabled=True,
audio_out_enabled=True,
camera_out_enabled=False,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
transcription_enabled=True,
async with aiohttp.ClientSession() as session:
# diallin_settings are only needed if Daily's SIP URI is used
# If you are handling this via Twilio, Telnyx, set this to None
# and handle call-forwarding when on_dialin_ready fires.
diallin_settings = DailyDialinSettings(
call_id=callId,
call_domain=callDomain
)
)
tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY", ""),
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
)
transport = DailyTransport(
room_url,
token,
"Chatbot",
DailyParams(
api_url=daily_api_url,
api_key=daily_api_key,
dialin_settings=diallin_settings,
audio_in_enabled=True,
audio_out_enabled=True,
camera_out_enabled=False,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
transcription_enabled=True,
)
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o"
)
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY", ""),
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
)
messages = [
{
"role": "system",
"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 saying 'Oh, hello! Who dares dial me at this hour?!'.",
},
]
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o")
tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages)
messages = [
{
"role": "system",
"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 saying 'Oh, hello! Who dares dial me at this hour?!'.",
},
]
pipeline = Pipeline([
transport.input(),
tma_in,
llm,
tts,
transport.output(),
tma_out,
])
tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages)
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
pipeline = Pipeline([
transport.input(),
tma_in,
llm,
tts,
transport.output(),
tma_out,
])
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
await task.queue_frames([LLMMessagesFrame(messages)])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason):
await task.queue_frame(EndFrame())
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
await task.queue_frames([LLMMessagesFrame(messages)])
runner = PipelineRunner()
@transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason):
await task.queue_frame(EndFrame())
await runner.run(task)
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":

View File

@@ -1,4 +1,5 @@
import asyncio
import aiohttp
import os
import sys
import argparse
@@ -35,81 +36,82 @@ daily_api_key = os.getenv("DAILY_API_KEY", "")
async def main(room_url: str, token: str, callId: str, sipUri: str):
# dialin_settings are only needed if Daily's SIP URI is used
# If you are handling this via Twilio, Telnyx, set this to None
# and handle call-forwarding when on_dialin_ready fires.
transport = DailyTransport(
room_url,
token,
"Chatbot",
DailyParams(
api_key=daily_api_key,
dialin_settings=None, # Not required for Twilio
audio_in_enabled=True,
audio_out_enabled=True,
camera_out_enabled=False,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
transcription_enabled=True,
)
)
tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY", ""),
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o"
)
messages = [
{
"role": "system",
"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 saying 'Hello! Who dares dial me at this hour?!'.",
},
]
tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages)
pipeline = Pipeline([
transport.input(),
tma_in,
llm,
tts,
transport.output(),
tma_out,
])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
await task.queue_frames([LLMMessagesFrame(messages)])
@transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason):
await task.queue_frame(EndFrame())
@transport.event_handler("on_dialin_ready")
async def on_dialin_ready(transport, cdata):
# For Twilio, Telnyx, etc. You need to update the state of the call
# and forward it to the sip_uri..
print(f"Forwarding call: {callId} {sipUri}")
try:
# The TwiML is updated using Twilio's client library
call = twilioclient.calls(callId).update(
twiml=f'<Response><Dial><Sip>{sipUri}</Sip></Dial></Response>'
async with aiohttp.ClientSession() as session:
# diallin_settings are only needed if Daily's SIP URI is used
# If you are handling this via Twilio, Telnyx, set this to None
# and handle call-forwarding when on_dialin_ready fires.
transport = DailyTransport(
room_url,
token,
"Chatbot",
DailyParams(
api_key=daily_api_key,
dialin_settings=None, # Not required for Twilio
audio_in_enabled=True,
audio_out_enabled=True,
camera_out_enabled=False,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
transcription_enabled=True,
)
except Exception as e:
raise Exception(f"Failed to forward call: {str(e)}")
)
runner = PipelineRunner()
await runner.run(task)
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY", ""),
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o")
messages = [
{
"role": "system",
"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 saying 'Hello! Who dares dial me at this hour?!'.",
},
]
tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages)
pipeline = Pipeline([
transport.input(),
tma_in,
llm,
tts,
transport.output(),
tma_out,
])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
await task.queue_frames([LLMMessagesFrame(messages)])
@transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason):
await task.queue_frame(EndFrame())
@transport.event_handler("on_dialin_ready")
async def on_dialin_ready(transport, cdata):
# For Twilio, Telnyx, etc. You need to update the state of the call
# and forward it to the sip_uri..
print(f"Forwarding call: {callId} {sipUri}")
try:
# The TwiML is updated using Twilio's client library
call = twilioclient.calls(callId).update(
twiml=f'<Response><Dial><Sip>{sipUri}</Sip></Dial></Response>'
)
except Exception as e:
raise Exception(f"Failed to forward call: {str(e)}")
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":

View File

@@ -3,4 +3,3 @@ fastapi
uvicorn
python-dotenv
twilio
python-multipart

View File

@@ -9,11 +9,11 @@ import aiohttp
import os
import sys
from pipecat.frames.frames import TextFrame
from pipecat.frames.frames import EndFrame, TextFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask
from pipecat.pipeline.runner import PipelineRunner
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from runner import configure
@@ -34,9 +34,10 @@ async def main():
transport = DailyTransport(
room_url, None, "Say One Thing", DailyParams(audio_out_enabled=True))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
runner = PipelineRunner()
@@ -48,7 +49,7 @@ async def main():
@transport.event_handler("on_participant_joined")
async def on_new_participant_joined(transport, participant):
participant_name = participant["info"]["userName"] or ''
await task.queue_frame(TextFrame(f"Hello there, {participant_name}!"))
await task.queue_frames([TextFrame(f"Hello there, {participant_name}!"), EndFrame()])
await runner.run(task)

View File

@@ -9,11 +9,11 @@ import aiohttp
import os
import sys
from pipecat.frames.frames import TextFrame
from pipecat.frames.frames import EndFrame, TextFrame
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.elevenlabs import ElevenLabsTTSService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.local.audio import LocalAudioTransport
@@ -30,9 +30,10 @@ async def main():
async with aiohttp.ClientSession() as session:
transport = LocalAudioTransport(TransportParams(audio_out_enabled=True))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
pipeline = Pipeline([tts, transport.output()])
@@ -41,7 +42,7 @@ async def main():
async def say_something():
await asyncio.sleep(1)
await task.queue_frame(TextFrame("Hello there!"))
await task.queue_frames([TextFrame("Hello there!"), EndFrame()])
runner = PipelineRunner()

View File

@@ -9,11 +9,11 @@ import aiohttp
import os
import sys
from pipecat.frames.frames import LLMMessagesFrame
from pipecat.frames.frames import EndFrame, LLMMessagesFrame
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.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -38,9 +38,10 @@ async def main():
"Say One Thing From an LLM",
DailyParams(audio_out_enabled=True))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
llm = OpenAILLMService(
@@ -59,7 +60,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))
await task.queue_frames([LLMMessagesFrame(messages), EndFrame()])
await runner.run(task)

View File

@@ -56,7 +56,7 @@ async def main():
runner = PipelineRunner()
async def run_tk():
while not task.has_finished():
while runner.is_active():
tk_root.update()
tk_root.update_idletasks()
await asyncio.sleep(0.1)

View File

@@ -13,6 +13,7 @@ from dataclasses import dataclass
from pipecat.frames.frames import (
AppFrame,
EndFrame,
Frame,
ImageRawFrame,
LLMFullResponseStartFrame,
@@ -89,6 +90,7 @@ async def main():
)
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
@@ -150,6 +152,8 @@ async def main():
frames.append(MonthFrame(month=month))
frames.append(LLMMessagesFrame(messages))
frames.append(EndFrame())
runner = PipelineRunner()
task = PipelineTask(pipeline)

View File

@@ -85,6 +85,7 @@ async def main():
model="gpt-4o")
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"))
@@ -136,11 +137,14 @@ async def main():
task = PipelineTask(pipeline)
# We only specify a few months as we create tasks all at once and we
# might get rate limited otherwise.
# We only specify 5 months as we create tasks all at once and we might
# get rate limited otherwise.
months: list[str] = [
"January",
"February",
# "March",
# "April",
# "May",
]
# We create one task per month. This will be executed concurrently.

View File

@@ -18,7 +18,8 @@ from pipecat.processors.aggregators.llm_response import (
LLMUserResponseAggregator,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.processors.logger import FrameLogger
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
@@ -58,9 +59,10 @@ async def main():
)
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
llm = OpenAILLMService(

View File

@@ -79,6 +79,7 @@ async def main():
)
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)

View File

@@ -15,7 +15,7 @@ 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.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
@@ -47,9 +47,10 @@ async def main():
)
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
llm = OpenAILLMService(

View File

@@ -15,7 +15,7 @@ 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.elevenlabs import ElevenLabsTTSService
from pipecat.services.anthropic import AnthropicLLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
@@ -47,9 +47,10 @@ async def main():
)
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
llm = AnthropicLLMService(

View File

@@ -17,7 +17,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_response import (
LLMAssistantResponseAggregator, LLMUserResponseAggregator)
from pipecat.processors.frameworks.langchain import LangchainProcessor
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
@@ -63,9 +63,10 @@ async def main():
),
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
prompt = ChatPromptTemplate.from_messages(

View File

@@ -15,11 +15,12 @@ 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.lmnt import LmntTTSService
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
@@ -40,17 +41,18 @@ async def main():
token,
"Respond bot",
DailyParams(
audio_out_sample_rate=44100,
audio_out_enabled=True,
audio_out_sample_rate=24000,
transcription_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer()
)
)
tts = LmntTTSService(
api_key=os.getenv("LMNT_API_KEY"),
voice_id="morgan"
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="a0e99841-438c-4a64-b679-ae501e7d6091", # Barbershop Man
sample_rate=44100,
)
llm = OpenAILLMService(
@@ -72,11 +74,11 @@ async def main():
tma_in, # User responses
llm, # LLM
tts, # TTS
tma_out, # Goes before the transport because cartesia has word-level timestamps!
transport.output(), # Transport bot output
tma_out # Assistant spoken responses
])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True))
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):

View File

@@ -1,99 +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.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
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_out_enabled=True,
transcription_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer()
)
)
tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY", ""),
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
)
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.",
},
]
tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages)
pipeline = Pipeline([
transport.input(), # Transport user input
tma_in, # User responses
llm, # LLM
tts, # TTS
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_first_participant_joined")
async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"])
# Kick off the conversation.
messages.append(
{"role": "system", "content": "Please introduce yourself to the user."})
await task.queue_frames([LLMMessagesFrame(messages)])
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":
asyncio.run(main())

View File

@@ -17,7 +17,7 @@ from pipecat.processors.aggregators.llm_response import (
LLMAssistantResponseAggregator,
LLMUserResponseAggregator,
)
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.openpipe import OpenPipeLLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
@@ -50,9 +50,10 @@ async def main():
)
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
timestamp = int(time.time())

View File

@@ -15,9 +15,10 @@ 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.deepgram import DeepgramSTTService, DeepgramTTSService
from pipecat.services.gladia import GladiaSTTService
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
@@ -52,9 +53,10 @@ async def main():
api_key=os.getenv("GLADIA_API_KEY"),
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = DeepgramTTSService(
aiohttp_session=session,
api_key=os.getenv("DEEPGRAM_API_KEY"),
voice="aura-helios-en"
)
llm = OpenAILLMService(

View File

@@ -15,7 +15,7 @@ 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.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
@@ -47,9 +47,10 @@ async def main():
)
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
llm = OpenAILLMService(

View File

@@ -104,6 +104,7 @@ async def main():
model="gpt-4o")
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id="ErXwobaYiN019PkySvjV",
)

View File

@@ -16,7 +16,7 @@ from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.user_response import UserResponseAggregator
from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.moondream import MoondreamService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
@@ -65,6 +65,12 @@ async def main():
)
)
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
user_response = UserResponseAggregator()
image_requester = UserImageRequester()
@@ -74,9 +80,10 @@ async def main():
# If you run into weird description, try with use_cpu=True
moondream = MoondreamService()
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
@transport.event_handler("on_first_participant_joined")

View File

@@ -16,7 +16,7 @@ from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.user_response import UserResponseAggregator
from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.google import GoogleLLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
@@ -76,9 +76,10 @@ async def main():
model="gemini-1.5-flash-latest",
api_key=os.getenv("GOOGLE_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
@transport.event_handler("on_first_participant_joined")

View File

@@ -16,7 +16,7 @@ from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.user_response import UserResponseAggregator
from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.cartesia import CartesiaTTSService
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
@@ -76,9 +76,10 @@ async def main():
model="gpt-4o"
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
@transport.event_handler("on_first_participant_joined")

View File

@@ -16,7 +16,7 @@ from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.user_response import UserResponseAggregator
from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.anthropic import AnthropicLLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
@@ -72,13 +72,14 @@ async def main():
vision_aggregator = VisionImageFrameAggregator()
anthropic = AnthropicLLMService(
api_key=os.getenv("ANTHROPIC_API_KEY")
api_key=os.getenv("ANTHROPIC_API_KEY"),
model="claude-3-sonnet-20240229"
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
sample_rate=16000,
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
@transport.event_handler("on_first_participant_joined")

View File

@@ -13,8 +13,12 @@ from pipecat.frames.frames import TextFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.llm_response import (
LLMAssistantContextAggregator,
LLMUserContextAggregator,
)
from pipecat.processors.logger import FrameLogger
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMContext, OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
@@ -32,12 +36,12 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def start_fetch_weather(function_name, llm, context):
await llm.push_frame(TextFrame("Let me check on that."))
async def start_fetch_weather(llm):
await llm.push_frame(TextFrame("Let me think."))
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 fetch_weather_from_api(llm, args):
return {"conditions": "nice", "temperature": "75"}
async def main():
@@ -56,18 +60,17 @@ async def main():
)
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o")
# 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,
"get_current_weather",
fetch_weather_from_api,
start_callback=start_fetch_weather)
@@ -108,17 +111,17 @@ async def main():
]
context = OpenAILLMContext(messages, tools)
context_aggregator = llm.create_context_aggregator(context)
tma_in = LLMUserContextAggregator(context)
tma_out = LLMAssistantContextAggregator(context)
pipeline = Pipeline([
fl_in,
transport.input(),
context_aggregator.user(),
tma_in,
llm,
fl_out,
tts,
transport.output(),
context_aggregator.assistant(),
tma_out
])
task = PipelineTask(pipeline)

View File

@@ -14,6 +14,10 @@ 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.llm_response import (
LLMAssistantContextAggregator,
LLMUserContextAggregator
)
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.filters.function_filter import FunctionFilter
from pipecat.services.cartesia import CartesiaTTSService
@@ -36,10 +40,10 @@ logger.add(sys.stderr, level="DEBUG")
current_voice = "News Lady"
async def switch_voice(function_name, tool_call_id, args, llm, context, result_callback):
async def switch_voice(llm, args):
global current_voice
current_voice = args["voice"]
await result_callback({"voice": f"You are now using your {current_voice} voice. Your responses should now be as if you were a {current_voice}."})
return {"voice": f"You are now using your {current_voice} voice. Your responses should now be as if you were a {current_voice}."}
async def news_lady_filter(frame) -> bool:
@@ -115,11 +119,12 @@ async def main():
]
context = OpenAILLMContext(messages, tools)
context_aggregator = llm.create_context_aggregator(context)
tma_in = LLMUserContextAggregator(context)
tma_out = LLMAssistantContextAggregator(context)
pipeline = Pipeline([
transport.input(), # Transport user input
context_aggregator.user(), # User responses
tma_in, # User responses
llm, # LLM
ParallelPipeline( # TTS (one of the following vocies)
[FunctionFilter(news_lady_filter), news_lady], # News Lady voice
@@ -127,7 +132,7 @@ async def main():
[FunctionFilter(barbershop_man_filter), barbershop_man], # Barbershop Man voice
),
transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses
tma_out # Assistant spoken responses
])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))

View File

@@ -14,9 +14,13 @@ 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.llm_response import (
LLMAssistantContextAggregator,
LLMUserContextAggregator
)
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.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.services.whisper import Model, WhisperSTTService
from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -37,10 +41,10 @@ logger.add(sys.stderr, level="DEBUG")
current_language = "English"
async def switch_language(function_name, tool_call_id, args, llm, context, result_callback):
async def switch_language(llm, args):
global current_language
current_language = args["language"]
await result_callback({"voice": f"Your answers from now on should be in {current_language}."})
return {"voice": f"Your answers from now on should be in {current_language}."}
async def english_filter(frame) -> bool:
@@ -70,14 +74,17 @@ async def main():
stt = WhisperSTTService(model=Model.LARGE)
english_tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
english_tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id="pNInz6obpgDQGcFmaJgB",
)
spanish_tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="846d6cb0-2301-48b6-9683-48f5618ea2f6", # Spanish-speaking Lady
spanish_tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
model="eleven_multilingual_v2",
voice_id="9F4C8ztpNUmXkdDDbz3J",
)
llm = OpenAILLMService(
@@ -110,19 +117,20 @@ async def main():
]
context = OpenAILLMContext(messages, tools)
context_aggregator = llm.create_context_aggregator(context)
tma_in = LLMUserContextAggregator(context)
tma_out = LLMAssistantContextAggregator(context)
pipeline = Pipeline([
transport.input(), # Transport user input
stt, # STT
context_aggregator.user(), # User responses
tma_in, # User responses
llm, # LLM
ParallelPipeline( # TTS (bot will speak the chosen language)
[FunctionFilter(english_filter), english_tts], # English
[FunctionFilter(spanish_filter), spanish_tts], # Spanish
),
transport.output(), # Transport bot output
context_aggregator.assistant() # Assistant spoken responses
tma_out # Assistant spoken responses
])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))

View File

@@ -15,8 +15,9 @@ from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_response import (
LLMAssistantResponseAggregator, LLMUserResponseAggregator)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.processors.user_idle_processor import UserIdleProcessor
from pipecat.services.cartesia import CartesiaTTSService
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
@@ -48,9 +49,10 @@ async def main():
)
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
llm = OpenAILLMService(

View File

@@ -1,116 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import aiohttp
import os
import sys
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
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_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 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 = AnthropicLLMService(
api_key=os.getenv("ANTHROPIC_API_KEY"),
model="claude-3-5-sonnet-20240620"
)
llm.register_function("get_weather", get_weather)
tools = [
{
"name": "get_weather",
"description": "Get the current weather in a given location",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
}
},
"required": ["location"],
},
}
]
# todo: test with very short initial user message
# messages = [{"role": "system",
# "content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation."},
# {"role": "user",
# "content": " Start the conversation by introducing yourself."}]
messages = [{"role": "user", "content": "Say 'hello' to start the conversation."}]
context = OpenAILLMContext(messages, tools)
context_aggregator = llm.create_context_aggregator(context)
pipeline = Pipeline([
transport.input(), # Transport user input
context_aggregator.user(), # User spoken responses
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([context_aggregator.user().get_context_frame()])
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":
asyncio.run(main())

View File

@@ -1,172 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import aiohttp
import os
import sys
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
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")
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):
question = arguments["question"]
await llm.request_image_frame(user_id=video_participant_id, text_content=question)
async def main():
global llm
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
)
llm.register_function("get_weather", get_weather)
llm.register_function("get_image", get_image)
tools = [
{
"name": "get_weather",
"description": "Get the current weather in a given location",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
}
},
"required": ["location"],
},
},
{
"name": "get_image",
"description": "Get an image from the video stream.",
"input_schema": {
"type": "object",
"properties": {
"question": {
"type": "string",
"description": "The question that the user is asking about the image.",
}
},
"required": ["question"],
},
}
]
# 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.
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?
If you need to use a tool, simply use the tool. Do not tell the user the tool you are using. Be brief and concise.
"""
messages = [
{
"role": "system",
"content": [
{
"type": "text",
"text": system_prompt,
}
]
},
{
"role": "user",
"content": "Start the conversation by introducing yourself."
}]
context = OpenAILLMContext(messages, tools)
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):
global video_participant_id
video_participant_id = participant["id"]
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()])
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":
asyncio.run(main())

View File

@@ -1,138 +0,0 @@
#
# 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())

View File

@@ -31,7 +31,7 @@ from pipecat.processors.aggregators.llm_response import LLMUserResponseAggregato
from pipecat.processors.aggregators.sentence import SentenceAggregator
from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.moondream import MoondreamService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -153,9 +153,10 @@ async def main():
)
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id="pNInz6obpgDQGcFmaJgB",
)
llm = OpenAILLMService(

View File

@@ -10,14 +10,24 @@ import os
import sys
import wave
from typing import List
from openai._types import NotGiven, NOT_GIVEN
from openai.types.chat import (
ChatCompletionToolParam,
)
from pipecat.frames.frames import AudioRawFrame
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 LLMUserContextAggregator, LLMAssistantContextAggregator
from pipecat.processors.logger import FrameLogger
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService
from pipecat.services.ai_services import AIService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
@@ -54,11 +64,20 @@ for file in sound_files:
class IntakeProcessor:
def __init__(self, context: OpenAILLMContext):
def __init__(
self,
context: OpenAILLMContext,
llm: AIService,
tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN,
*args,
**kwargs,
):
super().__init__(*args, **kwargs)
self._context: OpenAILLMContext = context
self._llm = llm
print(f"Initializing context from IntakeProcessor")
context.add_message({"role": "system", "content": "You are Jessica, an agent for a company called Tri-County Health Services. Your job is to collect important information from the user before their doctor visit. You're talking to Chad Bailey. You should address the user by their first name and be polite and professional. You're not a medical professional, so you shouldn't provide any advice. Keep your responses short. Your job is to collect information to give to a doctor. Don't make assumptions about what values to plug into functions. Ask for clarification if a user response is ambiguous. Start by introducing yourself. Then, ask the user to confirm their identity by telling you their birthday, including the year. When they answer with their birthday, call the verify_birthday function."})
context.set_tools([
self._context.add_message({"role": "system", "content": "You are Jessica, an agent for a company called Tri-County Health Services. Your job is to collect important information from the user before their doctor visit. You're talking to Chad Bailey. You should address the user by their first name and be polite and professional. You're not a medical professional, so you shouldn't provide any advice. Keep your responses short. Your job is to collect information to give to a doctor. Don't make assumptions about what values to plug into functions. Ask for clarification if a user response is ambiguous. Start by introducing yourself. Then, ask the user to confirm their identity by telling you their birthday, including the year. When they answer with their birthday, call the verify_birthday function."})
self._context.set_tools([
{
"type": "function",
"function": {
@@ -74,17 +93,18 @@ class IntakeProcessor:
},
},
}])
# Create an allowlist of functions that the LLM can call
self._functions = [
"verify_birthday",
"list_prescriptions",
"list_allergies",
"list_conditions",
"list_visit_reasons",
]
async def verify_birthday(
self,
function_name,
tool_call_id,
args,
llm,
context,
result_callback):
async def verify_birthday(self, llm, args):
if args["birthday"] == "1983-01-01":
context.set_tools(
self._context.set_tools(
[
{
"type": "function",
@@ -114,18 +134,18 @@ class IntakeProcessor:
},
}])
# It's a bit weird to push this to the LLM, but it gets it into the pipeline
# await llm.push_frame(sounds["ding2.wav"], FrameDirection.DOWNSTREAM)
await llm.push_frame(sounds["ding2.wav"], FrameDirection.DOWNSTREAM)
# We don't need the function call in the context, so just return a new
# system message and let the framework re-prompt
await result_callback([{"role": "system", "content": "Next, thank the user for confirming their identity, then ask the user to list their current prescriptions. Each prescription needs to have a medication name and a dosage. Do not call the list_prescriptions function with any unknown dosages."}])
return [{"role": "system", "content": "Next, thank the user for confirming their identity, then ask the user to list their current prescriptions. Each prescription needs to have a medication name and a dosage. Do not call the list_prescriptions function with any unknown dosages."}]
else:
# The user provided an incorrect birthday; ask them to try again
await result_callback([{"role": "system", "content": "The user provided an incorrect birthday. Ask them for their birthday again. When they answer, call the verify_birthday function."}])
return [{"role": "system", "content": "The user provided an incorrect birthday. Ask them for their birthday again. When they answer, call the verify_birthday function."}]
async def start_prescriptions(self, function_name, llm, context):
async def start_prescriptions(self, llm):
print(f"!!! doing start prescriptions")
# Move on to allergies
context.set_tools(
self._context.set_tools(
[
{
"type": "function",
@@ -149,18 +169,18 @@ class IntakeProcessor:
},
},
}])
context.add_message(
self._context.add_message(
{
"role": "system",
"content": "Next, ask the user if they have any allergies. Once they have listed their allergies or confirmed they don't have any, call the list_allergies function."})
print(f"!!! about to await llm process frame in start prescrpitions")
await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
await llm.process_frame(OpenAILLMContextFrame(self._context), FrameDirection.DOWNSTREAM)
print(f"!!! past await process frame in start prescriptions")
async def start_allergies(self, function_name, llm, context):
async def start_allergies(self, llm):
print("!!! doing start allergies")
# Move on to conditions
context.set_tools(
self._context.set_tools(
[
{
"type": "function",
@@ -185,16 +205,16 @@ class IntakeProcessor:
},
},
])
context.add_message(
self._context.add_message(
{
"role": "system",
"content": "Now ask the user if they have any medical conditions the doctor should know about. Once they've answered the question, call the list_conditions function."})
await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
await llm.process_frame(OpenAILLMContextFrame(self._context), FrameDirection.DOWNSTREAM)
async def start_conditions(self, function_name, llm, context):
async def start_conditions(self, llm):
print("!!! doing start conditions")
# Move on to visit reasons
context.set_tools(
self._context.set_tools(
[
{
"type": "function",
@@ -218,25 +238,23 @@ class IntakeProcessor:
},
},
}])
context.add_message(
{
"role": "system",
"content": "Finally, ask the user the reason for their doctor visit today. Once they answer, call the list_visit_reasons function."})
await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
self._context.add_message(
{"role": "system", "content": "Finally, ask the user the reason for their doctor visit today. Once they answer, call the list_visit_reasons function."})
await llm.process_frame(OpenAILLMContextFrame(self._context), FrameDirection.DOWNSTREAM)
async def start_visit_reasons(self, function_name, llm, context):
async def start_visit_reasons(self, llm):
print("!!! doing start visit reasons")
# move to finish call
context.set_tools([])
context.add_message({"role": "system",
"content": "Now, thank the user and end the conversation."})
await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
self._context.set_tools([])
self._context.add_message({"role": "system",
"content": "Now, thank the user and end the conversation."})
await llm.process_frame(OpenAILLMContextFrame(self._context), FrameDirection.DOWNSTREAM)
async def save_data(self, function_name, tool_call_id, args, llm, context, result_callback):
async def save_data(self, llm, args):
logger.info(f"!!! Saving data: {args}")
# Since this is supposed to be "async", returning None from the callback
# will prevent adding anything to context or re-prompting
await result_callback(None)
return None
async def main():
@@ -263,15 +281,20 @@ async def main():
)
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
)
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
#
# English
#
voice_id="pNInz6obpgDQGcFmaJgB",
# tts = CartesiaTTSService(
# api_key=os.getenv("CARTESIA_API_KEY"),
# voice_id="846d6cb0-2301-48b6-9683-48f5618ea2f6", # Spanish-speaking Lady
# )
#
# Spanish
#
# model="eleven_multilingual_v2",
# voice_id="gD1IexrzCvsXPHUuT0s3",
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
@@ -279,9 +302,10 @@ async def main():
messages = []
context = OpenAILLMContext(messages=messages)
context_aggregator = llm.create_context_aggregator(context)
user_context = LLMUserContextAggregator(context)
assistant_context = LLMAssistantContextAggregator(context)
intake = IntakeProcessor(context)
intake = IntakeProcessor(context, llm)
llm.register_function("verify_birthday", intake.verify_birthday)
llm.register_function(
"list_prescriptions",
@@ -304,12 +328,12 @@ async def main():
pipeline = Pipeline([
transport.input(), # Transport input
context_aggregator.user(), # User responses
user_context, # User responses
llm, # LLM
fl, # Frame logger
tts, # TTS
transport.output(), # Transport output
context_aggregator.assistant(), # Assistant responses
assistant_context, # Assistant responses
])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=False))

View File

@@ -26,7 +26,7 @@ from pipecat.frames.frames import (
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.transports.services.daily import DailyParams, DailyTranscriptionSettings, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
from runner import configure
@@ -111,6 +111,7 @@ async def main():
)
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
#
# English

View File

@@ -60,6 +60,7 @@ async def main(room_url, token=None):
)
tts_service = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)

View File

@@ -1,13 +0,0 @@
# studypal
### Have a conversation about any article on the web
studypal is a fast conversational AI built using [Daily](https://www.daily.co/) for real-time media transport and [Cartesia](https://cartesia.ai) for text-to-speech. Everything is orchestrated together (VAD -> STT -> LLM -> TTS) using [Pipecat](https://www.pipecat.ai/).
## Setup
1. Clone the repository
2. Copy `env.example` to a `.env` file and add API keys
3. Install the required packages: `pip install -r requirements.txt`
4. Run `python3 studypal.py` from your command line.
5. While the app is running, go to the `https://<yourdomain>.daily.co/<room_url>` set in `DAILY_SAMPLE_ROOM_URL` and talk to studypal!

View File

@@ -1,5 +0,0 @@
DAILY_SAMPLE_ROOM_URL= # Follow instructions here and put your https://YOURDOMAIN.daily.co/YOURROOM (Instructions: https://docs.pipecat.ai/quickstart#preparing-your-environment)
DAILY_API_KEY= # Create here: https://dashboard.daily.co/developers
OPENAI_API_KEY= # Create here: https://platform.openai.com/docs/overview
CARTESIA_API_KEY= # Create here: https://play.cartesia.ai/console
CARTESIA_VOICE_ID= # Find here: https://play.cartesia.ai/

View File

@@ -1,5 +0,0 @@
beautifulsoup4==4.12.3
pypdf==4.3.1
tiktoken==0.7.0
pipecat-ai[daily,cartesia,openai,silero]==0.0.40
python-dotenv==1.0.1

View File

@@ -1,61 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import argparse
import os
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
async def configure(aiohttp_session: aiohttp.ClientSession):
(url, token, _) = await configure_with_args(aiohttp_session)
return (url, token)
async def configure_with_args(
aiohttp_session: aiohttp.ClientSession,
parser: argparse.ArgumentParser | None = None):
if not parser:
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, args)

View File

@@ -1,170 +0,0 @@
import aiohttp
import asyncio
import os
import sys
import io
from bs4 import BeautifulSoup
from pypdf import PdfReader
import tiktoken
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.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)
# Run this script directly from your command line.
# This project was adapted from
# https://github.com/pipecat-ai/pipecat/blob/main/examples/foundational/07d-interruptible-cartesia.py
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
# Count number of tokens used in model and truncate the content
def truncate_content(content, model_name):
encoding = tiktoken.encoding_for_model(model_name)
tokens = encoding.encode(content)
max_tokens = 10000
if len(tokens) > max_tokens:
truncated_tokens = tokens[:max_tokens]
return encoding.decode(truncated_tokens)
return content
# Main function to extract content from url
async def get_article_content(url: str, aiohttp_session: aiohttp.ClientSession):
if 'arxiv.org' in url:
return await get_arxiv_content(url, aiohttp_session)
else:
return await get_wikipedia_content(url, aiohttp_session)
# Helper function to extract content from Wikipedia url (this is
# technically agnostic to URL type but will work best with Wikipedia
# articles)
async def get_wikipedia_content(url: str, aiohttp_session: aiohttp.ClientSession):
async with aiohttp_session.get(url) as response:
if response.status != 200:
return "Failed to download Wikipedia article."
text = await response.text()
soup = BeautifulSoup(text, 'html.parser')
content = soup.find('div', {'class': 'mw-parser-output'})
if content:
return content.get_text()
else:
return "Failed to extract Wikipedia article content."
# Helper function to extract content from arXiv url
async def get_arxiv_content(url: str, aiohttp_session: aiohttp.ClientSession):
if '/abs/' in url:
url = url.replace('/abs/', '/pdf/')
if not url.endswith('.pdf'):
url += '.pdf'
async with aiohttp_session.get(url) as response:
if response.status != 200:
return "Failed to download arXiv PDF."
content = await response.read()
pdf_file = io.BytesIO(content)
pdf_reader = PdfReader(pdf_file)
text = ""
for page in pdf_reader.pages:
text += page.extract_text()
return text
# This is the main function that handles STT -> LLM -> TTS
async def main():
url = input("Enter the URL of the article you would like to talk about: ")
async with aiohttp.ClientSession() as session:
article_content = await get_article_content(url, session)
article_content = truncate_content(article_content, model_name="gpt-4o-mini")
(room_url, token) = await configure(session)
transport = DailyTransport(
room_url,
token,
"studypal",
DailyParams(
audio_out_sample_rate=44100,
audio_out_enabled=True,
transcription_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer()
)
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id=os.getenv("CARTESIA_VOICE_ID", "4d2fd738-3b3d-4368-957a-bb4805275bd9"),
# British Narration Lady: 4d2fd738-3b3d-4368-957a-bb4805275bd9
sample_rate=44100,
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o-mini")
messages = [{
"role": "system", "content": f"""You are an AI study partner. You have been given the following article content:
{article_content}
Your task is to help the user understand and learn from this article in 2 sentences. THESE RESPONSES SHOULD BE ONLY MAX 2 SENTENCES. THIS INSTRUCTION IS VERY IMPORTANT. RESPONSES SHOULDN'T BE LONG.
""", }, ]
tma_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages)
pipeline = Pipeline([
transport.input(),
tma_in,
llm,
tts,
transport.output(),
tma_out,
])
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"])
messages.append(
{
"role": "system",
"content": "Hello! I'm ready to discuss the article with you. What would you like to learn about?"})
await task.queue_frames([LLMMessagesFrame(messages)])
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":
asyncio.run(main())

View File

@@ -10,9 +10,9 @@ 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.elevenlabs import ElevenLabsTTSService
from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketTransport, FastAPIWebsocketParams
from pipecat.vad.silero import SileroVADAnalyzer
from pipecat.serializers.twilio import TwilioFrameSerializer
@@ -46,9 +46,10 @@ async def run_bot(websocket_client, stream_sid):
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
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
messages = [

View File

@@ -17,8 +17,8 @@ from pipecat.processors.aggregators.llm_response import (
LLMAssistantResponseAggregator,
LLMUserResponseAggregator
)
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.network.websocket_server import WebsocketServerParams, WebsocketServerTransport
from pipecat.vad.silero import SileroVADAnalyzer
@@ -50,9 +50,10 @@ async def main():
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
tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
messages = [

View File

@@ -0,0 +1,502 @@
#
# This file is autogenerated by pip-compile with Python 3.10
# by the following command:
#
# pip-compile --all-extras pyproject.toml
#
aiofiles==24.1.0
# via deepgram-sdk
aiohttp==3.9.5
# via
# deepgram-sdk
# langchain
# langchain-community
# pipecat-ai (pyproject.toml)
aiosignal==1.3.1
# via aiohttp
annotated-types==0.7.0
# via pydantic
anthropic==0.28.1
# via
# openpipe
# pipecat-ai (pyproject.toml)
anyio==4.4.0
# via
# anthropic
# httpx
# openai
# starlette
# watchfiles
async-timeout==4.0.3
# via
# aiohttp
# langchain
attrs==23.2.0
# via
# aiohttp
# openpipe
av==12.3.0
# via faster-whisper
azure-cognitiveservices-speech==1.38.0
# via pipecat-ai (pyproject.toml)
blinker==1.8.2
# via flask
cachetools==5.4.0
# via google-auth
certifi==2024.7.4
# via
# httpcore
# httpx
# requests
charset-normalizer==3.3.2
# via requests
click==8.1.7
# via
# flask
# typer
# uvicorn
coloredlogs==15.0.1
# via onnxruntime
ctranslate2==4.3.1
# via faster-whisper
daily-python==0.10.1
# via pipecat-ai (pyproject.toml)
dataclasses-json==0.6.7
# via
# deepgram-sdk
# langchain-community
deepgram-sdk==3.2.7
# via pipecat-ai (pyproject.toml)
distro==1.9.0
# via
# anthropic
# openai
dnspython==2.6.1
# via email-validator
einops==0.8.0
# via pipecat-ai (pyproject.toml)
email-validator==2.2.0
# via fastapi
exceptiongroup==1.2.2
# via anyio
fal-client==0.4.1
# via pipecat-ai (pyproject.toml)
fastapi==0.111.1
# via pipecat-ai (pyproject.toml)
fastapi-cli==0.0.5
# via fastapi
faster-whisper==1.0.3
# via pipecat-ai (pyproject.toml)
filelock==3.15.4
# via
# huggingface-hub
# pyht
# torch
# transformers
# triton
flask==3.0.3
# via
# flask-cors
# pipecat-ai (pyproject.toml)
flask-cors==4.0.1
# via pipecat-ai (pyproject.toml)
flatbuffers==24.3.25
# via onnxruntime
frozenlist==1.4.1
# via
# aiohttp
# aiosignal
fsspec==2024.6.1
# via
# huggingface-hub
# torch
future==1.0.0
# via pyloudnorm
google-ai-generativelanguage==0.6.6
# via google-generativeai
google-api-core[grpc]==2.19.1
# via
# google-ai-generativelanguage
# google-api-python-client
# google-generativeai
google-api-python-client==2.140.0
# via google-generativeai
google-auth==2.33.0
# via
# google-ai-generativelanguage
# google-api-core
# google-api-python-client
# google-auth-httplib2
# google-generativeai
google-auth-httplib2==0.2.0
# via google-api-python-client
google-generativeai==0.7.2
# via pipecat-ai (pyproject.toml)
googleapis-common-protos==1.63.2
# via
# google-api-core
# grpcio-status
greenlet==3.0.3
# via sqlalchemy
grpcio==1.65.4
# via
# google-api-core
# grpcio-status
# pyht
grpcio-status==1.62.3
# via google-api-core
h11==0.14.0
# via
# httpcore
# uvicorn
httpcore==1.0.5
# via httpx
httplib2==0.22.0
# via
# google-api-python-client
# google-auth-httplib2
httptools==0.6.1
# via uvicorn
httpx==0.27.0
# via
# anthropic
# deepgram-sdk
# fal-client
# fastapi
# openai
# openpipe
httpx-sse==0.4.0
# via fal-client
huggingface-hub==0.24.5
# via
# faster-whisper
# timm
# tokenizers
# transformers
humanfriendly==10.0
# via coloredlogs
idna==3.7
# via
# anyio
# email-validator
# httpx
# requests
# yarl
itsdangerous==2.2.0
# via flask
jinja2==3.1.4
# via
# fastapi
# flask
# torch
jiter==0.5.0
# via anthropic
jsonpatch==1.33
# via langchain-core
jsonpointer==3.0.0
# via jsonpatch
langchain==0.2.12
# via
# langchain-community
# pipecat-ai (pyproject.toml)
langchain-community==0.2.11
# via pipecat-ai (pyproject.toml)
langchain-core==0.2.29
# via
# langchain
# langchain-community
# langchain-openai
# langchain-text-splitters
langchain-openai==0.1.20
# via pipecat-ai (pyproject.toml)
langchain-text-splitters==0.2.2
# via langchain
langsmith==0.1.98
# via
# langchain
# langchain-community
# langchain-core
llvmlite==0.43.0
# via numba
loguru==0.7.2
# via pipecat-ai (pyproject.toml)
markdown-it-py==3.0.0
# via rich
markupsafe==2.1.5
# via
# jinja2
# werkzeug
marshmallow==3.21.3
# via dataclasses-json
mdurl==0.1.2
# via markdown-it-py
mpmath==1.3.0
# via sympy
multidict==6.0.5
# via
# aiohttp
# yarl
mypy-extensions==1.0.0
# via typing-inspect
networkx==3.3
# via torch
numba==0.60.0
# via resampy
numpy==1.26.4
# via
# ctranslate2
# langchain
# langchain-community
# numba
# onnxruntime
# pipecat-ai (pyproject.toml)
# pyloudnorm
# resampy
# scipy
# torchvision
# transformers
nvidia-cublas-cu12==12.1.3.1
# via
# nvidia-cudnn-cu12
# nvidia-cusolver-cu12
# torch
nvidia-cuda-cupti-cu12==12.1.105
# via torch
nvidia-cuda-nvrtc-cu12==12.1.105
# via torch
nvidia-cuda-runtime-cu12==12.1.105
# via torch
nvidia-cudnn-cu12==9.1.0.70
# via torch
nvidia-cufft-cu12==11.0.2.54
# via torch
nvidia-curand-cu12==10.3.2.106
# via torch
nvidia-cusolver-cu12==11.4.5.107
# via torch
nvidia-cusparse-cu12==12.1.0.106
# via
# nvidia-cusolver-cu12
# torch
nvidia-nccl-cu12==2.20.5
# via torch
nvidia-nvjitlink-cu12==12.6.20
# via
# nvidia-cusolver-cu12
# nvidia-cusparse-cu12
nvidia-nvtx-cu12==12.1.105
# via torch
onnxruntime==1.18.1
# via
# faster-whisper
# silero-vad
openai==1.35.15
# via
# langchain-openai
# openpipe
# pipecat-ai (pyproject.toml)
openpipe==4.18.0
# via pipecat-ai (pyproject.toml)
orjson==3.10.7
# via langsmith
packaging==24.1
# via
# huggingface-hub
# langchain-core
# marshmallow
# onnxruntime
# transformers
pillow==10.3.0
# via
# pipecat-ai (pyproject.toml)
# torchvision
proto-plus==1.24.0
# via
# google-ai-generativelanguage
# google-api-core
protobuf==4.25.4
# via
# google-ai-generativelanguage
# google-api-core
# google-generativeai
# googleapis-common-protos
# grpcio-status
# onnxruntime
# pipecat-ai (pyproject.toml)
# proto-plus
# pyht
pyasn1==0.6.0
# via
# pyasn1-modules
# rsa
pyasn1-modules==0.4.0
# via google-auth
pyaudio==0.2.14
# via pipecat-ai (pyproject.toml)
pydantic==2.8.2
# via
# anthropic
# fastapi
# google-generativeai
# langchain
# langchain-core
# langsmith
# openai
pydantic-core==2.20.1
# via pydantic
pygments==2.18.0
# via rich
pyht==0.0.28
# via pipecat-ai (pyproject.toml)
pyloudnorm==0.1.1
# via pipecat-ai (pyproject.toml)
pyparsing==3.1.2
# via httplib2
python-dateutil==2.9.0.post0
# via openpipe
python-dotenv==1.0.1
# via
# pipecat-ai (pyproject.toml)
# uvicorn
python-multipart==0.0.9
# via fastapi
pyyaml==6.0.2
# via
# ctranslate2
# huggingface-hub
# langchain
# langchain-community
# langchain-core
# timm
# transformers
# uvicorn
regex==2024.7.24
# via
# tiktoken
# transformers
requests==2.32.3
# via
# google-api-core
# huggingface-hub
# langchain
# langchain-community
# langsmith
# pyht
# tiktoken
# transformers
resampy==0.4.3
# via pipecat-ai (pyproject.toml)
rich==13.7.1
# via typer
rsa==4.9
# via google-auth
safetensors==0.4.4
# via
# timm
# transformers
scipy==1.14.0
# via pyloudnorm
shellingham==1.5.4
# via typer
silero-vad==5.1
# via pipecat-ai (pyproject.toml)
six==1.16.0
# via python-dateutil
sniffio==1.3.1
# via
# anthropic
# anyio
# httpx
# openai
sqlalchemy==2.0.32
# via
# langchain
# langchain-community
starlette==0.37.2
# via fastapi
sympy==1.13.1
# via
# onnxruntime
# torch
tenacity==8.5.0
# via
# langchain
# langchain-community
# langchain-core
tiktoken==0.7.0
# via langchain-openai
timm==0.9.16
# via pipecat-ai (pyproject.toml)
tokenizers==0.19.1
# via
# anthropic
# faster-whisper
# transformers
torch==2.4.0
# via
# silero-vad
# timm
# torchaudio
# torchvision
torchaudio==2.4.0
# via silero-vad
torchvision==0.19.0
# via timm
tqdm==4.66.5
# via
# google-generativeai
# huggingface-hub
# openai
# transformers
transformers==4.40.2
# via pipecat-ai (pyproject.toml)
triton==3.0.0
# via torch
typer==0.12.3
# via fastapi-cli
typing-extensions==4.12.2
# via
# anthropic
# anyio
# deepgram-sdk
# fastapi
# google-generativeai
# huggingface-hub
# langchain-core
# openai
# pipecat-ai (pyproject.toml)
# pydantic
# pydantic-core
# sqlalchemy
# torch
# typer
# typing-inspect
# uvicorn
typing-inspect==0.9.0
# via dataclasses-json
uritemplate==4.1.1
# via google-api-python-client
urllib3==2.2.2
# via requests
uvicorn[standard]==0.30.5
# via
# fastapi
# fastapi-cli
uvloop==0.19.0
# via uvicorn
verboselogs==1.7
# via deepgram-sdk
watchfiles==0.23.0
# via uvicorn
websockets==12.0
# via
# deepgram-sdk
# pipecat-ai (pyproject.toml)
# uvicorn
werkzeug==3.0.3
# via flask
yarl==1.9.4
# via aiohttp
# The following packages are considered to be unsafe in a requirements file:
# setuptools

View File

@@ -0,0 +1,466 @@
#
# This file is autogenerated by pip-compile with Python 3.10
# by the following command:
#
# pip-compile --all-extras pyproject.toml
#
aiofiles==24.1.0
# via deepgram-sdk
aiohttp==3.9.5
# via
# deepgram-sdk
# langchain
# langchain-community
# pipecat-ai (pyproject.toml)
aiosignal==1.3.1
# via aiohttp
annotated-types==0.7.0
# via pydantic
anthropic==0.28.1
# via
# openpipe
# pipecat-ai (pyproject.toml)
anyio==4.4.0
# via
# anthropic
# httpx
# openai
# starlette
# watchfiles
async-timeout==4.0.3
# via
# aiohttp
# langchain
attrs==23.2.0
# via
# aiohttp
# openpipe
av==12.3.0
# via faster-whisper
azure-cognitiveservices-speech==1.38.0
# via pipecat-ai (pyproject.toml)
blinker==1.8.2
# via flask
cachetools==5.4.0
# via google-auth
certifi==2024.7.4
# via
# httpcore
# httpx
# requests
charset-normalizer==3.3.2
# via requests
click==8.1.7
# via
# flask
# typer
# uvicorn
coloredlogs==15.0.1
# via onnxruntime
ctranslate2==4.3.1
# via faster-whisper
daily-python==0.10.1
# via pipecat-ai (pyproject.toml)
dataclasses-json==0.6.7
# via
# deepgram-sdk
# langchain-community
deepgram-sdk==3.2.7
# via pipecat-ai (pyproject.toml)
distro==1.9.0
# via
# anthropic
# openai
dnspython==2.6.1
# via email-validator
einops==0.8.0
# via pipecat-ai (pyproject.toml)
email-validator==2.2.0
# via fastapi
exceptiongroup==1.2.2
# via anyio
fal-client==0.4.1
# via pipecat-ai (pyproject.toml)
fastapi==0.111.1
# via pipecat-ai (pyproject.toml)
fastapi-cli==0.0.5
# via fastapi
faster-whisper==1.0.3
# via pipecat-ai (pyproject.toml)
filelock==3.15.4
# via
# huggingface-hub
# pyht
# torch
# transformers
flask==3.0.3
# via
# flask-cors
# pipecat-ai (pyproject.toml)
flask-cors==4.0.1
# via pipecat-ai (pyproject.toml)
flatbuffers==24.3.25
# via onnxruntime
frozenlist==1.4.1
# via
# aiohttp
# aiosignal
fsspec==2024.6.1
# via
# huggingface-hub
# torch
future==1.0.0
# via pyloudnorm
google-ai-generativelanguage==0.6.6
# via google-generativeai
google-api-core[grpc]==2.19.1
# via
# google-ai-generativelanguage
# google-api-python-client
# google-generativeai
google-api-python-client==2.140.0
# via google-generativeai
google-auth==2.33.0
# via
# google-ai-generativelanguage
# google-api-core
# google-api-python-client
# google-auth-httplib2
# google-generativeai
google-auth-httplib2==0.2.0
# via google-api-python-client
google-generativeai==0.7.2
# via pipecat-ai (pyproject.toml)
googleapis-common-protos==1.63.2
# via
# google-api-core
# grpcio-status
grpcio==1.65.4
# via
# google-api-core
# grpcio-status
# pyht
grpcio-status==1.62.3
# via google-api-core
h11==0.14.0
# via
# httpcore
# uvicorn
httpcore==1.0.5
# via httpx
httplib2==0.22.0
# via
# google-api-python-client
# google-auth-httplib2
httptools==0.6.1
# via uvicorn
httpx==0.27.0
# via
# anthropic
# deepgram-sdk
# fal-client
# fastapi
# openai
# openpipe
httpx-sse==0.4.0
# via fal-client
huggingface-hub==0.24.5
# via
# faster-whisper
# timm
# tokenizers
# transformers
humanfriendly==10.0
# via coloredlogs
idna==3.7
# via
# anyio
# email-validator
# httpx
# requests
# yarl
itsdangerous==2.2.0
# via flask
jinja2==3.1.4
# via
# fastapi
# flask
# torch
jiter==0.5.0
# via anthropic
jsonpatch==1.33
# via langchain-core
jsonpointer==3.0.0
# via jsonpatch
langchain==0.2.12
# via
# langchain-community
# pipecat-ai (pyproject.toml)
langchain-community==0.2.11
# via pipecat-ai (pyproject.toml)
langchain-core==0.2.29
# via
# langchain
# langchain-community
# langchain-openai
# langchain-text-splitters
langchain-openai==0.1.20
# via pipecat-ai (pyproject.toml)
langchain-text-splitters==0.2.2
# via langchain
langsmith==0.1.98
# via
# langchain
# langchain-community
# langchain-core
llvmlite==0.43.0
# via numba
loguru==0.7.2
# via pipecat-ai (pyproject.toml)
markdown-it-py==3.0.0
# via rich
markupsafe==2.1.5
# via
# jinja2
# werkzeug
marshmallow==3.21.3
# via dataclasses-json
mdurl==0.1.2
# via markdown-it-py
mpmath==1.3.0
# via sympy
multidict==6.0.5
# via
# aiohttp
# yarl
mypy-extensions==1.0.0
# via typing-inspect
networkx==3.3
# via torch
numba==0.60.0
# via resampy
numpy==1.26.4
# via
# ctranslate2
# langchain
# langchain-community
# numba
# onnxruntime
# pipecat-ai (pyproject.toml)
# pyloudnorm
# resampy
# scipy
# torchvision
# transformers
onnxruntime==1.18.1
# via
# faster-whisper
# silero-vad
openai==1.35.15
# via
# langchain-openai
# openpipe
# pipecat-ai (pyproject.toml)
openpipe==4.18.0
# via pipecat-ai (pyproject.toml)
orjson==3.10.7
# via langsmith
packaging==24.1
# via
# huggingface-hub
# langchain-core
# marshmallow
# onnxruntime
# transformers
pillow==10.3.0
# via
# pipecat-ai (pyproject.toml)
# torchvision
proto-plus==1.24.0
# via
# google-ai-generativelanguage
# google-api-core
protobuf==4.25.4
# via
# google-ai-generativelanguage
# google-api-core
# google-generativeai
# googleapis-common-protos
# grpcio-status
# onnxruntime
# pipecat-ai (pyproject.toml)
# proto-plus
# pyht
pyasn1==0.6.0
# via
# pyasn1-modules
# rsa
pyasn1-modules==0.4.0
# via google-auth
pyaudio==0.2.14
# via pipecat-ai (pyproject.toml)
pydantic==2.8.2
# via
# anthropic
# fastapi
# google-generativeai
# langchain
# langchain-core
# langsmith
# openai
pydantic-core==2.20.1
# via pydantic
pygments==2.18.0
# via rich
pyht==0.0.28
# via pipecat-ai (pyproject.toml)
pyloudnorm==0.1.1
# via pipecat-ai (pyproject.toml)
pyparsing==3.1.2
# via httplib2
python-dateutil==2.9.0.post0
# via openpipe
python-dotenv==1.0.1
# via
# pipecat-ai (pyproject.toml)
# uvicorn
python-multipart==0.0.9
# via fastapi
pyyaml==6.0.2
# via
# ctranslate2
# huggingface-hub
# langchain
# langchain-community
# langchain-core
# timm
# transformers
# uvicorn
regex==2024.7.24
# via
# tiktoken
# transformers
requests==2.32.3
# via
# google-api-core
# huggingface-hub
# langchain
# langchain-community
# langsmith
# pyht
# tiktoken
# transformers
resampy==0.4.3
# via pipecat-ai (pyproject.toml)
rich==13.7.1
# via typer
rsa==4.9
# via google-auth
safetensors==0.4.4
# via
# timm
# transformers
scipy==1.14.0
# via pyloudnorm
shellingham==1.5.4
# via typer
silero-vad==5.1
# via pipecat-ai (pyproject.toml)
six==1.16.0
# via python-dateutil
sniffio==1.3.1
# via
# anthropic
# anyio
# httpx
# openai
sqlalchemy==2.0.32
# via
# langchain
# langchain-community
starlette==0.37.2
# via fastapi
sympy==1.13.1
# via
# onnxruntime
# torch
tenacity==8.5.0
# via
# langchain
# langchain-community
# langchain-core
tiktoken==0.7.0
# via langchain-openai
timm==0.9.16
# via pipecat-ai (pyproject.toml)
tokenizers==0.19.1
# via
# anthropic
# faster-whisper
# transformers
torch==2.4.0
# via
# silero-vad
# timm
# torchaudio
# torchvision
torchaudio==2.4.0
# via silero-vad
torchvision==0.19.0
# via timm
tqdm==4.66.5
# via
# google-generativeai
# huggingface-hub
# openai
# transformers
transformers==4.40.2
# via pipecat-ai (pyproject.toml)
typer==0.12.3
# via fastapi-cli
typing-extensions==4.12.2
# via
# anthropic
# anyio
# deepgram-sdk
# fastapi
# google-generativeai
# huggingface-hub
# langchain-core
# openai
# pipecat-ai (pyproject.toml)
# pydantic
# pydantic-core
# sqlalchemy
# torch
# typer
# typing-inspect
# uvicorn
typing-inspect==0.9.0
# via dataclasses-json
uritemplate==4.1.1
# via google-api-python-client
urllib3==2.2.2
# via requests
uvicorn[standard]==0.30.5
# via
# fastapi
# fastapi-cli
uvloop==0.19.0
# via uvicorn
verboselogs==1.7
# via deepgram-sdk
watchfiles==0.23.0
# via uvicorn
websockets==12.0
# via
# deepgram-sdk
# pipecat-ai (pyproject.toml)
# uvicorn
werkzeug==3.0.3
# via flask
yarl==1.9.4
# via aiohttp
# The following packages are considered to be unsafe in a requirements file:
# setuptools

View File

@@ -20,13 +20,13 @@ classifiers = [
"Topic :: Scientific/Engineering :: Artificial Intelligence"
]
dependencies = [
"aiohttp~=3.10.3",
"aiohttp~=3.9.5",
"numpy~=1.26.4",
"loguru~=0.7.2",
"Pillow~=10.4.0",
"protobuf~=4.25.4",
"pydantic~=2.8.2",
"loguru~=0.7.0",
"Pillow~=10.3.0",
"protobuf~=4.25.3",
"pyloudnorm~=0.1.1",
"typing-extensions~=4.12.1",
]
[project.urls]
@@ -34,29 +34,24 @@ Source = "https://github.com/pipecat-ai/pipecat"
Website = "https://pipecat.ai"
[project.optional-dependencies]
anthropic = [ "anthropic~=0.34.0" ]
azure = [ "azure-cognitiveservices-speech~=1.40.0" ]
anthropic = [ "anthropic~=0.28.1" ]
azure = [ "azure-cognitiveservices-speech~=1.38.0" ]
cartesia = [ "websockets~=12.0" ]
daily = [ "daily-python~=0.10.1" ]
deepgram = [ "deepgram-sdk~=3.5.0" ]
elevenlabs = [ "websockets~=12.0" ]
examples = [ "python-dotenv~=1.0.1", "flask~=3.0.3", "flask_cors~=4.0.1" ]
deepgram = [ "deepgram-sdk~=3.2.7" ]
examples = [ "python-dotenv~=1.0.0", "flask~=3.0.3", "flask_cors~=4.0.1" ]
fal = [ "fal-client~=0.4.1" ]
gladia = [ "websockets~=12.0" ]
google = [ "google-generativeai~=0.7.2" ]
gstreamer = [ "pygobject~=3.48.2" ]
fireworks = [ "openai~=1.37.2" ]
langchain = [ "langchain~=0.2.14", "langchain-community~=0.2.12", "langchain-openai~=0.1.20" ]
livekit = [ "livekit~=0.13.1" ]
lmnt = [ "lmnt~=1.1.4" ]
local = [ "pyaudio~=0.2.14" ]
moondream = [ "einops~=0.8.0", "timm~=1.0.8", "transformers~=4.44.0" ]
openai = [ "openai~=1.37.2" ]
openpipe = [ "openpipe~=4.24.0" ]
google = [ "google-generativeai~=0.7.1" ]
fireworks = [ "openai~=1.35.0" ]
langchain = [ "langchain~=0.2.10", "langchain-community~=0.2.9", "langchain-openai~=0.1.17" ]
local = [ "pyaudio~=0.2.0" ]
moondream = [ "einops~=0.8.0", "timm~=0.9.16", "transformers~=4.40.2" ]
openai = [ "openai~=1.35.0" ]
openpipe = [ "openpipe~=4.18.0" ]
playht = [ "pyht~=0.0.28" ]
silero = [ "silero-vad~=5.1" ]
together = [ "together~=1.2.7" ]
websocket = [ "websockets~=12.0", "fastapi~=0.112.1" ]
websocket = [ "websockets~=12.0", "fastapi~=0.111.0" ]
whisper = [ "faster-whisper~=1.0.3" ]
xtts = [ "resampy~=0.4.3" ]

View File

@@ -1,18 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from abc import ABC, abstractmethod
class BaseClock(ABC):
@abstractmethod
def get_time(self) -> int:
pass
@abstractmethod
def start(self):
pass

View File

@@ -1,21 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import time
from pipecat.clocks.base_clock import BaseClock
class SystemClock(BaseClock):
def __init__(self):
self._time = 0
def get_time(self) -> int:
return time.monotonic_ns() - self._time if self._time > 0 else 0
def start(self):
self._time = time.monotonic_ns()

View File

@@ -4,31 +4,21 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from typing import Any, List, Mapping, Optional, Tuple
from typing import Any, List, Mapping, Tuple
from dataclasses import dataclass, field
from pipecat.clocks.base_clock import BaseClock
from pipecat.transcriptions.language import Language
from pipecat.utils.time import nanoseconds_to_str
from pipecat.utils.utils import obj_count, obj_id
from pipecat.vad.vad_analyzer import VADParams
def format_pts(pts: int | None):
return nanoseconds_to_str(pts) if pts else None
@dataclass
class Frame:
id: int = field(init=False)
name: str = field(init=False)
pts: Optional[int] = field(init=False)
def __post_init__(self):
self.id: int = obj_id()
self.name: str = f"{self.__class__.__name__}#{obj_count(self)}"
self.pts: Optional[int] = None
def __str__(self):
return self.name
@@ -54,8 +44,7 @@ class AudioRawFrame(DataFrame):
self.num_frames = int(len(self.audio) / (self.num_channels * 2))
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, size: {len(self.audio)}, frames: {self.num_frames}, sample_rate: {self.sample_rate}, channels: {self.num_channels})"
return f"{self.name}(size: {len(self.audio)}, frames: {self.num_frames}, sample_rate: {self.sample_rate}, channels: {self.num_channels})"
@dataclass
@@ -69,8 +58,7 @@ class ImageRawFrame(DataFrame):
format: str | None
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, size: {self.size}, format: {self.format})"
return f"{self.name}(size: {self.size}, format: {self.format})"
@dataclass
@@ -82,8 +70,7 @@ class URLImageRawFrame(ImageRawFrame):
url: str | None
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, url: {self.url}, size: {self.size}, format: {self.format})"
return f"{self.name}(url: {self.url}, size: {self.size}, format: {self.format})"
@dataclass
@@ -95,8 +82,7 @@ class VisionImageRawFrame(ImageRawFrame):
text: str | None
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, text: {self.text}, size: {self.size}, format: {self.format})"
return f"{self.name}(text: {self.text}, size: {self.size}, format: {self.format})"
@dataclass
@@ -108,8 +94,7 @@ class UserImageRawFrame(ImageRawFrame):
user_id: str
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format})"
return f"{self.name}(user: {self.user_id}, size: {self.size}, format: {self.format})"
@dataclass
@@ -122,8 +107,7 @@ class SpriteFrame(Frame):
images: List[ImageRawFrame]
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, size: {len(self.images)})"
return f"{self.name}(size: {len(self.images)})"
@dataclass
@@ -135,8 +119,7 @@ class TextFrame(DataFrame):
text: str
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, text: {self.text})"
return f"{self.name}(text: {self.text})"
@dataclass
@@ -147,10 +130,9 @@ class TranscriptionFrame(TextFrame):
"""
user_id: str
timestamp: str
language: Language | None = None
def __str__(self):
return f"{self.name}(user: {self.user_id}, text: {self.text}, language: {self.language}, timestamp: {self.timestamp})"
return f"{self.name}(user: {self.user_id}, text: {self.text}, timestamp: {self.timestamp})"
@dataclass
@@ -159,10 +141,9 @@ class InterimTranscriptionFrame(TextFrame):
the transport's receive queue when a participant speaks."""
user_id: str
timestamp: str
language: Language | None = None
def __str__(self):
return f"{self.name}(user: {self.user_id}, text: {self.text}, language: {self.language}, timestamp: {self.timestamp})"
return f"{self.name}(user: {self.user_id}, text: {self.text}, timestamp: {self.timestamp})"
@dataclass
@@ -196,22 +177,6 @@ class LLMMessagesUpdateFrame(DataFrame):
messages: List[dict]
@dataclass
class LLMSetToolsFrame(DataFrame):
"""A frame containing a list of tools for an LLM to use for function calling.
The specific format depends on the LLM being used, but it should typically
contain JSON Schema objects.
"""
tools: List[dict]
@dataclass
class LLMEnablePromptCachingFrame(DataFrame):
"""A frame to enable/disable prompt caching in certain LLMs.
"""
enable: bool
@dataclass
class TTSSpeakFrame(DataFrame):
"""A frame that contains a text that should be spoken by the TTS in the
@@ -224,7 +189,6 @@ class TTSSpeakFrame(DataFrame):
@dataclass
class TransportMessageFrame(DataFrame):
message: Any
urgent: bool = False
def __str__(self):
return f"{self.name}(message: {self.message})"
@@ -257,31 +221,18 @@ class CancelFrame(SystemFrame):
@dataclass
class ErrorFrame(SystemFrame):
"""This is used notify upstream that an error has occurred downstream the
pipeline. A fatal error indicates the error is unrecoverable and that the
bot should exit.
"""
error: str
fatal: bool = False
pipeline."""
error: str | None
def __str__(self):
return f"{self.name}(error: {self.error}, fatal: {self.fatal})"
@dataclass
class FatalErrorFrame(ErrorFrame):
"""This is used notify upstream that an unrecoverable error has occurred and
that the bot should exit.
"""
fatal: bool = field(default=True, init=False)
return f"{self.name}(error: {self.error})"
@dataclass
class StopTaskFrame(SystemFrame):
"""Indicates that a pipeline task should be stopped but that the pipeline
processors should be kept in a running state. This is normally queued from
the pipeline task.
"""Indicates that a pipeline task should be stopped. This should inform the
pipeline processors that they should stop pushing frames but that they
should be kept in a running state.
"""
pass
@@ -341,7 +292,6 @@ class ControlFrame(Frame):
@dataclass
class StartFrame(ControlFrame):
"""This is the first frame that should be pushed down a pipeline."""
clock: BaseClock
allow_interruptions: bool = False
enable_metrics: bool = False
enable_usage_metrics: bool = False
@@ -439,7 +389,6 @@ class TTSStoppedFrame(ControlFrame):
class UserImageRequestFrame(ControlFrame):
"""A frame user to request an image from the given user."""
user_id: str
context: Optional[Any] = None
def __str__(self):
return f"{self.name}, user: {self.user_id}"
@@ -452,67 +401,8 @@ class LLMModelUpdateFrame(ControlFrame):
model: str
@dataclass
class TTSModelUpdateFrame(ControlFrame):
"""A control frame containing a request to update the TTS model.
"""
model: str
@dataclass
class TTSVoiceUpdateFrame(ControlFrame):
"""A control frame containing a request to update to a new TTS voice.
"""
voice: str
@dataclass
class TTSLanguageUpdateFrame(ControlFrame):
"""A control frame containing a request to update to a new TTS language and
optional voice.
"""
language: Language
@dataclass
class STTModelUpdateFrame(ControlFrame):
"""A control frame containing a request to update the STT model and optional
language.
"""
model: str
@dataclass
class STTLanguageUpdateFrame(ControlFrame):
"""A control frame containing a request to update to STT language.
"""
language: Language
@dataclass
class FunctionCallInProgressFrame(SystemFrame):
"""A frame signaling that a function call is in progress.
"""
function_name: str
tool_call_id: str
arguments: str
@dataclass
class FunctionCallResultFrame(DataFrame):
"""A frame containing the result of an LLM function (tool) call.
"""
function_name: str
tool_call_id: str
arguments: str
result: Any
@dataclass
class VADParamsUpdateFrame(ControlFrame):
"""A control frame containing a request to update VAD params. Intended
to be pushed upstream from RTVI processor.
"""
params: VADParams

View File

@@ -57,7 +57,6 @@ class ParallelTask(BasePipeline):
raise Exception(f"ParallelTask needs at least one argument")
self._sinks = []
self._sources = []
self._pipelines = []
self._up_queue = asyncio.Queue()
@@ -71,10 +70,10 @@ class ParallelTask(BasePipeline):
# We add a source at the beginning of the pipeline and a sink at the end.
source = Source(self._up_queue)
sink = Sink(self._down_queue)
processors: List[FrameProcessor] = [source] + processors + [sink]
processors: List[FrameProcessor] = [source] + processors
processors.append(sink)
# Keep track of sources and sinks.
self._sources.append(source)
# Keep track of sinks. We access the source through the pipeline.
self._sinks.append(sink)
# Create pipeline
@@ -100,8 +99,8 @@ class ParallelTask(BasePipeline):
# If we get an upstream frame we process it in each sink.
await asyncio.gather(*[s.process_frame(frame, direction) for s in self._sinks])
elif direction == FrameDirection.DOWNSTREAM:
# If we get a downstream frame we process it in each source.
await asyncio.gather(*[s.process_frame(frame, direction) for s in self._sources])
# If we get a downstream frame we process it in each source (using the pipeline).
await asyncio.gather(*[p.process_frame(frame, direction) for p in self._pipelines])
seen_ids = set()
while not self._up_queue.empty():

View File

@@ -10,16 +10,7 @@ from typing import AsyncIterable, Iterable
from pydantic import BaseModel
from pipecat.clocks.base_clock import BaseClock
from pipecat.clocks.system_clock import SystemClock
from pipecat.frames.frames import (
CancelFrame,
EndFrame,
ErrorFrame,
Frame,
MetricsFrame,
StartFrame,
StopTaskFrame)
from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame, Frame, MetricsFrame, StartFrame, StopTaskFrame
from pipecat.pipeline.base_pipeline import BasePipeline
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.utils import obj_count, obj_id
@@ -46,32 +37,18 @@ class Source(FrameProcessor):
match direction:
case FrameDirection.UPSTREAM:
await self._handle_upstream_frame(frame)
await self._up_queue.put(frame)
case FrameDirection.DOWNSTREAM:
await self.push_frame(frame, direction)
async def _handle_upstream_frame(self, frame: Frame):
if isinstance(frame, ErrorFrame):
logger.error(f"Error running app: {frame}")
if frame.fatal:
# Cancel all tasks downstream.
await self.push_frame(CancelFrame())
# Tell the task we should stop.
await self._up_queue.put(StopTaskFrame())
class PipelineTask:
def __init__(
self,
pipeline: BasePipeline,
params: PipelineParams = PipelineParams(),
clock: BaseClock = SystemClock()):
def __init__(self, pipeline: BasePipeline, params: PipelineParams = PipelineParams()):
self.id: int = obj_id()
self.name: str = f"{self.__class__.__name__}#{obj_count(self)}"
self._pipeline = pipeline
self._clock = clock
self._params = params
self._finished = False
@@ -93,7 +70,7 @@ class PipelineTask:
# Make sure everything is cleaned up downstream. This is sent
# out-of-band from the main streaming task which is what we want since
# we want to cancel right away.
await self._source.push_frame(CancelFrame())
await self._source.process_frame(CancelFrame(), FrameDirection.DOWNSTREAM)
self._process_down_task.cancel()
self._process_up_task.cancel()
await self._process_down_task
@@ -115,6 +92,8 @@ class PipelineTask:
elif isinstance(frames, Iterable):
for frame in frames:
await self.queue_frame(frame)
else:
raise Exception("Frames must be an iterable or async iterable")
def _initial_metrics_frame(self) -> MetricsFrame:
processors = self._pipeline.processors_with_metrics()
@@ -123,18 +102,15 @@ class PipelineTask:
return MetricsFrame(ttfb=ttfb, processing=processing)
async def _process_down_queue(self):
self._clock.start()
start_frame = StartFrame(
allow_interruptions=self._params.allow_interruptions,
enable_metrics=self._params.enable_metrics,
enable_usage_metrics=self._params.enable_metrics,
report_only_initial_ttfb=self._params.report_only_initial_ttfb,
clock=self._clock
report_only_initial_ttfb=self._params.report_only_initial_ttfb
)
await self._source.process_frame(start_frame, FrameDirection.DOWNSTREAM)
if self._params.enable_metrics and self._params.send_initial_empty_metrics:
if self._params.send_initial_empty_metrics:
await self._source.process_frame(self._initial_metrics_frame(), FrameDirection.DOWNSTREAM)
running = True
@@ -160,8 +136,9 @@ class PipelineTask:
while True:
try:
frame = await self._up_queue.get()
if isinstance(frame, StopTaskFrame):
await self.queue_frame(StopTaskFrame())
if isinstance(frame, ErrorFrame):
logger.error(f"Error running app: {frame.error}")
await self.queue_frame(CancelFrame())
self._up_queue.task_done()
except asyncio.CancelledError:
break

View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from typing import List, Tuple
from typing import List
from pipecat.frames.frames import Frame, SystemFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -40,14 +40,12 @@ class GatedAggregator(FrameProcessor):
Goodbye.
"""
def __init__(self, gate_open_fn, gate_close_fn, start_open,
direction: FrameDirection = FrameDirection.DOWNSTREAM):
def __init__(self, gate_open_fn, gate_close_fn, start_open):
super().__init__()
self._gate_open_fn = gate_open_fn
self._gate_close_fn = gate_close_fn
self._gate_open = start_open
self._direction = direction
self._accumulator: List[Tuple[Frame, FrameDirection]] = []
self._accumulator: List[Frame] = []
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -57,11 +55,6 @@ class GatedAggregator(FrameProcessor):
await self.push_frame(frame, direction)
return
# Ignore frames that are not following the direction of this gate.
if direction != self._direction:
await self.push_frame(frame, direction)
return
old_state = self._gate_open
if self._gate_open:
self._gate_open = not self._gate_close_fn(frame)
@@ -74,8 +67,8 @@ class GatedAggregator(FrameProcessor):
if self._gate_open:
await self.push_frame(frame, direction)
for (f, d) in self._accumulator:
await self.push_frame(f, d)
for frame in self._accumulator:
await self.push_frame(frame, direction)
self._accumulator = []
else:
self._accumulator.append((frame, direction))
self._accumulator.append(frame)

View File

@@ -4,10 +4,9 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import sys
from typing import List
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame, OpenAILLMContext
from pipecat.services.openai import OpenAILLMContextFrame, OpenAILLMContext
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.frames.frames import (
@@ -18,7 +17,6 @@ from pipecat.frames.frames import (
LLMMessagesAppendFrame,
LLMMessagesFrame,
LLMMessagesUpdateFrame,
LLMSetToolsFrame,
StartInterruptionFrame,
TranscriptionFrame,
TextFrame,
@@ -109,7 +107,7 @@ class LLMResponseAggregator(FrameProcessor):
await self.push_frame(frame, direction)
elif isinstance(frame, self._accumulator_frame):
if self._aggregating:
self._aggregation += f" {frame.text}" if self._aggregation else frame.text
self._aggregation += f" {frame.text}"
# We have recevied a complete sentence, so if we have seen the
# end frame and we were still aggregating, it means we should
# send the aggregation.
@@ -125,11 +123,18 @@ class LLMResponseAggregator(FrameProcessor):
self._reset()
await self.push_frame(frame, direction)
elif isinstance(frame, LLMMessagesAppendFrame):
self._add_messages(frame.messages)
self._messages.extend(frame.messages)
messages_frame = LLMMessagesFrame(self._messages)
await self.push_frame(messages_frame)
elif isinstance(frame, LLMMessagesUpdateFrame):
self._set_messages(frame.messages)
elif isinstance(frame, LLMSetToolsFrame):
self._set_tools(frame.tools)
# We push the frame downstream so the assistant aggregator gets
# updated as well.
await self.push_frame(frame)
# We can now reset this one.
self._reset()
self._messages = frame.messages
messages_frame = LLMMessagesFrame(self._messages)
await self.push_frame(messages_frame)
else:
await self.push_frame(frame, direction)
@@ -147,19 +152,6 @@ class LLMResponseAggregator(FrameProcessor):
frame = LLMMessagesFrame(self._messages)
await self.push_frame(frame)
# TODO-CB: Types
def _add_messages(self, messages):
self._messages.extend(messages)
def _set_messages(self, messages):
self._reset()
self._messages.clear()
self._messages.extend(messages)
def _set_tools(self, tools):
# noop in the base class
pass
def _reset(self):
self._aggregation = ""
self._aggregating = False
@@ -248,29 +240,9 @@ class LLMFullResponseAggregator(FrameProcessor):
class LLMContextAggregator(LLMResponseAggregator):
def __init__(self, *, context: OpenAILLMContext, **kwargs):
super().__init__(**kwargs)
self._context = context
@property
def context(self):
return self._context
def get_context_frame(self) -> OpenAILLMContextFrame:
return OpenAILLMContextFrame(context=self._context)
async def push_context_frame(self):
frame = self.get_context_frame()
await self.push_frame(frame)
# TODO-CB: Types
def _add_messages(self, messages):
self._context.add_messages(messages)
def _set_messages(self, messages):
self._context.set_messages(messages)
def _set_tools(self, tools: List):
self._context.set_tools(tools)
super().__init__(**kwargs)
async def _push_aggregation(self):
if len(self._aggregation) > 0:

View File

@@ -4,33 +4,23 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from dataclasses import dataclass
import io
import json
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, List
from typing import List
from PIL import Image
from pipecat.frames.frames import Frame, VisionImageRawFrame, FunctionCallInProgressFrame, FunctionCallResultFrame
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.frames.frames import Frame, VisionImageRawFrame
from loguru import logger
from openai._types import NOT_GIVEN, NotGiven
try:
from openai._types import NOT_GIVEN, NotGiven
from openai.types.chat import (
ChatCompletionToolParam,
ChatCompletionToolChoiceOptionParam,
ChatCompletionMessageParam
)
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use OpenAI, you need to `pip install pipecat-ai[openai]`. Also, set `OPENAI_API_KEY` environment variable.")
raise Exception(f"Missing module: {e}")
from openai.types.chat import (
ChatCompletionToolParam,
ChatCompletionToolChoiceOptionParam,
ChatCompletionMessageParam
)
# JSON custom encoder to handle bytes arrays so that we can log contexts
# with images to the console.
@@ -52,19 +42,20 @@ class OpenAILLMContext:
tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN,
tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = NOT_GIVEN
):
self._messages: List[ChatCompletionMessageParam] = messages if messages else [
self.messages: List[ChatCompletionMessageParam] = messages if messages else [
]
self._tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice
self._tools: List[ChatCompletionToolParam] | NotGiven = tools
self.tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice
self.tools: List[ChatCompletionToolParam] | NotGiven = tools
@staticmethod
def from_messages(messages: List[dict]) -> "OpenAILLMContext":
context = OpenAILLMContext()
for message in messages:
if "name" not in message:
message["name"] = message["role"]
context.add_message(message)
context.add_message({
"content": message["content"],
"role": message["role"],
"name": message["name"] if "name" in message else message["role"]
})
return context
@staticmethod
@@ -92,75 +83,25 @@ class OpenAILLMContext:
})
return context
@property
def messages(self) -> List[ChatCompletionMessageParam]:
return self._messages
@property
def tools(self) -> List[ChatCompletionToolParam] | NotGiven:
return self._tools
@property
def tool_choice(self) -> ChatCompletionToolChoiceOptionParam | NotGiven:
return self._tool_choice
def add_message(self, message: ChatCompletionMessageParam):
self._messages.append(message)
def add_messages(self, messages: List[ChatCompletionMessageParam]):
self._messages.extend(messages)
def set_messages(self, messages: List[ChatCompletionMessageParam]):
self._messages[:] = messages
self.messages.append(message)
def get_messages(self) -> List[ChatCompletionMessageParam]:
return self._messages
return self.messages
def get_messages_json(self) -> str:
return json.dumps(self._messages, cls=CustomEncoder)
return json.dumps(self.messages, cls=CustomEncoder)
def set_tool_choice(
self, tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven
):
self._tool_choice = tool_choice
self.tool_choice = tool_choice
def set_tools(self, tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN):
if tools != NOT_GIVEN and len(tools) == 0:
tools = NOT_GIVEN
self._tools = tools
async def call_function(self,
f: Callable[[str,
str,
Any,
FrameProcessor,
'OpenAILLMContext',
Callable[[Any],
Awaitable[None]]],
Awaitable[None]],
*,
function_name: str,
tool_call_id: str,
arguments: str,
llm: FrameProcessor) -> None:
# Push a SystemFrame downstream. This frame will let our assistant context aggregator
# know that we are in the middle of a function call. Some contexts/aggregators may
# not need this. But some definitely do (Anthropic, for example).
await llm.push_frame(FunctionCallInProgressFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
))
# Define a callback function that pushes a FunctionCallResultFrame downstream.
async def function_call_result_callback(result):
await llm.push_frame(FunctionCallResultFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
result=result))
await f(function_name, tool_call_id, arguments, llm, self, function_call_result_callback)
self.tools = tools
@dataclass

View File

@@ -4,9 +4,10 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import re
from pipecat.frames.frames import EndFrame, Frame, InterimTranscriptionFrame, TextFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.string import match_endofsentence
class SentenceAggregator(FrameProcessor):
@@ -39,10 +40,12 @@ class SentenceAggregator(FrameProcessor):
return
if isinstance(frame, TextFrame):
self._aggregation += frame.text
if match_endofsentence(self._aggregation):
await self.push_frame(TextFrame(self._aggregation))
self._aggregation = ""
m = re.search("(.*[?.!])(.*)", frame.text)
if m:
await self.push_frame(TextFrame(self._aggregation + m.group(1)))
self._aggregation = m.group(2)
else:
self._aggregation += frame.text
elif isinstance(frame, EndFrame):
if self._aggregation:
await self.push_frame(TextFrame(self._aggregation))

View File

@@ -9,14 +9,7 @@ import time
from enum import Enum
from pipecat.clocks.base_clock import BaseClock
from pipecat.frames.frames import (
ErrorFrame,
Frame,
MetricsFrame,
StartFrame,
StartInterruptionFrame,
UserStoppedSpeakingFrame)
from pipecat.frames.frames import ErrorFrame, Frame, MetricsFrame, StartFrame, StartInterruptionFrame, UserStoppedSpeakingFrame
from pipecat.utils.utils import obj_count, obj_id
from loguru import logger
@@ -97,9 +90,6 @@ class FrameProcessor:
self._next: "FrameProcessor" | None = None
self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_running_loop()
# Clock
self._clock: BaseClock | None = None
# Properties
self._allow_interruptions = False
self._enable_metrics = False
@@ -181,12 +171,8 @@ class FrameProcessor:
def get_parent(self) -> "FrameProcessor":
return self._parent
def get_clock(self) -> BaseClock:
return self._clock
async def process_frame(self, frame: Frame, direction: FrameDirection):
if isinstance(frame, StartFrame):
self._clock = frame.clock
self._allow_interruptions = frame.allow_interruptions
self._enable_metrics = frame.enable_metrics
self._enable_usage_metrics = frame.enable_usage_metrics

View File

@@ -5,46 +5,53 @@
#
import asyncio
import dataclasses
from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Union
from pydantic import BaseModel, Field, PrivateAttr, ValidationError
from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Type
from pydantic import PrivateAttr, BaseModel, ValidationError
from pipecat.frames.frames import (
BotInterruptionFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
ErrorFrame,
Frame,
InterimTranscriptionFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesAppendFrame,
LLMMessagesUpdateFrame,
LLMModelUpdateFrame,
MetricsFrame,
StartFrame,
SystemFrame,
TTSSpeakFrame,
TTSVoiceUpdateFrame,
TextFrame,
TranscriptionFrame,
TransportMessageFrame,
UserStartedSpeakingFrame,
FunctionCallResultFrame,
UserStoppedSpeakingFrame)
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.pipeline.pipeline import Pipeline
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, OpenAILLMContext
from pipecat.transports.base_transport import BaseTransport
from loguru import logger
RTVI_PROTOCOL_VERSION = "0.1"
ActionResult = Union[bool, int, float, str, list, dict]
class RTVIServiceOption(BaseModel):
name: str
type: Literal["bool", "number", "string", "array", "object"]
handler: Callable[["RTVIProcessor", str, "RTVIServiceOptionConfig"],
Awaitable[None]] = Field(exclude=True)
handler: Optional[Callable[['RTVIProcessor',
'RTVIServiceOptionConfig'],
Awaitable[None]]] = None
class RTVIService(BaseModel):
name: str
cls: Type[FrameProcessor]
options: List[RTVIServiceOption]
_options_dict: Dict[str, RTVIServiceOption] = PrivateAttr(default={})
@@ -54,31 +61,9 @@ class RTVIService(BaseModel):
self._options_dict[option.name] = option
return super().model_post_init(__context)
class RTVIActionArgumentData(BaseModel):
name: str
value: Any
class RTVIActionArgument(BaseModel):
name: str
type: Literal["bool", "number", "string", "array", "object"]
class RTVIAction(BaseModel):
service: str
action: str
arguments: List[RTVIActionArgument] = []
result: Literal["bool", "number", "string", "array", "object"]
handler: Callable[["RTVIProcessor", str, Dict[str, Any]],
Awaitable[ActionResult]] = Field(exclude=True)
_arguments_dict: Dict[str, RTVIActionArgument] = PrivateAttr(default={})
def model_post_init(self, __context: Any) -> None:
self._arguments_dict = {}
for arg in self.arguments:
self._arguments_dict[arg.name] = arg
return super().model_post_init(__context)
#
# Client -> Pipecat messages.
#
class RTVIServiceOptionConfig(BaseModel):
@@ -93,27 +78,22 @@ class RTVIServiceConfig(BaseModel):
class RTVIConfig(BaseModel):
config: List[RTVIServiceConfig]
_config_dict: Dict[str, RTVIServiceConfig] = PrivateAttr(default={})
def model_post_init(self, __context: Any) -> None:
self._config_dict = {}
for c in self.config:
self._config_dict[c.service] = c
return super().model_post_init(__context)
#
# Client -> Pipecat messages.
#
class RTVILLMContextData(BaseModel):
messages: List[dict]
class RTVIUpdateConfig(BaseModel):
config: List[RTVIServiceConfig]
interrupt: bool = False
class RTVIActionRunArgument(BaseModel):
name: str
value: Any
class RTVIActionRun(BaseModel):
service: str
action: str
arguments: Optional[List[RTVIActionRunArgument]] = None
class RTVITTSSpeakData(BaseModel):
text: str
interrupt: Optional[bool] = False
class RTVIMessage(BaseModel):
@@ -127,20 +107,20 @@ class RTVIMessage(BaseModel):
#
class RTVIErrorResponseData(BaseModel):
error: str
class RTVIResponseData(BaseModel):
success: bool
error: Optional[str] = None
class RTVIErrorResponse(BaseModel):
class RTVIResponse(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["error-response"] = "error-response"
type: Literal["response"] = "response"
id: str
data: RTVIErrorResponseData
data: RTVIResponseData
class RTVIErrorData(BaseModel):
error: str
fatal: bool
message: str
class RTVIError(BaseModel):
@@ -149,85 +129,29 @@ class RTVIError(BaseModel):
data: RTVIErrorData
class RTVIDescribeConfigData(BaseModel):
config: List[RTVIService]
class RTVILLMContextMessageData(BaseModel):
messages: List[dict]
class RTVIDescribeConfig(BaseModel):
class RTVILLMContextMessage(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["config-available"] = "config-available"
id: str
data: RTVIDescribeConfigData
type: Literal["llm-context"] = "llm-context"
data: RTVILLMContextMessageData
class RTVIDescribeActionsData(BaseModel):
actions: List[RTVIAction]
class RTVITTSTextMessageData(BaseModel):
text: str
class RTVIDescribeActions(BaseModel):
class RTVITTSTextMessage(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["actions-available"] = "actions-available"
id: str
data: RTVIDescribeActionsData
class RTVIConfigResponse(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["config"] = "config"
id: str
data: RTVIConfig
class RTVIActionResponseData(BaseModel):
result: ActionResult
class RTVIActionResponse(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["action-response"] = "action-response"
id: str
data: RTVIActionResponseData
class RTVIBotReadyData(BaseModel):
version: str
config: List[RTVIServiceConfig]
type: Literal["tts-text"] = "tts-text"
data: RTVITTSTextMessageData
class RTVIBotReady(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["bot-ready"] = "bot-ready"
id: str
data: RTVIBotReadyData
class RTVILLMFunctionCallMessageData(BaseModel):
function_name: str
tool_call_id: str
args: dict
class RTVILLMFunctionCallMessage(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["llm-function-call"] = "llm-function-call"
data: RTVILLMFunctionCallMessageData
class RTVILLMFunctionCallStartMessageData(BaseModel):
function_name: str
class RTVILLMFunctionCallStartMessage(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["llm-function-call-start"] = "llm-function-call-start"
data: RTVILLMFunctionCallStartMessageData
class RTVILLMFunctionCallResultData(BaseModel):
function_name: str
tool_call_id: str
arguments: dict
result: dict | str
class RTVITranscriptionMessageData(BaseModel):
@@ -253,93 +177,177 @@ class RTVIUserStoppedSpeakingMessage(BaseModel):
type: Literal["user-stopped-speaking"] = "user-stopped-speaking"
class RTVIBotStartedSpeakingMessage(BaseModel):
class RTVIJSONCompletion(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["bot-started-speaking"] = "bot-started-speaking"
type: Literal["json-completion"] = "json-completion"
data: str
class RTVIBotStoppedSpeakingMessage(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["bot-stopped-speaking"] = "bot-stopped-speaking"
class FunctionCaller(FrameProcessor):
def __init__(self, context):
super().__init__()
self._checking = False
self._aggregating = False
self._emitted_start = False
self._aggregation = ""
self._context = context
self._callbacks = {}
self._start_callbacks = {}
def register_function(self, function_name: str, callback, start_callback=None):
self._callbacks[function_name] = callback
if start_callback:
self._start_callbacks[function_name] = start_callback
def unregister_function(self, function_name: str):
del self._callbacks[function_name]
if self._start_callbacks[function_name]:
del self._start_callbacks[function_name]
def has_function(self, function_name: str):
return function_name in self._callbacks.keys()
async def call_function(self, function_name: str, args):
if function_name in self._callbacks.keys():
return await self._callbacks[function_name](self, args)
return None
async def call_start_function(self, function_name: str):
if function_name in self._start_callbacks.keys():
await self._start_callbacks[function_name](self)
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, LLMFullResponseStartFrame):
self._checking = True
await self.push_frame(frame, direction)
elif isinstance(frame, TextFrame) and self._checking:
# TODO-CB: should we expand this to any non-text character to start the completion?
if frame.text.strip().startswith("{") or frame.text.strip().startswith("```"):
self._emitted_start = False
self._checking = False
self._aggregation = frame.text
self._aggregating = True
else:
self._checking = False
self._aggregating = False
self._aggregation = ""
self._emitted_start = False
await self.push_frame(frame, direction)
elif isinstance(frame, TextFrame) and self._aggregating:
self._aggregation += frame.text
# TODO-CB: We can probably ignore function start I think
# if not self._emitted_start:
# fn = re.search(r'{"function_name":\s*"(.*)",', self._aggregation)
# if fn and fn.group(1):
# await self.call_start_function(fn.group(1))
# self._emitted_start = True
elif isinstance(frame, LLMFullResponseEndFrame) and self._aggregating:
try:
self._aggregation = self._aggregation.replace("```json", "").replace("```", "")
self._context.add_message({"role": "assistant", "content": self._aggregation})
message = RTVIJSONCompletion(data=self._aggregation)
msg = message.model_dump(exclude_none=True)
await self.push_frame(TransportMessageFrame(message=msg))
except Exception as e:
print(f"Error parsing function call json: {e}")
print(f"aggregation was: {self._aggregation}")
self._aggregating = False
self._aggregation = ""
self._emitted_start = False
elif isinstance(frame, LLMFullResponseEndFrame):
await self.push_frame(frame, direction)
else:
await self.push_frame(frame, direction)
class RTVIProcessorParams(BaseModel):
send_bot_ready: bool = True
class RTVITTSTextProcessor(FrameProcessor):
def __init__(self):
super().__init__()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
await self.push_frame(frame, direction)
if isinstance(frame, TextFrame):
message = RTVITTSTextMessage(data=RTVITTSTextMessageData(text=frame.text))
await self.push_frame(TransportMessageFrame(message=message.model_dump(exclude_none=True)))
async def handle_llm_model_update(rtvi: 'RTVIProcessor', option: RTVIServiceOptionConfig):
frame = LLMModelUpdateFrame(option.value)
await rtvi.push_frame(frame)
async def handle_llm_messages_update(rtvi: 'RTVIProcessor', option: RTVIServiceOptionConfig):
frame = LLMMessagesUpdateFrame(option.value)
await rtvi.push_frame(frame)
async def handle_tts_voice_update(rtvi: 'RTVIProcessor', option: RTVIServiceOptionConfig):
frame = TTSVoiceUpdateFrame(option.value)
await rtvi.push_frame(frame)
DEFAULT_LLM_SERVICE = RTVIService(
name="llm",
cls=OpenAILLMService,
options=[
RTVIServiceOption(name="model", handler=handle_llm_model_update),
RTVIServiceOption(name="messages", handler=handle_llm_messages_update)
])
DEFAULT_TTS_SERVICE = RTVIService(
name="tts",
cls=CartesiaTTSService,
options=[
RTVIServiceOption(name="voice_id", handler=handle_tts_voice_update),
])
class RTVIProcessor(FrameProcessor):
def __init__(self,
*,
config: RTVIConfig = RTVIConfig(config=[]),
params: RTVIProcessorParams = RTVIProcessorParams()):
def __init__(self, *, transport: BaseTransport):
super().__init__()
self._config = config
self._params = params
self._transport = transport
self._config: RTVIConfig | None = None
self._ctor_args: Dict[str, Any] = {}
self._start_frame: Frame | None = None
self._pipeline: FrameProcessor | None = None
self._pipeline_started = False
self._first_participant_joined: bool = False
self._client_ready = False
self._client_ready_id = ""
# Register transport event so we can send a `bot-ready` event (and maybe
# others) when the participant joins.
transport.add_event_handler(
"on_first_participant_joined",
self._on_first_participant_joined)
self._registered_actions: Dict[str, RTVIAction] = {}
# Register default services.
self._registered_services: Dict[str, RTVIService] = {}
self.register_service(DEFAULT_LLM_SERVICE)
self.register_service(DEFAULT_TTS_SERVICE)
self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler())
self._push_queue = asyncio.Queue()
self._message_task = self.get_event_loop().create_task(self._message_task_handler())
self._message_queue = asyncio.Queue()
def register_action(self, action: RTVIAction):
id = self._action_id(action.service, action.action)
self._registered_actions[id] = action
self._frame_handler_task = self.get_event_loop().create_task(self._frame_handler())
self._frame_queue = asyncio.Queue()
def register_service(self, service: RTVIService):
self._registered_services[service.name] = service
async def interrupt_bot(self):
await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
def setup_on_start(self, config: RTVIConfig | None, ctor_args: Dict[str, Any]):
self._config = config
self._ctor_args = ctor_args
async def send_error(self, error: str):
message = RTVIError(data=RTVIErrorData(error=error, fatal=False))
await self._push_transport_message(message)
async def set_client_ready(self):
if not self._client_ready:
self._client_ready = True
await self._maybe_send_bot_ready()
async def handle_function_call(
self,
function_name: str,
tool_call_id: str,
arguments: dict,
llm: FrameProcessor,
context: OpenAILLMContext,
result_callback):
fn = RTVILLMFunctionCallMessageData(
function_name=function_name,
tool_call_id=tool_call_id,
args=arguments)
message = RTVILLMFunctionCallMessage(data=fn)
await self._push_transport_message(message, exclude_none=False)
async def handle_function_call_start(
self,
function_name: str,
llm: FrameProcessor,
context: OpenAILLMContext):
fn = RTVILLMFunctionCallStartMessageData(function_name=function_name)
message = RTVILLMFunctionCallStartMessage(data=fn)
await self._push_transport_message(message, exclude_none=False)
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
if isinstance(frame, SystemFrame):
await super().push_frame(frame, direction)
else:
await self._internal_push_frame(frame, direction)
async def update_config(self, config: RTVIConfig):
if self._pipeline:
await self._handle_config_update(config)
self._config = config
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -348,86 +356,71 @@ class RTVIProcessor(FrameProcessor):
if isinstance(frame, CancelFrame):
await self._cancel(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, ErrorFrame):
await self._send_error_frame(frame)
await self.push_frame(frame, direction)
# All other system frames
elif isinstance(frame, SystemFrame):
await self.push_frame(frame, direction)
# Control frames
elif isinstance(frame, StartFrame):
# Push StartFrame before start(), because we want StartFrame to be
# processed by every processor before any other frame is processed.
await self.push_frame(frame, direction)
await self._start(frame)
await self._internal_push_frame(frame, direction)
elif isinstance(frame, EndFrame):
# Push EndFrame before stop(), because stop() waits on the task to
# finish and the task finishes when EndFrame is processed.
await self.push_frame(frame, direction)
await self._internal_push_frame(frame, direction)
await self._stop(frame)
elif isinstance(frame, UserStartedSpeakingFrame) or isinstance(frame, UserStoppedSpeakingFrame):
await self._handle_interruptions(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, BotStartedSpeakingFrame) or isinstance(frame, BotStoppedSpeakingFrame):
await self._handle_bot_speaking(frame)
await self.push_frame(frame, direction)
# Data frames
elif isinstance(frame, TranscriptionFrame) or isinstance(frame, InterimTranscriptionFrame):
await self._handle_transcriptions(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, TransportMessageFrame):
await self._message_queue.put(frame)
# Other frames
else:
await self.push_frame(frame, direction)
await self._internal_push_frame(frame, direction)
async def cleanup(self):
if self._pipeline:
await self._pipeline.cleanup()
async def _start(self, frame: StartFrame):
self._pipeline_started = True
await self._maybe_send_bot_ready()
try:
await self._handle_pipeline_setup(frame, self._config)
except Exception as e:
await self._send_error(f"unable to setup RTVI pipeline: {e}")
async def _stop(self, frame: EndFrame):
# We need to cancel the message task handler because that one is not
# processing EndFrames.
self._message_task.cancel()
await self._message_task
await self._push_frame_task
await self._frame_handler_task
async def _cancel(self, frame: CancelFrame):
self._message_task.cancel()
await self._message_task
self._push_frame_task.cancel()
await self._push_frame_task
self._frame_handler_task.cancel()
await self._frame_handler_task
async def _internal_push_frame(
self,
frame: Frame | None,
direction: FrameDirection | None = FrameDirection.DOWNSTREAM):
await self._push_queue.put((frame, direction))
await self._frame_queue.put((frame, direction))
async def _push_frame_task_handler(self):
async def _frame_handler(self):
running = True
while running:
try:
(frame, direction) = await self._push_queue.get()
await super().push_frame(frame, direction)
self._push_queue.task_done()
(frame, direction) = await self._frame_queue.get()
await self._handle_frame(frame, direction)
self._frame_queue.task_done()
running = not isinstance(frame, EndFrame)
except asyncio.CancelledError:
break
async def _push_transport_message(self, model: BaseModel, exclude_none: bool = True):
frame = TransportMessageFrame(
message=model.model_dump(exclude_none=exclude_none),
urgent=True)
await self.push_frame(frame)
async def _handle_frame(self, frame: Frame, direction: FrameDirection):
if isinstance(frame, TransportMessageFrame):
await self._handle_message(frame)
else:
await self.push_frame(frame, direction)
if isinstance(frame, TranscriptionFrame) or isinstance(frame, InterimTranscriptionFrame):
await self._handle_transcriptions(frame)
elif isinstance(frame, UserStartedSpeakingFrame) or isinstance(frame, UserStoppedSpeakingFrame):
await self._handle_interruptions(frame)
async def _handle_transcriptions(self, frame: Frame):
# TODO(aleix): Once we add support for using custom pipelines, the STTs will
# be in the pipeline after this processor.
# TODO(aleix): Once we add support for using custom piplines, the STTs will
# be in the pipeline after this processor. This means the STT will have to
# push transcriptions upstream as well.
message = None
if isinstance(frame, TranscriptionFrame):
@@ -446,7 +439,8 @@ class RTVIProcessor(FrameProcessor):
final=False))
if message:
await self._push_transport_message(message)
frame = TransportMessageFrame(message=message.model_dump(exclude_none=True))
await self.push_frame(frame)
async def _handle_interruptions(self, frame: Frame):
message = None
@@ -456,157 +450,170 @@ class RTVIProcessor(FrameProcessor):
message = RTVIUserStoppedSpeakingMessage()
if message:
await self._push_transport_message(message)
async def _handle_bot_speaking(self, frame: Frame):
message = None
if isinstance(frame, BotStartedSpeakingFrame):
message = RTVIBotStartedSpeakingMessage()
elif isinstance(frame, BotStoppedSpeakingFrame):
message = RTVIBotStoppedSpeakingMessage()
if message:
await self._push_transport_message(message)
async def _message_task_handler(self):
while True:
try:
frame = await self._message_queue.get()
await self._handle_message(frame)
self._message_queue.task_done()
except asyncio.CancelledError:
break
frame = TransportMessageFrame(message=message.model_dump(exclude_none=True))
await self.push_frame(frame)
async def _handle_message(self, frame: TransportMessageFrame):
try:
message = RTVIMessage.model_validate(frame.message)
except ValidationError as e:
await self.send_error(f"Invalid incoming message: {e}")
await self._send_error(f"Invalid incoming message: {e}")
logger.warning(f"Invalid incoming message: {e}")
return
try:
success = True
error = None
match message.type:
case "client-ready":
await self._handle_client_ready(message.id)
case "describe-actions":
await self._handle_describe_actions(message.id)
case "describe-config":
await self._handle_describe_config(message.id)
case "get-config":
await self._handle_get_config(message.id)
case "update-config":
update_config = RTVIUpdateConfig.model_validate(message.data)
await self._handle_update_config(message.id, update_config)
case "action":
action = RTVIActionRun.model_validate(message.data)
await self._handle_action(message.id, action)
case "llm-function-call-result":
data = RTVILLMFunctionCallResultData.model_validate(message.data)
await self._handle_function_call_result(data)
case "config-update":
await self._handle_config_update(RTVIConfig.model_validate(message.data))
case "llm-get-context":
await self._handle_llm_get_context()
case "llm-append-context":
await self._handle_llm_append_context(RTVILLMContextData.model_validate(message.data))
case "llm-update-context":
await self._handle_llm_update_context(RTVILLMContextData.model_validate(message.data))
case "tts-speak":
await self._handle_tts_speak(RTVITTSSpeakData.model_validate(message.data))
case "tts-interrupt":
await self._handle_tts_interrupt()
case _:
await self._send_error_response(message.id, f"Unsupported type {message.type}")
success = False
error = f"Unsupported type {message.type}"
await self._send_response(message.id, success, error)
except ValidationError as e:
await self._send_error_response(message.id, f"Invalid incoming message: {e}")
await self._send_response(message.id, False, f"Invalid incoming message: {e}")
logger.warning(f"Invalid incoming message: {e}")
except Exception as e:
await self._send_error_response(message.id, f"Exception processing message: {e}")
await self._send_response(message.id, False, f"Exception processing message: {e}")
logger.warning(f"Exception processing message: {e}")
async def _handle_client_ready(self, request_id: str):
self._client_ready = True
self._client_ready_id = request_id
async def _handle_pipeline_setup(self, start_frame: StartFrame, config: RTVIConfig | None):
# TODO(aleix): We shouldn't need to save this in `self._tma_in`.
self._tma_in = LLMUserResponseAggregator()
tma_out = LLMAssistantResponseAggregator()
llm_cls = self._registered_services["llm"].cls
llm_args = self._ctor_args["llm"]
llm = llm_cls(**llm_args)
tts_cls = self._registered_services["tts"].cls
tts_args = self._ctor_args["tts"]
tts = tts_cls(**tts_args)
# TODO-CB: Eventually we'll need to switch the context aggregators to use the
# OpenAI context frames instead of message frames
context = OpenAILLMContext()
fc = FunctionCaller(context)
tts_text = RTVITTSTextProcessor()
pipeline = Pipeline([
self._tma_in,
llm,
fc,
tts,
tts_text,
tma_out,
self._transport.output(),
])
parent = self.get_parent()
if parent:
parent.link(pipeline)
# We need to initialize the new pipeline with the same settings
# as the initial one.
start_frame = dataclasses.replace(start_frame)
await self.push_frame(start_frame)
# Configure the pipeline
if config:
await self._handle_config_update(config)
# Send new initial metrics with the new processors
processors = parent.processors_with_metrics()
processors.extend(pipeline.processors_with_metrics())
ttfb = [{"processor": p.name, "value": 0.0} for p in processors]
processing = [{"processor": p.name, "value": 0.0} for p in processors]
tokens = [{"processor": p.name, "value": {"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0}} for p in processors]
characters = [{"processor": p.name, "value": 0} for p in processors]
await self.push_frame(MetricsFrame(ttfb=ttfb, processing=processing, tokens=tokens, characters=characters))
self._pipeline = pipeline
await self._maybe_send_bot_ready()
async def _handle_describe_config(self, request_id: str):
services = list(self._registered_services.values())
message = RTVIDescribeConfig(id=request_id, data=RTVIDescribeConfigData(config=services))
await self._push_transport_message(message)
async def _handle_describe_actions(self, request_id: str):
actions = list(self._registered_actions.values())
message = RTVIDescribeActions(id=request_id, data=RTVIDescribeActionsData(actions=actions))
await self._push_transport_message(message)
async def _handle_get_config(self, request_id: str):
message = RTVIConfigResponse(id=request_id, data=self._config)
await self._push_transport_message(message)
def _update_config_option(self, service: str, config: RTVIServiceOptionConfig):
for service_config in self._config.config:
if service_config.service == service:
for option_config in service_config.options:
if option_config.name == config.name:
option_config.value = config.value
return
# If we couldn't find a value for this config, we simply need to
# add it.
service_config.options.append(config)
async def _update_service_config(self, config: RTVIServiceConfig):
async def _handle_config_service(self, config: RTVIServiceConfig):
service = self._registered_services[config.service]
for option in config.options:
handler = service._options_dict[option.name].handler
await handler(self, service.name, option)
self._update_config_option(service.name, option)
if handler:
await handler(self, option)
async def _update_config(self, data: RTVIConfig, interrupt: bool):
if interrupt:
await self.interrupt_bot()
for service_config in data.config:
await self._update_service_config(service_config)
async def _handle_config_update(self, data: RTVIConfig):
for config in data.config:
await self._handle_config_service(config)
async def _handle_update_config(self, request_id: str, data: RTVIUpdateConfig):
await self._update_config(RTVIConfig(config=data.config), data.interrupt)
await self._handle_get_config(request_id)
async def _handle_function_call_result(self, data):
frame = FunctionCallResultFrame(
function_name=data.function_name,
tool_call_id=data.tool_call_id,
arguments=data.arguments,
result=data.result)
async def _handle_llm_get_context(self):
data = RTVILLMContextMessageData(messages=self._tma_in.messages)
message = RTVILLMContextMessage(data=data)
frame = TransportMessageFrame(message=message.model_dump(exclude_none=True))
await self.push_frame(frame)
async def _handle_action(self, request_id: str, data: RTVIActionRun):
action_id = self._action_id(data.service, data.action)
if action_id not in self._registered_actions:
await self._send_error_response(request_id, f"Action {action_id} not registered")
return
action = self._registered_actions[action_id]
arguments = {}
if data.arguments:
for arg in data.arguments:
arguments[arg.name] = arg.value
result = await action.handler(self, action.service, arguments)
message = RTVIActionResponse(id=request_id, data=RTVIActionResponseData(result=result))
await self._push_transport_message(message)
async def _handle_llm_append_context(self, data: RTVILLMContextData):
if data and data.messages:
frame = LLMMessagesAppendFrame(data.messages)
await self.push_frame(frame)
async def _handle_llm_update_context(self, data: RTVILLMContextData):
if data and data.messages:
frame = LLMMessagesUpdateFrame(data.messages)
await self.push_frame(frame)
async def _handle_tts_speak(self, data: RTVITTSSpeakData):
if data and data.text:
if data.interrupt:
await self._handle_tts_interrupt()
frame = TTSSpeakFrame(text=data.text)
await self.push_frame(frame)
async def _handle_tts_interrupt(self):
await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
async def _on_first_participant_joined(self, transport, participant):
self._first_participant_joined = True
await self._maybe_send_bot_ready()
async def _maybe_send_bot_ready(self):
if self._pipeline_started and self._client_ready:
await self._send_bot_ready()
await self._update_config(self._config, False)
if self._pipeline and self._first_participant_joined:
message = RTVIBotReady()
frame = TransportMessageFrame(message=message.model_dump(exclude_none=True))
await self.push_frame(frame)
async def _send_bot_ready(self):
if not self._params.send_bot_ready:
return
async def _send_error(self, error: str):
message = RTVIError(data=RTVIErrorData(message=error))
frame = TransportMessageFrame(message=message.model_dump(exclude_none=True))
await self.push_frame(frame)
message = RTVIBotReady(
id=self._client_ready_id,
data=RTVIBotReadyData(
version=RTVI_PROTOCOL_VERSION,
config=self._config.config))
await self._push_transport_message(message)
async def _send_response(self, id: str, success: bool, error: str | None = None):
# TODO(aleix): This is a bit hacky, but we might get invalid
# configuration or something might going wrong during setup and we would
# like to send the error to the client. However, if the pipeline is not
# setup yet we don't have an output transport and therefore we can't
# send any messages. So, we setup a super basic pipeline with just the
# output transport so we can send messages.
if not self._pipeline:
pipeline = Pipeline([self._transport.output()])
self._pipeline = pipeline
async def _send_error_frame(self, frame: ErrorFrame):
message = RTVIError(data=RTVIErrorData(error=frame.error, fatal=frame.fatal))
await self._push_transport_message(message)
parent = self.get_parent()
if parent:
parent.link(pipeline)
async def _send_error_response(self, id: str, error: str):
message = RTVIErrorResponse(id=id, data=RTVIErrorResponseData(error=error))
await self._push_transport_message(message)
def _action_id(self, service: str, action: str) -> str:
return f"{service}:{action}"
message = RTVIResponse(id=id, data=RTVIResponseData(success=success, error=error))
frame = TransportMessageFrame(message=message.model_dump(exclude_none=True))
await self.push_frame(frame)

View File

@@ -27,8 +27,7 @@ try:
from gi.repository import Gst, GstApp
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use GStreamer, you need to `pip install pipecat-ai[gstreamer]`. Also, you need to install GStreamer in your system.")
logger.error("In order to use GStreamer processors, you need to install GStreamer in your system`.")
raise Exception(f"Missing module: {e}")
@@ -78,10 +77,8 @@ class GStreamerPipelineSource(FrameProcessor):
await self.push_frame(frame, direction)
# Control frames
elif isinstance(frame, StartFrame):
# Push StartFrame before start(), because we want StartFrame to be
# processed by every processor before any other frame is processed.
await self._internal_push_frame(frame, direction)
await self._start(frame)
await self._internal_push_frame(frame, direction)
elif isinstance(frame, EndFrame):
# Push EndFrame before stop(), because stop() waits on the task to
# finish and the task finishes when EndFrame is processed.

View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from pipecat.frames.frames import BotSpeakingFrame, Frame, AudioRawFrame, TransportMessageFrame
from pipecat.frames.frames import Frame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from loguru import logger
from typing import Optional
@@ -12,25 +12,16 @@ logger = logger.opt(ansi=True)
class FrameLogger(FrameProcessor):
def __init__(
self,
prefix="Frame",
color: Optional[str] = None,
ignored_frame_types: Optional[list] = [
BotSpeakingFrame,
AudioRawFrame,
TransportMessageFrame]):
def __init__(self, prefix="Frame", color: Optional[str] = None):
super().__init__()
self._prefix = prefix
self._color = color
self._ignored_frame_types = tuple(ignored_frame_types) if ignored_frame_types else None
async def process_frame(self, frame: Frame, direction: FrameDirection):
if self._ignored_frame_types and not isinstance(frame, self._ignored_frame_types):
dir = "<" if direction is FrameDirection.UPSTREAM else ">"
msg = f"{dir} {self._prefix}: {frame}"
if self._color:
msg = f"<{self._color}>{msg}</>"
logger.debug(msg)
dir = "<" if direction is FrameDirection.UPSTREAM else ">"
msg = f"{dir} {self._prefix}: {frame}"
if self._color:
msg = f"<{self._color}>{msg}</>"
logger.debug(msg)
await self.push_frame(frame, direction)

View File

@@ -1,46 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import ctypes
import pickle
from pipecat.frames.frames import AudioRawFrame, Frame
from pipecat.serializers.base_serializer import FrameSerializer
from loguru import logger
try:
from livekit.rtc import AudioFrame
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use LiveKit, you need to `pip install pipecat-ai[livekit]`.")
raise Exception(f"Missing module: {e}")
class LivekitFrameSerializer(FrameSerializer):
SERIALIZABLE_TYPES = {
AudioRawFrame: "audio",
}
def serialize(self, frame: Frame) -> str | bytes | None:
if not isinstance(frame, AudioRawFrame):
return None
audio_frame = AudioFrame(
data=frame.audio,
sample_rate=frame.sample_rate,
num_channels=frame.num_channels,
samples_per_channel=len(frame.audio) // ctypes.sizeof(ctypes.c_int16),
)
return pickle.dumps(audio_frame)
def deserialize(self, data: str | bytes) -> Frame | None:
audio_frame: AudioFrame = pickle.loads(data)['frame']
return AudioRawFrame(
audio=bytes(audio_frame.data),
sample_rate=audio_frame.sample_rate,
num_channels=audio_frame.num_channels,
)

View File

@@ -9,7 +9,7 @@ import json
from pydantic import BaseModel
from pipecat.frames.frames import AudioRawFrame, Frame, StartInterruptionFrame
from pipecat.frames.frames import AudioRawFrame, Frame
from pipecat.serializers.base_serializer import FrameSerializer
from pipecat.utils.audio import ulaw_to_pcm, pcm_to_ulaw
@@ -28,25 +28,22 @@ class TwilioFrameSerializer(FrameSerializer):
self._params = params
def serialize(self, frame: Frame) -> str | bytes | None:
if isinstance(frame, AudioRawFrame):
data = frame.audio
if not isinstance(frame, AudioRawFrame):
return None
serialized_data = pcm_to_ulaw(
data, frame.sample_rate, self._params.twilio_sample_rate)
payload = base64.b64encode(serialized_data).decode("utf-8")
answer = {
"event": "media",
"streamSid": self._stream_sid,
"media": {
"payload": payload
}
data = frame.audio
serialized_data = pcm_to_ulaw(data, frame.sample_rate, self._params.twilio_sample_rate)
payload = base64.b64encode(serialized_data).decode("utf-8")
answer = {
"event": "media",
"streamSid": self._stream_sid,
"media": {
"payload": payload
}
}
return json.dumps(answer)
if isinstance(frame, StartInterruptionFrame):
answer = {"event": "clear", "streamSid": self._stream_sid}
return json.dumps(answer)
return json.dumps(answer)
def deserialize(self, data: str | bytes) -> Frame | None:
message = json.loads(data)

View File

@@ -4,12 +4,11 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import io
import wave
from abc import abstractmethod
from typing import AsyncGenerator, List, Optional, Tuple
from typing import AsyncGenerator
from pipecat.frames.frames import (
AudioRawFrame,
@@ -18,30 +17,37 @@ from pipecat.frames.frames import (
ErrorFrame,
Frame,
LLMFullResponseEndFrame,
STTLanguageUpdateFrame,
STTModelUpdateFrame,
StartFrame,
StartInterruptionFrame,
TTSLanguageUpdateFrame,
TTSModelUpdateFrame,
TTSSpeakFrame,
TTSStartedFrame,
TTSStoppedFrame,
TTSVoiceUpdateFrame,
TextFrame,
UserImageRequestFrame,
VisionImageRawFrame
VisionImageRawFrame,
)
from pipecat.processors.async_frame_processor import AsyncFrameProcessor
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.transcriptions.language import Language
from pipecat.utils.audio import calculate_audio_volume
from pipecat.utils.string import match_endofsentence
from pipecat.utils.time import seconds_to_nanoseconds
from pipecat.utils.utils import exp_smoothing
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
import re
from loguru import logger
ENDOFSENTENCE_PATTERN_STR = r"""
(?<![A-Z]) # Negative lookbehind: not preceded by an uppercase letter (e.g., "U.S.A.")
(?<!\d) # Negative lookbehind: not preceded by a digit (e.g., "1. Let's start")
(?<!\d\s[ap]) # Negative lookbehind: not preceded by time (e.g., "3:00 a.m.")
(?<!Mr|Ms|Dr) # Negative lookbehind: not preceded by Mr, Ms, Dr (combined bc. length is the same)
(?<!Mrs) # Negative lookbehind: not preceded by "Mrs"
(?<!Prof) # Negative lookbehind: not preceded by "Prof"
[\.\?\!:] # Match a period, question mark, exclamation point, or colon
$ # End of string
"""
ENDOFSENTENCE_PATTERN = re.compile(ENDOFSENTENCE_PATTERN_STR, re.VERBOSE)
def match_endofsentence(text: str) -> bool:
return ENDOFSENTENCE_PATTERN.search(text.rstrip()) is not None
class AIService(FrameProcessor):
@@ -109,55 +115,27 @@ class LLMService(AIService):
self._start_callbacks = {}
# TODO-CB: callback function type
def register_function(self, function_name: str | None, callback, start_callback=None):
# Registering a function with the function_name set to None will run that callback
# for all functions
def register_function(self, function_name: str, callback, start_callback=None):
self._callbacks[function_name] = callback
# QUESTION FOR CB: maybe this isn't needed anymore?
if start_callback:
self._start_callbacks[function_name] = start_callback
def unregister_function(self, function_name: str | None):
def unregister_function(self, function_name: str):
del self._callbacks[function_name]
if self._start_callbacks[function_name]:
del self._start_callbacks[function_name]
def has_function(self, function_name: str):
if None in self._callbacks.keys():
return True
return function_name in self._callbacks.keys()
async def call_function(
self,
*,
context: OpenAILLMContext,
tool_call_id: str,
function_name: str,
arguments: str) -> None:
f = None
async def call_function(self, function_name: str, args):
if function_name in self._callbacks.keys():
f = self._callbacks[function_name]
elif None in self._callbacks.keys():
f = self._callbacks[None]
else:
return None
await context.call_function(
f,
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
llm=self)
return await self._callbacks[function_name](self, args)
return None
# QUESTION FOR CB: maybe this isn't needed anymore?
async def call_start_function(self, context: OpenAILLMContext, function_name: str):
async def call_start_function(self, function_name: str):
if function_name in self._start_callbacks.keys():
await self._start_callbacks[function_name](function_name, self, context)
elif None in self._start_callbacks.keys():
return await self._start_callbacks[None](function_name, self, context)
async def request_image_frame(self, user_id: str, *, text_content: str | None = None):
await self.push_frame(UserImageRequestFrame(user_id=user_id, context=text_content),
FrameDirection.UPSTREAM)
await self._start_callbacks[function_name](self)
class TTSService(AIService):
@@ -167,32 +145,16 @@ class TTSService(AIService):
aggregate_sentences: bool = True,
# if True, subclass is responsible for pushing TextFrames and LLMFullResponseEndFrames
push_text_frames: bool = True,
# if True, TTSService will push TTSStoppedFrames, otherwise subclass must do it
push_stop_frames: bool = False,
# if push_stop_frames is True, wait for this idle period before pushing TTSStoppedFrame
stop_frame_timeout_s: float = 1.0,
**kwargs):
super().__init__(**kwargs)
self._aggregate_sentences: bool = aggregate_sentences
self._push_text_frames: bool = push_text_frames
self._push_stop_frames: bool = push_stop_frames
self._stop_frame_timeout_s: float = stop_frame_timeout_s
self._stop_frame_task: Optional[asyncio.Task] = None
self._stop_frame_queue: asyncio.Queue = asyncio.Queue()
self._current_sentence: str = ""
@abstractmethod
async def set_model(self, model: str):
pass
@abstractmethod
async def set_voice(self, voice: str):
pass
@abstractmethod
async def set_language(self, language: Language):
pass
# Converts the text to audio.
@abstractmethod
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
@@ -223,9 +185,11 @@ class TTSService(AIService):
if not text:
return
await self.push_frame(TTSStartedFrame())
await self.start_processing_metrics()
await self.process_generator(self.run_tts(text))
await self.stop_processing_metrics()
await self.push_frame(TTSStoppedFrame())
if self._push_text_frames:
# We send the original text after the audio. This way, if we are
# interrupted, the text is not added to the assistant context.
@@ -249,175 +213,15 @@ class TTSService(AIService):
await self.push_frame(frame, direction)
elif isinstance(frame, TTSSpeakFrame):
await self._push_tts_frames(frame.text, False)
elif isinstance(frame, TTSModelUpdateFrame):
await self.set_model(frame.model)
elif isinstance(frame, TTSVoiceUpdateFrame):
await self.set_voice(frame.voice)
elif isinstance(frame, TTSLanguageUpdateFrame):
await self.set_language(frame.language)
else:
await self.push_frame(frame, direction)
async def start(self, frame: StartFrame):
await super().start(frame)
if self._push_stop_frames:
self._stop_frame_task = self.get_event_loop().create_task(self._stop_frame_handler())
async def stop(self, frame: EndFrame):
await super().stop(frame)
if self._stop_frame_task:
self._stop_frame_task.cancel()
await self._stop_frame_task
self._stop_frame_task = None
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
if self._stop_frame_task:
self._stop_frame_task.cancel()
await self._stop_frame_task
self._stop_frame_task = None
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
await super().push_frame(frame, direction)
if self._push_stop_frames and (
isinstance(frame, StartInterruptionFrame) or
isinstance(frame, TTSStartedFrame) or
isinstance(frame, AudioRawFrame) or
isinstance(frame, TTSStoppedFrame)):
await self._stop_frame_queue.put(frame)
async def _stop_frame_handler(self):
try:
has_started = False
while True:
try:
frame = await asyncio.wait_for(self._stop_frame_queue.get(),
self._stop_frame_timeout_s)
if isinstance(frame, TTSStartedFrame):
has_started = True
elif isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)):
has_started = False
except asyncio.TimeoutError:
if has_started:
await self.push_frame(TTSStoppedFrame())
has_started = False
except asyncio.CancelledError:
pass
class AsyncTTSService(TTSService):
def __init__(self, **kwargs):
super().__init__(**kwargs)
@abstractmethod
async def flush_audio(self):
pass
class AsyncWordTTSService(AsyncTTSService):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._initial_word_timestamp = -1
self._words_queue = asyncio.Queue()
self._words_task = self.get_event_loop().create_task(self._words_task_handler())
def start_word_timestamps(self):
if self._initial_word_timestamp == -1:
self._initial_word_timestamp = self.get_clock().get_time()
def reset_word_timestamps(self):
self._initial_word_timestamp = -1
self._word_timestamps = []
async def add_word_timestamps(self, word_times: List[Tuple[str, float]]):
for (word, timestamp) in word_times:
await self._words_queue.put((word, seconds_to_nanoseconds(timestamp)))
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._stop_words_task()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._stop_words_task()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, LLMFullResponseEndFrame) or isinstance(frame, EndFrame):
await self.flush_audio()
async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
await super()._handle_interruption(frame, direction)
self.reset_word_timestamps()
async def _stop_words_task(self):
if self._words_task:
self._words_task.cancel()
await self._words_task
async def _words_task_handler(self):
while True:
try:
(word, timestamp) = await self._words_queue.get()
if word == "LLMFullResponseEndFrame" and timestamp == 0:
await self.push_frame(LLMFullResponseEndFrame())
else:
frame = TextFrame(word)
frame.pts = self._initial_word_timestamp + timestamp
await self.push_frame(frame)
self._words_queue.task_done()
except asyncio.CancelledError:
break
except Exception as e:
logger.exception(f"{self} exception: {e}")
class STTService(AIService):
"""STTService is a base class for speech-to-text services."""
def __init__(self, **kwargs):
super().__init__(**kwargs)
@abstractmethod
async def set_model(self, model: str):
pass
@abstractmethod
async def set_language(self, language: Language):
pass
@abstractmethod
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Returns transcript as a string"""
pass
async def process_audio_frame(self, frame: AudioRawFrame):
await self.process_generator(self.run_stt(frame.audio))
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Processes a frame of audio data, either buffering or transcribing it."""
await super().process_frame(frame, direction)
if isinstance(frame, AudioRawFrame):
# In this service we accumulate audio internally and at the end we
# push a TextFrame. We don't really want to push audio frames down.
await self.process_audio_frame(frame)
elif isinstance(frame, STTModelUpdateFrame):
await self.set_model(frame.model)
elif isinstance(frame, STTLanguageUpdateFrame):
await self.set_language(frame.language)
else:
await self.push_frame(frame, direction)
class SegmentedSTTService(STTService):
"""SegmentedSTTService is an STTService that will detect speech and will run
speech-to-text on speech segments only, instead of a continous stream.
"""
def __init__(self,
*,
min_volume: float = 0.6,
@@ -438,7 +242,24 @@ class SegmentedSTTService(STTService):
self._smoothing_factor = 0.2
self._prev_volume = 0
async def process_audio_frame(self, frame: AudioRawFrame):
@abstractmethod
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Returns transcript as a string"""
pass
def _new_wave(self):
content = io.BytesIO()
ww = wave.open(content, "wb")
ww.setsampwidth(2)
ww.setnchannels(self._num_channels)
ww.setframerate(self._sample_rate)
return (content, ww)
def _get_smoothed_volume(self, frame: AudioRawFrame) -> float:
volume = calculate_audio_volume(frame.audio, frame.sample_rate)
return exp_smoothing(volume, self._prev_volume, self._smoothing_factor)
async def _append_audio(self, frame: AudioRawFrame):
# Try to filter out empty background noise
volume = self._get_smoothed_volume(frame)
if volume >= self._min_volume:
@@ -458,7 +279,9 @@ class SegmentedSTTService(STTService):
self._silence_num_frames = 0
self._wave.close()
self._content.seek(0)
await self.start_processing_metrics()
await self.process_generator(self.run_stt(self._content.read()))
await self.stop_processing_metrics()
(self._content, self._wave) = self._new_wave()
async def stop(self, frame: EndFrame):
@@ -467,17 +290,16 @@ class SegmentedSTTService(STTService):
async def cancel(self, frame: CancelFrame):
self._wave.close()
def _new_wave(self):
content = io.BytesIO()
ww = wave.open(content, "wb")
ww.setsampwidth(2)
ww.setnchannels(self._num_channels)
ww.setframerate(self._sample_rate)
return (content, ww)
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Processes a frame of audio data, either buffering or transcribing it."""
await super().process_frame(frame, direction)
def _get_smoothed_volume(self, frame: AudioRawFrame) -> float:
volume = calculate_audio_volume(frame.audio, frame.sample_rate)
return exp_smoothing(volume, self._prev_volume, self._smoothing_factor)
if isinstance(frame, AudioRawFrame):
# In this service we accumulate audio internally and at the end we
# push a TextFrame. We don't really want to push audio frames down.
await self._append_audio(frame)
else:
await self.push_frame(frame, direction)
class ImageGenService(AIService):

View File

@@ -5,137 +5,108 @@
#
import base64
import json
import io
import copy
from typing import List, Optional
from dataclasses import dataclass
from PIL import Image
from asyncio import CancelledError
import re
from pipecat.frames.frames import (
Frame,
LLMEnablePromptCachingFrame,
LLMModelUpdateFrame,
TextFrame,
VisionImageRawFrame,
UserImageRequestFrame,
UserImageRawFrame,
LLMMessagesFrame,
LLMFullResponseStartFrame,
LLMFullResponseEndFrame,
FunctionCallResultFrame,
FunctionCallInProgressFrame,
StartInterruptionFrame
LLMFullResponseEndFrame
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import LLMService
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame
)
from pipecat.processors.aggregators.llm_response import (
LLMUserContextAggregator,
LLMAssistantContextAggregator
)
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext, OpenAILLMContextFrame
from loguru import logger
try:
from anthropic import AsyncAnthropic, NOT_GIVEN, NotGiven
from anthropic import AsyncAnthropic
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`. " +
"Also, set `ANTHROPIC_API_KEY` environment variable.")
"In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`. Also, set `ANTHROPIC_API_KEY` environment variable.")
raise Exception(f"Missing module: {e}")
@dataclass
class AnthropicImageMessageFrame(Frame):
user_image_raw_frame: UserImageRawFrame
text: Optional[str] = None
@dataclass
class AnthropicContextAggregatorPair:
_user: 'AnthropicUserContextAggregator'
_assistant: 'AnthropicAssistantContextAggregator'
def user(self) -> 'AnthropicUserContextAggregator':
return self._user
def assistant(self) -> 'AnthropicAssistantContextAggregator':
return self._assistant
class AnthropicLLMService(LLMService):
"""This class implements inference with Anthropic's AI models
This service translates internally from OpenAILLMContext to the messages format
expected by the Anthropic Python SDK. We are using the OpenAILLMContext as a lingua
franca for all LLM services, so that it is easy to switch between different LLMs.
"""
def __init__(
self,
*,
api_key: str,
model: str = "claude-3-5-sonnet-20240620",
max_tokens: int = 4096,
enable_prompt_caching_beta: bool = False,
**kwargs):
super().__init__(**kwargs)
model: str = "claude-3-opus-20240229",
max_tokens: int = 1024):
super().__init__()
self._client = AsyncAnthropic(api_key=api_key)
self._model = model
self._max_tokens = max_tokens
self._enable_prompt_caching_beta = enable_prompt_caching_beta
def can_generate_metrics(self) -> bool:
return True
@property
def enable_prompt_caching_beta(self) -> bool:
return self._enable_prompt_caching_beta
def _get_messages_from_openai_context(
self, context: OpenAILLMContext):
openai_messages = context.get_messages()
anthropic_messages = []
@staticmethod
def create_context_aggregator(context: OpenAILLMContext) -> AnthropicContextAggregatorPair:
user = AnthropicUserContextAggregator(context)
assistant = AnthropicAssistantContextAggregator(user)
return AnthropicContextAggregatorPair(
_user=user,
_assistant=assistant
)
for message in openai_messages:
role = message["role"]
text = message["content"]
if role == "system":
role = "user"
if message.get("mime_type") == "image/jpeg":
# vision frame
encoded_image = base64.b64encode(message["data"].getvalue()).decode("utf-8")
anthropic_messages.append({
"role": role,
"content": [{
"type": "image",
"source": {
"type": "base64",
"media_type": message.get("mime_type"),
"data": encoded_image,
}
}, {
"type": "text",
"text": text
}]
})
else:
# Text frame. Anthropic needs the roles to alternate. This will
# cause an issue with interruptions. So, if we detect we are the
# ones asking again it probably means we were interrupted.
if role == "user" and len(anthropic_messages) > 1:
last_message = anthropic_messages[-1]
if last_message["role"] == "user":
anthropic_messages = anthropic_messages[:-1]
content = last_message["content"]
anthropic_messages.append(
{"role": "user", "content": f"Sorry, I just asked you about [{content}] but now I would like to know [{text}]."})
else:
anthropic_messages.append({"role": role, "content": text})
else:
anthropic_messages.append({"role": role, "content": text})
return anthropic_messages
async def _process_context(self, context: OpenAILLMContext):
# Usage tracking. We track the usage reported by Anthropic in prompt_tokens and
# completion_tokens. We also estimate the completion tokens from output text
# and use that estimate if we are interrupted, because we almost certainly won't
# get a complete usage report if the task we're running in is cancelled.
prompt_tokens = 0
completion_tokens = 0
completion_tokens_estimate = 0
use_completion_tokens_estimate = False
cache_creation_input_tokens = 0
cache_read_input_tokens = 0
await self.push_frame(LLMFullResponseStartFrame())
try:
await self.push_frame(LLMFullResponseStartFrame())
await self.start_processing_metrics()
logger.debug(f"Generating chat: {context.get_messages_json()}")
logger.debug(
f"Generating chat: {context.system} | {context.get_messages_for_logging()}")
messages = context.messages
if self._enable_prompt_caching_beta:
messages = context.get_messages_with_cache_control_markers()
api_call = self._client.messages.create
if self._enable_prompt_caching_beta:
api_call = self._client.beta.prompt_caching.messages.create
messages = self._get_messages_from_openai_context(context)
await self.start_ttfb_metrics()
response = await api_call(
tools=context.tools or [],
system=context.system,
response = await self._client.messages.create(
messages=messages,
model=self._model,
max_tokens=self._max_tokens,
@@ -143,396 +114,32 @@ class AnthropicLLMService(LLMService):
await self.stop_ttfb_metrics()
# Function calling
tool_use_block = None
json_accumulator = ''
async for event in response:
# logger.debug(f"Anthropic LLM event: {event}")
# Aggregate streaming content, create frames, trigger events
if (event.type == "content_block_delta"):
if hasattr(event.delta, 'text'):
await self.push_frame(TextFrame(event.delta.text))
completion_tokens_estimate += self._estimate_tokens(event.delta.text)
elif hasattr(event.delta, 'partial_json') and tool_use_block:
json_accumulator += event.delta.partial_json
completion_tokens_estimate += self._estimate_tokens(
event.delta.partial_json)
elif (event.type == "content_block_start"):
if event.content_block.type == "tool_use":
tool_use_block = event.content_block
json_accumulator = ''
elif ((event.type == "message_delta" and
hasattr(event.delta, 'stop_reason')
and event.delta.stop_reason == 'tool_use')):
if tool_use_block:
await self.call_function(context=context,
tool_call_id=tool_use_block.id,
function_name=tool_use_block.name,
arguments=json.loads(json_accumulator) if json_accumulator else dict()
)
await self.push_frame(TextFrame(event.delta.text))
# Calculate usage. Do this here in its own if statement, because there may be usage
# data embedded in messages that we do other processing for, above.
if hasattr(event, "usage"):
prompt_tokens += event.usage.input_tokens if hasattr(
event.usage, "input_tokens") else 0
completion_tokens += event.usage.output_tokens if hasattr(
event.usage, "output_tokens") else 0
elif hasattr(event, "message") and hasattr(event.message, "usage"):
prompt_tokens += event.message.usage.input_tokens if hasattr(
event.message.usage, "input_tokens") else 0
completion_tokens += event.message.usage.output_tokens if hasattr(
event.message.usage, "output_tokens") else 0
if hasattr(event.message.usage, "cache_creation_input_tokens"):
cache_creation_input_tokens += event.message.usage.cache_creation_input_tokens
logger.debug(f"Cache creation input tokens: {cache_creation_input_tokens}")
if hasattr(event.message.usage, "cache_read_input_tokens"):
cache_read_input_tokens += event.message.usage.cache_read_input_tokens
logger.debug(f"Cache read input tokens: {cache_read_input_tokens}")
total_input_tokens = prompt_tokens + cache_creation_input_tokens + cache_read_input_tokens
if total_input_tokens >= 1024:
context.turns_above_cache_threshold += 1
except CancelledError:
# If we're interrupted, we won't get a complete usage report. So set our flag to use the
# token estimate. The reraise the exception so all the processors running in this task
# also get cancelled.
use_completion_tokens_estimate = True
raise
except Exception as e:
logger.exception(f"{self} exception: {e}")
finally:
await self.stop_processing_metrics()
await self.push_frame(LLMFullResponseEndFrame())
comp_tokens = completion_tokens if not use_completion_tokens_estimate else completion_tokens_estimate
await self._report_usage_metrics(
prompt_tokens=prompt_tokens,
completion_tokens=comp_tokens,
cache_creation_input_tokens=cache_creation_input_tokens,
cache_read_input_tokens=cache_read_input_tokens
)
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
context = None
if isinstance(frame, OpenAILLMContextFrame):
context = frame.context
context: OpenAILLMContext = frame.context
elif isinstance(frame, LLMMessagesFrame):
context = AnthropicLLMContext.from_messages(frame.messages)
context = OpenAILLMContext.from_messages(frame.messages)
elif isinstance(frame, VisionImageRawFrame):
# This is only useful in very simple pipelines because it creates
# a new context. Generally we want a context manager to catch
# UserImageRawFrames coming through the pipeline and add them
# to the context.
context = AnthropicLLMContext.from_image_frame(frame)
context = OpenAILLMContext.from_image_frame(frame)
elif isinstance(frame, LLMModelUpdateFrame):
logger.debug(f"Switching LLM model to: [{frame.model}]")
self._model = frame.model
elif isinstance(frame, LLMEnablePromptCachingFrame):
logger.debug(f"Setting enable prompt caching to: [{frame.enable}]")
self._enable_prompt_caching_beta = frame.enable
else:
await self.push_frame(frame, direction)
if context:
await self._process_context(context)
def _estimate_tokens(self, text: str) -> int:
return int(len(re.split(r'[^\w]+', text)) * 1.3)
async def _report_usage_metrics(
self,
prompt_tokens: int,
completion_tokens: int,
cache_creation_input_tokens: int,
cache_read_input_tokens: int):
if prompt_tokens or completion_tokens or cache_creation_input_tokens or cache_read_input_tokens:
tokens = {
"processor": self.name,
"model": self._model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"cache_creation_input_tokens": cache_creation_input_tokens,
"cache_read_input_tokens": cache_read_input_tokens,
"total_tokens": prompt_tokens + completion_tokens
}
await self.start_llm_usage_metrics(tokens)
class AnthropicLLMContext(OpenAILLMContext):
def __init__(
self,
messages: list[dict] | None = None,
tools: list[dict] | None = None,
tool_choice: dict | None = None,
*,
system: str | NotGiven = NOT_GIVEN
):
super().__init__(messages=messages, tools=tools, tool_choice=tool_choice)
self._user_image_request_context = {}
# For beta prompt caching. This is a counter that tracks the number of turns
# we've seen above the cache threshold. We reset this when we reset the
# messages list. We only care about this number being 0, 1, or 2. But
# it's easiest just to treat it as a counter.
self.turns_above_cache_threshold = 0
self.system = system
@classmethod
def from_openai_context(cls, openai_context: OpenAILLMContext):
self = cls(
messages=openai_context.messages,
tools=openai_context.tools,
tool_choice=openai_context.tool_choice,
)
self._restructure_from_openai_messages()
return self
@classmethod
def from_messages(cls, messages: List[dict]) -> "AnthropicLLMContext":
self = cls(messages=messages)
self._restructure_from_openai_messages()
return self
@classmethod
def from_image_frame(cls, frame: VisionImageRawFrame) -> "AnthropicLLMContext":
context = cls()
context.add_image_frame_message(
format=frame.format,
size=frame.size,
image=frame.image,
text=frame.text)
return context
def set_messages(self, messages: List):
self.turns_above_cache_threshold = 0
self._messages[:] = messages
self._restructure_from_openai_messages()
def add_image_frame_message(
self, *, format: str, size: tuple[int, int], image: bytes, text: str = None):
buffer = io.BytesIO()
Image.frombytes(format, size, image).save(buffer, format="JPEG")
encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
# Anthropic docs say that the image should be the first content block in the message.
content = [{"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": encoded_image,
}}]
if text:
content.append({"type": "text", "text": text})
self.add_message({"role": "user", "content": content})
def add_message(self, message):
try:
if self.messages:
# Anthropic requires that roles alternate. If this message's role is the same as the
# last message, we should add this message's content to the last message.
if self.messages[-1]["role"] == message["role"]:
# if the last message has just a content string, convert it to a list
# in the proper format
if isinstance(self.messages[-1]["content"], str):
self.messages[-1]["content"] = [{"type": "text",
"text": self.messages[-1]["content"]}]
# if this message has just a content string, convert it to a list
# in the proper format
if isinstance(message["content"], str):
message["content"] = [{"type": "text", "text": message["content"]}]
# append the content of this message to the last message
self.messages[-1]["content"].extend(message["content"])
else:
self.messages.append(message)
else:
self.messages.append(message)
except Exception as e:
logger.error(f"Error adding message: {e}")
def get_messages_with_cache_control_markers(self) -> List[dict]:
try:
messages = copy.deepcopy(self.messages)
if self.turns_above_cache_threshold >= 1 and messages[-1]["role"] == "user":
if isinstance(messages[-1]["content"], str):
messages[-1]["content"] = [{"type": "text", "text": messages[-1]["content"]}]
messages[-1]["content"][-1]["cache_control"] = {"type": "ephemeral"}
if (self.turns_above_cache_threshold >= 2 and
len(messages) > 2 and messages[-3]["role"] == "user"):
if isinstance(messages[-3]["content"], str):
messages[-3]["content"] = [{"type": "text", "text": messages[-3]["content"]}]
messages[-3]["content"][-1]["cache_control"] = {"type": "ephemeral"}
return messages
except Exception as e:
logger.error(f"Error adding cache control marker: {e}")
return self.messages
def _restructure_from_openai_messages(self):
# See if we should pull the system message out of our context.messages list. (For
# compatibility with Open AI messages format.)
if self.messages and self.messages[0]["role"] == "system":
if len(self.messages) == 1:
# If we have only have a system message in the list, all we can really do
# without introducing too much magic is change the role to "user".
self.messages[0]["role"] = "user"
else:
# If we have more than one message, we'll pull the system message out of the
# list.
self.system = self.messages[0]["content"]
self.messages.pop(0)
def get_messages_for_logging(self) -> str:
msgs = []
for message in self.messages:
msg = copy.deepcopy(message)
if "content" in msg:
if isinstance(msg["content"], list):
for item in msg["content"]:
if item["type"] == "image":
item["source"]["data"] = "..."
msgs.append(msg)
return json.dumps(msgs)
class AnthropicUserContextAggregator(LLMUserContextAggregator):
def __init__(self, context: OpenAILLMContext | AnthropicLLMContext):
super().__init__(context=context)
if isinstance(context, OpenAILLMContext):
self._context = AnthropicLLMContext.from_openai_context(context)
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# Our parent method has already called push_frame(). So we can't interrupt the
# flow here and we don't need to call push_frame() ourselves. Possibly something
# to talk through (tagging @aleix). At some point we might need to refactor these
# context aggregators.
try:
if isinstance(frame, UserImageRequestFrame):
# The LLM sends a UserImageRequestFrame upstream. Cache any context provided with
# that frame so we can use it when we assemble the image message in the assistant
# context aggregator.
if (frame.context):
if isinstance(frame.context, str):
self._context._user_image_request_context[frame.user_id] = frame.context
else:
logger.error(
f"Unexpected UserImageRequestFrame context type: {type(frame.context)}")
del self._context._user_image_request_context[frame.user_id]
else:
if frame.user_id in self._context._user_image_request_context:
del self._context._user_image_request_context[frame.user_id]
elif isinstance(frame, UserImageRawFrame):
# Push a new AnthropicImageMessageFrame with the text context we cached
# downstream to be handled by our assistant context aggregator. This is
# necessary so that we add the message to the context in the right order.
text = self._context._user_image_request_context.get(frame.user_id) or ""
if text:
del self._context._user_image_request_context[frame.user_id]
frame = AnthropicImageMessageFrame(user_image_raw_frame=frame, text=text)
await self.push_frame(frame)
except Exception as e:
logger.error(f"Error processing frame: {e}")
#
# Claude returns a text content block along with a tool use content block. This works quite nicely
# with streaming. We get the text first, so we can start streaming it right away. Then we get the
# tool_use block. While the text is streaming to TTS and the transport, we can run the tool call.
#
# But Claude is verbose. It would be nice to come up with prompt language that suppresses Claude's
# chattiness about it's tool thinking.
#
class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
def __init__(self, user_context_aggregator: AnthropicUserContextAggregator):
super().__init__(context=user_context_aggregator._context)
self._user_context_aggregator = user_context_aggregator
self._function_call_in_progress = None
self._function_call_result = None
self._pending_image_frame_message = None
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# See note above about not calling push_frame() here.
if isinstance(frame, StartInterruptionFrame):
self._function_call_in_progress = None
self._function_call_finished = None
elif isinstance(frame, FunctionCallInProgressFrame):
self._function_call_in_progress = frame
elif isinstance(frame, FunctionCallResultFrame):
if (self._function_call_in_progress and self._function_call_in_progress.tool_call_id ==
frame.tool_call_id):
self._function_call_in_progress = None
self._function_call_result = frame
else:
logger.warning(
"FunctionCallResultFrame tool_call_id != InProgressFrame tool_call_id")
self._function_call_in_progress = None
self._function_call_result = None
elif isinstance(frame, AnthropicImageMessageFrame):
self._pending_image_frame_message = frame
async def _push_aggregation(self):
if not self._aggregation:
return
run_llm = False
aggregation = self._aggregation
self._aggregation = ""
try:
if self._function_call_result:
frame = self._function_call_result
self._function_call_result = None
if frame.result:
self._context.add_message({
"role": "assistant",
"content": [
{
"type": "text",
"text": aggregation
},
{
"type": "tool_use",
"id": frame.tool_call_id,
"name": frame.function_name,
"input": frame.arguments
}
]
})
self._context.add_message({
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": frame.tool_call_id,
"content": json.dumps(frame.result)
}
]
})
run_llm = True
else:
self._context.add_message({"role": "assistant", "content": aggregation})
if self._pending_image_frame_message:
frame = self._pending_image_frame_message
self._pending_image_frame_message = None
self._context.add_image_frame_message(
format=frame.user_image_raw_frame.format,
size=frame.user_image_raw_frame.size,
image=frame.user_image_raw_frame.image,
text=frame.text)
run_llm = True
if run_llm:
await self._user_context_aggregator.push_context_frame()
except Exception as e:
logger.error(f"Error processing frame: {e}")

View File

@@ -7,6 +7,7 @@
import aiohttp
import asyncio
import io
import time
from PIL import Image
from typing import AsyncGenerator
@@ -17,16 +18,14 @@ from pipecat.frames.frames import (
EndFrame,
ErrorFrame,
Frame,
MetricsFrame,
StartFrame,
SystemFrame,
TTSStartedFrame,
TTSStoppedFrame,
TranscriptionFrame,
URLImageRawFrame)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import AsyncAIService, TTSService, ImageGenService
from pipecat.services.openai import BaseOpenAILLMService
from pipecat.utils.time import time_now_iso8601
from loguru import logger
@@ -107,10 +106,8 @@ class AzureTTSService(TTSService):
if result.reason == ResultReason.SynthesizingAudioCompleted:
await self.start_tts_usage_metrics(text)
await self.stop_ttfb_metrics()
await self.push_frame(TTSStartedFrame())
# Azure always sends a 44-byte header. Strip it off.
yield AudioRawFrame(audio=result.audio_data[44:], sample_rate=16000, num_channels=1)
await self.push_frame(TTSStoppedFrame())
elif result.reason == ResultReason.Canceled:
cancellation_details = result.cancellation_details
logger.warning(f"Speech synthesis canceled: {cancellation_details.reason}")
@@ -167,7 +164,7 @@ class AzureSTTService(AsyncAIService):
def _on_handle_recognized(self, event):
if event.result.reason == ResultReason.RecognizedSpeech and len(event.result.text) > 0:
frame = TranscriptionFrame(event.result.text, "", time_now_iso8601())
frame = TranscriptionFrame(event.result.text, "", int(time.time_ns() / 1000000))
asyncio.run_coroutine_threadsafe(self.queue_frame(frame), self.get_event_loop())

View File

@@ -12,22 +12,19 @@ import time
from typing import AsyncGenerator
from pipecat.processors.frame_processor import FrameDirection
from pipecat.frames.frames import (
CancelFrame,
ErrorFrame,
Frame,
AudioRawFrame,
StartInterruptionFrame,
StartFrame,
EndFrame,
TTSStartedFrame,
TTSStoppedFrame,
TextFrame,
MetricsFrame,
LLMFullResponseEndFrame
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.transcriptions.language import Language
from pipecat.services.ai_services import AsyncWordTTSService
from pipecat.services.ai_services import TTSService
from loguru import logger
@@ -41,26 +38,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_cartesia_language(language: Language) -> str | None:
match language:
case Language.DE:
return "de"
case Language.EN:
return "en"
case Language.ES:
return "es"
case Language.FR:
return "fr"
case Language.JA:
return "ja"
case Language.PT:
return "pt"
case Language.ZH:
return "zh"
return None
class CartesiaTTSService(AsyncWordTTSService):
class CartesiaTTSService(TTSService):
def __init__(
self,
@@ -74,17 +52,19 @@ class CartesiaTTSService(AsyncWordTTSService):
sample_rate: int = 16000,
language: str = "en",
**kwargs):
super().__init__(**kwargs)
# Aggregating sentences still gives cleaner-sounding results and fewer
# artifacts than streaming one word at a time. On average, waiting for a
# full sentence should only "cost" us 15ms or so with GPT-4o or a Llama
# 3 model, and it's worth it for the better audio quality.
#
# We also don't want to automatically push LLM response text frames,
# because the context aggregators will add them to the LLM context even
# if we're interrupted. Cartesia gives us word-by-word timestamps. We
# can use those to generate text frames ourselves aligned with the
# playout timing of the audio!
super().__init__(aggregate_sentences=True, push_text_frames=False, **kwargs)
# artifacts than streaming one word at a time. On average, waiting for
# a full sentence should only "cost" us 15ms or so with GPT-4o or a Llama 3
# model, and it's worth it for the better audio quality.
self._aggregate_sentences = True
# we don't want to automatically push LLM response text frames, because the
# context aggregators will add them to the LLM context even if we're
# interrupted. cartesia gives us word-by-word timestamps. we can use those
# to generate text frames ourselves aligned with the playout timing of the audio!
self._push_text_frames = False
self._api_key = api_key
self._cartesia_version = cartesia_version
@@ -100,23 +80,18 @@ class CartesiaTTSService(AsyncWordTTSService):
self._websocket = None
self._context_id = None
self._context_id_start_timestamp = None
self._timestamped_words_buffer = []
self._receive_task = None
self._context_appending_task = None
def can_generate_metrics(self) -> bool:
return True
async def set_model(self, model: str):
logger.debug(f"Switching TTS model to: [{model}]")
self._model_id = model
async def set_voice(self, voice: str):
logger.debug(f"Switching TTS voice to: [{voice}]")
self._voice_id = voice
async def set_language(self, language: Language):
logger.debug(f"Switching TTS language to: [{language}]")
self._language = language_to_cartesia_language(language)
async def start(self, frame: StartFrame):
await super().start(frame)
await self._connect()
@@ -135,6 +110,7 @@ class CartesiaTTSService(AsyncWordTTSService):
f"{self._url}?api_key={self._api_key}&cartesia_version={self._cartesia_version}"
)
self._receive_task = self.get_event_loop().create_task(self._receive_task_handler())
self._context_appending_task = self.get_event_loop().create_task(self._context_appending_task_handler())
except Exception as e:
logger.exception(f"{self} initialization error: {e}")
self._websocket = None
@@ -143,6 +119,10 @@ class CartesiaTTSService(AsyncWordTTSService):
try:
await self.stop_all_metrics()
if self._context_appending_task:
self._context_appending_task.cancel()
await self._context_appending_task
self._context_appending_task = None
if self._receive_task:
self._receive_task.cancel()
await self._receive_task
@@ -152,33 +132,18 @@ class CartesiaTTSService(AsyncWordTTSService):
self._websocket = None
self._context_id = None
self._context_id_start_timestamp = None
self._timestamped_words_buffer = []
except Exception as e:
logger.exception(f"{self} error closing websocket: {e}")
async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
await super()._handle_interruption(frame, direction)
self._context_id = None
self._context_id_start_timestamp = None
self._timestamped_words_buffer = []
await self.stop_all_metrics()
await self.push_frame(LLMFullResponseEndFrame())
self._context_id = None
async def flush_audio(self):
if not self._context_id or not self._websocket:
return
logger.debug("Flushing audio")
msg = {
"transcript": "",
"continue": False,
"context_id": self._context_id,
"model_id": self._model_id,
"voice": {
"mode": "id",
"id": self._voice_id
},
"output_format": self._output_format,
"language": self._language,
"add_timestamps": True,
}
await self._websocket.send(json.dumps(msg))
async def _receive_task_handler(self):
try:
@@ -188,32 +153,47 @@ class CartesiaTTSService(AsyncWordTTSService):
continue
if msg["type"] == "done":
await self.stop_ttfb_metrics()
await self.push_frame(TTSStoppedFrame())
# Unset _context_id but not the _context_id_start_timestamp
# because we are likely still playing out audio and need the
# timestamp to set send context frames.
self._context_id = None
await self.add_word_timestamps([("LLMFullResponseEndFrame", 0)])
self._timestamped_words_buffer.append(("LLMFullResponseEndFrame", 0))
elif msg["type"] == "timestamps":
await self.add_word_timestamps(
list(zip(msg["word_timestamps"]["words"], msg["word_timestamps"]["start"]))
# logger.debug(f"TIMESTAMPS: {msg}")
self._timestamped_words_buffer.extend(
list(zip(msg["word_timestamps"]["words"], msg["word_timestamps"]["end"]))
)
elif msg["type"] == "chunk":
await self.stop_ttfb_metrics()
self.start_word_timestamps()
if not self._context_id_start_timestamp:
self._context_id_start_timestamp = time.time()
frame = AudioRawFrame(
audio=base64.b64decode(msg["data"]),
sample_rate=self._output_format["sample_rate"],
num_channels=1
)
await self.push_frame(frame)
elif msg["type"] == "error":
logger.error(f"{self} error: {msg}")
await self.push_frame(TTSStoppedFrame())
await self.stop_all_metrics()
await self.push_error(ErrorFrame(f'{self} error: {msg["error"]}'))
else:
logger.error(f"Cartesia error, unknown message type: {msg}")
except asyncio.CancelledError:
pass
except Exception as e:
logger.exception(f"{self} exception: {e}")
async def _context_appending_task_handler(self):
try:
while True:
await asyncio.sleep(0.1)
if not self._context_id_start_timestamp:
continue
elapsed_seconds = time.time() - self._context_id_start_timestamp
# Pop all words from self._timestamped_words_buffer that are
# older than the elapsed time and print a message about them to
# the console.
while self._timestamped_words_buffer and self._timestamped_words_buffer[0][1] <= elapsed_seconds:
word, timestamp = self._timestamped_words_buffer.pop(0)
if word == "LLMFullResponseEndFrame" and timestamp == 0:
await self.push_frame(LLMFullResponseEndFrame())
continue
await self.push_frame(TextFrame(word))
except asyncio.CancelledError:
pass
except Exception as e:
@@ -227,7 +207,6 @@ class CartesiaTTSService(AsyncWordTTSService):
await self._connect()
if not self._context_id:
await self.push_frame(TTSStartedFrame())
await self.start_ttfb_metrics()
self._context_id = str(uuid.uuid4())
@@ -248,8 +227,7 @@ class CartesiaTTSService(AsyncWordTTSService):
await self._websocket.send(json.dumps(msg))
await self.start_tts_usage_metrics(text)
except Exception as e:
logger.error(f"{self} error sending message: {e}")
await self.push_frame(TTSStoppedFrame())
logger.exception(f"{self} error sending message: {e}")
await self._disconnect()
await self._connect()
return

View File

@@ -15,12 +15,12 @@ from pipecat.frames.frames import (
ErrorFrame,
Frame,
InterimTranscriptionFrame,
MetricsFrame,
StartFrame,
TTSStartedFrame,
TTSStoppedFrame,
SystemFrame,
TranscriptionFrame)
from pipecat.services.ai_services import STTService, TTSService
from pipecat.transcriptions.language import Language
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import AsyncAIService, TTSService
from pipecat.utils.time import time_now_iso8601
from loguru import logger
@@ -29,12 +29,10 @@ from loguru import logger
# See .env.example for Deepgram configuration needed
try:
from deepgram import (
AsyncListenWebSocketClient,
DeepgramClient,
DeepgramClientOptions,
LiveTranscriptionEvents,
LiveOptions,
LiveResultResponse
)
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
@@ -98,17 +96,15 @@ class DeepgramTTSService(TTSService):
await self.start_tts_usage_metrics(text)
await self.push_frame(TTSStartedFrame())
async for data in r.content:
await self.stop_ttfb_metrics()
frame = AudioRawFrame(audio=data, sample_rate=self._sample_rate, num_channels=1)
yield frame
await self.push_frame(TTSStoppedFrame())
except Exception as e:
logger.exception(f"{self} exception: {e}")
class DeepgramSTTService(STTService):
class DeepgramSTTService(AsyncAIService):
def __init__(self,
*,
api_key: str,
@@ -121,8 +117,6 @@ class DeepgramSTTService(STTService):
channels=1,
interim_results=True,
smart_format=True,
punctuate=True,
profanity_filter=True,
),
**kwargs):
super().__init__(**kwargs)
@@ -131,62 +125,40 @@ class DeepgramSTTService(STTService):
self._client = DeepgramClient(
api_key, config=DeepgramClientOptions(url=url, options={"keepalive": "true"}))
self._connection: AsyncListenWebSocketClient = self._client.listen.asyncwebsocket.v("1")
self._connection = self._client.listen.asynclive.v("1")
self._connection.on(LiveTranscriptionEvents.Transcript, self._on_message)
async def set_model(self, model: str):
logger.debug(f"Switching STT model to: [{model}]")
self._live_options.model = model
await self._disconnect()
await self._connect()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
async def set_language(self, language: Language):
logger.debug(f"Switching STT language to: [{language}]")
self._live_options.language = language
await self._disconnect()
await self._connect()
if isinstance(frame, SystemFrame):
await self.push_frame(frame, direction)
elif isinstance(frame, AudioRawFrame):
await self._connection.send(frame.audio)
else:
await self.queue_frame(frame, direction)
async def start(self, frame: StartFrame):
await super().start(frame)
await self._connect()
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._disconnect()
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
await self.start_processing_metrics()
await self._connection.send(audio)
yield None
await self.stop_processing_metrics()
async def _connect(self):
if await self._connection.start(self._live_options):
logger.debug(f"{self}: Connected to Deepgram")
else:
logger.error(f"{self}: Unable to connect to Deepgram")
async def _disconnect(self):
if self._connection.is_connected:
await self._connection.finish()
logger.debug(f"{self}: Disconnected from Deepgram")
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._connection.finish()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._connection.finish()
async def _on_message(self, *args, **kwargs):
result: LiveResultResponse = kwargs["result"]
if len(result.channel.alternatives) == 0:
return
result = kwargs["result"]
is_final = result.is_final
transcript = result.channel.alternatives[0].transcript
language = None
if result.channel.alternatives[0].languages:
language = result.channel.alternatives[0].languages[0]
language = Language(language)
if len(transcript) > 0:
if is_final:
await self.push_frame(TranscriptionFrame(transcript, "", time_now_iso8601(), language))
await self.queue_frame(TranscriptionFrame(transcript, "", time_now_iso8601()))
else:
await self.push_frame(InterimTranscriptionFrame(transcript, "", time_now_iso8601(), language))
await self.queue_frame(InterimTranscriptionFrame(transcript, "", time_now_iso8601()))

View File

@@ -4,72 +4,18 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import base64
import json
import aiohttp
from typing import Any, AsyncGenerator, List, Literal, Mapping, Tuple
from typing import AsyncGenerator, Literal
from pydantic import BaseModel
from pipecat.frames.frames import (
AudioRawFrame,
CancelFrame,
EndFrame,
Frame,
StartFrame,
StartInterruptionFrame,
TTSStartedFrame,
TTSStoppedFrame)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import AsyncWordTTSService
from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, MetricsFrame
from pipecat.services.ai_services import TTSService
from loguru import logger
# See .env.example for ElevenLabs configuration needed
try:
import websockets
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use ElevenLabs, you need to `pip install pipecat-ai[elevenlabs]`. Also, set `ELEVENLABS_API_KEY` environment variable.")
raise Exception(f"Missing module: {e}")
def sample_rate_from_output_format(output_format: str) -> int:
match output_format:
case "pcm_16000":
return 16000
case "pcm_22050":
return 22050
case "pcm_24000":
return 24000
case "pcm_44100":
return 44100
return 16000
def calculate_word_times(
alignment_info: Mapping[str, Any], cumulative_time: float
) -> List[Tuple[str, float]]:
zipped_times = list(zip(alignment_info["chars"], alignment_info["charStartTimesMs"]))
words = "".join(alignment_info["chars"]).split(" ")
# Calculate start time for each word. We do this by finding a space character
# and using the previous word time, also taking into account there might not
# be a space at the end.
times = []
for (i, (a, b)) in enumerate(zipped_times):
if a == " " or i == len(zipped_times) - 1:
t = cumulative_time + (zipped_times[i - 1][1] / 1000.0)
times.append(t)
word_times = list(zip(words, times))
return word_times
class ElevenLabsTTSService(AsyncWordTTSService):
class ElevenLabsTTSService(TTSService):
class InputParams(BaseModel):
output_format: Literal["pcm_16000", "pcm_22050", "pcm_24000", "pcm_44100"] = "pcm_16000"
@@ -78,186 +24,54 @@ class ElevenLabsTTSService(AsyncWordTTSService):
*,
api_key: str,
voice_id: str,
aiohttp_session: aiohttp.ClientSession,
model: str = "eleven_turbo_v2_5",
url: str = "wss://api.elevenlabs.io",
params: InputParams = InputParams(),
**kwargs):
# Aggregating sentences still gives cleaner-sounding results and fewer
# artifacts than streaming one word at a time. On average, waiting for a
# full sentence should only "cost" us 15ms or so with GPT-4o or a Llama
# 3 model, and it's worth it for the better audio quality.
#
# We also don't want to automatically push LLM response text frames,
# because the context aggregators will add them to the LLM context even
# if we're interrupted. ElevenLabs gives us word-by-word timestamps. We
# can use those to generate text frames ourselves aligned with the
# playout timing of the audio!
#
# Finally, ElevenLabs doesn't provide information on when the bot stops
# speaking for a while, so we want the parent class to send TTSStopFrame
# after a short period not receiving any audio.
super().__init__(
aggregate_sentences=True,
push_text_frames=False,
push_stop_frames=True,
stop_frame_timeout_s=2.0,
**kwargs
)
super().__init__(**kwargs)
self._api_key = api_key
self._voice_id = voice_id
self._model = model
self._url = url
self._params = params
self._sample_rate = sample_rate_from_output_format(params.output_format)
# Websocket connection to ElevenLabs.
self._websocket = None
# Indicates if we have sent TTSStartedFrame. It will reset to False when
# there's an interruption or TTSStoppedFrame.
self._started = False
self._cumulative_time = 0
self._aiohttp_session = aiohttp_session
def can_generate_metrics(self) -> bool:
return True
async def set_model(self, model: str):
logger.debug(f"Switching TTS model to: [{model}]")
self._model = model
await self._disconnect()
await self._connect()
async def set_voice(self, voice: str):
logger.debug(f"Switching TTS voice to: [{voice}]")
self._voice_id = voice
await self._disconnect()
await self._connect()
async def start(self, frame: StartFrame):
await super().start(frame)
await self._connect()
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._disconnect()
async def flush_audio(self):
if self._websocket:
msg = {"text": " ", "flush": True}
await self._websocket.send(json.dumps(msg))
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)):
self._started = False
if isinstance(frame, TTSStoppedFrame):
await self.add_word_timestamps([("LLMFullResponseEndFrame", 0)])
async def _connect(self):
try:
voice_id = self._voice_id
model = self._model
output_format = self._params.output_format
url = f"{self._url}/v1/text-to-speech/{voice_id}/stream-input?model_id={model}&output_format={output_format}"
self._websocket = await websockets.connect(url)
self._receive_task = self.get_event_loop().create_task(self._receive_task_handler())
self._keepalive_task = self.get_event_loop().create_task(self._keepalive_task_handler())
# According to ElevenLabs, we should always start with a single space.
msg = {
"text": " ",
"xi_api_key": self._api_key,
}
await self._websocket.send(json.dumps(msg))
except Exception as e:
logger.exception(f"{self} initialization error: {e}")
self._websocket = None
async def _disconnect(self):
try:
await self.stop_all_metrics()
if self._receive_task:
self._receive_task.cancel()
await self._receive_task
self._receive_task = None
if self._keepalive_task:
self._keepalive_task.cancel()
await self._keepalive_task
self._keepalive_task = None
if self._websocket:
await self._websocket.close()
self._websocket = None
self._started = False
except Exception as e:
logger.exception(f"{self} error closing websocket: {e}")
async def _receive_task_handler(self):
try:
async for message in self._websocket:
msg = json.loads(message)
if msg.get("audio"):
await self.stop_ttfb_metrics()
self.start_word_timestamps()
audio = base64.b64decode(msg["audio"])
frame = AudioRawFrame(audio, self._sample_rate, 1)
await self.push_frame(frame)
if msg.get("alignment"):
word_times = calculate_word_times(msg["alignment"], self._cumulative_time)
await self.add_word_timestamps(word_times)
self._cumulative_time = word_times[-1][1]
except asyncio.CancelledError:
pass
except Exception as e:
logger.exception(f"{self} exception: {e}")
async def _keepalive_task_handler(self):
while True:
try:
await asyncio.sleep(10)
await self._send_text("")
except asyncio.CancelledError:
break
except Exception as e:
logger.exception(f"{self} exception: {e}")
async def _send_text(self, text: str):
if self._websocket:
msg = {"text": text + " "}
await self._websocket.send(json.dumps(msg))
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating TTS: [{text}]")
try:
if not self._websocket:
await self._connect()
url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream"
try:
if not self._started:
await self.push_frame(TTSStartedFrame())
await self.start_ttfb_metrics()
self._started = True
self._cumulative_time = 0
payload = {"text": text, "model_id": self._model}
await self._send_text(text)
await self.start_tts_usage_metrics(text)
except Exception as e:
logger.error(f"{self} error sending message: {e}")
await self.push_frame(TTSStoppedFrame())
await self._disconnect()
await self._connect()
querystring = {
"output_format": self._params.output_format
}
headers = {
"xi-api-key": self._api_key,
"Content-Type": "application/json",
}
await self.start_ttfb_metrics()
async with self._aiohttp_session.post(url, json=payload, headers=headers, params=querystring) as r:
if r.status != 200:
text = await r.text()
logger.error(f"{self} error getting audio (status: {r.status}, error: {text})")
yield ErrorFrame(f"Error getting audio (status: {r.status}, error: {text})")
return
yield None
except Exception as e:
logger.exception(f"{self} exception: {e}")
await self.start_tts_usage_metrics(text)
async for chunk in r.content:
if len(chunk) > 0:
await self.stop_ttfb_metrics()
frame = AudioRawFrame(chunk, 16000, 1)
yield frame

View File

@@ -1,166 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
from typing import AsyncGenerator
from pipecat.processors.frame_processor import FrameDirection
from pipecat.frames.frames import (
AudioRawFrame,
CancelFrame,
EndFrame,
ErrorFrame,
Frame,
StartFrame,
StartInterruptionFrame,
TTSStartedFrame,
TTSStoppedFrame,
)
from pipecat.services.ai_services import AsyncTTSService
from loguru import logger
# See .env.example for LMNT configuration needed
try:
from lmnt.api import Speech
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use LMNT, you need to `pip install pipecat-ai[lmnt]`. Also, set `LMNT_API_KEY` environment variable.")
raise Exception(f"Missing module: {e}")
class LmntTTSService(AsyncTTSService):
def __init__(
self,
*,
api_key: str,
voice_id: str,
sample_rate: int = 24000,
language: str = "en",
**kwargs):
# Let TTSService produce TTSStoppedFrames after a short delay of
# no activity.
super().__init__(push_stop_frames=True, **kwargs)
self._api_key = api_key
self._voice_id = voice_id
self._output_format = {
"container": "raw",
"encoding": "pcm_s16le",
"sample_rate": sample_rate,
}
self._language = language
self._speech = None
self._connection = None
self._receive_task = None
# Indicates if we have sent TTSStartedFrame. It will reset to False when
# there's an interruption or TTSStoppedFrame.
self._started = False
def can_generate_metrics(self) -> bool:
return True
async def set_voice(self, voice: str):
logger.debug(f"Switching TTS voice to: [{voice}]")
self._voice_id = voice
async def start(self, frame: StartFrame):
await super().start(frame)
await self._connect()
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._disconnect()
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)):
self._started = False
async def _connect(self):
try:
self._speech = Speech()
self._connection = await self._speech.synthesize_streaming(
self._voice_id, format="raw", sample_rate=self._output_format["sample_rate"])
self._receive_task = self.get_event_loop().create_task(self._receive_task_handler())
except Exception as e:
logger.exception(f"{self} initialization error: {e}")
self._connection = None
async def _disconnect(self):
try:
await self.stop_all_metrics()
if self._receive_task:
self._receive_task.cancel()
await self._receive_task
self._receive_task = None
if self._connection:
await self._connection.socket.close()
self._connection = None
if self._speech:
await self._speech.close()
self._speech = None
self._started = False
except Exception as e:
logger.exception(f"{self} error closing websocket: {e}")
async def _receive_task_handler(self):
try:
async for msg in self._connection:
if "error" in msg:
logger.error(f'{self} error: {msg["error"]}')
await self.push_frame(TTSStoppedFrame())
await self.stop_all_metrics()
await self.push_error(ErrorFrame(f'{self} error: {msg["error"]}'))
elif "audio" in msg:
await self.stop_ttfb_metrics()
frame = AudioRawFrame(
audio=msg["audio"],
sample_rate=self._output_format["sample_rate"],
num_channels=1
)
await self.push_frame(frame)
else:
logger.error(f"LMNT error, unknown message type: {msg}")
except asyncio.CancelledError:
pass
except Exception as e:
logger.exception(f"{self} exception: {e}")
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating TTS: [{text}]")
try:
if not self._connection:
await self._connect()
if not self._started:
await self.push_frame(TTSStartedFrame())
await self.start_ttfb_metrics()
self._started = True
try:
await self._connection.append_text(text)
await self._connection.flush()
await self.start_tts_usage_metrics(text)
except Exception as e:
logger.error(f"{self} error sending message: {e}")
await self.push_frame(TTSStoppedFrame())
await self._disconnect()
await self._connect()
return
yield None
except Exception as e:
logger.exception(f"{self} exception: {e}")

View File

@@ -9,7 +9,6 @@ import base64
import io
import json
import httpx
from dataclasses import dataclass
from typing import AsyncGenerator, List, Literal
@@ -24,17 +23,11 @@ from pipecat.frames.frames import (
LLMFullResponseStartFrame,
LLMMessagesFrame,
LLMModelUpdateFrame,
TTSStartedFrame,
TTSStoppedFrame,
MetricsFrame,
TextFrame,
URLImageRawFrame,
VisionImageRawFrame,
FunctionCallResultFrame,
FunctionCallInProgressFrame,
StartInterruptionFrame
VisionImageRawFrame
)
from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator, LLMAssistantContextAggregator
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame
@@ -48,7 +41,12 @@ from pipecat.services.ai_services import (
try:
from openai import AsyncOpenAI, AsyncStream, DefaultAsyncHttpxClient, BadRequestError
from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
from openai.types.chat import (
ChatCompletionChunk,
ChatCompletionFunctionMessageParam,
ChatCompletionMessageParam,
ChatCompletionToolParam
)
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
@@ -139,7 +137,6 @@ class BaseOpenAILLMService(LLMService):
if chunk.usage:
tokens = {
"processor": self.name,
"model": self._model,
"prompt_tokens": chunk.usage.prompt_tokens,
"completion_tokens": chunk.usage.completion_tokens,
"total_tokens": chunk.usage.total_tokens
@@ -167,7 +164,7 @@ class BaseOpenAILLMService(LLMService):
if tool_call.function and tool_call.function.name:
function_name += tool_call.function.name
tool_call_id = tool_call.id
await self.call_start_function(context, function_name)
await self.call_start_function(function_name)
if tool_call.function and tool_call.function.arguments:
# Keep iterating through the response to collect all the argument fragments
arguments += tool_call.function.arguments
@@ -193,12 +190,44 @@ class BaseOpenAILLMService(LLMService):
arguments
):
arguments = json.loads(arguments)
await self.call_function(
context=context,
tool_call_id=tool_call_id,
function_name=function_name,
arguments=arguments
)
result = await self.call_function(function_name, arguments)
arguments = json.dumps(arguments)
if isinstance(result, (str, dict)):
# Handle it in "full magic mode"
tool_call = ChatCompletionFunctionMessageParam({
"role": "assistant",
"tool_calls": [
{
"id": tool_call_id,
"function": {
"arguments": arguments,
"name": function_name
},
"type": "function"
}
]
})
context.add_message(tool_call)
if isinstance(result, dict):
result = json.dumps(result)
tool_result = ChatCompletionToolParam({
"tool_call_id": tool_call_id,
"role": "tool",
"content": result
})
context.add_message(tool_result)
# re-prompt to get a human answer
await self._process_context(context)
elif isinstance(result, list):
# reduced magic
for msg in result:
context.add_message(msg)
await self._process_context(context)
elif isinstance(result, type(None)):
pass
else:
raise TypeError(f"Unknown return type from function callback: {type(result)}")
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -224,32 +253,11 @@ class BaseOpenAILLMService(LLMService):
await self.push_frame(LLMFullResponseEndFrame())
@dataclass
class OpenAIContextAggregatorPair:
_user: 'OpenAIUserContextAggregator'
_assistant: 'OpenAIAssistantContextAggregator'
def user(self) -> 'OpenAIUserContextAggregator':
return self._user
def assistant(self) -> 'OpenAIAssistantContextAggregator':
return self._assistant
class OpenAILLMService(BaseOpenAILLMService):
def __init__(self, *, model: str = "gpt-4o", **kwargs):
super().__init__(model=model, **kwargs)
@staticmethod
def create_context_aggregator(context: OpenAILLMContext) -> OpenAIContextAggregatorPair:
user = OpenAIUserContextAggregator(context)
assistant = OpenAIAssistantContextAggregator(user)
return OpenAIContextAggregatorPair(
_user=user,
_assistant=assistant
)
class OpenAIImageGenService(ImageGenService):
@@ -344,87 +352,10 @@ class OpenAITTSService(TTSService):
await self.start_tts_usage_metrics(text)
await self.push_frame(TTSStartedFrame())
async for chunk in r.iter_bytes(8192):
if len(chunk) > 0:
await self.stop_ttfb_metrics()
frame = AudioRawFrame(chunk, 24_000, 1)
yield frame
await self.push_frame(TTSStoppedFrame())
except BadRequestError as e:
logger.exception(f"{self} error generating TTS: {e}")
class OpenAIUserContextAggregator(LLMUserContextAggregator):
def __init__(self, context: OpenAILLMContext):
super().__init__(context=context)
class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
def __init__(self, user_context_aggregator: OpenAIUserContextAggregator):
super().__init__(context=user_context_aggregator._context)
self._user_context_aggregator = user_context_aggregator
self._function_call_in_progress = None
self._function_call_result = None
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# See note above about not calling push_frame() here.
if isinstance(frame, StartInterruptionFrame):
self._function_call_in_progress = None
self._function_call_finished = None
elif isinstance(frame, FunctionCallInProgressFrame):
self._function_call_in_progress = frame
elif isinstance(frame, FunctionCallResultFrame):
if self._function_call_in_progress and self._function_call_in_progress.tool_call_id == frame.tool_call_id:
self._function_call_in_progress = None
self._function_call_result = frame
# TODO-CB: Kwin wants us to refactor this out of here but I REFUSE
await self._push_aggregation()
else:
logger.warning(
f"FunctionCallResultFrame tool_call_id does not match FunctionCallInProgressFrame tool_call_id")
self._function_call_in_progress = None
self._function_call_result = None
async def _push_aggregation(self):
if not (self._aggregation or self._function_call_result):
return
run_llm = False
aggregation = self._aggregation
self._aggregation = ""
try:
if self._function_call_result:
frame = self._function_call_result
self._function_call_result = None
if frame.result:
self._context.add_message({
"role": "assistant",
"tool_calls": [
{
"id": frame.tool_call_id,
"function": {
"name": frame.function_name,
"arguments": json.dumps(frame.arguments)
},
"type": "function"
}
]
})
self._context.add_message({
"role": "tool",
"content": json.dumps(frame.result),
"tool_call_id": frame.tool_call_id
})
run_llm = True
else:
self._context.add_message({"role": "assistant", "content": aggregation})
if run_llm:
await self._user_context_aggregator.push_context_frame()
except Exception as e:
logger.error(f"Error processing frame: {e}")

View File

@@ -9,7 +9,7 @@ import struct
from typing import AsyncGenerator
from pipecat.frames.frames import AudioRawFrame, Frame, TTSStartedFrame, TTSStoppedFrame
from pipecat.frames.frames import AudioRawFrame, Frame, MetricsFrame
from pipecat.services.ai_services import TTSService
from loguru import logger
@@ -62,7 +62,6 @@ class PlayHTTTSService(TTSService):
await self.start_tts_usage_metrics(text)
await self.push_frame(TTSStartedFrame())
async for chunk in playht_gen:
# skip the RIFF header.
if in_header:
@@ -82,6 +81,5 @@ class PlayHTTTSService(TTSService):
await self.stop_ttfb_metrics()
frame = AudioRawFrame(chunk, 16000, 1)
yield frame
await self.push_frame(TTSStoppedFrame())
except Exception as e:
logger.exception(f"{self} error generating TTS: {e}")

View File

@@ -1,315 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import base64
import json
import io
import copy
from typing import List, Optional
from dataclasses import dataclass
from asyncio import CancelledError
import re
import uuid
from pipecat.frames.frames import (
Frame,
LLMModelUpdateFrame,
TextFrame,
VisionImageRawFrame,
UserImageRequestFrame,
UserImageRawFrame,
LLMMessagesFrame,
LLMFullResponseStartFrame,
LLMFullResponseEndFrame,
FunctionCallResultFrame,
FunctionCallInProgressFrame,
StartInterruptionFrame
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import LLMService
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext, OpenAILLMContextFrame
from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator, LLMAssistantContextAggregator
from loguru import logger
try:
from together import AsyncTogether
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use Together.ai, you need to `pip install pipecat-ai[together]`. Also, set `TOGETHER_API_KEY` environment variable.")
raise Exception(f"Missing module: {e}")
@dataclass
class TogetherContextAggregatorPair:
_user: 'TogetherUserContextAggregator'
_assistant: 'TogetherAssistantContextAggregator'
def user(self) -> 'TogetherUserContextAggregator':
return self._user
def assistant(self) -> 'TogetherAssistantContextAggregator':
return self._assistant
class TogetherLLMService(LLMService):
"""This class implements inference with Together's Llama 3.1 models
"""
def __init__(
self,
*,
api_key: str,
model: str = "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo",
max_tokens: int = 4096,
**kwargs):
super().__init__(**kwargs)
self._client = AsyncTogether(api_key=api_key)
self._model = model
self._max_tokens = max_tokens
def can_generate_metrics(self) -> bool:
return True
@staticmethod
def create_context_aggregator(context: OpenAILLMContext) -> TogetherContextAggregatorPair:
user = TogetherUserContextAggregator(context)
assistant = TogetherAssistantContextAggregator(user)
return TogetherContextAggregatorPair(
_user=user,
_assistant=assistant
)
async def _process_context(self, context: OpenAILLMContext):
try:
await self.push_frame(LLMFullResponseStartFrame())
await self.start_processing_metrics()
logger.debug(f"Generating chat: {context.get_messages_for_logging()}")
await self.start_ttfb_metrics()
stream = await self._client.chat.completions.create(
messages=context.messages,
model=self._model,
max_tokens=self._max_tokens,
stream=True,
)
# Function calling
got_first_chunk = False
accumulating_function_call = False
function_call_accumulator = ""
async for chunk in stream:
# logger.debug(f"Together LLM event: {chunk}")
if chunk.usage:
tokens = {
"processor": self.name,
"model": self._model,
"prompt_tokens": chunk.usage.prompt_tokens,
"completion_tokens": chunk.usage.completion_tokens,
"total_tokens": chunk.usage.total_tokens
}
await self.start_llm_usage_metrics(tokens)
if len(chunk.choices) == 0:
continue
if not got_first_chunk:
await self.stop_ttfb_metrics()
if chunk.choices[0].delta.content:
got_first_chunk = True
if chunk.choices[0].delta.content[0] == "<":
accumulating_function_call = True
if chunk.choices[0].delta.content:
if accumulating_function_call:
function_call_accumulator += chunk.choices[0].delta.content
else:
await self.push_frame(TextFrame(chunk.choices[0].delta.content))
if chunk.choices[0].finish_reason == 'eos' and accumulating_function_call:
await self._extract_function_call(context, function_call_accumulator)
except CancelledError as e:
# todo: implement token counting estimates for use when the user interrupts a long generation
# we do this in the anthropic.py service
raise
except Exception as e:
logger.exception(f"{self} exception: {e}")
finally:
await self.stop_processing_metrics()
await self.push_frame(LLMFullResponseEndFrame())
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
context = None
if isinstance(frame, OpenAILLMContextFrame):
context = frame.context
elif isinstance(frame, LLMMessagesFrame):
context = TogetherLLMContext.from_messages(frame.messages)
elif isinstance(frame, LLMModelUpdateFrame):
logger.debug(f"Switching LLM model to: [{frame.model}]")
self._model = frame.model
else:
await self.push_frame(frame, direction)
if context:
await self._process_context(context)
async def _extract_function_call(self, context, function_call_accumulator):
context.add_message({"role": "assistant", "content": function_call_accumulator})
function_regex = r"<function=(\w+)>(.*?)</function>"
match = re.search(function_regex, function_call_accumulator)
if match:
function_name, args_string = match.groups()
try:
arguments = json.loads(args_string)
await self.call_function(context=context,
tool_call_id=str(uuid.uuid4()),
function_name=function_name,
arguments=arguments)
return
except json.JSONDecodeError as error:
# We get here if the LLM returns a function call with invalid JSON arguments. This could happen
# because of LLM non-determinism, or maybe more often because of user error in the prompt.
# Should we do anything more than log a warning?
logger.debug(f"Error parsing function arguments: {error}")
class TogetherLLMContext(OpenAILLMContext):
def __init__(
self,
messages: list[dict] | None = None,
):
super().__init__(messages=messages)
@classmethod
def from_openai_context(cls, openai_context: OpenAILLMContext):
self = cls(
messages=openai_context.messages,
)
return self
@classmethod
def from_messages(cls, messages: List[dict]) -> "TogetherLLMContext":
return cls(messages=messages)
def add_message(self, message):
try:
self.messages.append(message)
except Exception as e:
logger.error(f"Error adding message: {e}")
def get_messages_for_logging(self) -> str:
return json.dumps(self.messages)
class TogetherUserContextAggregator(LLMUserContextAggregator):
def __init__(self, context: OpenAILLMContext | TogetherLLMContext):
super().__init__(context=context)
if isinstance(context, OpenAILLMContext):
self._context = TogetherLLMContext.from_openai_context(context)
async def push_messages_frame(self):
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# Our parent method has already called push_frame(). So we can't interrupt the
# flow here and we don't need to call push_frame() ourselves. Possibly something
# to talk through (tagging @aleix). At some point we might need to refactor these
# context aggregators.
try:
if isinstance(frame, UserImageRequestFrame):
# The LLM sends a UserImageRequestFrame upstream. Cache any context provided with
# that frame so we can use it when we assemble the image message in the assistant
# context aggregator.
if (frame.context):
if isinstance(frame.context, str):
self._context._user_image_request_context[frame.user_id] = frame.context
else:
logger.error(
f"Unexpected UserImageRequestFrame context type: {type(frame.context)}")
del self._context._user_image_request_context[frame.user_id]
else:
if frame.user_id in self._context._user_image_request_context:
del self._context._user_image_request_context[frame.user_id]
except Exception as e:
logger.error(f"Error processing frame: {e}")
#
# Claude returns a text content block along with a tool use content block. This works quite nicely
# with streaming. We get the text first, so we can start streaming it right away. Then we get the
# tool_use block. While the text is streaming to TTS and the transport, we can run the tool call.
#
# But Claude is verbose. It would be nice to come up with prompt language that suppresses Claude's
# chattiness about it's tool thinking.
#
class TogetherAssistantContextAggregator(LLMAssistantContextAggregator):
def __init__(self, user_context_aggregator: TogetherUserContextAggregator):
super().__init__(context=user_context_aggregator._context)
self._user_context_aggregator = user_context_aggregator
self._function_call_in_progress = None
self._function_call_result = None
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# See note above about not calling push_frame() here.
if isinstance(frame, StartInterruptionFrame):
self._function_call_in_progress = None
self._function_call_finished = None
elif isinstance(frame, FunctionCallInProgressFrame):
self._function_call_in_progress = frame
elif isinstance(frame, FunctionCallResultFrame):
if self._function_call_in_progress and self._function_call_in_progress.tool_call_id == frame.tool_call_id:
self._function_call_in_progress = None
self._function_call_result = frame
await self._push_aggregation()
else:
logger.warning(
f"FunctionCallResultFrame tool_call_id does not match FunctionCallInProgressFrame tool_call_id")
self._function_call_in_progress = None
self._function_call_result = None
def add_message(self, message):
self._user_context_aggregator.add_message(message)
async def _push_aggregation(self):
if not (self._aggregation or self._function_call_result):
return
run_llm = False
aggregation = self._aggregation
self._aggregation = ""
try:
if self._function_call_result:
frame = self._function_call_result
self._function_call_result = None
self._context.add_message({
"role": "tool",
# Together expects the content here to be a string, so stringify it
"content": str(frame.result)
})
run_llm = True
else:
self._context.add_message({"role": "assistant", "content": aggregation})
if run_llm:
await self._user_context_aggregator.push_messages_frame()
except Exception as e:
logger.error(f"Error processing frame: {e}")

View File

@@ -9,12 +9,12 @@
import asyncio
from enum import Enum
from typing import AsyncGenerator
from typing_extensions import AsyncGenerator
import numpy as np
from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
from pipecat.services.ai_services import SegmentedSTTService
from pipecat.services.ai_services import STTService
from pipecat.utils.time import time_now_iso8601
from loguru import logger
@@ -38,7 +38,7 @@ class Model(Enum):
DISTIL_MEDIUM_EN = "Systran/faster-distil-whisper-medium.en"
class WhisperSTTService(SegmentedSTTService):
class WhisperSTTService(STTService):
"""Class to transcribe audio with a locally-downloaded Whisper model"""
def __init__(self,
@@ -77,7 +77,6 @@ class WhisperSTTService(SegmentedSTTService):
yield ErrorFrame("Whisper model not available")
return
await self.start_processing_metrics()
await self.start_ttfb_metrics()
# Divide by 32768 because we have signed 16-bit data.
@@ -89,9 +88,7 @@ class WhisperSTTService(SegmentedSTTService):
if segment.no_speech_prob < self._no_speech_prob:
text += f"{segment.text} "
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
if text:
await self.stop_ttfb_metrics()
logger.debug(f"Transcription: [{text}]")
yield TranscriptionFrame(text, "", time_now_iso8601())

View File

@@ -8,13 +8,7 @@ import aiohttp
from typing import Any, AsyncGenerator, Dict
from pipecat.frames.frames import (
AudioRawFrame,
ErrorFrame,
Frame,
StartFrame,
TTSStartedFrame,
TTSStoppedFrame)
from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, MetricsFrame, StartFrame
from pipecat.services.ai_services import TTSService
from loguru import logger
@@ -105,9 +99,8 @@ class XTTSService(TTSService):
await self.start_tts_usage_metrics(text)
await self.push_frame(TTSStartedFrame())
buffer = bytearray()
async for chunk in r.content.iter_chunked(1024):
if len(chunk) > 0:
await self.stop_ttfb_metrics()
@@ -138,5 +131,3 @@ class XTTSService(TTSService):
resampled_audio_bytes = resampled_audio.astype(np.int16).tobytes()
frame = AudioRawFrame(resampled_audio_bytes, 16000, 1)
yield frame
await self.push_frame(TTSStoppedFrame())

View File

@@ -1,64 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import sys
from enum import Enum
if sys.version_info < (3, 11):
class StrEnum(str, Enum):
def __new__(cls, value):
obj = str.__new__(cls, value)
obj._value_ = value
return obj
else:
from enum import StrEnum
class Language(StrEnum):
BG = "bg" # Bulgarian
CA = "ca" # Catalan
ZH = "zh" # Chinese simplified
ZH_TW = "zh-TW" # Chinese traditional
CS = "cs" # Czech
DA = "da" # Danish
NL = "nl" # Dutch
EN = "en" # English
EN_US = "en-US" # English (USA)
EN_AU = "en-AU" # English (Australia)
EN_GB = "en-GB" # English (Great Britain)
EN_NZ = "en-NZ" # English (New Zealand)
EN_IN = "en-IN" # English (India)
ET = "et" # Estonian
FI = "fi" # Finnish
NL_BE = "nl-BE" # Flemmish
FR = "fr" # French
FR_CA = "fr-CA" # French (Canada)
DE = "de" # German
DE_CH = "de-CH" # German (Switzerland)
EL = "el" # Greek
HI = "hi" # Hindi
HU = "hu" # Hungarian
ID = "id" # Indonesian
IT = "it" # Italian
JA = "ja" # Japanese
KO = "ko" # Korean
LV = "lv" # Latvian
LT = "lt" # Lithuanian
MS = "ms" # Malay
NO = "no" # Norwegian
PL = "pl" # Polish
PT = "pt" # Portuguese
PT_BR = "pt-BR" # Portuguese (Brazil)
RO = "ro" # Romanian
RU = "ru" # Russian
SK = "sk" # Slovak
ES = "es" # Spanish
SV = "sv" # Swedish
TH = "th" # Thai
TR = "tr" # Turkish
UK = "uk" # Ukrainian
VI = "vi" # Vietnamese

View File

@@ -20,8 +20,7 @@ from pipecat.frames.frames import (
StopInterruptionFrame,
SystemFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
VADParamsUpdateFrame)
UserStoppedSpeakingFrame)
from pipecat.transports.base_transport import TransportParams
from pipecat.vad.vad_analyzer import VADAnalyzer, VADState
@@ -96,19 +95,13 @@ class BaseInputTransport(FrameProcessor):
await self.push_frame(frame, direction)
# Control frames
elif isinstance(frame, StartFrame):
# Push StartFrame before start(), because we want StartFrame to be
# processed by every processor before any other frame is processed.
await self._internal_push_frame(frame, direction)
await self.start(frame)
await self._internal_push_frame(frame, direction)
elif isinstance(frame, EndFrame):
# Push EndFrame before stop(), because stop() waits on the task to
# finish and the task finishes when EndFrame is processed.
await self._internal_push_frame(frame, direction)
await self.stop(frame)
elif isinstance(frame, VADParamsUpdateFrame):
vad_analyzer = self.vad_analyzer()
if vad_analyzer:
vad_analyzer.set_params(frame.params)
# Other frames
else:
await self._internal_push_frame(frame, direction)

View File

@@ -8,7 +8,6 @@
import asyncio
import itertools
import time
import sys
from PIL import Image
from typing import List
@@ -31,14 +30,11 @@ from pipecat.frames.frames import (
SystemFrame,
TTSStartedFrame,
TTSStoppedFrame,
TextFrame,
TransportMessageFrame)
from pipecat.transports.base_transport import TransportParams
from loguru import logger
from pipecat.utils.time import nanoseconds_to_seconds
class BaseOutputTransport(FrameProcessor):
@@ -60,15 +56,10 @@ class BaseOutputTransport(FrameProcessor):
self._stopped_event = asyncio.Event()
# Indicates if the bot is currently speaking. This is useful when we
# have an interruption since all the queued messages will be thrown
# away and we would lose the TTSStoppedFrame.
self._bot_speaking = False
# Create sink frame task. This is the task that will actually write
# audio or video frames. We write audio/video in a task so we can keep
# generating frames upstream while, for example, the audio is playing.
self._create_sink_tasks()
self._create_sink_task()
# Create push frame task. This is the task that will push frames in
# order. We also guarantee that all frames are pushed in the same task.
@@ -138,8 +129,8 @@ class BaseOutputTransport(FrameProcessor):
# queue.
#
if isinstance(frame, CancelFrame):
await self.cancel(frame)
await self.push_frame(frame, direction)
await self.cancel(frame)
elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame):
await self.push_frame(frame, direction)
await self._handle_interruptions(frame)
@@ -153,7 +144,6 @@ class BaseOutputTransport(FrameProcessor):
await self._sink_queue.put(frame)
await self.start(frame)
elif isinstance(frame, EndFrame):
await self._sink_clock_queue.put((sys.maxsize, frame.id, frame))
await self._sink_queue.put(frame)
await self.stop(frame)
# Other frames.
@@ -161,11 +151,6 @@ class BaseOutputTransport(FrameProcessor):
await self._handle_audio(frame)
elif isinstance(frame, ImageRawFrame) or isinstance(frame, SpriteFrame):
await self._handle_image(frame)
elif isinstance(frame, TransportMessageFrame) and frame.urgent:
await self.send_message(frame)
# TODO(aleix): Images and audio should support presentation timestamps.
elif frame.pts:
await self._sink_clock_queue.put((frame.pts, frame.id, frame))
else:
await self._sink_queue.put(frame)
@@ -174,21 +159,14 @@ class BaseOutputTransport(FrameProcessor):
return
if isinstance(frame, StartInterruptionFrame):
# Stop sink tasks.
# Stop sink task.
self._sink_task.cancel()
await self._sink_task
# Stop sink clock tasks.
self._sink_clock_task.cancel()
await self._sink_clock_task
# Create sink tasks.
self._create_sink_tasks()
self._create_sink_task()
# Stop push task.
self._push_frame_task.cancel()
await self._push_frame_task
self._create_push_task()
# Let's send a bot stopped speaking if we have to.
if self._bot_speaking:
await self._bot_stopped_speaking()
async def _handle_audio(self, frame: AudioRawFrame):
if not self._params.audio_out_enabled:
@@ -213,93 +191,43 @@ class BaseOutputTransport(FrameProcessor):
else:
await self._sink_queue.put(frame)
#
# Sink tasks
#
def _create_sink_tasks(self):
def _create_sink_task(self):
loop = self.get_event_loop()
self._sink_queue = asyncio.Queue()
self._sink_task = loop.create_task(self._sink_task_handler())
self._sink_clock_queue = asyncio.PriorityQueue()
self._sink_clock_task = loop.create_task(self._sink_clock_task_handler())
async def _sink_frame_handler(self, frame: Frame):
if isinstance(frame, AudioRawFrame):
await self.write_raw_audio_frames(frame.audio)
await self._internal_push_frame(frame)
await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM)
elif isinstance(frame, ImageRawFrame):
await self._set_camera_image(frame)
elif isinstance(frame, SpriteFrame):
await self._set_camera_images(frame.images)
elif isinstance(frame, TransportMessageFrame):
await self.send_message(frame)
elif isinstance(frame, TTSStartedFrame):
await self._bot_started_speaking()
await self._internal_push_frame(frame)
elif isinstance(frame, TTSStoppedFrame):
await self._bot_stopped_speaking()
await self._internal_push_frame(frame)
else:
await self._internal_push_frame(frame)
async def _sink_task_handler(self):
running = True
while running:
try:
frame = await self._sink_queue.get()
await self._sink_frame_handler(frame)
if isinstance(frame, AudioRawFrame):
await self.write_raw_audio_frames(frame.audio)
await self._internal_push_frame(frame)
await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM)
elif isinstance(frame, ImageRawFrame):
await self._set_camera_image(frame)
elif isinstance(frame, SpriteFrame):
await self._set_camera_images(frame.images)
elif isinstance(frame, TransportMessageFrame):
await self.send_message(frame)
elif isinstance(frame, TTSStartedFrame):
await self._internal_push_frame(BotStartedSpeakingFrame(), FrameDirection.UPSTREAM)
await self._internal_push_frame(frame)
elif isinstance(frame, TTSStoppedFrame):
await self._internal_push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM)
await self._internal_push_frame(frame)
else:
await self._internal_push_frame(frame)
running = not isinstance(frame, EndFrame)
self._sink_queue.task_done()
except asyncio.CancelledError:
break
except Exception as e:
logger.exception(f"{self} error processing sink queue: {e}")
async def _sink_clock_frame_handler(self, frame: Frame):
# TODO(aleix): For now we just process TextFrame. But we should process
# audio and video as well.
if isinstance(frame, TextFrame):
await self._internal_push_frame(frame)
async def _sink_clock_task_handler(self):
running = True
while running:
try:
timestamp, _, frame = await self._sink_clock_queue.get()
# If we hit an EndFrame, we cna finish right away.
running = not isinstance(frame, EndFrame)
# If we have a frame we check it's presentation timestamp. If it
# has already passed we process it, otherwise we wait until it's
# time to process it.
if running:
current_time = self.get_clock().get_time()
if timestamp <= current_time:
await self._sink_clock_frame_handler(frame)
else:
wait_time = nanoseconds_to_seconds(timestamp - current_time)
await asyncio.sleep(wait_time)
await self._sink_frame_handler(frame)
self._sink_clock_queue.task_done()
except asyncio.CancelledError:
break
except Exception as e:
logger.exception(f"{self} error processing sink clock queue: {e}")
async def _bot_started_speaking(self):
logger.debug("Bot started speaking")
self._bot_speaking = True
await self._internal_push_frame(BotStartedSpeakingFrame(), FrameDirection.UPSTREAM)
async def _bot_stopped_speaking(self):
logger.debug("Bot stopped speaking")
self._bot_speaking = False
await self._internal_push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM)
#
# Push frames task
#

View File

@@ -32,7 +32,6 @@ class TransportParams(BaseModel):
audio_out_is_live: bool = False
audio_out_sample_rate: int = 16000
audio_out_channels: int = 1
audio_out_bitrate: int = 96000
audio_in_enabled: bool = False
audio_in_sample_rate: int = 16000
audio_in_channels: int = 1

View File

@@ -24,7 +24,7 @@ try:
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use local audio, you need to `pip install pipecat-ai[local]`. On MacOS, you also need to `brew install portaudio`.")
"In order to use local audio, you need to `pip install pipecat-ai[audio]`. On MacOS, you also need to `brew install portaudio`.")
raise Exception(f"Missing module: {e}")
try:

View File

@@ -1,117 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import json
import io
import wave
from typing import Awaitable, Callable
from pydantic.main import BaseModel
from pipecat.frames.frames import AudioRawFrame, CancelFrame, EndFrame, Frame, StartFrame, StartInterruptionFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.serializers.base_serializer import FrameSerializer
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams
from loguru import logger
try:
from fastapi import Request, Response
from starlette.background import BackgroundTask
from sse_starlette.sse import EventSourceResponse
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use FastAPI HTTP SSE, you need to `pip install pipecat-ai[http]`.")
raise Exception(f"Missing module: {e}")
class FastAPIHTTPParams(TransportParams):
serializer: FrameSerializer
class FastAPIHTTPInputTransport(BaseInputTransport):
def __init__(
self,
params: FastAPIHTTPParams,
**kwargs):
super().__init__(params, **kwargs)
self._params = params
self._request = None
# todo: this should probably expect a list of frames, not just one frame
async def handle_request(self, request: Request):
self._request = request
frames_list = await request.json()
logger.debug(f"Received frames: {frames_list}")
for frame in frames_list:
logger.debug(f"Received frame: {frame}")
frame = self._params.serializer.deserialize(frame)
if frame and isinstance(frame, AudioRawFrame):
await self.push_audio_frame(frame)
else:
await self.push_frame(frame)
class FastAPIHTTPOutputTransport(BaseOutputTransport):
def __init__(self, params: FastAPIHTTPParams, **kwargs):
super().__init__(params, **kwargs)
self._params = params
self._event_queue = asyncio.Queue()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
await self._write_frame(frame)
async def write_raw_audio_frames(self, frames: bytes):
pass
async def _write_frame(self, frame: Frame):
payload = self._params.serializer.serialize(frame)
await self._event_queue.put(payload)
async def event_generator(self):
while True:
event = await self._event_queue.get()
logger.debug(f"Sending event {event}")
yield event
class FastAPIHTTPTransport(BaseTransport):
def __init__(
self,
params: FastAPIHTTPParams,
input_name: str | None = None,
output_name: str | None = None,
loop: asyncio.AbstractEventLoop | None = None):
super().__init__(input_name=input_name, output_name=output_name, loop=loop)
self._params = params
self._request = None
self._input = FastAPIHTTPInputTransport(
self._params, name=self._input_name)
self._output = FastAPIHTTPOutputTransport(
self._params, name=self._output_name)
def input(self) -> FrameProcessor:
return self._input
def output(self) -> FrameProcessor:
return self._output
async def handle_request(self, request: Request):
self._request = request
await self._input.handle_request(request)
return EventSourceResponse(self._output.event_generator())

View File

@@ -12,8 +12,8 @@ import wave
from typing import Awaitable, Callable
from pydantic.main import BaseModel
from pipecat.frames.frames import AudioRawFrame, CancelFrame, EndFrame, Frame, StartFrame, StartInterruptionFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.frames.frames import AudioRawFrame, CancelFrame, EndFrame, StartFrame
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.serializers.base_serializer import FrameSerializer
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport
@@ -91,20 +91,13 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
self._websocket = websocket
self._params = params
self._websocket_audio_buffer = bytes()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, StartInterruptionFrame):
await self._write_frame(frame)
self._audio_buffer = bytes()
async def write_raw_audio_frames(self, frames: bytes):
self._websocket_audio_buffer += frames
while len(self._websocket_audio_buffer):
self._audio_buffer += frames
while len(self._audio_buffer) >= self._params.audio_frame_size:
frame = AudioRawFrame(
audio=self._websocket_audio_buffer[:
self._params.audio_frame_size],
audio=self._audio_buffer[:self._params.audio_frame_size],
sample_rate=self._params.audio_out_sample_rate,
num_channels=self._params.audio_out_channels
)
@@ -128,13 +121,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
if payload and self._websocket.client_state == WebSocketState.CONNECTED:
await self._websocket.send_text(payload)
self._websocket_audio_buffer = self._websocket_audio_buffer[
self._params.audio_frame_size:]
async def _write_frame(self, frame: Frame):
payload = self._params.serializer.serialize(frame)
if payload and self._websocket.client_state == WebSocketState.CONNECTED:
await self._websocket.send_text(payload)
self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:]
class FastAPIWebsocketTransport(BaseTransport):

View File

@@ -120,7 +120,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
self._websocket: websockets.WebSocketServerProtocol | None = None
self._websocket_audio_buffer = bytes()
self._audio_buffer = bytes()
async def set_client_connection(self, websocket: websockets.WebSocketServerProtocol | None):
if self._websocket:
@@ -132,10 +132,10 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
if not self._websocket:
return
self._websocket_audio_buffer += frames
while len(self._websocket_audio_buffer) >= self._params.audio_frame_size:
self._audio_buffer += frames
while len(self._audio_buffer) >= self._params.audio_frame_size:
frame = AudioRawFrame(
audio=self._websocket_audio_buffer[:self._params.audio_frame_size],
audio=self._audio_buffer[:self._params.audio_frame_size],
sample_rate=self._params.audio_out_sample_rate,
num_channels=self._params.audio_out_channels
)
@@ -159,7 +159,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
if proto:
await self._websocket.send(proto)
self._websocket_audio_buffer = self._websocket_audio_buffer[self._params.audio_frame_size:]
self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:]
class WebsocketServerTransport(BaseTransport):

View File

@@ -36,7 +36,6 @@ from pipecat.frames.frames import (
UserImageRawFrame,
UserImageRequestFrame)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.transcriptions.language import Language
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -159,8 +158,8 @@ class DailyTransportClient(EventHandler):
loop: asyncio.AbstractEventLoop):
super().__init__()
if not DailyTransportClient._daily_initialized:
DailyTransportClient._daily_initialized = True
if not self._daily_initialized:
self._daily_initialized = True
Daily.init()
self._room_url: str = room_url
@@ -183,39 +182,24 @@ class DailyTransportClient(EventHandler):
self._client: CallClient = CallClient(event_handler=self)
self._camera: VirtualCameraDevice | None = None
if self._params.camera_out_enabled:
self._camera = Daily.create_camera_device(
self._camera_name(),
width=self._params.camera_out_width,
height=self._params.camera_out_height,
color_format=self._params.camera_out_color_format)
self._camera: VirtualCameraDevice = Daily.create_camera_device(
"camera",
width=self._params.camera_out_width,
height=self._params.camera_out_height,
color_format=self._params.camera_out_color_format)
self._mic: VirtualMicrophoneDevice | None = None
if self._params.audio_out_enabled:
self._mic = Daily.create_microphone_device(
self._mic_name(),
sample_rate=self._params.audio_out_sample_rate,
channels=self._params.audio_out_channels,
non_blocking=True)
self._mic: VirtualMicrophoneDevice = Daily.create_microphone_device(
"mic",
sample_rate=self._params.audio_out_sample_rate,
channels=self._params.audio_out_channels,
non_blocking=True)
self._speaker: VirtualSpeakerDevice | None = None
if self._params.audio_in_enabled or self._params.vad_enabled:
self._speaker = Daily.create_speaker_device(
self._speaker_name(),
sample_rate=self._params.audio_in_sample_rate,
channels=self._params.audio_in_channels,
non_blocking=True)
Daily.select_speaker_device(self._speaker_name())
def _camera_name(self):
return f"camera-{self}"
def _mic_name(self):
return f"mic-{self}"
def _speaker_name(self):
return f"speaker-{self}"
self._speaker: VirtualSpeakerDevice = Daily.create_speaker_device(
"speaker",
sample_rate=self._params.audio_in_sample_rate,
channels=self._params.audio_in_channels,
non_blocking=True)
Daily.select_speaker_device("speaker")
@property
def participant_id(self) -> str:
@@ -240,9 +224,6 @@ class DailyTransportClient(EventHandler):
await future
async def read_next_audio_frame(self) -> AudioRawFrame | None:
if not self._speaker:
return None
sample_rate = self._params.audio_in_sample_rate
num_channels = self._params.audio_in_channels
num_frames = int(sample_rate / 100) * 2 # 20ms of audio
@@ -261,17 +242,11 @@ class DailyTransportClient(EventHandler):
return None
async def write_raw_audio_frames(self, frames: bytes):
if not self._mic:
return None
future = self._loop.create_future()
self._mic.write_frames(frames, completion=completion_callback(future))
await future
async def write_frame_to_camera(self, frame: ImageRawFrame):
if not self._camera:
return None
self._camera.write_frame(frame.image)
async def join(self):
@@ -306,7 +281,7 @@ class DailyTransportClient(EventHandler):
if self._token and self._params.transcription_enabled:
await self._start_transcription()
await self._callbacks.on_joined(data)
await self._callbacks.on_joined(data["participants"]["local"])
else:
error_msg = f"Error joining {self._room_url}: {error}"
logger.error(error_msg)
@@ -340,13 +315,13 @@ class DailyTransportClient(EventHandler):
"camera": {
"isEnabled": self._params.camera_out_enabled,
"settings": {
"deviceId": self._camera_name(),
"deviceId": "camera",
},
},
"microphone": {
"isEnabled": self._params.audio_out_enabled,
"settings": {
"deviceId": self._mic_name(),
"deviceId": "mic",
"customConstraints": {
"autoGainControl": {"exact": False},
"echoCancellation": {"exact": False},
@@ -366,12 +341,6 @@ class DailyTransportClient(EventHandler):
}
},
}
},
"microphone": {
"sendSettings": {
"channelConfig": "stereo" if self._params.audio_out_channels == 2 else "mono",
"bitrate": self._params.audio_out_bitrate,
}
}
},
})
@@ -565,7 +534,6 @@ class DailyInputTransport(BaseInputTransport):
self._client = client
self._video_renderers = {}
self._audio_in_task = None
self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer
if params.vad_enabled and not params.vad_analyzer:
@@ -589,7 +557,7 @@ class DailyInputTransport(BaseInputTransport):
# Leave the room.
await self._client.leave()
# Stop audio thread.
if self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled):
if self._params.audio_in_enabled or self._params.vad_enabled:
self._audio_in_task.cancel()
await self._audio_in_task
@@ -599,7 +567,7 @@ class DailyInputTransport(BaseInputTransport):
# Leave the room.
await self._client.leave()
# Stop audio thread.
if self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled):
if self._params.audio_in_enabled or self._params.vad_enabled:
self._audio_in_task.cancel()
await self._audio_in_task
@@ -760,7 +728,7 @@ class DailyTransport(BaseTransport):
room_url: str,
token: str | None,
bot_name: str,
params: DailyParams = DailyParams(),
params: DailyParams,
input_name: str | None = None,
output_name: str | None = None,
loop: asyncio.AbstractEventLoop | None = None):
@@ -825,7 +793,7 @@ class DailyTransport(BaseTransport):
# DailyTransport
#
@property
@ property
def participant_id(self) -> str:
return self._client.participant_id
@@ -871,8 +839,8 @@ class DailyTransport(BaseTransport):
self._input.capture_participant_video(
participant_id, framerate, video_source, color_format)
async def _on_joined(self, data):
await self._call_event_handler("on_joined", data)
async def _on_joined(self, participant):
await self._call_event_handler("on_joined", participant)
async def _on_left(self):
await self._call_event_handler("on_left")
@@ -957,16 +925,11 @@ class DailyTransport(BaseTransport):
text = message["text"]
timestamp = message["timestamp"]
is_final = message["rawResponse"]["is_final"]
try:
language = message["rawResponse"]["channel"]["alternatives"][0]["languages"][0]
language = Language(language)
except KeyError:
language = None
if is_final:
frame = TranscriptionFrame(text, participant_id, timestamp, language)
frame = TranscriptionFrame(text, participant_id, timestamp)
logger.debug(f"Transcription (from: {participant_id}): [{text}]")
else:
frame = InterimTranscriptionFrame(text, participant_id, timestamp, language)
frame = InterimTranscriptionFrame(text, participant_id, timestamp)
if self._input:
await self._input.push_transcription_frame(frame)

View File

@@ -70,20 +70,16 @@ class DailyRESTHelper:
self.daily_api_url = daily_api_url
self.aiohttp_session = aiohttp_session
def get_name_from_url(self, room_url: str) -> str:
def _get_name_from_url(self, room_url: str) -> str:
return urlparse(room_url).path[1:]
async def get_room_from_url(self, room_url: str) -> DailyRoomObject:
room_name = self.get_name_from_url(room_url)
return await self._get_room_from_name(room_name)
async def create_room(self, params: DailyRoomParams) -> DailyRoomObject:
headers = {"Authorization": f"Bearer {self.daily_api_key}"}
json = {**params.model_dump(exclude_none=True)}
async with self.aiohttp_session.post(f"{self.daily_api_url}/rooms", headers=headers, json=json) as r:
if r.status != 200:
text = await r.text()
raise Exception(f"Unable to create room (status: {r.status}): {text}")
raise Exception(f"Unable to create room: {text}")
data = await r.json()
@@ -94,49 +90,6 @@ class DailyRESTHelper:
return room
async def get_token(
self,
room_url: str,
expiry_time: float = 60 * 60,
owner: bool = True) -> str:
if not room_url:
raise Exception(
"No Daily room specified. You must specify a Daily room in order a token to be generated.")
expiration: float = time.time() + expiry_time
room_name = self.get_name_from_url(room_url)
headers = {"Authorization": f"Bearer {self.daily_api_key}"}
json = {
"properties": {
"room_name": room_name,
"is_owner": owner,
"exp": expiration
}
}
async with self.aiohttp_session.post(f"{self.daily_api_url}/meeting-tokens", headers=headers, json=json) as r:
if r.status != 200:
text = await r.text()
raise Exception(f"Failed to create meeting token (status: {r.status}): {text}")
data = await r.json()
return data["token"]
async def delete_room_by_url(self, room_url: str) -> bool:
room_name = self.get_name_from_url(room_url)
return await self.delete_room_by_name(room_name)
async def delete_room_by_name(self, room_name: str) -> bool:
headers = {"Authorization": f"Bearer {self.daily_api_key}"}
async with self.aiohttp_session.delete(f"{self.daily_api_url}/rooms/{room_name}", headers=headers) as r:
if r.status != 200 and r.status != 404:
text = await r.text()
raise Exception(f"Failed to delete room [{room_name}] (status: {r.status}): {text}")
return True
async def _get_room_from_name(self, room_name: str) -> DailyRoomObject:
headers = {"Authorization": f"Bearer {self.daily_api_key}"}
async with self.aiohttp_session.get(f"{self.daily_api_url}/rooms/{room_name}", headers=headers) as r:
@@ -151,3 +104,47 @@ class DailyRESTHelper:
raise Exception(f"Invalid response: {e}")
return room
async def get_room_from_url(self, room_url: str,) -> DailyRoomObject:
room_name = self._get_name_from_url(room_url)
return await self._get_room_from_name(room_name)
async def get_token(
self,
room_url: str,
expiry_time: float = 60 * 60,
owner: bool = True) -> str:
if not room_url:
raise Exception(
"No Daily room specified. You must specify a Daily room in order a token to be generated.")
expiration: float = time.time() + expiry_time
room_name = self._get_name_from_url(room_url)
headers = {"Authorization": f"Bearer {self.daily_api_key}"}
json = {
"properties": {
"room_name": room_name,
"is_owner": owner,
"exp": expiration
}
}
async with self.aiohttp_session.post(f"{self.daily_api_url}/meeting-tokens", headers=headers, json=json) as r:
if r.status != 200:
text = await r.text()
raise Exception(f"Failed to create meeting token: {r.status} {text}")
data = await r.json()
return data["token"]
async def delete_room_by_name(self, room_name: str) -> bool:
headers = {"Authorization": f"Bearer {self.daily_api_key}"}
async with self.aiohttp_session.delete(f"{self.daily_api_url}/rooms/{room_name}", headers=headers) as r:
if r.status != 200 and r.status != 404:
raise Exception(f"Failed to delete room: {room_name}")
data = await r.json()
return True

View File

@@ -1,24 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import re
ENDOFSENTENCE_PATTERN_STR = r"""
(?<![A-Z]) # Negative lookbehind: not preceded by an uppercase letter (e.g., "U.S.A.")
(?<!\d) # Negative lookbehind: not preceded by a digit (e.g., "1. Let's start")
(?<!\d\s[ap]) # Negative lookbehind: not preceded by time (e.g., "3:00 a.m.")
(?<!Mr|Ms|Dr) # Negative lookbehind: not preceded by Mr, Ms, Dr (combined bc. length is the same)
(?<!Mrs) # Negative lookbehind: not preceded by "Mrs"
(?<!Prof) # Negative lookbehind: not preceded by "Prof"
[\.\?\!:] # Match a period, question mark, exclamation point, or colon
$ # End of string
"""
ENDOFSENTENCE_PATTERN = re.compile(ENDOFSENTENCE_PATTERN_STR, re.VERBOSE)
def match_endofsentence(text: str) -> bool:
return ENDOFSENTENCE_PATTERN.search(text.rstrip()) is not None

View File

@@ -9,20 +9,3 @@ import datetime
def time_now_iso8601() -> str:
return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="milliseconds")
def seconds_to_nanoseconds(seconds: float) -> int:
return int(seconds * 1_000_000_000)
def nanoseconds_to_seconds(nanoseconds: int) -> float:
return nanoseconds / 1_000_000_000
def nanoseconds_to_str(nanoseconds: int) -> str:
total_seconds = nanoseconds_to_seconds(nanoseconds)
hours = int(total_seconds // 3600)
minutes = int((total_seconds % 3600) // 60)
seconds = int(total_seconds % 60)
microseconds = int((total_seconds - int(total_seconds)) * 1_000_000)
return f"{hours}:{minutes:02}:{seconds:02}.{microseconds:06}"

View File

@@ -3,39 +3,32 @@
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import collections
import itertools
_COUNTS = collections.defaultdict(itertools.count)
_ID = itertools.count()
from threading import Lock
_COUNTS = {}
_COUNTS_MUTEX = Lock()
_ID = 0
_ID_MUTEX = Lock()
def obj_id() -> int:
"""
Generate a unique id for an object.
>>> obj_id()
0
>>> obj_id()
1
>>> obj_id()
2
"""
return next(_ID)
global _ID, _ID_MUTEX
with _ID_MUTEX:
_ID += 1
return _ID
def obj_count(obj) -> int:
"""Generate a unique id for an object.
>>> obj_count(object())
0
>>> obj_count(object())
1
>>> new_type = type('NewType', (object,), {})
>>> obj_count(new_type())
0
"""
return next(_COUNTS[obj.__class__.__name__])
global _COUNTS, COUNTS_MUTEX
name = obj.__class__.__name__
with _COUNTS_MUTEX:
if name not in _COUNTS:
_COUNTS[name] = 0
else:
_COUNTS[name] += 1
return _COUNTS[name]
def exp_smoothing(value: float, prev_value: float, factor: float) -> float:

View File

@@ -11,8 +11,6 @@ from pydantic.main import BaseModel
from pipecat.utils.audio import calculate_audio_volume, exp_smoothing
from loguru import logger
class VADState(Enum):
QUIET = 1
@@ -33,8 +31,17 @@ class VADAnalyzer:
def __init__(self, *, sample_rate: int, num_channels: int, params: VADParams):
self._sample_rate = sample_rate
self._num_channels = num_channels
self._params = params
self._vad_frames = self.num_frames_required()
self._vad_frames_num_bytes = self._vad_frames * num_channels * 2
self.set_params(params)
vad_frames_per_sec = self._vad_frames / self._sample_rate
self._vad_start_frames = round(self._params.start_secs / vad_frames_per_sec)
self._vad_stop_frames = round(self._params.stop_secs / vad_frames_per_sec)
self._vad_starting_count = 0
self._vad_stopping_count = 0
self._vad_state: VADState = VADState.QUIET
self._vad_buffer = b""
@@ -46,10 +53,6 @@ class VADAnalyzer:
def sample_rate(self):
return self._sample_rate
@property
def num_channels(self):
return self._num_channels
@abstractmethod
def num_frames_required(self) -> int:
pass
@@ -58,20 +61,6 @@ class VADAnalyzer:
def voice_confidence(self, buffer) -> float:
pass
def set_params(self, params: VADParams):
logger.debug(f"Setting VAD params to: {params}")
self._params = params
self._vad_frames = self.num_frames_required()
self._vad_frames_num_bytes = self._vad_frames * self._num_channels * 2
vad_frames_per_sec = self._vad_frames / self._sample_rate
self._vad_start_frames = round(self._params.start_secs / vad_frames_per_sec)
self._vad_stop_frames = round(self._params.stop_secs / vad_frames_per_sec)
self._vad_starting_count = 0
self._vad_stopping_count = 0
self._vad_state: VADState = VADState.QUIET
def _get_smoothed_volume(self, audio: bytes) -> float:
volume = calculate_audio_volume(audio, self._sample_rate)
return exp_smoothing(volume, self._prev_volume, self._smoothing_factor)