From 51a86a509c64c04d0261d1ce164735b0183b8342 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 3 Feb 2025 10:35:11 -0800 Subject: [PATCH 1/2] examples: multiple twilio-chatbot improvements --- CHANGELOG.md | 5 + examples/twilio-chatbot/README.md | 31 +++++ examples/twilio-chatbot/bot.py | 63 +++++++++- examples/twilio-chatbot/client.py | 200 ++++++++++++++++++++++++++++++ examples/twilio-chatbot/server.py | 11 +- 5 files changed, 303 insertions(+), 7 deletions(-) create mode 100644 examples/twilio-chatbot/client.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ef8af2b3..3fd1a86d5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -101,6 +101,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Other +- Updated `twilio-chatbot` with a few new features: use 8000 sample rate and + avoid resampling, a new client useful for stress testing and testing locally + without the need to make phone calls. Also, added audio recording on both the + client and the server to make sure the audio sounds good. + - Updated examples to use `task.cancel()` to immediately exit the example when a participant leaves or disconnects, instead of pushing an `EndFrame`. Pushing an `EndFrame` causes the bot to run through everything that is internally diff --git a/examples/twilio-chatbot/README.md b/examples/twilio-chatbot/README.md index 32bd4de49..d06fd7f85 100644 --- a/examples/twilio-chatbot/README.md +++ b/examples/twilio-chatbot/README.md @@ -107,3 +107,34 @@ The server will start on port 8765. Keep this running while you test with Twilio ## Usage To start a call, simply make a call to your configured Twilio phone number. The webhook URL will direct the call to your FastAPI application, which will handle it accordingly. + +## Testing + +It is also possible to automatically test the server without making phone calls by using a software client. + +First, update `templates/streams.xml` to point to your server's websocket endpoint. For example: + +``` + + + + + + + +``` + +Then, start the server with `-t` to indicate we are testing: + +```sh +# Make sure you’re in the project directory and your virtual environment is activated +python server.py -t +``` + +Finally, just point the client to the server's URL: + +```sh +python client.py -u http://localhost:8765 -c 2 +``` + +where `-c` allows you to create multiple concurrent clients. diff --git a/examples/twilio-chatbot/bot.py b/examples/twilio-chatbot/bot.py index e715c02e1..e6511485c 100644 --- a/examples/twilio-chatbot/bot.py +++ b/examples/twilio-chatbot/bot.py @@ -4,10 +4,16 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import datetime +import io import os import sys +import wave +import aiofiles +from deepgram import LiveOptions from dotenv import load_dotenv +from fastapi import WebSocket from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer @@ -15,6 +21,7 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor from pipecat.serializers.twilio import TwilioFrameSerializer from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.deepgram import DeepgramSTTService @@ -30,38 +37,70 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def run_bot(websocket_client, stream_sid): +async def save_audio(server_name: str, audio: bytes, sample_rate: int, num_channels: int): + if len(audio) > 0: + filename = ( + f"{server_name}_recording_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav" + ) + with io.BytesIO() as buffer: + with wave.open(buffer, "wb") as wf: + wf.setsampwidth(2) + wf.setnchannels(num_channels) + wf.setframerate(sample_rate) + wf.writeframes(audio) + async with aiofiles.open(filename, "wb") as file: + await file.write(buffer.getvalue()) + logger.info(f"Merged audio saved to {filename}") + else: + logger.info("No audio data to save") + + +async def run_bot(websocket_client: WebSocket, stream_sid: str, testing: bool): transport = FastAPIWebsocketTransport( websocket=websocket_client, params=FastAPIWebsocketParams( + audio_in_enabled=True, audio_out_enabled=True, + audio_out_sample_rate=8000, add_wav_header=False, vad_enabled=True, - vad_analyzer=SileroVADAnalyzer(), + vad_analyzer=SileroVADAnalyzer(sample_rate=8000), vad_audio_passthrough=True, - serializer=TwilioFrameSerializer(stream_sid), + serializer=TwilioFrameSerializer( + stream_sid, TwilioFrameSerializer.InputParams(sample_rate=8000) + ), ), ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = DeepgramSTTService( + api_key=os.getenv("DEEPGRAM_API_KEY"), + live_options=LiveOptions(sample_rate=8000), + audio_passthrough=True, + ) tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + sample_rate=8000, + push_silence_after_stop=testing, ) messages = [ { "role": "system", - "content": "You are a helpful LLM in an audio 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.", + "content": "You are an elementary teacher in an audio call. Your output will be converted to audio so don't include special characters in your answers. Respond to what the student said in a short short sentence.", }, ] context = OpenAILLMContext(messages) context_aggregator = llm.create_context_aggregator(context) + # NOTE: Watch out! This will save all the conversation in memory. You can + # pass `buffer_size` to get periodic callbacks. + audiobuffer = AudioBufferProcessor(sample_rate=8000) + pipeline = Pipeline( [ transport.input(), # Websocket input from client @@ -70,6 +109,7 @@ async def run_bot(websocket_client, stream_sid): llm, # LLM tts, # Text-To-Speech transport.output(), # Websocket output to client + audiobuffer, # Used to buffer the audio in the pipeline context_aggregator.assistant(), ] ) @@ -78,6 +118,8 @@ async def run_bot(websocket_client, stream_sid): @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): + # Start recording. + await audiobuffer.start_recording() # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([context_aggregator.user().get_context_frame()]) @@ -86,6 +128,15 @@ async def run_bot(websocket_client, stream_sid): async def on_client_disconnected(transport, client): await task.cancel() - runner = PipelineRunner(handle_sigint=False) + @audiobuffer.event_handler("on_audio_data") + async def on_audio_data(buffer, audio, sample_rate, num_channels): + server_name = f"server_{websocket_client.client.port}" + await save_audio(server_name, audio, sample_rate, num_channels) + + # We use `handle_sigint=False` because `uvicorn` is controlling keyboard + # interruptions. We use `force_gc=True` to force garbage collection after + # the runner finishes running a task which could be useful for long running + # applications with multiple clients connecting. + runner = PipelineRunner(handle_sigint=False, force_gc=True) await runner.run(task) diff --git a/examples/twilio-chatbot/client.py b/examples/twilio-chatbot/client.py new file mode 100644 index 000000000..424ee80a4 --- /dev/null +++ b/examples/twilio-chatbot/client.py @@ -0,0 +1,200 @@ +# +# Copyright (c) 2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import asyncio +import datetime +import io +import os +import sys +import wave +import xml.etree.ElementTree as ET +from uuid import uuid4 + +import aiofiles +import aiohttp +from deepgram import LiveOptions +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.audio.vad.vad_analyzer import VADParams +from pipecat.frames.frames import EndFrame, TransportMessageUrgentFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor +from pipecat.serializers.twilio import TwilioFrameSerializer +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.deepgram import DeepgramSTTService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.network.websocket_client import ( + WebsocketClientParams, + WebsocketClientTransport, +) + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +DEFAULT_DURATION = 30 + + +async def download_twiml(server_url: str) -> str: + # TODO(aleix): add error checking. + async with aiohttp.ClientSession() as session: + async with session.post(server_url) as response: + return await response.text() + + +def get_stream_url_from_twiml(twiml: str) -> str: + root = ET.fromstring(twiml) + # TODO(aleix): add error checking. + stream_element = root.find(".//Stream") # Finds the first element + url = stream_element.get("url") + return url + + +async def save_audio(client_name: str, audio: bytes, sample_rate: int, num_channels: int): + if len(audio) > 0: + filename = ( + f"{client_name}_recording_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav" + ) + with io.BytesIO() as buffer: + with wave.open(buffer, "wb") as wf: + wf.setsampwidth(2) + wf.setnchannels(num_channels) + wf.setframerate(sample_rate) + wf.writeframes(audio) + async with aiofiles.open(filename, "wb") as file: + await file.write(buffer.getvalue()) + logger.info(f"Merged audio saved to {filename}") + else: + logger.info("No audio data to save") + + +async def run_client(client_name: str, server_url: str, duration_secs: int): + twiml = await download_twiml(server_url) + + stream_url = get_stream_url_from_twiml(twiml) + + stream_sid = str(uuid4()) + + transport = WebsocketClientTransport( + uri=stream_url, + params=WebsocketClientParams( + audio_in_enabled=True, + audio_out_enabled=True, + audio_out_sample_rate=8000, + add_wav_header=False, + serializer=TwilioFrameSerializer( + stream_sid, params=TwilioFrameSerializer.InputParams(sample_rate=8000) + ), + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=1.5), sample_rate=8000), + vad_audio_passthrough=True, + ), + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + + # We let the audio passthrough so we can record the conversation. + stt = DeepgramSTTService( + api_key=os.getenv("DEEPGRAM_API_KEY"), + live_options=LiveOptions(sample_rate=8000), + audio_passthrough=True, + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="e13cae5c-ec59-4f71-b0a6-266df3c9bb8e", # Madame Mischief + sample_rate=8000, + push_silence_after_stop=True, + ) + + messages = [ + { + "role": "system", + "content": "You are an 8 year old child. A teacher will explain you new concepts you want to know about. Feel free to change topics whnever you want. Once you are taught something you need to keep asking for clarifications if you think someone your age would not understand what you are being taught.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + # NOTE: Watch out! This will save all the conversation in memory. You can + # pass `buffer_size` to get periodic callbacks. + audiobuffer = AudioBufferProcessor(sample_rate=8000) + + pipeline = Pipeline( + [ + transport.input(), # Websocket input from server + stt, # Speech-To-Text + context_aggregator.user(), + llm, # LLM + tts, # Text-To-Speech + transport.output(), # Websocket output to server + audiobuffer, # Used to buffer the audio in the pipeline + context_aggregator.assistant(), + ] + ) + + task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) + + @transport.event_handler("on_connected") + async def on_connected(transport: WebsocketClientTransport, client): + # Start recording. + await audiobuffer.start_recording() + + message = TransportMessageUrgentFrame( + message={"event": "connected", "protocol": "Call", "version": "1.0.0"} + ) + await transport.output().send_message(message) + + message = TransportMessageUrgentFrame( + message={"event": "start", "streamSid": stream_sid, "start": {"streamSid": stream_sid}} + ) + await transport.output().send_message(message) + + @audiobuffer.event_handler("on_audio_data") + async def on_audio_data(buffer, audio, sample_rate, num_channels): + await save_audio(client_name, audio, sample_rate, num_channels) + + async def end_call(): + await asyncio.sleep(duration_secs) + await task.queue_frame(EndFrame()) + + runner = PipelineRunner() + + await asyncio.gather(runner.run(task), end_call()) + + +async def main(): + parser = argparse.ArgumentParser(description="Pipecat Twilio Chatbot Client") + parser.add_argument("-u", "--url", type=str, required=True, help="specify the server URL") + parser.add_argument( + "-c", "--clients", type=int, required=True, help="number of concurrent clients" + ) + parser.add_argument( + "-d", + "--duration", + type=int, + default=DEFAULT_DURATION, + help=f"duration of each client in seconds (default: {DEFAULT_DURATION})", + ) + args, _ = parser.parse_known_args() + + clients = [] + for i in range(args.clients): + clients.append(asyncio.create_task(run_client(f"client_{i}", args.url, args.duration))) + await asyncio.gather(*clients) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/twilio-chatbot/server.py b/examples/twilio-chatbot/server.py index bb67a86f3..d65015b5c 100644 --- a/examples/twilio-chatbot/server.py +++ b/examples/twilio-chatbot/server.py @@ -4,6 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import argparse import json import uvicorn @@ -38,8 +39,16 @@ async def websocket_endpoint(websocket: WebSocket): print(call_data, flush=True) stream_sid = call_data["start"]["streamSid"] print("WebSocket connection accepted") - await run_bot(websocket, stream_sid) + await run_bot(websocket, stream_sid, app.state.testing) if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Pipecat Twilio Chatbot Server") + parser.add_argument( + "-t", "--test", action="store_true", default=False, help="set the server in testing mode" + ) + args, _ = parser.parse_known_args() + + app.state.testing = args.test + uvicorn.run(app, host="0.0.0.0", port=8765) From 1cdb66f8895fe56ca6b8a9b574fbfbb30d87317d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 3 Feb 2025 10:58:06 -0800 Subject: [PATCH 2/2] examples(twilio-chatbot): create sample rate variable --- examples/twilio-chatbot/bot.py | 14 ++++++++------ examples/twilio-chatbot/client.py | 21 ++++++++++++--------- 2 files changed, 20 insertions(+), 15 deletions(-) diff --git a/examples/twilio-chatbot/bot.py b/examples/twilio-chatbot/bot.py index e6511485c..64c3c9cb5 100644 --- a/examples/twilio-chatbot/bot.py +++ b/examples/twilio-chatbot/bot.py @@ -36,6 +36,8 @@ load_dotenv(override=True) logger.remove(0) logger.add(sys.stderr, level="DEBUG") +SAMPLE_RATE = 8000 + async def save_audio(server_name: str, audio: bytes, sample_rate: int, num_channels: int): if len(audio) > 0: @@ -61,13 +63,13 @@ async def run_bot(websocket_client: WebSocket, stream_sid: str, testing: bool): params=FastAPIWebsocketParams( audio_in_enabled=True, audio_out_enabled=True, - audio_out_sample_rate=8000, + audio_out_sample_rate=SAMPLE_RATE, add_wav_header=False, vad_enabled=True, - vad_analyzer=SileroVADAnalyzer(sample_rate=8000), + vad_analyzer=SileroVADAnalyzer(sample_rate=SAMPLE_RATE), vad_audio_passthrough=True, serializer=TwilioFrameSerializer( - stream_sid, TwilioFrameSerializer.InputParams(sample_rate=8000) + stream_sid, TwilioFrameSerializer.InputParams(sample_rate=SAMPLE_RATE) ), ), ) @@ -76,14 +78,14 @@ async def run_bot(websocket_client: WebSocket, stream_sid: str, testing: bool): stt = DeepgramSTTService( api_key=os.getenv("DEEPGRAM_API_KEY"), - live_options=LiveOptions(sample_rate=8000), + live_options=LiveOptions(sample_rate=SAMPLE_RATE), audio_passthrough=True, ) tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady - sample_rate=8000, + sample_rate=SAMPLE_RATE, push_silence_after_stop=testing, ) @@ -99,7 +101,7 @@ async def run_bot(websocket_client: WebSocket, stream_sid: str, testing: bool): # NOTE: Watch out! This will save all the conversation in memory. You can # pass `buffer_size` to get periodic callbacks. - audiobuffer = AudioBufferProcessor(sample_rate=8000) + audiobuffer = AudioBufferProcessor(sample_rate=SAMPLE_RATE) pipeline = Pipeline( [ diff --git a/examples/twilio-chatbot/client.py b/examples/twilio-chatbot/client.py index 424ee80a4..fa710773b 100644 --- a/examples/twilio-chatbot/client.py +++ b/examples/twilio-chatbot/client.py @@ -43,7 +43,8 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -DEFAULT_DURATION = 30 +DEFAULT_CLIENT_DURATION = 30 +SAMPLE_RATE = 8000 async def download_twiml(server_url: str) -> str: @@ -91,13 +92,15 @@ async def run_client(client_name: str, server_url: str, duration_secs: int): params=WebsocketClientParams( audio_in_enabled=True, audio_out_enabled=True, - audio_out_sample_rate=8000, + audio_out_sample_rate=SAMPLE_RATE, add_wav_header=False, serializer=TwilioFrameSerializer( - stream_sid, params=TwilioFrameSerializer.InputParams(sample_rate=8000) + stream_sid, params=TwilioFrameSerializer.InputParams(sample_rate=SAMPLE_RATE) ), vad_enabled=True, - vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=1.5), sample_rate=8000), + vad_analyzer=SileroVADAnalyzer( + params=VADParams(stop_secs=1.5), sample_rate=SAMPLE_RATE + ), vad_audio_passthrough=True, ), ) @@ -107,14 +110,14 @@ async def run_client(client_name: str, server_url: str, duration_secs: int): # We let the audio passthrough so we can record the conversation. stt = DeepgramSTTService( api_key=os.getenv("DEEPGRAM_API_KEY"), - live_options=LiveOptions(sample_rate=8000), + live_options=LiveOptions(sample_rate=SAMPLE_RATE), audio_passthrough=True, ) tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="e13cae5c-ec59-4f71-b0a6-266df3c9bb8e", # Madame Mischief - sample_rate=8000, + sample_rate=SAMPLE_RATE, push_silence_after_stop=True, ) @@ -130,7 +133,7 @@ async def run_client(client_name: str, server_url: str, duration_secs: int): # NOTE: Watch out! This will save all the conversation in memory. You can # pass `buffer_size` to get periodic callbacks. - audiobuffer = AudioBufferProcessor(sample_rate=8000) + audiobuffer = AudioBufferProcessor(sample_rate=SAMPLE_RATE) pipeline = Pipeline( [ @@ -185,8 +188,8 @@ async def main(): "-d", "--duration", type=int, - default=DEFAULT_DURATION, - help=f"duration of each client in seconds (default: {DEFAULT_DURATION})", + default=DEFAULT_CLIENT_DURATION, + help=f"duration of each client in seconds (default: {DEFAULT_CLIENT_DURATION})", ) args, _ = parser.parse_known_args()