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

Add PlivoFrameSerializer

Browse Source
This commit is contained in:
Mark Backman
2025-05-23 18:56:23 -04:00
parent 2a6c01f634
commit e13c9fd42e
10 changed files with 761 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
CHANGELOG.md
View File

@@ -11,10 +11,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Added `GoogleHttpTTSService` which uses Google's HTTP TTS API.
- Added `TavusTransport`, a new transport implementation compatible with any
Pipecat pipeline. When using the `TavusTransport`the Pipecat bot will
- Added `TavusTransport`, a new transport implementation compatible with any
Pipecat pipeline. When using the `TavusTransport`the Pipecat bot will
connect in the same room as the Tavus Avatar and the user.
- Added `PlivoFrameSerializer` to support Plivo calls. A full running example
has also been added to `examples/plivo-chatbot`.
- Added `UserBotLatencyLogObserver`. This is an observer that logs the latency
between when the user stops speaking and when the bot starts speaking. This
gives you an initial idea on how quickly the AI services respond.
@@ -36,8 +39,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
correspond to the `StartFrame`, `StopFrame`, `EndFrame` and `CancelFrame`
respectively.
- Added additional languages to `LmntTTSService`. Languages include: `hi`, `id`,
`it`, `ja`, `nl`, `pl`, `ru`, `sv`, `th`, `tr`, `uk`, `vi`.
- Added additional languages to `LmntTTSService`. Languages include: `hi`,
`id`, `it`, `ja`, `nl`, `pl`, `ru`, `sv`, `th`, `tr`, `uk`, `vi`.
- Added a `model` parameter to the `LmntTTSService` constructor, allowing
switching between LMNT models.
@@ -75,8 +78,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
```
By default, Pipecat has implemented service decorators to trace execution of
STT, LLM, and TTS services. You can enable tracing by setting `enable_tracing`
to `True` in the PipelineTask.
STT, LLM, and TTS services. You can enable tracing by setting
`enable_tracing` to `True` in the PipelineTask.
- Added `TurnTrackingObserver`, which tracks the start and end of a user/bot
turn pair and emits events `on_turn_started` and `on_turn_stopped`
@@ -86,13 +89,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed
- Updated `GoogleTTSService` to use Google's streaming TTS API. The default voice also updated to `en-US-Chirp3-HD-Charon`.
- Updated `GoogleTTSService` to use Google's streaming TTS API. The default
voice also updated to `en-US-Chirp3-HD-Charon`.
- ⚠️Refactored the `TavusVideoService`, so it acts like a proxy, sending audio to
Tavus and receiving both audio and video. This will make `TavusVideoService` usable
with any Pipecat pipeline and with any transport. This is a **breaking change**,
check the `examples/foundational/21a-tavus-layer-small-webrtc.py` to see how to
use it.
- ⚠️ Refactored the `TavusVideoService`, so it acts like a proxy, sending audio
to Tavus and receiving both audio and video. This will make
`TavusVideoService` usable with any Pipecat pipeline and with any transport.
This is a **breaking change**, check the
`examples/foundational/21a-tavus-layer-small-webrtc.py` to see how to use it.
- `DailyTransport` now uses custom microphone audio tracks instead of virtual
microphones. Now, multiple Daily transports can be used in the same process.

161
examples/plivo-chatbot/.gitignore vendored Normal file
View File

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

20
examples/plivo-chatbot/Dockerfile Normal file
View File

@@ -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 /plivo-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"]

128
examples/plivo-chatbot/README.md Normal file
View File

@@ -0,0 +1,128 @@
# Plivo Chatbot
This project is a FastAPI-based chatbot that integrates with Plivo 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)
- [Configure Plivo URLs](#configure-plivo-urls)
- [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)
- Plivo 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**:
Copy the example environment file and update with your settings:
```sh
cp env.example .env
```
4. **Install ngrok**:
Follow the instructions on the [ngrok website](https://ngrok.com/download) to download and install ngrok.
## Configure Plivo URLs
1. **Start ngrok**:
In a new terminal, start ngrok to tunnel the local server:
```sh
ngrok http 8765
```
2. **Update the Plivo Application**:
- Go to your Plivo console and navigate to Voice > Applications > XML
- Select "Add New Application" or edit an existing one
- Set the Primary Answer URL to your ngrok URL (e.g., https://<ngrok_url>/)
- Ensure the Answer Method is set to POST
- Save the application
- Configure your number to use the newly created (or updated) application
- Phone Numbers > Active > Your number
- Select Application Type: XML Application
- Plivo Application: Your application
- Click "Update" to save
3. **Configure streams.xml**:
- Copy the template file to create your local version:
```sh
cp templates/streams.xml.template templates/streams.xml
```
- In `templates/streams.xml`, replace `<your server url>` with your ngrok URL (without `https://`)
- The final URL should look like: `wss://abc123.ngrok.io/ws`
4. **Assign the Application to a Plivo Number**:
- Go to Phone Numbers > Your Numbers in the Plivo console
- Edit your Plivo number
- Select the application you created/updated in the previous step
- Save the configuration
## Running the Application
Choose one of these two methods to run the application:
### Using Python (Option 1)
**Run the FastAPI application**:
```sh
# Make sure you're in the project directory and your virtual environment is activated
python server.py
```
### Using Docker (Option 2)
1. **Build the Docker image**:
```sh
docker build -t plivo-chatbot .
```
2. **Run the Docker container**:
```sh
docker run -it --rm -p 8765:8765 plivo-chatbot
```
The server will start on port 8765. Keep this running while you test with Plivo.
## Usage
To start a call, simply make a call to your configured Plivo phone number. The Answer URL will direct the call to your FastAPI application, which will handle it accordingly.
## Key Differences from Twilio
- Plivo uses `streamId` instead of `streamSid`
- Plivo uses `callId` instead of `callSid`
- Plivo uses `<Stream>` element instead of `<Connect><Stream>`
- Plivo's Stream element has `bidirectional`, `keepCallAlive`, and `contentType` attributes
- Plivo API authentication uses Auth ID and Auth Token (similar to Twilio's Account SID and Auth Token)

111
examples/plivo-chatbot/bot.py Normal file
View File

@@ -0,0 +1,111 @@
#
# Copyright (c) 2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
import sys
from typing import Optional
from dotenv import load_dotenv
from fastapi import WebSocket
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
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.serializers.plivo import PlivoFrameSerializer
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.network.fastapi_websocket import (
FastAPIWebsocketParams,
FastAPIWebsocketTransport,
)
load_dotenv()
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def run_bot(websocket_client: WebSocket, stream_id: str, call_id: Optional[str]):
logger.info(f"Starting bot for stream: {stream_id}")
serializer = PlivoFrameSerializer(
stream_id=stream_id,
call_id=call_id,
auth_id=os.getenv("PLIVO_AUTH_ID"),
auth_token=os.getenv("PLIVO_AUTH_TOKEN"),
)
transport = FastAPIWebsocketTransport(
websocket=websocket_client,
params=FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
add_wav_header=False,
vad_analyzer=SileroVADAnalyzer(),
serializer=serializer,
),
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
messages = [
{
"role": "system",
"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)
pipeline = Pipeline(
[
transport.input(), # Websocket input from client
stt, # Speech-To-Text
context_aggregator.user(),
llm, # LLM
tts, # Text-To-Speech
transport.output(), # Websocket output to client
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
audio_in_sample_rate=8000,
audio_out_sample_rate=8000,
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([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
await task.cancel()
# 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)

5
examples/plivo-chatbot/env.example Normal file
View File

@@ -0,0 +1,5 @@
OPENAI_API_KEY=
DEEPGRAM_API_KEY=
CARTESIA_API_KEY=
PLIVO_AUTH_ID=
PLIVO_AUTH_TOKEN=

5
examples/plivo-chatbot/requirements.txt Normal file
View File

@@ -0,0 +1,5 @@
pipecat-ai[cartesia,openai,silero,deepgram]
fastapi
uvicorn
python-dotenv
loguru

59
examples/plivo-chatbot/server.py Normal file
View File

@@ -0,0 +1,59 @@
#
# Copyright (c) 2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import json
import uvicorn
from bot import run_bot
from fastapi import FastAPI, WebSocket
from fastapi.middleware.cors import CORSMiddleware
from loguru import logger
from starlette.responses import HTMLResponse
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allow all origins for testing
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.post("/")
async def start_call():
print("POST Plivo XML")
return HTMLResponse(content=open("templates/streams.xml").read(), media_type="application/xml")
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
# Plivo sends a start event when the stream begins
start_data = websocket.iter_text()
start_message = json.loads(await start_data.__anext__())
print("Received start message:", start_message, flush=True)
# Extract stream_id and call_id from the start event
start_info = start_message.get("start", {})
stream_id = start_info.get("streamId")
call_id = start_info.get("callId")
if not stream_id:
logger.error("No streamId found in start message")
await websocket.close()
return
print(f"WebSocket connection accepted for stream: {stream_id}, call: {call_id}")
await run_bot(websocket, stream_id, call_id)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8765)

4
examples/plivo-chatbot/templates/streams.xml.template Normal file
View File

@@ -0,0 +1,4 @@
<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Stream bidirectional="true" keepCallAlive="true" contentType="audio/x-mulaw;rate=8000">wss://<your server url>/ws</Stream>
</Response>

252
src/pipecat/serializers/plivo.py Normal file
View File

@@ -0,0 +1,252 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import base64
import json
from typing import Optional
from loguru import logger
from pydantic import BaseModel
from pipecat.audio.utils import create_default_resampler, pcm_to_ulaw, ulaw_to_pcm
from pipecat.frames.frames import (
AudioRawFrame,
CancelFrame,
EndFrame,
Frame,
InputAudioRawFrame,
InputDTMFFrame,
KeypadEntry,
StartFrame,
StartInterruptionFrame,
TransportMessageFrame,
TransportMessageUrgentFrame,
)
from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType
class PlivoFrameSerializer(FrameSerializer):
"""Serializer for Plivo Audio Streaming WebSocket protocol.
This serializer handles converting between Pipecat frames and Plivo's WebSocket
audio streaming protocol. It supports audio conversion, DTMF events, and automatic
call termination.
When auto_hang_up is enabled (default), the serializer will automatically terminate
the Plivo call when an EndFrame or CancelFrame is processed, but requires Plivo
credentials to be provided.
Attributes:
_stream_id: The Plivo Stream ID.
_call_id: The associated Plivo Call ID.
_auth_id: Plivo auth ID for API access.
_auth_token: Plivo authentication token for API access.
_params: Configuration parameters.
_plivo_sample_rate: Sample rate used by Plivo (typically 8kHz).
_sample_rate: Input sample rate for the pipeline.
_resampler: Audio resampler for format conversion.
"""
class InputParams(BaseModel):
"""Configuration parameters for PlivoFrameSerializer.
Attributes:
plivo_sample_rate: Sample rate used by Plivo, defaults to 8000 Hz.
sample_rate: Optional override for pipeline input sample rate.
auto_hang_up: Whether to automatically terminate call on EndFrame.
"""
plivo_sample_rate: int = 8000
sample_rate: Optional[int] = None
auto_hang_up: bool = True
def __init__(
self,
stream_id: str,
call_id: Optional[str] = None,
auth_id: Optional[str] = None,
auth_token: Optional[str] = None,
params: Optional[InputParams] = None,
):
"""Initialize the PlivoFrameSerializer.
Args:
stream_id: The Plivo Stream ID.
call_id: The associated Plivo Call ID (optional, but required for auto hang-up).
auth_id: Plivo auth ID (required for auto hang-up).
auth_token: Plivo auth token (required for auto hang-up).
params: Configuration parameters.
"""
self._stream_id = stream_id
self._call_id = call_id
self._auth_id = auth_id
self._auth_token = auth_token
self._params = params or PlivoFrameSerializer.InputParams()
self._plivo_sample_rate = self._params.plivo_sample_rate
self._sample_rate = 0 # Pipeline input rate
self._resampler = create_default_resampler()
self._hangup_attempted = False
@property
def type(self) -> FrameSerializerType:
"""Gets the serializer type.
Returns:
The serializer type, either TEXT or BINARY.
"""
return FrameSerializerType.TEXT
async def setup(self, frame: StartFrame):
"""Sets up the serializer with pipeline configuration.
Args:
frame: The StartFrame containing pipeline configuration.
"""
self._sample_rate = self._params.sample_rate or frame.audio_in_sample_rate
async def serialize(self, frame: Frame) -> str | bytes | None:
"""Serializes a Pipecat frame to Plivo WebSocket format.
Handles conversion of various frame types to Plivo WebSocket messages.
For EndFrames, initiates call termination if auto_hang_up is enabled.
Args:
frame: The Pipecat frame to serialize.
Returns:
Serialized data as string or bytes, or None if the frame isn't handled.
"""
if (
self._params.auto_hang_up
and not self._hangup_attempted
and isinstance(frame, (EndFrame, CancelFrame))
):
self._hangup_attempted = True
await self._hang_up_call()
return None
elif isinstance(frame, StartInterruptionFrame):
answer = {"event": "clearAudio", "streamId": self._stream_id}
return json.dumps(answer)
elif isinstance(frame, AudioRawFrame):
data = frame.audio
# Output: Convert PCM at frame's rate to 8kHz μ-law for Plivo
serialized_data = await pcm_to_ulaw(
data, frame.sample_rate, self._plivo_sample_rate, self._resampler
)
payload = base64.b64encode(serialized_data).decode("utf-8")
answer = {
"event": "playAudio",
"media": {
"contentType": "audio/x-mulaw",
"sampleRate": self._plivo_sample_rate,
"payload": payload,
},
"streamId": self._stream_id,
}
return json.dumps(answer)
elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)):
return json.dumps(frame.message)
# Return None for unhandled frames
return None
async def _hang_up_call(self):
"""Hang up the Plivo call using Plivo's REST API."""
try:
import aiohttp
auth_id = self._auth_id
auth_token = self._auth_token
call_id = self._call_id
if not call_id or not auth_id or not auth_token:
missing = []
if not call_id:
missing.append("call_id")
if not auth_id:
missing.append("auth_id")
if not auth_token:
missing.append("auth_token")
logger.warning(
f"Cannot hang up Plivo call: missing required parameters: {', '.join(missing)}"
)
return
# Plivo API endpoint for hanging up calls
endpoint = f"https://api.plivo.com/v1/Account/{auth_id}/Call/{call_id}/"
# Create basic auth from auth_id and auth_token
auth = aiohttp.BasicAuth(auth_id, auth_token)
# Make the DELETE request to hang up the call
async with aiohttp.ClientSession() as session:
async with session.delete(endpoint, auth=auth) as response:
if response.status == 204: # Plivo returns 204 for successful hangup
logger.debug(f"Successfully terminated Plivo call {call_id}")
elif response.status == 404: # Call already ended
logger.debug(f"Plivo call {call_id} already terminated")
else:
# Get the error details for better debugging
error_text = await response.text()
logger.error(
f"Failed to terminate Plivo call {call_id}: "
f"Status {response.status}, Response: {error_text}"
)
except Exception as e:
logger.exception(f"Failed to hang up Plivo call: {e}")
async def deserialize(self, data: str | bytes) -> Frame | None:
"""Deserializes Plivo WebSocket data to Pipecat frames.
Handles conversion of Plivo media events to appropriate Pipecat frames.
Args:
data: The raw WebSocket data from Plivo.
Returns:
A Pipecat frame corresponding to the Plivo event, or None if unhandled.
"""
try:
message = json.loads(data)
except json.JSONDecodeError:
logger.warning(f"Failed to parse JSON message: {data}")
return None
if message.get("event") == "media":
media = message.get("media", {})
payload_base64 = media.get("payload")
if not payload_base64:
return None
payload = base64.b64decode(payload_base64)
# Input: Convert Plivo's 8kHz μ-law to PCM at pipeline input rate
deserialized_data = await ulaw_to_pcm(
payload, self._plivo_sample_rate, self._sample_rate, self._resampler
)
audio_frame = InputAudioRawFrame(
audio=deserialized_data, num_channels=1, sample_rate=self._sample_rate
)
return audio_frame
elif message.get("event") == "dtmf":
dtmf_data = message.get("dtmf", {})
digit = dtmf_data.get("digit")
if digit:
try:
return InputDTMFFrame(KeypadEntry(digit))
except ValueError:
# Handle case where string doesn't match any enum value
logger.warning(f"Invalid DTMF digit received: {digit}")
return None
else:
return None