Merge branch 'main' into m-ods/assemblyai-universal-streaming

This commit is contained in:
Martin Schweiger
2025-05-23 17:39:00 +08:00
committed by GitHub
94 changed files with 1592 additions and 534 deletions

View File

@@ -9,8 +9,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- Added `SarvamTTSService`, which implements Sarvam AI's TTS API:
https://docs.sarvam.ai/api-reference-docs/text-to-speech/convert.
- Added `PipelineTask.add_observer()` and `PipelineTask.remove_observer()` to
allow mangaging observers at runtime. This is useful for cases where the task
is passed around to other code components that might want to observe the
pipeline dynamically.
- Added `user_id` field to `TranscriptionMessage`. This allows identifying the
user in a multi-user scenario. Note that this requires that
`TranscriptionFrame` has the `user_id` properly set.
- Added new `PipelineTask` event handlers `on_pipeline_started`,
`on_pipeline_stopped`, `on_pipeline_ended` and `on_pipeline_cancelled`, which
correspond to the `StartFrame`, `StopFrame`, `EndFrame` and `CancelFrame`
respectively.
- Added additional languages to `LmntTTSService`. Languages include: `hi`, `id`,
`it`, `ja`, `nl`, `pl`, `ru`, `sv`, `th`, `tr`, `uk`, `vi`.
`it`, `ja`, `nl`, `pl`, `ru`, `sv`, `th`, `tr`, `uk`, `vi`.
- Added a `model` parameter to the `LmntTTSService` constructor, allowing
switching between LMNT models.
@@ -59,12 +76,36 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed
- Updated the default model for `AnthropicLLMService` to
`claude-sonnet-4-20250514`.
- Updated the default model for `GeminiMultimodalLiveLLMService` to
`models/gemini-2.5-flash-preview-native-audio-dialog`.
- `BaseTextFilter` methods `filter()`, `update_settings()`,
`handle_interruption()` and `reset_interruption()` are now async.
- `BaseTextAggregator` methods `aggregate()`, `handle_interruption()` and
`reset()` are now async.
- The API version for `CartesiaTTSService` and `CartesiaHttpTTSService` has
been updated. Also, the `cartesia` dependency has been updated to 2.x.
- `CartesiaTTSService` and `CartesiaHttpTTSService` now support Cartesia's new
`speed` parameter which accepts values of `slow`, `normal`, and `fast`.
- `GeminiMultimodalLiveLLMService` now uses the user transcription and usage
metrics provided by Gemini Live.
- `GoogleLLMService` has been updated to use `google-genai` instead of the
deprecated `google-generativeai`.
### Deprecated
- In `CartesiaTTSService` and `CartesiaHttpTTSService`, `emotion` has been
deprecated by Cartesia. Pipecat is following suit and deprecating `emotion`
as well.
### Removed
- Since `GeminiMultimodalLiveLLMService` now transcribes it's own audio, the
@@ -76,12 +117,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
- Fixed an issue with `CartesiaTTSService` where `TTSTextFrame` messages weren't being emitted when the model was set to `sonic`. This resulted in the assistant context not being updated with assistant messages.
- Fixed an issue with `ElevenLabsTTSService` where changing the model or voice
while the service is running wasn't working.
- Fixed an issue that would cause multiple instances of the same class to behave
incorrectly if any of the given constructor arguments defaulted to a mutable
value (e.g. lists, dictionaries, objects).
- Fixed an issue with `CartesiaTTSService` where `TTSTextFrame` messages weren't
being emitted when the model was set to `sonic`. This resulted in the
assistant context not being updated with assistant messages.
### Performance
- Don't create event handler tasks if no user event handlers have been
registered.
### Other
- Added foundation example `07y-minimax-http.py` to show how to use the
`MiniMaxHttpTTSService`.
- Added foundation examples `07y-interruptible-minimax.py` and
`07z-interruptible-sarvam.py`to show how to use the `MiniMaxHttpTTSService`
and `SarvamTTSService`, respectively.
- Added an `open-telemetry-tracing` example, showing how to setup tracing. The
example also includes Jaeger as an open source OpenTelemetry client to review

View File

@@ -8,6 +8,8 @@
**Pipecat** is an open-source Python framework for building real-time voice and multimodal conversational agents. Orchestrate audio and video, AI services, different transports, and conversation pipelines effortlessly—so you can focus on what makes your agent unique.
> Want to dive right in? [Install Pipecat](https://docs.pipecat.ai/getting-started/installation) then try the [quickstart](https://docs.pipecat.ai/getting-started/quickstart).
## 🚀 What You Can Build
- **Voice Assistants** natural, streaming conversations with AI
@@ -49,18 +51,18 @@ You can connect to Pipecat from any platform using our official SDKs:
## 🧩 Available services
| Category | Services |
| ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [AWS](https://docs.pipecat.ai/server/services/stt/aws), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Fal Wizper](https://docs.pipecat.ai/server/services/stt/fal), [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), [Parakeet (NVIDIA)](https://docs.pipecat.ai/server/services/stt/parakeet), [Ultravox](https://docs.pipecat.ai/server/services/stt/ultravox), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) |
| LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [AWS](https://docs.pipecat.ai/server/services/llm/aws), [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), [Qwen](https://docs.pipecat.ai/server/services/llm/qwen), [Together AI](https://docs.pipecat.ai/server/services/llm/together) |
| 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), [FastPitch (NVIDIA)](https://docs.pipecat.ai/server/services/tts/fastpitch), [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), [MiniMax](https://docs.pipecat.ai/server/services/tts/minimax), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [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) |
| Speech-to-Speech | [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) |
| Transport | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), [FastAPI Websocket](https://docs.pipecat.ai/server/services/transport/fastapi-websocket), [SmallWebRTCTransport](https://docs.pipecat.ai/server/services/transport/small-webrtc), [WebSocket Server](https://docs.pipecat.ai/server/services/transport/websocket-server), Local |
| Video | [Tavus](https://docs.pipecat.ai/server/services/video/tavus), [Simli](https://docs.pipecat.ai/server/services/video/simli) |
| Memory | [mem0](https://docs.pipecat.ai/server/services/memory/mem0) |
| Vision & Image | [fal](https://docs.pipecat.ai/server/services/image-generation/fal), [Google Imagen](https://docs.pipecat.ai/server/services/image-generation/fal), [Moondream](https://docs.pipecat.ai/server/services/vision/moondream) |
| Audio Processing | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Koala](https://docs.pipecat.ai/server/utilities/audio/koala-filter), [Noisereduce](https://docs.pipecat.ai/server/utilities/audio/noisereduce-filter) |
| Analytics & Metrics | [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry) |
| Category | Services |
| ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [AWS](https://docs.pipecat.ai/server/services/stt/aws), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Fal Wizper](https://docs.pipecat.ai/server/services/stt/fal), [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), [Parakeet (NVIDIA)](https://docs.pipecat.ai/server/services/stt/parakeet), [Ultravox](https://docs.pipecat.ai/server/services/stt/ultravox), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) |
| LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [AWS](https://docs.pipecat.ai/server/services/llm/aws), [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), [Qwen](https://docs.pipecat.ai/server/services/llm/qwen), [Together AI](https://docs.pipecat.ai/server/services/llm/together) |
| 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), [FastPitch (NVIDIA)](https://docs.pipecat.ai/server/services/tts/fastpitch), [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), [MiniMax](https://docs.pipecat.ai/server/services/tts/minimax), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [Sarvam](https://docs.pipecat.ai/server/services/tts/sarvam), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) |
| Speech-to-Speech | [AWS Nova Sonic](https://docs.pipecat.ai/server/services/s2s/aws), [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) |
| Transport | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), [FastAPI Websocket](https://docs.pipecat.ai/server/services/transport/fastapi-websocket), [SmallWebRTCTransport](https://docs.pipecat.ai/server/services/transport/small-webrtc), [WebSocket Server](https://docs.pipecat.ai/server/services/transport/websocket-server), Local |
| Video | [Tavus](https://docs.pipecat.ai/server/services/video/tavus), [Simli](https://docs.pipecat.ai/server/services/video/simli) |
| Memory | [mem0](https://docs.pipecat.ai/server/services/memory/mem0) |
| Vision & Image | [fal](https://docs.pipecat.ai/server/services/image-generation/fal), [Google Imagen](https://docs.pipecat.ai/server/services/image-generation/fal), [Moondream](https://docs.pipecat.ai/server/services/vision/moondream) |
| Audio Processing | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Koala](https://docs.pipecat.ai/server/utilities/audio/koala-filter), [Noisereduce](https://docs.pipecat.ai/server/utilities/audio/noisereduce-filter) |
| Analytics & Metrics | [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry) |
📚 [View full services documentation →](https://docs.pipecat.ai/server/services/supported-services)

View File

@@ -105,3 +105,6 @@ TWILIO_AUTH_TOKEN=...
# MiniMax
MINIMAX_API_KEY=...
MINIMAX_GROUP_ID=...
# Sarvam AI
SARVAM_API_KEY=...

View File

@@ -0,0 +1,105 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
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.services.cartesia.tts import CartesiaHttpTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
logger.info(f"Starting bot")
transport = SmallWebRTCTransport(
webrtc_connection=webrtc_connection,
params=TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaHttpTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{
"role": "system",
"content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
},
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
context_aggregator.user(), # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True,
enable_usage_metrics=True,
report_only_initial_ttfb=True,
),
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
messages.append({"role": "system", "content": "Please introduce yourself to the user."})
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
@transport.event_handler("on_client_closed")
async def on_client_closed(transport, client):
logger.info(f"Client closed connection")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
if __name__ == "__main__":
from run import main
main()

View File

@@ -0,0 +1,109 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
import aiohttp
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.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.sarvam.tts import SarvamTTSService
from pipecat.transcriptions.language import Language
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
logger.info(f"Starting bot")
transport = SmallWebRTCTransport(
webrtc_connection=webrtc_connection,
params=TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
# Create an HTTP session
async with aiohttp.ClientSession() as session:
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = SarvamTTSService(
api_key=os.getenv("SARVAM_API_KEY"),
aiohttp_session=session,
params=SarvamTTSService.InputParams(language=Language.EN),
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{
"role": "system",
"content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
},
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
context_aggregator.user(), # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True,
enable_usage_metrics=True,
report_only_initial_ttfb=True,
),
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
messages.append({"role": "system", "content": "Please introduce yourself to the user."})
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
@transport.event_handler("on_client_closed")
async def on_client_closed(transport, client):
logger.info(f"Client closed connection")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
if __name__ == "__main__":
from run import main
main()

View File

@@ -99,21 +99,14 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
)
# Register handler for voice switching
def on_voice_tag(match: PatternMatch):
async def on_voice_tag(match: PatternMatch):
voice_name = match.content.strip().lower()
if voice_name in VOICE_IDS:
voice_id = VOICE_IDS[voice_name]
# Create task to reset the TTS context after voice change
async def change_voice():
# First flush any existing audio to finish the current context
await tts.flush_audio()
# Then set the new voice
tts.set_voice(voice_id)
logger.info(f"Switched to {voice_name} voice")
# Schedule the voice change task
asyncio.create_task(change_voice())
# First flush any existing audio to finish the current context
await tts.flush_audio()
# Then set the new voice
tts.set_voice(VOICE_IDS[voice_name])
logger.info(f"Switched to {voice_name} voice")
else:
logger.warning(f"Unknown voice: {voice_name}")

View File

@@ -1,169 +0,0 @@
# OpenTelemetry Tracing for Pipecat
This demo showcases OpenTelemetry tracing integration for Pipecat services, allowing you to visualize service calls, performance metrics, and dependencies in a Jaeger dashboard.
## Features
- **Hierarchical Tracing**: Track entire conversations, turns, and service calls
- **Service Tracing**: Detailed spans for TTS, STT, and LLM services with rich context
- **TTFB Metrics**: Capture Time To First Byte metrics for latency analysis
- **Usage Statistics**: Track character counts for TTS and token usage for LLMs
- **Flexible Exporters**: Use Jaeger, Zipkin, or any OpenTelemetry-compatible backend
## Trace Structure
Traces are organized hierarchically:
```
Conversation (conversation-uuid)
├── turn-1
│ ├── stt_deepgramsttservice
│ ├── llm_openaillmservice
│ └── tts_cartesiattsservice
└── turn-2
├── stt_deepgramsttservice
├── llm_openaillmservice
└── tts_cartesiattsservice
turn-N
└── ...
```
This organization helps you track conversation-to-conversation and turn-to-turn.
## Setup Instructions
### 1. Start the Jaeger Container
Run Jaeger in Docker to collect and visualize traces:
```bash
docker run -d --name jaeger \
-e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \
-p 16686:16686 \
-p 4317:4317 \
-p 4318:4318 \
jaegertracing/all-in-one:latest
```
### 2. Environment Configuration
Create a `.env` file with your API keys and enable tracing:
```
ENABLE_TRACING=true
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 # Point to your preferred backend
# OTEL_CONSOLE_EXPORT=true # Set to any value for debug output to console
# Service API keys
DEEPGRAM_API_KEY=your_key_here
CARTESIA_API_KEY=your_key_here
OPENAI_API_KEY=your_key_here
```
### 3. Configure Your Pipeline Task
Enable tracing in your Pipecat application:
```python
# Initialize OpenTelemetry with your chosen exporter
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
exporter = OTLPSpanExporter(
endpoint="http://localhost:4317", # Jaeger OTLP endpoint
insecure=True,
)
setup_tracing(
service_name="pipecat-demo",
exporter=exporter,
console_export=os.getenv("OTEL_CONSOLE_EXPORT", "false").lower() == "true",
)
# Enable tracing in your PipelineTask
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True, # Required for some service metrics
),
enable_tracing=True, # Enables both turn and conversation tracing
conversation_id="customer-123", # Optional - will auto-generate if not provided
)
```
### 4. Exporter Options
While this demo uses Jaeger, you can configure any OpenTelemetry-compatible exporter:
#### Jaeger (Default for the demo)
```python
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
exporter = OTLPSpanExporter(
endpoint="http://localhost:4317", # Jaeger OTLP endpoint
insecure=True,
)
```
#### Cloud Providers
Many cloud providers offer OpenTelemetry-compatible observability services:
- AWS X-Ray
- Google Cloud Trace
- Azure Monitor
- Datadog APM
See the OpenTelemetry documentation for specific exporter configurations:
https://opentelemetry.io/ecosystem/vendors/
### 5. Install Dependencies
```bash
pip install -r requirements.txt
```
### 6. Run the Demo
```bash
python bot.py
```
### 7. View Traces in Jaeger
Open your browser to [http://localhost:16686](http://localhost:16686) and select the "pipecat-demo" service to view traces.
## Understanding the Traces
- **Conversation Spans**: The top-level span representing an entire conversation
- **Turn Spans**: Child spans of conversations that represent each turn in the dialog
- **Service Spans**: Detailed service operations nested under turns
- **Service Attributes**: Each service includes rich context about its operation:
- **TTS**: Voice ID, character count, service type
- **STT**: Transcription text, language, model
- **LLM**: Messages, tokens used, model, service configuration
- **Metrics**: Performance data like `metrics.ttfb_ms` and processing durations
## How It Works
The tracing system consists of:
1. **TurnTrackingObserver**: Detects conversation turns
2. **TurnTraceObserver**: Creates spans for turns and conversations
3. **Service Decorators**: `@traced_tts`, `@traced_stt`, `@traced_llm` for service-specific tracing
4. **Context Providers**: Share context between different parts of the pipeline
## Troubleshooting
- **No Traces in Jaeger**: Ensure the Docker container is running and the OTLP endpoint is correct
- **Debugging Traces**: Set `OTEL_CONSOLE_EXPORT=true` to print traces to the console for debugging
- **Missing Metrics**: Check that `enable_metrics=True` in PipelineParams
- **Connection Errors**: Verify network connectivity to the Jaeger container
- **Exporter Issues**: Try the Console exporter (`OTEL_CONSOLE_EXPORT=true`) to verify tracing works
- **Other Backends**: If using a different backend, ensure you've configured the correct exporter and endpoint
## References
- [OpenTelemetry Python Documentation](https://opentelemetry-python.readthedocs.io/)
- [Jaeger Documentation](https://www.jaegertracing.io/docs/latest/)

View File

@@ -0,0 +1,69 @@
# OpenTelemetry Tracing with Pipecat
This repository demonstrates OpenTelemetry tracing integration for Pipecat services, with examples for different backends.
## Tracing Features in Pipecat
- **Hierarchical Tracing**: Track entire conversations, turns, and service calls
- **Service Tracing**: Detailed spans for TTS, STT, and LLM services with rich context
- **TTFB Metrics**: Capture Time To First Byte metrics for latency analysis
- **Usage Statistics**: Track character counts for TTS and token usage for LLMs
## Trace Structure
Traces are organized hierarchically:
```
Conversation (conversation)
├── turn
│ ├── stt_deepgramsttservice
│ ├── llm_openaillmservice
│ └── tts_cartesiattsservice
└── turn
├── stt_deepgramsttservice
├── llm_openaillmservice
└── tts_cartesiattsservice
turn
└── ...
```
This organization helps you track conversation-to-conversation and turn-to-turn interactions.
## Available Demos
| Demo | Description |
| ------------------------------- | ------------------------------------------------------------------------- |
| [Jaeger Tracing](./jaeger/) | Tracing with Jaeger, an open-source end-to-end distributed tracing system |
| [Langfuse Tracing](./langfuse/) | Tracing with Langfuse, a specialized platform for LLM observability |
## Common Requirements
- Python 3.10+
- Pipecat and its dependencies
- API keys for the services used (Deepgram, Cartesia, OpenAI)
- The appropriate OpenTelemetry exporters
## How Tracing Works
The tracing system consists of:
1. **TurnTrackingObserver**: Detects conversation turns
2. **TurnTraceObserver**: Creates spans for turns and conversations
3. **Service Decorators**: `@traced_tts`, `@traced_stt`, `@traced_llm` for service-specific tracing
4. **Context Providers**: Share context between different parts of the pipeline
## Getting Started
1. Choose one of the demos from the table above
2. Follow the README instructions in the respective directory
## Common Troubleshooting
- **Debugging Traces**: Set `OTEL_CONSOLE_EXPORT=true` to print traces to the console for debugging
- **Missing Metrics**: Check that `enable_metrics=True` in PipelineParams
- **API Key Issues**: Verify your API keys are set correctly in the .env file
## References
- [OpenTelemetry Python Documentation](https://opentelemetry-python.readthedocs.io/)
- [Pipecat Documentation](https://docs.pipecat.ai/server/utilities/opentelemetry)

View File

@@ -0,0 +1,80 @@
# Jaeger Tracing for Pipecat
This demo showcases OpenTelemetry tracing integration for Pipecat services using Jaeger, allowing you to visualize service calls, performance metrics, and dependencies.
## Setup Instructions
### 1. Start the Jaeger Container
Run Jaeger in Docker to collect and visualize traces:
```bash
docker run -d --name jaeger \
-e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \
-p 16686:16686 \
-p 4317:4317 \
-p 4318:4318 \
jaegertracing/all-in-one:latest
```
### 2. Environment Configuration
Create a `.env` file with your API keys and enable tracing:
```
ENABLE_TRACING=true
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 # Point to your Jaeger backend
# OTEL_CONSOLE_EXPORT=true # Set to any value for debug output to console
# Service API keys
DEEPGRAM_API_KEY=your_key_here
CARTESIA_API_KEY=your_key_here
OPENAI_API_KEY=your_key_here
```
### 3. Install Dependencies
```bash
pip install -r requirements.txt
```
### 4. Run the Demo
```bash
python bot.py
```
### 5. View Traces in Jaeger
Open your browser to [http://localhost:16686](http://localhost:16686) and select the "pipecat-demo" service to view traces.
## Jaeger-Specific Configuration
In the `bot.py` file, note the GRPC exporter configuration:
```python
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
# Create the exporter
otlp_exporter = OTLPSpanExporter(
endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
insecure=True,
)
# Set up tracing with the exporter
setup_tracing(
service_name="pipecat-demo",
exporter=otlp_exporter,
console_export=bool(os.getenv("OTEL_CONSOLE_EXPORT")),
)
```
## Troubleshooting
- **No Traces in Jaeger**: Ensure the Docker container is running and the OTLP endpoint is correct
- **Connection Errors**: Verify network connectivity to the Jaeger container
- **Exporter Issues**: Try the Console exporter (`OTEL_CONSOLE_EXPORT=true`) to verify tracing works
## References
- [Jaeger Documentation](https://www.jaegertracing.io/docs/latest/)

View File

@@ -6,6 +6,7 @@
import argparse
import os
import sys
from dotenv import load_dotenv
from loguru import logger
@@ -154,6 +155,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
if __name__ == "__main__":
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from run import main
main()

View File

@@ -0,0 +1,82 @@
# Langfuse Tracing for Pipecat
This demo showcases [Langfuse](https://langfuse.com) tracing integration for Pipecat services via OpenTelemetry, allowing you to visualize service calls, performance metrics, and dependencies with a focus on LLM observability.
Pipecat trace in Langfuse:
https://github.com/user-attachments/assets/13dd7431-bf5e-42e3-8d6d-2ed84c51195d
## Setup Instructions
### 1. Create a Langfuse Project and get API keys
[Self-host](https://langfuse.com/self-hosting) Langfuse or create a free [Langfuse Cloud](https://cloud.langfuse.com) account.
Create a new project and get the API keys.
### 2. Environment Configuration
Base64 encode your Langfuse public and secret key:
```bash
echo -n "pk-lf-1234567890:sk-lf-1234567890" | base64
```
Create a `.env` file with your API keys to enable tracing:
```
ENABLE_TRACING=true
# OTLP endpoint for Langfuse
OTEL_EXPORTER_OTLP_ENDPOINT=http://cloud.langfuse.com/api/public/otel
OTEL_EXPORTER_OTLP_HEADERS=Authorization=Basic%20<base64_encoded_api_key>
# Set to any value to enable console output for debugging
# OTEL_CONSOLE_EXPORT=true
# Service API keys
DEEPGRAM_API_KEY=your_key_here
CARTESIA_API_KEY=your_key_here
OPENAI_API_KEY=your_key_here
```
### 3. Install Dependencies
```bash
pip install -r requirements.txt
```
### 4. Run the Demo
```bash
python bot.py
```
### 5. View Traces in Langfuse
Open your browser to [https://cloud.langfuse.com](https://cloud.langfuse.com) to view traces.
## Langfuse-Specific Configuration
In the `bot.py` file, note the HTTP exporter configuration:
```python
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
# Create the exporter - configured from environment variables
otlp_exporter = OTLPSpanExporter()
# Set up tracing with the exporter
setup_tracing(
service_name="pipecat-demo",
exporter=otlp_exporter,
console_export=bool(os.getenv("OTEL_CONSOLE_EXPORT")),
)
```
## Troubleshooting
- **No Traces in Langfuse**: Ensure that your credentials are correct and follow this [troubleshooting guide](https://langfuse.com/faq/all/missing-traces)
- **Connection Errors**: Verify network connectivity to Langfuse
- **Authorization Issues**: Check that your base64 encoding is correct and the API keys are valid
## References
- [Langfuse OpenTelemetry Documentation](https://langfuse.com/docs/opentelemetry/get-started)

View File

@@ -0,0 +1,158 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
import sys
from dotenv import load_dotenv
from loguru import logger
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
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
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
from pipecat.utils.tracing.setup import setup_tracing
load_dotenv(override=True)
IS_TRACING_ENABLED = bool(os.getenv("ENABLE_TRACING"))
# Initialize tracing if enabled
if IS_TRACING_ENABLED:
# Create the exporter
otlp_exporter = OTLPSpanExporter()
# Set up tracing with the exporter
setup_tracing(
service_name="pipecat-demo",
exporter=otlp_exporter,
console_export=bool(os.getenv("OTEL_CONSOLE_EXPORT")),
)
logger.info("OpenTelemetry tracing initialized")
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
logger.info(f"Starting bot")
transport = SmallWebRTCTransport(
webrtc_connection=webrtc_connection,
params=TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"), params=OpenAILLMService.InputParams(temperature=0.5)
)
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the user's location.",
},
},
required=["location", "format"],
)
tools = ToolsSchema(standard_tools=[weather_function])
messages = [
{
"role": "system",
"content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
},
]
context = OpenAILLMContext(messages, tools)
context_aggregator = llm.create_context_aggregator(context)
pipeline = Pipeline(
[
transport.input(),
stt,
context_aggregator.user(),
llm,
tts,
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True,
enable_usage_metrics=True,
),
enable_tracing=IS_TRACING_ENABLED,
# Optionally, add a conversation ID to track the conversation
# conversation_id="8df26cc1-6db0-4a7a-9930-1e037c8f1fa2",
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
@transport.event_handler("on_client_closed")
async def on_client_closed(transport, client):
logger.info(f"Client closed connection")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
if __name__ == "__main__":
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from run import main
main()

View File

@@ -0,0 +1,11 @@
DEEPGRAM_API_KEY=your_deepgram_key
CARTESIA_API_KEY=your_cartesia_key
OPENAI_API_KEY=your_openai_key
# Set to any value to enable tracing
ENABLE_TRACING=true
# OTLP endpoint (change to us.cloud.langfuse.com if you use the US data region)
OTEL_EXPORTER_OTLP_ENDPOINT=http://cloud.langfuse.com/api/public/otel
OTEL_EXPORTER_OTLP_HEADERS=Authorization=Basic%20<base64_encoded_api_keys>
# Set to any value to enable console output for debugging
# OTEL_CONSOLE_EXPORT=true

View File

@@ -0,0 +1,6 @@
fastapi
uvicorn
python-dotenv
pipecat-ai[webrtc,silero,cartesia,deepgram,openai,tracing]
pipecat-ai-small-webrtc-prebuilt
opentelemetry-exporter-otlp-proto-http

View File

@@ -183,6 +183,7 @@ async def bot(args: DailySessionArguments):
args.token,
"Word Wrangler Bot",
DailyParams(
audio_in_enabled=True,
audio_in_filter=None if LOCAL_RUN else KrispFilter(),
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
@@ -210,6 +211,7 @@ async def local_daily():
token,
bot_name="Bot",
params=DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),

View File

@@ -188,7 +188,7 @@ class HostResponseTextFilter(BaseTextFilter):
# No settings to update for this filter
pass
def filter(self, text: str) -> str:
async def filter(self, text: str) -> str:
# Remove case and whitespace for comparison
clean_text = text.strip().upper()
@@ -198,10 +198,10 @@ class HostResponseTextFilter(BaseTextFilter):
return text
def handle_interruption(self):
async def handle_interruption(self):
self._interrupted = True
def reset_interruption(self):
async def reset_interruption(self):
self._interrupted = False
@@ -568,6 +568,7 @@ async def main(room_url: str, token: str):
token,
"Word Wrangler Bot",
DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),

View File

@@ -178,7 +178,7 @@ class HostResponseTextFilter(BaseTextFilter):
# No settings to update for this filter
pass
def filter(self, text: str) -> str:
async def filter(self, text: str) -> str:
# Remove case and whitespace for comparison
clean_text = text.strip().upper()
@@ -188,10 +188,10 @@ class HostResponseTextFilter(BaseTextFilter):
return text
def handle_interruption(self):
async def handle_interruption(self):
self._interrupted = True
def reset_interruption(self):
async def reset_interruption(self):
self._interrupted = False

View File

@@ -44,7 +44,7 @@ assemblyai = [ "assemblyai~=0.37.0" ]
aws = [ "boto3~=1.37.16", "websockets~=13.1" ]
aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.0.2" ]
azure = [ "azure-cognitiveservices-speech~=1.42.0"]
cartesia = [ "cartesia~=1.4.0", "websockets~=13.1" ]
cartesia = [ "cartesia~=2.0.3", "websockets~=13.1" ]
cerebras = []
deepseek = []
daily = [ "daily-python~=0.18.2" ]

View File

@@ -5,7 +5,7 @@
#
from enum import Enum
from typing import Any, Dict, List
from typing import Any, Dict, List, Optional
from pipecat.adapters.schemas.function_schema import FunctionSchema
@@ -18,7 +18,7 @@ class ToolsSchema:
def __init__(
self,
standard_tools: List[FunctionSchema],
custom_tools: Dict[AdapterType, List[Dict[str, Any]]] = None,
custom_tools: Optional[Dict[AdapterType, List[Dict[str, Any]]]] = None,
) -> None:
"""
A schema for tools that includes both standardized function schemas

View File

@@ -36,10 +36,10 @@ class SmartTurnTimeoutException(Exception):
class BaseSmartTurn(BaseTurnAnalyzer):
def __init__(
self, *, sample_rate: Optional[int] = None, params: SmartTurnParams = SmartTurnParams()
self, *, sample_rate: Optional[int] = None, params: Optional[SmartTurnParams] = None
):
super().__init__(sample_rate=sample_rate)
self._params = params
self._params = params or SmartTurnParams()
# Configuration
self._stop_ms = self._params.stop_secs * 1000 # silence threshold in ms
# Inference state

View File

@@ -6,7 +6,7 @@
import asyncio
import io
from typing import Any, Dict
from typing import Any, Dict, Optional
import aiohttp
import numpy as np
@@ -21,12 +21,12 @@ class HttpSmartTurnAnalyzer(BaseSmartTurn):
*,
url: str,
aiohttp_session: aiohttp.ClientSession,
headers: Dict[str, str] = {},
headers: Optional[Dict[str, str]] = None,
**kwargs,
):
super().__init__(**kwargs)
self._url = url
self._headers = headers
self._headers = headers or {}
self._aiohttp_session = aiohttp_session
def _serialize_array(self, audio_array: np.ndarray) -> bytes:

View File

@@ -105,7 +105,7 @@ class SileroOnnxModel:
class SileroVADAnalyzer(VADAnalyzer):
def __init__(self, *, sample_rate: Optional[int] = None, params: VADParams = VADParams()):
def __init__(self, *, sample_rate: Optional[int] = None, params: Optional[VADParams] = None):
super().__init__(sample_rate=sample_rate, params=params)
logger.debug("Loading Silero VAD model...")

View File

@@ -34,10 +34,10 @@ class VADParams(BaseModel):
class VADAnalyzer(ABC):
def __init__(self, *, sample_rate: Optional[int] = None, params: VADParams):
def __init__(self, *, sample_rate: Optional[int] = None, params: Optional[VADParams] = None):
self._init_sample_rate = sample_rate
self._sample_rate = 0
self._params = params
self._params = params or VADParams()
self._num_channels = 1
self._vad_buffer = b""

View File

@@ -288,6 +288,7 @@ class TranscriptionMessage:
role: Literal["user", "assistant"]
content: str
user_id: Optional[str] = None
timestamp: Optional[str] = None

View File

@@ -4,12 +4,13 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from abc import ABC, abstractmethod
from abc import abstractmethod
from dataclasses import dataclass
from typing_extensions import TYPE_CHECKING
from pipecat.frames.frames import Frame
from pipecat.utils.base_object import BaseObject
if TYPE_CHECKING:
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -39,7 +40,7 @@ class FramePushed:
timestamp: int
class BaseObserver(ABC):
class BaseObserver(BaseObject):
"""This is the base class for pipeline frame observers. Observers can view
all the frames that go through the pipeline without the need to inject
processors in the pipeline. This can be useful, for example, to implement

View File

@@ -74,6 +74,7 @@ class DebugLogObserver(BaseObserver):
Union[Tuple[Type[Frame], ...], Dict[Type[Frame], Optional[Tuple[Type, FrameEndpoint]]]]
] = None,
exclude_fields: Optional[Set[str]] = None,
**kwargs,
):
"""Initialize the debug log observer.
@@ -87,6 +88,8 @@ class DebugLogObserver(BaseObserver):
exclude_fields: Set of field names to exclude from logging. If None, only binary
data fields are excluded.
"""
super().__init__(**kwargs)
# Process frame filters
self.frame_filters = {}

View File

@@ -30,7 +30,7 @@ class TurnTrackingObserver(BaseObserver):
"""
def __init__(self, max_frames=100, turn_end_timeout_secs=2.5, **kwargs):
super().__init__()
super().__init__(**kwargs)
self._turn_count = 0
self._is_turn_active = False
self._is_bot_speaking = False
@@ -154,32 +154,3 @@ class TurnTrackingObserver(BaseObserver):
status = "interrupted" if was_interrupted else "completed"
logger.trace(f"Turn {self._turn_count} {status} after {duration:.2f}s")
await self._call_event_handler("on_turn_ended", self._turn_count, duration, was_interrupted)
def _register_event_handler(self, event_name):
"""Register an event handler."""
if not hasattr(self, "_event_handlers"):
self._event_handlers = {}
if event_name not in self._event_handlers:
self._event_handlers[event_name] = []
async def _call_event_handler(self, event_name, *args, **kwargs):
"""Call registered event handlers."""
if not hasattr(self, "_event_handlers"):
return
if event_name in self._event_handlers:
for handler in self._event_handlers[event_name]:
await handler(self, *args, **kwargs)
def event_handler(self, event_name):
"""Decorator for registering event handlers."""
def decorator(func):
if not hasattr(self, "_event_handlers"):
self._event_handlers = {}
if event_name not in self._event_handlers:
self._event_handlers[event_name] = []
self._event_handlers[event_name].append(func)
return func
return decorator

View File

@@ -144,7 +144,26 @@ class PipelineTask(BaseTask):
`LLMFullResponseEndFrame` are received within `idle_timeout_secs`.
@task.event_handler("on_idle_timeout")
async def on_idle_timeout(task):
async def on_pipeline_idle_timeout(task):
...
There are also events to know if a pipeline has been started, stopped, ended
or cancelled.
@task.event_handler("on_pipeline_started")
async def on_pipeline_started(task, frame: StartFrame):
...
@task.event_handler("on_pipeline_stopped")
async def on_pipeline_stopped(task, frame: StopFrame):
...
@task.event_handler("on_pipeline_ended")
async def on_pipeline_ended(task, frame: EndFrame):
...
@task.event_handler("on_pipeline_cancelled")
async def on_pipeline_cancelled(task, frame: CancelFrame):
...
Args:
@@ -169,9 +188,9 @@ class PipelineTask(BaseTask):
self,
pipeline: BasePipeline,
*,
params: PipelineParams = PipelineParams(),
observers: List[BaseObserver] = [],
clock: BaseClock = SystemClock(),
params: Optional[PipelineParams] = None,
observers: Optional[List[BaseObserver]] = None,
clock: Optional[BaseClock] = None,
task_manager: Optional[BaseTaskManager] = None,
check_dangling_tasks: bool = True,
idle_timeout_secs: Optional[float] = 300,
@@ -186,8 +205,8 @@ class PipelineTask(BaseTask):
):
super().__init__()
self._pipeline = pipeline
self._clock = clock
self._params = params
self._clock = clock or SystemClock()
self._params = params or PipelineParams()
self._check_dangling_tasks = check_dangling_tasks
self._idle_timeout_secs = idle_timeout_secs
self._idle_timeout_frames = idle_timeout_frames
@@ -205,14 +224,17 @@ class PipelineTask(BaseTask):
DeprecationWarning,
)
observers = self._params.observers
observers = observers or []
self._turn_tracking_observer: Optional[TurnTrackingObserver] = None
self._turn_trace_observer: Optional[TurnTraceObserver] = None
if self._enable_turn_tracking:
self._turn_tracking_observer = TurnTrackingObserver()
observers = [self._turn_tracking_observer] + list(observers)
if self._enable_turn_tracking and self._enable_tracing:
observers.append(self._turn_tracking_observer)
if self._enable_tracing and self._turn_tracking_observer:
self._turn_trace_observer = TurnTraceObserver(
self._turn_tracking_observer, conversation_id=self._conversation_id
)
observers = [self._turn_trace_observer] + list(observers)
observers.append(self._turn_trace_observer)
self._finished = False
# This queue receives frames coming from the pipeline upstream.
@@ -264,6 +286,10 @@ class PipelineTask(BaseTask):
self._register_event_handler("on_frame_reached_upstream")
self._register_event_handler("on_frame_reached_downstream")
self._register_event_handler("on_idle_timeout")
self._register_event_handler("on_pipeline_started")
self._register_event_handler("on_pipeline_stopped")
self._register_event_handler("on_pipeline_ended")
self._register_event_handler("on_pipeline_cancelled")
@property
def params(self) -> PipelineParams:
@@ -273,12 +299,18 @@ class PipelineTask(BaseTask):
@property
def turn_tracking_observer(self) -> Optional[TurnTrackingObserver]:
"""Return the turn tracking observer if enabled."""
return getattr(self, "_turn_tracking_observer", None)
return self._turn_tracking_observer
@property
def turn_trace_observer(self) -> Optional[TurnTraceObserver]:
"""Return the turn trace observer if enabled."""
return getattr(self, "_turn_trace_observer", None)
return self._turn_trace_observer
async def add_observer(self, observer: BaseObserver):
await self._observer.add_observer(observer)
async def remove_observer(self, observer: BaseObserver):
await self._observer.remove_observer(observer)
def set_event_loop(self, loop: asyncio.AbstractEventLoop):
self._task_manager.set_event_loop(loop)
@@ -552,8 +584,16 @@ class PipelineTask(BaseTask):
if isinstance(frame, self._reached_downstream_types):
await self._call_event_handler("on_frame_reached_downstream", frame)
if isinstance(frame, (EndFrame, StopFrame)):
if isinstance(frame, StartFrame):
await self._call_event_handler("on_pipeline_started", frame)
elif isinstance(frame, EndFrame):
await self._call_event_handler("on_pipeline_ended", frame)
self._pipeline_end_event.set()
elif isinstance(frame, StopFrame):
await self._call_event_handler("on_pipeline_stopped", frame)
self._pipeline_end_event.set()
elif isinstance(frame, CancelFrame):
await self._call_event_handler("on_pipeline_cancelled", frame)
elif isinstance(frame, HeartbeatFrame):
await self._heartbeat_queue.put(frame)
self._down_queue.task_done()

View File

@@ -6,7 +6,7 @@
import asyncio
import inspect
from typing import List
from typing import Dict, List, Optional
from attr import dataclass
@@ -39,10 +39,32 @@ class TaskObserver(BaseObserver):
"""
def __init__(self, *, observers: List[BaseObserver] = [], task_manager: BaseTaskManager):
self._observers = observers
def __init__(
self,
*,
observers: Optional[List[BaseObserver]] = None,
task_manager: BaseTaskManager,
**kwargs,
):
super().__init__(**kwargs)
self._observers = observers or []
self._task_manager = task_manager
self._proxies: List[Proxy] = []
self._proxies: Dict[BaseObserver, Proxy] = {}
async def add_observer(self, observer: BaseObserver):
proxy = self._create_proxy(observer)
self._proxies[observer] = proxy
self._observers.append(observer)
async def remove_observer(self, observer: BaseObserver):
if observer in self._proxies:
proxy = self._proxies[observer]
# Remove the proxy so it doesn't get called anymore.
del self._proxies[observer]
# Cancel the proxy task right away.
await self._task_manager.cancel_task(proxy.task)
# Remove the observer.
self._observers.remove(observer)
async def start(self):
"""Starts all proxy observer tasks."""
@@ -50,23 +72,27 @@ class TaskObserver(BaseObserver):
async def stop(self):
"""Stops all proxy observer tasks."""
for proxy in self._proxies:
for proxy in self._proxies.values():
await self._task_manager.cancel_task(proxy.task)
async def on_push_frame(self, data: FramePushed):
for proxy in self._proxies:
for proxy in self._proxies.values():
await proxy.queue.put(data)
def _create_proxies(self, observers) -> List[Proxy]:
proxies = []
def _create_proxy(self, observer: BaseObserver) -> Proxy:
queue = asyncio.Queue()
task = self._task_manager.create_task(
self._proxy_task_handler(queue, observer),
f"TaskObserver::{observer}::_proxy_task_handler",
)
proxy = Proxy(queue=queue, task=task, observer=observer)
return proxy
def _create_proxies(self, observers: List[BaseObserver]) -> Dict[BaseObserver, Proxy]:
proxies = {}
for observer in observers:
queue = asyncio.Queue()
task = self._task_manager.create_task(
self._proxy_task_handler(queue, observer),
f"TaskObserver::{observer.__class__.__name__}::_proxy_task_handler",
)
proxy = Proxy(queue=queue, task=task, observer=observer)
proxies.append(proxy)
proxy = self._create_proxy(observer)
proxies[observer] = proxy
return proxies
async def _proxy_task_handler(self, queue: asyncio.Queue, observer: BaseObserver):

View File

@@ -7,7 +7,7 @@
import asyncio
from abc import abstractmethod
from dataclasses import dataclass
from typing import Dict, List, Literal, Set
from typing import Dict, List, Literal, Optional, Set
from loguru import logger
@@ -243,11 +243,11 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
self,
context: OpenAILLMContext,
*,
params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
params: Optional[LLMUserAggregatorParams] = None,
**kwargs,
):
super().__init__(context=context, role="user", **kwargs)
self._params = params
self._params = params or LLMUserAggregatorParams()
if "aggregation_timeout" in kwargs:
import warnings
@@ -446,11 +446,11 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
self,
context: OpenAILLMContext,
*,
params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
params: Optional[LLMAssistantAggregatorParams] = None,
**kwargs,
):
super().__init__(context=context, role="assistant", **kwargs)
self._params = params
self._params = params or LLMAssistantAggregatorParams()
if "expect_stripped_words" in kwargs:
import warnings
@@ -640,9 +640,9 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
class LLMUserResponseAggregator(LLMUserContextAggregator):
def __init__(
self,
messages: List[dict] = [],
messages: Optional[List[dict]] = None,
*,
params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
params: Optional[LLMUserAggregatorParams] = None,
**kwargs,
):
super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs)
@@ -662,9 +662,9 @@ class LLMUserResponseAggregator(LLMUserContextAggregator):
class LLMAssistantResponseAggregator(LLMAssistantContextAggregator):
def __init__(
self,
messages: List[dict] = [],
messages: Optional[List[dict]] = None,
*,
params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
params: Optional[LLMAssistantAggregatorParams] = None,
**kwargs,
):
super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs)

View File

@@ -7,6 +7,7 @@
import re
import time
from enum import Enum
from typing import List
from loguru import logger
@@ -31,7 +32,7 @@ class WakeCheckFilter(FrameProcessor):
self.wake_timer = 0.0
self.accumulator = ""
def __init__(self, wake_phrases: list[str], keepalive_timeout: float = 3):
def __init__(self, wake_phrases: List[str], keepalive_timeout: float = 3):
super().__init__()
self._participant_states = {}
self._keepalive_timeout = keepalive_timeout

View File

@@ -22,7 +22,7 @@ class WakeNotifierFilter(FrameProcessor):
self,
notifier: BaseNotifier,
*,
types: Tuple[Type[Frame]],
types: Tuple[Type[Frame], ...],
filter: Callable[[Frame], Awaitable[bool]],
**kwargs,
):

View File

@@ -437,10 +437,12 @@ class RTVIObserver(BaseObserver):
params (RTVIObserverParams): Settings to enable/disable specific messages.
"""
def __init__(self, rtvi: "RTVIProcessor", *, params: RTVIObserverParams = RTVIObserverParams()):
super().__init__()
def __init__(
self, rtvi: "RTVIProcessor", *, params: Optional[RTVIObserverParams] = None, **kwargs
):
super().__init__(**kwargs)
self._rtvi = rtvi
self._params = params
self._params = params or RTVIObserverParams()
self._bot_transcription = ""
self._frames_seen = set()
rtvi.set_errors_enabled(self._params.errors_enabled)
@@ -632,12 +634,12 @@ class RTVIProcessor(FrameProcessor):
def __init__(
self,
*,
config: RTVIConfig = RTVIConfig(config=[]),
config: Optional[RTVIConfig] = None,
transport: Optional[BaseTransport] = None,
**kwargs,
):
super().__init__(**kwargs)
self._config = config
self._config = config or RTVIConfig(config=[])
self._bot_ready = False
self._client_ready = False

View File

@@ -43,10 +43,10 @@ class GStreamerPipelineSource(FrameProcessor):
audio_channels: int = 1
clock_sync: bool = True
def __init__(self, *, pipeline: str, out_params: OutputParams = OutputParams(), **kwargs):
def __init__(self, *, pipeline: str, out_params: Optional[OutputParams] = None, **kwargs):
super().__init__(**kwargs)
self._out_params = out_params
self._out_params = out_params or GStreamerPipelineSource.OutputParams()
self._sample_rate = 0
Gst.init()

View File

@@ -5,7 +5,7 @@
#
import asyncio
from typing import Awaitable, Callable, List
from typing import Awaitable, Callable, List, Optional
from pipecat.frames.frames import Frame, StartFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -22,14 +22,14 @@ class IdleFrameProcessor(FrameProcessor):
*,
callback: Callable[["IdleFrameProcessor"], Awaitable[None]],
timeout: float,
types: List[type] = [],
types: Optional[List[type]] = None,
**kwargs,
):
super().__init__(**kwargs)
self._callback = callback
self._timeout = timeout
self._types = types
self._types = types or []
self._idle_task = None
async def process_frame(self, frame: Frame, direction: FrameDirection):

View File

@@ -4,11 +4,17 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from typing import Optional
from typing import Optional, Tuple, Type
from loguru import logger
from pipecat.frames.frames import AudioRawFrame, BotSpeakingFrame, Frame, TransportMessageFrame
from pipecat.frames.frames import (
BotSpeakingFrame,
Frame,
InputAudioRawFrame,
OutputAudioRawFrame,
TransportMessageFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
logger = logger.opt(ansi=True)
@@ -19,16 +25,17 @@ class FrameLogger(FrameProcessor):
self,
prefix="Frame",
color: Optional[str] = None,
ignored_frame_types: Optional[list] = [
ignored_frame_types: Tuple[Type[Frame], ...] = (
BotSpeakingFrame,
AudioRawFrame,
InputAudioRawFrame,
OutputAudioRawFrame,
TransportMessageFrame,
],
),
):
super().__init__()
self._prefix = prefix
self._color = color
self._ignored_frame_types = tuple(ignored_frame_types) if ignored_frame_types else None
self._ignored_frame_types = ignored_frame_types
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)

View File

@@ -62,7 +62,7 @@ class UserTranscriptProcessor(BaseTranscriptProcessor):
if isinstance(frame, TranscriptionFrame):
message = TranscriptionMessage(
role="user", content=frame.text, timestamp=frame.timestamp
role="user", user_id=frame.user_id, content=frame.text, timestamp=frame.timestamp
)
await self._emit_update([message])

View File

@@ -79,7 +79,7 @@ class TelnyxFrameSerializer(FrameSerializer):
inbound_encoding: str,
call_control_id: Optional[str] = None,
api_key: Optional[str] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
):
"""Initialize the TelnyxFrameSerializer.
@@ -92,11 +92,11 @@ class TelnyxFrameSerializer(FrameSerializer):
params: Configuration parameters.
"""
self._stream_id = stream_id
params.outbound_encoding = outbound_encoding
params.inbound_encoding = inbound_encoding
self._call_control_id = call_control_id
self._api_key = api_key
self._params = params
self._params = params or TelnyxFrameSerializer.InputParams()
self._params.outbound_encoding = outbound_encoding
self._params.inbound_encoding = inbound_encoding
self._telnyx_sample_rate = self._params.telnyx_sample_rate
self._sample_rate = 0 # Pipeline input rate

View File

@@ -69,7 +69,7 @@ class TwilioFrameSerializer(FrameSerializer):
call_sid: Optional[str] = None,
account_sid: Optional[str] = None,
auth_token: Optional[str] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
):
"""Initialize the TwilioFrameSerializer.
@@ -84,7 +84,7 @@ class TwilioFrameSerializer(FrameSerializer):
self._call_sid = call_sid
self._account_sid = account_sid
self._auth_token = auth_token
self._params = params
self._params = params or TwilioFrameSerializer.InputParams()
self._twilio_sample_rate = self._params.twilio_sample_rate
self._sample_rate = 0 # Pipeline input rate

View File

@@ -90,12 +90,13 @@ class AnthropicLLMService(LLMService):
self,
*,
api_key: str,
model: str = "claude-3-7-sonnet-20250219",
params: InputParams = InputParams(),
model: str = "claude-sonnet-4-20250514",
params: Optional[InputParams] = None,
client=None,
**kwargs,
):
super().__init__(**kwargs)
params = params or AnthropicLLMService.InputParams()
self._client = client or AsyncAnthropic(
api_key=api_key
) # if the client is provided, use it and remove it, otherwise create a new one

View File

@@ -58,7 +58,7 @@ class AssemblyAISTTService(STTService):
self._connection_params = connection_params
super().__init__(sample_rate=self._connection_params.sample_rate, **kwargs)
self._websocket = None
self._termination_event = asyncio.Event()
self._received_termination = False

View File

@@ -530,17 +530,19 @@ class AWSBedrockLLMService(LLMService):
def __init__(
self,
*,
model: str,
aws_access_key: Optional[str] = None,
aws_secret_key: Optional[str] = None,
aws_session_token: Optional[str] = None,
aws_region: str = "us-east-1",
model: str,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
client_config: Optional[Config] = None,
**kwargs,
):
super().__init__(**kwargs)
params = params or AWSBedrockLLMService.InputParams()
# Initialize the AWS Bedrock client
if not client_config:
client_config = Config(

View File

@@ -125,11 +125,13 @@ class AWSPollyTTSService(TTSService):
region: Optional[str] = None,
voice_id: str = "Joanna",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or AWSPollyTTSService.InputParams()
self._polly_client = boto3.client(
"polly",
aws_access_key_id=aws_access_key_id,

View File

@@ -142,7 +142,7 @@ class AWSNovaSonicLLMService(LLMService):
region: str,
model: str = "amazon.nova-sonic-v1:0",
voice_id: str = "matthew", # matthew, tiffany, amy
params: Params = Params(),
params: Optional[Params] = None,
system_instruction: Optional[str] = None,
tools: Optional[ToolsSchema] = None,
send_transcription_frames: bool = True,
@@ -155,7 +155,7 @@ class AWSNovaSonicLLMService(LLMService):
self._model = model
self._client: Optional[BedrockRuntimeClient] = None
self._voice_id = voice_id
self._params = params
self._params = params or Params()
self._system_instruction = system_instruction
self._tools = tools
self._send_transcription_frames = send_transcription_frames

View File

@@ -68,11 +68,13 @@ class AzureBaseTTSService(TTSService):
region: str,
voice="en-US-SaraNeural",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or AzureBaseTTSService.InputParams()
self._settings = {
"emphasis": params.emphasis,
"language": self.language_to_service_language(params.language)

View File

@@ -7,10 +7,11 @@
import base64
import json
import uuid
import warnings
from typing import AsyncGenerator, List, Optional, Union
from loguru import logger
from pydantic import BaseModel
from pydantic import BaseModel, Field
from pipecat.frames.frames import (
CancelFrame,
@@ -83,13 +84,13 @@ class CartesiaTTSService(AudioContextWordTTSService):
*,
api_key: str,
voice_id: str,
cartesia_version: str = "2024-06-10",
cartesia_version: str = "2025-04-16",
url: str = "wss://api.cartesia.ai/tts/websocket",
model: str = "sonic-2",
sample_rate: Optional[int] = None,
encoding: str = "pcm_s16le",
container: str = "raw",
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
text_aggregator: Optional[BaseTextAggregator] = None,
**kwargs,
):
@@ -112,6 +113,8 @@ class CartesiaTTSService(AudioContextWordTTSService):
**kwargs,
)
params = params or CartesiaTTSService.InputParams()
self._api_key = api_key
self._cartesia_version = cartesia_version
self._url = url
@@ -151,10 +154,13 @@ class CartesiaTTSService(AudioContextWordTTSService):
voice_config["mode"] = "id"
voice_config["id"] = self._voice_id
if self._settings["speed"] or self._settings["emotion"]:
if self._settings["emotion"]:
warnings.warn(
"The 'emotion' parameter in __experimental_controls is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
voice_config["__experimental_controls"] = {}
if self._settings["speed"]:
voice_config["__experimental_controls"]["speed"] = self._settings["speed"]
if self._settings["emotion"]:
voice_config["__experimental_controls"]["emotion"] = self._settings["emotion"]
@@ -169,6 +175,10 @@ class CartesiaTTSService(AudioContextWordTTSService):
"add_timestamps": add_timestamps,
"use_original_timestamps": False if self.model_name == "sonic" else True,
}
if self._settings["speed"]:
msg["speed"] = self._settings["speed"]
return json.dumps(msg)
async def start(self, frame: StartFrame):
@@ -309,7 +319,7 @@ class CartesiaHttpTTSService(TTSService):
class InputParams(BaseModel):
language: Optional[Language] = Language.EN
speed: Optional[Union[str, float]] = ""
emotion: Optional[List[str]] = []
emotion: Optional[List[str]] = Field(default_factory=list)
def __init__(
self,
@@ -318,15 +328,20 @@ class CartesiaHttpTTSService(TTSService):
voice_id: str,
model: str = "sonic-2",
base_url: str = "https://api.cartesia.ai",
cartesia_version: str = "2024-11-13",
sample_rate: Optional[int] = None,
encoding: str = "pcm_s16le",
container: str = "raw",
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or CartesiaHttpTTSService.InputParams()
self._api_key = api_key
self._base_url = base_url
self._cartesia_version = cartesia_version
self._settings = {
"output_format": {
"container": container,
@@ -342,7 +357,10 @@ class CartesiaHttpTTSService(TTSService):
self.set_voice(voice_id)
self.set_model_name(model)
self._client = AsyncCartesia(api_key=api_key, base_url=base_url)
self._client = AsyncCartesia(
api_key=api_key,
base_url=base_url,
)
def can_generate_metrics(self) -> bool:
return True
@@ -367,37 +385,63 @@ class CartesiaHttpTTSService(TTSService):
logger.debug(f"{self}: Generating TTS [{text}]")
try:
voice_controls = None
if self._settings["speed"] or self._settings["emotion"]:
voice_controls = {}
if self._settings["speed"]:
voice_controls["speed"] = self._settings["speed"]
if self._settings["emotion"]:
voice_controls["emotion"] = self._settings["emotion"]
voice_config = {"mode": "id", "id": self._voice_id}
if self._settings["emotion"]:
warnings.warn(
"The 'emotion' parameter in voice.__experimental_controls is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
voice_config["__experimental_controls"] = {"emotion": self._settings["emotion"]}
await self.start_ttfb_metrics()
output = await self._client.tts.sse(
model_id=self._model_name,
transcript=text,
voice_id=self._voice_id,
output_format=self._settings["output_format"],
language=self._settings["language"],
stream=False,
_experimental_voice_controls=voice_controls,
)
payload = {
"model_id": self._model_name,
"transcript": text,
"voice": voice_config,
"output_format": self._settings["output_format"],
"language": self._settings["language"],
}
await self.start_tts_usage_metrics(text)
if self._settings["speed"]:
payload["speed"] = self._settings["speed"]
yield TTSStartedFrame()
session = await self._client._get_session()
headers = {
"Cartesia-Version": self._cartesia_version,
"X-API-Key": self._api_key,
"Content-Type": "application/json",
}
url = f"{self._base_url}/tts/bytes"
async with session.post(url, json=payload, headers=headers) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Cartesia API error: {error_text}")
await self.push_error(ErrorFrame(f"Cartesia API error: {error_text}"))
raise Exception(f"Cartesia API returned status {response.status}: {error_text}")
audio_data = await response.read()
await self.start_tts_usage_metrics(text)
frame = TTSAudioRawFrame(
audio=output["audio"], sample_rate=self.sample_rate, num_channels=1
audio=audio_data,
sample_rate=self.sample_rate,
num_channels=1,
)
yield frame
except Exception as e:
logger.error(f"{self} exception: {e}")
await self.push_error(ErrorFrame(f"Error generating TTS: {e}"))
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()

View File

@@ -184,7 +184,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
model: str = "eleven_flash_v2_5",
url: str = "wss://api.elevenlabs.io",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
# Aggregating sentences still gives cleaner-sounding results and fewer
@@ -210,6 +210,8 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
**kwargs,
)
params = params or ElevenLabsTTSService.InputParams()
self._api_key = api_key
self._url = url
self._settings = {
@@ -252,14 +254,16 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
async def set_model(self, model: str):
await super().set_model(model)
logger.info(f"Switching TTS model to: [{model}]")
# No need to disconnect/reconnect for model changes with multi-context API
await self._disconnect()
await self._connect()
async def _update_settings(self, settings: Mapping[str, Any]):
prev_voice = self._voice_id
await super()._update_settings(settings)
# If voice changes, we don't need to reconnect, just use a new context
if not prev_voice == self._voice_id:
logger.info(f"Switching TTS voice to: [{self._voice_id}]")
await self._disconnect()
await self._connect()
async def start(self, frame: StartFrame):
await super().start(frame)
@@ -386,7 +390,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
msg = json.loads(message)
# Check if this message belongs to the current context
# The default context may return null/None for context_id
received_ctx_id = msg.get("context_id")
received_ctx_id = msg.get("contextId")
if (
self._context_id is not None
and received_ctx_id is not None
@@ -406,7 +410,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
word_times = calculate_word_times(msg["alignment"], self._cumulative_time)
await self.add_word_timestamps(word_times)
self._cumulative_time = word_times[-1][1]
if msg.get("is_final"):
if msg.get("isFinal"):
logger.trace(f"Received final message for context {received_ctx_id}")
# Context has finished
if self._context_id == received_ctx_id:
@@ -512,7 +516,7 @@ class ElevenLabsHttpTTSService(WordTTSService):
model: str = "eleven_flash_v2_5",
base_url: str = "https://api.elevenlabs.io",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(
@@ -523,6 +527,8 @@ class ElevenLabsHttpTTSService(WordTTSService):
**kwargs,
)
params = params or ElevenLabsHttpTTSService.InputParams()
self._api_key = api_key
self._base_url = base_url
self._params = params

View File

@@ -173,7 +173,7 @@ class FalSTTService(SegmentedSTTService):
*,
api_key: Optional[str] = None,
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(
@@ -181,6 +181,8 @@ class FalSTTService(SegmentedSTTService):
**kwargs,
)
params = params or FalSTTService.InputParams()
if api_key:
os.environ["FAL_KEY"] = api_key
elif "FAL_KEY" not in os.environ:

View File

@@ -52,7 +52,7 @@ class FishAudioTTSService(InterruptibleTTSService):
model: str, # This is the reference_id
output_format: FishAudioOutputFormat = "pcm",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(
@@ -62,6 +62,8 @@ class FishAudioTTSService(InterruptibleTTSService):
**kwargs,
)
params = params or FishAudioTTSService.InputParams()
self._api_key = api_key
self._base_url = "wss://api.fish.audio/v1/tts/live"
self._websocket = None

View File

@@ -335,17 +335,20 @@ class GeminiMultimodalLiveLLMService(LLMService):
*,
api_key: str,
base_url: str = "generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent",
model="models/gemini-2.0-flash-live-001",
model="models/gemini-2.5-flash-preview-native-audio-dialog",
voice_id: str = "Charon",
start_audio_paused: bool = False,
start_video_paused: bool = False,
system_instruction: Optional[str] = None,
tools: Optional[Union[List[dict], ToolsSchema]] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
inference_on_context_initialization: bool = True,
**kwargs,
):
super().__init__(base_url=base_url, **kwargs)
params = params or InputParams()
self._last_sent_time = 0
self._api_key = api_key
self._base_url = base_url
@@ -960,11 +963,17 @@ class GeminiMultimodalLiveLLMService(LLMService):
usage = evt.usageMetadata
# Ensure we have valid integers for all token counts
prompt_tokens = usage.promptTokenCount or 0
completion_tokens = usage.responseTokenCount or 0
total_tokens = usage.totalTokenCount or (prompt_tokens + completion_tokens)
tokens = LLMTokenUsage(
prompt_tokens=usage.promptTokenCount,
completion_tokens=usage.responseTokenCount,
total_tokens=usage.totalTokenCount,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
)
await self.start_llm_usage_metrics(tokens)
def create_context_aggregator(

View File

@@ -194,7 +194,7 @@ class GladiaSTTService(STTService):
confidence: float = 0.5,
sample_rate: Optional[int] = None,
model: str = "solaria-1",
params: GladiaInputParams = GladiaInputParams(),
params: Optional[GladiaInputParams] = None,
**kwargs,
):
"""Initialize the Gladia STT service.
@@ -211,6 +211,8 @@ class GladiaSTTService(STTService):
"""
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or GladiaInputParams()
# Warn about deprecated language parameter if it's used
if params.language is not None:
warnings.warn(

View File

@@ -10,7 +10,7 @@ import os
# Suppress gRPC fork warnings
os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
from typing import AsyncGenerator
from typing import AsyncGenerator, Optional
from loguru import logger
from PIL import Image
@@ -32,19 +32,19 @@ class GoogleImageGenService(ImageGenService):
class InputParams(BaseModel):
number_of_images: int = Field(default=1, ge=1, le=8)
model: str = Field(default="imagen-3.0-generate-002")
negative_prompt: str = Field(default=None)
negative_prompt: Optional[str] = Field(default=None)
def __init__(
self,
*,
params: InputParams = InputParams(),
api_key: str,
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(**kwargs)
self.set_model_name(params.model)
self._params = params
self._params = params or GoogleImageGenService.InputParams()
self._client = genai.Client(api_key=api_key)
self.set_model_name(self._params.model)
def can_generate_metrics(self) -> bool:
return True

View File

@@ -467,13 +467,16 @@ class GoogleLLMService(LLMService):
*,
api_key: str,
model: str = "gemini-2.0-flash",
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
system_instruction: Optional[str] = None,
tools: Optional[List[Dict[str, Any]]] = None,
tool_config: Optional[Dict[str, Any]] = None,
**kwargs,
):
super().__init__(**kwargs)
params = params or GoogleLLMService.InputParams()
self.set_model_name(model)
self._api_key = api_key
self._system_instruction = system_instruction

View File

@@ -52,7 +52,7 @@ class GoogleVertexLLMService(OpenAILLMService):
credentials: Optional[str] = None,
credentials_path: Optional[str] = None,
model: str = "google/gemini-2.0-flash-001",
params: InputParams = OpenAILLMService.InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
"""Initializes the VertexLLMService.
@@ -64,6 +64,7 @@ class GoogleVertexLLMService(OpenAILLMService):
params (InputParams): Vertex AI input parameters.
**kwargs: Additional arguments for OpenAILLMService.
"""
params = params or OpenAILLMService.InputParams()
base_url = self._get_base_url(params)
self._api_key = self._get_api_token(credentials, credentials_path)

View File

@@ -412,7 +412,7 @@ class GoogleSTTService(STTService):
credentials_path: Optional[str] = None,
location: str = "global",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
"""Initialize the Google STT service.
@@ -431,6 +431,8 @@ class GoogleSTTService(STTService):
"""
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or GoogleSTTService.InputParams()
self._location = location
self._stream = None
self._config = None

View File

@@ -219,11 +219,13 @@ class GoogleTTSService(TTSService):
credentials_path: Optional[str] = None,
voice_id: str = "en-US-Neural2-A",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or GoogleTTSService.InputParams()
self._settings = {
"pitch": params.pitch,
"rate": params.rate,

View File

@@ -34,7 +34,7 @@ class GroqTTSService(TTSService):
*,
api_key: str,
output_format: str = "wav",
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
model_name: str = "playai-tts",
voice_id: str = "Celeste-PlayAI",
sample_rate: Optional[int] = GROQ_SAMPLE_RATE,
@@ -42,12 +42,15 @@ class GroqTTSService(TTSService):
):
if sample_rate != self.GROQ_SAMPLE_RATE:
logger.warning(f"Groq TTS only supports {self.GROQ_SAMPLE_RATE}Hz sample rate. ")
super().__init__(
pause_frame_processing=True,
sample_rate=sample_rate,
**kwargs,
)
params = params or GroqTTSService.InputParams()
self._api_key = api_key
self._model_name = model_name
self._output_format = output_format

View File

@@ -155,8 +155,7 @@ class MCPClient(BaseObject):
error_msg = f"Error calling mcp tool {function_name}: {str(e)}"
logger.error(error_msg)
response = "Sorry, could not call the mcp tool"
image_url = None
response = ""
if results:
if hasattr(results, "content") and results.content:
for i, content in enumerate(results.content):
@@ -171,7 +170,8 @@ class MCPClient(BaseObject):
else:
logger.error(f"Error getting content from {function_name} results.")
await result_callback(response)
final_response = response if len(response) else "Sorry, could not call the mcp tool"
await result_callback(final_response)
async def _list_tools(self, session, mcp_tool_wrapper, llm):
available_tools = await session.list_tools()

View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from typing import Any, Dict, List
from typing import Any, Dict, List, Optional
from loguru import logger
from pydantic import BaseModel, Field
@@ -49,16 +49,19 @@ class Mem0MemoryService(FrameProcessor):
def __init__(
self,
*,
api_key: str = None,
local_config: Dict[str, Any] = {},
user_id: str = None,
agent_id: str = None,
run_id: str = None,
params: InputParams = InputParams(),
api_key: Optional[str] = None,
local_config: Optional[Dict[str, Any]] = None,
user_id: Optional[str] = None,
agent_id: Optional[str] = None,
run_id: Optional[str] = None,
params: Optional[InputParams] = None,
):
# Important: Call the parent class __init__ first
super().__init__()
local_config = local_config or {}
params = params or Mem0MemoryService.InputParams()
if local_config:
self.memory_client = Memory.from_config(local_config)
else:

View File

@@ -114,11 +114,13 @@ class MiniMaxHttpTTSService(TTSService):
voice_id: str = "Calm_Woman",
aiohttp_session: aiohttp.ClientSession,
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or MiniMaxHttpTTSService.InputParams()
self._api_key = api_key
self._group_id = group_id
self._base_url = f"https://api.minimaxi.chat/v1/t2a_v2?GroupId={group_id}"

View File

@@ -80,7 +80,7 @@ class NeuphonicTTSService(InterruptibleTTSService):
url: str = "wss://api.neuphonic.com",
sample_rate: Optional[int] = 22050,
encoding: str = "pcm_linear",
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(
@@ -92,6 +92,8 @@ class NeuphonicTTSService(InterruptibleTTSService):
**kwargs,
)
params = params or NeuphonicTTSService.InputParams()
self._api_key = api_key
self._url = url
self._settings = {
@@ -293,11 +295,13 @@ class NeuphonicHttpTTSService(TTSService):
url: str = "https://api.neuphonic.com",
sample_rate: Optional[int] = 22050,
encoding: str = "pcm_linear",
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or NeuphonicHttpTTSService.InputParams()
self._api_key = api_key
self._url = url
self._settings = {

View File

@@ -77,11 +77,14 @@ class BaseOpenAILLMService(LLMService):
base_url=None,
organization=None,
project=None,
default_headers: Mapping[str, str] | None = None,
params: InputParams = InputParams(),
default_headers: Optional[Mapping[str, str]] = None,
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(**kwargs)
params = params or BaseOpenAILLMService.InputParams()
self._settings = {
"frequency_penalty": params.frequency_penalty,
"presence_penalty": params.presence_penalty,

View File

@@ -6,7 +6,7 @@
import json
from dataclasses import dataclass
from typing import Any
from typing import Any, Optional
from pipecat.frames.frames import (
FunctionCallCancelFrame,
@@ -41,7 +41,7 @@ class OpenAILLMService(BaseOpenAILLMService):
self,
*,
model: str = "gpt-4.1",
params: BaseOpenAILLMService.InputParams = BaseOpenAILLMService.InputParams(),
params: Optional[BaseOpenAILLMService.InputParams] = None,
**kwargs,
):
super().__init__(model=model, params=params, **kwargs)

View File

@@ -8,6 +8,7 @@ import base64
import json
import time
from dataclasses import dataclass
from typing import Optional
from loguru import logger
@@ -89,17 +90,20 @@ class OpenAIRealtimeBetaLLMService(LLMService):
api_key: str,
model: str = "gpt-4o-realtime-preview-2024-12-17",
base_url: str = "wss://api.openai.com/v1/realtime",
session_properties: events.SessionProperties = events.SessionProperties(),
session_properties: Optional[events.SessionProperties] = None,
start_audio_paused: bool = False,
send_transcription_frames: bool = True,
**kwargs,
):
full_url = f"{base_url}?model={model}"
super().__init__(base_url=full_url, **kwargs)
self.api_key = api_key
self.base_url = full_url
self._session_properties: events.SessionProperties = session_properties
self._session_properties: events.SessionProperties = (
session_properties or events.SessionProperties()
)
self._audio_input_paused = start_audio_paused
self._send_transcription_frames = send_transcription_frames
self._websocket = None

View File

@@ -110,7 +110,7 @@ class PlayHTTTSService(InterruptibleTTSService):
voice_engine: str = "Play3.0-mini",
sample_rate: Optional[int] = None,
output_format: str = "wav",
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(
@@ -119,6 +119,8 @@ class PlayHTTTSService(InterruptibleTTSService):
**kwargs,
)
params = params or PlayHTTTSService.InputParams()
self._api_key = api_key
self._user_id = user_id
self._websocket_url = None
@@ -328,11 +330,13 @@ class PlayHTHttpTTSService(TTSService):
voice_engine: str = "Play3.0-mini",
protocol: str = "http", # Options: http, ws
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or PlayHTHttpTTSService.InputParams()
self._user_id = user_id
self._api_key = api_key

View File

@@ -80,7 +80,7 @@ class RimeTTSService(AudioContextWordTTSService):
url: str = "wss://users.rime.ai/ws2",
model: str = "mistv2",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
text_aggregator: Optional[BaseTextAggregator] = None,
**kwargs,
):
@@ -105,6 +105,8 @@ class RimeTTSService(AudioContextWordTTSService):
**kwargs,
)
params = params or RimeTTSService.InputParams()
# Store service configuration
self._api_key = api_key
self._url = url
@@ -364,11 +366,13 @@ class RimeHttpTTSService(TTSService):
aiohttp_session: aiohttp.ClientSession,
model: str = "mistv2",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or RimeHttpTTSService.InputParams()
self._api_key = api_key
self._session = aiohttp_session
self._base_url = "https://users.rime.ai/v1/rime-tts"

View File

@@ -99,10 +99,13 @@ class RivaSTTService(STTService):
"model_name": "parakeet-ctc-1.1b-asr",
},
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or RivaSTTService.InputParams()
self._api_key = api_key
self._profanity_filter = False
self._automatic_punctuation = True
@@ -322,11 +325,13 @@ class RivaSegmentedSTTService(SegmentedSTTService):
"model_name": "canary-1b-asr",
},
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or RivaSegmentedSTTService.InputParams()
# Set model name
self.set_model_name(model_function_map.get("model_name"))
@@ -533,7 +538,7 @@ class ParakeetSTTService(RivaSTTService):
"model_name": "parakeet-ctc-1.1b-asr",
},
sample_rate: Optional[int] = None,
params: RivaSTTService.InputParams = RivaSTTService.InputParams(), # Use parent class's type
params: Optional[RivaSTTService.InputParams] = None, # Use parent class's type
**kwargs,
):
super().__init__(

View File

@@ -44,7 +44,7 @@ class RivaTTSService(TTSService):
def __init__(
self,
*,
api_key: str = None,
api_key: str,
server: str = "grpc.nvcf.nvidia.com:443",
voice_id: str = "Magpie-Multilingual.EN-US.Ray",
sample_rate: Optional[int] = None,
@@ -52,10 +52,13 @@ class RivaTTSService(TTSService):
"function_id": "877104f7-e885-42b9-8de8-f6e4c6303969",
"model_name": "magpie-tts-multilingual",
},
params: InputParams = InputParams(),
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or RivaTTSService.InputParams()
self._api_key = api_key
self._voice_id = voice_id
self._language_code = params.language
@@ -136,14 +139,10 @@ class RivaTTSService(TTSService):
class FastPitchTTSService(RivaTTSService):
class InputParams(BaseModel):
language: Optional[Language] = Language.EN_US
quality: Optional[int] = 20
def __init__(
self,
*,
api_key: str = None,
api_key: str,
server: str = "grpc.nvcf.nvidia.com:443",
voice_id: str = "English-US.Female-1",
sample_rate: Optional[int] = None,
@@ -151,11 +150,12 @@ class FastPitchTTSService(RivaTTSService):
"function_id": "0149dedb-2be8-4195-b9a0-e57e0e14f972",
"model_name": "fastpitch-hifigan-tts",
},
params: InputParams = InputParams(),
params: Optional[RivaTTSService.InputParams] = None,
**kwargs,
):
super().__init__(
api_key=api_key,
server=server,
voice_id=voice_id,
sample_rate=sample_rate,
model_function_map=model_function_map,

View File

@@ -0,0 +1,8 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from .tts import *

View File

@@ -0,0 +1,195 @@
#
# Copyright (c) 20242025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import base64
from typing import AsyncGenerator, Optional
import aiohttp
from loguru import logger
from pydantic import BaseModel, Field
from pipecat.frames.frames import (
ErrorFrame,
Frame,
StartFrame,
TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
)
from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
def language_to_sarvam_language(language: Language) -> Optional[str]:
"""Convert Pipecat Language enum to Sarvam AI language codes."""
LANGUAGE_MAP = {
Language.BN: "bn-IN", # Bengali
Language.EN: "en-IN", # English (India)
Language.GU: "gu-IN", # Gujarati
Language.HI: "hi-IN", # Hindi
Language.KN: "kn-IN", # Kannada
Language.ML: "ml-IN", # Malayalam
Language.MR: "mr-IN", # Marathi
Language.OR: "od-IN", # Odia
Language.PA: "pa-IN", # Punjabi
Language.TA: "ta-IN", # Tamil
Language.TE: "te-IN", # Telugu
}
return LANGUAGE_MAP.get(language)
class SarvamTTSService(TTSService):
"""Text-to-Speech service using Sarvam AI's API.
Converts text to speech using Sarvam AI's TTS models with support for multiple
Indian languages. Provides control over voice characteristics like pitch, pace,
and loudness.
Args:
api_key: Sarvam AI API subscription key.
voice_id: Speaker voice ID (e.g., "anushka", "meera").
model: TTS model to use ("bulbul:v1" or "bulbul:v2").
aiohttp_session: Shared aiohttp session for making requests.
base_url: Sarvam AI API base URL.
sample_rate: Audio sample rate in Hz (8000, 16000, 22050, 24000).
params: Additional voice and preprocessing parameters.
Example:
```python
tts = SarvamTTSService(
api_key="your-api-key",
voice_id="anushka",
model="bulbul:v2",
aiohttp_session=session,
params=SarvamTTSService.InputParams(
language=Language.HI,
pitch=0.1,
pace=1.2
)
)
```
"""
class InputParams(BaseModel):
language: Optional[Language] = Language.EN
pitch: Optional[float] = Field(default=0.0, ge=-0.75, le=0.75)
pace: Optional[float] = Field(default=1.0, ge=0.3, le=3.0)
loudness: Optional[float] = Field(default=1.0, ge=0.1, le=3.0)
enable_preprocessing: Optional[bool] = False
def __init__(
self,
*,
api_key: str,
voice_id: str = "anushka",
model: str = "bulbul:v2",
aiohttp_session: aiohttp.ClientSession,
base_url: str = "https://api.sarvam.ai",
sample_rate: Optional[int] = None,
params: Optional[InputParams] = None,
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
params = params or SarvamTTSService.InputParams()
self._api_key = api_key
self._base_url = base_url
self._session = aiohttp_session
self._settings = {
"language": self.language_to_service_language(params.language)
if params.language
else "en-IN",
"pitch": params.pitch,
"pace": params.pace,
"loudness": params.loudness,
"enable_preprocessing": params.enable_preprocessing,
}
self.set_model_name(model)
self.set_voice(voice_id)
def can_generate_metrics(self) -> bool:
return True
def language_to_service_language(self, language: Language) -> Optional[str]:
return language_to_sarvam_language(language)
async def start(self, frame: StartFrame):
await super().start(frame)
self._settings["sample_rate"] = self.sample_rate
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")
try:
await self.start_ttfb_metrics()
payload = {
"text": text,
"target_language_code": self._settings["language"],
"speaker": self._voice_id,
"pitch": self._settings["pitch"],
"pace": self._settings["pace"],
"loudness": self._settings["loudness"],
"speech_sample_rate": self.sample_rate,
"enable_preprocessing": self._settings["enable_preprocessing"],
"model": self._model_name,
}
headers = {
"api-subscription-key": self._api_key,
"Content-Type": "application/json",
}
url = f"{self._base_url}/text-to-speech"
yield TTSStartedFrame()
async with self._session.post(url, json=payload, headers=headers) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Sarvam API error: {error_text}")
await self.push_error(ErrorFrame(f"Sarvam API error: {error_text}"))
return
response_data = await response.json()
await self.start_tts_usage_metrics(text)
# Decode base64 audio data
if "audios" not in response_data or not response_data["audios"]:
logger.error("No audio data received from Sarvam API")
await self.push_error(ErrorFrame("No audio data received"))
return
# Get the first audio (there should be only one for single text input)
base64_audio = response_data["audios"][0]
audio_data = base64.b64decode(base64_audio)
# Strip WAV header (first 44 bytes) if present
if audio_data.startswith(b"RIFF"):
logger.debug("Stripping WAV header from Sarvam audio data")
audio_data = audio_data[44:]
frame = TTSAudioRawFrame(
audio=audio_data,
sample_rate=self.sample_rate,
num_channels=1,
)
yield frame
except Exception as e:
logger.error(f"{self} exception: {e}")
await self.push_error(ErrorFrame(f"Error generating TTS: {e}"))
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()

View File

@@ -64,7 +64,7 @@ class TTSService(AIService):
# Text aggregator to aggregate incoming tokens and decide when to push to the TTS.
text_aggregator: Optional[BaseTextAggregator] = None,
# Text filter executed after text has been aggregated.
text_filters: Sequence[BaseTextFilter] = [],
text_filters: Optional[Sequence[BaseTextFilter]] = None,
text_filter: Optional[BaseTextFilter] = None,
# Audio transport destination of the generated frames.
transport_destination: Optional[str] = None,
@@ -83,7 +83,7 @@ class TTSService(AIService):
self._voice_id: str = ""
self._settings: Dict[str, Any] = {}
self._text_aggregator: BaseTextAggregator = text_aggregator or SimpleTextAggregator()
self._text_filters: Sequence[BaseTextFilter] = text_filters
self._text_filters: Sequence[BaseTextFilter] = text_filters or []
self._transport_destination: Optional[str] = transport_destination
if text_filter:
@@ -157,7 +157,7 @@ class TTSService(AIService):
self.set_voice(value)
elif key == "text_filter":
for filter in self._text_filters:
filter.update_settings(value)
await filter.update_settings(value)
else:
logger.warning(f"Unknown setting for TTS service: {key}")
@@ -183,7 +183,7 @@ class TTSService(AIService):
await self._maybe_pause_frame_processing()
sentence = self._text_aggregator.text
self._text_aggregator.reset()
await self._text_aggregator.reset()
self._processing_text = False
await self._push_tts_frames(sentence)
if isinstance(frame, LLMFullResponseEndFrame):
@@ -234,9 +234,9 @@ class TTSService(AIService):
async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
self._processing_text = False
self._text_aggregator.handle_interruption()
await self._text_aggregator.handle_interruption()
for filter in self._text_filters:
filter.handle_interruption()
await filter.handle_interruption()
async def _maybe_pause_frame_processing(self):
if self._processing_text and self._pause_frame_processing:
@@ -251,7 +251,7 @@ class TTSService(AIService):
if not self._aggregate_sentences:
text = frame.text
else:
text = self._text_aggregator.aggregate(frame.text)
text = await self._text_aggregator.aggregate(frame.text)
if text:
await self._push_tts_frames(text)
@@ -274,8 +274,8 @@ class TTSService(AIService):
# Process all filter.
for filter in self._text_filters:
filter.reset_interruption()
text = filter.filter(text)
await filter.reset_interruption()
text = await filter.filter(text)
if text:
await self.process_generator(self.run_tts(text))

View File

@@ -38,7 +38,9 @@ class HeartbeatsObserver(BaseObserver):
*,
target: FrameProcessor,
heartbeat_callback: Callable[[FrameProcessor, HeartbeatFrame], Awaitable[None]],
**kwargs,
):
super().__init__(**kwargs)
self._target = target
self._callback = heartbeat_callback
@@ -79,10 +81,13 @@ async def run_test(
expected_down_frames: Optional[Sequence[type]] = None,
expected_up_frames: Optional[Sequence[type]] = None,
ignore_start: bool = True,
observers: List[BaseObserver] = [],
start_metadata: Dict[str, Any] = {},
observers: Optional[List[BaseObserver]] = None,
start_metadata: Optional[Dict[str, Any]] = None,
send_end_frame: bool = True,
) -> Tuple[Sequence[Frame], Sequence[Frame]]:
observers = observers or []
start_metadata = start_metadata or {}
received_up = asyncio.Queue()
received_down = asyncio.Queue()
source = QueuedFrameProcessor(

View File

@@ -250,11 +250,11 @@ class WebsocketClientTransport(BaseTransport):
def __init__(
self,
uri: str,
params: WebsocketClientParams = WebsocketClientParams(),
params: Optional[WebsocketClientParams] = None,
):
super().__init__()
self._params = params
self._params = params or WebsocketClientParams()
callbacks = WebsocketClientCallbacks(
on_connected=self._on_connected,
@@ -262,7 +262,7 @@ class WebsocketClientTransport(BaseTransport):
on_message=self._on_message,
)
self._session = WebsocketClientSession(uri, params, callbacks, self.name)
self._session = WebsocketClientSession(uri, self._params, callbacks, self.name)
self._input: Optional[WebsocketClientInputTransport] = None
self._output: Optional[WebsocketClientOutputTransport] = None

View File

@@ -95,7 +95,7 @@ class WebRTCVADAnalyzer(VADAnalyzer):
params: VAD configuration parameters (VADParams).
"""
def __init__(self, *, sample_rate: Optional[int] = None, params: VADParams = VADParams()):
def __init__(self, *, sample_rate: Optional[int] = None, params: Optional[VADParams] = None):
super().__init__(sample_rate=sample_rate, params=params)
self._webrtc_vad = Daily.create_native_vad(
@@ -1223,7 +1223,7 @@ class DailyTransport(BaseTransport):
room_url: str,
token: Optional[str],
bot_name: str,
params: DailyParams = DailyParams(),
params: Optional[DailyParams] = None,
input_name: Optional[str] = None,
output_name: Optional[str] = None,
):
@@ -1256,9 +1256,11 @@ class DailyTransport(BaseTransport):
on_recording_stopped=self._on_recording_stopped,
on_recording_error=self._on_recording_error,
)
self._params = params
self._params = params or DailyParams()
self._client = DailyTransportClient(room_url, token, bot_name, params, callbacks, self.name)
self._client = DailyTransportClient(
room_url, token, bot_name, self._params, callbacks, self.name
)
self._input: Optional[DailyInputTransport] = None
self._output: Optional[DailyOutputTransport] = None

View File

@@ -499,7 +499,7 @@ class LiveKitTransport(BaseTransport):
url: str,
token: str,
room_name: str,
params: LiveKitParams = LiveKitParams(),
params: Optional[LiveKitParams] = None,
input_name: Optional[str] = None,
output_name: Optional[str] = None,
):
@@ -515,7 +515,7 @@ class LiveKitTransport(BaseTransport):
on_data_received=self._on_data_received,
on_first_participant_joined=self._on_first_participant_joined,
)
self._params = params
self._params = params or LiveKitParams()
self._client = LiveKitTransportClient(
url, token, room_name, self._params, callbacks, self.name

View File

@@ -59,6 +59,11 @@ class BaseObject(ABC):
self._event_handlers[event_name] = []
async def _call_event_handler(self, event_name: str, *args, **kwargs):
# If we haven't registered an event handler, we don't need to do
# anything.
if not self._event_handlers.get(event_name):
return
# Create the task.
task = asyncio.create_task(self._run_task(event_name, *args, **kwargs))

View File

@@ -25,7 +25,7 @@ class BaseTextAggregator(ABC):
pass
@abstractmethod
def aggregate(self, text: str) -> Optional[str]:
async def aggregate(self, text: str) -> Optional[str]:
"""Aggregates the specified text with the currently accumulated text.
This method should be implemented to define how the new text contributes
@@ -43,7 +43,7 @@ class BaseTextAggregator(ABC):
pass
@abstractmethod
def handle_interruption(self):
async def handle_interruption(self):
"""Handles interruptions. When an interruption occurs it is possible
that we might want to discard the aggregated text or do some internal
modifications to the aggregated text.
@@ -52,6 +52,6 @@ class BaseTextAggregator(ABC):
pass
@abstractmethod
def reset(self):
async def reset(self):
"""Clears the internally aggregated text."""
pass

View File

@@ -10,17 +10,17 @@ from typing import Any, Mapping
class BaseTextFilter(ABC):
@abstractmethod
def update_settings(self, settings: Mapping[str, Any]):
async def update_settings(self, settings: Mapping[str, Any]):
pass
@abstractmethod
def filter(self, text: str) -> str:
async def filter(self, text: str) -> str:
pass
@abstractmethod
def handle_interruption(self):
async def handle_interruption(self):
pass
@abstractmethod
def reset_interruption(self):
async def reset_interruption(self):
pass

View File

@@ -26,19 +26,19 @@ class MarkdownTextFilter(BaseTextFilter):
filter_code: Optional[bool] = False
filter_tables: Optional[bool] = False
def __init__(self, params: InputParams = InputParams(), **kwargs):
def __init__(self, params: Optional[InputParams] = None, **kwargs):
super().__init__(**kwargs)
self._settings = params
self._settings = params or MarkdownTextFilter.InputParams()
self._in_code_block = False
self._in_table = False
self._interrupted = False
def update_settings(self, settings: Mapping[str, Any]):
async def update_settings(self, settings: Mapping[str, Any]):
for key, value in settings.items():
if hasattr(self._settings, key):
setattr(self._settings, key, value)
def filter(self, text: str) -> str:
async def filter(self, text: str) -> str:
if self._settings.enable_text_filter:
# Remove newlines and replace with a space only when there's no text before or after
filtered_text = re.sub(r"^\s*\n", " ", text, flags=re.MULTILINE)
@@ -100,16 +100,19 @@ class MarkdownTextFilter(BaseTextFilter):
# Restore leading and trailing spaces
filtered_text = re.sub("§", " ", filtered_text)
## Make links more readable
filtered_text = re.sub(r"https?://", "", filtered_text)
return filtered_text
else:
return text
def handle_interruption(self):
async def handle_interruption(self):
self._interrupted = True
self._in_code_block = False
self._in_table = False
def reset_interruption(self):
async def reset_interruption(self):
self._interrupted = False
#

View File

@@ -5,7 +5,7 @@
#
import re
from typing import Callable, Optional, Tuple
from typing import Awaitable, Callable, Optional, Tuple
from loguru import logger
@@ -106,7 +106,7 @@ class PatternPairAggregator(BaseTextAggregator):
return self
def on_pattern_match(
self, pattern_id: str, handler: Callable[[PatternMatch], None]
self, pattern_id: str, handler: Callable[[PatternMatch], Awaitable[None]]
) -> "PatternPairAggregator":
"""Register a handler for when a pattern pair is matched.
@@ -124,7 +124,7 @@ class PatternPairAggregator(BaseTextAggregator):
self._handlers[pattern_id] = handler
return self
def _process_complete_patterns(self, text: str) -> Tuple[str, bool]:
async def _process_complete_patterns(self, text: str) -> Tuple[str, bool]:
"""Process all complete pattern pairs in the text.
Searches for all complete pattern pairs in the text, calls the
@@ -167,7 +167,7 @@ class PatternPairAggregator(BaseTextAggregator):
# Call the appropriate handler if registered
if pattern_id in self._handlers:
try:
self._handlers[pattern_id](pattern_match)
await self._handlers[pattern_id](pattern_match)
except Exception as e:
logger.error(f"Error in pattern handler for {pattern_id}: {e}")
@@ -204,7 +204,7 @@ class PatternPairAggregator(BaseTextAggregator):
return False
def aggregate(self, text: str) -> Optional[str]:
async def aggregate(self, text: str) -> Optional[str]:
"""Aggregate text and process pattern pairs.
This method adds the new text to the buffer, processes any complete pattern
@@ -223,7 +223,7 @@ class PatternPairAggregator(BaseTextAggregator):
self._text += text
# Process any complete patterns in the buffer
processed_text, modified = self._process_complete_patterns(self._text)
processed_text, modified = await self._process_complete_patterns(self._text)
# Only update the buffer if modifications were made
if modified:
@@ -245,7 +245,7 @@ class PatternPairAggregator(BaseTextAggregator):
# No complete sentence found yet
return None
def handle_interruption(self):
async def handle_interruption(self):
"""Handle interruptions by clearing the buffer.
Called when an interruption occurs in the processing pipeline,
@@ -253,7 +253,7 @@ class PatternPairAggregator(BaseTextAggregator):
"""
self._text = ""
def reset(self):
async def reset(self):
"""Clear the internally aggregated text.
Resets the aggregator to its initial state, discarding any

View File

@@ -23,7 +23,7 @@ class SimpleTextAggregator(BaseTextAggregator):
def text(self) -> str:
return self._text
def aggregate(self, text: str) -> Optional[str]:
async def aggregate(self, text: str) -> Optional[str]:
result: Optional[str] = None
self._text += text
@@ -35,8 +35,8 @@ class SimpleTextAggregator(BaseTextAggregator):
return result
def handle_interruption(self):
async def handle_interruption(self):
self._text = ""
def reset(self):
async def reset(self):
self._text = ""

View File

@@ -43,7 +43,7 @@ class SkipTagsAggregator(BaseTextAggregator):
"""
return self._text
def aggregate(self, text: str) -> Optional[str]:
async def aggregate(self, text: str) -> Optional[str]:
"""Aggregate text and process pattern pairs.
This method adds the new text to the buffer, processes any complete pattern
@@ -77,7 +77,7 @@ class SkipTagsAggregator(BaseTextAggregator):
# No complete sentence found yet
return None
def handle_interruption(self):
async def handle_interruption(self):
"""Handle interruptions by clearing the buffer.
Called when an interruption occurs in the processing pipeline,
@@ -85,7 +85,7 @@ class SkipTagsAggregator(BaseTextAggregator):
"""
self._text = ""
def reset(self):
async def reset(self):
"""Clear the internally aggregated text.
Resets the aggregator to its initial state, discarding any

View File

@@ -18,6 +18,39 @@ if is_tracing_available():
from opentelemetry.trace import Span
def _get_gen_ai_system_from_service_name(service_name: str) -> str:
"""Extract the standardized gen_ai.system value from a service class name.
Source:
https://opentelemetry.io/docs/specs/semconv/attributes-registry/gen-ai/#gen-ai-system
Uses standard OTel names where possible, with special case mappings for
service names that don't follow the pattern.
"""
SPECIAL_CASE_MAPPINGS = {
# AWS
"AWSBedrockLLMService": "aws.bedrock",
# Azure
"AzureLLMService": "az.ai.openai",
# Google
"GoogleLLMService": "gcp.gemini",
"GoogleLLMOpenAIBetaService": "gcp.gemini",
"GoogleVertexLLMService": "gcp.vertex_ai",
# Others
"GrokLLMService": "xai",
}
if service_name in SPECIAL_CASE_MAPPINGS:
return SPECIAL_CASE_MAPPINGS[service_name]
if service_name.endswith("LLMService"):
provider = service_name[:-10].lower()
else:
provider = service_name.lower()
return provider
def add_tts_span_attributes(
span: "Span",
service_name: str,
@@ -45,17 +78,18 @@ def add_tts_span_attributes(
**kwargs: Additional attributes to add
"""
# Add standard attributes
span.set_attribute("service.name", service_name)
span.set_attribute("model", model)
span.set_attribute("gen_ai.system", service_name.replace("TTSService", "").lower())
span.set_attribute("gen_ai.request.model", model)
span.set_attribute("gen_ai.operation.name", operation_name)
span.set_attribute("gen_ai.output.type", "speech")
span.set_attribute("voice_id", voice_id)
span.set_attribute("operation", operation_name)
# Add optional attributes
if text:
span.set_attribute("text", text)
if character_count is not None:
span.set_attribute("metrics.tts.character_count", character_count)
span.set_attribute("metrics.character_count", character_count)
if ttfb_ms is not None:
span.set_attribute("metrics.ttfb_ms", ttfb_ms)
@@ -76,6 +110,7 @@ def add_stt_span_attributes(
span: "Span",
service_name: str,
model: str,
operation_name: str = "stt",
transcript: Optional[str] = None,
is_final: Optional[bool] = None,
language: Optional[str] = None,
@@ -90,6 +125,7 @@ def add_stt_span_attributes(
span: The span to add attributes to
service_name: Name of the STT service (e.g., "deepgram")
model: Model name/identifier
operation_name: Name of the operation (default: "stt")
transcript: The transcribed text
is_final: Whether this is a final transcript
language: Detected or configured language
@@ -99,8 +135,9 @@ def add_stt_span_attributes(
**kwargs: Additional attributes to add
"""
# Add standard attributes
span.set_attribute("service.name", service_name)
span.set_attribute("model", model)
span.set_attribute("gen_ai.system", service_name.replace("STTService", "").lower())
span.set_attribute("gen_ai.request.model", model)
span.set_attribute("gen_ai.operation.name", operation_name)
span.set_attribute("vad_enabled", vad_enabled)
# Add optional attributes
@@ -161,13 +198,15 @@ def add_llm_span_attributes(
**kwargs: Additional attributes to add
"""
# Add standard attributes
span.set_attribute("service.name", service_name)
span.set_attribute("model", model)
span.set_attribute("gen_ai.system", _get_gen_ai_system_from_service_name(service_name))
span.set_attribute("gen_ai.request.model", model)
span.set_attribute("gen_ai.operation.name", "chat")
span.set_attribute("gen_ai.output.type", "text")
span.set_attribute("stream", stream)
# Add optional attributes
if messages:
span.set_attribute("messages", messages)
span.set_attribute("input", messages)
if tools:
span.set_attribute("tools", tools)
@@ -188,7 +227,19 @@ def add_llm_span_attributes(
if parameters:
for key, value in parameters.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(f"param.{key}", value)
if key in [
"temperature",
"max_tokens",
"max_completion_tokens",
"top_p",
"top_k",
"frequency_penalty",
"presence_penalty",
"seed",
]:
span.set_attribute(f"gen_ai.request.{key}", value)
else:
span.set_attribute(f"param.{key}", value)
# Add extra parameters if provided
if extra_parameters:

View File

@@ -67,20 +67,6 @@ def _get_parent_service_context(self):
return context_api.get_current()
def _get_service_name(self, service_prefix: str) -> str:
"""Generate a default span name using service type and class name.
Args:
self: The service instance.
service_prefix: The service type (e.g., 'llm', 'stt', 'tts').
Returns:
A default span name string like "type_classname" (e.g. llm_openaillmservice).
"""
service_class_name = self.__class__.__name__.lower()
return f"{service_prefix}_{service_class_name}"
def _add_token_usage_to_span(span, token_usage):
"""Add token usage metrics to a span (internal use only).
@@ -93,13 +79,15 @@ def _add_token_usage_to_span(span, token_usage):
if isinstance(token_usage, dict):
if "prompt_tokens" in token_usage:
span.set_attribute("llm.prompt_tokens", token_usage["prompt_tokens"])
span.set_attribute("gen_ai.usage.input_tokens", token_usage["prompt_tokens"])
if "completion_tokens" in token_usage:
span.set_attribute("llm.completion_tokens", token_usage["completion_tokens"])
span.set_attribute("gen_ai.usage.output_tokens", token_usage["completion_tokens"])
else:
# Handle LLMTokenUsage object
span.set_attribute("llm.prompt_tokens", getattr(token_usage, "prompt_tokens", 0))
span.set_attribute("llm.completion_tokens", getattr(token_usage, "completion_tokens", 0))
span.set_attribute("gen_ai.usage.input_tokens", getattr(token_usage, "prompt_tokens", 0))
span.set_attribute(
"gen_ai.usage.output_tokens", getattr(token_usage, "completion_tokens", 0)
)
def traced_tts(func: Optional[Callable] = None, *, name: Optional[str] = None) -> Callable:
@@ -134,7 +122,7 @@ def traced_tts(func: Optional[Callable] = None, *, name: Optional[str] = None) -
return
service_class_name = self.__class__.__name__
span_name = name or _get_service_name(self, "tts")
span_name = "tts"
# Get parent context
turn_context = get_current_turn_context()
@@ -237,7 +225,7 @@ def traced_stt(func: Optional[Callable] = None, *, name: Optional[str] = None) -
return await f(self, transcript, is_final, language)
service_class_name = self.__class__.__name__
span_name = name or _get_service_name(self, "stt")
span_name = "stt"
# Get the turn context first, then fall back to service context
turn_context = get_current_turn_context()
@@ -313,7 +301,7 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
return await f(self, context, *args, **kwargs)
service_class_name = self.__class__.__name__
span_name = name or _get_service_name(self, "llm")
span_name = "llm"
# Get the parent context - turn context if available, otherwise service context
turn_context = get_current_turn_context()

View File

@@ -34,8 +34,10 @@ class TurnTraceObserver(BaseObserver):
conversation span that encapsulates the entire session.
"""
def __init__(self, turn_tracker: TurnTrackingObserver, conversation_id: Optional[str] = None):
super().__init__()
def __init__(
self, turn_tracker: TurnTrackingObserver, conversation_id: Optional[str] = None, **kwargs
):
super().__init__(**kwargs)
self._turn_tracker = turn_tracker
self._current_span: Optional["Span"] = None
self._current_turn_number: int = 0
@@ -82,7 +84,7 @@ class TurnTraceObserver(BaseObserver):
self._conversation_id = conversation_id
# Create a new span for this conversation
self._conversation_span = self._tracer.start_span(f"conversation-{conversation_id}")
self._conversation_span = self._tracer.start_span("conversation")
# Set span attributes
self._conversation_span.set_attribute("conversation.id", conversation_id)
@@ -143,7 +145,7 @@ class TurnTraceObserver(BaseObserver):
parent_context = context_provider.get_current_conversation_context()
# Create a new span for this turn
self._current_span = self._tracer.start_span(f"turn-{turn_number}", context=parent_context)
self._current_span = self._tracer.start_span("turn", context=parent_context)
self._current_turn_number = turn_number
# Set span attributes

View File

@@ -30,7 +30,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
Some inline code here
"""
result = self.filter.filter(input_text)
result = await self.filter.filter(input_text)
self.assertEqual(result.strip(), expected_text.strip())
async def test_space_preservation(self):
@@ -45,7 +45,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
]
for text in input_text:
result = self.filter.filter(text)
result = await self.filter.filter(text)
self.assertEqual(
len(result), len(text), f"Space preservation failed for: '{text}'\nGot: '{result}'"
)
@@ -71,7 +71,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
}
for input_text, expected in test_cases.items():
result = self.filter.filter(input_text)
result = await self.filter.filter(input_text)
self.assertEqual(
result,
expected,
@@ -88,7 +88,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
2. Second item
3. Third item with bold"""
result = self.filter.filter(input_text)
result = await self.filter.filter(input_text)
self.assertEqual(
result.strip(),
expected.strip(),
@@ -106,7 +106,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
}
for input_text, expected in test_cases.items():
result = self.filter.filter(input_text)
result = await self.filter.filter(input_text)
self.assertEqual(result, expected, f"HTML entity conversion failed for: '{input_text}'")
async def test_asterisk_removal(self):
@@ -120,7 +120,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
}
for input_text, expected in test_cases.items():
result = self.filter.filter(input_text)
result = await self.filter.filter(input_text)
self.assertEqual(result, expected, f"Asterisk removal failed for: '{input_text}'")
async def test_newline_handling(self):
@@ -132,11 +132,23 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
}
for input_text, expected in test_cases.items():
result = self.filter.filter(input_text)
result = await self.filter.filter(input_text)
self.assertEqual(
result, expected, f"Newline handling failed for:\n{input_text}\nGot:\n{result}"
)
async def test_links_cleaning(self):
"""Test cleaning of links and URLs, i.e. https?:// is removed."""
test_cases = {
"Please check http://example.com": "Please check example.com",
"Visit https://www.google.com for more": "Visit www.google.com for more",
"No link here": "No link here", # No link to clean
}
for input_text, expected in test_cases.items():
result = await self.filter.filter(input_text)
self.assertEqual(result, expected, f"Link cleaning failed for: '{input_text}'")
async def test_numbered_list_marker_handling(self):
"""Test handling of numbered lists with the special §NUM§ marker."""
test_cases = {
@@ -148,7 +160,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
}
for input_text, expected in test_cases.items():
result = self.filter.filter(input_text)
result = await self.filter.filter(input_text)
self.assertEqual(
result,
expected,
@@ -166,7 +178,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
}
for input_text, expected in test_cases.items():
result = self.filter.filter(input_text)
result = await self.filter.filter(input_text)
self.assertEqual(result, expected, f"Inline code handling failed for: '{input_text}'")
async def test_simple_table_removal(self):
@@ -177,7 +189,7 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
expected = ""
result = filter.filter(input_text)
result = await filter.filter(input_text)
self.assertEqual(
result.strip(),
expected.strip(),
@@ -198,15 +210,15 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
# Test with text filtering disabled
text_with_markdown = "**bold** and *italic* with `code`"
self.assertEqual(
filter.filter(text_with_markdown),
await filter.filter(text_with_markdown),
text_with_markdown,
"Disabled filter should not modify text",
)
# Enable just text filtering
filter.update_settings({"enable_text_filter": True})
await filter.update_settings({"enable_text_filter": True})
self.assertEqual(
filter.filter(text_with_markdown),
await filter.filter(text_with_markdown),
"bold and italic with code",
"Enabled filter should remove markdown",
)
@@ -217,14 +229,18 @@ class TestMarkdownTextFilter(unittest.IsolatedAsyncioTestCase):
# Initial state - formatting should be removed
input_text = "**bold** and *italic*"
self.assertEqual(filter.filter(input_text), "bold and italic")
self.assertEqual(await filter.filter(input_text), "bold and italic")
# Disable text filtering
filter.update_settings({"enable_text_filter": False})
self.assertEqual(filter.filter(input_text), input_text, "Text filtering should be disabled")
await filter.update_settings({"enable_text_filter": False})
self.assertEqual(
await filter.filter(input_text), input_text, "Text filtering should be disabled"
)
# Re-enable text filtering
filter.update_settings({"enable_text_filter": True})
await filter.update_settings({"enable_text_filter": True})
self.assertEqual(
filter.filter(input_text), "bold and italic", "Text filtering should be re-enabled"
await filter.filter(input_text),
"bold and italic",
"Text filtering should be re-enabled",
)

View File

@@ -5,7 +5,7 @@
#
import unittest
from unittest.mock import Mock
from unittest.mock import AsyncMock
from pipecat.utils.text.pattern_pair_aggregator import PatternMatch, PatternPairAggregator
@@ -13,7 +13,7 @@ from pipecat.utils.text.pattern_pair_aggregator import PatternMatch, PatternPair
class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self.aggregator = PatternPairAggregator()
self.test_handler = Mock()
self.test_handler = AsyncMock()
# Add a test pattern
self.aggregator.add_pattern_pair(
@@ -28,12 +28,12 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
async def test_pattern_match_and_removal(self):
# First part doesn't complete the pattern
result = self.aggregator.aggregate("Hello <test>pattern")
result = await self.aggregator.aggregate("Hello <test>pattern")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text, "Hello <test>pattern")
# Second part completes the pattern and includes an exclamation point
result = self.aggregator.aggregate(" content</test>!")
result = await self.aggregator.aggregate(" content</test>!")
# Verify the handler was called with correct PatternMatch object
self.test_handler.assert_called_once()
@@ -48,7 +48,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
self.assertEqual(result, "Hello !")
# Next sentence should be processed separately
result = self.aggregator.aggregate(" This is another sentence.")
result = await self.aggregator.aggregate(" This is another sentence.")
self.assertEqual(result, " This is another sentence.")
# Buffer should be empty after returning a complete sentence
@@ -56,7 +56,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
async def test_incomplete_pattern(self):
# Add text with incomplete pattern
result = self.aggregator.aggregate("Hello <test>pattern content")
result = await self.aggregator.aggregate("Hello <test>pattern content")
# No complete pattern yet, so nothing should be returned
self.assertIsNone(result)
@@ -68,13 +68,13 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
self.assertEqual(self.aggregator.text, "Hello <test>pattern content")
# Reset and confirm buffer is cleared
self.aggregator.reset()
await self.aggregator.reset()
self.assertEqual(self.aggregator.text, "")
async def test_multiple_patterns(self):
# Set up multiple patterns and handlers
voice_handler = Mock()
emphasis_handler = Mock()
voice_handler = AsyncMock()
emphasis_handler = AsyncMock()
self.aggregator.add_pattern_pair(
pattern_id="voice", start_pattern="<voice>", end_pattern="</voice>", remove_match=True
@@ -92,7 +92,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
# Test with multiple patterns in one text block
text = "Hello <voice>female</voice> I am <em>very</em> excited to meet you!"
result = self.aggregator.aggregate(text)
result = await self.aggregator.aggregate(text)
# Both handlers should be called with correct data
voice_handler.assert_called_once()
@@ -113,11 +113,11 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
async def test_handle_interruption(self):
# Start with incomplete pattern
result = self.aggregator.aggregate("Hello <test>pattern")
result = await self.aggregator.aggregate("Hello <test>pattern")
self.assertIsNone(result)
# Simulate interruption
self.aggregator.handle_interruption()
await self.aggregator.handle_interruption()
# Buffer should be cleared
self.assertEqual(self.aggregator.text, "")
@@ -127,13 +127,13 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
async def test_pattern_across_sentences(self):
# Test pattern that spans multiple sentences
result = self.aggregator.aggregate("Hello <test>This is sentence one.")
result = await self.aggregator.aggregate("Hello <test>This is sentence one.")
# First sentence contains start of pattern but no end, so no complete pattern yet
self.assertIsNone(result)
# Add second part with pattern end
result = self.aggregator.aggregate(" This is sentence two.</test> Final sentence.")
result = await self.aggregator.aggregate(" This is sentence two.</test> Final sentence.")
# Handler should be called with entire content
self.test_handler.assert_called_once()

View File

@@ -8,7 +8,8 @@ import asyncio
import time
import unittest
from pipecat.frames.frames import EndFrame, HeartbeatFrame, StartFrame, TextFrame
from pipecat.frames.frames import EndFrame, HeartbeatFrame, StartFrame, StopFrame, TextFrame
from pipecat.observers.base_observer import BaseObserver, FramePushed
from pipecat.pipeline.parallel_pipeline import ParallelPipeline
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -95,7 +96,114 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
await task.run()
assert task.has_finished()
async def test_task_event_handlers(self):
async def test_task_observers(self):
frame_received = False
class CustomObserver(BaseObserver):
async def on_push_frame(self, data: FramePushed):
nonlocal frame_received
if isinstance(data.frame, TextFrame):
frame_received = True
identity = IdentityFilter()
pipeline = Pipeline([identity])
task = PipelineTask(pipeline, observers=[CustomObserver()])
task.set_event_loop(asyncio.get_event_loop())
await task.queue_frames([TextFrame(text="Hello Downstream!"), EndFrame()])
await task.run()
assert frame_received
async def test_task_add_observer(self):
frame_received = False
frame_add_count = 0
class CustomObserver(BaseObserver):
async def on_push_frame(self, data: FramePushed):
nonlocal frame_received
if isinstance(data.frame, TextFrame):
frame_received = True
class CustomAddObserver(BaseObserver):
async def on_push_frame(self, data: FramePushed):
nonlocal frame_add_count
if isinstance(data.source, IdentityFilter) and isinstance(data.frame, TextFrame):
frame_add_count += 1
identity = IdentityFilter()
pipeline = Pipeline([identity])
task = PipelineTask(pipeline, observers=[CustomObserver()])
task.set_event_loop(asyncio.get_event_loop())
async def delayed_add_observer():
observer = CustomAddObserver()
# Wait after the pipeline is started and add an observer.
await asyncio.sleep(0.1)
await task.add_observer(observer)
# Push a TextFrame and wait for the observer to pick it up.
await task.queue_frame(TextFrame(text="Hello Downstream!"))
await asyncio.sleep(0.1)
# Remove the observer
await task.remove_observer(observer)
# Push another TextFrame. This time the counter should not
# increments since we have removed the observer.
await task.queue_frame(TextFrame(text="Hello Downstream!"))
await asyncio.sleep(0.1)
# Finally end the pipeline.
await task.queue_frame(EndFrame())
await asyncio.gather(task.run(), delayed_add_observer())
assert frame_received
assert frame_add_count == 1
async def test_task_started_ended_event_handler(self):
start_received = False
end_received = False
identity = IdentityFilter()
pipeline = Pipeline([identity])
task = PipelineTask(pipeline)
task.set_event_loop(asyncio.get_event_loop())
@task.event_handler("on_pipeline_started")
async def on_pipeline_started(task, frame: StartFrame):
nonlocal start_received
start_received = True
@task.event_handler("on_pipeline_ended")
async def on_pipeline_ended(task, frame: EndFrame):
nonlocal end_received
end_received = True
await task.queue_frame(EndFrame())
await task.run()
assert start_received
assert end_received
async def test_task_stopped_event_handler(self):
stop_received = False
identity = IdentityFilter()
pipeline = Pipeline([identity])
task = PipelineTask(pipeline)
task.set_event_loop(asyncio.get_event_loop())
@task.event_handler("on_pipeline_stopped")
async def on_pipeline_ended(task, frame: StopFrame):
nonlocal stop_received
stop_received = True
await task.queue_frame(StopFrame())
await task.run()
assert stop_received
async def test_task_frame_reached_event_handlers(self):
upstream_received = False
downstream_received = False

View File

@@ -14,16 +14,16 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase):
self.aggregator = SimpleTextAggregator()
async def test_reset_aggregations(self):
assert self.aggregator.aggregate("Hello ") == None
assert await self.aggregator.aggregate("Hello ") == None
assert self.aggregator.text == "Hello "
self.aggregator.reset()
await self.aggregator.reset()
assert self.aggregator.text == ""
async def test_simple_sentence(self):
assert self.aggregator.aggregate("Hello ") == None
assert self.aggregator.aggregate("Pipecat!") == "Hello Pipecat!"
assert await self.aggregator.aggregate("Hello ") == None
assert await self.aggregator.aggregate("Pipecat!") == "Hello Pipecat!"
assert self.aggregator.text == ""
async def test_multiple_sentences(self):
assert self.aggregator.aggregate("Hello Pipecat! How are ") == "Hello Pipecat!"
assert self.aggregator.aggregate("you?") == " How are you?"
assert await self.aggregator.aggregate("Hello Pipecat! How are ") == "Hello Pipecat!"
assert await self.aggregator.aggregate("you?") == " How are you?"

View File

@@ -14,41 +14,41 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase):
self.aggregator = SkipTagsAggregator([("<spell>", "</spell>")])
async def test_no_tags(self):
self.aggregator.reset()
await self.aggregator.reset()
# No tags involved, aggregate at end of sentence.
result = self.aggregator.aggregate("Hello Pipecat!")
result = await self.aggregator.aggregate("Hello Pipecat!")
self.assertEqual(result, "Hello Pipecat!")
self.assertEqual(self.aggregator.text, "")
async def test_basic_tags(self):
self.aggregator.reset()
await self.aggregator.reset()
# Tags involved, avoid aggregation during tags.
result = self.aggregator.aggregate("My email is <spell>foo@pipecat.ai</spell>.")
result = await self.aggregator.aggregate("My email is <spell>foo@pipecat.ai</spell>.")
self.assertEqual(result, "My email is <spell>foo@pipecat.ai</spell>.")
self.assertEqual(self.aggregator.text, "")
async def test_streaming_tags(self):
self.aggregator.reset()
await self.aggregator.reset()
# Tags involved, stream small chunk of texts.
result = self.aggregator.aggregate("My email is <sp")
result = await self.aggregator.aggregate("My email is <sp")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text, "My email is <sp")
result = self.aggregator.aggregate("ell>foo.")
result = await self.aggregator.aggregate("ell>foo.")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text, "My email is <spell>foo.")
result = self.aggregator.aggregate("bar@pipecat.")
result = await self.aggregator.aggregate("bar@pipecat.")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text, "My email is <spell>foo.bar@pipecat.")
result = self.aggregator.aggregate("ai</spe")
result = await self.aggregator.aggregate("ai</spe")
self.assertIsNone(result)
self.assertEqual(self.aggregator.text, "My email is <spell>foo.bar@pipecat.ai</spe")
result = self.aggregator.aggregate("ll>.")
result = await self.aggregator.aggregate("ll>.")
self.assertEqual(result, "My email is <spell>foo.bar@pipecat.ai</spell>.")
self.assertEqual(self.aggregator.text, "")

View File

@@ -64,6 +64,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
message = update_frame.messages[0]
self.assertEqual(message.role, "user")
self.assertEqual(message.content, "Hello, world!")
self.assertEqual(message.user_id, "test_user")
self.assertEqual(message.timestamp, timestamp)
async def test_event_handler(self):