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

Merge pull request #1125 from pipecat-ai/aleix/twilio-chatbot-improvements

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-02-03 11:10:33 -08:00
committed by GitHub
parent 824898f7b7 1cdb66f889
commit cc54255c41
5 changed files with 308 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGELOG.md
View File

@@ -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

31
examples/twilio-chatbot/README.md
View File

@@ -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:
```
<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Connect>
<Stream url="ws://localhost:8765/ws"></Stream>
</Connect>
<Pause length="40"/>
</Response>
```
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.

65
examples/twilio-chatbot/bot.py
View File

@@ -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
@@ -29,39 +36,73 @@ load_dotenv(override=True)
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
SAMPLE_RATE = 8000
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=SAMPLE_RATE,
add_wav_header=False,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
vad_analyzer=SileroVADAnalyzer(sample_rate=SAMPLE_RATE),
vad_audio_passthrough=True,
serializer=TwilioFrameSerializer(stream_sid),
serializer=TwilioFrameSerializer(
stream_sid, TwilioFrameSerializer.InputParams(sample_rate=SAMPLE_RATE)
),
),
)
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=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=SAMPLE_RATE,
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=SAMPLE_RATE)
pipeline = Pipeline(
[
transport.input(), # Websocket input from client
@@ -70,6 +111,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 +120,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 +130,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)

203
examples/twilio-chatbot/client.py Normal file
View File

@@ -0,0 +1,203 @@
#
# 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_CLIENT_DURATION = 30
SAMPLE_RATE = 8000
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 <Stream> 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=SAMPLE_RATE,
add_wav_header=False,
serializer=TwilioFrameSerializer(
stream_sid, params=TwilioFrameSerializer.InputParams(sample_rate=SAMPLE_RATE)
),
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(
params=VADParams(stop_secs=1.5), sample_rate=SAMPLE_RATE
),
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=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=SAMPLE_RATE,
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=SAMPLE_RATE)
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_CLIENT_DURATION,
help=f"duration of each client in seconds (default: {DEFAULT_CLIENT_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())

11
examples/twilio-chatbot/server.py
View File

@@ -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)