Merge pull request #2004 from pipecat-ai/filipi/pipeline_freeze
Pipeline freeze improvements
This commit is contained in:
Filipi da Silva Fuchter
12
CHANGELOG.md
12
CHANGELOG.md
@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
|
||||
### Added
|
||||
|
||||
- Added logging and improved error handling to help diagnose and prevent potential
|
||||
Pipeline freezes.
|
||||
|
||||
- Introduce task watchdog timers. Watchdog timers are used to detect if a
|
||||
Pipecat task is taking longer than expected (by default 5 seconds). It is
|
||||
possible to change the default watchdog timer timeout by using the
|
||||
@@ -72,6 +75,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
|
||||
### Fixed
|
||||
|
||||
- Fixed an issue in `FastAPIWebsocketClient` to ensure proper disconnection
|
||||
when the websocket is already closed.
|
||||
|
||||
- Fixed an issue where the `UserStoppedSpeakingFrame` was not received if the
|
||||
transport was not receiving new audio frames.
|
||||
|
||||
- Fixed an edge case where if the user interrupted the bot but no new aggregation
|
||||
was received, the bot would not resume speaking.
|
||||
|
||||
- Fixed an issue with `ElevenLabsTTSService` where the context was not being
|
||||
closed.
|
||||
|
||||
|
||||
59
examples/freeze-test/README.md
Normal file
59
examples/freeze-test/README.md
Normal file
@@ -0,0 +1,59 @@
|
||||
# Freeze Test Client
|
||||
|
||||
The purpose of this example is to create an environment for testing the bot and try to create freezing conditions.
|
||||
|
||||
### Approach 1: Server-Side Testing with `SimulateFreezeInput`
|
||||
|
||||
- Utilize only the bot `freeze_test_bot.py` with the `SimulateFreezeInput` processor. This input continuously injects frames, simulating user speech interruptions at random intervals.
|
||||
- This approach excludes the use of input transport and speech-to-text (STT) functionalities.
|
||||
|
||||
### Approach 2: Server-Side with TypeScript Client
|
||||
|
||||
- Combine server-side operations with a TypeScript client.
|
||||
- The client initially records a segment of audio, e.g., 5–10 seconds long. It can be anything.
|
||||
- After that, it replays this recorded audio to the server at random intervals, mimicking user input interruptions.
|
||||
- This helps testing interruptions in the pipeline as if real users were interacting with the bot.
|
||||
|
||||
## Setup
|
||||
|
||||
Follow these steps to set up and run the Freeze Test Client:
|
||||
|
||||
1. **Run the Bot Server**
|
||||
- Set up and activate your virtual environment:
|
||||
```bash
|
||||
python3 -m venv venv
|
||||
source venv/bin/activate # On Windows: venv\Scripts\activate
|
||||
```
|
||||
|
||||
- Install dependencies:
|
||||
```bash
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
- Create your `.env` file and set your env vars:
|
||||
```bash
|
||||
cp env.example .env
|
||||
```
|
||||
|
||||
- Run the server:
|
||||
```bash
|
||||
python freeze_test_bot.py
|
||||
```
|
||||
|
||||
2. **Navigate to the Client Directory**
|
||||
```bash
|
||||
cd client
|
||||
```
|
||||
|
||||
3. **Install Dependencies**
|
||||
```bash
|
||||
npm install
|
||||
```
|
||||
|
||||
4. **Run the Client Application**
|
||||
```bash
|
||||
npm run dev
|
||||
```
|
||||
|
||||
5. **Access the Client in Your Browser**
|
||||
Visit [http://localhost:5173](http://localhost:5173) to interact with the Freeze Test Client.
|
||||
43
examples/freeze-test/client/index.html
Normal file
43
examples/freeze-test/client/index.html
Normal file
@@ -0,0 +1,43 @@
|
||||
<!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>
|
||||
<div class="status-bar">
|
||||
<div class="status">
|
||||
Playing audio: <span id="play-audio-status"></span>
|
||||
</div>
|
||||
<div class="controls">
|
||||
<button id="play-btn">Start</button>
|
||||
<button id="stop-btn" disabled>Stop</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/freeze-test/client/package-lock.json
generated
Normal file
1770
examples/freeze-test/client/package-lock.json
generated
Normal file
File diff suppressed because it is too large
Load Diff
26
examples/freeze-test/client/package.json
Normal file
26
examples/freeze-test/client/package.json
Normal 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"
|
||||
}
|
||||
}
|
||||
328
examples/freeze-test/client/src/app.ts
Normal file
328
examples/freeze-test/client/src/app.ts
Normal file
@@ -0,0 +1,328 @@
|
||||
/**
|
||||
* 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 {
|
||||
ProtobufFrameSerializer,
|
||||
WebSocketTransport
|
||||
} from "@pipecat-ai/websocket-transport";
|
||||
|
||||
class RecordingSerializer extends ProtobufFrameSerializer {
|
||||
|
||||
private lastTimestamp: number | null = null;
|
||||
private recordingAudioToSend: boolean = false;
|
||||
private _recordedAudio: { data: ArrayBuffer; delay: number }[] = [];
|
||||
|
||||
public startRecording() {
|
||||
this.recordingAudioToSend = true;
|
||||
this._recordedAudio = [];
|
||||
this.lastTimestamp = null;
|
||||
}
|
||||
|
||||
public stopRecording() {
|
||||
this.recordingAudioToSend = false;
|
||||
}
|
||||
|
||||
// @ts-ignore
|
||||
serializeAudio(data: ArrayBuffer, sampleRate: number, numChannels: number): Uint8Array | null {
|
||||
if (this.recordingAudioToSend) {
|
||||
const now = Date.now();
|
||||
// Compute delay since last packet
|
||||
const delay = this.lastTimestamp ? now - this.lastTimestamp : 0;
|
||||
this.lastTimestamp = now;
|
||||
// Save audio chunk and delay
|
||||
this._recordedAudio.push({ data, delay });
|
||||
return null;
|
||||
} else {
|
||||
return super.serializeAudio(data, sampleRate, numChannels);
|
||||
}
|
||||
}
|
||||
|
||||
public get recordedAudio() {
|
||||
return this._recordedAudio
|
||||
}
|
||||
}
|
||||
|
||||
class WebsocketClientApp {
|
||||
private ENABLE_RECORDING_MODE = false
|
||||
private RECORDING_TIME_MS = 10000
|
||||
|
||||
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;
|
||||
|
||||
private declare websocketTransport: WebSocketTransport;
|
||||
private sendRecordedAudio: boolean = false
|
||||
private declare recordingSerializer: RecordingSerializer;
|
||||
|
||||
private playBtn: HTMLButtonElement | null = null;
|
||||
private stopBtn: HTMLButtonElement | null = null;
|
||||
|
||||
constructor() {
|
||||
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');
|
||||
this.playBtn = document.getElementById('play-btn') as HTMLButtonElement;
|
||||
this.stopBtn = document.getElementById('stop-btn') as HTMLButtonElement;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up event listeners for connect/disconnect buttons
|
||||
*/
|
||||
private setupEventListeners(): void {
|
||||
this.connectBtn?.addEventListener('click', () => this.connect());
|
||||
this.disconnectBtn?.addEventListener('click', () => this.disconnect());
|
||||
this.playBtn?.addEventListener('click', () => this.startSendingRecordedAudio());
|
||||
this.stopBtn?.addEventListener('click', () => this.stopSendingRecordedAudio());
|
||||
}
|
||||
|
||||
/**
|
||||
* 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();
|
||||
|
||||
this.recordingSerializer = new RecordingSerializer()
|
||||
const transport = this.ENABLE_RECORDING_MODE ? new WebSocketTransport({serializer: this.recordingSerializer}) : new WebSocketTransport();
|
||||
this.websocketTransport = transport
|
||||
|
||||
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}`);
|
||||
|
||||
if (this.ENABLE_RECORDING_MODE) {
|
||||
this.log(`Starting to recording the next ${(this.RECORDING_TIME_MS/1000)}s of audio`);
|
||||
this.recordingSerializer.startRecording()
|
||||
await this.sleep(this.RECORDING_TIME_MS)
|
||||
this.recordingSerializer.stopRecording()
|
||||
this.log("Recording stopped");
|
||||
this.rtviClient.enableMic(false)
|
||||
this.startSendingRecordedAudio()
|
||||
}
|
||||
} 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 {
|
||||
this.stopSendingRecordedAudio()
|
||||
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}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private startSendingRecordedAudio() {
|
||||
this.sendRecordedAudio = true
|
||||
if (this.playBtn) this.playBtn.disabled = true;
|
||||
if (this.stopBtn) this.stopBtn.disabled = false;
|
||||
void this.replayAudio()
|
||||
}
|
||||
|
||||
private stopSendingRecordedAudio() {
|
||||
if (this.stopBtn) this.stopBtn.disabled = true;
|
||||
if (this.playBtn) this.playBtn.disabled = false;
|
||||
this.sendRecordedAudio = false
|
||||
}
|
||||
|
||||
private async replayAudio() {
|
||||
if (this.sendRecordedAudio) {
|
||||
this.log("Sending recorded audio")
|
||||
for (const chunk of this.recordingSerializer.recordedAudio) {
|
||||
await this.sleep(chunk.delay);
|
||||
this.websocketTransport.handleUserAudioStream(chunk.data);
|
||||
}
|
||||
const randomDelay = 1000 + Math.random() * (10000 - 500);
|
||||
await this.sleep(randomDelay);
|
||||
|
||||
void this.replayAudio()
|
||||
}
|
||||
}
|
||||
|
||||
private sleep(ms: number): Promise<void> {
|
||||
return new Promise(resolve => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
declare global {
|
||||
interface Window {
|
||||
WebsocketClientApp: typeof WebsocketClientApp;
|
||||
}
|
||||
}
|
||||
|
||||
window.addEventListener('DOMContentLoaded', () => {
|
||||
window.WebsocketClientApp = WebsocketClientApp;
|
||||
new WebsocketClientApp();
|
||||
});
|
||||
98
examples/freeze-test/client/src/style.css
Normal file
98
examples/freeze-test/client/src/style.css
Normal 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/freeze-test/client/tsconfig.json
Normal file
111
examples/freeze-test/client/tsconfig.json
Normal 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/freeze-test/client/vite.config.js
Normal file
15
examples/freeze-test/client/vite.config.js
Normal 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,
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
312
examples/freeze-test/freeze_test_bot.py
Normal file
312
examples/freeze-test/freeze_test_bot.py
Normal file
@@ -0,0 +1,312 @@
|
||||
#
|
||||
# Copyright (c) 2024–2025, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import os
|
||||
import random
|
||||
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
|
||||
from fastapi.responses import RedirectResponse
|
||||
from loguru import logger
|
||||
from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI
|
||||
|
||||
from pipecat.audio.vad.silero import SileroVADAnalyzer
|
||||
from pipecat.frames.frames import (
|
||||
CancelFrame,
|
||||
EndFrame,
|
||||
Frame,
|
||||
InterimTranscriptionFrame,
|
||||
LLMFullResponseEndFrame,
|
||||
LLMTextFrame,
|
||||
StartFrame,
|
||||
StartInterruptionFrame,
|
||||
StopFrame,
|
||||
StopInterruptionFrame,
|
||||
TranscriptionFrame,
|
||||
TTSTextFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
)
|
||||
from pipecat.observers.loggers.debug_log_observer import DebugLogObserver
|
||||
from pipecat.pipeline.parallel_pipeline import ParallelPipeline
|
||||
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,
|
||||
OpenAILLMContextFrame,
|
||||
)
|
||||
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||||
from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIProcessor
|
||||
from pipecat.serializers.protobuf import ProtobufFrameSerializer
|
||||
from pipecat.services.cartesia.tts import CartesiaTTSService
|
||||
from pipecat.services.deepgram import DeepgramSTTService
|
||||
from pipecat.services.openai.llm import OpenAILLMService
|
||||
from pipecat.transports.network.fastapi_websocket import (
|
||||
FastAPIWebsocketParams,
|
||||
FastAPIWebsocketTransport,
|
||||
)
|
||||
from pipecat.utils.time import time_now_iso8601
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
|
||||
@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=["*"],
|
||||
)
|
||||
|
||||
# Mount the frontend at /
|
||||
app.mount("/client", SmallWebRTCPrebuiltUI)
|
||||
|
||||
|
||||
class SimulateFreezeInput(FrameProcessor):
|
||||
def __init__(
|
||||
self,
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(**kwargs)
|
||||
# Whether we have seen a StartFrame already.
|
||||
self._initialized = False
|
||||
self._send_frames_task = None
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
if isinstance(frame, StartFrame):
|
||||
# Push StartFrame before start(), because we want StartFrame to be
|
||||
# processed by every processor before any other frame is processed.
|
||||
await self.push_frame(frame, direction)
|
||||
await self._start(frame)
|
||||
elif isinstance(frame, CancelFrame):
|
||||
logger.info("SimulateFreezeInput: Received cancel frame")
|
||||
await self._stop()
|
||||
await self.push_frame(frame, direction)
|
||||
elif isinstance(frame, EndFrame):
|
||||
logger.info("SimulateFreezeInput: Received end frame")
|
||||
await self.push_frame(frame, direction)
|
||||
await self._stop()
|
||||
elif isinstance(frame, StopFrame):
|
||||
logger.info("SimulateFreezeInput: Received stop frame")
|
||||
await self.push_frame(frame, direction)
|
||||
await self._stop()
|
||||
|
||||
async def _start(self, frame: StartFrame):
|
||||
if self._initialized:
|
||||
return
|
||||
logger.info(f"Starting SimulateFreezeInput")
|
||||
self._initialized = True
|
||||
if not self._send_frames_task:
|
||||
self._send_frames_task = self.create_task(self._send_frames())
|
||||
|
||||
async def _stop(self):
|
||||
logger.info(f"Stopping SimulateFreezeInput")
|
||||
self._initialized = False
|
||||
if self._send_frames_task:
|
||||
await self.cancel_task(self._send_frames_task)
|
||||
self._send_frames_task = None
|
||||
|
||||
async def _send_user_text(self, text: str):
|
||||
# Emulation as if the user has spoken and the stt transcribed
|
||||
await self.push_frame(UserStartedSpeakingFrame())
|
||||
await self.push_frame(StartInterruptionFrame())
|
||||
await self.push_frame(
|
||||
TranscriptionFrame(
|
||||
text,
|
||||
"",
|
||||
time_now_iso8601(),
|
||||
)
|
||||
)
|
||||
# Need to wait before sending the UserStoppedSpeakingFrame,
|
||||
# otherwise TranscriptionFrame will be processed
|
||||
# later than the UserStoppedSpeakingFrame
|
||||
await asyncio.sleep(0.1)
|
||||
await self.push_frame(UserStoppedSpeakingFrame())
|
||||
await self.push_frame(StopInterruptionFrame())
|
||||
|
||||
async def _send_frames(self):
|
||||
try:
|
||||
i = 0
|
||||
while True:
|
||||
logger.debug("SimulateFreezeInput _send_frames")
|
||||
await self._send_user_text("Tell me a brief history of Brazil!")
|
||||
await asyncio.sleep(3)
|
||||
await self._send_user_text("")
|
||||
break
|
||||
# i += 1
|
||||
# if i >= 5:
|
||||
# break
|
||||
# sleeping 1s before interrupting
|
||||
# wait_time = random.uniform(1, 10)
|
||||
# await asyncio.sleep(wait_time)
|
||||
except Exception as e:
|
||||
logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})")
|
||||
|
||||
|
||||
async def run_example(websocket_client):
|
||||
logger.info(f"Starting bot")
|
||||
|
||||
# Create a transport using the WebRTC connection
|
||||
transport = FastAPIWebsocketTransport(
|
||||
websocket=websocket_client,
|
||||
params=FastAPIWebsocketParams(
|
||||
audio_in_enabled=True,
|
||||
audio_out_enabled=True,
|
||||
add_wav_header=False,
|
||||
vad_analyzer=SileroVADAnalyzer(),
|
||||
serializer=ProtobufFrameSerializer(),
|
||||
),
|
||||
)
|
||||
|
||||
freeze = SimulateFreezeInput()
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
|
||||
|
||||
rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
|
||||
|
||||
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(
|
||||
[
|
||||
ParallelPipeline(
|
||||
[
|
||||
freeze,
|
||||
],
|
||||
[
|
||||
transport.input(),
|
||||
stt,
|
||||
],
|
||||
),
|
||||
rtvi,
|
||||
context_aggregator.user(), # User responses
|
||||
llm, # LLM
|
||||
tts, # TTS
|
||||
transport.output(), # Transport bot output
|
||||
context_aggregator.assistant(), # Assistant spoken responses
|
||||
]
|
||||
)
|
||||
|
||||
task = PipelineTask(
|
||||
pipeline,
|
||||
params=PipelineParams(
|
||||
allow_interruptions=True,
|
||||
enable_metrics=True,
|
||||
enable_usage_metrics=True,
|
||||
report_only_initial_ttfb=True,
|
||||
),
|
||||
idle_timeout_secs=120,
|
||||
observers=[
|
||||
DebugLogObserver(
|
||||
frame_types={
|
||||
InterimTranscriptionFrame: None,
|
||||
TranscriptionFrame: None,
|
||||
# TTSTextFrame: None,
|
||||
# LLMTextFrame: None,
|
||||
OpenAILLMContextFrame: None,
|
||||
LLMFullResponseEndFrame: None,
|
||||
},
|
||||
exclude_fields={
|
||||
"result",
|
||||
"metadata",
|
||||
"audio",
|
||||
"image",
|
||||
"images",
|
||||
},
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
@transport.event_handler("on_client_connected")
|
||||
async def on_client_connected(transport, client):
|
||||
logger.info(f"Client connected")
|
||||
|
||||
@rtvi.event_handler("on_client_ready")
|
||||
async def on_client_ready(rtvi):
|
||||
logger.info(f"Client ready")
|
||||
await rtvi.set_bot_ready()
|
||||
# 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):
|
||||
logger.info(f"Client disconnected")
|
||||
await task.cancel()
|
||||
|
||||
runner = PipelineRunner(handle_sigint=False)
|
||||
|
||||
await runner.run(task)
|
||||
|
||||
|
||||
@app.get("/", include_in_schema=False)
|
||||
async def root_redirect():
|
||||
return RedirectResponse(url="/client/")
|
||||
|
||||
|
||||
@app.websocket("/ws")
|
||||
async def websocket_endpoint(websocket: WebSocket):
|
||||
await websocket.accept()
|
||||
print("WebSocket connection accepted")
|
||||
try:
|
||||
await run_example(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}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description="Pipecat Bot Runner")
|
||||
parser.add_argument(
|
||||
"--host", default="localhost", help="Host for HTTP server (default: localhost)"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--port", type=int, default=7860, help="Port for HTTP server (default: 7860)"
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
uvicorn.run(app, host=args.host, port=args.port)
|
||||
@@ -78,3 +78,8 @@ class BaseTurnAnalyzer(ABC):
|
||||
EndOfTurnState: The result of the end of turn analysis.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def clear(self):
|
||||
"""Reset the turn analyzer to its initial state."""
|
||||
pass
|
||||
|
||||
@@ -98,6 +98,9 @@ class BaseSmartTurn(BaseTurnAnalyzer):
|
||||
logger.debug(f"End of Turn result: {state}")
|
||||
return state, result
|
||||
|
||||
def clear(self):
|
||||
self._clear(EndOfTurnState.COMPLETE)
|
||||
|
||||
def _clear(self, turn_state: EndOfTurnState):
|
||||
# If the state is still incomplete, keep the _speech_triggered as True
|
||||
self._speech_triggered = turn_state == EndOfTurnState.INCOMPLETE
|
||||
|
||||
@@ -6,7 +6,8 @@
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Tuple, Type
|
||||
from collections import deque
|
||||
from typing import Any, AsyncIterable, Deque, Dict, Iterable, List, Optional, Tuple, Type
|
||||
|
||||
from loguru import logger
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
@@ -23,6 +24,7 @@ from pipecat.frames.frames import (
|
||||
ErrorFrame,
|
||||
Frame,
|
||||
HeartbeatFrame,
|
||||
InputAudioRawFrame,
|
||||
LLMFullResponseEndFrame,
|
||||
MetricsFrame,
|
||||
StartFrame,
|
||||
@@ -663,12 +665,17 @@ class PipelineTask(BasePipelineTask):
|
||||
"""
|
||||
running = True
|
||||
last_frame_time = 0
|
||||
frame_buffer = deque(maxlen=10) # Store last 10 frames
|
||||
|
||||
while running:
|
||||
try:
|
||||
frame = await asyncio.wait_for(
|
||||
self._idle_queue.get(), timeout=self._idle_timeout_secs
|
||||
)
|
||||
|
||||
if not isinstance(frame, InputAudioRawFrame):
|
||||
frame_buffer.append(frame)
|
||||
|
||||
if isinstance(frame, StartFrame) or isinstance(frame, self._idle_timeout_frames):
|
||||
# If we find a StartFrame or one of the frames that prevents a
|
||||
# time out we update the time.
|
||||
@@ -679,7 +686,7 @@ class PipelineTask(BasePipelineTask):
|
||||
# valid frames.
|
||||
diff_time = time.time() - last_frame_time
|
||||
if diff_time >= self._idle_timeout_secs:
|
||||
running = await self._idle_timeout_detected()
|
||||
running = await self._idle_timeout_detected(frame_buffer)
|
||||
# Reset `last_frame_time` so we don't trigger another
|
||||
# immediate idle timeout if we are not cancelling. For
|
||||
# example, we might want to force the bot to say goodbye
|
||||
@@ -687,15 +694,20 @@ class PipelineTask(BasePipelineTask):
|
||||
last_frame_time = time.time()
|
||||
|
||||
self._idle_queue.task_done()
|
||||
except asyncio.TimeoutError:
|
||||
running = await self._idle_timeout_detected()
|
||||
|
||||
async def _idle_timeout_detected(self) -> bool:
|
||||
except asyncio.TimeoutError:
|
||||
running = await self._idle_timeout_detected(frame_buffer)
|
||||
|
||||
async def _idle_timeout_detected(self, last_frames: Deque[Frame]) -> bool:
|
||||
"""Logic for when the pipeline is idle.
|
||||
|
||||
Returns:
|
||||
bool: Whther the pipeline task is being cancelled or not.
|
||||
"""
|
||||
logger.warning("Idle timeout detected. Last 10 frames received:")
|
||||
for i, frame in enumerate(last_frames, 1):
|
||||
logger.warning(f"Frame {i}: {frame}")
|
||||
|
||||
await self._call_event_handler("on_idle_timeout")
|
||||
if self._cancel_on_idle_timeout:
|
||||
logger.warning(f"Idle pipeline detected, cancelling pipeline task...")
|
||||
|
||||
@@ -266,6 +266,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
|
||||
|
||||
self._user_speaking = False
|
||||
self._bot_speaking = False
|
||||
self._was_bot_speaking = False
|
||||
self._emulating_vad = False
|
||||
self._seen_interim_results = False
|
||||
self._waiting_for_aggregation = False
|
||||
@@ -275,6 +276,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
|
||||
|
||||
async def reset(self):
|
||||
await super().reset()
|
||||
self._was_bot_speaking = False
|
||||
self._seen_interim_results = False
|
||||
self._waiting_for_aggregation = False
|
||||
[await s.reset() for s in self._interruption_strategies]
|
||||
@@ -355,6 +357,20 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
|
||||
else:
|
||||
# No interruption config - normal behavior (always push aggregation)
|
||||
await self._process_aggregation()
|
||||
# Handles the case where both the user and the bot are not speaking,
|
||||
# and the bot was previously speaking before the user interruption.
|
||||
# Normally, when the user stops speaking, new text is expected,
|
||||
# which triggers the bot to respond. However, if no new text
|
||||
# is received, this safeguard ensures
|
||||
# the bot doesn't hang indefinitely while waiting to speak again.
|
||||
elif not self._seen_interim_results and self._was_bot_speaking and not self._bot_speaking:
|
||||
logger.warning(
|
||||
"User stopped speaking but no new aggregation received. Forcing aggregation processing to resume bot response."
|
||||
)
|
||||
# Resetting it so we don't trigger this twice
|
||||
self._was_bot_speaking = False
|
||||
# We are just pushing the same previous context to be processed again in this case
|
||||
await self.push_frame(OpenAILLMContextFrame(self._context))
|
||||
|
||||
async def _should_interrupt_based_on_strategies(self) -> bool:
|
||||
"""Check if interruption should occur based on configured strategies."""
|
||||
@@ -381,6 +397,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
|
||||
async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame):
|
||||
self._user_speaking = True
|
||||
self._waiting_for_aggregation = True
|
||||
self._was_bot_speaking = self._bot_speaking
|
||||
|
||||
# If we get a non-emulated UserStartedSpeakingFrame but we are in the
|
||||
# middle of emulating VAD, let's stop emulating VAD (i.e. don't send the
|
||||
@@ -393,7 +410,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
|
||||
# We just stopped speaking. Let's see if there's some aggregation to
|
||||
# push. If the last thing we saw is an interim transcription, let's wait
|
||||
# pushing the aggregation as we will probably get a final transcription.
|
||||
if not self._seen_interim_results:
|
||||
if len(self._aggregation) > 0 and not self._seen_interim_results:
|
||||
await self.push_aggregation()
|
||||
|
||||
async def _handle_bot_started_speaking(self, _: BotStartedSpeakingFrame):
|
||||
|
||||
@@ -337,9 +337,8 @@ class FrameProcessor(BaseObject):
|
||||
# Cancel the input task. This will stop processing queued frames.
|
||||
await self.__cancel_input_task()
|
||||
except Exception as e:
|
||||
logger.exception(f"Uncaught exception in {self}: {e}")
|
||||
logger.exception(f"Uncaught exception in {self} when handling _start_interruption: {e}")
|
||||
await self.push_error(ErrorFrame(str(e)))
|
||||
raise
|
||||
|
||||
# Create a new input queue and task.
|
||||
self.__create_input_task()
|
||||
@@ -382,7 +381,6 @@ class FrameProcessor(BaseObject):
|
||||
except Exception as e:
|
||||
logger.exception(f"Uncaught exception in {self}: {e}")
|
||||
await self.push_error(ErrorFrame(str(e)))
|
||||
raise
|
||||
|
||||
def _check_started(self, frame: Frame):
|
||||
if not self.__started:
|
||||
@@ -411,19 +409,19 @@ class FrameProcessor(BaseObject):
|
||||
logger.trace(f"{self}: frame processing resumed")
|
||||
|
||||
(frame, direction, callback) = await self.__input_queue.get()
|
||||
|
||||
self.start_watchdog()
|
||||
|
||||
# Process the frame.
|
||||
await self.process_frame(frame, direction)
|
||||
|
||||
# If this frame has an associated callback, call it now.
|
||||
if callback:
|
||||
await callback(self, frame, direction)
|
||||
|
||||
self.__input_queue.task_done()
|
||||
|
||||
self.reset_watchdog()
|
||||
try:
|
||||
self.start_watchdog()
|
||||
# Process the frame.
|
||||
await self.process_frame(frame, direction)
|
||||
# If this frame has an associated callback, call it now.
|
||||
if callback:
|
||||
await callback(self, frame, direction)
|
||||
except Exception as e:
|
||||
logger.exception(f"{self}: error processing frame: {e}")
|
||||
await self.push_error(ErrorFrame(str(e)))
|
||||
finally:
|
||||
self.__input_queue.task_done()
|
||||
self.reset_watchdog()
|
||||
|
||||
def __create_push_task(self):
|
||||
if not self.__push_frame_task:
|
||||
|
||||
@@ -43,6 +43,8 @@ from pipecat.metrics.metrics import MetricsData
|
||||
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||||
from pipecat.transports.base_transport import TransportParams
|
||||
|
||||
AUDIO_INPUT_TIMEOUT_SECS = 0.5
|
||||
|
||||
|
||||
class BaseInputTransport(FrameProcessor):
|
||||
def __init__(self, params: TransportParams, **kwargs):
|
||||
@@ -56,6 +58,9 @@ class BaseInputTransport(FrameProcessor):
|
||||
# Track bot speaking state for interruption logic
|
||||
self._bot_speaking = False
|
||||
|
||||
# Track user speaking state for interruption logic
|
||||
self._user_speaking = False
|
||||
|
||||
# We read audio from a single queue one at a time and we then run VAD in
|
||||
# a thread. Therefore, only one thread should be necessary.
|
||||
self._executor = ThreadPoolExecutor(max_workers=1)
|
||||
@@ -130,6 +135,7 @@ class BaseInputTransport(FrameProcessor):
|
||||
|
||||
async def start(self, frame: StartFrame):
|
||||
self._paused = False
|
||||
self._user_speaking = False
|
||||
|
||||
self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
|
||||
|
||||
@@ -240,6 +246,7 @@ class BaseInputTransport(FrameProcessor):
|
||||
async def _handle_user_interruption(self, frame: Frame):
|
||||
if isinstance(frame, UserStartedSpeakingFrame):
|
||||
logger.debug("User started speaking")
|
||||
self._user_speaking = True
|
||||
await self.push_frame(frame)
|
||||
|
||||
# Only push StartInterruptionFrame if:
|
||||
@@ -263,6 +270,7 @@ class BaseInputTransport(FrameProcessor):
|
||||
)
|
||||
elif isinstance(frame, UserStoppedSpeakingFrame):
|
||||
logger.debug("User stopped speaking")
|
||||
self._user_speaking = False
|
||||
await self.push_frame(frame)
|
||||
if self.interruptions_allowed:
|
||||
await self._stop_interruption()
|
||||
@@ -355,30 +363,42 @@ class BaseInputTransport(FrameProcessor):
|
||||
async def _audio_task_handler(self):
|
||||
vad_state: VADState = VADState.QUIET
|
||||
while True:
|
||||
frame: InputAudioRawFrame = await self._audio_in_queue.get()
|
||||
try:
|
||||
frame: InputAudioRawFrame = await asyncio.wait_for(
|
||||
self._audio_in_queue.get(), timeout=AUDIO_INPUT_TIMEOUT_SECS
|
||||
)
|
||||
|
||||
self.start_watchdog()
|
||||
self.start_watchdog()
|
||||
|
||||
# If an audio filter is available, run it before VAD.
|
||||
if self._params.audio_in_filter:
|
||||
frame.audio = await self._params.audio_in_filter.filter(frame.audio)
|
||||
# If an audio filter is available, run it before VAD.
|
||||
if self._params.audio_in_filter:
|
||||
frame.audio = await self._params.audio_in_filter.filter(frame.audio)
|
||||
|
||||
# Check VAD and push event if necessary. We just care about
|
||||
# changes from QUIET to SPEAKING and vice versa.
|
||||
previous_vad_state = vad_state
|
||||
if self._params.vad_analyzer:
|
||||
vad_state = await self._handle_vad(frame, vad_state)
|
||||
# Check VAD and push event if necessary. We just care about
|
||||
# changes from QUIET to SPEAKING and vice versa.
|
||||
previous_vad_state = vad_state
|
||||
if self._params.vad_analyzer:
|
||||
vad_state = await self._handle_vad(frame, vad_state)
|
||||
|
||||
if self._params.turn_analyzer:
|
||||
await self._run_turn_analyzer(frame, vad_state, previous_vad_state)
|
||||
if self._params.turn_analyzer:
|
||||
await self._run_turn_analyzer(frame, vad_state, previous_vad_state)
|
||||
|
||||
# Push audio downstream if passthrough is set.
|
||||
if self._params.audio_in_passthrough:
|
||||
await self.push_frame(frame)
|
||||
# Push audio downstream if passthrough is set.
|
||||
if self._params.audio_in_passthrough:
|
||||
await self.push_frame(frame)
|
||||
|
||||
self._audio_in_queue.task_done()
|
||||
|
||||
self.reset_watchdog()
|
||||
self._audio_in_queue.task_done()
|
||||
except asyncio.TimeoutError:
|
||||
if self._user_speaking:
|
||||
logger.warning(
|
||||
"Forcing user stopped speaking due to timeout receiving audio frame!"
|
||||
)
|
||||
vad_state = VADState.QUIET
|
||||
if self._params.turn_analyzer:
|
||||
self._params.turn_analyzer.clear()
|
||||
await self._handle_user_interruption(UserStoppedSpeakingFrame())
|
||||
finally:
|
||||
self.reset_watchdog()
|
||||
|
||||
async def _handle_prediction_result(self, result: MetricsData):
|
||||
"""Handle a prediction result event from the turn analyzer.
|
||||
|
||||
@@ -70,11 +70,22 @@ class FastAPIWebsocketClient:
|
||||
return self._websocket.iter_bytes() if self._is_binary else self._websocket.iter_text()
|
||||
|
||||
async def send(self, data: str | bytes):
|
||||
if self._can_send():
|
||||
if self._is_binary:
|
||||
await self._websocket.send_bytes(data)
|
||||
else:
|
||||
await self._websocket.send_text(data)
|
||||
try:
|
||||
if self._can_send():
|
||||
if self._is_binary:
|
||||
await self._websocket.send_bytes(data)
|
||||
else:
|
||||
await self._websocket.send_text(data)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"{self} exception sending data: {e.__class__.__name__} ({e}), application_state: {self._websocket.application_state}"
|
||||
)
|
||||
# For some reason the websocket is disconnected, and we are not able to send data
|
||||
# So let's properly handle it and disconnect the transport
|
||||
if self._websocket.application_state == WebSocketState.DISCONNECTED:
|
||||
logger.warning("Closing already disconnected websocket!")
|
||||
self._closing = True
|
||||
await self.trigger_client_disconnected()
|
||||
|
||||
async def disconnect(self):
|
||||
self._leave_counter -= 1
|
||||
|
||||
@@ -261,6 +261,9 @@ class TaskManager(BaseTaskManager):
|
||||
pass
|
||||
except Exception as e:
|
||||
logger.exception(f"{name}: unexpected exception while cancelling task: {e}")
|
||||
except BaseException as e:
|
||||
logger.critical(f"{name}: fatal base exception while cancelling task: {e}")
|
||||
raise
|
||||
finally:
|
||||
self._remove_task(task)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user