Compare commits

..

1 Commits

Author SHA1 Message Date
Aleix Conchillo Flaqué
b4326b253f CartesiaTTSService: reset context id when flushing audio 2025-02-11 23:36:08 -08:00
113 changed files with 864 additions and 18905 deletions

15
.gitignore vendored
View File

@@ -32,21 +32,6 @@ fly.toml
# Example files
pipecat/examples/twilio-chatbot/templates/streams.xml
pipecat/examples/bot-ready-signalling/client/react-native/node_modules/
pipecat/examples/bot-ready-signalling/client/react-native/.expo/
pipecat/examples/bot-ready-signalling/client/react-native/dist/
pipecat/examples/bot-ready-signalling/client/react-native/npm-debug.*
pipecat/examples/bot-ready-signalling/client/react-native/*.jks
pipecat/examples/bot-ready-signalling/client/react-native/*.p8
pipecat/examples/bot-ready-signalling/client/react-native/*.p12
pipecat/examples/bot-ready-signalling/client/react-native/*.key
pipecat/examples/bot-ready-signalling/client/react-native/*.mobileprovision
pipecat/examples/bot-ready-signalling/client/react-native/*.orig.*
pipecat/examples/bot-ready-signalling/client/react-native/web-build/
# macOS
.DS_Store
# Documentation
docs/api/_build/

View File

@@ -5,113 +5,10 @@ All notable changes to **Pipecat** will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## Unreleased
## [Unreleased]
### Added
- Added retry logic to `get_chat_completions()` fo all OpenAI-based LLM
services.
- Added new log observers `LLMLogObserver` and `TranscriptionLogObserver` that
can be useful for debugging your pipelines.
- Added `room_url` property to `DailyTransport`.
- Added `addons` argument to `DeepgramSTTService`.
- Added `exponential_backoff_time()` to `utils.network` module.
### Changed
- `RimeHttpTTSService` needs an `aiohttp.ClientSession` to be passed to the
constructor as all the other HTTP-based services.
- `RimeHttpTTSService` doesn't use a default voice anymore.
- `DeepgramSTTService` now uses the new `nova-3` model by default. If you want
to use the previous model you can pass `LiveOptions(model="nova-2-general")`.
(see https://deepgram.com/learn/introducing-nova-3-speech-to-text-api)
```python
stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general"))
```
### Removed
- Remove `TransportParams.audio_out_is_live` since it was not being used at all.
### Fixed
- Fixed `match_endofsentence` support for ellipses.
- Fixed an issue that would cause undesired interruptions via
`EmulateUserStartedSpeakingFrame` when only interim transcriptions (i.e. no
final transcriptions) where received.
- Fixed an issue where `EndTaskFrame` was not triggering
`on_client_disconnected` or closing the WebSocket in FastAPI.
- Fixed an issue in `DeepgramSTTService` where the `sample_rate` passed to the
`LiveOptions` was not being used, causing the service to use the default
sample rate of pipeline.
- Fixed a context aggregator issue that would not append the LLM text response
to the context if a function call happened in the same LLM turn.
- Fixed an issue that was causing HTTP TTS services to push `TTSStoppedFrame`
more than once.
- Fixed a `FishAudioTTSService` issue where `TTSStoppedFrame` was not being
pushed.
- Fixed an issue that `start_callback` was not invoked for some LLM services.
- Fixed an issue that would cause `DeepgramSTTService` to stop working after an
error occurred (e.g. sudden network loss). If the network recovered we would
not reconnect.
- Fixed a `STTMuteFilter` issue that would not mute user audio frames causing
transcriptions to be generated by the STT service.
### Other
- Added Gemini support to `examples/phone-chatbot`.
## [0.0.57] - 2025-02-14
### Added
- Added new `AudioContextWordTTSService`. This is a TTS base class for TTS
services that handling multiple separate audio requests.
- Added new frames `EmulateUserStartedSpeakingFrame` and
`EmulateUserStoppedSpeakingFrame` which can be used to emulated VAD behavior
without VAD being present or not being triggered.
- Added a new `audio_in_stream_on_start` field to `TransportParams`.
- Added a new method `start_audio_in_streaming` in the `BaseInputTransport`.
- This method should be used to start receiving the input audio in case the
field `audio_in_stream_on_start` is set to `false`.
- Added support for the `RTVIProcessor` to handle buffered audio in `base64`
format, converting it into InputAudioRawFrame for transport.
- Added support for the `RTVIProcessor` to trigger `start_audio_in_streaming`
only after the `client-ready` message.
- Added new `MUTE_UNTIL_FIRST_BOT_COMPLETE` strategy to `STTMuteStrategy`. This
strategy starts muted and remains muted until the first bot speech completes,
ensuring the bot's first response cannot be interrupted. This complements the
existing `FIRST_SPEECH` strategy which only mutes during the first detected
bot speech.
- Added support for Google Cloud Speech-to-Text V2 through `GoogleSTTService`.
- Added `RimeTTSService`, a new `WordTTSService`. Updated the foundational
example `07q-interruptible-rime.py` to use `RimeTTSService`.
- Added support for Groq's Whisper API through the new `GroqSTTService` and
OpenAI's Whisper API through the new `OpenAISTTService`. Introduced a new
base class `BaseWhisperSTTService` to handle common Whisper API
@@ -121,33 +18,10 @@ stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general"))
OpenAI-compatible interface. Also, added foundational example
`14n-function-calling-perplexity.py`.
- Added `DailyTransport.update_remote_participants()`. This allows you to update
remote participant's settings, like their permissions or which of their
devices are enabled. Requires that the local participant have participant
admin permission.
- Added `DailyTransport.update_remote_participants()`. This allows you to update remote participant's settings, like their permissions or which of their devices are enabled. Requires that the local participant have participant admin permission.
### Changed
- We don't consider a colon `:` and end of sentence any more.
- Updated `DailyTransport` to respect the `audio_in_stream_on_start` field,
ensuring it only starts receiving the audio input if it is enabled.
- Updated `FastAPIWebsocketOutputTransport` to send `TransportMessageFrame` and
`TransportMessageUrgentFrame` to the serializer.
- Updated `WebsocketServerOutputTransport` to send `TransportMessageFrame` and
`TransportMessageUrgentFrame` to the serializer.
- Enhanced `STTMuteConfig` to validate strategy combinations, preventing
`MUTE_UNTIL_FIRST_BOT_COMPLETE` and `FIRST_SPEECH` from being used together
as they handle first bot speech differently.
- Updated foundational example `07n-interruptible-google.py` to use all Google
services.
- `RimeHttpTTSService` now uses the `mistv2` model by default.
- Improved error handling in `AzureTTSService` to properly detect and log
synthesis cancellation errors.
@@ -161,8 +35,7 @@ stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general"))
model.
- `RTVIObserver` doesn't handle `LLMSearchResponseFrame` frames anymore. For
now, to handle those frames you need to create a `GoogleRTVIObserver`
instead.
now, to handle those frames you need to create a `GoogleRTVIObserver` instead.
### Deprecated
@@ -179,28 +52,8 @@ stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general"))
### Fixed
- Fixed a `FalImageGenService` issue that was causing the event loop to be
blocked while loading the downloadded image.
- Fixed a `CartesiaTTSService` service issue that would cause audio overlapping
in some cases.
- Fixed a websocket-based service issue (e.g. `CartesiaTTSService`) that was
preventing a reconnection after the server disconnected cleanly, which was
causing an inifite loop instead.
- Fixed a `BaseOutputTransport` issue that was causing upstream frames to no be
pushed upstream.
- Fixed multiple issue where user transcriptions where not being handled
properly. It was possible for short utterances to not trigger VAD which would
cause user transcriptions to be ignored. It was also possible for one or more
transcriptions to be generated after VAD in which case they would also be
ignored.
- Fixed an issue that was causing `BotStoppedSpeakingFrame` to be generated too
late. This could then cause issues unblocking `STTMuteFilter` later than
desired.
- Fixed a `CartesiaTTSService` issue that could cause audio overlapping in some
cases.
- Fixed an issue that was causing `AudioBufferProcessor` to not record
synchronized audio.
@@ -208,20 +61,8 @@ stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general"))
- Fixed an `RTVI` issue that was causing `bot-tts-text` messages to be sent
before being processed by the output transport.
- Fixed an issue[#1192] in 11labs where we are trying to reconnect/disconnect
the websocket connection even when the connection is already closed.
- Fixed an issue where `has_regular_messages` condition was always true in
`GoogleLLMContext` due to `Part` having `function_call` & `function_response`
with `None` values.
### Other
- Added new `instant-voice` example. This example showcases how to enable
instant voice communication as soon as a user connects.
- Added new `local-input-select-stt` example. This examples allows you to play
with local audio inputs by slecting them through a nice text interface.
- Fixed an issue[#1192] in 11labs where we are trying to reconnect/disconnect the
websocket connection even when the connection is already closed.
## [0.0.56] - 2025-02-06
@@ -242,8 +83,7 @@ stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general"))
logging setup.
- Fixed a `SentryMetrics` issue that was preventing any metrics to be sent to
Sentry and also was preventing from metrics frames to be pushed to the
pipeline.
Sentry and also was preventing from metrics frames to be pushed to the pipeline.
- Fixed an issue in `BaseOutputTransport` where incoming audio would not be
resampled to the desired output sample rate.
@@ -402,8 +242,7 @@ stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general"))
- Added `enable_recording` and `geo` parameters to `DailyRoomProperties`.
- Added `RecordingsBucketConfig` to `DailyRoomProperties` to upload recordings
to a custom AWS bucket.
- Added `RecordingsBucketConfig` to `DailyRoomProperties` to upload recordings to a custom AWS bucket.
### Changed

View File

@@ -57,7 +57,7 @@ pip install "pipecat-ai[option,...]"
| Category | Services | Install Command Example |
| ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------- |
| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Google](https://docs.pipecat.ai/server/services/stt/google), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | `pip install "pipecat-ai[deepgram]"` |
| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | `pip install "pipecat-ai[deepgram]"` |
| LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Perplexity](https://docs.pipecat.ai/server/services/llm/perplexity), [Together AI](https://docs.pipecat.ai/server/services/llm/together) | `pip install "pipecat-ai[openai]"` |
| Text-to-Speech | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | `pip install "pipecat-ai[cartesia]"` |
| Speech-to-Speech | [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) | `pip install "pipecat-ai[google]"` |

View File

@@ -1,11 +0,0 @@
coverage:
range: 50..90 # coverage lower than 50 is red, higher than 90 green, between color code
status:
project:
default:
target: auto # auto % coverage target
threshold: 5% # allow for 5% reduction of coverage without failing
# do not run coverage on patch nor changes
patch: false

View File

@@ -1,60 +0,0 @@
# React Native Implementation
Basic implementation using the [Pipecat React Native SDK](https://docs.pipecat.ai/client/react-native/introduction).
## Usage
### Expo requirements
This project cannot be used with an [Expo Go](https://docs.expo.dev/workflow/expo-go/) app because [it requires custom native code](https://docs.expo.io/workflow/customizing/).
When a project requires custom native code or a config plugin, we need to transition from using [Expo Go](https://docs.expo.dev/workflow/expo-go/)
to a [development build](https://docs.expo.dev/development/introduction/).
More details about the custom native code used by this demo can be found in [rn-daily-js-expo-config-plugin](https://github.com/daily-co/rn-daily-js-expo-config-plugin).
### Building remotely
If you do not have experience with Xcode and Android Studio builds or do not have them installed locally on your computer, you will need to follow [this guide from Expo to use EAS Build](https://docs.expo.dev/development/create-development-builds/#create-and-install-eas-build).
### Building locally
You will need to have installed locally on your computer:
- [Xcode](https://developer.apple.com/xcode/) to build for iOS;
- [Android Studio](https://developer.android.com/studio) to build for Android;
#### Install the demo dependencies
```bash
# Use the version of node specified in .nvmrc
nvm i
# Install dependencies
npm i
# Before a native app can be compiled, the native source code must be generated.
npx expo prebuild
# Configure the environment variable to connect to the local server
cp env.example .env
# edit .env and add your local ip address, for example: http://192.168.1.16:7860
```
#### Running on Android
After plugging in an Android device [configured for debugging](https://developer.android.com/studio/debug/dev-options), run the following command:
```
npm run android
```
#### Running on iOS
Run the following command:
```
npm run ios
```
#### Connect to the server
Use the http://localhost:5173 in your app.

View File

@@ -1,75 +0,0 @@
{
"expo": {
"name": "bot-ready-rn",
"slug": "bot-ready-rn",
"version": "1.0.0",
"orientation": "portrait",
"icon": "./assets/icon.png",
"userInterfaceStyle": "light",
"splash": {
"image": "./assets/splash.png",
"resizeMode": "contain",
"backgroundColor": "#ffffff"
},
"updates": {
"fallbackToCacheTimeout": 0
},
"assetBundlePatterns": [
"**/*"
],
"ios": {
"supportsTablet": true,
"bitcode": false,
"bundleIdentifier": "co.daily.expo.BotReady",
"infoPlist": {
"UIBackgroundModes": [
"voip"
]
},
"appleTeamId": "EEBGKV9N3N"
},
"android": {
"adaptiveIcon": {
"foregroundImage": "./assets/adaptive-icon.png",
"backgroundColor": "#FFFFFF"
},
"package": "co.daily.expo.BotReady",
"permissions": [
"android.permission.ACCESS_NETWORK_STATE",
"android.permission.BLUETOOTH",
"android.permission.CAMERA",
"android.permission.INTERNET",
"android.permission.MODIFY_AUDIO_SETTINGS",
"android.permission.RECORD_AUDIO",
"android.permission.SYSTEM_ALERT_WINDOW",
"android.permission.WAKE_LOCK",
"android.permission.FOREGROUND_SERVICE",
"android.permission.FOREGROUND_SERVICE_CAMERA",
"android.permission.FOREGROUND_SERVICE_MICROPHONE",
"android.permission.FOREGROUND_SERVICE_MEDIA_PROJECTION",
"android.permission.POST_NOTIFICATIONS"
]
},
"web": {
"favicon": "./assets/favicon.png"
},
"plugins": [
"@config-plugins/react-native-webrtc",
"@daily-co/config-plugin-rn-daily-js",
[
"expo-build-properties",
{
"android": {
"minSdkVersion": 24,
"compileSdkVersion": 35,
"targetSdkVersion": 34,
"buildToolsVersion": "35.0.0"
},
"ios": {
"deploymentTarget": "15.1"
}
}
]
]
}
}

Binary file not shown.

Before

Width:  |  Height:  |  Size: 17 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.4 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 22 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 46 KiB

View File

@@ -1,7 +0,0 @@
module.exports = function(api) {
api.cache(true);
return {
presets: ['babel-preset-expo'],
plugins: [["module:react-native-dotenv"]],
};
};

View File

@@ -1 +0,0 @@
API_BASE_URL=http://YOUR_LOCAL_IP:7860

View File

@@ -1,7 +0,0 @@
import { registerRootComponent } from "expo";
import App from "./src/App";
// registerRootComponent calls AppRegistry.registerComponent('main', () => App);
// It also ensures that the environment is set up appropriately
registerRootComponent(App);

View File

@@ -1,4 +0,0 @@
// Learn more https://docs.expo.io/guides/customizing-metro
const { getDefaultConfig } = require('expo/metro-config');
module.exports = getDefaultConfig(__dirname);

File diff suppressed because it is too large Load Diff

View File

@@ -1,31 +0,0 @@
{
"name": "bot-ready-rn",
"version": "1.0.0",
"scripts": {
"start": "expo start --dev-client",
"android": "expo run:android --device",
"ios": "expo run:ios --device",
"web": "expo start --web"
},
"dependencies": {
"@config-plugins/react-native-webrtc": "^10.0.0",
"@daily-co/config-plugin-rn-daily-js": "0.0.7",
"@daily-co/react-native-daily-js": "^0.70.0",
"@daily-co/react-native-webrtc": "^118.0.3-daily.2",
"@react-native-async-storage/async-storage": "1.23.1",
"expo": "^52.0.0",
"expo-build-properties": "~0.13.1",
"expo-dev-client": "~5.0.5",
"expo-splash-screen": "~0.29.16",
"expo-status-bar": "~2.0.0",
"react": "18.3.1",
"react-native": "0.76.3",
"react-native-background-timer": "^2.4.1",
"react-native-dotenv": "^3.4.11",
"react-native-get-random-values": "^1.11.0"
},
"devDependencies": {
"@babel/core": "^7.12.9"
},
"private": true
}

View File

@@ -1,121 +0,0 @@
import React, { useState, useEffect } from 'react';
import {SafeAreaView, View, Text, Button, StyleSheet, ScrollView} from 'react-native';
import Daily from "@daily-co/react-native-daily-js";
import { API_BASE_URL } from "@env";
const CallScreen = () => {
const [connectionStatus, setConnectionStatus] = useState('Disconnected');
const [isConnected, setIsConnected] = useState(false);
const [callObject, setCallObject] = useState(null);
const [logs, setLogs] = useState([]);
useEffect(() => {
if (callObject) {
setupTrackListeners(callObject);
}
}, [callObject]);
const log = (message) => {
setLogs((prevLogs) => [...prevLogs, `${new Date().toISOString()} - ${message}`]);
console.log(message);
};
const setupTrackListeners = (callObject) => {
callObject.on("joined-meeting", () => {
setConnectionStatus('Connected');
setIsConnected(true);
log('Client connected');
});
callObject.on("left-meeting", () => {
setConnectionStatus('Disconnected');
setIsConnected(false);
log('Client disconnected');
});
callObject.on("participant-left", () => {
// When the bot leaves, we are also disconnecting from the call
disconnect().catch((err) => {
log(`Failed to disconnect ${err}`);
})
});
// Trigger so the bot can start sending audio
callObject.on("track-started", (evt) => {
if (evt.track.kind === "audio" && evt.participant.local === false) {
handleEventToConsole(evt)
log("Sending the message that will trigger the bot to play the audio.")
callObject.sendAppMessage("playable")
}
});
callObject.on("error", (evt) => log(`Error: ${evt.error}`));
// Other events just for awareness
callObject.on("track-stopped", handleEventToConsole);
callObject.on("participant-joined", handleEventToConsole);
callObject.on("participant-updated", handleEventToConsole);
};
const handleEventToConsole = (evt) => {
log(`Received event: ${evt.action}`);
};
const connect = async () => {
try {
const callObject = Daily.createCallObject({ subscribeToTracksAutomatically: true });
setCallObject(callObject);
const connectionUrl = `${API_BASE_URL}/connect`
const res = await fetch(connectionUrl, { method: "POST", headers: { "Content-Type": "application/json" } });
const roomInfo = await res.json();
await callObject.join({ url: roomInfo.room_url });
} catch (error) {
log(`Error connecting: ${error.message}`);
}
};
const disconnect = async () => {
if (callObject) {
try {
await callObject.leave();
await callObject.destroy();
setCallObject(null);
} catch (error) {
log(`Error disconnecting: ${error.message}`);
}
}
};
return (
<SafeAreaView style={styles.safeArea}>
<View style={styles.container}>
<View style={styles.statusBar}>
<Text>Status: <Text style={styles.status}>{connectionStatus}</Text></Text>
<View style={styles.controls}>
<Button
title={isConnected ? "Disconnect" : "Connect"}
onPress={isConnected ? disconnect : connect}
/>
</View>
</View>
<View style={styles.debugPanel}>
<Text style={styles.debugTitle}>Debug Info</Text>
<ScrollView style={styles.debugLog}>
{logs.map((logEntry, index) => (
<Text key={index} style={styles.logText}>{logEntry}</Text>
))}
</ScrollView>
</View>
</View>
</SafeAreaView>
);
};
const styles = StyleSheet.create({
safeArea: { flex: 1, backgroundColor: '#f0f0f0', padding: 20 },
container: { flex: 1, margin: 20 },
statusBar: { flexDirection: 'row', justifyContent: 'space-between', alignItems: 'center', padding: 10, backgroundColor: '#fff', borderRadius: 8, marginBottom: 20 },
status: { fontWeight: 'bold' },
controls: { flexDirection: 'row', gap: 10 },
debugPanel: { height: '80%', backgroundColor: '#fff', borderRadius: 8, padding: 20},
debugTitle: { fontSize: 16, fontWeight: 'bold' },
debugLog: { height: '100%', overflow: 'scroll', backgroundColor: '#f8f8f8', padding: 10, borderRadius: 4, fontFamily: 'monospace', fontSize: 12, lineHeight: 1.4 },
});
export default CallScreen;

View File

@@ -16,7 +16,7 @@ from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams
from pipecat.transports.local.audio import LocalAudioTransport, LocalTransportParams
load_dotenv(override=True)
@@ -25,7 +25,7 @@ logger.add(sys.stderr, level="DEBUG")
async def main():
transport = LocalAudioTransport(LocalAudioTransportParams(audio_out_enabled=True))
transport = LocalAudioTransport(LocalTransportParams(audio_out_enabled=True))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),

View File

@@ -27,7 +27,7 @@ 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.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.cartesia import CartesiaHttpTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -91,7 +91,7 @@ async def main():
),
)
tts = CartesiaTTSService(
tts = CartesiaHttpTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
)

View File

@@ -18,7 +18,9 @@ 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.services.google import GoogleLLMService, GoogleSTTService, GoogleTTSService
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.services.google import GoogleTTSService
from pipecat.services.openai import OpenAILLMService
from pipecat.transcriptions.language import Language
from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -44,16 +46,14 @@ async def main():
),
)
stt = GoogleSTTService(
params=GoogleSTTService.InputParams(languages=Language.EN_US),
)
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = GoogleTTSService(
voice_id="en-US-Journey-F",
params=GoogleTTSService.InputParams(language=Language.EN_US),
)
llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"))
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
messages = [
{

View File

@@ -19,7 +19,7 @@ from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai import OpenAILLMService
from pipecat.services.rime import RimeTTSService
from pipecat.services.rime import RimeHttpTTSService
from pipecat.transports.services.daily import DailyParams, DailyTransport
load_dotenv(override=True)
@@ -44,9 +44,10 @@ async def main():
),
)
tts = RimeTTSService(
tts = RimeHttpTTSService(
api_key=os.getenv("RIME_API_KEY", ""),
voice_id="rex",
params=RimeHttpTTSService.InputParams(reduce_latency=True),
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")

View File

@@ -16,7 +16,7 @@ from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.whisper import WhisperSTTService
from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams
from pipecat.transports.local.audio import LocalAudioTransport, LocalTransportParams
load_dotenv(override=True)
@@ -33,7 +33,7 @@ class TranscriptionLogger(FrameProcessor):
async def main():
transport = LocalAudioTransport(LocalAudioTransportParams(audio_in_enabled=True))
transport = LocalAudioTransport(LocalTransportParams(audio_in_enabled=True))
stt = WhisperSTTService()

View File

@@ -14,7 +14,6 @@ from loguru import logger
from runner import configure
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -31,12 +30,6 @@ logger.add(sys.stderr, level="DEBUG")
video_participant_id = None
async def start_fetch_weather(function_name, llm, context):
"""Push a frame to the LLM; this is handy when the LLM response might take a while."""
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")
async def get_weather(function_name, tool_call_id, arguments, llm, context, result_callback):
location = arguments["location"]
await result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
@@ -70,7 +63,7 @@ async def main():
)
llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001")
llm.register_function("get_weather", get_weather, start_fetch_weather)
llm.register_function("get_weather", get_weather)
llm.register_function("get_image", get_image)
tools = [

View File

@@ -38,6 +38,7 @@ async def main():
"GStreamer",
DailyParams(
audio_out_enabled=True,
audio_out_is_live=True,
camera_out_enabled=True,
camera_out_width=1280,
camera_out_height=720,

View File

@@ -497,7 +497,7 @@ class UserAggregatorBuffer(LLMResponseAggregator):
if isinstance(frame, UserStartedSpeakingFrame):
self._transcription = ""
async def push_aggregation(self):
async def _push_aggregation(self):
if self._aggregation:
self._transcription = self._aggregation
self._aggregation = ""

View File

@@ -62,10 +62,7 @@ async def main():
# Configure the mute processor with both strategies
stt_mute_processor = STTMuteFilter(
config=STTMuteConfig(
strategies={
STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE,
STTMuteStrategy.FUNCTION_CALL,
}
strategies={STTMuteStrategy.FIRST_SPEECH, STTMuteStrategy.FUNCTION_CALL}
),
)

View File

@@ -143,14 +143,7 @@ class InputTranscriptionContextFilter(FrameProcessor):
return
try:
# Make sure we're working with a GoogleLLMContext
context = GoogleLLMContext.upgrade_to_google(frame.context)
message = context.messages[-1]
if not isinstance(message, glm.Content):
logger.error(f"Expected glm.Content, got {type(message)}")
return
message = frame.context.messages[-1]
last_part = message.parts[-1]
if not (
message.role == "user"

View File

@@ -18,10 +18,12 @@ from pipecat.frames.frames import (
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
Frame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMTextFrame,
StartInterruptionFrame,
)
from pipecat.observers.base_observer import BaseObserver
from pipecat.observers.loggers.llm_log_observer import LLMLogObserver
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -71,6 +73,38 @@ class DebugObserver(BaseObserver):
logger.info(f"🤖 BOT STOP SPEAKING: {src} {arrow} {dst} at {time_sec:.2f}s")
class LLMLogObserver(BaseObserver):
"""Observer to log LLM activity to the console.
Logs all frame instances of:
- LLMFullResponseStartFrame (only from LLM service)
- LLMTextFrame
- LLMFullResponseEndFrame (only from LLM service)
This allows you to track when the LLM starts responding, what it generates, and when it finishes.
Log format: [LLM EVENT]: [details] at [timestamp]s
"""
async def on_push_frame(
self,
src: FrameProcessor,
dst: FrameProcessor,
frame: Frame,
direction: FrameDirection,
timestamp: int,
):
time_sec = timestamp / 1_000_000_000
# Only log start/end frames from OpenAILLMService
if isinstance(frame, (LLMFullResponseStartFrame, LLMFullResponseEndFrame)):
if isinstance(src, OpenAILLMService):
event = "START" if isinstance(frame, LLMFullResponseStartFrame) else "END"
logger.info(f"🧠 LLM {event} RESPONSE at {time_sec:.2f}s")
# Log all LLMTextFrames
elif isinstance(frame, LLMTextFrame):
logger.info(f"🧠 LLM GENERATING: {frame.text!r} at {time_sec:.2f}s")
async def main():
async with aiohttp.ClientSession() as session:
(room_url, token) = await configure(session)

View File

@@ -1,7 +0,0 @@
venv
.idea
server/.env
client/javascript/node_modules
client/javascript/.vite
bkp

View File

@@ -1,55 +0,0 @@
# Instant Voice
This demo showcases how to enable instant voice communication as soon as a user connects.
By leveraging optimizations on both the server and client sides, users can start speaking immediately after pressing the connect button.
## How It Works
### Server-Side Improvements:
- A **pool of Daily rooms** is managed to ensure quick connections.
- When a user connects, an existing room from the pool is assigned.
- A new room is created asynchronously to maintain the predefined pool size.
### Client-Side Improvements:
- Using the **DailyTransport** property `bufferLocalAudioUntilBotReady` set to enabled, users can start speaking immediately
upon receiving the `AUDIO_BUFFERING_STARTED` event (typically within ~1s).
- This allows users to speak even before the bot is fully ready or the WebRTC connection is fully established.
## Quick Start
### 1. Start the Bot Server
1. Navigate to the server directory:
```bash
cd server
```
2. Create and activate a virtual environment:
```bash
python3 -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
```
3. Install dependencies:
```bash
pip install -r requirements.txt
```
4. Copy the `.env.example` file to `.env` and configure it:
- Add your API keys.
5. Start the server:
```bash
python src/server.py
```
### 2. Connect Using the Client App
For client-side setup, refer to the [JavaScript Guide](client/javascript/README.md).
## Important Notes
- The bot server **must** be running before using the client implementation.
- Ensure your environment variables are correctly set up.
## Requirements
- **Python 3.10+**
- **Node.js 16+** (for JavaScript/React client)
- **Daily API key**
- **Google API key**
- **Modern web browser with WebRTC support**

View File

@@ -1,27 +0,0 @@
# 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.

View File

@@ -1,37 +0,0 @@
<!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">
Buffering audio: <span id="buffering-status">No</span>
</div>
<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>

File diff suppressed because it is too large Load Diff

View File

@@ -1,24 +0,0 @@
{
"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.13.1",
"@vitejs/plugin-react-swc": "^3.7.2",
"typescript": "^5.7.3",
"vite": "^6.0.2"
},
"dependencies": {
"@pipecat-ai/client-js": "^0.3.2",
"@pipecat-ai/daily-transport": "^0.3.5"
}
}

View File

@@ -1,268 +0,0 @@
/**
* Copyright (c) 20242025, Daily
*
* SPDX-License-Identifier: BSD 2-Clause License
*/
/**
* RTVI Client Implementation
*
* This client connects to an RTVI-compatible bot server using WebRTC (via Daily).
* It handles audio/video streaming and manages the connection lifecycle.
*
* Requirements:
* - A running RTVI bot server (defaults to http://localhost:7860)
* - The server must implement the /connect endpoint that returns Daily.co room credentials
* - Browser with WebRTC support
*/
import {
Participant,
RTVIClient,
RTVIClientOptions,
RTVIEvent,
} from '@pipecat-ai/client-js';
import { DailyTransport } from '@pipecat-ai/daily-transport';
import SoundUtils from "./util/soundUtils";
import { InstantVoiceHelper } from "./util/instantVoiceHelper";
/**
* InstantVoiceClient handles the connection and media management for a real-time
* voice and video interaction with an AI bot.
*/
class InstantVoiceClient {
private declare rtviClient: RTVIClient;
private connectBtn: HTMLButtonElement | null = null;
private disconnectBtn: HTMLButtonElement | null = null;
private statusSpan: HTMLElement | null = null;
private bufferingAudioSpan: HTMLElement | null = null;
private debugLog: HTMLElement | null = null;
private botAudio: HTMLAudioElement;
private declare startTime: number;
constructor() {
this.botAudio = document.createElement('audio');
this.botAudio.autoplay = true;
document.body.appendChild(this.botAudio);
this.setupDOMElements();
this.setupEventListeners();
this.initializeRTVIClient();
}
/**
* 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.bufferingAudioSpan = document.getElementById('buffering-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());
}
private initializeRTVIClient(): void {
const transport = new DailyTransport({
bufferLocalAudioUntilBotReady: true
});
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');
this.updateBufferingStatus('No');
if (this.connectBtn) this.connectBtn.disabled = false;
if (this.disconnectBtn) this.disconnectBtn.disabled = true;
this.log('Client disconnected');
},
onBotConnected: (participant: Participant) => {
this.log(`onBotConnected, timeTaken: ${Date.now() - this.startTime}`);
},
onBotReady: (data) => {
this.log(`onBotReady, timeTaken: ${Date.now() - this.startTime}`);
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.rtviClient.registerHelper("transport", new InstantVoiceHelper({
callbacks: {
onAudioBufferingStarted: () => {
SoundUtils.beep()
this.updateBufferingStatus('Yes');
this.log(`onMicCaptureStarted, timeTaken: ${Date.now() - this.startTime}`);
},
onAudioBufferingStopped: () => {
this.updateBufferingStatus('No');
this.log(`onMicCaptureStopped, timeTaken: ${Date.now() - this.startTime}`);
}
}
}
));
this.setupTrackListeners();
}
/**
* 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}`);
}
/**
* Update the connection status display
*/
private updateBufferingStatus(status: string): void {
if (this.bufferingAudioSpan) {
this.bufferingAudioSpan.textContent = status;
}
this.log(`BufferingStatus: ${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 {
this.startTime = Date.now();
this.log('Connecting to bot...');
await this.rtviClient.connect();
} catch (error) {
this.log(`Error connecting: ${(error as Error).message}`);
this.updateStatus('Error');
this.updateBufferingStatus('No');
// 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> {
try {
await this.rtviClient.disconnect();
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 {
InstantVoiceClient: typeof InstantVoiceClient;
}
}
window.addEventListener('DOMContentLoaded', () => {
window.InstantVoiceClient = InstantVoiceClient;
new InstantVoiceClient();
});

View File

@@ -1,98 +0,0 @@
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;
}

View File

@@ -1,39 +0,0 @@
import {RTVIClientHelper, RTVIClientHelperOptions, RTVIMessage} from "@pipecat-ai/client-js";
import {DailyRTVIMessageType} from '@pipecat-ai/daily-transport';
export type InstantVoiceHelperCallbacks = Partial<{
onAudioBufferingStarted: () => void;
onAudioBufferingStopped: () => void;
}>;
// --- Interface and class
export interface InstantVoiceHelperOptions extends RTVIClientHelperOptions {
callbacks?: InstantVoiceHelperCallbacks;
}
export class InstantVoiceHelper extends RTVIClientHelper {
protected declare _options: InstantVoiceHelperOptions;
constructor(options: InstantVoiceHelperOptions) {
super(options);
}
handleMessage(rtviMessage: RTVIMessage): void {
switch (rtviMessage.type) {
case DailyRTVIMessageType.AUDIO_BUFFERING_STARTED:
if (this._options.callbacks?.onAudioBufferingStarted) {
this._options.callbacks?.onAudioBufferingStarted()
}
break;
case DailyRTVIMessageType.AUDIO_BUFFERING_STOPPED:
if (this._options.callbacks?.onAudioBufferingStopped) {
this._options.callbacks?.onAudioBufferingStopped()
}
break;
}
}
getMessageTypes(): string[] {
return [DailyRTVIMessageType.AUDIO_BUFFERING_STARTED, DailyRTVIMessageType.AUDIO_BUFFERING_STOPPED];
}
}

View File

@@ -1,8 +0,0 @@
class SoundUtils {
static beep() {
const snd = new Audio("data:audio/wav;base64,//uQRAAAAWMSLwUIYAAsYkXgoQwAEaYLWfkWgAI0wWs/ItAAAGDgYtAgAyN+QWaAAihwMWm4G8QQRDiMcCBcH3Cc+CDv/7xA4Tvh9Rz/y8QADBwMWgQAZG/ILNAARQ4GLTcDeIIIhxGOBAuD7hOfBB3/94gcJ3w+o5/5eIAIAAAVwWgQAVQ2ORaIQwEMAJiDg95G4nQL7mQVWI6GwRcfsZAcsKkJvxgxEjzFUgfHoSQ9Qq7KNwqHwuB13MA4a1q/DmBrHgPcmjiGoh//EwC5nGPEmS4RcfkVKOhJf+WOgoxJclFz3kgn//dBA+ya1GhurNn8zb//9NNutNuhz31f////9vt///z+IdAEAAAK4LQIAKobHItEIYCGAExBwe8jcToF9zIKrEdDYIuP2MgOWFSE34wYiR5iqQPj0JIeoVdlG4VD4XA67mAcNa1fhzA1jwHuTRxDUQ//iYBczjHiTJcIuPyKlHQkv/LHQUYkuSi57yQT//uggfZNajQ3Vmz+Zt//+mm3Wm3Q576v////+32///5/EOgAAADVghQAAAAA//uQZAUAB1WI0PZugAAAAAoQwAAAEk3nRd2qAAAAACiDgAAAAAAABCqEEQRLCgwpBGMlJkIz8jKhGvj4k6jzRnqasNKIeoh5gI7BJaC1A1AoNBjJgbyApVS4IDlZgDU5WUAxEKDNmmALHzZp0Fkz1FMTmGFl1FMEyodIavcCAUHDWrKAIA4aa2oCgILEBupZgHvAhEBcZ6joQBxS76AgccrFlczBvKLC0QI2cBoCFvfTDAo7eoOQInqDPBtvrDEZBNYN5xwNwxQRfw8ZQ5wQVLvO8OYU+mHvFLlDh05Mdg7BT6YrRPpCBznMB2r//xKJjyyOh+cImr2/4doscwD6neZjuZR4AgAABYAAAABy1xcdQtxYBYYZdifkUDgzzXaXn98Z0oi9ILU5mBjFANmRwlVJ3/6jYDAmxaiDG3/6xjQQCCKkRb/6kg/wW+kSJ5//rLobkLSiKmqP/0ikJuDaSaSf/6JiLYLEYnW/+kXg1WRVJL/9EmQ1YZIsv/6Qzwy5qk7/+tEU0nkls3/zIUMPKNX/6yZLf+kFgAfgGyLFAUwY//uQZAUABcd5UiNPVXAAAApAAAAAE0VZQKw9ISAAACgAAAAAVQIygIElVrFkBS+Jhi+EAuu+lKAkYUEIsmEAEoMeDmCETMvfSHTGkF5RWH7kz/ESHWPAq/kcCRhqBtMdokPdM7vil7RG98A2sc7zO6ZvTdM7pmOUAZTnJW+NXxqmd41dqJ6mLTXxrPpnV8avaIf5SvL7pndPvPpndJR9Kuu8fePvuiuhorgWjp7Mf/PRjxcFCPDkW31srioCExivv9lcwKEaHsf/7ow2Fl1T/9RkXgEhYElAoCLFtMArxwivDJJ+bR1HTKJdlEoTELCIqgEwVGSQ+hIm0NbK8WXcTEI0UPoa2NbG4y2K00JEWbZavJXkYaqo9CRHS55FcZTjKEk3NKoCYUnSQ0rWxrZbFKbKIhOKPZe1cJKzZSaQrIyULHDZmV5K4xySsDRKWOruanGtjLJXFEmwaIbDLX0hIPBUQPVFVkQkDoUNfSoDgQGKPekoxeGzA4DUvnn4bxzcZrtJyipKfPNy5w+9lnXwgqsiyHNeSVpemw4bWb9psYeq//uQZBoABQt4yMVxYAIAAAkQoAAAHvYpL5m6AAgAACXDAAAAD59jblTirQe9upFsmZbpMudy7Lz1X1DYsxOOSWpfPqNX2WqktK0DMvuGwlbNj44TleLPQ+Gsfb+GOWOKJoIrWb3cIMeeON6lz2umTqMXV8Mj30yWPpjoSa9ujK8SyeJP5y5mOW1D6hvLepeveEAEDo0mgCRClOEgANv3B9a6fikgUSu/DmAMATrGx7nng5p5iimPNZsfQLYB2sDLIkzRKZOHGAaUyDcpFBSLG9MCQALgAIgQs2YunOszLSAyQYPVC2YdGGeHD2dTdJk1pAHGAWDjnkcLKFymS3RQZTInzySoBwMG0QueC3gMsCEYxUqlrcxK6k1LQQcsmyYeQPdC2YfuGPASCBkcVMQQqpVJshui1tkXQJQV0OXGAZMXSOEEBRirXbVRQW7ugq7IM7rPWSZyDlM3IuNEkxzCOJ0ny2ThNkyRai1b6ev//3dzNGzNb//4uAvHT5sURcZCFcuKLhOFs8mLAAEAt4UWAAIABAAAAAB4qbHo0tIjVkUU//uQZAwABfSFz3ZqQAAAAAngwAAAE1HjMp2qAAAAACZDgAAAD5UkTE1UgZEUExqYynN1qZvqIOREEFmBcJQkwdxiFtw0qEOkGYfRDifBui9MQg4QAHAqWtAWHoCxu1Yf4VfWLPIM2mHDFsbQEVGwyqQoQcwnfHeIkNt9YnkiaS1oizycqJrx4KOQjahZxWbcZgztj2c49nKmkId44S71j0c8eV9yDK6uPRzx5X18eDvjvQ6yKo9ZSS6l//8elePK/Lf//IInrOF/FvDoADYAGBMGb7FtErm5MXMlmPAJQVgWta7Zx2go+8xJ0UiCb8LHHdftWyLJE0QIAIsI+UbXu67dZMjmgDGCGl1H+vpF4NSDckSIkk7Vd+sxEhBQMRU8j/12UIRhzSaUdQ+rQU5kGeFxm+hb1oh6pWWmv3uvmReDl0UnvtapVaIzo1jZbf/pD6ElLqSX+rUmOQNpJFa/r+sa4e/pBlAABoAAAAA3CUgShLdGIxsY7AUABPRrgCABdDuQ5GC7DqPQCgbbJUAoRSUj+NIEig0YfyWUho1VBBBA//uQZB4ABZx5zfMakeAAAAmwAAAAF5F3P0w9GtAAACfAAAAAwLhMDmAYWMgVEG1U0FIGCBgXBXAtfMH10000EEEEEECUBYln03TTTdNBDZopopYvrTTdNa325mImNg3TTPV9q3pmY0xoO6bv3r00y+IDGid/9aaaZTGMuj9mpu9Mpio1dXrr5HERTZSmqU36A3CumzN/9Robv/Xx4v9ijkSRSNLQhAWumap82WRSBUqXStV/YcS+XVLnSS+WLDroqArFkMEsAS+eWmrUzrO0oEmE40RlMZ5+ODIkAyKAGUwZ3mVKmcamcJnMW26MRPgUw6j+LkhyHGVGYjSUUKNpuJUQoOIAyDvEyG8S5yfK6dhZc0Tx1KI/gviKL6qvvFs1+bWtaz58uUNnryq6kt5RzOCkPWlVqVX2a/EEBUdU1KrXLf40GoiiFXK///qpoiDXrOgqDR38JB0bw7SoL+ZB9o1RCkQjQ2CBYZKd/+VJxZRRZlqSkKiws0WFxUyCwsKiMy7hUVFhIaCrNQsKkTIsLivwKKigsj8XYlwt/WKi2N4d//uQRCSAAjURNIHpMZBGYiaQPSYyAAABLAAAAAAAACWAAAAApUF/Mg+0aohSIRobBAsMlO//Kk4soosy1JSFRYWaLC4qZBYWFRGZdwqKiwkNBVmoWFSJkWFxX4FFRQWR+LsS4W/rFRb/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////VEFHAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAU291bmRib3kuZGUAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAMjAwNGh0dHA6Ly93d3cuc291bmRib3kuZGUAAAAAAAAAACU=");
void snd.play();
}
}
export default SoundUtils;

View File

@@ -1,111 +0,0 @@
{
"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. */
}
}

View File

@@ -1,15 +0,0 @@
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,
},
},
},
});

View File

@@ -1,339 +0,0 @@
# THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY.
# yarn lockfile v1
"@babel/runtime@^7.12.5":
version "7.26.0"
resolved "https://registry.npmjs.org/@babel/runtime/-/runtime-7.26.0.tgz"
integrity sha512-FDSOghenHTiToteC/QRlv2q3DhPZ/oOXTBoirfWNx1Cx3TMVcGWQtMMmQcSvb/JjpNeGzx8Pq/b4fKEJuWm1sw==
dependencies:
regenerator-runtime "^0.14.0"
"@daily-co/daily-js@^0.73.0":
version "0.73.0"
resolved "https://registry.npmjs.org/@daily-co/daily-js/-/daily-js-0.73.0.tgz"
integrity sha512-Wz8c60hgmkx8fcEeDAi4L4J0rbafiihWKyXFyhYoFYPsw2OdChHpA4RYwIB+1enRws5IK+/HdmzFDYLQsB4A6w==
dependencies:
"@babel/runtime" "^7.12.5"
"@sentry/browser" "^8.33.1"
bowser "^2.8.1"
dequal "^2.0.3"
events "^3.1.0"
"@esbuild/darwin-arm64@0.24.0":
version "0.24.0"
resolved "https://registry.npmjs.org/@esbuild/darwin-arm64/-/darwin-arm64-0.24.0.tgz"
integrity sha512-CKyDpRbK1hXwv79soeTJNHb5EiG6ct3efd/FTPdzOWdbZZfGhpbcqIpiD0+vwmpu0wTIL97ZRPZu8vUt46nBSw==
"@pipecat-ai/client-js@^0.3.2", "@pipecat-ai/client-js@~0.3.2":
version "0.3.2"
resolved "https://registry.npmjs.org/@pipecat-ai/client-js/-/client-js-0.3.2.tgz"
integrity sha512-psunOVrJjPka2SWlq53vxVWCA0Vt8pSXsXtn8pOLC0YTKFsUx+b7Z6quYUJcDZjCe1aAg9cKETek3Xal3Co8Tg==
dependencies:
"@types/events" "^3.0.3"
clone-deep "^4.0.1"
events "^3.3.0"
typed-emitter "^2.1.0"
uuid "^10.0.0"
"@pipecat-ai/daily-transport@^0.3.5":
version "0.3.5"
resolved "https://registry.npmjs.org/@pipecat-ai/daily-transport/-/daily-transport-0.3.5.tgz"
integrity sha512-nJ0TvWPCqXPmU81U8cXOqk5mUEEvEuI06Mis+N0jN8KZUrNy1pP08iWbs07ObmIXdnQcoL+kQmHOerT4q/bF0w==
dependencies:
"@daily-co/daily-js" "^0.73.0"
"@rollup/rollup-darwin-arm64@4.28.0":
version "4.28.0"
resolved "https://registry.npmjs.org/@rollup/rollup-darwin-arm64/-/rollup-darwin-arm64-4.28.0.tgz"
integrity sha512-lmKx9yHsppblnLQZOGxdO66gT77bvdBtr/0P+TPOseowE7D9AJoBw8ZDULRasXRWf1Z86/gcOdpBrV6VDUY36Q==
"@sentry-internal/browser-utils@8.49.0":
version "8.49.0"
resolved "https://registry.npmjs.org/@sentry-internal/browser-utils/-/browser-utils-8.49.0.tgz"
integrity sha512-XkPHHdFqsN7EPaB+QGUOEmpFqXiqP67t2rRZ1HG1UwJoe0PhJEKNy7b4+WRwmT7ODSt+PvFk1gNBlJBpThwH7Q==
dependencies:
"@sentry/core" "8.49.0"
"@sentry-internal/feedback@8.49.0":
version "8.49.0"
resolved "https://registry.npmjs.org/@sentry-internal/feedback/-/feedback-8.49.0.tgz"
integrity sha512-v/wf7WvPxEvZUB7xrCnecI3fhevVo84hw8WlxgZIz6mLUHXEIX8xYWc9H8Yet/KKJ2uEB8GQ8aDsY6S1hVEIUA==
dependencies:
"@sentry/core" "8.49.0"
"@sentry-internal/replay-canvas@8.49.0":
version "8.49.0"
resolved "https://registry.npmjs.org/@sentry-internal/replay-canvas/-/replay-canvas-8.49.0.tgz"
integrity sha512-/yXxI7f+Wu24FIYoRE7A0AidNxORuhAyPzb5ey1wFqMXP72nG8dXhOpcl0w+bi554FkqkLjdeUDhSOBWYZXH9g==
dependencies:
"@sentry-internal/replay" "8.49.0"
"@sentry/core" "8.49.0"
"@sentry-internal/replay@8.49.0":
version "8.49.0"
resolved "https://registry.npmjs.org/@sentry-internal/replay/-/replay-8.49.0.tgz"
integrity sha512-BDiiCBxskkktTd6FNplBc9V8l14R4T/AwRIZj2itX4xnuHewTTDjVbeyvGol4roA4r+V0Mzoi31hLEGI6yFQ5Q==
dependencies:
"@sentry-internal/browser-utils" "8.49.0"
"@sentry/core" "8.49.0"
"@sentry/browser@^8.33.1":
version "8.49.0"
resolved "https://registry.npmjs.org/@sentry/browser/-/browser-8.49.0.tgz"
integrity sha512-dS4Sw2h8EixHeXOIR++XEVMTen6xCGcIQ/XhJbsjqvddXeIijW0WkxSeTfPkfs0dsqFHSisWmlmo0xhHbXvEsQ==
dependencies:
"@sentry-internal/browser-utils" "8.49.0"
"@sentry-internal/feedback" "8.49.0"
"@sentry-internal/replay" "8.49.0"
"@sentry-internal/replay-canvas" "8.49.0"
"@sentry/core" "8.49.0"
"@sentry/core@8.49.0":
version "8.49.0"
resolved "https://registry.npmjs.org/@sentry/core/-/core-8.49.0.tgz"
integrity sha512-/OAm6LdHhh8TvfDAucWfSJV7M03IOHrJm5LVjrrKr4gwQ1HKd4CDbARsBbPwHIzSRAle0IgG3sbJxEvv52JUIw==
"@swc/core-darwin-arm64@1.10.14":
version "1.10.14"
resolved "https://registry.npmjs.org/@swc/core-darwin-arm64/-/core-darwin-arm64-1.10.14.tgz"
integrity sha512-Dh4VyrhDDb05tdRmqJ/MucOPMTnrB4pRJol18HVyLlqu1HOT5EzonUniNTCdQbUXjgdv5UVJSTE1lYTzrp+myA==
"@swc/core@^1.7.26":
version "1.10.14"
resolved "https://registry.npmjs.org/@swc/core/-/core-1.10.14.tgz"
integrity sha512-WSrnE6JRnH20ZYjOOgSS4aOaPv9gxlkI2KRkN24kagbZnPZMnN8bZZyzw1rrLvwgpuRGv17Uz+hflosbR+SP6w==
dependencies:
"@swc/counter" "^0.1.3"
"@swc/types" "^0.1.17"
optionalDependencies:
"@swc/core-darwin-arm64" "1.10.14"
"@swc/core-darwin-x64" "1.10.14"
"@swc/core-linux-arm-gnueabihf" "1.10.14"
"@swc/core-linux-arm64-gnu" "1.10.14"
"@swc/core-linux-arm64-musl" "1.10.14"
"@swc/core-linux-x64-gnu" "1.10.14"
"@swc/core-linux-x64-musl" "1.10.14"
"@swc/core-win32-arm64-msvc" "1.10.14"
"@swc/core-win32-ia32-msvc" "1.10.14"
"@swc/core-win32-x64-msvc" "1.10.14"
"@swc/counter@^0.1.3":
version "0.1.3"
resolved "https://registry.npmjs.org/@swc/counter/-/counter-0.1.3.tgz"
integrity sha512-e2BR4lsJkkRlKZ/qCHPw9ZaSxc0MVUd7gtbtaB7aMvHeJVYe8sOB8DBZkP2DtISHGSku9sCK6T6cnY0CtXrOCQ==
"@swc/types@^0.1.17":
version "0.1.17"
resolved "https://registry.npmjs.org/@swc/types/-/types-0.1.17.tgz"
integrity sha512-V5gRru+aD8YVyCOMAjMpWR1Ui577DD5KSJsHP8RAxopAH22jFz6GZd/qxqjO6MJHQhcsjvjOFXyDhyLQUnMveQ==
dependencies:
"@swc/counter" "^0.1.3"
"@types/estree@1.0.6":
version "1.0.6"
resolved "https://registry.npmjs.org/@types/estree/-/estree-1.0.6.tgz"
integrity sha512-AYnb1nQyY49te+VRAVgmzfcgjYS91mY5P0TKUDCLEM+gNnA+3T6rWITXRLYCpahpqSQbN5cE+gHpnPyXjHWxcw==
"@types/events@^3.0.3":
version "3.0.3"
resolved "https://registry.npmjs.org/@types/events/-/events-3.0.3.tgz"
integrity sha512-trOc4AAUThEz9hapPtSd7wf5tiQKvTtu5b371UxXdTuqzIh0ArcRspRP0i0Viu+LXstIQ1z96t1nsPxT9ol01g==
"@types/node@^18.0.0 || ^20.0.0 || >=22.0.0", "@types/node@^22.13.1":
version "22.13.1"
resolved "https://registry.npmjs.org/@types/node/-/node-22.13.1.tgz"
integrity sha512-jK8uzQlrvXqEU91UxiK5J7pKHyzgnI1Qnl0QDHIgVGuolJhRb9EEl28Cj9b3rGR8B2lhFCtvIm5os8lFnO/1Ew==
dependencies:
undici-types "~6.20.0"
"@vitejs/plugin-react-swc@^3.7.2":
version "3.7.2"
resolved "https://registry.npmjs.org/@vitejs/plugin-react-swc/-/plugin-react-swc-3.7.2.tgz"
integrity sha512-y0byko2b2tSVVf5Gpng1eEhX1OvPC7x8yns1Fx8jDzlJp4LS6CMkCPfLw47cjyoMrshQDoQw4qcgjsU9VvlCew==
dependencies:
"@swc/core" "^1.7.26"
bowser@^2.8.1:
version "2.11.0"
resolved "https://registry.npmjs.org/bowser/-/bowser-2.11.0.tgz"
integrity sha512-AlcaJBi/pqqJBIQ8U9Mcpc9i8Aqxn88Skv5d+xBX006BY5u8N3mGLHa5Lgppa7L/HfwgwLgZ6NYs+Ag6uUmJRA==
clone-deep@^4.0.1:
version "4.0.1"
resolved "https://registry.npmjs.org/clone-deep/-/clone-deep-4.0.1.tgz"
integrity sha512-neHB9xuzh/wk0dIHweyAXv2aPGZIVk3pLMe+/RNzINf17fe0OG96QroktYAUm7SM1PBnzTabaLboqqxDyMU+SQ==
dependencies:
is-plain-object "^2.0.4"
kind-of "^6.0.2"
shallow-clone "^3.0.0"
dequal@^2.0.3:
version "2.0.3"
resolved "https://registry.npmjs.org/dequal/-/dequal-2.0.3.tgz"
integrity sha512-0je+qPKHEMohvfRTCEo3CrPG6cAzAYgmzKyxRiYSSDkS6eGJdyVJm7WaYA5ECaAD9wLB2T4EEeymA5aFVcYXCA==
esbuild@^0.24.0:
version "0.24.0"
resolved "https://registry.npmjs.org/esbuild/-/esbuild-0.24.0.tgz"
integrity sha512-FuLPevChGDshgSicjisSooU0cemp/sGXR841D5LHMB7mTVOmsEHcAxaH3irL53+8YDIeVNQEySh4DaYU/iuPqQ==
optionalDependencies:
"@esbuild/aix-ppc64" "0.24.0"
"@esbuild/android-arm" "0.24.0"
"@esbuild/android-arm64" "0.24.0"
"@esbuild/android-x64" "0.24.0"
"@esbuild/darwin-arm64" "0.24.0"
"@esbuild/darwin-x64" "0.24.0"
"@esbuild/freebsd-arm64" "0.24.0"
"@esbuild/freebsd-x64" "0.24.0"
"@esbuild/linux-arm" "0.24.0"
"@esbuild/linux-arm64" "0.24.0"
"@esbuild/linux-ia32" "0.24.0"
"@esbuild/linux-loong64" "0.24.0"
"@esbuild/linux-mips64el" "0.24.0"
"@esbuild/linux-ppc64" "0.24.0"
"@esbuild/linux-riscv64" "0.24.0"
"@esbuild/linux-s390x" "0.24.0"
"@esbuild/linux-x64" "0.24.0"
"@esbuild/netbsd-x64" "0.24.0"
"@esbuild/openbsd-arm64" "0.24.0"
"@esbuild/openbsd-x64" "0.24.0"
"@esbuild/sunos-x64" "0.24.0"
"@esbuild/win32-arm64" "0.24.0"
"@esbuild/win32-ia32" "0.24.0"
"@esbuild/win32-x64" "0.24.0"
events@^3.1.0, events@^3.3.0:
version "3.3.0"
resolved "https://registry.npmjs.org/events/-/events-3.3.0.tgz"
integrity sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==
fsevents@~2.3.2, fsevents@~2.3.3:
version "2.3.3"
resolved "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz"
integrity sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==
is-plain-object@^2.0.4:
version "2.0.4"
resolved "https://registry.npmjs.org/is-plain-object/-/is-plain-object-2.0.4.tgz"
integrity sha512-h5PpgXkWitc38BBMYawTYMWJHFZJVnBquFE57xFpjB8pJFiF6gZ+bU+WyI/yqXiFR5mdLsgYNaPe8uao6Uv9Og==
dependencies:
isobject "^3.0.1"
isobject@^3.0.1:
version "3.0.1"
resolved "https://registry.npmjs.org/isobject/-/isobject-3.0.1.tgz"
integrity sha512-WhB9zCku7EGTj/HQQRz5aUQEUeoQZH2bWcltRErOpymJ4boYE6wL9Tbr23krRPSZ+C5zqNSrSw+Cc7sZZ4b7vg==
kind-of@^6.0.2:
version "6.0.3"
resolved "https://registry.npmjs.org/kind-of/-/kind-of-6.0.3.tgz"
integrity sha512-dcS1ul+9tmeD95T+x28/ehLgd9mENa3LsvDTtzm3vyBEO7RPptvAD+t44WVXaUjTBRcrpFeFlC8WCruUR456hw==
nanoid@^3.3.7:
version "3.3.8"
resolved "https://registry.npmjs.org/nanoid/-/nanoid-3.3.8.tgz"
integrity sha512-WNLf5Sd8oZxOm+TzppcYk8gVOgP+l58xNy58D0nbUnOxOWRWvlcCV4kUF7ltmI6PsrLl/BgKEyS4mqsGChFN0w==
picocolors@^1.1.1:
version "1.1.1"
resolved "https://registry.npmjs.org/picocolors/-/picocolors-1.1.1.tgz"
integrity sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA==
postcss@^8.4.49:
version "8.4.49"
resolved "https://registry.npmjs.org/postcss/-/postcss-8.4.49.tgz"
integrity sha512-OCVPnIObs4N29kxTjzLfUryOkvZEq+pf8jTF0lg8E7uETuWHA+v7j3c/xJmiqpX450191LlmZfUKkXxkTry7nA==
dependencies:
nanoid "^3.3.7"
picocolors "^1.1.1"
source-map-js "^1.2.1"
regenerator-runtime@^0.14.0:
version "0.14.1"
resolved "https://registry.npmjs.org/regenerator-runtime/-/regenerator-runtime-0.14.1.tgz"
integrity sha512-dYnhHh0nJoMfnkZs6GmmhFknAGRrLznOu5nc9ML+EJxGvrx6H7teuevqVqCuPcPK//3eDrrjQhehXVx9cnkGdw==
rollup@^4.23.0:
version "4.28.0"
resolved "https://registry.npmjs.org/rollup/-/rollup-4.28.0.tgz"
integrity sha512-G9GOrmgWHBma4YfCcX8PjH0qhXSdH8B4HDE2o4/jaxj93S4DPCIDoLcXz99eWMji4hB29UFCEd7B2gwGJDR9cQ==
dependencies:
"@types/estree" "1.0.6"
optionalDependencies:
"@rollup/rollup-android-arm-eabi" "4.28.0"
"@rollup/rollup-android-arm64" "4.28.0"
"@rollup/rollup-darwin-arm64" "4.28.0"
"@rollup/rollup-darwin-x64" "4.28.0"
"@rollup/rollup-freebsd-arm64" "4.28.0"
"@rollup/rollup-freebsd-x64" "4.28.0"
"@rollup/rollup-linux-arm-gnueabihf" "4.28.0"
"@rollup/rollup-linux-arm-musleabihf" "4.28.0"
"@rollup/rollup-linux-arm64-gnu" "4.28.0"
"@rollup/rollup-linux-arm64-musl" "4.28.0"
"@rollup/rollup-linux-powerpc64le-gnu" "4.28.0"
"@rollup/rollup-linux-riscv64-gnu" "4.28.0"
"@rollup/rollup-linux-s390x-gnu" "4.28.0"
"@rollup/rollup-linux-x64-gnu" "4.28.0"
"@rollup/rollup-linux-x64-musl" "4.28.0"
"@rollup/rollup-win32-arm64-msvc" "4.28.0"
"@rollup/rollup-win32-ia32-msvc" "4.28.0"
"@rollup/rollup-win32-x64-msvc" "4.28.0"
fsevents "~2.3.2"
rxjs@*:
version "7.8.1"
resolved "https://registry.npmjs.org/rxjs/-/rxjs-7.8.1.tgz"
integrity sha512-AA3TVj+0A2iuIoQkWEK/tqFjBq2j+6PO6Y0zJcvzLAFhEFIO3HL0vls9hWLncZbAAbK0mar7oZ4V079I/qPMxg==
dependencies:
tslib "^2.1.0"
shallow-clone@^3.0.0:
version "3.0.1"
resolved "https://registry.npmjs.org/shallow-clone/-/shallow-clone-3.0.1.tgz"
integrity sha512-/6KqX+GVUdqPuPPd2LxDDxzX6CAbjJehAAOKlNpqqUpAqPM6HeL8f+o3a+JsyGjn2lv0WY8UsTgUJjU9Ok55NA==
dependencies:
kind-of "^6.0.2"
source-map-js@^1.2.1:
version "1.2.1"
resolved "https://registry.npmjs.org/source-map-js/-/source-map-js-1.2.1.tgz"
integrity sha512-UXWMKhLOwVKb728IUtQPXxfYU+usdybtUrK/8uGE8CQMvrhOpwvzDBwj0QhSL7MQc7vIsISBG8VQ8+IDQxpfQA==
tslib@^2.1.0:
version "2.8.1"
resolved "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz"
integrity sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==
typed-emitter@^2.1.0:
version "2.1.0"
resolved "https://registry.npmjs.org/typed-emitter/-/typed-emitter-2.1.0.tgz"
integrity sha512-g/KzbYKbH5C2vPkaXGu8DJlHrGKHLsM25Zg9WuC9pMGfuvT+X25tZQWo5fK1BjBm8+UrVE9LDCvaY0CQk+fXDA==
optionalDependencies:
rxjs "*"
typescript@^5.7.3:
version "5.7.3"
resolved "https://registry.npmjs.org/typescript/-/typescript-5.7.3.tgz"
integrity sha512-84MVSjMEHP+FQRPy3pX9sTVV/INIex71s9TL2Gm5FG/WG1SqXeKyZ0k7/blY/4FdOzI12CBy1vGc4og/eus0fw==
undici-types@~6.20.0:
version "6.20.0"
resolved "https://registry.npmjs.org/undici-types/-/undici-types-6.20.0.tgz"
integrity sha512-Ny6QZ2Nju20vw1SRHe3d9jVu6gJ+4e3+MMpqu7pqE5HT6WsTSlce++GQmK5UXS8mzV8DSYHrQH+Xrf2jVcuKNg==
uuid@^10.0.0:
version "10.0.0"
resolved "https://registry.npmjs.org/uuid/-/uuid-10.0.0.tgz"
integrity sha512-8XkAphELsDnEGrDxUOHB3RGvXz6TeuYSGEZBOjtTtPm2lwhGBjLgOzLHB63IUWfBpNucQjND6d3AOudO+H3RWQ==
"vite@^4 || ^5 || ^6", vite@^6.0.2:
version "6.0.2"
resolved "https://registry.npmjs.org/vite/-/vite-6.0.2.tgz"
integrity sha512-XdQ+VsY2tJpBsKGs0wf3U/+azx8BBpYRHFAyKm5VeEZNOJZRB63q7Sc8Iup3k0TrN3KO6QgyzFf+opSbfY1y0g==
dependencies:
esbuild "^0.24.0"
postcss "^8.4.49"
rollup "^4.23.0"
optionalDependencies:
fsevents "~2.3.3"

View File

@@ -1,3 +0,0 @@
DAILY_API_KEY=
GOOGLE_API_KEY=
DAILY_SAMPLE_ROOM_URL=https://yourdomain.daily.co/yourroom # (for joining the bot to the same room repeatedly for local dev)

View File

@@ -1,11 +0,0 @@
[tool.ruff]
exclude = [".git", "*_pb2.py"]
line-length = 100
[tool.ruff.lint]
select = [
"I", # Import rules
]
[tool.ruff.lint.pydocstyle]
convention = "google"

View File

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

View File

@@ -1,200 +0,0 @@
#
# Copyright (c) 2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from contextlib import asynccontextmanager
from typing import Any, Dict, List
import aiohttp
import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
# Load environment variables
load_dotenv(override=True)
NUMBER_OF_ROOMS = 1
class RoomPool:
"""Manages a pool of pre-created rooms for quick allocation."""
def __init__(self, daily_rest_helper: DailyRESTHelper):
self.daily_rest_helper = daily_rest_helper
self.pool: List[Dict[str, str]] = []
self.lock = asyncio.Lock()
async def fill_pool(self, count: int):
"""Fills the pool with `count` new rooms."""
for _ in range(count):
await self.add_room()
async def add_room(self):
"""Creates a new room and adds it to the pool."""
try:
room = await self.daily_rest_helper.create_room(DailyRoomParams())
if not room.url:
raise HTTPException(status_code=500, detail="Failed to create room")
user_token = await self.daily_rest_helper.get_token(room.url)
if not user_token:
raise HTTPException(status_code=500, detail="Failed to get user token")
bot_token = await self.daily_rest_helper.get_token(room.url)
if not bot_token:
raise HTTPException(status_code=500, detail="Failed to get bot token")
async with self.lock:
self.pool.append(
{"room_url": room.url, "user_token": user_token, "bot_token": bot_token}
)
except Exception as e:
print(f"Error adding room to pool: {e}")
async def get_room(self) -> Dict[str, str]:
"""Retrieves a room from the pool and requests a new one to maintain the size."""
async with self.lock:
if not self.pool:
raise HTTPException(status_code=503, detail="No available rooms")
room = self.pool.pop(0) # Get first available room
# Start a background task to replenish the pool
asyncio.create_task(self.add_room())
return room
async def delete_room(self, room_url: str):
"""Deletes a room when it is not needed anymore"""
await self.daily_rest_helper.delete_room_by_url(room_url)
async def cleanup(self):
for rooms in self.pool:
room_url = rooms["room_url"]
await self.delete_room(room_url)
class BotManager:
"""Manages bot subprocesses asynchronously."""
def __init__(self):
self.bot_procs: Dict[int, asyncio.subprocess.Process] = {}
self.room_mappings: Dict[int, str] = {} # Maps process ID to room URL
async def start_bot(self, room_url: str, token: str) -> int:
bot_file = "single_bot"
command = f"python3 -m {bot_file} -u {room_url} -t {token}"
try:
proc = await asyncio.create_subprocess_shell(
command,
cwd=os.path.dirname(os.path.abspath(__file__)),
)
if proc.pid is None:
raise HTTPException(status_code=500, detail="Failed to get subprocess PID")
self.bot_procs[proc.pid] = proc
self.room_mappings[proc.pid] = room_url
# Monitor the process and delete the room when it exits
asyncio.create_task(self._monitor_process(proc.pid))
return proc.pid
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}")
async def _monitor_process(self, pid: int):
"""Monitors a bot process and deletes the associated room when it exits."""
proc = self.bot_procs.get(pid)
if proc:
await proc.wait() # Wait for the process to exit
room_url = self.room_mappings.pop(pid, None)
if room_url:
await room_pool.delete_room(room_url)
print(f"Deleted room: {room_url}")
del self.bot_procs[pid]
async def cleanup(self):
"""Terminates all running bot processes and deletes associated rooms."""
for pid, proc in list(self.bot_procs.items()):
try:
proc.terminate()
await asyncio.wait_for(proc.wait(), timeout=5)
room_url = self.room_mappings.pop(pid, None)
if room_url:
await room_pool.delete_room(room_url) # Delete room when process terminates
print(f"Deleted room: {room_url}")
except asyncio.TimeoutError:
print(f"Process {pid} did not terminate in time.")
except Exception as e:
print(f"Error terminating process {pid}: {e}")
# Clear remaining mappings
self.bot_procs.clear()
self.room_mappings.clear()
# Global instances
bot_manager = BotManager()
room_pool: RoomPool # Will be initialized in lifespan
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Handles FastAPI startup and shutdown."""
global room_pool
aiohttp_session = aiohttp.ClientSession()
daily_rest_helper = DailyRESTHelper(
daily_api_key=os.getenv("DAILY_API_KEY", ""),
daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
aiohttp_session=aiohttp_session,
)
room_pool = RoomPool(daily_rest_helper)
await room_pool.fill_pool(NUMBER_OF_ROOMS) # Fill pool on startup
yield # Run app
await bot_manager.cleanup()
await room_pool.cleanup()
await aiohttp_session.close()
# 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.post("/connect")
async def bot_connect(request: Request) -> Dict[Any, Any]:
try:
room = await room_pool.get_room()
await bot_manager.start_bot(room["room_url"], room["bot_token"])
except HTTPException as e:
return {"error": str(e)}
return {
"room_url": room["room_url"],
"token": room["user_token"],
}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=7860)

View File

@@ -1,120 +0,0 @@
#
# Copyright (c) 2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import asyncio
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, RTVIProcessor
from pipecat.services.gemini_multimodal_live import GeminiMultimodalLiveLLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
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.
"""
def extract_arguments():
parser = argparse.ArgumentParser(description="Instant Voice Example")
parser.add_argument(
"-u", "--url", type=str, required=True, help="URL of the Daily room to join"
)
parser.add_argument(
"-t", "--token", type=str, required=False, help="Token of the Daily room to join"
)
args, unknown = parser.parse_known_args()
url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
token = args.token
return url, token
async def main():
room_url, token = extract_arguments()
print(f"room_url: {room_url}")
daily_transport = DailyTransport(
room_url,
token,
"Instant voice Chatbot",
DailyParams(
audio_out_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
vad_audio_passthrough=True,
),
)
llm = GeminiMultimodalLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
voice_id="Puck", # Aoede, Charon, Fenrir, Kore, Puck
transcribe_user_audio=True,
transcribe_model_audio=True,
system_instruction=SYSTEM_INSTRUCTION,
)
context = OpenAILLMContext()
context_aggregator = llm.create_context_aggregator(context)
# RTVI events for Pipecat client UI
rtvi = RTVIProcessor(config=RTVIConfig(config=[]), transport=daily_transport)
pipeline = Pipeline(
[
daily_transport.input(),
context_aggregator.user(),
rtvi,
llm, # LLM
daily_transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
observers=[rtvi.observer()],
),
)
@rtvi.event_handler("on_client_ready")
async def on_client_ready(rtvi):
await rtvi.set_bot_ready()
@daily_transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
await task.queue_frames([context_aggregator.user().get_context_frame()])
@daily_transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason):
print(f"Participant left: {participant}")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
if __name__ == "__main__":
asyncio.run(main())

View File

@@ -18,7 +18,7 @@ from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.whisper import Model, WhisperSTTService
from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams
from pipecat.transports.local.audio import LocalAudioTransport, LocalTransportParams
load_dotenv(override=True)
@@ -36,7 +36,7 @@ class TranscriptionLogger(FrameProcessor):
async def main(input_device: int, output_device: int):
transport = LocalAudioTransport(
LocalAudioTransportParams(
LocalTransportParams(
audio_in_enabled=True,
audio_out_enabled=False,
input_device_index=input_device,

View File

@@ -1,5 +1,3 @@
<!-- @format -->
<div align="center">
 <img alt="pipecat" width="300px" height="auto" src="image.png">
</div>
@@ -106,21 +104,6 @@ curl -X POST "http://localhost:7860/daily_start_bot" \
-d '{"dialoutNumber": "+18057145330", "detectVoicemail": true}'
```
### New! Using Gemini with Daily
We have introduced a new example file that uses Gemini. You can find the code within bot_daily_gemini.py.
If you want to spin up a Gemini-based bot for this demo, instead of an OpenAI-based bot, call the same properties above but on the `daily_gemini_start_bot` endpoint instead.
For example:
```shell
curl -X POST "http://localhost:7860/daily_gemini_start_bot" \ py pipecat
-H "Content-Type: application/json" \
-d '{"detectVoicemail": true}'
```
Any request body properties supported by `/daily_start_bot` (such as "detectVoicemail", "dialoutnumber", etc) can also be passed to `/daily_gemini_start_bot`. The only difference is that calling the Gemini endpoint will start a Gemini bot session.
### More information
For more configuration options, please consult [Daily's API documentation](https://docs.daily.co).

View File

@@ -98,7 +98,6 @@ async def main(
- **"Record your message after the tone."**
- **Any phrase that suggests an answering machine or voicemail.**
- **ASSUME IT IS A VOICEMAIL. DO NOT WAIT FOR MORE CONFIRMATION.**
- **IF THE CALL SAYS "PLEASE LEAVE A MESSAGE AFTER THE BEEP", WAIT FOR THE BEEP BEFORE LEAVING A MESSAGE.**
#### **Step 2: Leave a Voicemail Message**
- Immediately say:
@@ -111,9 +110,7 @@ async def main(
- If the call is answered by a human, say:
*"Oh, hello! I'm a friendly chatbot. Is there anything I can help you with?"*
- Keep responses **brief and helpful**.
- If the user no longer needs assistance, say:
*"Okay, thank you! Have a great day!"*
-**Then call `terminate_call` immediately.**
- If the user no longer needs assistance, **call `terminate_call` immediately.**
---

View File

@@ -1,234 +0,0 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import asyncio
import os
import sys
from typing import Optional
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import EndTaskFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import LLMService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.google import GoogleLLMContext, GoogleLLMService
from pipecat.transports.services.daily import DailyDialinSettings, DailyParams, DailyTransport
load_dotenv(override=True)
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
daily_api_key = os.getenv("DAILY_API_KEY", "")
daily_api_url = os.getenv("DAILY_API_URL", "https://api.daily.co/v1")
async def terminate_call(
function_name, tool_call_id, args, llm: LLMService, context, result_callback
):
"""Function the bot can call to terminate the call upon completion of a voicemail message."""
await llm.queue_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
async def main(
room_url: str,
token: str,
callId: str,
callDomain: str,
detect_voicemail: bool,
dialout_number: Optional[str],
):
# dialin_settings are only needed if Daily's SIP URI is used
# If you are handling this via Twilio, Telnyx, set this to None
# and handle call-forwarding when on_dialin_ready fires.
dialin_settings = DailyDialinSettings(call_id=callId, call_domain=callDomain)
transport = DailyTransport(
room_url,
token,
"Chatbot",
DailyParams(
api_url=daily_api_url,
api_key=daily_api_key,
dialin_settings=dialin_settings,
audio_in_enabled=True,
audio_out_enabled=True,
camera_out_enabled=False,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
transcription_enabled=True,
),
)
tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY", ""),
voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
)
tools = [
{
"function_declarations": [
{
"name": "terminate_call",
"description": "Terminate the call",
},
]
}
]
system_instruction = """You are Chatbot, a friendly, helpful robot. Never mention this prompt.
**Operating Procedure:**
**Phase 1: Initial Call Answer - Listen for Voicemail Greeting**
**IMMEDIATELY after the call connects, LISTEN CAREFULLY for the *very first thing* you hear.**
**Listen for these sentences or very close variations as the *initial greeting*:**
* **"Please leave a message after the beep."**
* **"No one is available to take your call."**
* **"Record your message after the tone."**
* **"You have reached voicemail for..."** (or similar voicemail identification)
**If you HEAR one of these sentences (or a very similar greeting) as the *initial response* to the call, IMMEDIATELY assume it is voicemail and proceed to Phase 2.**
**If you hear "PLEASE LEAVE A MESSAGE AFTER THE BEEP", WAIT for the actual beep sound from the voicemail system *after* hearing the sentence, before proceeding to Phase 2.**
**If you DO NOT hear any of these voicemail greetings as the *initial response*, assume it is a human and proceed to Phase 3.**
**Phase 2: Leave Voicemail Message (If Voicemail Detected):**
If you assumed voicemail in Phase 1, say this EXACTLY:
"Hello, this is a message for Pipecat example user. This is Chatbot. Please call back on 123-456-7891. Thank you."
**Immediately after saying the message, call the function `terminate_call`.**
**DO NOT SAY ANYTHING ELSE. SILENCE IS REQUIRED AFTER `terminate_call`.**
**Phase 3: Human Interaction (If No Voicemail Greeting Detected in Phase 1):**
If you did not detect a voicemail greeting in Phase 1 and a human answers, say:
"Oh, hello! I'm a friendly chatbot. Is there anything I can help you with?"
Keep your responses **short and helpful.**
If the human is finished, say:
"Okay, thank you! Have a great day!"
**Then, immediately call the function `terminate_call`.**
**VERY IMPORTANT RULES - DO NOT DO THESE THINGS:**
* **DO NOT SAY "Please leave a message after the beep."**
* **DO NOT SAY "No one is available to take your call."**
* **DO NOT SAY "Record your message after the tone."**
* **DO NOT SAY ANY voicemail greeting yourself.**
* **Only check for voicemail greetings in Phase 1, *immediately after the call connects*.**
* **After voicemail or human interaction, ALWAYS call `terminate_call` immediately.**
* **Do not speak after calling `terminate_call`.**
* Your speech will be audio, so use simple language without special characters.
"""
llm = GoogleLLMService(
model="models/gemini-2.0-flash-exp",
api_key=os.getenv("GOOGLE_API_KEY"),
system_instruction=system_instruction,
tools=tools,
)
llm.register_function("terminate_call", terminate_call)
context = GoogleLLMContext()
context_aggregator = llm.create_context_aggregator(context)
pipeline = Pipeline(
[
transport.input(), # Transport user input
context_aggregator.user(), # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
PipelineParams(allow_interruptions=True),
)
if dialout_number:
logger.debug("dialout number detected; doing dialout")
# Configure some handlers for dialing out
@transport.event_handler("on_joined")
async def on_joined(transport, data):
logger.debug(f"Joined; starting dialout to: {dialout_number}")
await transport.start_dialout({"phoneNumber": dialout_number})
@transport.event_handler("on_dialout_connected")
async def on_dialout_connected(transport, data):
logger.debug(f"Dial-out connected: {data}")
@transport.event_handler("on_dialout_answered")
async def on_dialout_answered(transport, data):
logger.debug(f"Dial-out answered: {data}")
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
await transport.capture_participant_transcription(participant["id"])
# unlike the dialin case, for the dialout case, the caller will speak first. Presumably
# they will answer the phone and say "Hello?" Since we've captured their transcript,
# That will put a frame into the pipeline and prompt an LLM completion, which is how the
# bot will then greet the user.
elif detect_voicemail:
logger.debug("Detect voicemail example. You can test this in example in Daily Prebuilt")
# For the voicemail detection case, we do not want the bot to answer the phone. We want it to wait for the voicemail
# machine to say something like 'Leave a message after the beep', or for the user to say 'Hello?'.
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
await transport.capture_participant_transcription(participant["id"])
else:
logger.debug("no dialout number; assuming dialin")
# Different handlers for dialin
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
await transport.capture_participant_transcription(participant["id"])
# For the dialin case, we want the bot to answer the phone and greet the user. We
# can prompt the bot to speak by putting the context into the pipeline.
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason):
await task.cancel()
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Pipecat Simple ChatBot")
parser.add_argument("-u", type=str, help="Room URL")
parser.add_argument("-t", type=str, help="Token")
parser.add_argument("-i", type=str, help="Call ID")
parser.add_argument("-d", type=str, help="Call Domain")
parser.add_argument("-v", action="store_true", help="Detect voicemail")
parser.add_argument("-o", type=str, help="Dialout number", default=None)
config = parser.parse_args()
asyncio.run(main(config.u, config.t, config.i, config.d, config.v, config.o))

View File

@@ -110,15 +110,10 @@ async def _create_daily_room(
# Spawn a new agent, and join the user session
# Note: this is mostly for demonstration purposes (refer to 'deployment' in docs)
print(f"Vendor: {vendor}")
if vendor == "daily":
bot_proc = f"python3 -m bot_daily -u {room.url} -t {token} -i {callId} -d {callDomain}{' -v' if detect_voicemail else ''}"
if dialoutNumber:
bot_proc += f" -o {dialoutNumber}"
elif vendor == "daily-gemini":
bot_proc = f"python3 -m bot_daily_gemini -u {room.url} -t {token} -i {callId} -d {callDomain}{' -v' if detect_voicemail else ''}"
if dialoutNumber:
bot_proc += f" -o {dialoutNumber}"
else:
bot_proc = f"python3 -m bot_twilio -u {room.url} -t {token} -i {callId} -s {room.config.sip_endpoint}"
@@ -206,38 +201,6 @@ async def daily_start_bot(request: Request) -> JSONResponse:
return JSONResponse({"room_url": room.url, "sipUri": room.config.sip_endpoint})
@app.post("/daily_gemini_start_bot")
async def daily_gemini_start_bot(request: Request) -> JSONResponse:
# The /daily_start_bot is invoked when a call is received on Daily's SIP URI
# daily_start_bot will create the room, put the call on hold until
# the bot and sip worker are ready. Daily will automatically
# forward the call to the SIP URi when dialin_ready fires.
# Use specified room URL, or create a new one if not specified
room_url = os.getenv("DAILY_SAMPLE_ROOM_URL", None)
# Get the dial-in properties from the request
try:
data = await request.json()
if "test" in data:
# Pass through any webhook checks
return JSONResponse({"test": True})
detect_voicemail = data.get("detectVoicemail", False)
callId = data.get("callId", None)
callDomain = data.get("callDomain", None)
dialoutNumber = data.get("dialoutNumber", None)
except Exception:
raise HTTPException(
status_code=500, detail="Missing properties 'callId', 'callDomain', or 'dialoutNumber'"
)
room: DailyRoomObject = await _create_daily_room(
room_url, callId, callDomain, dialoutNumber, "daily-gemini", detect_voicemail
)
# Grab a token for the user to join with
return JSONResponse({"room_url": room.url, "sipUri": room.config.sip_endpoint})
# ----------------- Main ----------------- #

View File

@@ -24,15 +24,17 @@ from pipecat.frames.frames import (
)
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIProcessor
from pipecat.processors.transcript_processor import TranscriptProcessor
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.transports.services.daily import (
DailyParams,
DailyTransport,
)
load_dotenv(override=True)
@@ -164,32 +166,21 @@ async def main():
async def on_transcript_update(processor, frame):
await transcript_handler.on_transcript_update(processor, frame)
rtvi = RTVIProcessor()
pipeline = Pipeline(
[
transport.input(),
rtvi,
stt,
transcript.user(), # User transcripts
tp,
llm,
tts,
transport.output(),
transcript.assistant(),
context_aggregator.assistant(),
transcript.assistant(), # Assistant transcripts
]
)
task = PipelineTask(
pipeline,
PipelineParams(
allow_interruptions=False, # We don't want to interrupt the translator bot
enable_metrics=True,
enable_usage_metrics=True,
observers=[RTVIObserver(rtvi)],
),
)
task = PipelineTask(pipeline)
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):

View File

@@ -1,4 +1,4 @@
python-dotenv
fastapi[all]
pipecat-ai[cartesia,daily,deepgram,openai,silero]
pipecat-ai[daily,openai,azure]
aiohttp

View File

@@ -33,8 +33,7 @@ dependencies = [
"pydantic~=2.10.5",
"pyloudnorm~=0.1.1",
"resampy~=0.4.3",
"soxr~=0.5.0",
"tenacity~=9.0.0"
"soxr~=0.5.0"
]
[project.urls]
@@ -56,7 +55,7 @@ elevenlabs = [ "websockets~=13.1" ]
fal = [ "fal-client~=0.5.6" ]
fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ]
gladia = [ "websockets~=13.1" ]
google = [ "google-cloud-speech~=2.31.0", "google-cloud-texttospeech~=2.25.0", "google-genai~=1.2.0", "google-generativeai~=0.8.4" ]
google = [ "google-generativeai~=0.8.3", "google-cloud-texttospeech~=2.24.0", "google-genai~=1.0.0" ]
grok = [ "openai~=1.59.6" ]
groq = [ "openai~=1.59.6" ]
gstreamer = [ "pygobject~=3.50.0" ]
@@ -74,7 +73,6 @@ openai = [ "openai~=1.59.6", "websockets~=13.1", "python-deepcompare~=2.1.0" ]
openpipe = [ "openpipe~=4.45.0" ]
perplexity = [ "openai~=1.59.6" ]
playht = [ "pyht~=0.1.6", "websockets~=13.1" ]
rime = [ "websockets~=13.1" ]
riva = [ "nvidia-riva-client~=2.18.0" ]
sentry = [ "sentry-sdk~=2.20.0" ]
silero = [ "onnxruntime~=1.20.1" ]

View File

@@ -565,22 +565,6 @@ class UserStoppedSpeakingFrame(SystemFrame):
pass
@dataclass
class EmulateUserStartedSpeakingFrame(SystemFrame):
"""Emitted by internal processors upstream to emulate VAD behavior when a
user starts speaking."""
pass
@dataclass
class EmulateUserStoppedSpeakingFrame(SystemFrame):
"""Emitted by internal processors upstream to emulate VAD behavior when a
user stops speaking."""
pass
@dataclass
class BotInterruptionFrame(SystemFrame):
"""Emitted by when the bot should be interrupted. This will mainly cause the
@@ -634,13 +618,6 @@ class FunctionCallInProgressFrame(SystemFrame):
arguments: str
@dataclass
class STTMuteFrame(SystemFrame):
"""System frame to mute/unmute the STT service."""
mute: bool
@dataclass
class TransportMessageUrgentFrame(SystemFrame):
message: Any
@@ -775,6 +752,13 @@ class TTSUpdateSettingsFrame(ServiceUpdateSettingsFrame):
pass
@dataclass
class STTMuteFrame(ControlFrame):
"""Control frame to mute/unmute the STT service."""
mute: bool
@dataclass
class STTUpdateSettingsFrame(ServiceUpdateSettingsFrame):
pass

View File

@@ -1,85 +0,0 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from loguru import logger
from pipecat.frames.frames import (
Frame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
LLMTextFrame,
)
from pipecat.observers.base_observer import BaseObserver
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.ai_services import LLMService
class LLMLogObserver(BaseObserver):
"""Observer to log LLM activity to the console.
Logs all frame instances (only from/to LLM service) of:
- LLMFullResponseStartFrame
- LLMFullResponseEndFrame
- LLMTextFrame
- FunctionCallInProgressFrame
- LLMMessagesFrame
- OpenAILLMContextFrame
This allows you to track when the LLM starts responding, what it generates,
and when it finishes.
"""
async def on_push_frame(
self,
src: FrameProcessor,
dst: FrameProcessor,
frame: Frame,
direction: FrameDirection,
timestamp: int,
):
if not isinstance(src, LLMService) and not isinstance(dst, LLMService):
return
time_sec = timestamp / 1_000_000_000
arrow = ""
# Log LLM start/end frames (output)
if isinstance(frame, (LLMFullResponseStartFrame, LLMFullResponseEndFrame)):
event = "START" if isinstance(frame, LLMFullResponseStartFrame) else "END"
logger.debug(f"🧠 {src} {arrow} LLM {event} RESPONSE at {time_sec:.2f}s")
# Log all LLMTextFrames (output)
elif isinstance(frame, LLMTextFrame):
logger.debug(f"🧠 {src} {arrow} LLM GENERATING: {frame.text!r} at {time_sec:.2f}s")
# Log function calling (output)
elif (
isinstance(frame, FunctionCallInProgressFrame)
and direction != FrameDirection.DOWNSTREAM
):
logger.debug(
f"🧠 {src} {arrow} LLM FUNCTION CALL ({frame.tool_call_id}): {frame.function_name!r}({frame.arguments}) at {time_sec:.2f}s"
)
# Log LLMMessagesFrame (input)
elif isinstance(frame, LLMMessagesFrame):
logger.debug(
f"🧠 {arrow} {dst} LLM MESSAGES FRAME: {frame.messages} at {time_sec:.2f}s"
)
# Log OpenAILLMContextFrame (input)
elif isinstance(frame, OpenAILLMContextFrame):
logger.debug(
f"🧠 {arrow} {dst} LLM CONTEXT FRAME: {frame.context.messages} at {time_sec:.2f}s"
)
# Log function call result (input)
elif isinstance(frame, FunctionCallResultFrame):
logger.debug(
f"🧠 {arrow} {src} LLM FUNCTION CALL RESULT ({frame.tool_call_id}): {frame.result} at {time_sec:.2f}s"
)

View File

@@ -1,54 +0,0 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from loguru import logger
from pipecat.frames.frames import (
Frame,
InterimTranscriptionFrame,
TranscriptionFrame,
)
from pipecat.observers.base_observer import BaseObserver
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.ai_services import STTService
class TranscriptionLogObserver(BaseObserver):
"""Observer to log transcription activity to the console.
Logs all frame instances (only from STT service) of:
- TranscriptionFrame
- InterimTranscriptionFrame
This allows you to track when the LLM starts responding, what it generates,
and when it finishes.
"""
async def on_push_frame(
self,
src: FrameProcessor,
dst: FrameProcessor,
frame: Frame,
direction: FrameDirection,
timestamp: int,
):
if not isinstance(src, STTService):
return
time_sec = timestamp / 1_000_000_000
arrow = ""
if isinstance(frame, TranscriptionFrame):
logger.debug(
f"💬 {src} {arrow} TRANSCRIPTION: {frame.text!r} from {frame.user_id!r} at {time_sec:.2f}s"
)
elif isinstance(frame, InterimTranscriptionFrame):
logger.debug(
f"💬 {src} {arrow} INTERIM TRANSCRIPTION: {frame.text!r} from {frame.user_id!r} at {time_sec:.2f}s"
)

View File

@@ -4,16 +4,9 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import time
from abc import abstractmethod
from typing import List
from typing import List, Optional, Type
from pipecat.frames.frames import (
CancelFrame,
EmulateUserStartedSpeakingFrame,
EmulateUserStoppedSpeakingFrame,
EndFrame,
Frame,
InterimTranscriptionFrame,
LLMFullResponseEndFrame,
@@ -22,7 +15,6 @@ from pipecat.frames.frames import (
LLMMessagesFrame,
LLMMessagesUpdateFrame,
LLMSetToolsFrame,
StartFrame,
StartInterruptionFrame,
TextFrame,
TranscriptionFrame,
@@ -36,105 +28,121 @@ from pipecat.processors.aggregators.openai_llm_context import (
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
class BaseLLMResponseAggregator(FrameProcessor):
"""This is the base class for all LLM response aggregators. These
aggregators process incoming frames and aggregate content until they are
ready to push the aggregation. In the case of a user, an aggregation might
be a full transcription received from the STT service.
The LLM response aggregators also keep a store (e.g. a message list or an
LLM context) of the current conversation, that is, it stores the messages
said by the user or by the bot.
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
@property
@abstractmethod
def messages(self) -> List[dict]:
"""Returns the messages from the current conversation."""
pass
@property
@abstractmethod
def role(self) -> str:
"""Returns the role (e.g. user, assistant...) for this aggregator."""
pass
@abstractmethod
def add_messages(self, messages):
"""Add the given messages to the conversation."""
pass
@abstractmethod
def set_messages(self, messages):
"""Reset the conversation with the given messages."""
pass
@abstractmethod
def set_tools(self, tools):
"""Set LLM tools to be used in the current conversation."""
pass
@abstractmethod
def reset(self):
"""Reset the internals of this aggregator. This should not modify the
internal messages."""
pass
@abstractmethod
async def push_aggregation(self):
pass
class LLMResponseAggregator(BaseLLMResponseAggregator):
"""This is a base LLM aggregator that uses a simple list of messages to
store the conversation. It pushes `LLMMessagesFrame` as an aggregation
frame.
"""
class LLMResponseAggregator(FrameProcessor):
def __init__(
self,
*,
messages: List[dict],
role: str = "user",
**kwargs,
role: str,
start_frame,
end_frame,
accumulator_frame: Type[TextFrame],
interim_accumulator_frame: Optional[Type[TextFrame]] = None,
handle_interruptions: bool = False,
expect_stripped_words: bool = True, # if True, need to add spaces between words
):
super().__init__(**kwargs)
super().__init__()
self._messages = messages
self._role = role
self._start_frame = start_frame
self._end_frame = end_frame
self._accumulator_frame = accumulator_frame
self._interim_accumulator_frame = interim_accumulator_frame
self._handle_interruptions = handle_interruptions
self._expect_stripped_words = expect_stripped_words
self._aggregation = ""
self.reset()
# Reset our accumulator state.
self._reset()
@property
def messages(self) -> List[dict]:
def messages(self):
return self._messages
@property
def role(self) -> str:
def role(self):
return self._role
def add_messages(self, messages):
self._messages.extend(messages)
#
# Frame processor
#
def set_messages(self, messages):
self.reset()
self._messages.clear()
self._messages.extend(messages)
# Use cases implemented:
#
# S: Start, E: End, T: Transcription, I: Interim, X: Text
#
# S E -> None
# S T E -> X
# S I T E -> X
# S I E T -> X
# S I E I T -> X
# S E T -> X
# S E I T -> X
#
# The following case would not be supported:
#
# S I E T1 I T2 -> X
#
# and T2 would be dropped.
def set_tools(self, tools):
pass
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
def reset(self):
self._aggregation = ""
send_aggregation = False
async def push_aggregation(self):
if isinstance(frame, self._start_frame):
self._aggregation = ""
self._aggregating = True
self._seen_start_frame = True
self._seen_end_frame = False
self._seen_interim_results = False
await self.push_frame(frame, direction)
elif isinstance(frame, self._end_frame):
self._seen_end_frame = True
self._seen_start_frame = False
# We might have received the end frame but we might still be
# aggregating (i.e. we have seen interim results but not the final
# text).
self._aggregating = self._seen_interim_results or len(self._aggregation) == 0
# Send the aggregation if we are not aggregating anymore (i.e. no
# more interim results received).
send_aggregation = not self._aggregating
await self.push_frame(frame, direction)
elif isinstance(frame, self._accumulator_frame):
if self._aggregating:
if self._expect_stripped_words:
self._aggregation += f" {frame.text}" if self._aggregation else frame.text
else:
self._aggregation += frame.text
# We have recevied a complete sentence, so if we have seen the
# end frame and we were still aggregating, it means we should
# send the aggregation.
send_aggregation = self._seen_end_frame
# We just got our final result, so let's reset interim results.
self._seen_interim_results = False
elif self._interim_accumulator_frame and isinstance(frame, self._interim_accumulator_frame):
self._seen_interim_results = True
elif self._handle_interruptions and isinstance(frame, StartInterruptionFrame):
await self._push_aggregation()
# Reset anyways
self._reset()
await self.push_frame(frame, direction)
elif isinstance(frame, LLMMessagesAppendFrame):
self._add_messages(frame.messages)
elif isinstance(frame, LLMMessagesUpdateFrame):
self._set_messages(frame.messages)
elif isinstance(frame, LLMSetToolsFrame):
self._set_tools(frame.tools)
else:
await self.push_frame(frame, direction)
if send_aggregation:
await self._push_aggregation()
async def _push_aggregation(self):
if len(self._aggregation) > 0:
self._messages.append({"role": self._role, "content": self._aggregation})
@@ -145,30 +153,109 @@ class LLMResponseAggregator(BaseLLMResponseAggregator):
frame = LLMMessagesFrame(self._messages)
await self.push_frame(frame)
# Reset our accumulator state.
self.reset()
# TODO-CB: Types
def _add_messages(self, messages):
self._messages.extend(messages)
def _set_messages(self, messages):
self._reset()
self._messages.clear()
self._messages.extend(messages)
def _set_tools(self, tools):
# noop in the base class
pass
def _reset(self):
self._aggregation = ""
self._aggregating = False
self._seen_start_frame = False
self._seen_end_frame = False
self._seen_interim_results = False
class LLMContextResponseAggregator(BaseLLMResponseAggregator):
"""This is a base LLM aggregator that uses an LLM context to store the
conversation. It pushes `OpenAILLMContextFrame` as an aggregation frame.
class LLMAssistantResponseAggregator(LLMResponseAggregator):
def __init__(self, messages: List[dict] = []):
super().__init__(
messages=messages,
role="assistant",
start_frame=LLMFullResponseStartFrame,
end_frame=LLMFullResponseEndFrame,
accumulator_frame=TextFrame,
handle_interruptions=True,
)
class LLMUserResponseAggregator(LLMResponseAggregator):
def __init__(self, messages: List[dict] = []):
super().__init__(
messages=messages,
role="user",
start_frame=UserStartedSpeakingFrame,
end_frame=UserStoppedSpeakingFrame,
accumulator_frame=TranscriptionFrame,
interim_accumulator_frame=InterimTranscriptionFrame,
)
class LLMFullResponseAggregator(FrameProcessor):
"""This class aggregates Text frames until it receives a
LLMFullResponseEndFrame, then emits the concatenated text as
a single text frame.
given the following frames:
TextFrame("Hello,")
TextFrame(" world.")
TextFrame(" I am")
TextFrame(" an LLM.")
LLMFullResponseEndFrame()]
this processor will yield nothing for the first 4 frames, then
TextFrame("Hello, world. I am an LLM.")
LLMFullResponseEndFrame()
when passed the last frame.
>>> async def print_frames(aggregator, frame):
... async for frame in aggregator.process_frame(frame):
... if isinstance(frame, TextFrame):
... print(frame.text)
... else:
... print(frame.__class__.__name__)
>>> aggregator = LLMFullResponseAggregator()
>>> asyncio.run(print_frames(aggregator, TextFrame("Hello,")))
>>> asyncio.run(print_frames(aggregator, TextFrame(" world.")))
>>> asyncio.run(print_frames(aggregator, TextFrame(" I am")))
>>> asyncio.run(print_frames(aggregator, TextFrame(" an LLM.")))
>>> asyncio.run(print_frames(aggregator, LLMFullResponseEndFrame()))
Hello, world. I am an LLM.
LLMFullResponseEndFrame
"""
def __init__(self, *, context: OpenAILLMContext, role: str, **kwargs):
super().__init__(**kwargs)
self._context = context
self._role = role
def __init__(self):
super().__init__()
self._aggregation = ""
@property
def messages(self) -> List[dict]:
return self._context.get_messages()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@property
def role(self) -> str:
return self._role
if isinstance(frame, TextFrame):
self._aggregation += frame.text
elif isinstance(frame, LLMFullResponseEndFrame):
await self.push_frame(TextFrame(self._aggregation))
await self.push_frame(frame)
self._aggregation = ""
else:
await self.push_frame(frame, direction)
class LLMContextAggregator(LLMResponseAggregator):
def __init__(self, *, context: OpenAILLMContext, **kwargs):
super().__init__(**kwargs)
self._context = context
@property
def context(self):
@@ -177,25 +264,23 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator):
def get_context_frame(self) -> OpenAILLMContextFrame:
return OpenAILLMContextFrame(context=self._context)
async def push_context_frame(self, direction: FrameDirection = FrameDirection.DOWNSTREAM):
async def push_context_frame(self):
frame = self.get_context_frame()
await self.push_frame(frame, direction)
await self.push_frame(frame)
def add_messages(self, messages):
# TODO-CB: Types
def _add_messages(self, messages):
self._context.add_messages(messages)
def set_messages(self, messages):
def _set_messages(self, messages):
self._context.set_messages(messages)
def set_tools(self, tools: List):
def _set_tools(self, tools: List):
self._context.set_tools(tools)
def reset(self):
self._aggregation = ""
async def push_aggregation(self):
async def _push_aggregation(self):
if len(self._aggregation) > 0:
self._context.add_message({"role": self.role, "content": self._aggregation})
self._context.add_message({"role": self._role, "content": self._aggregation})
# Reset the aggregation. Reset it before pushing it down, otherwise
# if the tasks gets cancelled we won't be able to clear things up.
@@ -205,243 +290,31 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator):
await self.push_frame(frame)
# Reset our accumulator state.
self.reset()
self._reset()
class LLMUserContextAggregator(LLMContextResponseAggregator):
"""This is a user LLM aggregator that uses an LLM context to store the
conversation. It aggregates transcriptions from the STT service and it has
logic to handle multiple scenarios where transcriptions are received between
VAD events (`UserStartedSpeakingFrame` and `UserStoppedSpeakingFrame`) or
even outside or no VAD events at all.
"""
def __init__(
self,
context: OpenAILLMContext,
aggregation_timeout: float = 1.0,
bot_interruption_timeout: float = 2.0,
**kwargs,
):
super().__init__(context=context, role="user", **kwargs)
self._aggregation_timeout = aggregation_timeout
self._bot_interruption_timeout = bot_interruption_timeout
self._seen_interim_results = False
self._user_speaking = False
self._last_user_speaking_time = 0
self._emulating_vad = False
self._aggregation_event = asyncio.Event()
self._aggregation_task = None
self.reset()
def reset(self):
super().reset()
self._seen_interim_results = False
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, StartFrame):
await self._start(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, EndFrame):
await self._stop(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, CancelFrame):
await self._cancel(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, UserStartedSpeakingFrame):
await self._handle_user_started_speaking(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, UserStoppedSpeakingFrame):
await self._handle_user_stopped_speaking(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, TranscriptionFrame):
await self._handle_transcription(frame)
elif isinstance(frame, InterimTranscriptionFrame):
await self._handle_interim_transcription(frame)
elif isinstance(frame, LLMMessagesAppendFrame):
self.add_messages(frame.messages)
elif isinstance(frame, LLMMessagesUpdateFrame):
self.set_messages(frame.messages)
elif isinstance(frame, LLMSetToolsFrame):
self.set_tools(frame.tools)
else:
await self.push_frame(frame, direction)
async def _start(self, frame: StartFrame):
self._create_aggregation_task()
async def _stop(self, frame: EndFrame):
await self._cancel_aggregation_task()
async def _cancel(self, frame: CancelFrame):
await self._cancel_aggregation_task()
async def _handle_user_started_speaking(self, _: UserStartedSpeakingFrame):
self._last_user_speaking_time = time.time()
self._user_speaking = True
async def _handle_user_stopped_speaking(self, _: UserStoppedSpeakingFrame):
self._last_user_speaking_time = time.time()
self._user_speaking = False
if not self._seen_interim_results:
await self.push_aggregation()
async def _handle_transcription(self, frame: TranscriptionFrame):
text = frame.text
# Make sure we really have some text.
if not text.strip():
return
self._aggregation += f" {text}" if self._aggregation else text
# We just got a final result, so let's reset interim results.
self._seen_interim_results = False
# Reset aggregation timer.
self._aggregation_event.set()
async def _handle_interim_transcription(self, _: InterimTranscriptionFrame):
self._seen_interim_results = True
def _create_aggregation_task(self):
self._aggregation_task = self.create_task(self._aggregation_task_handler())
async def _cancel_aggregation_task(self):
if self._aggregation_task:
await self.cancel_task(self._aggregation_task)
self._aggregation_task = None
async def _aggregation_task_handler(self):
while True:
try:
await asyncio.wait_for(self._aggregation_event.wait(), self._aggregation_timeout)
await self._maybe_push_bot_interruption()
except asyncio.TimeoutError:
if not self._user_speaking:
await self.push_aggregation()
# If we are emulating VAD we still need to send the user stopped
# speaking frame.
if self._emulating_vad:
await self.push_frame(
EmulateUserStoppedSpeakingFrame(), FrameDirection.UPSTREAM
)
self._emulating_vad = False
finally:
self._aggregation_event.clear()
async def _maybe_push_bot_interruption(self):
"""If the user stopped speaking a while back and we got a transcription
frame we might want to interrupt the bot.
"""
if not self._user_speaking:
diff_time = time.time() - self._last_user_speaking_time
if diff_time > self._bot_interruption_timeout:
# If we reach this case we received a transcription but VAD was
# not able to detect voice (e.g. when you whisper a short
# utterance). So, we need to emulate VAD (i.e. user
# start/stopped speaking).
await self.push_frame(EmulateUserStartedSpeakingFrame(), FrameDirection.UPSTREAM)
self._emulating_vad = True
# Reset time so we don't interrupt again right away.
self._last_user_speaking_time = time.time()
class LLMAssistantContextAggregator(LLMContextAggregator):
def __init__(self, context: OpenAILLMContext, *, expect_stripped_words: bool = True):
super().__init__(
messages=[],
context=context,
role="assistant",
start_frame=LLMFullResponseStartFrame,
end_frame=LLMFullResponseEndFrame,
accumulator_frame=TextFrame,
handle_interruptions=True,
expect_stripped_words=expect_stripped_words,
)
class LLMAssistantContextAggregator(LLMContextResponseAggregator):
"""This is an assistant LLM aggregator that uses an LLM context to store the
conversation. It aggregates text frames received between
`LLMFullResponseStartFrame` and `LLMFullResponseEndFrame`.
"""
def __init__(self, context: OpenAILLMContext, *, expect_stripped_words: bool = True, **kwargs):
super().__init__(context=context, role="assistant", **kwargs)
self._expect_stripped_words = expect_stripped_words
self._started = False
self.reset()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, StartInterruptionFrame):
await self.push_aggregation()
# Reset anyways
self.reset()
await self.push_frame(frame, direction)
elif isinstance(frame, LLMFullResponseStartFrame):
await self._handle_llm_start(frame)
elif isinstance(frame, LLMFullResponseEndFrame):
await self._handle_llm_end(frame)
elif isinstance(frame, TextFrame):
await self._handle_text(frame)
elif isinstance(frame, LLMMessagesAppendFrame):
self.add_messages(frame.messages)
elif isinstance(frame, LLMMessagesUpdateFrame):
self.set_messages(frame.messages)
elif isinstance(frame, LLMSetToolsFrame):
self.set_tools(frame.tools)
else:
await self.push_frame(frame, direction)
async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
self._started = True
async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
self._started = False
await self.push_aggregation()
async def _handle_text(self, frame: TextFrame):
if not self._started:
return
if self._expect_stripped_words:
self._aggregation += f" {frame.text}" if self._aggregation else frame.text
else:
self._aggregation += frame.text
class LLMUserResponseAggregator(LLMUserContextAggregator):
def __init__(self, messages: List[dict] = [], **kwargs):
super().__init__(context=OpenAILLMContext(messages), **kwargs)
async def push_aggregation(self):
if len(self._aggregation) > 0:
self._context.add_message({"role": self.role, "content": self._aggregation})
# Reset the aggregation. Reset it before pushing it down, otherwise
# if the tasks gets cancelled we won't be able to clear things up.
self._aggregation = ""
frame = LLMMessagesFrame(self._context.messages)
await self.push_frame(frame)
# Reset our accumulator state.
self.reset()
class LLMAssistantResponseAggregator(LLMAssistantContextAggregator):
def __init__(self, messages: List[dict] = [], **kwargs):
super().__init__(context=OpenAILLMContext(messages), **kwargs)
async def push_aggregation(self):
if len(self._aggregation) > 0:
self._context.add_message({"role": self.role, "content": self._aggregation})
# Reset the aggregation. Reset it before pushing it down, otherwise
# if the tasks gets cancelled we won't be able to clear things up.
self._aggregation = ""
frame = LLMMessagesFrame(self._context.messages)
await self.push_frame(frame)
# Reset our accumulator state.
self.reset()
class LLMUserContextAggregator(LLMContextAggregator):
def __init__(self, context: OpenAILLMContext):
super().__init__(
messages=[],
context=context,
role="user",
start_frame=UserStartedSpeakingFrame,
end_frame=UserStoppedSpeakingFrame,
accumulator_frame=TranscriptionFrame,
interim_accumulator_frame=InterimTranscriptionFrame,
)

View File

@@ -4,15 +4,131 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from pipecat.frames.frames import TextFrame
from pipecat.processors.aggregators.llm_response import LLMUserResponseAggregator
from typing import Optional
from pipecat.frames.frames import (
Frame,
InterimTranscriptionFrame,
StartInterruptionFrame,
TextFrame,
TranscriptionFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
class UserResponseAggregator(LLMUserResponseAggregator):
def __init__(self, **kwargs):
super().__init__(**kwargs)
class ResponseAggregator(FrameProcessor):
"""This frame processor aggregates frames between a start and an end frame
into complete text frame sentences.
async def push_aggregation(self):
For example, frame input/output:
UserStartedSpeakingFrame() -> None
TranscriptionFrame("Hello,") -> None
TranscriptionFrame(" world.") -> None
UserStoppedSpeakingFrame() -> TextFrame("Hello world.")
Doctest: FIXME to work with asyncio
>>> async def print_frames(aggregator, frame):
... async for frame in aggregator.process_frame(frame):
... if isinstance(frame, TextFrame):
... print(frame.text)
>>> aggregator = ResponseAggregator(start_frame = UserStartedSpeakingFrame,
... end_frame=UserStoppedSpeakingFrame,
... accumulator_frame=TranscriptionFrame,
... pass_through=False)
>>> asyncio.run(print_frames(aggregator, UserStartedSpeakingFrame()))
>>> asyncio.run(print_frames(aggregator, TranscriptionFrame("Hello,", 1, 1)))
>>> asyncio.run(print_frames(aggregator, TranscriptionFrame("world.", 1, 2)))
>>> asyncio.run(print_frames(aggregator, UserStoppedSpeakingFrame()))
Hello, world.
"""
def __init__(
self,
*,
start_frame,
end_frame,
accumulator_frame: TextFrame,
interim_accumulator_frame: Optional[TextFrame] = None,
):
super().__init__()
self._start_frame = start_frame
self._end_frame = end_frame
self._accumulator_frame = accumulator_frame
self._interim_accumulator_frame = interim_accumulator_frame
# Reset our accumulator state.
self._reset()
#
# Frame processor
#
# Use cases implemented:
#
# S: Start, E: End, T: Transcription, I: Interim, X: Text
#
# S E -> None
# S T E -> X
# S I T E -> X
# S I E T -> X
# S I E I T -> X
# S E T -> X
# S E I T -> X
#
# The following case would not be supported:
#
# S I E T1 I T2 -> X
#
# and T2 would be dropped.
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
send_aggregation = False
if isinstance(frame, self._start_frame):
self._aggregating = True
self._seen_start_frame = True
self._seen_end_frame = False
self._seen_interim_results = False
await self.push_frame(frame, direction)
elif isinstance(frame, self._end_frame):
self._seen_end_frame = True
self._seen_start_frame = False
# We might have received the end frame but we might still be
# aggregating (i.e. we have seen interim results but not the final
# text).
self._aggregating = self._seen_interim_results or len(self._aggregation) == 0
# Send the aggregation if we are not aggregating anymore (i.e. no
# more interim results received).
send_aggregation = not self._aggregating
await self.push_frame(frame, direction)
elif isinstance(frame, self._accumulator_frame):
if self._aggregating:
self._aggregation += f" {frame.text}"
# We have recevied a complete sentence, so if we have seen the
# end frame and we were still aggregating, it means we should
# send the aggregation.
send_aggregation = self._seen_end_frame
# We just got our final result, so let's reset interim results.
self._seen_interim_results = False
elif self._interim_accumulator_frame and isinstance(frame, self._interim_accumulator_frame):
self._seen_interim_results = True
else:
await self.push_frame(frame, direction)
if send_aggregation:
await self._push_aggregation()
async def _push_aggregation(self):
if len(self._aggregation) > 0:
frame = TextFrame(self._aggregation.strip())
@@ -23,4 +139,21 @@ class UserResponseAggregator(LLMUserResponseAggregator):
await self.push_frame(frame)
# Reset our accumulator state.
self.reset()
self._reset()
def _reset(self):
self._aggregation = ""
self._aggregating = False
self._seen_start_frame = False
self._seen_end_frame = False
self._seen_interim_results = False
class UserResponseAggregator(ResponseAggregator):
def __init__(self):
super().__init__(
start_frame=UserStartedSpeakingFrame,
end_frame=UserStoppedSpeakingFrame,
accumulator_frame=TranscriptionFrame,
interim_accumulator_frame=InterimTranscriptionFrame,
)

View File

@@ -23,8 +23,6 @@ from pipecat.frames.frames import (
Frame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
InputAudioRawFrame,
StartFrame,
StartInterruptionFrame,
StopInterruptionFrame,
STTMuteFrame,
@@ -39,18 +37,16 @@ class STTMuteStrategy(Enum):
"""Strategies determining when STT should be muted.
Attributes:
FIRST_SPEECH: Mute only during first detected bot speech
MUTE_UNTIL_FIRST_BOT_COMPLETE: Start muted and remain muted until first bot speech completes
FIRST_SPEECH: Mute only during first bot speech
FUNCTION_CALL: Mute during function calls
ALWAYS: Mute during all bot speech
CUSTOM: Allow custom logic via callback
"""
FIRST_SPEECH = "first_speech"
MUTE_UNTIL_FIRST_BOT_COMPLETE = "mute_until_first_bot_complete"
FUNCTION_CALL = "function_call"
ALWAYS = "always"
CUSTOM = "custom"
FIRST_SPEECH = "first_speech" # Mute only during first bot speech
FUNCTION_CALL = "function_call" # Mute during function calls
ALWAYS = "always" # Mute during all bot speech
CUSTOM = "custom" # Allow custom logic via callback
@dataclass
@@ -61,25 +57,12 @@ class STTMuteConfig:
strategies: Set of muting strategies to apply
should_mute_callback: Optional callback for custom muting logic.
Only required when using STTMuteStrategy.CUSTOM
Note:
MUTE_UNTIL_FIRST_BOT_COMPLETE and FIRST_SPEECH strategies should not be used together
as they handle the first bot speech differently.
"""
strategies: set[STTMuteStrategy]
# Optional callback for custom muting logic
should_mute_callback: Optional[Callable[["STTMuteFilter"], Awaitable[bool]]] = None
def __post_init__(self):
if (
STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE in self.strategies
and STTMuteStrategy.FIRST_SPEECH in self.strategies
):
raise ValueError(
"MUTE_UNTIL_FIRST_BOT_COMPLETE and FIRST_SPEECH strategies should not be used together"
)
class STTMuteFilter(FrameProcessor):
"""A processor that handles STT muting and interruption control.
@@ -110,7 +93,7 @@ class STTMuteFilter(FrameProcessor):
self._first_speech_handled = False
self._bot_is_speaking = False
self._function_call_in_progress = False
self._is_muted = False # Initialize as unmuted, will set state on StartFrame if needed
self._is_muted = False
@property
def is_muted(self) -> bool:
@@ -141,10 +124,6 @@ class STTMuteFilter(FrameProcessor):
self._first_speech_handled = True
return True
case STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE:
if not self._first_speech_handled:
return True
case STTMuteStrategy.CUSTOM:
if self._bot_is_speaking and self._config.should_mute_callback:
should_mute = await self._config.should_mute_callback(self)
@@ -154,31 +133,25 @@ class STTMuteFilter(FrameProcessor):
return False
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Processes incoming frames and manages muting state."""
await super().process_frame(frame, direction)
# Determine if we need to change mute state based on frame type
should_mute = None
# Process frames to determine mute state
if isinstance(frame, StartFrame):
should_mute = await self._should_mute()
elif isinstance(frame, FunctionCallInProgressFrame):
"""Processes incoming frames and manages muting state."""
# Handle function call state changes
if isinstance(frame, FunctionCallInProgressFrame):
self._function_call_in_progress = True
should_mute = await self._should_mute()
await self._handle_mute_state(await self._should_mute())
elif isinstance(frame, FunctionCallResultFrame):
self._function_call_in_progress = False
should_mute = await self._should_mute()
await self._handle_mute_state(await self._should_mute())
# Handle bot speaking state changes
elif isinstance(frame, BotStartedSpeakingFrame):
self._bot_is_speaking = True
should_mute = await self._should_mute()
await self._handle_mute_state(await self._should_mute())
elif isinstance(frame, BotStoppedSpeakingFrame):
self._bot_is_speaking = False
if not self._first_speech_handled:
self._first_speech_handled = True
should_mute = await self._should_mute()
await self._handle_mute_state(await self._should_mute())
# Then push the original frame
# Handle frame propagation
if isinstance(
frame,
(
@@ -186,18 +159,13 @@ class STTMuteFilter(FrameProcessor):
StopInterruptionFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
InputAudioRawFrame,
),
):
# Only pass VAD-related frames when not muted
if not self.is_muted:
await self.push_frame(frame, direction)
else:
logger.trace(f"{frame.__class__.__name__} suppressed - STT currently muted")
logger.debug(f"{frame.__class__.__name__} suppressed - STT currently muted")
else:
# Pass all other frames through
await self.push_frame(frame, direction)
# Finally handle mute state change if needed
if should_mute is not None and should_mute != self.is_muted:
await self._handle_mute_state(should_mute)

View File

@@ -73,11 +73,10 @@ class FrameProcessor:
self._metrics.set_processor_name(self.name)
# Processors have an input queue. The input queue will be processed
# immediately (default) or it will block if `pause_processing_frames()`
# is called. To resume processing frames we need to call
# `resume_processing_frames()` which will wake up the event.
# immediately (default) or it will block if `pause_processing_frames()` is
# called. To resume processing frames we need to call
# `resume_processing_frames()`.
self.__should_block_frames = False
self.__input_event = asyncio.Event()
self.__input_frame_task: Optional[asyncio.Task] = None
# Every processor in Pipecat should only output frames from a single
@@ -336,8 +335,8 @@ class FrameProcessor:
def __create_input_task(self):
if not self.__input_frame_task:
self.__should_block_frames = False
self.__input_event.clear()
self.__input_queue = asyncio.Queue()
self.__input_event = asyncio.Event()
self.__input_frame_task = self.create_task(self.__input_frame_task_handler())
async def __cancel_input_task(self):

View File

@@ -5,7 +5,6 @@
#
import asyncio
import base64
from dataclasses import dataclass
from typing import (
Any,
@@ -32,7 +31,6 @@ from pipecat.frames.frames import (
ErrorFrame,
Frame,
FunctionCallResultFrame,
InputAudioRawFrame,
InterimTranscriptionFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
@@ -60,9 +58,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContextFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.transports.base_input import BaseInputTransport
from pipecat.transports.base_output import BaseOutputTransport
from pipecat.transports.base_transport import BaseTransport
from pipecat.utils.string import match_endofsentence
RTVI_PROTOCOL_VERSION = "0.3.0"
@@ -823,7 +819,6 @@ class RTVIProcessor(FrameProcessor):
self,
*,
config: RTVIConfig = RTVIConfig(config=[]),
transport: Optional[BaseTransport] = None,
**kwargs,
):
super().__init__(**kwargs)
@@ -849,14 +844,6 @@ class RTVIProcessor(FrameProcessor):
self._register_event_handler("on_bot_started")
self._register_event_handler("on_client_ready")
self._input_transport = None
self._transport = transport
if self._transport:
input_transport = self._transport.input()
if isinstance(input_transport, BaseInputTransport):
self._input_transport = input_transport
self._input_transport.enable_audio_in_stream_on_start(False)
def observer(self) -> RTVIObserver:
import warnings
@@ -1026,8 +1013,6 @@ class RTVIProcessor(FrameProcessor):
case "llm-function-call-result":
data = RTVILLMFunctionCallResultData.model_validate(message.data)
await self._handle_function_call_result(data)
case "raw-audio" | "raw-audio-batch":
await self._handle_audio_buffer(message.data)
case _:
await self._send_error_response(message.id, f"Unsupported type {message.type}")
@@ -1040,34 +1025,9 @@ class RTVIProcessor(FrameProcessor):
logger.warning(f"Exception processing message: {e}")
async def _handle_client_ready(self, request_id: str):
logger.debug("Received client-ready")
if self._input_transport:
self._input_transport.start_audio_in_streaming()
self._client_ready_id = request_id
await self.set_client_ready()
async def _handle_audio_buffer(self, data):
if not self._input_transport:
return
# Extract audio batch ensuring it's a list
audio_list = data.get("base64AudioBatch") or [data.get("base64Audio")]
try:
for base64_audio in filter(None, audio_list): # Filter out None values
pcm_bytes = base64.b64decode(base64_audio)
frame = InputAudioRawFrame(
audio=pcm_bytes,
sample_rate=data["sampleRate"],
num_channels=data["numChannels"],
)
await self._input_transport.push_audio_frame(frame)
except (KeyError, TypeError, ValueError) as e:
# Handle missing keys, decoding errors, and invalid types
logger.error(f"Error processing audio buffer: {e}")
async def _handle_describe_config(self, request_id: str):
services = list(self._registered_services.values())
message = RTVIDescribeConfig(id=request_id, data=RTVIDescribeConfigData(config=services))

View File

@@ -15,7 +15,6 @@ from loguru import logger
from pipecat.audio.utils import calculate_audio_volume, exp_smoothing
from pipecat.frames.frames import (
AudioRawFrame,
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
ErrorFrame,
@@ -76,13 +75,13 @@ class AIService(FrameProcessor):
)
for key, value in settings.items():
logger.debug("Update request for:", key, value)
print("Update request for:", key, value)
if key in self._settings:
logger.info(f"Updating LLM setting {key} to: [{value}]")
self._settings[key] = value
elif key in SessionProperties.model_fields:
logger.debug("Attempting to update", key, value)
print("Attempting to update", key, value)
try:
from pipecat.services.openai_realtime_beta.events import (
@@ -175,7 +174,6 @@ class LLMService(AIService):
f = self._callbacks[None]
else:
return None
await self.call_start_function(context, function_name)
await context.call_function(
f,
function_name=function_name,
@@ -209,13 +207,11 @@ class TTSService(AIService):
# if True, TTSService will push TTSStoppedFrames, otherwise subclass must do it
push_stop_frames: bool = False,
# if push_stop_frames is True, wait for this idle period before pushing TTSStoppedFrame
stop_frame_timeout_s: float = 2.0,
stop_frame_timeout_s: float = 1.0,
# if True, TTSService will push silence audio frames after TTSStoppedFrame
push_silence_after_stop: bool = False,
# if push_silence_after_stop is True, send this amount of audio silence
silence_time_s: float = 2.0,
# if True, we will pause processing frames while we are receiving audio
pause_frame_processing: bool = False,
# TTS output sample rate
sample_rate: Optional[int] = None,
text_filter: Optional[BaseTextFilter] = None,
@@ -228,7 +224,6 @@ class TTSService(AIService):
self._stop_frame_timeout_s: float = stop_frame_timeout_s
self._push_silence_after_stop: bool = push_silence_after_stop
self._silence_time_s: float = silence_time_s
self._pause_frame_processing: bool = pause_frame_processing
self._init_sample_rate = sample_rate
self._sample_rate = 0
self._voice_id: str = ""
@@ -239,7 +234,6 @@ class TTSService(AIService):
self._stop_frame_queue: asyncio.Queue = asyncio.Queue()
self._current_sentence: str = ""
self._processing_text: bool = False
@property
def sample_rate(self) -> int:
@@ -305,7 +299,6 @@ class TTSService(AIService):
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if (
isinstance(frame, TextFrame)
and not isinstance(frame, InterimTranscriptionFrame)
@@ -314,16 +307,9 @@ class TTSService(AIService):
await self._process_text_frame(frame)
elif isinstance(frame, StartInterruptionFrame):
await self._handle_interruption(frame, direction)
await self.push_frame(frame, direction)
elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
# We pause processing incoming frames if the LLM response included
# text (it might be that it's only a function calling response). We
# pause to avoid audio overlapping.
await self._maybe_pause_frame_processing()
sentence = self._current_sentence
self._current_sentence = ""
self._processing_text = False
await self._push_tts_frames(sentence)
if isinstance(frame, LLMFullResponseEndFrame):
if self._push_text_frames:
@@ -332,16 +318,9 @@ class TTSService(AIService):
await self.push_frame(frame, direction)
elif isinstance(frame, TTSSpeakFrame):
await self._push_tts_frames(frame.text)
# We pause processing incoming frames because we are sending data to
# the TTS. We pause to avoid audio overlapping.
await self._maybe_pause_frame_processing()
await self.flush_audio()
self._processing_text = False
elif isinstance(frame, TTSUpdateSettingsFrame):
await self._update_settings(frame.settings)
elif isinstance(frame, BotStoppedSpeakingFrame):
await self._maybe_resume_frame_processing()
await self.push_frame(frame, direction)
else:
await self.push_frame(frame, direction)
@@ -368,17 +347,9 @@ class TTSService(AIService):
async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
self._current_sentence = ""
self._processing_text = False
if self._text_filter:
self._text_filter.handle_interruption()
async def _maybe_pause_frame_processing(self):
if self._processing_text and self._pause_frame_processing:
await self.pause_processing_frames()
async def _maybe_resume_frame_processing(self):
if self._pause_frame_processing:
await self.resume_processing_frames()
await self.push_frame(frame, direction)
async def _process_text_frame(self, frame: TextFrame):
text: Optional[str] = None
@@ -400,11 +371,6 @@ class TTSService(AIService):
if not text.strip():
return
# This is just a flag that indicates if we sent something to the TTS
# service. It will be cleared if we sent text because of a TTSSpeakFrame
# or when we received an LLMFullResponseEndFrame
self._processing_text = True
await self.start_processing_metrics()
if self._text_filter:
self._text_filter.reset_interruption()
@@ -453,7 +419,7 @@ class WordTTSService(TTSService):
async def start(self, frame: StartFrame):
await super().start(frame)
self._create_words_task()
await self._create_words_task()
async def stop(self, frame: EndFrame):
await super().stop(frame)
@@ -473,7 +439,7 @@ class WordTTSService(TTSService):
await super()._handle_interruption(frame, direction)
self.reset_word_timestamps()
def _create_words_task(self):
async def _create_words_task(self):
self._words_task = self.create_task(self._words_task_handler())
async def _stop_words_task(self):
@@ -503,115 +469,6 @@ class WordTTSService(TTSService):
self._words_queue.task_done()
class AudioContextWordTTSService(WordTTSService):
"""This services allow us to send multiple TTS request to the services. Each
request could be multiple sentences long which are grouped by context. For
this to work, the TTS service needs to support handling multiple requests at
once (i.e. multiple simultaneous contexts).
The audio received from the TTS will be played in context order. That is, if
we requested audio for a context "A" and then audio for context "B", the
audio from context ID "A" will be played first.
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._contexts_queue = asyncio.Queue()
self._contexts: Dict[str, asyncio.Queue] = {}
self._audio_context_task = None
async def create_audio_context(self, context_id: str):
"""Create a new audio context."""
await self._contexts_queue.put(context_id)
self._contexts[context_id] = asyncio.Queue()
logger.trace(f"{self} created audio context {context_id}")
async def append_to_audio_context(self, context_id: str, frame: TTSAudioRawFrame):
"""Append audio to an existing context."""
if self.audio_context_available(context_id):
logger.trace(f"{self} appending audio {frame} to audio context {context_id}")
await self._contexts[context_id].put(frame)
else:
logger.warning(f"{self} unable to append audio to context {context_id}")
async def remove_audio_context(self, context_id: str):
"""Remove an existing audio context."""
if self.audio_context_available(context_id):
# We just mark the audio context for deletion by appending
# None. Once we reach None while handling audio we know we can
# safely remove the context.
logger.trace(f"{self} marking audio context {context_id} for deletion")
await self._contexts[context_id].put(None)
else:
logger.warning(f"{self} unable to remove context {context_id}")
def audio_context_available(self, context_id: str) -> bool:
"""Checks whether the given audio context is registered."""
return context_id in self._contexts
async def start(self, frame: StartFrame):
await super().start(frame)
self._create_audio_context_task()
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._stop_audio_context_task()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._stop_audio_context_task()
async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
await super()._handle_interruption(frame, direction)
await self._stop_audio_context_task()
self._create_audio_context_task()
def _create_audio_context_task(self):
self._contexts_queue = asyncio.Queue()
self._contexts: Dict[str, asyncio.Queue] = {}
self._audio_context_task = self.create_task(self._audio_context_task_handler())
async def _stop_audio_context_task(self):
if self._audio_context_task:
await self.cancel_task(self._audio_context_task)
self._audio_context_task = None
async def _audio_context_task_handler(self):
"""In this task we process audio contexts in order."""
while True:
context_id = await self._contexts_queue.get()
# Process the audio context until the context doesn't have more
# audio available (i.e. we find None).
await self._handle_audio_context(context_id)
# We just finished processing the context, so we can safely remove it.
del self._contexts[context_id]
self._contexts_queue.task_done()
# Append some silence between sentences.
silence = b"\x00" * self.sample_rate
frame = TTSAudioRawFrame(audio=silence, sample_rate=self.sample_rate, num_channels=1)
await self.push_frame(frame)
async def _handle_audio_context(self, context_id: str):
# If we don't receive any audio during this time, we consider the context finished.
AUDIO_CONTEXT_TIMEOUT = 3.0
queue = self._contexts[context_id]
running = True
while running:
try:
frame = await asyncio.wait_for(queue.get(), timeout=AUDIO_CONTEXT_TIMEOUT)
if frame:
await self.push_frame(frame)
running = frame is not None
except asyncio.TimeoutError:
# We didn't get audio, so let's consider this context finished.
logger.trace(f"{self} time out on audio context {context_id}")
break
class STTService(AIService):
"""STTService is a base class for speech-to-text services."""
@@ -668,13 +525,9 @@ class STTService(AIService):
else:
logger.warning(f"Unknown setting for STT service: {key}")
async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection):
if self._muted:
return
await self.process_generator(self.run_stt(frame.audio))
if self._audio_passthrough:
await self.push_frame(frame, direction)
async def process_audio_frame(self, frame: AudioRawFrame):
if not self._muted:
await self.process_generator(self.run_stt(frame.audio))
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Processes a frame of audio data, either buffering or transcribing it."""
@@ -684,7 +537,9 @@ class STTService(AIService):
# In this service we accumulate audio internally and at the end we
# push a TextFrame. We also push audio downstream in case someone
# else needs it.
await self.process_audio_frame(frame, direction)
await self.process_audio_frame(frame)
if self._audio_passthrough:
await self.push_frame(frame, direction)
elif isinstance(frame, STTUpdateSettingsFrame):
await self._update_settings(frame.settings)
elif isinstance(frame, STTMuteFrame):
@@ -720,7 +575,7 @@ class SegmentedSTTService(STTService):
self._smoothing_factor = 0.2
self._prev_volume = 0
async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection):
async def process_audio_frame(self, frame: AudioRawFrame):
# Try to filter out empty background noise
volume = self._get_smoothed_volume(frame)
if volume >= self._min_volume:

View File

@@ -126,11 +126,9 @@ class AnthropicLLMService(LLMService):
def create_context_aggregator(
context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True
) -> AnthropicContextAggregatorPair:
if isinstance(context, OpenAILLMContext):
context = AnthropicLLMContext.from_openai_context(context)
user = AnthropicUserContextAggregator(context)
assistant = AnthropicAssistantContextAggregator(
context, expect_stripped_words=assistant_expect_stripped_words
user, expect_stripped_words=assistant_expect_stripped_words
)
return AnthropicContextAggregatorPair(_user=user, _assistant=assistant)
@@ -653,8 +651,11 @@ class AnthropicLLMContext(OpenAILLMContext):
class AnthropicUserContextAggregator(LLMUserContextAggregator):
def __init__(self, context: OpenAILLMContext | AnthropicLLMContext, **kwargs):
super().__init__(context=context, **kwargs)
def __init__(self, context: OpenAILLMContext | AnthropicLLMContext):
super().__init__(context=context)
if isinstance(context, OpenAILLMContext):
self._context = AnthropicLLMContext.from_openai_context(context)
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
@@ -702,8 +703,9 @@ class AnthropicUserContextAggregator(LLMUserContextAggregator):
class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
def __init__(self, context: OpenAILLMContext | AnthropicLLMContext, **kwargs):
super().__init__(context=context, **kwargs)
def __init__(self, user_context_aggregator: AnthropicUserContextAggregator, **kwargs):
super().__init__(context=user_context_aggregator._context, **kwargs)
self._user_context_aggregator = user_context_aggregator
self._function_call_in_progress = None
self._function_call_result = None
self._pending_image_frame_message = None
@@ -723,7 +725,7 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
):
self._function_call_in_progress = None
self._function_call_result = frame
await self.push_aggregation()
await self._push_aggregation()
else:
logger.warning(
"FunctionCallResultFrame tool_call_id != InProgressFrame tool_call_id"
@@ -732,9 +734,9 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
self._function_call_result = None
elif isinstance(frame, AnthropicImageMessageFrame):
self._pending_image_frame_message = frame
await self.push_aggregation()
await self._push_aggregation()
async def push_aggregation(self):
async def _push_aggregation(self):
if not (
self._aggregation or self._function_call_result or self._pending_image_frame_message
):
@@ -743,19 +745,18 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
run_llm = False
properties: Optional[FunctionCallResultProperties] = None
aggregation = self._aggregation.strip()
self.reset()
aggregation = self._aggregation
self._reset()
try:
if aggregation:
self._context.add_message({"role": "assistant", "content": aggregation})
if self._function_call_result:
frame = self._function_call_result
properties = frame.properties
self._function_call_result = None
if frame.result:
assistant_message = {"role": "assistant", "content": []}
if aggregation:
assistant_message["content"].append({"type": "text", "text": aggregation})
assistant_message["content"].append(
{
"type": "tool_use",
@@ -783,6 +784,8 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
else:
# Default behavior
run_llm = True
elif aggregation:
self._context.add_message({"role": "assistant", "content": aggregation})
if self._pending_image_frame_message:
frame = self._pending_image_frame_message
@@ -796,14 +799,15 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
run_llm = True
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
await self._user_context_aggregator.push_context_frame()
# Emit the on_context_updated callback once the function call result is added to the context
if properties and properties.on_context_updated is not None:
await properties.on_context_updated()
# Push context frame
await self.push_context_frame()
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
# Push timestamp frame with current time
timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())

View File

@@ -111,8 +111,6 @@ class BaseWhisperSTTService(SegmentedSTTService):
api_key: Service API key. Defaults to None.
base_url: Service API base URL. Defaults to None.
language: Language of the audio input. Defaults to English.
prompt: Optional text to guide the model's style or continue a previous segment.
temperature: Sampling temperature between 0 and 1. Defaults to 0.0.
**kwargs: Additional arguments passed to SegmentedSTTService.
"""
@@ -123,16 +121,12 @@ class BaseWhisperSTTService(SegmentedSTTService):
api_key: Optional[str] = None,
base_url: Optional[str] = None,
language: Optional[Language] = Language.EN,
prompt: Optional[str] = None,
temperature: Optional[float] = None,
**kwargs,
):
super().__init__(**kwargs)
self.set_model_name(model)
self._client = self._create_client(api_key, base_url)
self._language = self.language_to_service_language(language or Language.EN)
self._prompt = prompt
self._temperature = temperature
def _create_client(self, api_key: Optional[str], base_url: Optional[str]):
return AsyncOpenAI(api_key=api_key, base_url=base_url)

View File

@@ -27,7 +27,7 @@ from pipecat.frames.frames import (
TTSStoppedFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import AudioContextWordTTSService, TTSService
from pipecat.services.ai_services import TTSService, WordTTSService
from pipecat.services.websocket_service import WebsocketService
from pipecat.transcriptions.language import Language
@@ -75,7 +75,7 @@ def language_to_cartesia_language(language: Language) -> Optional[str]:
return result
class CartesiaTTSService(AudioContextWordTTSService, WebsocketService):
class CartesiaTTSService(WordTTSService, WebsocketService):
class InputParams(BaseModel):
language: Optional[Language] = Language.EN
speed: Optional[Union[str, float]] = ""
@@ -105,11 +105,10 @@ class CartesiaTTSService(AudioContextWordTTSService, WebsocketService):
# if we're interrupted. Cartesia gives us word-by-word timestamps. We
# can use those to generate text frames ourselves aligned with the
# playout timing of the audio!
AudioContextWordTTSService.__init__(
WordTTSService.__init__(
self,
aggregate_sentences=True,
push_text_frames=False,
pause_frame_processing=True,
sample_rate=sample_rate,
**kwargs,
)
@@ -192,12 +191,12 @@ class CartesiaTTSService(AudioContextWordTTSService, WebsocketService):
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
async def _disconnect(self):
await self._disconnect_websocket()
if self._receive_task:
await self.cancel_task(self._receive_task)
self._receive_task = None
await self._disconnect_websocket()
async def _connect_websocket(self):
try:
logger.debug("Connecting to Cartesia")
@@ -245,14 +244,17 @@ class CartesiaTTSService(AudioContextWordTTSService, WebsocketService):
async def _receive_messages(self):
async for message in self._get_websocket():
msg = json.loads(message)
if not msg or not self.audio_context_available(msg["context_id"]):
if not msg or msg["context_id"] != self._context_id:
continue
if msg["type"] == "done":
await self.stop_ttfb_metrics()
# Unset _context_id but not the _context_id_start_timestamp
# because we are likely still playing out audio and need the
# timestamp to set send context frames.
self._context_id = None
await self.add_word_timestamps(
[("TTSStoppedFrame", 0), ("LLMFullResponseEndFrame", 0), ("Reset", 0)]
)
await self.remove_audio_context(msg["context_id"])
elif msg["type"] == "timestamps":
await self.add_word_timestamps(
list(zip(msg["word_timestamps"]["words"], msg["word_timestamps"]["start"]))
@@ -265,16 +267,28 @@ class CartesiaTTSService(AudioContextWordTTSService, WebsocketService):
sample_rate=self.sample_rate,
num_channels=1,
)
await self.append_to_audio_context(msg["context_id"], frame)
await self.push_frame(frame)
elif msg["type"] == "error":
logger.error(f"{self} error: {msg}")
await self.push_frame(TTSStoppedFrame())
await self.stop_all_metrics()
await self.push_error(ErrorFrame(f"{self} error: {msg['error']}"))
self._context_id = None
else:
logger.error(f"{self} error, unknown message type: {msg}")
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
# If we received a TTSSpeakFrame and the LLM response included text (it
# might be that it's only a function calling response) we pause
# processing more frames until we receive a BotStoppedSpeakingFrame.
if isinstance(frame, TTSSpeakFrame):
await self.pause_processing_frames()
elif isinstance(frame, LLMFullResponseEndFrame) and self._context_id:
await self.pause_processing_frames()
elif isinstance(frame, BotStoppedSpeakingFrame):
await self.resume_processing_frames()
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating TTS: [{text}]")
@@ -286,7 +300,6 @@ class CartesiaTTSService(AudioContextWordTTSService, WebsocketService):
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._context_id = str(uuid.uuid4())
await self.create_audio_context(self._context_id)
msg = self._build_msg(text=text or " ") # Text must contain at least one character
@@ -364,6 +377,9 @@ class CartesiaHttpTTSService(TTSService):
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating TTS: [{text}]")
await self.start_ttfb_metrics()
yield TTSStartedFrame()
try:
voice_controls = None
if self._settings["speed"] or self._settings["emotion"]:
@@ -373,8 +389,6 @@ class CartesiaHttpTTSService(TTSService):
if self._settings["emotion"]:
voice_controls["emotion"] = self._settings["emotion"]
await self.start_ttfb_metrics()
output = await self._client.tts.sse(
model_id=self._model_name,
transcript=text,
@@ -385,17 +399,14 @@ class CartesiaHttpTTSService(TTSService):
_experimental_voice_controls=voice_controls,
)
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
frame = TTSAudioRawFrame(
audio=output["audio"], sample_rate=self.sample_rate, num_channels=1
)
yield frame
except Exception as e:
logger.error(f"{self} exception: {e}")
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
await self.start_tts_usage_metrics(text)
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()

View File

@@ -7,7 +7,6 @@
from typing import List
from loguru import logger
from tenacity import retry, stop_after_attempt, wait_fixed
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai import OpenAILLMService
@@ -53,7 +52,6 @@ class CerebrasLLMService(OpenAILLMService):
logger.debug(f"Creating Cerebras client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
@retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
async def get_chat_completions(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> AsyncStream[ChatCompletionChunk]:

View File

@@ -5,7 +5,7 @@
#
import asyncio
from typing import AsyncGenerator, Dict, Optional
from typing import AsyncGenerator, Optional
from loguru import logger
@@ -34,7 +34,6 @@ try:
AsyncListenWebSocketClient,
DeepgramClient,
DeepgramClientOptions,
ErrorResponse,
LiveOptions,
LiveResultResponse,
LiveTranscriptionEvents,
@@ -121,16 +120,14 @@ class DeepgramSTTService(STTService):
url: str = "",
sample_rate: Optional[int] = None,
live_options: Optional[LiveOptions] = None,
addons: Optional[Dict] = None,
**kwargs,
):
sample_rate = sample_rate or (live_options.sample_rate if live_options else None)
super().__init__(sample_rate=sample_rate, **kwargs)
default_options = LiveOptions(
encoding="linear16",
language=Language.EN,
model="nova-3-general",
model="nova-2-general",
channels=1,
interim_results=True,
smart_format=True,
@@ -150,7 +147,6 @@ class DeepgramSTTService(STTService):
merged_options.language = merged_options.language.value
self._settings = merged_options.to_dict()
self._addons = addons
self._client = DeepgramClient(
api_key,
@@ -159,10 +155,13 @@ class DeepgramSTTService(STTService):
options={"keepalive": "true"}, # verbose=logging.DEBUG
),
)
self._connection: AsyncListenWebSocketClient = self._client.listen.asyncwebsocket.v("1")
self._connection.on(LiveTranscriptionEvents.Transcript, self._on_message)
if self.vad_enabled:
self._register_event_handler("on_speech_started")
self._register_event_handler("on_utterance_end")
self._connection.on(LiveTranscriptionEvents.SpeechStarted, self._on_speech_started)
self._connection.on(LiveTranscriptionEvents.UtteranceEnd, self._on_utterance_end)
@property
def vad_enabled(self):
@@ -203,25 +202,7 @@ class DeepgramSTTService(STTService):
async def _connect(self):
logger.debug("Connecting to Deepgram")
self._connection: AsyncListenWebSocketClient = self._client.listen.asyncwebsocket.v("1")
self._connection.on(
LiveTranscriptionEvents(LiveTranscriptionEvents.Transcript), self._on_message
)
self._connection.on(LiveTranscriptionEvents(LiveTranscriptionEvents.Error), self._on_error)
if self.vad_enabled:
self._connection.on(
LiveTranscriptionEvents(LiveTranscriptionEvents.SpeechStarted),
self._on_speech_started,
)
self._connection.on(
LiveTranscriptionEvents(LiveTranscriptionEvents.UtteranceEnd),
self._on_utterance_end,
)
if not await self._connection.start(options=self._settings, addons=self._addons):
if not await self._connection.start(self._settings):
logger.error(f"{self}: unable to connect to Deepgram")
async def _disconnect(self):
@@ -233,15 +214,6 @@ class DeepgramSTTService(STTService):
await self.start_ttfb_metrics()
await self.start_processing_metrics()
async def _on_error(self, *args, **kwargs):
error: ErrorResponse = kwargs["error"]
logger.warning(f"{self} connection error, will retry: {error}")
await self.stop_all_metrics()
# NOTE(aleix): we don't disconnect (i.e. call finish on the connection)
# because this triggers more errors internally in the Deepgram SDK. So,
# we just forget about the previous connection and create a new one.
await self._connect()
async def _on_speech_started(self, *args, **kwargs):
await self.start_metrics()
await self._call_event_handler("on_speech_started", *args, **kwargs)

View File

@@ -8,7 +8,6 @@
from typing import List
from loguru import logger
from tenacity import retry, stop_after_attempt, wait_fixed
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai import OpenAILLMService
@@ -54,7 +53,6 @@ class DeepSeekLLMService(OpenAILLMService):
logger.debug(f"Creating DeepSeek client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
@retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
async def get_chat_completions(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> AsyncStream[ChatCompletionChunk]:

View File

@@ -191,7 +191,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
aggregate_sentences=True,
push_text_frames=False,
push_stop_frames=True,
pause_frame_processing=True,
stop_frame_timeout_s=2.0,
sample_rate=sample_rate,
**kwargs,
)
@@ -289,6 +289,19 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
if isinstance(frame, TTSStoppedFrame):
await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)])
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
# If we received a TTSSpeakFrame and the LLM response included text (it
# might be that it's only a function calling response) we pause
# processing more frames until we receive a BotStoppedSpeakingFrame.
if isinstance(frame, TTSSpeakFrame):
await self.pause_processing_frames()
elif isinstance(frame, LLMFullResponseEndFrame) and self._started:
await self.pause_processing_frames()
elif isinstance(frame, BotStoppedSpeakingFrame):
await self.resume_processing_frames()
async def _connect(self):
await self._connect_websocket()
@@ -357,13 +370,8 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
except Exception as e:
logger.error(f"{self} error closing websocket: {e}")
def _get_websocket(self):
if self._websocket:
return self._websocket
raise Exception("Websocket not connected")
async def _receive_messages(self):
async for message in self._get_websocket():
async for message in self._websocket:
msg = json.loads(message)
if msg.get("audio"):
await self.stop_ttfb_metrics()
@@ -566,16 +574,18 @@ class ElevenLabsHttpTTSService(TTSService):
return
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
async for chunk in response.content:
if chunk:
await self.stop_ttfb_metrics()
yield TTSAudioRawFrame(chunk, self.sample_rate, 1)
yield TTSStoppedFrame()
except Exception as e:
logger.error(f"Error in run_tts: {e}")
yield ErrorFrame(error=str(e))
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()

View File

@@ -4,7 +4,6 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import io
import os
from typing import AsyncGenerator, Dict, Optional, Union
@@ -54,11 +53,6 @@ class FalImageGenService(ImageGenService):
os.environ["FAL_KEY"] = key
async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
def load_image_bytes(encoded_image: bytes):
buffer = io.BytesIO(encoded_image)
image = Image.open(buffer)
return (image.tobytes(), image.size, image.format)
logger.debug(f"Generating image from prompt: {prompt}")
response = await fal_client.run_async(
@@ -79,8 +73,10 @@ class FalImageGenService(ImageGenService):
logger.debug(f"Downloading image {image_url} ...")
async with self._aiohttp_session.get(image_url) as response:
logger.debug(f"Downloaded image {image_url}")
encoded_image = await response.content.read()
(image_bytes, size, format) = await asyncio.to_thread(load_image_bytes, encoded_image)
image_stream = io.BytesIO(await response.content.read())
image = Image.open(image_stream)
frame = URLImageRawFrame(url=image_url, image=image_bytes, size=size, format=format)
frame = URLImageRawFrame(
url=image_url, image=image.tobytes(), size=image.size, format=image.format
)
yield frame

View File

@@ -8,7 +8,6 @@
from typing import List
from loguru import logger
from tenacity import retry, stop_after_attempt, wait_fixed
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai import OpenAILLMService
@@ -51,7 +50,6 @@ class FireworksLLMService(OpenAILLMService):
logger.debug(f"Creating Fireworks client with api {base_url}")
return super().create_client(api_key, base_url, **kwargs)
@retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
async def get_chat_completions(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
):

View File

@@ -11,13 +11,16 @@ from loguru import logger
from pydantic import BaseModel
from pipecat.frames.frames import (
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
ErrorFrame,
Frame,
LLMFullResponseEndFrame,
StartFrame,
StartInterruptionFrame,
TTSAudioRawFrame,
TTSSpeakFrame,
TTSStartedFrame,
TTSStoppedFrame,
)
@@ -57,12 +60,7 @@ class FishAudioTTSService(TTSService, WebsocketService):
params: InputParams = InputParams(),
**kwargs,
):
super().__init__(
push_stop_frames=True,
pause_frame_processing=True,
sample_rate=sample_rate,
**kwargs,
)
super().__init__(sample_rate=sample_rate, **kwargs)
self._api_key = api_key
self._base_url = "wss://api.fish.audio/v1/tts/live"
@@ -168,6 +166,16 @@ class FishAudioTTSService(TTSService, WebsocketService):
except Exception as e:
logger.error(f"Error processing message: {e}")
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, TTSSpeakFrame):
await self.pause_processing_frames()
elif isinstance(frame, LLMFullResponseEndFrame) and self._request_id:
await self.pause_processing_frames()
elif isinstance(frame, BotStoppedSpeakingFrame):
await self.resume_processing_frames()
async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
await super()._handle_interruption(frame, direction)
await self.stop_all_metrics()

View File

@@ -115,10 +115,10 @@ class GeminiMultimodalLiveUserContextAggregator(OpenAIUserContextAggregator):
class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
async def push_aggregation(self):
async def _push_aggregation(self):
# We don't want to store any images in the context. Revisit this later when the API evolves.
self._pending_image_frame_message = None
await super().push_aggregation()
await super()._push_aggregation()
@dataclass
@@ -706,6 +706,6 @@ class GeminiMultimodalLiveLLMService(LLMService):
GeminiMultimodalLiveContext.upgrade(context)
user = GeminiMultimodalLiveUserContextAggregator(context)
assistant = GeminiMultimodalLiveAssistantContextAggregator(
context, expect_stripped_words=assistant_expect_stripped_words
user, expect_stripped_words=assistant_expect_stripped_words
)
return GeminiMultimodalLiveContextAggregatorPair(_user=user, _assistant=assistant)

View File

@@ -8,35 +8,24 @@ import asyncio
import base64
import io
import json
import os
import time
# Suppress gRPC fork warnings
os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
from dataclasses import dataclass
from typing import Any, AsyncGenerator, Dict, List, Literal, Optional, Union
from typing import Any, AsyncGenerator, Dict, List, Literal, Optional
from loguru import logger
from PIL import Image
from pydantic import BaseModel, Field, field_validator
from pydantic import BaseModel, Field
from pipecat.frames.frames import (
AudioRawFrame,
CancelFrame,
EndFrame,
ErrorFrame,
Frame,
FunctionCallResultProperties,
InterimTranscriptionFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
LLMTextFrame,
LLMUpdateSettingsFrame,
OpenAILLMContextAssistantTimestampFrame,
StartFrame,
TranscriptionFrame,
TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
@@ -49,7 +38,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContextFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import ImageGenService, LLMService, STTService, TTSService
from pipecat.services.ai_services import ImageGenService, LLMService, TTSService
from pipecat.services.google.frames import LLMSearchResponseFrame
from pipecat.services.openai import (
OpenAIAssistantContextAggregator,
@@ -62,13 +51,10 @@ try:
import google.ai.generativelanguage as glm
import google.generativeai as gai
from google import genai
from google.api_core.client_options import ClientOptions
from google.cloud import speech_v2, texttospeech_v1
from google.cloud.speech_v2.types import cloud_speech
from google.cloud import texttospeech_v1
from google.genai import types
from google.generativeai.types import GenerationConfig
from google.oauth2 import service_account
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
@@ -77,7 +63,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
def language_to_google_tts_language(language: Language) -> Optional[str]:
def language_to_google_language(language: Language) -> Optional[str]:
language_map = {
# Afrikaans
Language.AF: "af-ZA",
@@ -237,307 +223,8 @@ def language_to_google_tts_language(language: Language) -> Optional[str]:
return language_map.get(language)
def language_to_google_stt_language(language: Language) -> Optional[str]:
"""Maps Language enum to Google Speech-to-Text V2 language codes.
Args:
language: Language enum value.
Returns:
Optional[str]: Google STT language code or None if not supported.
"""
language_map = {
# Afrikaans
Language.AF: "af-ZA",
Language.AF_ZA: "af-ZA",
# Albanian
Language.SQ: "sq-AL",
Language.SQ_AL: "sq-AL",
# Amharic
Language.AM: "am-ET",
Language.AM_ET: "am-ET",
# Arabic
Language.AR: "ar-EG", # Default to Egypt
Language.AR_AE: "ar-AE",
Language.AR_BH: "ar-BH",
Language.AR_DZ: "ar-DZ",
Language.AR_EG: "ar-EG",
Language.AR_IQ: "ar-IQ",
Language.AR_JO: "ar-JO",
Language.AR_KW: "ar-KW",
Language.AR_LB: "ar-LB",
Language.AR_MA: "ar-MA",
Language.AR_OM: "ar-OM",
Language.AR_QA: "ar-QA",
Language.AR_SA: "ar-SA",
Language.AR_SY: "ar-SY",
Language.AR_TN: "ar-TN",
Language.AR_YE: "ar-YE",
# Armenian
Language.HY: "hy-AM",
Language.HY_AM: "hy-AM",
# Azerbaijani
Language.AZ: "az-AZ",
Language.AZ_AZ: "az-AZ",
# Basque
Language.EU: "eu-ES",
Language.EU_ES: "eu-ES",
# Bengali
Language.BN: "bn-IN", # Default to India
Language.BN_BD: "bn-BD",
Language.BN_IN: "bn-IN",
# Bosnian
Language.BS: "bs-BA",
Language.BS_BA: "bs-BA",
# Bulgarian
Language.BG: "bg-BG",
Language.BG_BG: "bg-BG",
# Burmese
Language.MY: "my-MM",
Language.MY_MM: "my-MM",
# Catalan
Language.CA: "ca-ES",
Language.CA_ES: "ca-ES",
# Chinese
Language.ZH: "cmn-Hans-CN", # Default to Simplified Chinese
Language.ZH_CN: "cmn-Hans-CN",
Language.ZH_HK: "cmn-Hans-HK",
Language.ZH_TW: "cmn-Hant-TW",
Language.YUE: "yue-Hant-HK", # Cantonese
Language.YUE_CN: "yue-Hant-HK",
# Croatian
Language.HR: "hr-HR",
Language.HR_HR: "hr-HR",
# Czech
Language.CS: "cs-CZ",
Language.CS_CZ: "cs-CZ",
# Danish
Language.DA: "da-DK",
Language.DA_DK: "da-DK",
# Dutch
Language.NL: "nl-NL", # Default to Netherlands
Language.NL_BE: "nl-BE",
Language.NL_NL: "nl-NL",
# English
Language.EN: "en-US", # Default to US
Language.EN_AU: "en-AU",
Language.EN_CA: "en-CA",
Language.EN_GB: "en-GB",
Language.EN_GH: "en-GH",
Language.EN_HK: "en-HK",
Language.EN_IN: "en-IN",
Language.EN_IE: "en-IE",
Language.EN_KE: "en-KE",
Language.EN_NG: "en-NG",
Language.EN_NZ: "en-NZ",
Language.EN_PH: "en-PH",
Language.EN_SG: "en-SG",
Language.EN_TZ: "en-TZ",
Language.EN_US: "en-US",
Language.EN_ZA: "en-ZA",
# Estonian
Language.ET: "et-EE",
Language.ET_EE: "et-EE",
# Filipino
Language.FIL: "fil-PH",
Language.FIL_PH: "fil-PH",
# Finnish
Language.FI: "fi-FI",
Language.FI_FI: "fi-FI",
# French
Language.FR: "fr-FR", # Default to France
Language.FR_BE: "fr-BE",
Language.FR_CA: "fr-CA",
Language.FR_CH: "fr-CH",
Language.FR_FR: "fr-FR",
# Galician
Language.GL: "gl-ES",
Language.GL_ES: "gl-ES",
# Georgian
Language.KA: "ka-GE",
Language.KA_GE: "ka-GE",
# German
Language.DE: "de-DE", # Default to Germany
Language.DE_AT: "de-AT",
Language.DE_CH: "de-CH",
Language.DE_DE: "de-DE",
# Greek
Language.EL: "el-GR",
Language.EL_GR: "el-GR",
# Gujarati
Language.GU: "gu-IN",
Language.GU_IN: "gu-IN",
# Hebrew
Language.HE: "iw-IL",
Language.HE_IL: "iw-IL",
# Hindi
Language.HI: "hi-IN",
Language.HI_IN: "hi-IN",
# Hungarian
Language.HU: "hu-HU",
Language.HU_HU: "hu-HU",
# Icelandic
Language.IS: "is-IS",
Language.IS_IS: "is-IS",
# Indonesian
Language.ID: "id-ID",
Language.ID_ID: "id-ID",
# Italian
Language.IT: "it-IT",
Language.IT_IT: "it-IT",
Language.IT_CH: "it-CH",
# Japanese
Language.JA: "ja-JP",
Language.JA_JP: "ja-JP",
# Javanese
Language.JV: "jv-ID",
Language.JV_ID: "jv-ID",
# Kannada
Language.KN: "kn-IN",
Language.KN_IN: "kn-IN",
# Kazakh
Language.KK: "kk-KZ",
Language.KK_KZ: "kk-KZ",
# Khmer
Language.KM: "km-KH",
Language.KM_KH: "km-KH",
# Korean
Language.KO: "ko-KR",
Language.KO_KR: "ko-KR",
# Lao
Language.LO: "lo-LA",
Language.LO_LA: "lo-LA",
# Latvian
Language.LV: "lv-LV",
Language.LV_LV: "lv-LV",
# Lithuanian
Language.LT: "lt-LT",
Language.LT_LT: "lt-LT",
# Macedonian
Language.MK: "mk-MK",
Language.MK_MK: "mk-MK",
# Malay
Language.MS: "ms-MY",
Language.MS_MY: "ms-MY",
# Malayalam
Language.ML: "ml-IN",
Language.ML_IN: "ml-IN",
# Marathi
Language.MR: "mr-IN",
Language.MR_IN: "mr-IN",
# Mongolian
Language.MN: "mn-MN",
Language.MN_MN: "mn-MN",
# Nepali
Language.NE: "ne-NP",
Language.NE_NP: "ne-NP",
# Norwegian
Language.NO: "no-NO",
Language.NB: "no-NO",
Language.NB_NO: "no-NO",
# Persian
Language.FA: "fa-IR",
Language.FA_IR: "fa-IR",
# Polish
Language.PL: "pl-PL",
Language.PL_PL: "pl-PL",
# Portuguese
Language.PT: "pt-PT", # Default to Portugal
Language.PT_BR: "pt-BR",
Language.PT_PT: "pt-PT",
# Punjabi
Language.PA: "pa-Guru-IN",
Language.PA_IN: "pa-Guru-IN",
# Romanian
Language.RO: "ro-RO",
Language.RO_RO: "ro-RO",
# Russian
Language.RU: "ru-RU",
Language.RU_RU: "ru-RU",
# Serbian
Language.SR: "sr-RS",
Language.SR_RS: "sr-RS",
# Sinhala
Language.SI: "si-LK",
Language.SI_LK: "si-LK",
# Slovak
Language.SK: "sk-SK",
Language.SK_SK: "sk-SK",
# Slovenian
Language.SL: "sl-SI",
Language.SL_SI: "sl-SI",
# Spanish
Language.ES: "es-ES", # Default to Spain
Language.ES_AR: "es-AR",
Language.ES_BO: "es-BO",
Language.ES_CL: "es-CL",
Language.ES_CO: "es-CO",
Language.ES_CR: "es-CR",
Language.ES_DO: "es-DO",
Language.ES_EC: "es-EC",
Language.ES_ES: "es-ES",
Language.ES_GT: "es-GT",
Language.ES_HN: "es-HN",
Language.ES_MX: "es-MX",
Language.ES_NI: "es-NI",
Language.ES_PA: "es-PA",
Language.ES_PE: "es-PE",
Language.ES_PR: "es-PR",
Language.ES_PY: "es-PY",
Language.ES_SV: "es-SV",
Language.ES_US: "es-US",
Language.ES_UY: "es-UY",
Language.ES_VE: "es-VE",
# Sundanese
Language.SU: "su-ID",
Language.SU_ID: "su-ID",
# Swahili
Language.SW: "sw-TZ", # Default to Tanzania
Language.SW_KE: "sw-KE",
Language.SW_TZ: "sw-TZ",
# Swedish
Language.SV: "sv-SE",
Language.SV_SE: "sv-SE",
# Tamil
Language.TA: "ta-IN", # Default to India
Language.TA_IN: "ta-IN",
Language.TA_MY: "ta-MY",
Language.TA_SG: "ta-SG",
Language.TA_LK: "ta-LK",
# Telugu
Language.TE: "te-IN",
Language.TE_IN: "te-IN",
# Thai
Language.TH: "th-TH",
Language.TH_TH: "th-TH",
# Turkish
Language.TR: "tr-TR",
Language.TR_TR: "tr-TR",
# Ukrainian
Language.UK: "uk-UA",
Language.UK_UA: "uk-UA",
# Urdu
Language.UR: "ur-IN", # Default to India
Language.UR_IN: "ur-IN",
Language.UR_PK: "ur-PK",
# Uzbek
Language.UZ: "uz-UZ",
Language.UZ_UZ: "uz-UZ",
# Vietnamese
Language.VI: "vi-VN",
Language.VI_VN: "vi-VN",
# Xhosa
Language.XH: "xh-ZA",
# Zulu
Language.ZU: "zu-ZA",
Language.ZU_ZA: "zu-ZA",
}
return language_map.get(language)
class GoogleUserContextAggregator(OpenAIUserContextAggregator):
async def push_aggregation(self):
async def _push_aggregation(self):
if len(self._aggregation) > 0:
self._context.add_message(
glm.Content(role="user", parts=[glm.Part(text=self._aggregation)])
@@ -552,11 +239,11 @@ class GoogleUserContextAggregator(OpenAIUserContextAggregator):
await self.push_frame(frame)
# Reset our accumulator state.
self.reset()
self._reset()
class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
async def push_aggregation(self):
async def _push_aggregation(self):
if not (
self._aggregation or self._function_call_result or self._pending_image_frame_message
):
@@ -565,15 +252,10 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
run_llm = False
properties: Optional[FunctionCallResultProperties] = None
aggregation = self._aggregation.strip()
self.reset()
aggregation = self._aggregation
self._reset()
try:
if aggregation:
self._context.add_message(
glm.Content(role="model", parts=[glm.Part(text=aggregation)])
)
if self._function_call_result:
frame = self._function_call_result
properties = frame.properties
@@ -613,6 +295,11 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
else:
# Default behavior is to run the LLM if there are no function calls in progress
run_llm = not bool(self._function_calls_in_progress)
else:
if aggregation.strip():
self._context.add_message(
glm.Content(role="model", parts=[glm.Part(text=aggregation)])
)
if self._pending_image_frame_message:
frame = self._pending_image_frame_message
@@ -626,14 +313,15 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
run_llm = True
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
await self._user_context_aggregator.push_context_frame()
# Emit the on_context_updated callback once the function call result is added to the context
if properties and properties.on_context_updated is not None:
await properties.on_context_updated()
# Push context frame
await self.push_context_frame()
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
# Push timestamp frame with current time
timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
@@ -916,9 +604,9 @@ class GoogleLLMContext(OpenAILLMContext):
# Check if we only have function-related messages (no regular text)
has_regular_messages = any(
len(msg.parts) == 1
and not getattr(msg.parts[0], "text", None)
and getattr(msg.parts[0], "function_call", None)
and getattr(msg.parts[0], "function_response", None)
and hasattr(msg.parts[0], "text")
and not hasattr(msg.parts[0], "function_call")
and not hasattr(msg.parts[0], "function_response")
for msg in self._messages
)
@@ -933,7 +621,7 @@ class GoogleLLMContext(OpenAILLMContext):
class GoogleLLMService(LLMService):
"""This class implements inference with Google's AI models.
"""This class implements inference with Google's AI models
This service translates internally from OpenAILLMContext to the messages format
expected by the Google AI model. We are using the OpenAILLMContext as a lingua
@@ -1174,7 +862,7 @@ class GoogleLLMService(LLMService):
) -> GoogleContextAggregatorPair:
user = GoogleUserContextAggregator(context)
assistant = GoogleAssistantContextAggregator(
context, expect_stripped_words=assistant_expect_stripped_words
user, expect_stripped_words=assistant_expect_stripped_words
)
return GoogleContextAggregatorPair(_user=user, _assistant=assistant)
@@ -1239,7 +927,7 @@ class GoogleTTSService(TTSService):
return True
def language_to_service_language(self, language: Language) -> Optional[str]:
return language_to_google_tts_language(language)
return language_to_google_language(language)
def _construct_ssml(self, text: str) -> str:
ssml = "<speak>"
@@ -1409,451 +1097,3 @@ class GoogleImageGenService(ImageGenService):
except Exception as e:
logger.error(f"{self} error generating image: {e}")
yield ErrorFrame(f"Image generation error: {str(e)}")
class GoogleSTTService(STTService):
"""Google Cloud Speech-to-Text V2 service implementation.
Provides real-time speech recognition using Google Cloud's Speech-to-Text V2 API
with streaming support. Handles audio transcription and optional voice activity detection.
Attributes:
InputParams: Configuration parameters for the STT service.
"""
# Google Cloud's STT service has a connection time limit of 5 minutes per stream.
# They've shared an "endless streaming" example that guided this implementation:
# https://cloud.google.com/speech-to-text/docs/transcribe-streaming-audio#endless-streaming
STREAMING_LIMIT = 240000 # 4 minutes in milliseconds
class InputParams(BaseModel):
"""Configuration parameters for Google Speech-to-Text.
Attributes:
languages: Single language or list of recognition languages. First language is primary.
model: Speech recognition model to use.
use_separate_recognition_per_channel: Process each audio channel separately.
enable_automatic_punctuation: Add punctuation to transcripts.
enable_spoken_punctuation: Include spoken punctuation in transcript.
enable_spoken_emojis: Include spoken emojis in transcript.
profanity_filter: Filter profanity from transcript.
enable_word_time_offsets: Include timing information for each word.
enable_word_confidence: Include confidence scores for each word.
enable_interim_results: Stream partial recognition results.
enable_voice_activity_events: Detect voice activity in audio.
"""
languages: Union[Language, List[Language]] = Field(default_factory=lambda: [Language.EN_US])
model: Optional[str] = "latest_long"
use_separate_recognition_per_channel: Optional[bool] = False
enable_automatic_punctuation: Optional[bool] = True
enable_spoken_punctuation: Optional[bool] = False
enable_spoken_emojis: Optional[bool] = False
profanity_filter: Optional[bool] = False
enable_word_time_offsets: Optional[bool] = False
enable_word_confidence: Optional[bool] = False
enable_interim_results: Optional[bool] = True
enable_voice_activity_events: Optional[bool] = False
@field_validator("languages", mode="before")
@classmethod
def validate_languages(cls, v) -> List[Language]:
if isinstance(v, Language):
return [v]
return v
@property
def language_list(self) -> List[Language]:
"""Get languages as a guaranteed list."""
assert isinstance(self.languages, list)
return self.languages
def __init__(
self,
*,
credentials: Optional[str] = None,
credentials_path: Optional[str] = None,
location: str = "global",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
"""Initialize the Google STT service.
Args:
credentials: JSON string containing Google Cloud service account credentials.
credentials_path: Path to service account credentials JSON file.
location: Google Cloud location (e.g., "global", "us-central1").
sample_rate: Audio sample rate in Hertz.
params: Configuration parameters for the service.
**kwargs: Additional arguments passed to STTService.
Raises:
ValueError: If neither credentials nor credentials_path is provided.
ValueError: If project ID is not found in credentials.
"""
super().__init__(sample_rate=sample_rate, **kwargs)
self._location = location
self._stream = None
self._config = None
self._request_queue = asyncio.Queue()
self._streaming_task = None
# Used for keep-alive logic
self._stream_start_time = 0
self._last_audio_input = []
self._audio_input = []
self._result_end_time = 0
self._is_final_end_time = 0
self._final_request_end_time = 0
self._bridging_offset = 0
self._last_transcript_was_final = False
self._new_stream = True
self._restart_counter = 0
# Configure client options based on location
client_options = None
if self._location != "global":
client_options = ClientOptions(api_endpoint=f"{self._location}-speech.googleapis.com")
# Extract project ID and create client
if credentials:
json_account_info = json.loads(credentials)
self._project_id = json_account_info.get("project_id")
creds = service_account.Credentials.from_service_account_info(json_account_info)
elif credentials_path:
with open(credentials_path) as f:
json_account_info = json.load(f)
self._project_id = json_account_info.get("project_id")
creds = service_account.Credentials.from_service_account_file(credentials_path)
else:
raise ValueError("Either credentials or credentials_path must be provided")
if not self._project_id:
raise ValueError("Project ID not found in credentials")
self._client = speech_v2.SpeechAsyncClient(credentials=creds, client_options=client_options)
self._settings = {
"language_codes": [
self.language_to_service_language(lang) for lang in params.language_list
],
"model": params.model,
"use_separate_recognition_per_channel": params.use_separate_recognition_per_channel,
"enable_automatic_punctuation": params.enable_automatic_punctuation,
"enable_spoken_punctuation": params.enable_spoken_punctuation,
"enable_spoken_emojis": params.enable_spoken_emojis,
"profanity_filter": params.profanity_filter,
"enable_word_time_offsets": params.enable_word_time_offsets,
"enable_word_confidence": params.enable_word_confidence,
"enable_interim_results": params.enable_interim_results,
"enable_voice_activity_events": params.enable_voice_activity_events,
}
def language_to_service_language(self, language: Language | List[Language]) -> str | List[str]:
"""Convert Language enum(s) to Google STT language code(s).
Args:
language: Single Language enum or list of Language enums.
Returns:
str | List[str]: Google STT language code(s).
"""
if isinstance(language, list):
return [language_to_google_stt_language(lang) or "en-US" for lang in language]
return language_to_google_stt_language(language) or "en-US"
async def _reconnect_if_needed(self):
"""Reconnect the stream if it's currently active."""
if self._streaming_task:
logger.debug("Reconnecting stream due to configuration changes")
await self._disconnect()
await self._connect()
async def set_languages(self, languages: List[Language]):
"""Update the service's recognition languages.
Args:
languages: List of languages for recognition. First language is primary.
"""
logger.debug(f"Switching STT languages to: {languages}")
self._settings["language_codes"] = [
self.language_to_service_language(lang) for lang in languages
]
# Recreate stream with new languages
await self._reconnect_if_needed()
async def set_model(self, model: str):
"""Update the service's recognition model."""
logger.debug(f"Switching STT model to: {model}")
await super().set_model(model)
self._settings["model"] = model
# Recreate stream with new model
await self._reconnect_if_needed()
async def start(self, frame: StartFrame):
await super().start(frame)
await self._connect()
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._disconnect()
async def update_options(
self,
*,
languages: Optional[List[Language]] = None,
model: Optional[str] = None,
enable_automatic_punctuation: Optional[bool] = None,
enable_spoken_punctuation: Optional[bool] = None,
enable_spoken_emojis: Optional[bool] = None,
profanity_filter: Optional[bool] = None,
enable_word_time_offsets: Optional[bool] = None,
enable_word_confidence: Optional[bool] = None,
enable_interim_results: Optional[bool] = None,
enable_voice_activity_events: Optional[bool] = None,
location: Optional[str] = None,
) -> None:
"""Update service options dynamically.
Args:
languages: New list of recongition languages.
model: New recognition model.
enable_automatic_punctuation: Enable/disable automatic punctuation.
enable_spoken_punctuation: Enable/disable spoken punctuation.
enable_spoken_emojis: Enable/disable spoken emojis.
profanity_filter: Enable/disable profanity filter.
enable_word_time_offsets: Enable/disable word timing info.
enable_word_confidence: Enable/disable word confidence scores.
enable_interim_results: Enable/disable interim results.
enable_voice_activity_events: Enable/disable voice activity detection.
location: New Google Cloud location.
Note:
Changes that affect the streaming configuration will cause
the stream to be reconnected.
"""
# Update settings with new values
if languages is not None:
logger.debug(f"Updating language to: {languages}")
self._settings["language_codes"] = [
self.language_to_service_language(lang) for lang in languages
]
if model is not None:
logger.debug(f"Updating model to: {model}")
self._settings["model"] = model
if enable_automatic_punctuation is not None:
logger.debug(f"Updating automatic punctuation to: {enable_automatic_punctuation}")
self._settings["enable_automatic_punctuation"] = enable_automatic_punctuation
if enable_spoken_punctuation is not None:
logger.debug(f"Updating spoken punctuation to: {enable_spoken_punctuation}")
self._settings["enable_spoken_punctuation"] = enable_spoken_punctuation
if enable_spoken_emojis is not None:
logger.debug(f"Updating spoken emojis to: {enable_spoken_emojis}")
self._settings["enable_spoken_emojis"] = enable_spoken_emojis
if profanity_filter is not None:
logger.debug(f"Updating profanity filter to: {profanity_filter}")
self._settings["profanity_filter"] = profanity_filter
if enable_word_time_offsets is not None:
logger.debug(f"Updating word time offsets to: {enable_word_time_offsets}")
self._settings["enable_word_time_offsets"] = enable_word_time_offsets
if enable_word_confidence is not None:
logger.debug(f"Updating word confidence to: {enable_word_confidence}")
self._settings["enable_word_confidence"] = enable_word_confidence
if enable_interim_results is not None:
logger.debug(f"Updating interim results to: {enable_interim_results}")
self._settings["enable_interim_results"] = enable_interim_results
if enable_voice_activity_events is not None:
logger.debug(f"Updating voice activity events to: {enable_voice_activity_events}")
self._settings["enable_voice_activity_events"] = enable_voice_activity_events
if location is not None:
logger.debug(f"Updating location to: {location}")
self._location = location
# Reconnect the stream for updates
await self._reconnect_if_needed()
async def _connect(self):
"""Initialize streaming recognition config and stream."""
logger.debug("Connecting to Google Speech-to-Text")
# Set stream start time
self._stream_start_time = int(time.time() * 1000)
self._new_stream = True
self._config = cloud_speech.StreamingRecognitionConfig(
config=cloud_speech.RecognitionConfig(
explicit_decoding_config=cloud_speech.ExplicitDecodingConfig(
encoding=cloud_speech.ExplicitDecodingConfig.AudioEncoding.LINEAR16,
sample_rate_hertz=self.sample_rate,
audio_channel_count=1,
),
language_codes=self._settings["language_codes"],
model=self._settings["model"],
features=cloud_speech.RecognitionFeatures(
enable_automatic_punctuation=self._settings["enable_automatic_punctuation"],
enable_spoken_punctuation=self._settings["enable_spoken_punctuation"],
enable_spoken_emojis=self._settings["enable_spoken_emojis"],
profanity_filter=self._settings["profanity_filter"],
enable_word_time_offsets=self._settings["enable_word_time_offsets"],
enable_word_confidence=self._settings["enable_word_confidence"],
),
),
streaming_features=cloud_speech.StreamingRecognitionFeatures(
enable_voice_activity_events=self._settings["enable_voice_activity_events"],
interim_results=self._settings["enable_interim_results"],
),
)
self._streaming_task = self.create_task(self._stream_audio())
async def _disconnect(self):
"""Clean up streaming recognition resources."""
if self._streaming_task:
logger.debug("Disconnecting from Google Speech-to-Text")
# Send sentinel value to stop request generator
await self._request_queue.put(None)
await self.cancel_task(self._streaming_task)
self._streaming_task = None
# Clear any remaining items in the queue
while not self._request_queue.empty():
try:
self._request_queue.get_nowait()
self._request_queue.task_done()
except asyncio.QueueEmpty:
break
async def _request_generator(self):
"""Generates requests for the streaming recognize method."""
recognizer_path = f"projects/{self._project_id}/locations/{self._location}/recognizers/_"
logger.trace(f"Using recognizer path: {recognizer_path}")
try:
# Send initial config
yield cloud_speech.StreamingRecognizeRequest(
recognizer=recognizer_path,
streaming_config=self._config,
)
while True:
try:
audio_data = await self._request_queue.get()
if audio_data is None: # Sentinel value to stop
break
# Check streaming limit
if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT:
logger.debug("Streaming limit reached, initiating graceful reconnection")
# Instead of immediate reconnection, we'll break and let the stream close naturally
self._last_audio_input = self._audio_input
self._audio_input = []
self._restart_counter += 1
# Put the current audio chunk back in the queue
await self._request_queue.put(audio_data)
break
self._audio_input.append(audio_data)
yield cloud_speech.StreamingRecognizeRequest(audio=audio_data)
except asyncio.CancelledError:
break
finally:
self._request_queue.task_done()
except Exception as e:
logger.error(f"Error in request generator: {e}")
raise
async def _stream_audio(self):
"""Handle bi-directional streaming with Google STT."""
try:
while True:
try:
# Start bi-directional streaming
streaming_recognize = await self._client.streaming_recognize(
requests=self._request_generator()
)
# Process responses
await self._process_responses(streaming_recognize)
# If we're here, check if we need to reconnect
if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT:
logger.debug("Reconnecting stream after timeout")
# Reset stream start time
self._stream_start_time = int(time.time() * 1000)
continue
else:
# Normal stream end
break
except Exception as e:
logger.error(f"Stream error, attempting to reconnect: {e}")
await asyncio.sleep(1) # Brief delay before reconnecting
self._stream_start_time = int(time.time() * 1000)
continue
except Exception as e:
logger.error(f"Error in streaming task: {e}")
await self.push_frame(ErrorFrame(str(e)))
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Process an audio chunk for STT transcription."""
if self._streaming_task:
# Queue the audio data
await self._request_queue.put(audio)
yield None
async def _process_responses(self, streaming_recognize):
"""Process streaming recognition responses."""
try:
async for response in streaming_recognize:
# Check streaming limit
if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT:
logger.debug("Stream timeout reached in response processing")
break
if not response.results:
continue
for result in response.results:
if not result.alternatives:
continue
transcript = result.alternatives[0].transcript
if not transcript:
continue
primary_language = self._settings["language_codes"][0]
if result.is_final:
self._last_transcript_was_final = True
await self.push_frame(
TranscriptionFrame(transcript, "", time_now_iso8601(), primary_language)
)
else:
self._last_transcript_was_final = False
await self.push_frame(
InterimTranscriptionFrame(
transcript, "", time_now_iso8601(), primary_language
)
)
except Exception as e:
logger.error(f"Error processing Google STT responses: {e}")

View File

@@ -17,7 +17,6 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.openai import (
OpenAIAssistantContextAggregator,
OpenAILLMService,
@@ -28,7 +27,7 @@ from pipecat.services.openai import (
class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
"""Custom assistant context aggregator for Grok that handles empty content requirement."""
async def push_aggregation(self):
async def _push_aggregation(self):
if not (
self._aggregation or self._function_call_result or self._pending_image_frame_message
):
@@ -37,13 +36,10 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
run_llm = False
properties: Optional[FunctionCallResultProperties] = None
aggregation = self._aggregation.strip()
self.reset()
aggregation = self._aggregation
self._reset()
try:
if aggregation:
self._context.add_message({"role": "assistant", "content": aggregation})
if self._function_call_result:
frame = self._function_call_result
properties = frame.properties
@@ -80,6 +76,9 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
# Default behavior is to run the LLM if there are no function calls in progress
run_llm = not bool(self._function_calls_in_progress)
else:
self._context.add_message({"role": "assistant", "content": aggregation})
if self._pending_image_frame_message:
frame = self._pending_image_frame_message
self._pending_image_frame_message = None
@@ -92,13 +91,14 @@ class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
run_llm = True
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
await self._user_context_aggregator.push_context_frame()
# Emit the on_context_updated callback once the function call result is added to the context
if properties and properties.on_context_updated is not None:
await properties.on_context_updated()
await self.push_context_frame()
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
except Exception as e:
logger.error(f"Error processing frame: {e}")
@@ -212,6 +212,6 @@ class GrokLLMService(OpenAILLMService):
) -> GrokContextAggregatorPair:
user = OpenAIUserContextAggregator(context)
assistant = GrokAssistantContextAggregator(
context, expect_stripped_words=assistant_expect_stripped_words
user, expect_stripped_words=assistant_expect_stripped_words
)
return GrokContextAggregatorPair(_user=user, _assistant=assistant)

View File

@@ -54,8 +54,6 @@ class GroqSTTService(BaseWhisperSTTService):
api_key: Groq API key. Defaults to None.
base_url: API base URL. Defaults to "https://api.groq.com/openai/v1".
language: Language of the audio input. Defaults to English.
prompt: Optional text to guide the model's style or continue a previous segment.
temperature: Optional sampling temperature between 0 and 1. Defaults to 0.0.
**kwargs: Additional arguments passed to BaseWhisperSTTService.
"""
@@ -66,35 +64,17 @@ class GroqSTTService(BaseWhisperSTTService):
api_key: Optional[str] = None,
base_url: str = "https://api.groq.com/openai/v1",
language: Optional[Language] = Language.EN,
prompt: Optional[str] = None,
temperature: Optional[float] = None,
**kwargs,
):
super().__init__(
model=model,
api_key=api_key,
base_url=base_url,
language=language,
prompt=prompt,
temperature=temperature,
**kwargs,
model=model, api_key=api_key, base_url=base_url, language=language, **kwargs
)
async def _transcribe(self, audio: bytes) -> Transcription:
assert self._language is not None # Assigned in the BaseWhisperSTTService class
# Build kwargs dict with only set parameters
kwargs = {
"file": ("audio.wav", audio, "audio/wav"),
"model": self.model_name,
"response_format": "json",
"language": self._language,
}
if self._prompt is not None:
kwargs["prompt"] = self._prompt
if self._temperature is not None:
kwargs["temperature"] = self._temperature
return await self._client.audio.transcriptions.create(**kwargs)
return await self._client.audio.transcriptions.create(
file=("audio.wav", audio, "audio/wav"),
model=self.model_name,
response_format="json",
language=self._language,
)

View File

@@ -73,7 +73,6 @@ class LmntTTSService(TTSService, WebsocketService):
TTSService.__init__(
self,
push_stop_frames=True,
pause_frame_processing=True,
sample_rate=sample_rate,
**kwargs,
)

View File

@@ -15,7 +15,6 @@ import httpx
from loguru import logger
from PIL import Image
from pydantic import BaseModel, Field
from tenacity import retry, stop_after_attempt, wait_fixed
from pipecat.frames.frames import (
ErrorFrame,
@@ -111,7 +110,7 @@ class BaseOpenAILLMService(LLMService):
seed: Optional[int] = Field(default_factory=lambda: NOT_GIVEN, ge=0)
temperature: Optional[float] = Field(default_factory=lambda: NOT_GIVEN, ge=0.0, le=2.0)
# Note: top_k is currently not supported by the OpenAI client library,
# so top_k is ignored right now.
# so top_k is ignore right now.
top_k: Optional[int] = Field(default=None, ge=0)
top_p: Optional[float] = Field(default_factory=lambda: NOT_GIVEN, ge=0.0, le=1.0)
max_tokens: Optional[int] = Field(default_factory=lambda: NOT_GIVEN, ge=1)
@@ -161,7 +160,6 @@ class BaseOpenAILLMService(LLMService):
def can_generate_metrics(self) -> bool:
return True
@retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
async def get_chat_completions(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> AsyncStream[ChatCompletionChunk]:
@@ -268,6 +266,7 @@ class BaseOpenAILLMService(LLMService):
if tool_call.function and tool_call.function.name:
function_name += tool_call.function.name
tool_call_id = tool_call.id
await self.call_start_function(context, function_name)
if tool_call.function and tool_call.function.arguments:
# Keep iterating through the response to collect all the argument fragments
arguments += tool_call.function.arguments
@@ -356,7 +355,7 @@ class OpenAILLMService(BaseOpenAILLMService):
) -> OpenAIContextAggregatorPair:
user = OpenAIUserContextAggregator(context)
assistant = OpenAIAssistantContextAggregator(
context, expect_stripped_words=assistant_expect_stripped_words
user, expect_stripped_words=assistant_expect_stripped_words
)
return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)
@@ -409,8 +408,6 @@ class OpenAISTTService(BaseWhisperSTTService):
api_key: OpenAI API key. Defaults to None.
base_url: API base URL. Defaults to None.
language: Language of the audio input. Defaults to English.
prompt: Optional text to guide the model's style or continue a previous segment.
temperature: Optional sampling temperature between 0 and 1. Defaults to 0.0.
**kwargs: Additional arguments passed to BaseWhisperSTTService.
"""
@@ -421,37 +418,17 @@ class OpenAISTTService(BaseWhisperSTTService):
api_key: Optional[str] = None,
base_url: Optional[str] = None,
language: Optional[Language] = Language.EN,
prompt: Optional[str] = None,
temperature: Optional[float] = None,
**kwargs,
):
super().__init__(
model=model,
api_key=api_key,
base_url=base_url,
language=language,
prompt=prompt,
temperature=temperature,
**kwargs,
model=model, api_key=api_key, base_url=base_url, language=language, **kwargs
)
async def _transcribe(self, audio: bytes) -> Transcription:
assert self._language is not None # Assigned in the BaseWhisperSTTService class
# Build kwargs dict with only set parameters
kwargs = {
"file": ("audio.wav", audio, "audio/wav"),
"model": self.model_name,
"language": self._language,
}
if self._prompt is not None:
kwargs["prompt"] = self._prompt
if self._temperature is not None:
kwargs["temperature"] = self._temperature
return await self._client.audio.transcriptions.create(**kwargs)
return await self._client.audio.transcriptions.create(
file=("audio.wav", audio, "audio/wav"), model=self.model_name, language=self._language
)
class OpenAITTSService(TTSService):
@@ -556,8 +533,8 @@ class OpenAIImageMessageFrame(Frame):
class OpenAIUserContextAggregator(LLMUserContextAggregator):
def __init__(self, context: OpenAILLMContext, **kwargs):
super().__init__(context=context, **kwargs)
def __init__(self, context: OpenAILLMContext):
super().__init__(context=context)
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
@@ -593,8 +570,9 @@ class OpenAIUserContextAggregator(LLMUserContextAggregator):
class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
def __init__(self, context: OpenAILLMContext, **kwargs):
super().__init__(context=context, **kwargs)
def __init__(self, user_context_aggregator: OpenAIUserContextAggregator, **kwargs):
super().__init__(context=user_context_aggregator._context, **kwargs)
self._user_context_aggregator = user_context_aggregator
self._function_calls_in_progress = {}
self._function_call_result = None
self._pending_image_frame_message = None
@@ -614,7 +592,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
del self._function_calls_in_progress[frame.tool_call_id]
self._function_call_result = frame
# TODO-CB: Kwin wants us to refactor this out of here but I REFUSE
await self.push_aggregation()
await self._push_aggregation()
else:
logger.warning(
"FunctionCallResultFrame tool_call_id does not match any function call in progress"
@@ -622,9 +600,9 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
self._function_call_result = None
elif isinstance(frame, OpenAIImageMessageFrame):
self._pending_image_frame_message = frame
await self.push_aggregation()
await self._push_aggregation()
async def push_aggregation(self):
async def _push_aggregation(self):
if not (
self._aggregation or self._function_call_result or self._pending_image_frame_message
):
@@ -633,13 +611,10 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
run_llm = False
properties: Optional[FunctionCallResultProperties] = None
aggregation = self._aggregation.strip()
self.reset()
aggregation = self._aggregation
self._reset()
try:
if aggregation:
self._context.add_message({"role": "assistant", "content": aggregation})
if self._function_call_result:
frame = self._function_call_result
properties = frame.properties
@@ -674,6 +649,9 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
# Default behavior is to run the LLM if there are no function calls in progress
run_llm = not bool(self._function_calls_in_progress)
else:
self._context.add_message({"role": "assistant", "content": aggregation})
if self._pending_image_frame_message:
frame = self._pending_image_frame_message
self._pending_image_frame_message = None
@@ -686,14 +664,15 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
run_llm = True
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
await self._user_context_aggregator.push_context_frame()
# Emit the on_context_updated callback once the function call result is added to the context
if properties and properties.on_context_updated is not None:
await properties.on_context_updated()
# Push context frame
await self.push_context_frame()
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
# Push timestamp frame with current time
timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())

View File

@@ -166,7 +166,7 @@ class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator):
if isinstance(frame, LLMSetToolsFrame):
await self.push_frame(frame, direction)
async def push_aggregation(self):
async def _push_aggregation(self):
# for the moment, ignore all user input coming into the pipeline.
# todo: think about whether/how to fix this to allow for text input from
# upstream (transport/transcription, or other sources)
@@ -174,7 +174,7 @@ class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator):
class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator):
async def push_aggregation(self):
async def _push_aggregation(self):
# the only thing we implement here is function calling. in all other cases, messages
# are added to the context when we receive openai realtime api events
if not self._function_call_result:
@@ -182,7 +182,7 @@ class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator)
properties: Optional[FunctionCallResultProperties] = None
self.reset()
self._reset()
try:
run_llm = True
frame = self._function_call_result
@@ -217,8 +217,8 @@ class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator)
# The standard function callback code path pushes the FunctionCallResultFrame from the llm itself,
# so we didn't have a chance to add the result to the openai realtime api context. Let's push a
# special frame to do that.
await self.push_frame(
RealtimeFunctionCallResultFrame(result_frame=frame), FrameDirection.UPSTREAM
await self._user_context_aggregator.push_frame(
RealtimeFunctionCallResultFrame(result_frame=frame)
)
if properties and properties.run_llm is not None:
# If the tool call result has a run_llm property, use it
@@ -228,13 +228,14 @@ class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator)
run_llm = not bool(self._function_calls_in_progress)
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
await self._user_context_aggregator.push_context_frame()
# Emit the on_context_updated callback once the function call result is added to the context
if properties and properties.on_context_updated is not None:
await properties.on_context_updated()
await self.push_context_frame()
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
except Exception as e:
logger.error(f"Error processing frame: {e}")

View File

@@ -568,6 +568,6 @@ class OpenAIRealtimeBetaLLMService(LLMService):
OpenAIRealtimeLLMContext.upgrade_to_realtime(context)
user = OpenAIRealtimeUserContextAggregator(context)
assistant = OpenAIRealtimeAssistantContextAggregator(
context, expect_stripped_words=assistant_expect_stripped_words
user, expect_stripped_words=assistant_expect_stripped_words
)
return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)

View File

@@ -7,7 +7,6 @@
from typing import Dict, List, Optional
from loguru import logger
from tenacity import retry, stop_after_attempt, wait_fixed
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai import OpenAILLMService
@@ -56,7 +55,6 @@ class OpenPipeLLMService(OpenAILLMService):
)
return client
@retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
async def get_chat_completions(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> AsyncStream[ChatCompletionChunk]:

View File

@@ -7,7 +7,6 @@
from typing import List
from loguru import logger
from tenacity import retry, stop_after_attempt, wait_fixed
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
@@ -57,7 +56,6 @@ class PerplexityLLMService(OpenAILLMService):
self._has_reported_prompt_tokens = False
self._is_processing = False
@retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
async def get_chat_completions(
self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
) -> AsyncStream[ChatCompletionChunk]:

View File

@@ -120,7 +120,6 @@ class PlayHTTTSService(TTSService, WebsocketService):
):
TTSService.__init__(
self,
pause_frame_processing=True,
sample_rate=sample_rate,
**kwargs,
)
@@ -270,6 +269,19 @@ class PlayHTTTSService(TTSService, WebsocketService):
except json.JSONDecodeError:
logger.error(f"Invalid JSON message: {message}")
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
# If we received a TTSSpeakFrame and the LLM response included text (it
# might be that it's only a function calling response) we pause
# processing more frames until we receive a BotStoppedSpeakingFrame.
if isinstance(frame, TTSSpeakFrame):
await self.pause_processing_frames()
elif isinstance(frame, LLMFullResponseEndFrame) and self._request_id:
await self.pause_processing_frames()
elif isinstance(frame, BotStoppedSpeakingFrame):
await self.resume_processing_frames()
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating TTS: [{text}]")
@@ -397,7 +409,6 @@ class PlayHTHttpTTSService(TTSService):
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
async for chunk in playht_gen:
# skip the RIFF header.
if in_header:
@@ -417,8 +428,6 @@ class PlayHTHttpTTSService(TTSService):
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
yield frame
yield TTSStoppedFrame()
except Exception as e:
logger.error(f"{self} error generating TTS: {e}")
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()

View File

@@ -4,9 +4,6 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import base64
import json
import uuid
from typing import AsyncGenerator, Optional
import aiohttp
@@ -14,323 +11,13 @@ from loguru import logger
from pydantic import BaseModel
from pipecat.frames.frames import (
CancelFrame,
EndFrame,
ErrorFrame,
Frame,
StartFrame,
StartInterruptionFrame,
TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import AudioContextWordTTSService, TTSService
from pipecat.services.websocket_service import WebsocketService
from pipecat.transcriptions.language import Language
try:
import websockets
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use Rime, you need to `pip install pipecat-ai[rime]`. Also, set `RIME_API_KEY` environment variable."
)
raise Exception(f"Missing module: {e}")
def language_to_rime_language(language: Language) -> str:
"""Convert pipecat Language to Rime language code.
Args:
language: The pipecat Language enum value.
Returns:
str: Three-letter language code used by Rime (e.g., 'eng' for English).
"""
LANGUAGE_MAP = {
Language.EN: "eng",
Language.ES: "spa",
}
return LANGUAGE_MAP.get(language, "eng")
class RimeTTSService(AudioContextWordTTSService, WebsocketService):
"""Text-to-Speech service using Rime's websocket API.
Uses Rime's websocket JSON API to convert text to speech with word-level timing
information. Supports interruptions and maintains context across multiple messages
within a turn.
"""
class InputParams(BaseModel):
"""Configuration parameters for Rime TTS service."""
language: Optional[Language] = Language.EN
speed_alpha: Optional[float] = 1.0
reduce_latency: Optional[bool] = False
def __init__(
self,
*,
api_key: str,
voice_id: str,
url: str = "wss://users-ws.rime.ai/ws2",
model: str = "mistv2",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
"""Initialize Rime TTS service.
Args:
api_key: Rime API key for authentication.
voice_id: ID of the voice to use.
url: Rime websocket API endpoint.
model: Model ID to use for synthesis.
sample_rate: Audio sample rate in Hz.
params: Additional configuration parameters.
"""
# Initialize with parent class settings for proper frame handling
AudioContextWordTTSService.__init__(
self,
aggregate_sentences=True,
push_text_frames=False,
push_stop_frames=True,
pause_frame_processing=True,
sample_rate=sample_rate,
**kwargs,
)
WebsocketService.__init__(self)
# Store service configuration
self._api_key = api_key
self._url = url
self._voice_id = voice_id
self._model = model
self._settings = {
"speaker": voice_id,
"modelId": model,
"audioFormat": "pcm",
"samplingRate": 0,
"lang": self.language_to_service_language(params.language)
if params.language
else "eng",
"speedAlpha": params.speed_alpha,
"reduceLatency": params.reduce_latency,
}
# State tracking
self._context_id = None # Tracks current turn
self._receive_task = None
self._cumulative_time = 0 # Accumulates time across messages
def can_generate_metrics(self) -> bool:
return True
def language_to_service_language(self, language: Language) -> str | None:
"""Convert pipecat language to Rime language code."""
return language_to_rime_language(language)
async def set_model(self, model: str):
"""Update the TTS model."""
self._model = model
await super().set_model(model)
def _build_msg(self, text: str = "") -> dict:
"""Build JSON message for Rime API."""
return {"text": text, "contextId": self._context_id}
def _build_clear_msg(self) -> dict:
"""Build clear operation message."""
return {"operation": "clear"}
def _build_eos_msg(self) -> dict:
"""Build end-of-stream operation message."""
return {"operation": "eos"}
async def start(self, frame: StartFrame):
"""Start the service and establish websocket connection."""
await super().start(frame)
self._settings["samplingRate"] = self.sample_rate
await self._connect()
async def stop(self, frame: EndFrame):
"""Stop the service and close connection."""
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
"""Cancel current operation and clean up."""
await super().cancel(frame)
await self._disconnect()
async def _connect(self):
"""Establish websocket connection and start receive task."""
await self._connect_websocket()
self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
async def _disconnect(self):
"""Close websocket connection and clean up tasks."""
await self._disconnect_websocket()
if self._receive_task:
await self.cancel_task(self._receive_task)
self._receive_task = None
async def _connect_websocket(self):
"""Connect to Rime websocket API with configured settings."""
try:
params = "&".join(f"{k}={v}" for k, v in self._settings.items())
url = f"{self._url}?{params}"
headers = {"Authorization": f"Bearer {self._api_key}"}
self._websocket = await websockets.connect(url, extra_headers=headers)
except Exception as e:
logger.error(f"{self} initialization error: {e}")
self._websocket = None
async def _disconnect_websocket(self):
"""Close websocket connection and reset state."""
try:
await self.stop_all_metrics()
if self._websocket:
await self._websocket.send(json.dumps(self._build_eos_msg()))
await self._websocket.close()
self._websocket = None
self._context_id = None
except Exception as e:
logger.error(f"{self} error closing websocket: {e}")
def _get_websocket(self):
"""Get active websocket connection or raise exception."""
if self._websocket:
return self._websocket
raise Exception("Websocket not connected")
async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
"""Handle interruption by clearing current context."""
await super()._handle_interruption(frame, direction)
await self.stop_all_metrics()
if self._context_id:
await self._get_websocket().send(json.dumps(self._build_clear_msg()))
self._context_id = None
def _calculate_word_times(self, words: list, starts: list, ends: list) -> list:
"""Calculate word timing pairs with proper spacing and punctuation.
Args:
words: List of words from Rime.
starts: List of start times for each word.
ends: List of end times for each word.
Returns:
List of (word, timestamp) pairs with proper timing.
"""
word_pairs = []
for i, (word, start_time, _) in enumerate(zip(words, starts, ends)):
if not word.strip():
continue
# Adjust timing by adding cumulative time
adjusted_start = start_time + self._cumulative_time
# Handle punctuation by appending to previous word
is_punctuation = bool(word.strip(",.!?") == "")
if is_punctuation and word_pairs:
prev_word, prev_time = word_pairs[-1]
word_pairs[-1] = (prev_word + word, prev_time)
else:
word_pairs.append((word, adjusted_start))
return word_pairs
async def flush_audio(self):
if not self._context_id or not self._websocket:
return
logger.trace(f"{self}: flushing audio")
self._context_id = None
async def _receive_messages(self):
"""Process incoming websocket messages."""
async for message in self._get_websocket():
msg = json.loads(message)
if not msg or not self.audio_context_available(msg["contextId"]):
continue
if msg["type"] == "chunk":
# Process audio chunk
await self.stop_ttfb_metrics()
self.start_word_timestamps()
frame = TTSAudioRawFrame(
audio=base64.b64decode(msg["data"]),
sample_rate=self.sample_rate,
num_channels=1,
)
await self.append_to_audio_context(msg["contextId"], frame)
elif msg["type"] == "timestamps":
# Process word timing information
timestamps = msg.get("word_timestamps", {})
words = timestamps.get("words", [])
starts = timestamps.get("start", [])
ends = timestamps.get("end", [])
if words and starts:
# Calculate word timing pairs
word_pairs = self._calculate_word_times(words, starts, ends)
if word_pairs:
await self.add_word_timestamps(word_pairs)
self._cumulative_time = ends[-1] + self._cumulative_time
logger.debug(f"Updated cumulative time to: {self._cumulative_time}")
elif msg["type"] == "error":
logger.error(f"{self} error: {msg}")
await self.push_frame(TTSStoppedFrame())
await self.stop_all_metrics()
await self.push_error(ErrorFrame(f"{self} error: {msg['message']}"))
self._context_id = None
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
"""Push frame and handle end-of-turn conditions."""
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)):
if isinstance(frame, TTSStoppedFrame):
await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)])
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text.
Args:
text: The text to convert to speech.
Yields:
Frames containing audio data and timing information.
"""
logger.debug(f"Generating TTS: [{text}]")
try:
if not self._websocket:
await self._connect()
try:
if not self._context_id:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._cumulative_time = 0
self._context_id = str(uuid.uuid4())
await self.create_audio_context(self._context_id)
msg = self._build_msg(text=text)
await self._get_websocket().send(json.dumps(msg))
await self.start_tts_usage_metrics(text)
except Exception as e:
logger.error(f"{self} error sending message: {e}")
yield TTSStoppedFrame()
await self._disconnect()
await self._connect()
return
yield None
except Exception as e:
logger.error(f"{self} exception: {e}")
from pipecat.services.ai_services import TTSService
class RimeHttpTTSService(TTSService):
@@ -345,9 +32,8 @@ class RimeHttpTTSService(TTSService):
self,
*,
api_key: str,
voice_id: str,
aiohttp_session: aiohttp.ClientSession,
model: str = "mistv2",
voice_id: str = "eva",
model: str = "mist",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
@@ -355,7 +41,6 @@ class RimeHttpTTSService(TTSService):
super().__init__(sample_rate=sample_rate, **kwargs)
self._api_key = api_key
self._session = aiohttp_session
self._base_url = "https://users.rime.ai/v1/rime-tts"
self._settings = {
"speedAlpha": params.speed_alpha,
@@ -389,31 +74,36 @@ class RimeHttpTTSService(TTSService):
try:
await self.start_ttfb_metrics()
await self.start_tts_usage_metrics(text)
async with self._session.post(
self._base_url, json=payload, headers=headers
) as response:
if response.status != 200:
error_message = f"Rime TTS error: HTTP {response.status}"
logger.error(error_message)
yield ErrorFrame(error=error_message)
return
yield TTSStartedFrame()
await self.start_tts_usage_metrics(text)
async with aiohttp.ClientSession() as session:
async with session.post(self._base_url, json=payload, headers=headers) as response:
if response.status != 200:
error_message = f"Rime TTS error: HTTP {response.status}"
logger.error(error_message)
yield ErrorFrame(error=error_message)
return
yield TTSStartedFrame()
# Process the streaming response
chunk_size = 8192
first_chunk = True
# Process the streaming response
chunk_size = 8192
async for chunk in response.content.iter_chunked(chunk_size):
if first_chunk:
await self.stop_ttfb_metrics()
first_chunk = False
if chunk:
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
yield frame
yield TTSStoppedFrame()
async for chunk in response.content.iter_chunked(chunk_size):
if chunk:
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
yield frame
except Exception as e:
logger.exception(f"Error generating TTS: {e}")
yield ErrorFrame(error=f"Rime TTS error: {str(e)}")
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()

View File

@@ -41,7 +41,7 @@ class SimliVideoService(FrameProcessor):
self._pipecat_resampler_event = asyncio.Event()
self._pipecat_resampler: AudioResampler = None
self._simli_resampler = AudioResampler("s16", "mono", 16000)
self._simli_resampler = AudioResampler("s16", 1, 16000)
self._audio_task: asyncio.Task = None
self._video_task: asyncio.Task = None

View File

@@ -10,10 +10,8 @@ from typing import Awaitable, Callable, Optional
import websockets
from loguru import logger
from websockets.protocol import State
from pipecat.frames.frames import ErrorFrame
from pipecat.utils.network import exponential_backoff_time
class WebsocketService(ABC):
@@ -52,6 +50,27 @@ class WebsocketService(ABC):
await self._connect_websocket()
return await self._verify_connection()
def _calculate_wait_time(
self, attempt: int, min_wait: float = 4, max_wait: float = 10, multiplier: float = 1
) -> float:
"""Calculate exponential backoff wait time.
Args:
attempt: Current attempt number (1-based)
min_wait: Minimum wait time in seconds
max_wait: Maximum wait time in seconds
multiplier: Base multiplier for exponential calculation
Returns:
Wait time in seconds
"""
try:
exp = 2 ** (attempt - 1) * multiplier
result = max(0, min(exp, max_wait))
return max(min_wait, result)
except (ValueError, ArithmeticError):
return max_wait
async def _receive_task_handler(self, report_error: Callable[[ErrorFrame], Awaitable[None]]):
"""Handles WebSocket message receiving with automatic retry logic.
@@ -64,13 +83,11 @@ class WebsocketService(ABC):
while True:
try:
await self._receive_messages()
logger.debug(f"{self} connection established successfully")
retry_count = 0 # Reset counter on successful message receive
if self._websocket and self._websocket.state == State.CLOSED:
raise websockets.ConnectionClosedOK(
self._websocket.close_rcvd,
self._websocket.close_sent,
self._websocket.close_rcvd_then_sent,
)
except websockets.ConnectionClosed as e:
logger.warning(f"{self} connection closed: {e}")
break
except Exception as e:
retry_count += 1
if retry_count >= MAX_RETRIES:
@@ -84,7 +101,7 @@ class WebsocketService(ABC):
try:
if await self._reconnect_websocket(retry_count):
retry_count = 0 # Reset counter on successful reconnection
wait_time = exponential_backoff_time(retry_count)
wait_time = self._calculate_wait_time(retry_count)
await asyncio.sleep(wait_time)
except Exception as reconnect_error:
logger.error(f"{self} reconnection failed: {reconnect_error}")

View File

@@ -5,7 +5,6 @@
#
import asyncio
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, Dict, Sequence, Tuple
from pipecat.frames.frames import (
@@ -13,7 +12,6 @@ from pipecat.frames.frames import (
Frame,
HeartbeatFrame,
StartFrame,
SystemFrame,
)
from pipecat.observers.base_observer import BaseObserver
from pipecat.pipeline.pipeline import Pipeline
@@ -22,16 +20,6 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@dataclass
class SleepFrame(SystemFrame):
"""This frame is used by test framework to introduce some sleep time before
the next frame is pushed. This is useful to control system frames vs data or
control frames.
"""
sleep: float = 0.1
class HeartbeatsObserver(BaseObserver):
def __init__(
self,
@@ -56,11 +44,7 @@ class HeartbeatsObserver(BaseObserver):
class QueuedFrameProcessor(FrameProcessor):
def __init__(
self,
*,
queue: asyncio.Queue,
queue_direction: FrameDirection,
ignore_start: bool = True,
self, queue: asyncio.Queue, queue_direction: FrameDirection, ignore_start: bool = True
):
super().__init__()
self._queue = queue
@@ -88,35 +72,21 @@ async def run_test(
) -> Tuple[Sequence[Frame], Sequence[Frame]]:
received_up = asyncio.Queue()
received_down = asyncio.Queue()
source = QueuedFrameProcessor(
queue=received_up,
queue_direction=FrameDirection.UPSTREAM,
ignore_start=ignore_start,
)
sink = QueuedFrameProcessor(
queue=received_down,
queue_direction=FrameDirection.DOWNSTREAM,
ignore_start=ignore_start,
)
source = QueuedFrameProcessor(received_up, FrameDirection.UPSTREAM, ignore_start)
sink = QueuedFrameProcessor(received_down, FrameDirection.DOWNSTREAM, ignore_start)
pipeline = Pipeline([source, processor, sink])
task = PipelineTask(pipeline, params=PipelineParams(start_metadata=start_metadata))
async def push_frames():
# Just give a little head start to the runner.
await asyncio.sleep(0.01)
for frame in frames_to_send:
if isinstance(frame, SleepFrame):
await asyncio.sleep(frame.sleep)
else:
await task.queue_frame(frame)
for frame in frames_to_send:
await task.queue_frame(frame)
if send_end_frame:
await task.queue_frame(EndFrame())
if send_end_frame:
await task.queue_frame(EndFrame())
runner = PipelineRunner()
await asyncio.gather(runner.run(task), push_frames())
await runner.run(task)
#
# Down frames
@@ -128,7 +98,6 @@ async def run_test(
received_down_frames.append(frame)
print("received DOWN frames =", received_down_frames)
print("expected DOWN frames =", expected_down_frames)
assert len(received_down_frames) == len(expected_down_frames)
@@ -144,7 +113,6 @@ async def run_test(
received_up_frames.append(frame)
print("received UP frames =", received_up_frames)
print("expected UP frames =", expected_up_frames)
assert len(received_up_frames) == len(expected_up_frames)

View File

@@ -101,7 +101,6 @@ class Language(StrEnum):
EN_AU = "en-AU"
EN_CA = "en-CA"
EN_GB = "en-GB"
EN_GH = "en-GH"
EN_HK = "en-HK"
EN_IE = "en-IE"
EN_IN = "en-IN"
@@ -209,7 +208,6 @@ class Language(StrEnum):
# Italian
IT = "it"
IT_IT = "it-IT"
IT_CH = "it-CH"
# Inuktitut
IU_CANS = "iu-Cans"

View File

@@ -14,8 +14,6 @@ from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState
from pipecat.frames.frames import (
BotInterruptionFrame,
CancelFrame,
EmulateUserStartedSpeakingFrame,
EmulateUserStoppedSpeakingFrame,
EndFrame,
FilterUpdateSettingsFrame,
Frame,
@@ -49,13 +47,6 @@ class BaseInputTransport(FrameProcessor):
# if passthrough is enabled.
self._audio_task = None
def enable_audio_in_stream_on_start(self, enabled: bool) -> None:
logger.debug(f"Enabling audio on start. {enabled}")
self._params.audio_in_stream_on_start = enabled
def start_audio_in_streaming(self):
pass
@property
def sample_rate(self) -> int:
return self._sample_rate
@@ -114,13 +105,9 @@ class BaseInputTransport(FrameProcessor):
await self.cancel(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, BotInterruptionFrame):
await self._handle_bot_interruption(frame)
elif isinstance(frame, EmulateUserStartedSpeakingFrame):
logger.debug("Emulating user started speaking")
await self._handle_user_interruption(UserStartedSpeakingFrame())
elif isinstance(frame, EmulateUserStoppedSpeakingFrame):
logger.debug("Emulating user stopped speaking")
await self._handle_user_interruption(UserStoppedSpeakingFrame())
logger.debug("Bot interruption")
await self._start_interruption()
await self.push_frame(StartInterruptionFrame())
# All other system frames
elif isinstance(frame, SystemFrame):
await self.push_frame(frame, direction)
@@ -143,13 +130,7 @@ class BaseInputTransport(FrameProcessor):
# Handle interruptions
#
async def _handle_bot_interruption(self, frame: BotInterruptionFrame):
logger.debug("Bot interruption")
if self.interruptions_allowed:
await self._start_interruption()
await self.push_frame(StartInterruptionFrame())
async def _handle_user_interruption(self, frame: Frame):
async def _handle_interruptions(self, frame: Frame):
if isinstance(frame, UserStartedSpeakingFrame):
logger.debug("User started speaking")
# Make sure we notify about interruptions quickly out-of-band.
@@ -174,9 +155,11 @@ class BaseInputTransport(FrameProcessor):
async def _vad_analyze(self, audio_frame: InputAudioRawFrame) -> VADState:
state = VADState.QUIET
if self.vad_analyzer:
logger.trace(f"{self}: analyzing VAD on {audio_frame}")
state = await self.get_event_loop().run_in_executor(
self._executor, self.vad_analyzer.analyze_audio, audio_frame.audio
)
logger.trace(f"{self}: done analyzing VAD on {audio_frame}")
return state
async def _handle_vad(self, audio_frame: InputAudioRawFrame, vad_state: VADState):
@@ -193,7 +176,7 @@ class BaseInputTransport(FrameProcessor):
frame = UserStoppedSpeakingFrame()
if frame:
await self._handle_user_interruption(frame)
await self._handle_interruptions(frame)
vad_state = new_vad_state
return vad_state

View File

@@ -14,6 +14,7 @@ from loguru import logger
from PIL import Image
from pipecat.audio.utils import create_default_resampler
from pipecat.audio.vad.vad_analyzer import VAD_STOP_SECS
from pipecat.frames.frames import (
BotSpeakingFrame,
BotStartedSpeakingFrame,
@@ -37,8 +38,6 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.transports.base_transport import TransportParams
from pipecat.utils.time import nanoseconds_to_seconds
BOT_VAD_STOP_SECS = 0.3
class BaseOutputTransport(FrameProcessor):
def __init__(self, params: TransportParams, **kwargs):
@@ -170,8 +169,6 @@ class BaseOutputTransport(FrameProcessor):
# TODO(aleix): Images and audio should support presentation timestamps.
elif frame.pts:
await self._sink_clock_queue.put((frame.pts, frame.id, frame))
elif direction == FrameDirection.UPSTREAM:
await self.push_frame(frame, direction)
else:
await self._sink_queue.put(frame)
@@ -324,10 +321,15 @@ class BaseOutputTransport(FrameProcessor):
)
yield frame
vad_stop_secs = (
self._params.vad_analyzer.params.stop_secs
if self._params.vad_analyzer
else VAD_STOP_SECS
)
if self._params.audio_out_mixer:
return with_mixer(BOT_VAD_STOP_SECS)
return with_mixer(vad_stop_secs)
else:
return without_mixer(BOT_VAD_STOP_SECS)
return without_mixer(vad_stop_secs)
async def _sink_task_handler(self):
async for frame in self._next_frame():

View File

@@ -4,6 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import inspect
from abc import ABC, abstractmethod
from typing import Optional
@@ -29,6 +30,7 @@ class TransportParams(BaseModel):
camera_out_framerate: int = 30
camera_out_color_format: str = "RGB"
audio_out_enabled: bool = False
audio_out_is_live: bool = False
audio_out_sample_rate: Optional[int] = None
audio_out_channels: int = 1
audio_out_bitrate: int = 96000
@@ -37,7 +39,6 @@ class TransportParams(BaseModel):
audio_in_sample_rate: Optional[int] = None
audio_in_channels: int = 1
audio_in_filter: Optional[BaseAudioFilter] = None
audio_in_stream_on_start: bool = True
vad_enabled: bool = False
vad_audio_passthrough: bool = False
vad_analyzer: Optional[VADAnalyzer] = None

View File

@@ -26,18 +26,17 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
class LocalAudioTransportParams(TransportParams):
input_device_index: Optional[int] = None
output_device_index: Optional[int] = None
class LocalTransportParams(TransportParams):
input_device_index: int = 0
output_device_index: int = 0
class LocalAudioInputTransport(BaseInputTransport):
_params: LocalAudioTransportParams
_params: LocalTransportParams
def __init__(self, py_audio: pyaudio.PyAudio, params: LocalAudioTransportParams):
def __init__(self, py_audio: pyaudio.PyAudio, params: LocalTransportParams):
super().__init__(params)
self._py_audio = py_audio
self._in_stream = None
self._sample_rate = 0
@@ -78,12 +77,11 @@ class LocalAudioInputTransport(BaseInputTransport):
class LocalAudioOutputTransport(BaseOutputTransport):
_params: LocalAudioTransportParams
_params: LocalTransportParams
def __init__(self, py_audio: pyaudio.PyAudio, params: LocalAudioTransportParams):
def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
super().__init__(params)
self._py_audio = py_audio
self._out_stream = None
self._sample_rate = 0
@@ -119,7 +117,7 @@ class LocalAudioOutputTransport(BaseOutputTransport):
class LocalAudioTransport(BaseTransport):
def __init__(self, params: LocalAudioTransportParams):
def __init__(self, params: LocalTransportParams):
super().__init__()
self._params = params
self._pyaudio = pyaudio.PyAudio()

View File

@@ -34,15 +34,8 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
class TkTransportParams(TransportParams):
audio_input_device_index: Optional[int] = None
audio_output_device_index: Optional[int] = None
class TkInputTransport(BaseInputTransport):
_params: TkTransportParams
def __init__(self, py_audio: pyaudio.PyAudio, params: TkTransportParams):
def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams):
super().__init__(params)
self._py_audio = py_audio
self._in_stream = None
@@ -61,7 +54,6 @@ class TkInputTransport(BaseInputTransport):
frames_per_buffer=num_frames,
stream_callback=self._audio_in_callback,
input=True,
input_device_index=self._params.audio_input_device_index,
)
self._in_stream.start_stream()
@@ -84,8 +76,6 @@ class TkInputTransport(BaseInputTransport):
class TkOutputTransport(BaseOutputTransport):
_params: TkTransportParams
def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams):
super().__init__(params)
self._py_audio = py_audio
@@ -113,7 +103,6 @@ class TkOutputTransport(BaseOutputTransport):
channels=self._params.audio_out_channels,
rate=self._sample_rate,
output=True,
output_device_index=self._params.audio_output_device_index,
)
self._out_stream.start_stream()

View File

@@ -23,8 +23,6 @@ from pipecat.frames.frames import (
OutputAudioRawFrame,
StartFrame,
StartInterruptionFrame,
TransportMessageFrame,
TransportMessageUrgentFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType
@@ -55,89 +53,45 @@ class FastAPIWebsocketCallbacks(BaseModel):
on_session_timeout: Callable[[WebSocket], Awaitable[None]]
class FastAPIWebsocketClient:
def __init__(self, websocket: WebSocket, is_binary: bool, callbacks: FastAPIWebsocketCallbacks):
self._websocket = websocket
self._closing = False
self._is_binary = is_binary
self._callbacks = callbacks
def receive(self) -> typing.AsyncIterator[bytes | str]:
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)
async def disconnect(self):
if self.is_connected and not self.is_closing:
self._closing = True
await self._websocket.close()
await self.trigger_client_disconnected()
async def trigger_client_disconnected(self):
await self._callbacks.on_client_disconnected(self._websocket)
async def trigger_client_connected(self):
await self._callbacks.on_client_connected(self._websocket)
async def trigger_client_timout(self):
await self._callbacks.on_session_timeout(self._websocket)
def _can_send(self):
return self.is_connected and not self.is_closing
@property
def is_connected(self) -> bool:
return self._websocket.client_state == WebSocketState.CONNECTED
@property
def is_closing(self) -> bool:
return self._closing
class FastAPIWebsocketInputTransport(BaseInputTransport):
def __init__(
self,
client: FastAPIWebsocketClient,
websocket: WebSocket,
params: FastAPIWebsocketParams,
callbacks: FastAPIWebsocketCallbacks,
**kwargs,
):
super().__init__(params, **kwargs)
self._client = client
self._websocket = websocket
self._params = params
self._receive_task = None
self._monitor_websocket_task = None
self._callbacks = callbacks
async def start(self, frame: StartFrame):
await super().start(frame)
await self._params.serializer.setup(frame)
if self._params.session_timeout:
self._monitor_websocket_task = self.create_task(self._monitor_websocket())
await self._client.trigger_client_connected()
await self._callbacks.on_client_connected(self._websocket)
self._receive_task = self.create_task(self._receive_messages())
async def _stop_tasks(self):
if self._monitor_websocket_task:
await self.cancel_task(self._monitor_websocket_task)
await self.cancel_task(self._receive_task)
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._stop_tasks()
await self._client.disconnect()
await self.cancel_task(self._receive_task)
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._stop_tasks()
await self._client.disconnect()
await self.cancel_task(self._receive_task)
def _iter_data(self) -> typing.AsyncIterator[bytes | str]:
if self._params.serializer.type == FrameSerializerType.BINARY:
return self._websocket.iter_bytes()
else:
return self._websocket.iter_text()
async def _receive_messages(self):
try:
async for message in self._client.receive():
async for message in self._iter_data():
frame = await self._params.serializer.deserialize(message)
if not frame:
@@ -150,23 +104,19 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
except Exception as e:
logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})")
await self._client.trigger_client_disconnected()
await self._callbacks.on_client_disconnected(self._websocket)
async def _monitor_websocket(self):
"""Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event."""
await asyncio.sleep(self._params.session_timeout)
await self._client.trigger_client_timout()
await self._callbacks.on_session_timeout(self._websocket)
class FastAPIWebsocketOutputTransport(BaseOutputTransport):
def __init__(
self,
client: FastAPIWebsocketClient,
params: FastAPIWebsocketParams,
**kwargs,
):
def __init__(self, websocket: WebSocket, params: FastAPIWebsocketParams, **kwargs):
super().__init__(params, **kwargs)
self._client = client
self._websocket = websocket
self._params = params
# write_raw_audio_frames() is called quickly, as soon as we get audio
@@ -182,14 +132,6 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
await self._params.serializer.setup(frame)
self._send_interval = (self._audio_chunk_size / self.sample_rate) / 2
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._client.disconnect()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._client.disconnect()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -197,14 +139,8 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
await self._write_frame(frame)
self._next_send_time = 0
async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
await self._write_frame(frame)
async def write_raw_audio_frames(self, frames: bytes):
if self._client.is_closing:
return
if not self._client.is_connected:
if self._websocket.client_state != WebSocketState.CONNECTED:
# Simulate audio playback with a sleep.
await self._write_audio_sleep()
return
@@ -231,17 +167,25 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
await self._write_frame(frame)
self._websocket_audio_buffer = bytes()
# Simulate audio playback with a sleep.
await self._write_audio_sleep()
async def _write_frame(self, frame: Frame):
try:
payload = await self._params.serializer.serialize(frame)
if payload:
await self._client.send(payload)
if payload and self._websocket.client_state == WebSocketState.CONNECTED:
await self._send_data(payload)
except Exception as e:
logger.error(f"{self} exception sending data: {e.__class__.__name__} ({e})")
def _send_data(self, data: str | bytes):
if self._params.serializer.type == FrameSerializerType.BINARY:
return self._websocket.send_bytes(data)
else:
return self._websocket.send_text(data)
async def _write_audio_sleep(self):
# Simulate a clock.
current_time = time.monotonic()
@@ -270,14 +214,11 @@ class FastAPIWebsocketTransport(BaseTransport):
on_session_timeout=self._on_session_timeout,
)
is_binary = self._params.serializer.type == FrameSerializerType.BINARY
self._client = FastAPIWebsocketClient(websocket, is_binary, self._callbacks)
self._input = FastAPIWebsocketInputTransport(
self._client, self._params, name=self._input_name
websocket, self._params, self._callbacks, name=self._input_name
)
self._output = FastAPIWebsocketOutputTransport(
self._client, self._params, name=self._output_name
websocket, self._params, name=self._output_name
)
# Register supported handlers. The user will only be able to register

View File

@@ -6,7 +6,6 @@
import asyncio
import io
import json
import time
import wave
from typing import Awaitable, Callable, Optional
@@ -22,9 +21,6 @@ from pipecat.frames.frames import (
OutputAudioRawFrame,
StartFrame,
StartInterruptionFrame,
TextFrame,
TransportMessageFrame,
TransportMessageUrgentFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.serializers.base_serializer import FrameSerializer
@@ -50,7 +46,6 @@ class WebsocketServerCallbacks(BaseModel):
on_client_connected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]]
on_client_disconnected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]]
on_session_timeout: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]]
on_websocket_ready: Callable[[], Awaitable[None]]
class WebsocketServerInputTransport(BaseInputTransport):
@@ -101,7 +96,6 @@ class WebsocketServerInputTransport(BaseInputTransport):
async def _server_task_handler(self):
logger.info(f"Starting websocket server on {self._host}:{self._port}")
async with websockets.serve(self._client_handler, self._host, self._port) as server:
await self._callbacks.on_websocket_ready()
await self._stop_server_event.wait()
async def _client_handler(self, websocket: websockets.WebSocketServerProtocol, path):
@@ -192,9 +186,6 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
await self._write_frame(frame)
self._next_send_time = 0
async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
await self._write_frame(frame)
async def write_raw_audio_frames(self, frames: bytes):
if not self._websocket:
# Simulate audio playback with a sleep.
@@ -263,7 +254,6 @@ class WebsocketServerTransport(BaseTransport):
on_client_connected=self._on_client_connected,
on_client_disconnected=self._on_client_disconnected,
on_session_timeout=self._on_session_timeout,
on_websocket_ready=self._on_websocket_ready,
)
self._input: Optional[WebsocketServerInputTransport] = None
self._output: Optional[WebsocketServerOutputTransport] = None
@@ -274,7 +264,6 @@ class WebsocketServerTransport(BaseTransport):
self._register_event_handler("on_client_connected")
self._register_event_handler("on_client_disconnected")
self._register_event_handler("on_session_timeout")
self._register_event_handler("on_websocket_ready")
def input(self) -> WebsocketServerInputTransport:
if not self._input:
@@ -304,6 +293,3 @@ class WebsocketServerTransport(BaseTransport):
async def _on_session_timeout(self, websocket):
await self._call_event_handler("on_session_timeout", websocket)
async def _on_websocket_ready(self):
await self._call_event_handler("on_websocket_ready")

View File

@@ -13,6 +13,9 @@ from typing import Any, Awaitable, Callable, Mapping, Optional
import aiohttp
from daily import (
CallClient,
Daily,
EventHandler,
VirtualCameraDevice,
VirtualMicrophoneDevice,
VirtualSpeakerDevice,
@@ -288,7 +291,6 @@ class DailyTransportClient(EventHandler):
self._transcription_ids = []
self._transcription_status = None
self._joining = False
self._joined = False
self._joined_event = asyncio.Event()
self._leave_counter = 0
@@ -329,10 +331,6 @@ class DailyTransportClient(EventHandler):
def _speaker_name(self):
return f"speaker-{self}"
@property
def room_url(self) -> str:
return self._room_url
@property
def participant_id(self) -> str:
return self._participant_id
@@ -425,14 +423,13 @@ class DailyTransportClient(EventHandler):
)
async def join(self):
# Transport already joined or joining, ignore.
if self._joined or self._joining:
# Transport already joined, ignore.
if self._joined:
# Increment leave counter if we already joined.
self._leave_counter += 1
return
logger.info(f"Joining {self._room_url}")
self._joining = True
# For performance reasons, never subscribe to video streams (unless a
# video renderer is registered).
@@ -447,7 +444,6 @@ class DailyTransportClient(EventHandler):
if not error:
self._joined = True
self._joining = False
# Increment leave counter if we successfully joined.
self._leave_counter += 1
@@ -466,7 +462,6 @@ class DailyTransportClient(EventHandler):
except asyncio.TimeoutError:
error_msg = f"Time out joining {self._room_url}"
logger.error(error_msg)
self._joining = False
await self._callbacks.on_error(error_msg)
async def _start_transcription(self):
@@ -842,13 +837,6 @@ class DailyInputTransport(BaseInputTransport):
def vad_analyzer(self) -> Optional[VADAnalyzer]:
return self._vad_analyzer
def start_audio_in_streaming(self):
# Create audio task. It reads audio frames from Daily and push them
# internally for VAD processing.
if self._params.audio_in_enabled or self._params.vad_enabled:
logger.debug(f"Start receiving audio")
self._audio_in_task = self.create_task(self._audio_in_task_handler())
async def start(self, frame: StartFrame):
# Parent start.
await super().start(frame)
@@ -859,8 +847,10 @@ class DailyInputTransport(BaseInputTransport):
# Inialize WebRTC VAD if needed.
if self._params.vad_enabled and not self._params.vad_analyzer:
self._vad_analyzer = WebRTCVADAnalyzer(sample_rate=self.sample_rate)
if self._params.audio_in_stream_on_start:
self.start_audio_in_streaming()
# Create audio task. It reads audio frames from Daily and push them
# internally for VAD processing.
if self._params.audio_in_enabled or self._params.vad_enabled:
self._audio_in_task = self.create_task(self._audio_in_task_handler())
async def stop(self, frame: EndFrame):
# Parent stop.
@@ -1116,10 +1106,6 @@ class DailyTransport(BaseTransport):
# DailyTransport
#
@property
def room_url(self) -> str:
return self._client.room_url
@property
def participant_id(self) -> str:
return self._client.participant_id

View File

@@ -1,27 +0,0 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
def exponential_backoff_time(
attempt: int, min_wait: float = 4, max_wait: float = 10, multiplier: float = 1
) -> float:
"""Calculate exponential backoff wait time.
Args:
attempt: Current attempt number (1-based)
min_wait: Minimum wait time in seconds
max_wait: Maximum wait time in seconds
multiplier: Base multiplier for exponential calculation
Returns:
Wait time in seconds
"""
try:
exp = 2 ** (attempt - 1) * multiplier
result = max(0, min(exp, max_wait))
return max(min_wait, result)
except (ValueError, ArithmeticError):
return max_wait

Some files were not shown because too many files have changed in this diff Show More