From 5753869e5ea0f5d5c549b3010aa44fb72454202e Mon Sep 17 00:00:00 2001 From: daniil5701133 Date: Wed, 19 Jun 2024 15:51:38 +0300 Subject: [PATCH] add twilio-chatbot example with README.md info how to start app created twilio_websocket_service.py, TwilioFrameSerializer.py moved pcm_16000_to_ulaw_8000 and ulaw_8000_to_pcm_16000 to src/pipecat/utils/audio.py fixed callback on disconnect --- examples/twilio-chatbot/.env.example | 4 + examples/twilio-chatbot/.gitignore | 161 ++++++++++++++++++ examples/twilio-chatbot/Dockerfile | 20 +++ examples/twilio-chatbot/README.md | 94 ++++++++++ examples/twilio-chatbot/requirements.txt | 5 + examples/twilio-chatbot/server.py | 32 ++++ examples/twilio-chatbot/templates/streams.xml | 7 + examples/twilio-chatbot/test_bot.py | 88 ++++++++++ .../serializers/TwilioFrameSerializer.py | 42 +++++ .../transports/network/fastapi_websocket.py | 132 ++++++++++++++ src/pipecat/utils/audio.py | 19 +++ 11 files changed, 604 insertions(+) create mode 100644 examples/twilio-chatbot/.env.example create mode 100644 examples/twilio-chatbot/.gitignore create mode 100644 examples/twilio-chatbot/Dockerfile create mode 100644 examples/twilio-chatbot/README.md create mode 100644 examples/twilio-chatbot/requirements.txt create mode 100644 examples/twilio-chatbot/server.py create mode 100644 examples/twilio-chatbot/templates/streams.xml create mode 100644 examples/twilio-chatbot/test_bot.py create mode 100644 src/pipecat/serializers/TwilioFrameSerializer.py create mode 100644 src/pipecat/transports/network/fastapi_websocket.py diff --git a/examples/twilio-chatbot/.env.example b/examples/twilio-chatbot/.env.example new file mode 100644 index 000000000..069f74445 --- /dev/null +++ b/examples/twilio-chatbot/.env.example @@ -0,0 +1,4 @@ +OPENAI_API_KEY= +DEEPGRAM_API_KEY= +ELEVENLABS_API_KEY= +ELEVENLABS_VOICE_ID= \ No newline at end of file diff --git a/examples/twilio-chatbot/.gitignore b/examples/twilio-chatbot/.gitignore new file mode 100644 index 000000000..2bc1403d1 --- /dev/null +++ b/examples/twilio-chatbot/.gitignore @@ -0,0 +1,161 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/#use-with-ide +.pdm.toml + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ +runpod.toml diff --git a/examples/twilio-chatbot/Dockerfile b/examples/twilio-chatbot/Dockerfile new file mode 100644 index 000000000..6e22d4cda --- /dev/null +++ b/examples/twilio-chatbot/Dockerfile @@ -0,0 +1,20 @@ +# Use an official Python runtime as a parent image +FROM python:3.10-bullseye + +# Set the working directory in the container +WORKDIR /twilio-chatbot + +# Copy the requirements file into the container +COPY requirements.txt . + +# Install any needed packages specified in requirements.txt +RUN pip install --no-cache-dir -r requirements.txt + +# Copy the current directory contents into the container +COPY . . + +# Expose the desired port +EXPOSE 8765 + +# Run the application +CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8765"] diff --git a/examples/twilio-chatbot/README.md b/examples/twilio-chatbot/README.md new file mode 100644 index 000000000..66a626ef7 --- /dev/null +++ b/examples/twilio-chatbot/README.md @@ -0,0 +1,94 @@ +# Twilio Chatbot + +This project is a FastAPI-based chatbot that integrates with Twilio to handle WebSocket connections and provide real-time communication. The project includes endpoints for starting a call and handling WebSocket connections. + +## Table of Contents + +- [Features](#features) +- [Requirements](#requirements) +- [Installation](#installation) +- [Running the Application](#running-the-application) +- [Usage](#usage) + +## Features + +- **FastAPI**: A modern, fast (high-performance), web framework for building APIs with Python 3.6+. +- **WebSocket Support**: Real-time communication using WebSockets. +- **CORS Middleware**: Allowing cross-origin requests for testing. +- **Dockerized**: Easily deployable using Docker. + +## Requirements + +- Python 3.10 +- Docker (for containerized deployment) +- ngrok (for tunneling) +- Twilio Account + +## Installation + +1. **Set up a virtual environment** (optional but recommended): + ```sh + python -m venv venv + source venv/bin/activate # On Windows, use `venv\Scripts\activate` + ``` + +2. **Install dependencies**: + ```sh + pip install -r requirements.txt + ``` + +3. **Create .env**: + create .env based on .env.example + +4. **Install ngrok**: + Follow the instructions on the [ngrok website](https://ngrok.com/download) to download and install ngrok. + +## Running the Application + +### Using Python + +1. **Run the FastAPI application**: + ```sh + python server.py + ``` + +2. **Start ngrok**: + In a new terminal, start ngrok to tunnel the local server: + ```sh + ngrok http 8765 + ``` + +3. **Update the Twilio Webhook**: + Copy the ngrok URL and update your Twilio phone number webhook URL to `http:///start_call`. + +3. **Update the streams.xml**: + Copy the ngrok URL and update your .xml URL to `wss:///ws`. + +### Using Docker + +1. **Build the Docker image**: + ```sh + docker build -t twilio-chatbot . + ``` + +2. **Run the Docker container**: + ```sh + docker build -t twilio-chatbot . + docker run -it --rm -p 8765:8765 twilio-chatbot + ``` + +3. **Start ngrok**: + In a new terminal, start ngrok to tunnel the local server: + ```sh + ngrok http 8765 + ``` + +4. **Update the Twilio Webhook**: + Copy the ngrok URL and update your Twilio phone number webhook URL to `http:///start_call`. + +5. **Update the streams.xml**: + Copy the ngrok URL and update your .xml URL to `wss:///ws`. + +## Usage + +To start a call, simply make a call to your Twilio phone number. The webhook URL will direct the call to your FastAPI application, which will handle it accordingly. diff --git a/examples/twilio-chatbot/requirements.txt b/examples/twilio-chatbot/requirements.txt new file mode 100644 index 000000000..f0456fcd5 --- /dev/null +++ b/examples/twilio-chatbot/requirements.txt @@ -0,0 +1,5 @@ +pipecat-ai[daily,openai,silero,deepgram] +fastapi +uvicorn +python-dotenv +loguru diff --git a/examples/twilio-chatbot/server.py b/examples/twilio-chatbot/server.py new file mode 100644 index 000000000..b671d1fd6 --- /dev/null +++ b/examples/twilio-chatbot/server.py @@ -0,0 +1,32 @@ +from fastapi import FastAPI, WebSocket, WebSocketDisconnect +from fastapi.middleware.cors import CORSMiddleware +import uvicorn +from starlette.responses import HTMLResponse +from test_bot import run_bot + +app = FastAPI() + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], # Allow all origins for testing + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + + +@app.post('/start_call') +async def start_call(): + print("POST TwiML") + return HTMLResponse(content=open("templates/streams.xml").read(), media_type="application/xml") + + +@app.websocket("/ws") +async def websocket_endpoint(websocket: WebSocket): + await websocket.accept() + print("WebSocket connection accepted") + await run_bot(websocket) + + +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=8765) diff --git a/examples/twilio-chatbot/templates/streams.xml b/examples/twilio-chatbot/templates/streams.xml new file mode 100644 index 000000000..1ab00a0e8 --- /dev/null +++ b/examples/twilio-chatbot/templates/streams.xml @@ -0,0 +1,7 @@ + + + + + + + \ No newline at end of file diff --git a/examples/twilio-chatbot/test_bot.py b/examples/twilio-chatbot/test_bot.py new file mode 100644 index 000000000..d49bef0aa --- /dev/null +++ b/examples/twilio-chatbot/test_bot.py @@ -0,0 +1,88 @@ +import aiohttp +import os +import sys + +from pipecat.frames.frames import LLMMessagesFrame, Frame, AudioRawFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantResponseAggregator, + LLMUserResponseAggregator +) +from pipecat.processors.frame_processor import FrameProcessor, FrameDirection + +from pipecat.services.openai import OpenAILLMService +from pipecat.services.deepgram import DeepgramSTTService +from pipecat.services.elevenlabs import ElevenLabsTTSService +from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketTransport, FastAPIWebsocketParams +from pipecat.vad.silero import SileroVADAnalyzer + +from loguru import logger + +from dotenv import load_dotenv +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def run_bot(websocket_client): + async with aiohttp.ClientSession() as session: + transport = FastAPIWebsocketTransport( + websocket=websocket_client, + params=FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + add_wav_header=False, + transcription_enabled=False, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True + ) + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o") + + stt = DeepgramSTTService(api_key=os.getenv('DEEPGRAM_API_KEY')) + + tts = ElevenLabsTTSService( + aiohttp_session=session, + api_key=os.getenv("ELEVENLABS_API_KEY"), + voice_id=os.getenv("ELEVENLABS_VOICE_ID"), + ) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) + + pipeline = Pipeline([ + transport.input(), # Websocket input from client + stt, # Speech-To-Text + tma_in, # User responses + llm, # LLM + tts, # Text-To-Speech + transport.output(), # Websocket output to client + tma_out # LLM responses + ]) + + task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + # Kick off the conversation. + messages.append( + {"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner(handle_sigint=False) + + await runner.run(task) diff --git a/src/pipecat/serializers/TwilioFrameSerializer.py b/src/pipecat/serializers/TwilioFrameSerializer.py new file mode 100644 index 000000000..5ac6227dd --- /dev/null +++ b/src/pipecat/serializers/TwilioFrameSerializer.py @@ -0,0 +1,42 @@ +import base64 +import json + +from pipecat.frames.frames import AudioRawFrame, Frame +from pipecat.serializers.base_serializer import FrameSerializer +from pipecat.utils.audio import ulaw_8000_to_pcm_16000, pcm_16000_to_ulaw_8000 + + +class TwilioFrameSerializer(FrameSerializer): + SERIALIZABLE_TYPES = { + AudioRawFrame: "audio", + } + + def __init__(self): + self.sid = None + + + def serialize(self, frame: AudioRawFrame) -> dict: + data = frame.audio + + serialized_data = pcm_16000_to_ulaw_8000(data) + payload = base64.b64encode(serialized_data).decode('utf-8') + answer_dict = {"event": "media", + "streamSid": self.sid, + "media": {"payload": payload}} + + return answer_dict + + def deserialize(self, message: bytes) -> AudioRawFrame | None: + data = json.loads(message) + if not self.sid: + self.sid = data['streamSid'] if data.get("streamSid") else None + + if data['event'] != 'media': + return None + else: + payload_base64 = data['media']['payload'] + payload = base64.b64decode(payload_base64) + + deserialized_data = ulaw_8000_to_pcm_16000(payload) + audio_frame = AudioRawFrame(audio=deserialized_data, num_channels=1, sample_rate=16000) + return audio_frame diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py new file mode 100644 index 000000000..fedcd9b4c --- /dev/null +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -0,0 +1,132 @@ +import asyncio +import io +import wave +from fastapi import WebSocket + +from typing import Awaitable, Callable +from pydantic.main import BaseModel + +from pipecat.serializers.TwilioFrameSerializer import TwilioFrameSerializer +from pipecat.frames.frames import AudioRawFrame, StartFrame +from pipecat.processors.frame_processor import FrameProcessor +from pipecat.serializers.base_serializer import FrameSerializer +from pipecat.transports.base_input import BaseInputTransport +from pipecat.transports.base_output import BaseOutputTransport +from pipecat.transports.base_transport import BaseTransport, TransportParams + +from loguru import logger + + +class FastAPIWebsocketParams(TransportParams): + add_wav_header: bool = False + audio_frame_size: int = 6400 # 200ms + serializer: FrameSerializer = TwilioFrameSerializer() + + +class FastAPIWebsocketCallbacks(BaseModel): + on_client_connected: Callable[[WebSocket], Awaitable[None]] + on_client_disconnected: Callable[[WebSocket], Awaitable[None]] + + +class FastAPIWebsocketInputTransport(BaseInputTransport): + + def __init__(self, websocket: WebSocket, params: FastAPIWebsocketParams, callbacks: FastAPIWebsocketCallbacks, **kwargs): + super().__init__(params, **kwargs) + + self._websocket = websocket + self._params = params + self._callbacks = callbacks + + async def start(self, frame: StartFrame): + await self._callbacks.on_client_connected(self._websocket) + await super().start(frame) + self._receive_task = self.get_event_loop().create_task(self._receive_messages()) + + async def stop(self): + await self._websocket.close() + await super().stop() + + async def _receive_messages(self): + async for message in self._websocket.iter_text(): + frame = self._params.serializer.deserialize(message) + + if not frame: + continue + + if isinstance(frame, AudioRawFrame): + self.push_audio_frame(frame) + + await self._callbacks.on_client_disconnected(self._websocket) + +class FastAPIWebsocketOutputTransport(BaseOutputTransport): + + def __init__(self, websocket: WebSocket, params: FastAPIWebsocketParams, **kwargs): + super().__init__(params, **kwargs) + + self._websocket = websocket + self._params = params + self._audio_buffer = bytes() + + def write_raw_audio_frames(self, frames: bytes): + self._audio_buffer += frames + while len(self._audio_buffer) >= self._params.audio_frame_size: + frame = AudioRawFrame( + audio=self._audio_buffer[:self._params.audio_frame_size], + sample_rate=self._params.audio_out_sample_rate, + num_channels=self._params.audio_out_channels + ) + + if self._params.add_wav_header: + content = io.BytesIO() + ww = wave.open(content, "wb") + ww.setsampwidth(2) + ww.setnchannels(frame.num_channels) + ww.setframerate(frame.sample_rate) + ww.writeframes(frame.audio) + ww.close() + content.seek(0) + wav_frame = AudioRawFrame( + content.read(), + sample_rate=frame.sample_rate, + num_channels=frame.num_channels) + frame = wav_frame + + payload = self._params.serializer.serialize(frame) + + future = asyncio.run_coroutine_threadsafe( + self._websocket.send_json(payload), self.get_event_loop()) + future.result() + + self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:] + + +class FastAPIWebsocketTransport(BaseTransport): + + def __init__(self, websocket: WebSocket, params: FastAPIWebsocketParams = FastAPIWebsocketParams(), input_name: str | None = None, output_name: str | None = None, loop: asyncio.AbstractEventLoop | None = None): + super().__init__(input_name=input_name, output_name=output_name, loop=loop) + self._params = params + + self._callbacks = FastAPIWebsocketCallbacks( + on_client_connected=self._on_client_connected, + on_client_disconnected=self._on_client_disconnected + ) + + self._input = FastAPIWebsocketInputTransport(websocket, self._params, self._callbacks, name=self._input_name) + self._output = FastAPIWebsocketOutputTransport(websocket, self._params, name=self._output_name) + + # Register supported handlers. The user will only be able to register + # these handlers. + self._register_event_handler("on_client_connected") + self._register_event_handler("on_client_disconnected") + + def input(self) -> FrameProcessor: + return self._input + + def output(self) -> FrameProcessor: + return self._output + + async def _on_client_connected(self, websocket): + await self._call_event_handler("on_client_connected", websocket) + + async def _on_client_disconnected(self, websocket): + await self._call_event_handler("on_client_disconnected", websocket) diff --git a/src/pipecat/utils/audio.py b/src/pipecat/utils/audio.py index 3c1118fc5..e1ee4698b 100644 --- a/src/pipecat/utils/audio.py +++ b/src/pipecat/utils/audio.py @@ -4,6 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import audioop import numpy as np import pyloudnorm as pyln @@ -31,3 +32,21 @@ def calculate_audio_volume(audio: bytes, sample_rate: int) -> float: def exp_smoothing(value: float, prev_value: float, factor: float) -> float: return prev_value + factor * (value - prev_value) + +def ulaw_8000_to_pcm_16000(ulaw_8000_bytes): + # Convert μ-law to PCM + pcm_8000_bytes = audioop.ulaw2lin(ulaw_8000_bytes, 2) + + # Resample from 8000 Hz to 16000 Hz + pcm_16000_bytes = audioop.ratecv(pcm_8000_bytes, 2, 1, 8000, 16000, None)[0] + + return pcm_16000_bytes + +def pcm_16000_to_ulaw_8000(pcm_16000_bytes): + # Resample from 16000 Hz to 8000 Hz + pcm_8000_bytes = audioop.ratecv(pcm_16000_bytes, 2, 1, 16000, 8000, None)[0] + + # Convert PCM to μ-law + ulaw_8000_bytes = audioop.lin2ulaw(pcm_8000_bytes, 2) + + return ulaw_8000_bytes \ No newline at end of file