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

add speechmatics tts

Browse Source
This commit is contained in:
Aaron Ng
2025-10-29 14:53:20 +00:00
parent 493d6bf91e
commit 4050e8b7dc
3 changed files with 217 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
examples/foundational/07a-interruptible-speechmatics-vad.py
View File

@@ -20,10 +20,10 @@ from pipecat.processors.aggregators.llm_response import (
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
from pipecat.services.openai.base_llm import BaseOpenAILLMService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.speechmatics.stt import SpeechmaticsSTTService
from pipecat.services.speechmatics.tts import SpeechmaticsTTSService
from pipecat.transcriptions.language import Language
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
@@ -51,35 +51,41 @@ transport_params = {
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
"""Speechmatics STT Service Example
"""Speechmatics STT and TTS Service Example
This example demonstrates using Speechmatics Speech-to-Text service with speaker diarization and intelligent speaker management. Key features:
This example demonstrates using Speechmatics Speech-to-Text and Text-to-Speech services
with speaker diarization and intelligent speaker management. Key features:
1. Speaker Diarization
1. Speaker Diarization (STT)
- Automatically identifies and distinguishes between different speakers
- First speaker is identified as 'S1', others get subsequent IDs
- Uses `enable_diarization` parameter to manage speaker detection
2. Smart Speaker Control
2. Smart Speaker Control (STT)
- `focus_speakers` parameter lets you target specific speakers (e.g. ["S1"])
- Other speakers will be wrapped in PASSIVE tags
- Only processes speech from focused speakers
- Words from all speakers are wrapped with XML tags for clear speaker identification
- Other speakers' speech only sent when focused speaker is active
3. Voice Activity Detection
3. Voice Activity Detection (STT)
- Built-in VAD using `enable_vad` parameter
- Remove `vad_analyzer` from `transport` config to use module's VAD
- Emits speaker started/stopped events
4. Configuration Options
4. Text-to-Speech (TTS)
- Low latency streaming audio synthesis
- Multiple voice options available including `sarah`, `theo`, and `megan`
5. Configuration Options
- `operating_point` parameter defaults to `ENHANCED` for optimal accuracy
- Configurable `end_of_utterance_silence_trigger` (default 0.5s)
- Customizable speaker formatting
- Additional diarization settings available
For detailed information about operating points and configuration:
https://docs.speechmatics.com/rt-api-ref
For detailed information:
- STT: https://docs.speechmatics.com/rt-api-ref
- TTS: https://docs.speechmatics.com/text-to-speech/quickstart
"""
logger.info(f"Starting bot")
@@ -97,10 +103,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
),
)
tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
model="eleven_turbo_v2_5",
tts = SpeechmaticsTTSService(
api_key=os.getenv("SPEECHMATICS_API_KEY"),
params=SpeechmaticsTTSService.InputParams(
voice="sarah",
),
)
llm = OpenAILLMService(
@@ -112,7 +119,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
{
"role": "system",
"content": (
"You are a helpful British assistant called Alfred. "
"You are a helpful British assistant called Sarah. "
"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. "
"Always include punctuation in your responses. "

38
examples/foundational/07a-interruptible-speechmatics.py
View File

@@ -24,10 +24,10 @@ from pipecat.processors.aggregators.llm_response import (
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
from pipecat.services.openai.base_llm import BaseOpenAILLMService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.speechmatics.stt import SpeechmaticsSTTService
from pipecat.services.speechmatics.tts import SpeechmaticsTTSService
from pipecat.transcriptions.language import Language
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
@@ -61,19 +61,24 @@ transport_params = {
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
"""Run example using Speechmatics STT.
"""Run example using Speechmatics STT and TTS.
This example will use diarization within our STT service and output the words spoken by
each individual speaker and wrap them with XML tags for the LLM to process. Note the
instructions in the system context for the LLM. This greatly improves the conversation
experience by allowing the LLM to understand who is speaking in a multi-party call.
This example demonstrates a complete Speechmatics integration with both Speech-to-Text
and Text-to-Speech services:
By default, this example will use our ENHANCED operating point, which is optimized for
high accuracy. You can change this by setting the `operating_point` parameter to a different
value.
STT Features:
- Diarization to identify and distinguish between different speakers
- Words spoken by each speaker are wrapped with XML tags for LLM processing
- System context instructions help the LLM understand multi-party conversations
- ENHANCED operating point by default for optimal accuracy
For more information on operating points, see the Speechmatics documentation:
https://docs.speechmatics.com/rt-api-ref
TTS Features:
- Low latency streaming audio synthesis
- Multiple voice options available including `sarah`, `theo`, and `megan`
For more information:
- STT: https://docs.speechmatics.com/rt-api-ref
- TTS: https://docs.speechmatics.com/text-to-speech/quickstart
"""
logger.info(f"Starting bot")
@@ -87,10 +92,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
),
)
tts = ElevenLabsTTSService(
api_key=os.getenv("ELEVENLABS_API_KEY"),
voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
model="eleven_turbo_v2_5",
tts = SpeechmaticsTTSService(
api_key=os.getenv("SPEECHMATICS_API_KEY"),
params=SpeechmaticsTTSService.InputParams(
voice="sarah",
),
)
llm = OpenAILLMService(
@@ -102,7 +108,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
{
"role": "system",
"content": (
"You are a helpful British assistant called Alfred. "
"You are a helpful British assistant called Sarah. "
"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. "
"Always include punctuation in your responses. "

174
src/pipecat/services/speechmatics/tts.py Normal file
View File

@@ -0,0 +1,174 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Speechmatics TTS service integration."""
import os
from typing import AsyncGenerator, Optional
import aiohttp
import numpy as np
from loguru import logger
from pydantic import BaseModel
from pipecat.frames.frames import (
ErrorFrame,
Frame,
TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
)
from pipecat.services.tts_service import TTSService
from pipecat.utils.tracing.service_decorators import traced_tts
class SpeechmaticsTTSService(TTSService):
"""Speechmatics TTS service implementation.
This service provides text-to-speech synthesis using the Speechmatics HTTP API.
It converts text to speech and returns raw PCM audio data for real-time playback.
"""
class InputParams(BaseModel):
"""Configuration parameters for Speechmatics TTS service.
Parameters:
voice: Voice model to use for synthesis. Defaults to "sarah".
"""
voice: str = "sarah"
def __init__(
self,
*,
api_key: str | None = None,
base_url: str | None = None,
aiohttp_session: aiohttp.ClientSession | None = None,
sample_rate: Optional[int] = 16000,
params: InputParams | None = None,
**kwargs,
):
"""Initialize the Speechmatics TTS service.
Args:
api_key: Speechmatics API key for authentication. Uses environment variable
`SPEECHMATICS_API_KEY` if not provided.
base_url: Base URL for Speechmatics TTS API. Defaults to
`https://preview.tts.speechmatics.com`.
aiohttp_session: Shared aiohttp session for HTTP requests.
sample_rate: Audio sample rate in Hz. Defaults to 16000.
params: Optional[InputParams]: Input parameters for the service.
**kwargs: Additional arguments passed to TTSService.
"""
super().__init__(sample_rate=sample_rate, **kwargs)
# Service parameters
self._api_key: str = api_key or os.getenv("SPEECHMATICS_API_KEY")
self._base_url: str = base_url or "https://preview.tts.speechmatics.com"
self._session = aiohttp_session or aiohttp.ClientSession()
# Check we have required attributes
if not self._api_key:
raise ValueError("Missing Speechmatics API key")
if not self._base_url:
raise ValueError("Missing Speechmatics base URL")
# Default parameters
self._params = params or SpeechmaticsTTSService.InputParams()
# Set voice from parameters
self.set_voice(self._params.voice)
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
Returns:
True, as Speechmatics service supports metrics generation.
"""
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Speechmatics' HTTP API.
Args:
text: The text to synthesize into speech.
Yields:
Frame: Audio frames containing the synthesized speech.
"""
logger.debug(f"{self}: Generating TTS [{text}]")
headers = {
"Authorization": f"Bearer {self._api_key}",
"Content-Type": "application/json",
}
payload = {
"text": text,
}
url = f"{self._base_url}/generate/{self._voice_id}?output_format=pcm_16000"
try:
await self.start_ttfb_metrics()
async with self._session.post(url, json=payload, headers=headers) as response:
if response.status != 200:
error_message = f"Speechmatics TTS error: HTTP {response.status}"
logger.error(error_message)
yield ErrorFrame(error=error_message)
return
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
# Process the response in streaming chunks
first_chunk = True
buffer = b""
# Helper to move all complete 2-byte int16 samples from buffer into a frame
def _emit_complete_samples():
nonlocal buffer
if len(buffer) < 2:
return None
complete_samples = len(buffer) // 2
complete_bytes = complete_samples * 2
audio_data = buffer[:complete_bytes]
buffer = buffer[complete_bytes:] # Keep remaining bytes for next iteration
return TTSAudioRawFrame(
audio=audio_data,
sample_rate=self.sample_rate,
num_channels=1,
)
async for chunk in response.content.iter_any():
if not chunk:
continue
if first_chunk:
await self.stop_ttfb_metrics()
first_chunk = False
buffer += chunk
# Emit a frame for all complete samples currently in buffer
frame = _emit_complete_samples()
if frame:
yield frame
# Process any remaining bytes in buffer after streaming ends
frame = _emit_complete_samples()
if frame:
yield frame
except Exception as e:
logger.exception(f"Error generating TTS: {e}")
yield ErrorFrame(error=f"Speechmatics TTS error: {str(e)}")
finally:
yield TTSStoppedFrame()