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

Merge pull request #1911 from pipecat-ai/filipi/websocket_transport_example

Browse Source 
Adding support to WebsocketTransport
This commit is contained in:
Filipi da Silva Fuchter
2025-06-06 17:25:07 -03:00
committed by GitHub
parent 07fb1a2c39 1e74476a71
commit c9575eaef9
23 changed files with 2693 additions and 463 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
CHANGELOG.md
View File

@@ -26,6 +26,12 @@ asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
- Fixed a typo in Livekit transport that prevented initialization.
### Added
- Added an `websocket` example, showing how to use the new Pipecat client
`WebsocketTransport` to connect with Pipecat `FastAPIWebsocketTransport` or
`WebsocketServerTransport`.
## [0.0.69] - 2025-06-02 "AI Engineer World's Fair release" ✨
### Added

15
examples/websocket-server/Dockerfile
View File

@@ -1,15 +0,0 @@
FROM python:3.10-bullseye
RUN mkdir /app
COPY *.py /app/
COPY requirements.txt /app/
COPY .env /app/
WORKDIR /app
RUN pip3 install -r requirements.txt
EXPOSE 7860
CMD ["python3", "bot.py"]

28
examples/websocket-server/README.md
View File

@@ -1,28 +0,0 @@
# Websocket Server
This is an example that shows how to use `WebsocketServerTransport` to communicate with a web client.
## Get started
```python
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
cp env.example .env # and add your credentials
```
## Run the bot
```bash
python bot.py
```
## Run the HTTP server
This will host the static web client:
```bash
python -m http.server
```
Then, visit `http://localhost:8000` in your browser to start a session.

153
examples/websocket-server/bot.py
View File

@@ -1,153 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
import sys
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import BotInterruptionFrame, EndFrame
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.protobuf import ProtobufFrameSerializer
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.network.websocket_server import (
WebsocketServerParams,
WebsocketServerTransport,
)
load_dotenv(override=True)
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
class SessionTimeoutHandler:
"""Handles actions to be performed when a session times out.
Inputs:
- task: Pipeline task (used to queue frames).
- tts: TTS service (used to generate speech output).
"""
def __init__(self, task, tts):
self.task = task
self.tts = tts
self.background_tasks = set()
async def handle_timeout(self, client_address):
"""Handles the timeout event for a session."""
try:
logger.info(f"Connection timeout for {client_address}")
# Queue a BotInterruptionFrame to notify the user
await self.task.queue_frames([BotInterruptionFrame()])
# Send the TTS message to inform the user about the timeout
await self.tts.say(
"I'm sorry, we are ending the call now. Please feel free to reach out again if you need assistance."
)
# Start the process to gracefully end the call in the background
end_call_task = asyncio.create_task(self._end_call())
self.background_tasks.add(end_call_task)
end_call_task.add_done_callback(self.background_tasks.discard)
except Exception as e:
logger.error(f"Error during session timeout handling: {e}")
async def _end_call(self):
"""Completes the session termination process after the TTS message."""
try:
# Wait for a duration to ensure TTS has completed
await asyncio.sleep(15)
# Queue both BotInterruptionFrame and EndFrame to conclude the session
await self.task.queue_frames([BotInterruptionFrame(), EndFrame()])
logger.info("TTS completed and EndFrame pushed successfully.")
except Exception as e:
logger.error(f"Error during call termination: {e}")
async def main():
transport = WebsocketServerTransport(
params=WebsocketServerParams(
serializer=ProtobufFrameSerializer(),
audio_in_enabled=True,
audio_out_enabled=True,
add_wav_header=True,
vad_analyzer=SileroVADAnalyzer(),
session_timeout=60 * 3, # 3 minutes
)
)
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 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.",
},
]
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=16000,
audio_out_sample_rate=16000,
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_session_timeout")
async def on_session_timeout(transport, client):
logger.info(f"Entering in timeout for {client.remote_address}")
timeout_handler = SessionTimeoutHandler(task, tts)
await timeout_handler.handle_timeout(client)
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":
asyncio.run(main())

8
examples/websocket-server/env.example
View File

@@ -1,8 +0,0 @@
# OpenAI API Key
OPENAI_API_KEY=your_openai_api_key_here
# Deepgram API Key
DEEPGRAM_API_KEY=your_deepgram_api_key_here
# Cartesia API Key
CARTESIA_API_KEY=your_cartesia_api_key_here

44
examples/websocket-server/frames.proto
View File

@@ -1,44 +0,0 @@
//
// Copyright (c) 2024–2025, Daily
//
// SPDX-License-Identifier: BSD 2-Clause License
//
// Generate frames_pb2.py with:
//
// python -m grpc_tools.protoc --proto_path=./ --python_out=./protobufs frames.proto
syntax = "proto3";
package pipecat;
message TextFrame {
uint64 id = 1;
string name = 2;
string text = 3;
}
message AudioRawFrame {
uint64 id = 1;
string name = 2;
bytes audio = 3;
uint32 sample_rate = 4;
uint32 num_channels = 5;
optional uint64 pts = 6;
}
message TranscriptionFrame {
uint64 id = 1;
string name = 2;
string text = 3;
string user_id = 4;
string timestamp = 5;
}
message Frame {
oneof frame {
TextFrame text = 1;
AudioRawFrame audio = 2;
TranscriptionFrame transcription = 3;
}
}

211
examples/websocket-server/index.html
View File

@@ -1,211 +0,0 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<script src="https://cdn.jsdelivr.net/npm/protobufjs@7.X.X/dist/protobuf.min.js"></script>
<title>Pipecat WebSocket Client Example</title>
</head>
<body>
<h1>Pipecat WebSocket Client Example</h1>
<h3><div id="progressText">Loading, wait...</div></h3>
<button id="startAudioBtn">Start Audio</button>
<button id="stopAudioBtn">Stop Audio</button>
<script>
const SAMPLE_RATE = 16000;
const NUM_CHANNELS = 1;
const PLAY_TIME_RESET_THRESHOLD_MS = 1.0;
// The protobuf type. We will load it later.
let Frame = null;
// The websocket connection.
let ws = null;
// The audio context
let audioContext = null;
// The audio context media stream source
let source = null;
// The microphone stream from getUserMedia. SHould be sampled to the
// proper sample rate.
let microphoneStream = null;
// Script processor to get data from microphone.
let scriptProcessor = null;
// AudioContext play time.
let playTime = 0;
// Last time we received a websocket message.
let lastMessageTime = 0;
// Whether we should be playing audio.
let isPlaying = false;
let startBtn = document.getElementById('startAudioBtn');
let stopBtn = document.getElementById('stopAudioBtn');
const proto = protobuf.load('frames.proto', (err, root) => {
if (err) {
throw err;
}
Frame = root.lookupType('pipecat.Frame');
const progressText = document.getElementById('progressText');
progressText.textContent = 'We are ready! Make sure to run the server and then click `Start Audio`.';
startBtn.disabled = false;
stopBtn.disabled = true;
});
function initWebSocket() {
ws = new WebSocket('ws://localhost:8765');
// This is so `event.data` is already an ArrayBuffer.
ws.binaryType = 'arraybuffer';
ws.addEventListener('open', handleWebSocketOpen);
ws.addEventListener('message', handleWebSocketMessage);
ws.addEventListener('close', (event) => {
console.log('WebSocket connection closed.', event.code, event.reason);
stopAudio(false);
});
ws.addEventListener('error', (event) => console.error('WebSocket error:', event));
}
function handleWebSocketOpen(event) {
console.log('WebSocket connection established.', event)
navigator.mediaDevices.getUserMedia({
audio: {
sampleRate: SAMPLE_RATE,
channelCount: NUM_CHANNELS,
autoGainControl: true,
echoCancellation: true,
noiseSuppression: true,
}
}).then((stream) => {
microphoneStream = stream;
// 512 is closest thing to 200ms.
scriptProcessor = audioContext.createScriptProcessor(512, 1, 1);
source = audioContext.createMediaStreamSource(stream);
source.connect(scriptProcessor);
scriptProcessor.connect(audioContext.destination);
scriptProcessor.onaudioprocess = (event) => {
if (!ws) {
return;
}
const audioData = event.inputBuffer.getChannelData(0);
const pcmS16Array = convertFloat32ToS16PCM(audioData);
const pcmByteArray = new Uint8Array(pcmS16Array.buffer);
const frame = Frame.create({
audio: {
audio: Array.from(pcmByteArray),
sampleRate: SAMPLE_RATE,
numChannels: NUM_CHANNELS
}
});
const encodedFrame = new Uint8Array(Frame.encode(frame).finish());
ws.send(encodedFrame);
};
}).catch((error) => console.error('Error accessing microphone:', error));
}
function handleWebSocketMessage(event) {
const arrayBuffer = event.data;
if (isPlaying) {
enqueueAudioFromProto(arrayBuffer);
}
}
function enqueueAudioFromProto(arrayBuffer) {
const parsedFrame = Frame.decode(new Uint8Array(arrayBuffer));
if (!parsedFrame?.audio) {
return false;
}
// Reset play time if it's been a while we haven't played anything.
const diffTime = audioContext.currentTime - lastMessageTime;
if ((playTime == 0) || (diffTime > PLAY_TIME_RESET_THRESHOLD_MS)) {
playTime = audioContext.currentTime;
}
lastMessageTime = audioContext.currentTime;
// We should be able to use parsedFrame.audio.audio.buffer but for
// some reason that contains all the bytes from the protobuf message.
const audioVector = Array.from(parsedFrame.audio.audio);
const audioArray = new Uint8Array(audioVector);
audioContext.decodeAudioData(audioArray.buffer, function(buffer) {
const source = new AudioBufferSourceNode(audioContext);
source.buffer = buffer;
source.start(playTime);
source.connect(audioContext.destination);
playTime = playTime + buffer.duration;
});
}
function convertFloat32ToS16PCM(float32Array) {
let int16Array = new Int16Array(float32Array.length);
for (let i = 0; i < float32Array.length; i++) {
let clampedValue = Math.max(-1, Math.min(1, float32Array[i]));
int16Array[i] = clampedValue < 0 ? clampedValue * 32768 : clampedValue * 32767;
}
return int16Array;
}
function startAudioBtnHandler() {
if (!navigator.mediaDevices || !navigator.mediaDevices.getUserMedia) {
alert('getUserMedia is not supported in your browser.');
return;
}
startBtn.disabled = true;
stopBtn.disabled = false;
audioContext = new (window.AudioContext || window.webkitAudioContext)({
latencyHint: 'interactive',
sampleRate: SAMPLE_RATE
});
isPlaying = true;
initWebSocket();
}
function stopAudio(closeWebsocket) {
playTime = 0;
isPlaying = false;
startBtn.disabled = false;
stopBtn.disabled = true;
if (ws && closeWebsocket) {
ws.close();
ws = null;
}
if (scriptProcessor) {
scriptProcessor.disconnect();
}
if (source) {
source.disconnect();
}
}
function stopAudioBtnHandler() {
stopAudio(true);
}
startBtn.addEventListener('click', startAudioBtnHandler);
stopBtn.addEventListener('click', stopAudioBtnHandler);
startBtn.disabled = true;
stopBtn.disabled = true;
</script>
</body>
</html>

2
examples/websocket-server/requirements.txt
View File

@@ -1,2 +0,0 @@
python-dotenv
pipecat-ai[cartesia,openai,silero,websocket,deepgram]

54
examples/websocket/README.md Normal file
View File

@@ -0,0 +1,54 @@
# Voice Agent
A Pipecat example demonstrating the simplest way to create a voice agent using `WebsocketTransport`.
## 🚀 Quick Start
### 1️⃣ Start the Bot Server
#### 🔧 Set Up the Environment
1. Create and activate a virtual environment:
```bash
python3 -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
```
2. Install dependencies:
```bash
pip install -r requirements.txt
```
3. Configure environment variables:
- Copy `env.example` to `.env`
```bash
cp env.example .env
```
- Add your API keys
- Choose what do you wish to use, 'fast_api' or 'websocket_server'
#### ▶️ Run the Server
```bash
python server/server.py
```
### 3️⃣ Connect Using a Custom Client App
For client-side setup, refer to the:
- [Typescript Guide](client/README.md).
## ⚠️ Important Note
Ensure the bot server is running before using any client implementations.
## 📌 Requirements
- Python **3.10+**
- Node.js **16+** (for JavaScript components)
- Google API Key
---
### 💡 Notes
- Ensure all dependencies are installed before running the server.
- Check the `.env` file for missing configurations.
Happy coding! 🎉

27
examples/websocket/client/README.md Normal file
View File

@@ -0,0 +1,27 @@
# JavaScript Implementation
Basic implementation using the [Pipecat JavaScript SDK](https://docs.pipecat.ai/client/js/introduction).
## Setup
1. Run the bot server. See the [server README](../README).
2. Navigate to the `client/javascript` directory:
```bash
cd client/javascript
```
3. Install dependencies:
```bash
npm install
```
4. Run the client app:
```
npm run dev
```
5. Visit http://localhost:5173 in your browser.

34
examples/websocket/client/index.html Normal file
View File

@@ -0,0 +1,34 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>AI Chatbot</title>
</head>
<body>
<div class="container">
<div class="status-bar">
<div class="status">
Transport: <span id="connection-status">Disconnected</span>
</div>
<div class="controls">
<button id="connect-btn">Connect</button>
<button id="disconnect-btn" disabled>Disconnect</button>
</div>
</div>
<audio id="bot-audio" autoplay></audio>
<div class="debug-panel">
<h3>Debug Info</h3>
<div id="debug-log"></div>
</div>
</div>
<script type="module" src="/src/app.ts"></script>
<link rel="stylesheet" href="/src/style.css">
</body>
</html>

1770
examples/websocket/client/package-lock.json generated Normal file
View File

File diff suppressed because it is too large Load Diff

26
examples/websocket/client/package.json Normal file
View File

@@ -0,0 +1,26 @@
{
"name": "client",
"version": "1.0.0",
"main": "index.js",
"scripts": {
"dev": "vite",
"build": "tsc && vite build",
"preview": "vite preview"
},
"keywords": [],
"author": "",
"license": "ISC",
"description": "",
"devDependencies": {
"@types/node": "^22.15.30",
"@types/protobufjs": "^6.0.0",
"@vitejs/plugin-react-swc": "^3.10.1",
"typescript": "^5.8.3",
"vite": "^6.3.5"
},
"dependencies": {
"@pipecat-ai/client-js": "^0.4.0",
"@pipecat-ai/websocket-transport": "^0.4.1",
"protobufjs": "^7.4.0"
}
}

234
examples/websocket/client/src/app.ts Normal file
View File

@@ -0,0 +1,234 @@
/**
* Copyright (c) 2024–2025, Daily
*
* SPDX-License-Identifier: BSD 2-Clause License
*/
/**
* RTVI Client Implementation
*
* This client connects to an RTVI-compatible bot server using WebSocket.
*
* Requirements:
* - A running RTVI bot server (defaults to http://localhost:7860)
*/
import {
RTVIClient,
RTVIClientOptions,
RTVIEvent,
} from '@pipecat-ai/client-js';
import {
WebSocketTransport
} from "@pipecat-ai/websocket-transport";
class WebsocketClientApp {
private rtviClient: RTVIClient | null = null;
private connectBtn: HTMLButtonElement | null = null;
private disconnectBtn: HTMLButtonElement | null = null;
private statusSpan: HTMLElement | null = null;
private debugLog: HTMLElement | null = null;
private botAudio: HTMLAudioElement;
constructor() {
console.log("WebsocketClientApp");
this.botAudio = document.createElement('audio');
this.botAudio.autoplay = true;
//this.botAudio.playsInline = true;
document.body.appendChild(this.botAudio);
this.setupDOMElements();
this.setupEventListeners();
}
/**
* Set up references to DOM elements and create necessary media elements
*/
private setupDOMElements(): void {
this.connectBtn = document.getElementById('connect-btn') as HTMLButtonElement;
this.disconnectBtn = document.getElementById('disconnect-btn') as HTMLButtonElement;
this.statusSpan = document.getElementById('connection-status');
this.debugLog = document.getElementById('debug-log');
}
/**
* Set up event listeners for connect/disconnect buttons
*/
private setupEventListeners(): void {
this.connectBtn?.addEventListener('click', () => this.connect());
this.disconnectBtn?.addEventListener('click', () => this.disconnect());
}
/**
* Add a timestamped message to the debug log
*/
private log(message: string): void {
if (!this.debugLog) return;
const entry = document.createElement('div');
entry.textContent = `${new Date().toISOString()} - ${message}`;
if (message.startsWith('User: ')) {
entry.style.color = '#2196F3';
} else if (message.startsWith('Bot: ')) {
entry.style.color = '#4CAF50';
}
this.debugLog.appendChild(entry);
this.debugLog.scrollTop = this.debugLog.scrollHeight;
console.log(message);
}
/**
* Update the connection status display
*/
private updateStatus(status: string): void {
if (this.statusSpan) {
this.statusSpan.textContent = status;
}
this.log(`Status: ${status}`);
}
/**
* Check for available media tracks and set them up if present
* This is called when the bot is ready or when the transport state changes to ready
*/
setupMediaTracks() {
if (!this.rtviClient) return;
const tracks = this.rtviClient.tracks();
if (tracks.bot?.audio) {
this.setupAudioTrack(tracks.bot.audio);
}
}
/**
* Set up listeners for track events (start/stop)
* This handles new tracks being added during the session
*/
setupTrackListeners() {
if (!this.rtviClient) return;
// Listen for new tracks starting
this.rtviClient.on(RTVIEvent.TrackStarted, (track, participant) => {
// Only handle non-local (bot) tracks
if (!participant?.local && track.kind === 'audio') {
this.setupAudioTrack(track);
}
});
// Listen for tracks stopping
this.rtviClient.on(RTVIEvent.TrackStopped, (track, participant) => {
this.log(`Track stopped: ${track.kind} from ${participant?.name || 'unknown'}`);
});
}
/**
* Set up an audio track for playback
* Handles both initial setup and track updates
*/
private setupAudioTrack(track: MediaStreamTrack): void {
this.log('Setting up audio track');
if (this.botAudio.srcObject && "getAudioTracks" in this.botAudio.srcObject) {
const oldTrack = this.botAudio.srcObject.getAudioTracks()[0];
if (oldTrack?.id === track.id) return;
}
this.botAudio.srcObject = new MediaStream([track]);
}
/**
* Initialize and connect to the bot
* This sets up the RTVI client, initializes devices, and establishes the connection
*/
public async connect(): Promise<void> {
try {
const startTime = Date.now();
//const transport = new DailyTransport();
const transport = new WebSocketTransport();
const RTVIConfig: RTVIClientOptions = {
transport,
params: {
// The baseURL and endpoint of your bot server that the client will connect to
baseUrl: 'http://localhost:7860',
endpoints: { connect: '/connect' },
},
enableMic: true,
enableCam: false,
callbacks: {
onConnected: () => {
this.updateStatus('Connected');
if (this.connectBtn) this.connectBtn.disabled = true;
if (this.disconnectBtn) this.disconnectBtn.disabled = false;
},
onDisconnected: () => {
this.updateStatus('Disconnected');
if (this.connectBtn) this.connectBtn.disabled = false;
if (this.disconnectBtn) this.disconnectBtn.disabled = true;
this.log('Client disconnected');
},
onBotReady: (data) => {
this.log(`Bot ready: ${JSON.stringify(data)}`);
this.setupMediaTracks();
},
onUserTranscript: (data) => {
if (data.final) {
this.log(`User: ${data.text}`);
}
},
onBotTranscript: (data) => this.log(`Bot: ${data.text}`),
onMessageError: (error) => console.error('Message error:', error),
onError: (error) => console.error('Error:', error),
},
}
this.rtviClient = new RTVIClient(RTVIConfig);
this.setupTrackListeners();
this.log('Initializing devices...');
await this.rtviClient.initDevices();
this.log('Connecting to bot...');
await this.rtviClient.connect();
const timeTaken = Date.now() - startTime;
this.log(`Connection complete, timeTaken: ${timeTaken}`);
} catch (error) {
this.log(`Error connecting: ${(error as Error).message}`);
this.updateStatus('Error');
// Clean up if there's an error
if (this.rtviClient) {
try {
await this.rtviClient.disconnect();
} catch (disconnectError) {
this.log(`Error during disconnect: ${disconnectError}`);
}
}
}
}
/**
* Disconnect from the bot and clean up media resources
*/
public async disconnect(): Promise<void> {
if (this.rtviClient) {
try {
await this.rtviClient.disconnect();
this.rtviClient = null;
if (this.botAudio.srcObject && "getAudioTracks" in this.botAudio.srcObject) {
this.botAudio.srcObject.getAudioTracks().forEach((track) => track.stop());
this.botAudio.srcObject = null;
}
} catch (error) {
this.log(`Error disconnecting: ${(error as Error).message}`);
}
}
}
}
declare global {
interface Window {
WebsocketClientApp: typeof WebsocketClientApp;
}
}
window.addEventListener('DOMContentLoaded', () => {
window.WebsocketClientApp = WebsocketClientApp;
new WebsocketClientApp();
});

98
examples/websocket/client/src/style.css Normal file
View File

@@ -0,0 +1,98 @@
body {
margin: 0;
padding: 20px;
font-family: Arial, sans-serif;
background-color: #f0f0f0;
}
.container {
max-width: 1200px;
margin: 0 auto;
}
.status-bar {
display: flex;
justify-content: space-between;
align-items: center;
padding: 10px;
background-color: #fff;
border-radius: 8px;
margin-bottom: 20px;
}
.controls button {
padding: 8px 16px;
margin-left: 10px;
border: none;
border-radius: 4px;
cursor: pointer;
}
#connect-btn {
background-color: #4caf50;
color: white;
}
#disconnect-btn {
background-color: #f44336;
color: white;
}
button:disabled {
opacity: 0.5;
cursor: not-allowed;
}
.main-content {
background-color: #fff;
border-radius: 8px;
padding: 20px;
margin-bottom: 20px;
}
.bot-container {
display: flex;
flex-direction: column;
align-items: center;
}
#bot-video-container {
width: 640px;
height: 360px;
background-color: #e0e0e0;
border-radius: 8px;
margin: 20px auto;
overflow: hidden;
display: flex;
align-items: center;
justify-content: center;
}
#bot-video-container video {
width: 100%;
height: 100%;
object-fit: cover;
}
.debug-panel {
background-color: #fff;
border-radius: 8px;
padding: 20px;
}
.debug-panel h3 {
margin: 0 0 10px 0;
font-size: 16px;
font-weight: bold;
}
#debug-log {
height: 500px;
overflow-y: auto;
background-color: #f8f8f8;
padding: 10px;
border-radius: 4px;
font-family: monospace;
font-size: 12px;
line-height: 1.4;
}

111
examples/websocket/client/tsconfig.json Normal file
View File

@@ -0,0 +1,111 @@
{
"compilerOptions": {
/* Visit https://aka.ms/tsconfig to read more about this file */
/* Projects */
// "incremental": true, /* Save .tsbuildinfo files to allow for incremental compilation of projects. */
// "composite": true, /* Enable constraints that allow a TypeScript project to be used with project references. */
// "tsBuildInfoFile": "./.tsbuildinfo", /* Specify the path to .tsbuildinfo incremental compilation file. */
// "disableSourceOfProjectReferenceRedirect": true, /* Disable preferring source files instead of declaration files when referencing composite projects. */
// "disableSolutionSearching": true, /* Opt a project out of multi-project reference checking when editing. */
// "disableReferencedProjectLoad": true, /* Reduce the number of projects loaded automatically by TypeScript. */
/* Language and Environment */
"target": "es2016", /* Set the JavaScript language version for emitted JavaScript and include compatible library declarations. */
// "lib": [], /* Specify a set of bundled library declaration files that describe the target runtime environment. */
// "jsx": "preserve", /* Specify what JSX code is generated. */
// "experimentalDecorators": true, /* Enable experimental support for legacy experimental decorators. */
// "emitDecoratorMetadata": true, /* Emit design-type metadata for decorated declarations in source files. */
// "jsxFactory": "", /* Specify the JSX factory function used when targeting React JSX emit, e.g. 'React.createElement' or 'h'. */
// "jsxFragmentFactory": "", /* Specify the JSX Fragment reference used for fragments when targeting React JSX emit e.g. 'React.Fragment' or 'Fragment'. */
// "jsxImportSource": "", /* Specify module specifier used to import the JSX factory functions when using 'jsx: react-jsx*'. */
// "reactNamespace": "", /* Specify the object invoked for 'createElement'. This only applies when targeting 'react' JSX emit. */
// "noLib": true, /* Disable including any library files, including the default lib.d.ts. */
// "useDefineForClassFields": true, /* Emit ECMAScript-standard-compliant class fields. */
// "moduleDetection": "auto", /* Control what method is used to detect module-format JS files. */
/* Modules */
"module": "commonjs", /* Specify what module code is generated. */
// "rootDir": "./", /* Specify the root folder within your source files. */
// "moduleResolution": "node10", /* Specify how TypeScript looks up a file from a given module specifier. */
// "baseUrl": "./", /* Specify the base directory to resolve non-relative module names. */
// "paths": {}, /* Specify a set of entries that re-map imports to additional lookup locations. */
// "rootDirs": [], /* Allow multiple folders to be treated as one when resolving modules. */
// "typeRoots": [], /* Specify multiple folders that act like './node_modules/@types'. */
// "types": [], /* Specify type package names to be included without being referenced in a source file. */
// "allowUmdGlobalAccess": true, /* Allow accessing UMD globals from modules. */
// "moduleSuffixes": [], /* List of file name suffixes to search when resolving a module. */
// "allowImportingTsExtensions": true, /* Allow imports to include TypeScript file extensions. Requires '--moduleResolution bundler' and either '--noEmit' or '--emitDeclarationOnly' to be set. */
// "rewriteRelativeImportExtensions": true, /* Rewrite '.ts', '.tsx', '.mts', and '.cts' file extensions in relative import paths to their JavaScript equivalent in output files. */
// "resolvePackageJsonExports": true, /* Use the package.json 'exports' field when resolving package imports. */
// "resolvePackageJsonImports": true, /* Use the package.json 'imports' field when resolving imports. */
// "customConditions": [], /* Conditions to set in addition to the resolver-specific defaults when resolving imports. */
// "noUncheckedSideEffectImports": true, /* Check side effect imports. */
// "resolveJsonModule": true, /* Enable importing .json files. */
// "allowArbitraryExtensions": true, /* Enable importing files with any extension, provided a declaration file is present. */
// "noResolve": true, /* Disallow 'import's, 'require's or '<reference>'s from expanding the number of files TypeScript should add to a project. */
/* JavaScript Support */
// "allowJs": true, /* Allow JavaScript files to be a part of your program. Use the 'checkJS' option to get errors from these files. */
// "checkJs": true, /* Enable error reporting in type-checked JavaScript files. */
// "maxNodeModuleJsDepth": 1, /* Specify the maximum folder depth used for checking JavaScript files from 'node_modules'. Only applicable with 'allowJs'. */
/* Emit */
// "declaration": true, /* Generate .d.ts files from TypeScript and JavaScript files in your project. */
// "declarationMap": true, /* Create sourcemaps for d.ts files. */
// "emitDeclarationOnly": true, /* Only output d.ts files and not JavaScript files. */
// "sourceMap": true, /* Create source map files for emitted JavaScript files. */
// "inlineSourceMap": true, /* Include sourcemap files inside the emitted JavaScript. */
// "noEmit": true, /* Disable emitting files from a compilation. */
// "outFile": "./", /* Specify a file that bundles all outputs into one JavaScript file. If 'declaration' is true, also designates a file that bundles all .d.ts output. */
// "outDir": "./", /* Specify an output folder for all emitted files. */
// "removeComments": true, /* Disable emitting comments. */
// "importHelpers": true, /* Allow importing helper functions from tslib once per project, instead of including them per-file. */
// "downlevelIteration": true, /* Emit more compliant, but verbose and less performant JavaScript for iteration. */
// "sourceRoot": "", /* Specify the root path for debuggers to find the reference source code. */
// "mapRoot": "", /* Specify the location where debugger should locate map files instead of generated locations. */
// "inlineSources": true, /* Include source code in the sourcemaps inside the emitted JavaScript. */
// "emitBOM": true, /* Emit a UTF-8 Byte Order Mark (BOM) in the beginning of output files. */
// "newLine": "crlf", /* Set the newline character for emitting files. */
// "stripInternal": true, /* Disable emitting declarations that have '@internal' in their JSDoc comments. */
// "noEmitHelpers": true, /* Disable generating custom helper functions like '__extends' in compiled output. */
// "noEmitOnError": true, /* Disable emitting files if any type checking errors are reported. */
// "preserveConstEnums": true, /* Disable erasing 'const enum' declarations in generated code. */
// "declarationDir": "./", /* Specify the output directory for generated declaration files. */
/* Interop Constraints */
// "isolatedModules": true, /* Ensure that each file can be safely transpiled without relying on other imports. */
// "verbatimModuleSyntax": true, /* Do not transform or elide any imports or exports not marked as type-only, ensuring they are written in the output file's format based on the 'module' setting. */
// "isolatedDeclarations": true, /* Require sufficient annotation on exports so other tools can trivially generate declaration files. */
// "allowSyntheticDefaultImports": true, /* Allow 'import x from y' when a module doesn't have a default export. */
"esModuleInterop": true, /* Emit additional JavaScript to ease support for importing CommonJS modules. This enables 'allowSyntheticDefaultImports' for type compatibility. */
// "preserveSymlinks": true, /* Disable resolving symlinks to their realpath. This correlates to the same flag in node. */
"forceConsistentCasingInFileNames": true, /* Ensure that casing is correct in imports. */
/* Type Checking */
"strict": true, /* Enable all strict type-checking options. */
// "noImplicitAny": true, /* Enable error reporting for expressions and declarations with an implied 'any' type. */
// "strictNullChecks": true, /* When type checking, take into account 'null' and 'undefined'. */
// "strictFunctionTypes": true, /* When assigning functions, check to ensure parameters and the return values are subtype-compatible. */
// "strictBindCallApply": true, /* Check that the arguments for 'bind', 'call', and 'apply' methods match the original function. */
// "strictPropertyInitialization": true, /* Check for class properties that are declared but not set in the constructor. */
// "strictBuiltinIteratorReturn": true, /* Built-in iterators are instantiated with a 'TReturn' type of 'undefined' instead of 'any'. */
// "noImplicitThis": true, /* Enable error reporting when 'this' is given the type 'any'. */
// "useUnknownInCatchVariables": true, /* Default catch clause variables as 'unknown' instead of 'any'. */
// "alwaysStrict": true, /* Ensure 'use strict' is always emitted. */
// "noUnusedLocals": true, /* Enable error reporting when local variables aren't read. */
// "noUnusedParameters": true, /* Raise an error when a function parameter isn't read. */
// "exactOptionalPropertyTypes": true, /* Interpret optional property types as written, rather than adding 'undefined'. */
// "noImplicitReturns": true, /* Enable error reporting for codepaths that do not explicitly return in a function. */
// "noFallthroughCasesInSwitch": true, /* Enable error reporting for fallthrough cases in switch statements. */
// "noUncheckedIndexedAccess": true, /* Add 'undefined' to a type when accessed using an index. */
// "noImplicitOverride": true, /* Ensure overriding members in derived classes are marked with an override modifier. */
// "noPropertyAccessFromIndexSignature": true, /* Enforces using indexed accessors for keys declared using an indexed type. */
// "allowUnusedLabels": true, /* Disable error reporting for unused labels. */
// "allowUnreachableCode": true, /* Disable error reporting for unreachable code. */
/* Completeness */
// "skipDefaultLibCheck": true, /* Skip type checking .d.ts files that are included with TypeScript. */
"skipLibCheck": true /* Skip type checking all .d.ts files. */
}
}

15
examples/websocket/client/vite.config.js Normal file
View File

@@ -0,0 +1,15 @@
import { defineConfig } from 'vite';
import react from '@vitejs/plugin-react-swc';
export default defineConfig({
plugins: [react()],
server: {
proxy: {
// Proxy /api requests to the backend server
'/connect': {
target: 'http://0.0.0.0:7860', // Replace with your backend URL
changeOrigin: true,
},
},
},
});

111
examples/websocket/server/bot_fast_api.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 dotenv import load_dotenv
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.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
from pipecat.serializers.protobuf import ProtobufFrameSerializer
from pipecat.services.gemini_multimodal_live import GeminiMultimodalLiveLLMService
from pipecat.transports.network.fastapi_websocket import (
FastAPIWebsocketParams,
FastAPIWebsocketTransport,
)
load_dotenv(override=True)
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
SYSTEM_INSTRUCTION = f"""
"You are Gemini 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. Keep your responses brief. One or two sentences at most.
"""
async def run_bot(websocket_client):
ws_transport = FastAPIWebsocketTransport(
websocket=websocket_client,
params=FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
add_wav_header=False,
vad_analyzer=SileroVADAnalyzer(),
serializer=ProtobufFrameSerializer(),
),
)
llm = GeminiMultimodalLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
voice_id="Puck", # Aoede, Charon, Fenrir, Kore, Puck
transcribe_model_audio=True,
system_instruction=SYSTEM_INSTRUCTION,
)
context = OpenAILLMContext(
[
{
"role": "user",
"content": "Start by greeting the user warmly and introducing yourself.",
}
],
)
context_aggregator = llm.create_context_aggregator(context)
# RTVI events for Pipecat client UI
rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
pipeline = Pipeline(
[
ws_transport.input(),
context_aggregator.user(),
rtvi,
llm, # LLM
ws_transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
),
observers=[RTVIObserver(rtvi)],
)
@rtvi.event_handler("on_client_ready")
async def on_client_ready(rtvi):
logger.info("Pipecat client ready.")
await rtvi.set_bot_ready()
# Kick off the conversation.
await task.queue_frames([context_aggregator.user().get_context_frame()])
@ws_transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info("Pipecat Client connected")
@ws_transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info("Pipecat Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)

109
examples/websocket/server/bot_websocket_server.py Normal file
View File

@@ -0,0 +1,109 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
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.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
from pipecat.serializers.protobuf import ProtobufFrameSerializer
from pipecat.services.gemini_multimodal_live import GeminiMultimodalLiveLLMService
from pipecat.transports.network.websocket_server import (
WebsocketServerParams,
WebsocketServerTransport,
)
SYSTEM_INSTRUCTION = f"""
"You are Gemini 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. Keep your responses brief. One or two sentences at most.
"""
async def run_bot_websocket_server():
ws_transport = WebsocketServerTransport(
params=WebsocketServerParams(
serializer=ProtobufFrameSerializer(),
audio_in_enabled=True,
audio_out_enabled=True,
add_wav_header=False,
vad_analyzer=SileroVADAnalyzer(),
session_timeout=60 * 3, # 3 minutes
)
)
llm = GeminiMultimodalLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
voice_id="Puck", # Aoede, Charon, Fenrir, Kore, Puck
transcribe_model_audio=True,
system_instruction=SYSTEM_INSTRUCTION,
)
context = OpenAILLMContext(
[
{
"role": "user",
"content": "Start by greeting the user warmly and introducing yourself.",
}
],
)
context_aggregator = llm.create_context_aggregator(context)
# RTVI events for Pipecat client UI
rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
pipeline = Pipeline(
[
ws_transport.input(),
context_aggregator.user(),
rtvi,
llm, # LLM
ws_transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
),
observers=[RTVIObserver(rtvi)],
)
@rtvi.event_handler("on_client_ready")
async def on_client_ready(rtvi):
logger.info("Pipecat client ready.")
await rtvi.set_bot_ready()
# Kick off the conversation.
await task.queue_frames([context_aggregator.user().get_context_frame()])
@ws_transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info("Pipecat Client connected")
@ws_transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info("Pipecat Client disconnected")
await task.cancel()
@ws_transport.event_handler("on_session_timeout")
async def on_session_timeout(transport, client):
logger.info(f"Entering in timeout for {client.remote_address}")
await task.cancel()
runner = PipelineRunner()
await runner.run(task)

2
examples/websocket/server/env.example Normal file
View File

@@ -0,0 +1,2 @@
GOOGLE_API_KEY=
WEBSOCKET_SERVER= # Options: 'fast_api' or 'websocket_server'

4
examples/websocket/server/requirements.txt Normal file
View File

@@ -0,0 +1,4 @@
python-dotenv
fastapi[all]
uvicorn
pipecat-ai[silero,websocket,google]

79
examples/websocket/server/server.py Normal file
View File

@@ -0,0 +1,79 @@
#
# Copyright (c) 2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from contextlib import asynccontextmanager
from typing import Any, Dict
import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, Request, WebSocket
from fastapi.middleware.cors import CORSMiddleware
# Load environment variables
load_dotenv(override=True)
from bot_fast_api import run_bot
from bot_websocket_server import run_bot_websocket_server
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Handles FastAPI startup and shutdown."""
yield # Run app
# Initialize FastAPI app with lifespan manager
app = FastAPI(lifespan=lifespan)
# Configure CORS to allow requests from any origin
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
print("WebSocket connection accepted")
try:
await run_bot(websocket)
except Exception as e:
print(f"Exception in run_bot: {e}")
@app.post("/connect")
async def bot_connect(request: Request) -> Dict[Any, Any]:
server_mode = os.getenv("WEBSOCKET_SERVER", "fast_api")
if server_mode == "websocket_server":
ws_url = "ws://localhost:8765"
else:
ws_url = "ws://localhost:7860/ws"
return {"ws_url": ws_url}
async def main():
server_mode = os.getenv("WEBSOCKET_SERVER", "fast_api")
tasks = []
try:
if server_mode == "websocket_server":
tasks.append(run_bot_websocket_server())
config = uvicorn.Config(app, host="0.0.0.0", port=7860)
server = uvicorn.Server(config)
tasks.append(server.serve())
await asyncio.gather(*tasks)
except asyncio.CancelledError:
print("Tasks cancelled (probably due to shutdown).")
if __name__ == "__main__":
asyncio.run(main())

15
src/pipecat/serializers/protobuf.py
View File

@@ -41,6 +41,7 @@ class ProtobufFrameSerializer(FrameSerializer):
TextFrame: "text",
InputAudioRawFrame: "audio",
TranscriptionFrame: "transcription",
MessageFrame: "message",
}
DESERIALIZABLE_FIELDS = {v: k for k, v in DESERIALIZABLE_TYPES.items()}
@@ -97,8 +98,18 @@ class ProtobufFrameSerializer(FrameSerializer):
if "pts" in args_dict:
del args_dict["pts"]
# Create the instance
instance = class_name(**args_dict)
# Special handling for MessageFrame -> TransportMessageUrgentFrame
if class_name == MessageFrame:
try:
msg = json.loads(args_dict["data"])
instance = TransportMessageUrgentFrame(message=msg)
logger.debug(f"ProtobufFrameSerializer: Transport message {instance}")
except Exception as e:
logger.error(f"Error parsing MessageFrame data: {e}")
return None
else:
# Normal deserialization, create the instance
instance = class_name(**args_dict)
# Set special fields
if id: