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

examples: websocket-server updates

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-05-29 16:21:23 -07:00
parent e31e87aabd
commit 5f45a9d90f
2 changed files with 207 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

267
examples/foundational/websocket-server/index.html
View File

@@ -1,134 +1,197 @@
<!DOCTYPE html> <!DOCTYPE html>
<html lang="en"> <html lang="en">
<head> <head>
<meta charset="UTF-8"> <meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0"> <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> <script src="https://cdn.jsdelivr.net/npm/protobufjs@7.X.X/dist/protobuf.min.js"></script>
<title>WebSocket Audio Stream</title> <title>Pipecat WebSocket Client Example</title>
</head> </head>
<body> <body>
<h1>WebSocket Audio Stream</h1> <h1>Pipecat WebSocket Client Example</h1>
<h3><div id="progressText">Loading, wait...</div></h2>
<button id="startAudioBtn">Start Audio</button> <button id="startAudioBtn">Start Audio</button>
<button id="stopAudioBtn">Stop Audio</button> <button id="stopAudioBtn">Stop Audio</button>
<script> <script>
const SAMPLE_RATE = 16000; const SAMPLE_RATE = 16000;
const BUFFER_SIZE = 8192; const NUM_CHANNELS = 1;
const MIN_AUDIO_SIZE = 6400;
let audioContext; // The protobuf type. We will load it later.
let microphoneStream; let Frame = null;
let scriptProcessor;
let source;
let frame;
let audioChunks = [];
let isPlaying = false;
let ws;
const proto = protobuf.load("frames.proto", (err, root) => { // The websocket connection.
if (err) throw err; let ws = null;
frame = root.lookupType("pipecat.Frame");
});
function initWebSocket() { // The audio context
ws = new WebSocket('ws://localhost:8765'); let audioContext = null;
ws.addEventListener('open', () => console.log('WebSocket connection established.')); // The audio context media stream source
ws.addEventListener('message', handleWebSocketMessage); let source = null;
ws.addEventListener('close', (event) => console.log("WebSocket connection closed.", event.code, event.reason));
ws.addEventListener('error', (event) => console.error('WebSocket error:', event));
}
async function handleWebSocketMessage(event) { // The microphone stream from getUserMedia. SHould be sampled to the
const arrayBuffer = await event.data.arrayBuffer(); // proper sample rate.
enqueueAudioFromProto(arrayBuffer); let microphoneStream = null;
}
function enqueueAudioFromProto(arrayBuffer) { // Script processor to get data from microphone.
const parsedFrame = frame.decode(new Uint8Array(arrayBuffer)); let scriptProcessor = null;
if (!parsedFrame?.audio) return false;
const frameCount = parsedFrame.audio.data.length / 2; // AudioContext play time.
const audioOutBuffer = audioContext.createBuffer(1, frameCount, SAMPLE_RATE); let playTime = 0;
const nowBuffering = audioOutBuffer.getChannelData(0);
const view = new Int16Array(parsedFrame.audio.data.buffer);
for (let i = 0; i < frameCount; i++) { // Whether we should be playing audio.
const word = view[i]; let isPlaying = false;
nowBuffering[i] = ((word + 32768) % 65536 - 32768) / 32768.0;
}
audioChunks.push(audioOutBuffer); let startBtn = document.getElementById('startAudioBtn');
if (!isPlaying) playNextChunk(); let stopBtn = document.getElementById('stopAudioBtn');
}
function playNextChunk() { const proto = protobuf.load("frames.proto", (err, root) => {
if (audioChunks.length === 0) { if (err) {
isPlaying = false; throw err;
return; }
} 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`.";
isPlaying = true; startBtn.disabled = false;
const audioOutBuffer = audioChunks.shift(); stopBtn.disabled = true;
const source = audioContext.createBufferSource(); });
source.buffer = audioOutBuffer;
source.connect(audioContext.destination);
source.onended = playNextChunk;
source.start();
}
function startAudio() { function initWebSocket() {
if (!navigator.mediaDevices || !navigator.mediaDevices.getUserMedia) { ws = new WebSocket('ws://localhost:8765');
alert('getUserMedia is not supported in your browser.');
return;
}
navigator.mediaDevices.getUserMedia({ audio: true }) ws.addEventListener('open', () => console.log('WebSocket connection established.'));
.then((stream) => { ws.addEventListener('message', handleWebSocketMessage);
microphoneStream = stream; ws.addEventListener('close', (event) => {
audioContext = new (window.AudioContext || window.webkitAudioContext)(); console.log("WebSocket connection closed.", event.code, event.reason);
scriptProcessor = audioContext.createScriptProcessor(BUFFER_SIZE, 1, 1); stopAudio(false);
source = audioContext.createMediaStreamSource(stream); });
source.connect(scriptProcessor); ws.addEventListener('error', (event) => console.error('WebSocket error:', event));
scriptProcessor.connect(audioContext.destination); }
const audioBuffer = []; async function handleWebSocketMessage(event) {
const skipRatio = Math.floor(audioContext.sampleRate / (SAMPLE_RATE * 2)); const arrayBuffer = await event.data.arrayBuffer();
if (isPlaying) {
enqueueAudioFromProto(arrayBuffer);
}
}
scriptProcessor.onaudioprocess = (event) => { function enqueueAudioFromProto(arrayBuffer) {
const rawLeftChannelData = event.inputBuffer.getChannelData(0); const parsedFrame = Frame.decode(new Uint8Array(arrayBuffer));
for (let i = 0; i < rawLeftChannelData.length; i += skipRatio) { if (!parsedFrame?.audio) {
const normalized = ((rawLeftChannelData[i] * 32768.0) + 32768) % 65536 - 32768; return false;
const swappedBytes = ((normalized & 0xff) << 8) | ((normalized >> 8) & 0xff); }
audioBuffer.push(swappedBytes);
}
if (audioBuffer.length >= MIN_AUDIO_SIZE) { if (playTime == 0) {
const audioFrame = frame.create({ audio: { audio: audioBuffer.slice(0, MIN_AUDIO_SIZE) } }); playTime = audioContext.currentTime;
const encodedFrame = new Uint8Array(frame.encode(audioFrame).finish()); }
ws.send(encodedFrame);
audioBuffer.splice(0, MIN_AUDIO_SIZE);
}
};
initWebSocket(); // We should be able to use parsedFrame.audio.audio.buffer but for
}) // some reason that contains all the bytes from the protobuf message.
.catch((error) => console.error('Error accessing microphone:', error)); const audioVector = Array.from(parsedFrame.audio.audio);
} const audioArray = new Uint8Array(audioVector);
function stopAudio() { audioContext.decodeAudioData(audioArray.buffer, function(buffer) {
if (ws) { const source = new AudioBufferSourceNode(audioContext);
ws.close(); source.buffer = buffer;
scriptProcessor.disconnect(); source.start(playTime);
source.disconnect(); source.connect(audioContext.destination);
ws = undefined; playTime = playTime + buffer.duration;
} });
} }
document.getElementById('startAudioBtn').addEventListener('click', startAudio); function convertFloat32ToS16PCM(float32Array) {
document.getElementById('stopAudioBtn').addEventListener('click', stopAudio); 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();
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 stopAudio(closeWebsocket) {
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> </script>
</body> </body>
</html> </html>

63
examples/foundational/websocket-server/server.py
View File

@@ -9,32 +9,37 @@ import asyncio
import os import os
import sys import sys
from loguru import logger from pipecat.frames.frames import LLMMessagesFrame
from pipecat.frames.frames import Frame, TextFrame, TranscriptionFrame
from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.llm_response import LLMAssistantResponseAggregator, LLMUserResponseAggregator
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.services.whisper import WhisperSTTService from pipecat.services.whisper import WhisperSTTService
from pipecat.transports.network.websocket_server import WebsocketServerTransport from pipecat.transports.network.websocket_server import WebsocketServerParams, WebsocketServerTransport
from pipecat.vad.silero import SileroVAD
from loguru import logger
from dotenv import load_dotenv
load_dotenv(override=True)
logger.remove(0) logger.remove(0)
logger.add(sys.stderr, level="DEBUG") logger.add(sys.stderr, level="TRACE")
class WhisperTranscriber(FrameProcessor):
async def process_frame(self, frame: Frame, direction: FrameDirection):
if isinstance(frame, TranscriptionFrame):
print(f"Transcribed: {frame.text}")
else:
await self.push_frame(frame, direction)
async def main(): async def main():
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
transport = WebsocketServerTransport() transport = WebsocketServerTransport(params=WebsocketServerParams(add_wav_header=True))
vad = SileroVAD(audio_passthrough=True)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4-turbo-preview")
stt = WhisperSTTService()
tts = ElevenLabsTTSService( tts = ElevenLabsTTSService(
aiohttp_session=session, aiohttp_session=session,
@@ -42,19 +47,35 @@ async def main():
voice_id=os.getenv("ELEVENLABS_VOICE_ID"), 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([ pipeline = Pipeline([
transport.input(), transport.input(), # Websocket input from client
WhisperSTTService(), vad, # VAD to detect user speech
WhisperTranscriber(), stt, # Speech-To-Text
tts, tma_in, # User responses
transport.output(), llm, # LLM
tts, # Text-To-Speech
transport.output(), # Websocket output to client
tma_out # LLM responses
]) ])
task = PipelineTask(pipeline) task = PipelineTask(pipeline)
@transport.event_handler("on_client_connected") @transport.event_handler("on_client_connected")
async def on_client_connected(transport, client): async def on_client_connected(transport, client):
await task.queue_frame(TextFrame("Hello there!")) # Kick off the conversation.
messages.append(
{"role": "system", "content": "Please introduce yourself to the user."})
await task.queue_frames([LLMMessagesFrame(messages)])
runner = PipelineRunner() runner = PipelineRunner()