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

address changes

Browse Source
This commit is contained in:
Aaron Ng
2025-10-30 16:25:10 +00:00
parent b0acbeffb9
commit 9d509bb409
3 changed files with 182 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -6,6 +6,7 @@
import os import os
import aiohttp
from dotenv import load_dotenv from dotenv import load_dotenv
from loguru import logger from loguru import logger
@@ -89,90 +90,89 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
""" """
logger.info(f"Starting bot") logger.info(f"Starting bot")
async with aiohttp.ClientSession() as session:
stt = SpeechmaticsSTTService( stt = SpeechmaticsSTTService(
api_key=os.getenv("SPEECHMATICS_API_KEY"), api_key=os.getenv("SPEECHMATICS_API_KEY"),
params=SpeechmaticsSTTService.InputParams( params=SpeechmaticsSTTService.InputParams(
language=Language.EN, language=Language.EN,
enable_vad=True, enable_vad=True,
enable_diarization=True, enable_diarization=True,
focus_speakers=["S1"], focus_speakers=["S1"],
end_of_utterance_silence_trigger=0.5, end_of_utterance_silence_trigger=0.5,
speaker_active_format="<{speaker_id}>{text}</{speaker_id}>", speaker_active_format="<{speaker_id}>{text}</{speaker_id}>",
speaker_passive_format="<PASSIVE><{speaker_id}>{text}</{speaker_id}></PASSIVE>", speaker_passive_format="<PASSIVE><{speaker_id}>{text}</{speaker_id}></PASSIVE>",
),
)
tts = SpeechmaticsTTSService(
api_key=os.getenv("SPEECHMATICS_API_KEY"),
params=SpeechmaticsTTSService.InputParams(
voice="sarah",
),
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
params=BaseOpenAILLMService.InputParams(temperature=0.75),
)
messages = [
{
"role": "system",
"content": (
"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. "
"Give very short replies - do not give longer replies unless strictly necessary. "
"Respond to what the user said in a concise, funny, creative and helpful way. "
"Use `<Sn/>` tags to identify different speakers - do not use tags in your replies. "
"Do not respond to speakers within `<PASSIVE/>` tags unless explicitly asked to. "
), ),
}, )
]
context = LLMContext(messages) tts = SpeechmaticsTTSService(
context_aggregator = LLMContextAggregatorPair( api_key=os.getenv("SPEECHMATICS_API_KEY"),
context, voice_id="sarah",
user_params=LLMUserAggregatorParams(aggregation_timeout=0.005), aiohttp_session=session,
) )
pipeline = Pipeline( llm = OpenAILLMService(
[ api_key=os.getenv("OPENAI_API_KEY"),
transport.input(), # Transport user input params=BaseOpenAILLMService.InputParams(temperature=0.75),
stt, )
context_aggregator.user(), # User responses
llm, # LLM messages = [
tts, # TTS {
transport.output(), # Transport bot output "role": "system",
context_aggregator.assistant(), # Assistant spoken responses "content": (
"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. "
"Give very short replies - do not give longer replies unless strictly necessary. "
"Respond to what the user said in a concise, funny, creative and helpful way. "
"Use `<Sn/>` tags to identify different speakers - do not use tags in your replies. "
"Do not respond to speakers within `<PASSIVE/>` tags unless explicitly asked to. "
),
},
] ]
)
task = PipelineTask( context = LLMContext(messages)
pipeline, context_aggregator = LLMContextAggregatorPair(
params=PipelineParams( context,
enable_metrics=True, user_params=LLMUserAggregatorParams(aggregation_timeout=0.005),
enable_usage_metrics=True, )
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected") pipeline = Pipeline(
async def on_client_connected(transport, client): [
logger.info(f"Client connected") transport.input(), # Transport user input
# Kick off the conversation. stt,
messages.append({"role": "system", "content": "Say a short hello to the user."}) context_aggregator.user(), # User responses
await task.queue_frames([LLMRunFrame()]) llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses
]
)
@transport.event_handler("on_client_disconnected") task = PipelineTask(
async def on_client_disconnected(transport, client): pipeline,
logger.info(f"Client disconnected") params=PipelineParams(
await task.cancel() enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) @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": "Say a short hello to the user."})
await task.queue_frames([LLMRunFrame()])
await runner.run(task) @transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments): async def bot(runner_args: RunnerArguments):

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

@@ -6,6 +6,7 @@
import os import os
import aiohttp
from dotenv import load_dotenv from dotenv import load_dotenv
from loguru import logger from loguru import logger
@@ -82,85 +83,85 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
""" """
logger.info(f"Starting bot") logger.info(f"Starting bot")
stt = SpeechmaticsSTTService( async with aiohttp.ClientSession() as session:
api_key=os.getenv("SPEECHMATICS_API_KEY"), stt = SpeechmaticsSTTService(
params=SpeechmaticsSTTService.InputParams( api_key=os.getenv("SPEECHMATICS_API_KEY"),
language=Language.EN, params=SpeechmaticsSTTService.InputParams(
enable_diarization=True, language=Language.EN,
end_of_utterance_silence_trigger=0.5, enable_diarization=True,
speaker_active_format="<{speaker_id}>{text}</{speaker_id}>", end_of_utterance_silence_trigger=0.5,
), speaker_active_format="<{speaker_id}>{text}</{speaker_id}>",
)
tts = SpeechmaticsTTSService(
api_key=os.getenv("SPEECHMATICS_API_KEY"),
params=SpeechmaticsTTSService.InputParams(
voice="sarah",
),
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
params=BaseOpenAILLMService.InputParams(temperature=0.75),
)
messages = [
{
"role": "system",
"content": (
"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. "
"Give very short replies - do not give longer replies unless strictly necessary. "
"Respond to what the user said in a concise, funny, creative and helpful way. "
"Use `<Sn/>` tags to identify different speakers - do not use tags in your replies."
), ),
}, )
]
context = LLMContext(messages) tts = SpeechmaticsTTSService(
context_aggregator = LLMContextAggregatorPair( api_key=os.getenv("SPEECHMATICS_API_KEY"),
context, voice_id="sarah",
user_params=LLMUserAggregatorParams(aggregation_timeout=0.005), aiohttp_session=session,
) )
pipeline = Pipeline( llm = OpenAILLMService(
[ api_key=os.getenv("OPENAI_API_KEY"),
transport.input(), # Transport user input params=BaseOpenAILLMService.InputParams(temperature=0.75),
stt, # STT )
context_aggregator.user(), # User responses
llm, # LLM messages = [
tts, # TTS {
transport.output(), # Transport bot output "role": "system",
context_aggregator.assistant(), # Assistant spoken responses "content": (
"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. "
"Give very short replies - do not give longer replies unless strictly necessary. "
"Respond to what the user said in a concise, funny, creative and helpful way. "
"Use `<Sn/>` tags to identify different speakers - do not use tags in your replies."
),
},
] ]
)
task = PipelineTask( context = LLMContext(messages)
pipeline, context_aggregator = LLMContextAggregatorPair(
params=PipelineParams( context,
enable_metrics=True, user_params=LLMUserAggregatorParams(aggregation_timeout=0.005),
enable_usage_metrics=True, )
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected") pipeline = Pipeline(
async def on_client_connected(transport, client): [
logger.info(f"Client connected") transport.input(), # Transport user input
# Kick off the conversation. stt, # STT
messages.append({"role": "system", "content": "Say a short hello to the user."}) context_aggregator.user(), # User responses
await task.queue_frames([LLMRunFrame()]) llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses
]
)
@transport.event_handler("on_client_disconnected") task = PipelineTask(
async def on_client_disconnected(transport, client): pipeline,
logger.info(f"Client disconnected") params=PipelineParams(
await task.cancel() enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) @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": "Say a short hello to the user."})
await task.queue_frames([LLMRunFrame()])
await runner.run(task) @transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments): async def bot(runner_args: RunnerArguments):

85
src/pipecat/services/speechmatics/tts.py
View File

@@ -6,7 +6,6 @@
"""Speechmatics TTS service integration.""" """Speechmatics TTS service integration."""
import os
from typing import AsyncGenerator, Optional from typing import AsyncGenerator, Optional
from urllib.parse import urlencode from urllib.parse import urlencode
@@ -41,55 +40,56 @@ class SpeechmaticsTTSService(TTSService):
It converts text to speech and returns raw PCM audio data for real-time playback. It converts text to speech and returns raw PCM audio data for real-time playback.
""" """
SPEECHMATICS_SAMPLE_RATE = 16000
class InputParams(BaseModel): class InputParams(BaseModel):
"""Configuration parameters for Speechmatics TTS service. """Optional input parameters for Speechmatics TTS configuration."""
Parameters: pass
voice: Voice model to use for synthesis. Defaults to "sarah".
"""
voice: str = "sarah"
def __init__( def __init__(
self, self,
*, *,
api_key: str | None = None, api_key: str,
base_url: str | None = None, base_url: str = "https://preview.tts.speechmatics.com",
aiohttp_session: aiohttp.ClientSession | None = None, voice_id: str = "sarah",
sample_rate: Optional[int] = 16000, aiohttp_session: aiohttp.ClientSession,
params: InputParams | None = None, sample_rate: Optional[int] = SPEECHMATICS_SAMPLE_RATE,
params: Optional[InputParams] = None,
**kwargs, **kwargs,
): ):
"""Initialize the Speechmatics TTS service. """Initialize the Speechmatics TTS service.
Args: Args:
api_key: Speechmatics API key for authentication. Uses environment variable api_key: Speechmatics API key for authentication.
`SPEECHMATICS_API_KEY` if not provided. base_url: Base URL for Speechmatics TTS API.
base_url: Base URL for Speechmatics TTS API. Defaults to voice_id: Voice model to use for synthesis.
`https://preview.tts.speechmatics.com`.
aiohttp_session: Shared aiohttp session for HTTP requests. aiohttp_session: Shared aiohttp session for HTTP requests.
sample_rate: Audio sample rate in Hz. Defaults to 16000. sample_rate: Audio sample rate in Hz.
params: Optional[InputParams]: Input parameters for the service. params: Optional[InputParams]: Input parameters for the service.
**kwargs: Additional arguments passed to TTSService. **kwargs: Additional arguments passed to TTSService.
""" """
if sample_rate and sample_rate != self.SPEECHMATICS_SAMPLE_RATE:
logger.warning(
f"Speechmatics TTS only supports {self.SPEECHMATICS_SAMPLE_RATE}Hz sample rate. "
f"Current rate of {sample_rate}Hz may cause issues."
)
super().__init__(sample_rate=sample_rate, **kwargs) super().__init__(sample_rate=sample_rate, **kwargs)
# Service parameters # Service parameters
self._api_key: str = api_key or os.getenv("SPEECHMATICS_API_KEY") self._api_key: str = api_key
self._base_url: str = base_url or "https://preview.tts.speechmatics.com" self._base_url: str = base_url
self._session = aiohttp_session or aiohttp.ClientSession() self._session = aiohttp_session
# Check we have required attributes # Check we have required attributes
if not self._api_key: if not self._api_key:
raise ValueError("Missing Speechmatics API key") raise ValueError("Missing Speechmatics API key")
if not self._base_url:
raise ValueError("Missing Speechmatics base URL")
# Default parameters # Default parameters
self._params = params or SpeechmaticsTTSService.InputParams() self._params = params or SpeechmaticsTTSService.InputParams()
# Set voice from parameters # Set voice from constructor parameter
self.set_voice(self._params.voice) self.set_voice(voice_id)
def can_generate_metrics(self) -> bool: def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics. """Check if this service can generate processing metrics.
@@ -140,23 +140,6 @@ class SpeechmaticsTTSService(TTSService):
first_chunk = True first_chunk = True
buffer = b"" 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(): async for chunk in response.content.iter_any():
if not chunk: if not chunk:
continue continue
@@ -166,15 +149,19 @@ class SpeechmaticsTTSService(TTSService):
buffer += chunk buffer += chunk
# Emit a frame for all complete samples currently in buffer # Emit all complete 2-byte int16 samples from buffer
frame = _emit_complete_samples() if len(buffer) >= 2:
if frame: complete_samples = len(buffer) // 2
yield frame complete_bytes = complete_samples * 2
# Process any remaining bytes in buffer after streaming ends audio_data = buffer[:complete_bytes]
frame = _emit_complete_samples() buffer = buffer[complete_bytes:] # Keep remaining bytes for next iteration
if frame:
yield frame yield TTSAudioRawFrame(
audio=audio_data,
sample_rate=self.sample_rate,
num_channels=1,
)
except Exception as e: except Exception as e:
logger.exception(f"Error generating TTS: {e}") logger.exception(f"Error generating TTS: {e}")