diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7bf00326d..a757c7ffd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,26 @@ 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/)
@@ -20,6 +40,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### 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.
@@ -32,6 +58,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
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
diff --git a/examples/deployment/flyio-example/bot.py b/examples/deployment/flyio-example/bot.py
index cc68f5522..c6380f6f3 100644
--- a/examples/deployment/flyio-example/bot.py
+++ b/examples/deployment/flyio-example/bot.py
@@ -1,5 +1,4 @@
import asyncio
-import aiohttp
import os
import sys
import argparse
@@ -27,71 +26,69 @@ daily_api_url = os.getenv("DAILY_API_URL", "https://api.daily.co/v1")
async def main(room_url: str, token: str):
- 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,
- )
+ 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(
- aiohttp_session=session,
- api_key=os.getenv("ELEVENLABS_API_KEY", ""),
- voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
- )
+ 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")
+ 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):
+ @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":
await task.queue_frame(EndFrame())
- @transport.event_handler("on_call_state_updated")
- async def on_call_state_updated(transport, state):
- if state == "left":
- await task.queue_frame(EndFrame())
+ runner = PipelineRunner()
- runner = PipelineRunner()
-
- await runner.run(task)
+ await runner.run(task)
if __name__ == "__main__":
diff --git a/examples/dialin-chatbot/bot_daily.py b/examples/dialin-chatbot/bot_daily.py
index ea30cd2d5..cd6afdad0 100644
--- a/examples/dialin-chatbot/bot_daily.py
+++ b/examples/dialin-chatbot/bot_daily.py
@@ -1,5 +1,4 @@
import asyncio
-import aiohttp
import os
import sys
import argparse
@@ -29,75 +28,74 @@ 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):
- 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
+ # 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,
)
+ )
- 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,
- )
- )
+ 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 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?!'.",
+ },
+ ]
- 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?!'.",
- },
- ]
+ 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_participant_left")
- async def on_participant_left(transport, participant, reason):
- await task.queue_frame(EndFrame())
+ runner = PipelineRunner()
- runner = PipelineRunner()
-
- await runner.run(task)
+ await runner.run(task)
if __name__ == "__main__":
diff --git a/examples/dialin-chatbot/bot_twilio.py b/examples/dialin-chatbot/bot_twilio.py
index 6ae8a24b3..e6653babd 100644
--- a/examples/dialin-chatbot/bot_twilio.py
+++ b/examples/dialin-chatbot/bot_twilio.py
@@ -1,5 +1,4 @@
import asyncio
-import aiohttp
import os
import sys
import argparse
@@ -36,82 +35,81 @@ daily_api_key = os.getenv("DAILY_API_KEY", "")
async def main(room_url: str, token: str, callId: str, sipUri: str):
- 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,
+ # 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'{sipUri}'
)
- )
+ except Exception as e:
+ raise Exception(f"Failed to forward call: {str(e)}")
- 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'{sipUri}'
- )
- except Exception as e:
- raise Exception(f"Failed to forward call: {str(e)}")
-
- runner = PipelineRunner()
- await runner.run(task)
+ runner = PipelineRunner()
+ await runner.run(task)
if __name__ == "__main__":
diff --git a/examples/dialin-chatbot/requirements.txt b/examples/dialin-chatbot/requirements.txt
index 38a0a93b0..e59a9c3d2 100644
--- a/examples/dialin-chatbot/requirements.txt
+++ b/examples/dialin-chatbot/requirements.txt
@@ -3,3 +3,4 @@ fastapi
uvicorn
python-dotenv
twilio
+python-multipart
diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py
index e3965e857..ca3ff9557 100644
--- a/examples/foundational/05-sync-speech-and-image.py
+++ b/examples/foundational/05-sync-speech-and-image.py
@@ -89,7 +89,6 @@ async def main():
)
tts = ElevenLabsTTSService(
- aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
diff --git a/examples/foundational/05a-local-sync-speech-and-image.py b/examples/foundational/05a-local-sync-speech-and-image.py
index 5decffcb5..63bcf1e9d 100644
--- a/examples/foundational/05a-local-sync-speech-and-image.py
+++ b/examples/foundational/05a-local-sync-speech-and-image.py
@@ -85,7 +85,6 @@ 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"))
diff --git a/examples/foundational/06a-image-sync.py b/examples/foundational/06a-image-sync.py
index fb1824ed8..812dab137 100644
--- a/examples/foundational/06a-image-sync.py
+++ b/examples/foundational/06a-image-sync.py
@@ -79,7 +79,6 @@ async def main():
)
tts = ElevenLabsTTSService(
- aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
)
diff --git a/examples/foundational/07b-interruptible-langchain.py b/examples/foundational/07b-interruptible-langchain.py
index c517ff27a..872dbf9bb 100644
--- a/examples/foundational/07b-interruptible-langchain.py
+++ b/examples/foundational/07b-interruptible-langchain.py
@@ -18,7 +18,6 @@ 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
diff --git a/examples/foundational/07d-interruptible-cartesia.py b/examples/foundational/07d-interruptible-elevenlabs.py
similarity index 82%
rename from examples/foundational/07d-interruptible-cartesia.py
rename to examples/foundational/07d-interruptible-elevenlabs.py
index 7bcc7476b..19bd4ad01 100644
--- a/examples/foundational/07d-interruptible-cartesia.py
+++ b/examples/foundational/07d-interruptible-elevenlabs.py
@@ -4,8 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
-import aiohttp
import asyncio
+import aiohttp
import os
import sys
@@ -15,12 +15,11 @@ 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
-
from runner import configure
from loguru import logger
@@ -41,7 +40,6 @@ async def main():
token,
"Respond bot",
DailyParams(
- audio_out_sample_rate=44100,
audio_out_enabled=True,
transcription_enabled=True,
vad_enabled=True,
@@ -49,12 +47,9 @@ async def main():
)
)
- tts = CartesiaTTSService(
- api_key=os.getenv("CARTESIA_API_KEY"),
- voice_id="a0e99841-438c-4a64-b679-ae501e7d6091", # Barbershop Man
- params=CartesiaTTSService.InputParams(
- sample_rate=44100,
- ),
+ tts = ElevenLabsTTSService(
+ api_key=os.getenv("ELEVENLABS_API_KEY", ""),
+ voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
)
llm = OpenAILLMService(
@@ -76,11 +71,16 @@ 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, enable_metrics=True))
+ 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):
diff --git a/examples/foundational/11-sound-effects.py b/examples/foundational/11-sound-effects.py
index 00fb0c9be..146b3bd09 100644
--- a/examples/foundational/11-sound-effects.py
+++ b/examples/foundational/11-sound-effects.py
@@ -104,7 +104,6 @@ async def main():
model="gpt-4o")
tts = ElevenLabsTTSService(
- aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id="ErXwobaYiN019PkySvjV",
)
diff --git a/examples/simple-chatbot/bot.py b/examples/simple-chatbot/bot.py
index d00f6acd1..1664e47fb 100644
--- a/examples/simple-chatbot/bot.py
+++ b/examples/simple-chatbot/bot.py
@@ -111,7 +111,6 @@ async def main():
)
tts = ElevenLabsTTSService(
- aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"),
#
# English
diff --git a/examples/storytelling-chatbot/src/bot.py b/examples/storytelling-chatbot/src/bot.py
index 4bd50fe42..91452dd75 100644
--- a/examples/storytelling-chatbot/src/bot.py
+++ b/examples/storytelling-chatbot/src/bot.py
@@ -60,7 +60,6 @@ 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"),
)
diff --git a/examples/studypal/studypal.py b/examples/studypal/studypal.py
index f14bd3def..79fd41764 100644
--- a/examples/studypal/studypal.py
+++ b/examples/studypal/studypal.py
@@ -149,8 +149,8 @@ Your task is to help the user understand and learn from this article in 2 senten
tma_in,
llm,
tts,
- tma_out,
transport.output(),
+ tma_out,
])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True))
diff --git a/pyproject.toml b/pyproject.toml
index 721f4be19..8a1e3a800 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -39,6 +39,7 @@ azure = [ "azure-cognitiveservices-speech~=1.40.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" ]
fal = [ "fal-client~=0.4.1" ]
gladia = [ "websockets~=12.0" ]
diff --git a/src/pipecat/clocks/__init__.py b/src/pipecat/clocks/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/src/pipecat/clocks/base_clock.py b/src/pipecat/clocks/base_clock.py
new file mode 100644
index 000000000..aa7b7b806
--- /dev/null
+++ b/src/pipecat/clocks/base_clock.py
@@ -0,0 +1,18 @@
+#
+# 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
diff --git a/src/pipecat/clocks/system_clock.py b/src/pipecat/clocks/system_clock.py
new file mode 100644
index 000000000..20319cff6
--- /dev/null
+++ b/src/pipecat/clocks/system_clock.py
@@ -0,0 +1,21 @@
+#
+# 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()
diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py
index 13c2f53f1..51770dff1 100644
--- a/src/pipecat/frames/frames.py
+++ b/src/pipecat/frames/frames.py
@@ -8,19 +8,27 @@ from typing import Any, List, Mapping, Optional, 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
@@ -46,7 +54,8 @@ class AudioRawFrame(DataFrame):
self.num_frames = int(len(self.audio) / (self.num_channels * 2))
def __str__(self):
- return f"{self.name}(size: {len(self.audio)}, frames: {self.num_frames}, sample_rate: {self.sample_rate}, channels: {self.num_channels})"
+ 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})"
@dataclass
@@ -60,7 +69,8 @@ class ImageRawFrame(DataFrame):
format: str | None
def __str__(self):
- return f"{self.name}(size: {self.size}, format: {self.format})"
+ pts = format_pts(self.pts)
+ return f"{self.name}(pts: {pts}, size: {self.size}, format: {self.format})"
@dataclass
@@ -72,7 +82,8 @@ class URLImageRawFrame(ImageRawFrame):
url: str | None
def __str__(self):
- return f"{self.name}(url: {self.url}, size: {self.size}, format: {self.format})"
+ pts = format_pts(self.pts)
+ return f"{self.name}(pts: {pts}, url: {self.url}, size: {self.size}, format: {self.format})"
@dataclass
@@ -84,7 +95,8 @@ class VisionImageRawFrame(ImageRawFrame):
text: str | None
def __str__(self):
- return f"{self.name}(text: {self.text}, size: {self.size}, format: {self.format})"
+ pts = format_pts(self.pts)
+ return f"{self.name}(pts: {pts}, text: {self.text}, size: {self.size}, format: {self.format})"
@dataclass
@@ -96,7 +108,8 @@ class UserImageRawFrame(ImageRawFrame):
user_id: str
def __str__(self):
- return f"{self.name}(user: {self.user_id}, size: {self.size}, format: {self.format})"
+ pts = format_pts(self.pts)
+ return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format})"
@dataclass
@@ -109,7 +122,8 @@ class SpriteFrame(Frame):
images: List[ImageRawFrame]
def __str__(self):
- return f"{self.name}(size: {len(self.images)})"
+ pts = format_pts(self.pts)
+ return f"{self.name}(pts: {pts}, size: {len(self.images)})"
@dataclass
@@ -121,7 +135,8 @@ class TextFrame(DataFrame):
text: str
def __str__(self):
- return f"{self.name}(text: {self.text})"
+ pts = format_pts(self.pts)
+ return f"{self.name}(pts: {pts}, text: {self.text})"
@dataclass
@@ -326,6 +341,7 @@ 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
diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index 102b4528b..03fd5c734 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -10,6 +10,8 @@ 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,
@@ -60,11 +62,16 @@ class Source(FrameProcessor):
class PipelineTask:
- def __init__(self, pipeline: BasePipeline, params: PipelineParams = PipelineParams()):
+ def __init__(
+ self,
+ pipeline: BasePipeline,
+ params: PipelineParams = PipelineParams(),
+ clock: BaseClock = SystemClock()):
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
@@ -116,11 +123,14 @@ 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
+ report_only_initial_ttfb=self._params.report_only_initial_ttfb,
+ clock=self._clock
)
await self._source.process_frame(start_frame, FrameDirection.DOWNSTREAM)
diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py
index 7c38e62ad..ab0552578 100644
--- a/src/pipecat/processors/aggregators/llm_response.py
+++ b/src/pipecat/processors/aggregators/llm_response.py
@@ -109,7 +109,7 @@ class LLMResponseAggregator(FrameProcessor):
await self.push_frame(frame, direction)
elif isinstance(frame, self._accumulator_frame):
if self._aggregating:
- self._aggregation += f" {frame.text}"
+ self._aggregation += f" {frame.text}" if self._aggregation else 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.
diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py
index 156e1c0ae..dfdee7d40 100644
--- a/src/pipecat/processors/frame_processor.py
+++ b/src/pipecat/processors/frame_processor.py
@@ -9,6 +9,7 @@ import time
from enum import Enum
+from pipecat.clocks.base_clock import BaseClock
from pipecat.frames.frames import (
ErrorFrame,
Frame,
@@ -96,6 +97,9 @@ 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
@@ -177,8 +181,12 @@ 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
diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py
index 8836fcd6d..583234ae4 100644
--- a/src/pipecat/serializers/twilio.py
+++ b/src/pipecat/serializers/twilio.py
@@ -9,7 +9,7 @@ import json
from pydantic import BaseModel
-from pipecat.frames.frames import AudioRawFrame, Frame
+from pipecat.frames.frames import AudioRawFrame, Frame, StartInterruptionFrame
from pipecat.serializers.base_serializer import FrameSerializer
from pipecat.utils.audio import ulaw_to_pcm, pcm_to_ulaw
@@ -28,22 +28,25 @@ class TwilioFrameSerializer(FrameSerializer):
self._params = params
def serialize(self, frame: Frame) -> str | bytes | None:
- if not isinstance(frame, AudioRawFrame):
- return None
+ if isinstance(frame, AudioRawFrame):
+ data = frame.audio
- 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
+ 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)
+ return json.dumps(answer)
+
+ if isinstance(frame, StartInterruptionFrame):
+ answer = {"event": "clear", "streamSid": self._stream_sid}
+ return json.dumps(answer)
def deserialize(self, data: str | bytes) -> Frame | None:
message = json.loads(data)
diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py
index c5cb7cc0d..dcba578c5 100644
--- a/src/pipecat/services/ai_services.py
+++ b/src/pipecat/services/ai_services.py
@@ -9,7 +9,7 @@ import io
import wave
from abc import abstractmethod
-from typing import AsyncGenerator, Optional
+from typing import AsyncGenerator, List, Optional, Tuple
from pipecat.frames.frames import (
AudioRawFrame,
@@ -37,9 +37,12 @@ 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
+from loguru import logger
+
class AIService(FrameProcessor):
def __init__(self, **kwargs):
@@ -167,7 +170,7 @@ class TTSService(AIService):
# 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 = 0.8,
+ stop_frame_timeout_s: float = 1.0,
**kwargs):
super().__init__(**kwargs)
self._aggregate_sentences: bool = aggregate_sentences
@@ -303,6 +306,74 @@ class TTSService(AIService):
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."""
diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py
index 927da53f0..25f54bf11 100644
--- a/src/pipecat/services/cartesia.py
+++ b/src/pipecat/services/cartesia.py
@@ -28,7 +28,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.transcriptions.language import Language
-from pipecat.services.ai_services import TTSService
+from pipecat.services.ai_services import AsyncWordTTSService
from loguru import logger
@@ -61,7 +61,7 @@ def language_to_cartesia_language(language: Language) -> str | None:
return None
-class CartesiaTTSService(TTSService):
+class CartesiaTTSService(AsyncWordTTSService):
class InputParams(BaseModel):
model_id: Optional[str] = "sonic-english"
encoding: Optional[str] = "pcm_s16le"
@@ -80,19 +80,17 @@ class CartesiaTTSService(TTSService):
url: str = "wss://api.cartesia.ai/tts/websocket",
params: InputParams = InputParams(),
**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.
- 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
+ # 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)
self._api_key = api_key
self._cartesia_version = cartesia_version
@@ -110,10 +108,7 @@ class CartesiaTTSService(TTSService):
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
@@ -145,43 +140,55 @@ class CartesiaTTSService(TTSService):
async def _connect(self):
try:
self._websocket = await websockets.connect(
- f"{self._url}?api_key={self._api_key}&cartesia_version={self._cartesia_version}"
+ 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}")
+ logger.error(f"{self} initialization error: {e}")
self._websocket = None
async def _disconnect(self):
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
- self._receive_task = None
if self._websocket:
await self._websocket.close()
self._websocket = None
+ if self._receive_task:
+ self._receive_task.cancel()
+ await self._receive_task
+ self._receive_task = 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}")
+ logger.error(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:
@@ -196,16 +203,15 @@ class CartesiaTTSService(TTSService):
# because we are likely still playing out audio and need the
# timestamp to set send context frames.
self._context_id = None
- self._timestamped_words_buffer.append(("LLMFullResponseEndFrame", 0))
+ await self.add_word_timestamps([("LLMFullResponseEndFrame", 0)])
elif msg["type"] == "timestamps":
- # logger.debug(f"TIMESTAMPS: {msg}")
- self._timestamped_words_buffer.extend(
- list(zip(msg["word_timestamps"]["words"], msg["word_timestamps"]["end"]))
+ await self.add_word_timestamps(
+ list(zip(msg["word_timestamps"]["words"],
+ msg["word_timestamps"]["start"]))
)
elif msg["type"] == "chunk":
await self.stop_ttfb_metrics()
- if not self._context_id_start_timestamp:
- self._context_id_start_timestamp = time.time()
+ self.start_word_timestamps()
frame = AudioRawFrame(
audio=base64.b64decode(msg["data"]),
sample_rate=self._output_format["sample_rate"],
@@ -218,32 +224,12 @@ class CartesiaTTSService(TTSService):
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}")
+ 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:
- logger.exception(f"{self} exception: {e}")
+ logger.error(f"{self} exception: {e}")
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating TTS: [{text}]")
@@ -290,4 +276,4 @@ class CartesiaTTSService(TTSService):
return
yield None
except Exception as e:
- logger.exception(f"{self} exception: {e}")
+ logger.error(f"{self} exception: {e}")
diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py
index 974619ea8..a7a80033e 100644
--- a/src/pipecat/services/elevenlabs.py
+++ b/src/pipecat/services/elevenlabs.py
@@ -4,18 +4,72 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
-import aiohttp
+import asyncio
+import base64
+import json
-from typing import AsyncGenerator, Literal
+from typing import Any, AsyncGenerator, List, Literal, Mapping, Tuple
from pydantic import BaseModel
-from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, TTSStartedFrame, TTSStoppedFrame
-from pipecat.services.ai_services import TTSService
+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 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}")
-class ElevenLabsTTSService(TTSService):
+
+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 InputParams(BaseModel):
output_format: Literal["pcm_16000", "pcm_22050", "pcm_24000", "pcm_44100"] = "pcm_16000"
@@ -24,56 +78,186 @@ class ElevenLabsTTSService(TTSService):
*,
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):
- 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. 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
+ )
self._api_key = api_key
self._voice_id = voice_id
self._model = model
+ self._url = url
self._params = params
- self._aiohttp_session = aiohttp_session
+ 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
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.error(f"{self} initialization error: {e}")
+ self._websocket = None
+
+ async def _disconnect(self):
+ try:
+ await self.stop_all_metrics()
+
+ if self._websocket:
+ await self._websocket.send(json.dumps({"text": ""}))
+ await self._websocket.close()
+ self._websocket = None
+
+ 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
+
+ self._started = False
+ except Exception as e:
+ logger.error(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.error(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.error(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}]")
- url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream"
+ try:
+ if not self._websocket:
+ await self._connect()
- payload = {"text": text, "model_id": self._model}
+ try:
+ if not self._started:
+ await self.push_frame(TTSStartedFrame())
+ await self.start_ttfb_metrics()
+ self._started = True
+ self._cumulative_time = 0
- 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})")
+ 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()
return
-
- await self.start_tts_usage_metrics(text)
-
- await self.push_frame(TTSStartedFrame())
- async for chunk in r.content:
- if len(chunk) > 0:
- await self.stop_ttfb_metrics()
- frame = AudioRawFrame(chunk, 16000, 1)
- yield frame
- await self.push_frame(TTSStoppedFrame())
+ yield None
+ except Exception as e:
+ logger.error(f"{self} exception: {e}")
diff --git a/src/pipecat/services/lmnt.py b/src/pipecat/services/lmnt.py
index f7afd6a41..f5ad8aa1a 100644
--- a/src/pipecat/services/lmnt.py
+++ b/src/pipecat/services/lmnt.py
@@ -20,7 +20,7 @@ from pipecat.frames.frames import (
TTSStartedFrame,
TTSStoppedFrame,
)
-from pipecat.services.ai_services import TTSService
+from pipecat.services.ai_services import AsyncTTSService
from loguru import logger
@@ -34,7 +34,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
-class LmntTTSService(TTSService):
+class LmntTTSService(AsyncTTSService):
def __init__(
self,
@@ -44,11 +44,9 @@ class LmntTTSService(TTSService):
sample_rate: int = 24000,
language: str = "en",
**kwargs):
- super().__init__(**kwargs)
-
# Let TTSService produce TTSStoppedFrames after a short delay of
# no activity.
- self._push_stop_frames = True
+ super().__init__(push_stop_frames=True, **kwargs)
self._api_key = api_key
self._voice_id = voice_id
@@ -62,6 +60,8 @@ class LmntTTSService(TTSService):
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:
diff --git a/src/pipecat/transcriptions/__init__.py b/src/pipecat/transcriptions/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py
index d2c6add2b..a24c9f4d2 100644
--- a/src/pipecat/transports/base_output.py
+++ b/src/pipecat/transports/base_output.py
@@ -8,6 +8,7 @@
import asyncio
import itertools
import time
+import sys
from PIL import Image
from typing import List
@@ -30,11 +31,14 @@ 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):
@@ -64,7 +68,7 @@ class BaseOutputTransport(FrameProcessor):
# 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_task()
+ self._create_sink_tasks()
# 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.
@@ -149,6 +153,7 @@ 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.
@@ -158,6 +163,9 @@ class BaseOutputTransport(FrameProcessor):
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)
@@ -166,10 +174,14 @@ class BaseOutputTransport(FrameProcessor):
return
if isinstance(frame, StartInterruptionFrame):
- # Stop sink task.
+ # Stop sink tasks.
self._sink_task.cancel()
await self._sink_task
- self._create_sink_task()
+ # Stop sink clock tasks.
+ self._sink_clock_task.cancel()
+ await self._sink_clock_task
+ # Create sink tasks.
+ self._create_sink_tasks()
# Stop push task.
self._push_frame_task.cancel()
await self._push_frame_task
@@ -201,43 +213,83 @@ class BaseOutputTransport(FrameProcessor):
else:
await self._sink_queue.put(frame)
- def _create_sink_task(self):
+ #
+ # Sink tasks
+ #
+
+ def _create_sink_tasks(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()
- 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)
-
+ await self._sink_frame_handler(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
diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py
index 72e609263..083aeac37 100644
--- a/src/pipecat/transports/base_transport.py
+++ b/src/pipecat/transports/base_transport.py
@@ -32,6 +32,7 @@ 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
diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py
index 914870114..2c4bd187b 100644
--- a/src/pipecat/transports/network/fastapi_websocket.py
+++ b/src/pipecat/transports/network/fastapi_websocket.py
@@ -12,8 +12,8 @@ import wave
from typing import Awaitable, Callable
from pydantic.main import BaseModel
-from pipecat.frames.frames import AudioRawFrame, CancelFrame, EndFrame, StartFrame
-from pipecat.processors.frame_processor import FrameProcessor
+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
@@ -93,11 +93,18 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
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)
+
async def write_raw_audio_frames(self, frames: bytes):
self._websocket_audio_buffer += frames
- while len(self._websocket_audio_buffer) >= self._params.audio_frame_size:
+ while len(self._websocket_audio_buffer):
frame = AudioRawFrame(
- audio=self._websocket_audio_buffer[:self._params.audio_frame_size],
+ audio=self._websocket_audio_buffer[:
+ self._params.audio_frame_size],
sample_rate=self._params.audio_out_sample_rate,
num_channels=self._params.audio_out_channels
)
@@ -121,7 +128,13 @@ 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:]
+ 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)
class FastAPIWebsocketTransport(BaseTransport):
diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py
index bb6032fa4..7cf330b9e 100644
--- a/src/pipecat/transports/services/daily.py
+++ b/src/pipecat/transports/services/daily.py
@@ -366,6 +366,12 @@ class DailyTransportClient(EventHandler):
}
},
}
+ },
+ "microphone": {
+ "sendSettings": {
+ "channelConfig": "stereo" if self._params.audio_out_channels == 2 else "mono",
+ "bitrate": self._params.audio_out_bitrate,
+ }
}
},
})
diff --git a/src/pipecat/utils/time.py b/src/pipecat/utils/time.py
index af493e77b..0f6ca1076 100644
--- a/src/pipecat/utils/time.py
+++ b/src/pipecat/utils/time.py
@@ -9,3 +9,20 @@ 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}"
diff --git a/src/pipecat/utils/utils.py b/src/pipecat/utils/utils.py
index 0be73191f..e2df99389 100644
--- a/src/pipecat/utils/utils.py
+++ b/src/pipecat/utils/utils.py
@@ -3,32 +3,39 @@
#
# SPDX-License-Identifier: BSD 2-Clause License
#
+import collections
+import itertools
-from threading import Lock
-
-_COUNTS = {}
-_COUNTS_MUTEX = Lock()
-
-_ID = 0
-_ID_MUTEX = Lock()
+_COUNTS = collections.defaultdict(itertools.count)
+_ID = itertools.count()
def obj_id() -> int:
- global _ID, _ID_MUTEX
- with _ID_MUTEX:
- _ID += 1
- return _ID
+ """
+ Generate a unique id for an object.
+
+ >>> obj_id()
+ 0
+ >>> obj_id()
+ 1
+ >>> obj_id()
+ 2
+ """
+ return next(_ID)
def obj_count(obj) -> int:
- 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]
+ """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__])
def exp_smoothing(value: float, prev_value: float, factor: float) -> float: