wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

services(elevenlabs): add elevenlabs package and use streaming

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-09-05 11:12:52 -07:00
parent 51cd7fd285
commit e405d7af9f
13 changed files with 230 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
examples/deployment/flyio-example/bot.py
View File

@@ -1,5 +1,4 @@
import asyncio import asyncio
import aiohttp
import os import os
import sys import sys
import argparse 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 def main(room_url: str, token: str):
async with aiohttp.ClientSession() as session: transport = DailyTransport(
transport = DailyTransport( room_url,
room_url, token,
token, "Chatbot",
"Chatbot", DailyParams(
DailyParams( api_url=daily_api_url,
api_url=daily_api_url, api_key=daily_api_key,
api_key=daily_api_key, audio_in_enabled=True,
audio_in_enabled=True, audio_out_enabled=True,
audio_out_enabled=True, camera_out_enabled=False,
camera_out_enabled=False, vad_enabled=True,
vad_enabled=True, vad_analyzer=SileroVADAnalyzer(),
vad_analyzer=SileroVADAnalyzer(), transcription_enabled=True,
transcription_enabled=True,
)
) )
)
tts = ElevenLabsTTSService( tts = ElevenLabsTTSService(
aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY", ""),
api_key=os.getenv("ELEVENLABS_API_KEY", ""), voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), )
)
llm = OpenAILLMService( llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"), api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4o") model="gpt-4o")
messages = [ messages = [
{ {
"role": "system", "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.", "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_in = LLMUserResponseAggregator(messages)
tma_out = LLMAssistantResponseAggregator(messages) tma_out = LLMAssistantResponseAggregator(messages)
pipeline = Pipeline([ pipeline = Pipeline([
transport.input(), transport.input(),
tma_in, tma_in,
llm, llm,
tts, tts,
transport.output(), transport.output(),
tma_out, tma_out,
]) ])
task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
@transport.event_handler("on_first_participant_joined") @transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant): async def on_first_participant_joined(transport, participant):
transport.capture_participant_transcription(participant["id"]) transport.capture_participant_transcription(participant["id"])
await task.queue_frames([LLMMessagesFrame(messages)]) await task.queue_frames([LLMMessagesFrame(messages)])
@transport.event_handler("on_participant_left") @transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason): 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()) await task.queue_frame(EndFrame())
@transport.event_handler("on_call_state_updated") runner = PipelineRunner()
async def on_call_state_updated(transport, state):
if state == "left":
await task.queue_frame(EndFrame())
runner = PipelineRunner() await runner.run(task)
await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":

116
examples/dialin-chatbot/bot_daily.py
View File

@@ -1,5 +1,4 @@
import asyncio import asyncio
import aiohttp
import os import os
import sys import sys
import argparse 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 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
# diallin_settings are only needed if Daily's SIP URI is used # If you are handling this via Twilio, Telnyx, set this to None
# If you are handling this via Twilio, Telnyx, set this to None # and handle call-forwarding when on_dialin_ready fires.
# and handle call-forwarding when on_dialin_ready fires. diallin_settings = DailyDialinSettings(
diallin_settings = DailyDialinSettings( call_id=callId,
call_id=callId, call_domain=callDomain
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( tts = ElevenLabsTTSService(
room_url, api_key=os.getenv("ELEVENLABS_API_KEY", ""),
token, voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
"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( llm = OpenAILLMService(
aiohttp_session=session, api_key=os.getenv("OPENAI_API_KEY"),
api_key=os.getenv("ELEVENLABS_API_KEY", ""), model="gpt-4o"
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), )
)
llm = OpenAILLMService( messages = [
api_key=os.getenv("OPENAI_API_KEY"), {
model="gpt-4o") "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 = [ tma_in = LLMUserResponseAggregator(messages)
{ tma_out = LLMAssistantResponseAggregator(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) pipeline = Pipeline([
tma_out = LLMAssistantResponseAggregator(messages) transport.input(),
tma_in,
llm,
tts,
transport.output(),
tma_out,
])
pipeline = Pipeline([ task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
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_first_participant_joined") @transport.event_handler("on_participant_left")
async def on_first_participant_joined(transport, participant): async def on_participant_left(transport, participant, reason):
transport.capture_participant_transcription(participant["id"]) await task.queue_frame(EndFrame())
await task.queue_frames([LLMMessagesFrame(messages)])
@transport.event_handler("on_participant_left") runner = PipelineRunner()
async def on_participant_left(transport, participant, reason):
await task.queue_frame(EndFrame())
runner = PipelineRunner() await runner.run(task)
await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":

148
examples/dialin-chatbot/bot_twilio.py
View File

@@ -1,5 +1,4 @@
import asyncio import asyncio
import aiohttp
import os import os
import sys import sys
import argparse 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 def main(room_url: str, token: str, callId: str, sipUri: str):
async with aiohttp.ClientSession() as session: # dialin_settings are only needed if Daily's SIP URI is used
# diallin_settings are only needed if Daily's SIP URI is used # If you are handling this via Twilio, Telnyx, set this to None
# If you are handling this via Twilio, Telnyx, set this to None # and handle call-forwarding when on_dialin_ready fires.
# and handle call-forwarding when on_dialin_ready fires. transport = DailyTransport(
transport = DailyTransport( room_url,
room_url, token,
token, "Chatbot",
"Chatbot", DailyParams(
DailyParams( api_key=daily_api_key,
api_key=daily_api_key, dialin_settings=None, # Not required for Twilio
dialin_settings=None, # Not required for Twilio audio_in_enabled=True,
audio_in_enabled=True, audio_out_enabled=True,
audio_out_enabled=True, camera_out_enabled=False,
camera_out_enabled=False, vad_enabled=True,
vad_enabled=True, vad_analyzer=SileroVADAnalyzer(),
vad_analyzer=SileroVADAnalyzer(), transcription_enabled=True,
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>'
) )
) except Exception as e:
raise Exception(f"Failed to forward call: {str(e)}")
tts = ElevenLabsTTSService( runner = PipelineRunner()
aiohttp_session=session, await runner.run(task)
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__": if __name__ == "__main__":

1
examples/foundational/05-sync-speech-and-image.py
View File

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

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

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

1
examples/foundational/06a-image-sync.py
View File

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

1
examples/foundational/07b-interruptible-langchain.py
View File

@@ -18,7 +18,6 @@ from pipecat.processors.aggregators.llm_response import (
LLMAssistantResponseAggregator, LLMUserResponseAggregator) LLMAssistantResponseAggregator, LLMUserResponseAggregator)
from pipecat.processors.frameworks.langchain import LangchainProcessor from pipecat.processors.frameworks.langchain import LangchainProcessor
from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer from pipecat.vad.silero import SileroVADAnalyzer

1
examples/foundational/11-sound-effects.py
View File

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

1
examples/simple-chatbot/bot.py
View File

@@ -111,7 +111,6 @@ async def main():
) )
tts = ElevenLabsTTSService( tts = ElevenLabsTTSService(
aiohttp_session=session,
api_key=os.getenv("ELEVENLABS_API_KEY"), api_key=os.getenv("ELEVENLABS_API_KEY"),
# #
# English # English

1
examples/storytelling-chatbot/src/bot.py
View File

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

1
pyproject.toml
View File

@@ -39,6 +39,7 @@ azure = [ "azure-cognitiveservices-speech~=1.40.0" ]
cartesia = [ "websockets~=12.0" ] cartesia = [ "websockets~=12.0" ]
daily = [ "daily-python~=0.10.1" ] daily = [ "daily-python~=0.10.1" ]
deepgram = [ "deepgram-sdk~=3.5.0" ] deepgram = [ "deepgram-sdk~=3.5.0" ]
elevenlabs = [ "elevenlabs~=1.7.0" ]
examples = [ "python-dotenv~=1.0.1", "flask~=3.0.3", "flask_cors~=4.0.1" ] examples = [ "python-dotenv~=1.0.1", "flask~=3.0.3", "flask_cors~=4.0.1" ]
fal = [ "fal-client~=0.4.1" ] fal = [ "fal-client~=0.4.1" ]
gladia = [ "websockets~=12.0" ] gladia = [ "websockets~=12.0" ]

2
src/pipecat/services/cartesia.py
View File

@@ -10,7 +10,7 @@ import base64
import asyncio import asyncio
import time import time
from typing import AsyncGenerator, Mapping from typing import AsyncGenerator
from pipecat.frames.frames import ( from pipecat.frames.frames import (
CancelFrame, CancelFrame,

80
src/pipecat/services/elevenlabs.py
View File

@@ -4,16 +4,36 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
import aiohttp
from typing import AsyncGenerator, Literal from typing import AsyncGenerator, Literal
from pydantic import BaseModel from pydantic import BaseModel
from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, TTSStartedFrame, TTSStoppedFrame from pipecat.frames.frames import AudioRawFrame, Frame, TTSStartedFrame, TTSStoppedFrame
from pipecat.services.ai_services import TTSService from pipecat.services.ai_services import TTSService
from loguru import logger from loguru import logger
# See .env.example for ElevenLabs configuration needed
try:
from elevenlabs.client import AsyncElevenLabs
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
class ElevenLabsTTSService(TTSService): class ElevenLabsTTSService(TTSService):
class InputParams(BaseModel): class InputParams(BaseModel):
@@ -24,21 +44,24 @@ class ElevenLabsTTSService(TTSService):
*, *,
api_key: str, api_key: str,
voice_id: str, voice_id: str,
aiohttp_session: aiohttp.ClientSession,
model: str = "eleven_turbo_v2_5", model: str = "eleven_turbo_v2_5",
params: InputParams = InputParams(), params: InputParams = InputParams(),
**kwargs): **kwargs):
super().__init__(**kwargs) super().__init__(**kwargs)
self._api_key = api_key
self._voice_id = voice_id self._voice_id = voice_id
self._model = model self._model = model
self._params = params self._params = params
self._aiohttp_session = aiohttp_session self._client = AsyncElevenLabs(api_key=api_key)
self._sample_rate = sample_rate_from_output_format(params.output_format)
def can_generate_metrics(self) -> bool: def can_generate_metrics(self) -> bool:
return True return True
async def set_model(self, model: str):
logger.debug(f"Switching TTS model to: [{model}]")
self._model = model
async def set_voice(self, voice: str): async def set_voice(self, voice: str):
logger.debug(f"Switching TTS voice to: [{voice}]") logger.debug(f"Switching TTS voice to: [{voice}]")
self._voice_id = voice self._voice_id = voice
@@ -46,34 +69,25 @@ class ElevenLabsTTSService(TTSService):
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating TTS: [{text}]") logger.debug(f"Generating TTS: [{text}]")
url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream" await self.start_tts_usage_metrics(text)
payload = {"text": text, "model_id": self._model}
querystring = {
"output_format": self._params.output_format
}
headers = {
"xi-api-key": self._api_key,
"Content-Type": "application/json",
}
await self.start_ttfb_metrics() await self.start_ttfb_metrics()
async with self._aiohttp_session.post(url, json=payload, headers=headers, params=querystring) as r: results = await self._client.generate(
if r.status != 200: text=text,
text = await r.text() voice=self._voice_id,
logger.error(f"{self} error getting audio (status: {r.status}, error: {text})") model=self._model,
yield ErrorFrame(f"Error getting audio (status: {r.status}, error: {text})") output_format=self._params.output_format
return )
await self.start_tts_usage_metrics(text) tts_started = False
async for audio in results:
# This is so we send TTSStartedFrame when we have the first audio
# bytes.
if not tts_started:
await self.push_frame(TTSStartedFrame())
tts_started = True
await self.stop_ttfb_metrics()
frame = AudioRawFrame(audio, self._sample_rate, 1)
yield frame
await self.push_frame(TTSStartedFrame()) await self.push_frame(TTSStoppedFrame())
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())