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

Align Together STT/TTS services with Pipecat patterns

Browse Source 
STT:
- Add Settings class alias and 4-step init pattern
- Add resampler to convert pipeline audio to 16kHz for Together API
- Add keepalive support and _update_settings with reconnect
- Pass language to transcription frames
- Remove unnecessary OpenAI-Beta header

TTS:
- Add Settings class alias and 4-step init pattern
- Use push_start_frame=True for base class audio context management
- Route audio through append_to_audio_context instead of push_frame
- Track pending commits for proper audio context lifecycle
- Replace _handle_interruption with on_audio_context_interrupted
- Add _update_settings with reconnect
- Guard against stale audio after interruption
This commit is contained in:
Mark Backman
2026-03-20 22:22:27 -04:00
parent 4262410812
commit 4cb699b64c
5 changed files with 259 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
examples/foundational/07e-interruptible-together.py
View File

@@ -21,7 +21,6 @@ from pipecat.processors.aggregators.llm_response_universal import (
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.together.llm import TogetherLLMService
from pipecat.services.together.stt import TogetherSTTService
from pipecat.services.together.tts import TogetherTTSService
@@ -55,12 +54,17 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
stt = TogetherSTTService(api_key=os.getenv("TOGETHER_API_KEY"))
tts = TogetherTTSService(api_key=os.getenv("TOGETHER_API_KEY"))
tts = TogetherTTSService(
api_key=os.getenv("TOGETHER_API_KEY"),
settings=TogetherTTSService.Settings(
voice="tara",
),
)
llm = TogetherLLMService(
api_key=os.getenv("TOGETHER_API_KEY"),
settings=TogetherLLMService.Settings(
model="Qwen/Qwen3.5-9B",
model="openai/gpt-oss-120b",
system_instruction="You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
),
)

83
examples/foundational/13n-together-transcription.py Normal file
View File

@@ -0,0 +1,83 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import Frame, TranscriptionFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.audio.vad_processor import VADProcessor
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.together.stt import TogetherSTTService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
load_dotenv(override=True)
class TranscriptionLogger(FrameProcessor):
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, TranscriptionFrame):
print(f"Transcription: {frame.text}")
# Push all frames through
await self.push_frame(frame, direction)
# We use lambdas to defer transport parameter creation until the transport
# type is selected at runtime.
transport_params = {
"daily": lambda: DailyParams(audio_in_enabled=True),
"twilio": lambda: FastAPIWebsocketParams(audio_in_enabled=True),
"webrtc": lambda: TransportParams(audio_in_enabled=True),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = TogetherSTTService(api_key=os.getenv("TOGETHER_API_KEY"))
tl = TranscriptionLogger()
vad_processor = VADProcessor(vad_analyzer=SileroVADAnalyzer())
pipeline = Pipeline([transport.input(), vad_processor, stt, tl])
task = PipelineTask(
pipeline,
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@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):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

4
src/pipecat/services/together/llm.py
View File

@@ -56,7 +56,9 @@ class TogetherLLMService(OpenAILLMService):
**kwargs: Additional keyword arguments passed to OpenAILLMService.
"""
# 1. Initialize default_settings with hardcoded defaults
default_settings = self.Settings(model=model)
default_settings = self.Settings(
model="openai/gpt-oss-120b",
)
# 2. Apply direct init arg overrides (deprecated)
if model is not None:

93
src/pipecat/services/together/stt.py
View File

@@ -9,10 +9,11 @@
import base64
import json
from dataclasses import dataclass
from typing import AsyncGenerator, Optional
from typing import Any, AsyncGenerator, Optional
from loguru import logger
from pipecat.audio.utils import create_stream_resampler
from pipecat.services.settings import STTSettings
from pipecat.services.stt_latency import TOGETHER_TTFS_P99
@@ -40,6 +41,9 @@ from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
# Together requires 16 kHz 16-bit mono PCM input.
_TOGETHER_SAMPLE_RATE = 16000
@dataclass
class TogetherSTTSettings(STTSettings):
@@ -50,8 +54,7 @@ class TogetherSTTSettings(STTSettings):
language: Language of the audio input.
"""
model: str = "openai/whisper-large-v3"
language: Language = Language.EN
pass
class TogetherSTTService(WebsocketSTTService):
@@ -59,18 +62,26 @@ class TogetherSTTService(WebsocketSTTService):
Provides real-time speech recognition using Together AI's WebSocket API
with OpenAI-compatible speech-to-text endpoints.
Example::
stt = TogetherSTTService(
api_key="...",
settings=TogetherSTTService.Settings(
model="openai/whisper-large-v3",
),
)
"""
_settings: TogetherSTTSettings
Settings = TogetherSTTSettings
_settings: Settings
def __init__(
self,
*,
api_key: str,
model: str = "openai/whisper-large-v3",
language: Language = Language.EN,
sample_rate: int = 16000,
base_url: str = "wss://api.together.xyz/v1",
base_url: str = "wss://api.together.ai/v1",
settings: Optional[Settings] = None,
ttfs_p99_latency: float = TOGETHER_TTFS_P99,
**kwargs,
):
@@ -78,27 +89,34 @@ class TogetherSTTService(WebsocketSTTService):
Args:
api_key: Together AI API key for authentication.
model: Together AI transcription model. Defaults to "openai/whisper-large-v3".
language: Language of the audio input. Defaults to English.
sample_rate: Audio sample rate (default: 16000). Together AI requires 16kHz input.
base_url: The URL of the Together AI WebSocket API.
settings: Runtime-updatable settings for model and language configuration.
ttfs_p99_latency: P99 latency from speech end to final transcript in seconds.
Override for your deployment. See https://github.com/pipecat-ai/stt-benchmark
**kwargs: Additional arguments passed to the parent WebsocketSTTService.
"""
# Hardcoded defaults
default_settings = self.Settings(
model="openai/whisper-large-v3",
language=Language.EN,
)
# Apply settings delta
if settings is not None:
default_settings.apply_update(settings)
super().__init__(
sample_rate=sample_rate,
ttfs_p99_latency=ttfs_p99_latency,
settings=TogetherSTTSettings(
model=model,
language=language,
),
keepalive_timeout=20,
keepalive_interval=5,
settings=default_settings,
**kwargs,
)
self._api_key = api_key
self._base_url = base_url
self._receive_task = None
self._resampler = create_stream_resampler()
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -108,6 +126,39 @@ class TogetherSTTService(WebsocketSTTService):
"""
return True
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
"""Apply a settings delta and reconnect to apply changes.
Together passes model/language as URL query params, so a reconnect
is needed to apply changes.
Args:
delta: A settings delta with updated values.
Returns:
Dict mapping changed field names to their previous values.
"""
changed = await super()._update_settings(delta)
if not changed:
return changed
# Reconnect to apply updated settings (they become WS URL params)
await self._disconnect()
await self._connect()
return changed
async def _send_keepalive(self, silence: bytes):
"""Send silent audio to keep the Together AI connection alive.
Wraps silence in the ``input_audio_buffer.append`` JSON protocol.
Args:
silence: Silent 16-bit mono PCM audio bytes.
"""
await self._send_audio(silence)
async def start(self, frame: StartFrame):
"""Start the Together AI STT service.
@@ -193,7 +244,6 @@ class TogetherSTTService(WebsocketSTTService):
)
headers = {
"Authorization": f"Bearer {self._api_key}",
"OpenAI-Beta": "realtime=v1",
}
self._websocket = await websocket_connect(url, additional_headers=headers)
@@ -225,11 +275,18 @@ class TogetherSTTService(WebsocketSTTService):
async def _send_audio(self, audio: bytes):
"""Send audio data via ``input_audio_buffer.append``.
Resamples from the pipeline sample rate to 16 kHz if needed.
Args:
audio: Raw audio bytes at the pipeline sample rate.
"""
try:
if not self._disconnecting and self._websocket:
audio = await self._resampler.resample(
audio, self.sample_rate, _TOGETHER_SAMPLE_RATE
)
if not audio:
return
payload = base64.b64encode(audio).decode("utf-8")
await self._websocket.send(
json.dumps({"type": "input_audio_buffer.append", "audio": payload})
@@ -330,6 +387,7 @@ class TogetherSTTService(WebsocketSTTService):
delta,
self._user_id,
time_now_iso8601(),
self._settings.language,
result=evt,
)
)
@@ -347,6 +405,7 @@ class TogetherSTTService(WebsocketSTTService):
transcript,
self._user_id,
time_now_iso8601(),
self._settings.language,
result=evt,
finalized=True,
)

134
src/pipecat/services/together/tts.py
View File

@@ -9,7 +9,7 @@
import base64
import json
from dataclasses import dataclass
from typing import AsyncGenerator, Optional
from typing import Any, AsyncGenerator, Optional
from loguru import logger
@@ -27,13 +27,10 @@ from pipecat.frames.frames import (
CancelFrame,
EndFrame,
Frame,
InterruptionFrame,
StartFrame,
TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import WebsocketTTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.tracing.service_decorators import traced_tts
@@ -44,15 +41,9 @@ class TogetherTTSSettings(TTSSettings):
"""Settings for the Together AI TTS service.
Parameters:
model: Together AI TTS model to use.
voice: Voice to use for synthesis.
language: Language of the text input.
max_partial_length: Maximum partial text length for streaming.
"""
model: str = "canopylabs/orpheus-3b-0.1-ft"
language: Optional[Language] = Language.EN
voice: Optional[str] = "tara"
max_partial_length: Optional[int] = None
@@ -63,40 +54,42 @@ class TogetherTTSService(WebsocketTTSService):
Supports streaming synthesis with configurable voice and model options.
"""
_settings: TogetherTTSSettings
Settings = TogetherTTSSettings
_settings: Settings
def __init__(
self,
*,
api_key: str,
model: str = "canopylabs/orpheus-3b-0.1-ft",
voice: str = "tara",
language: Optional[Language] = Language.EN,
max_partial_length: Optional[int] = None,
url: str = "wss://api.together.ai/v1/audio/speech/websocket",
sample_rate: Optional[int] = 24000,
sample_rate: Optional[int] = None,
settings: Optional[Settings] = None,
**kwargs,
):
"""Initialize the Together AI TTS service.
Args:
api_key: Together AI API key for authentication.
model: Together AI TTS model. Defaults to "canopylabs/orpheus-3b-0.1-ft".
voice: Voice to use for synthesis. Defaults to "tara".
language: Language of the text input. Defaults to English.
max_partial_length: Maximum partial text length for streaming.
url: WebSocket URL for Together AI TTS API.
sample_rate: Audio sample rate (default: 24000).
settings: Runtime-updatable settings for model, voice, and language configuration.
**kwargs: Additional arguments passed to the parent service.
"""
# Hardcoded defaults
default_settings = self.Settings(
model="canopylabs/orpheus-3b-0.1-ft",
voice="tara",
language=Language.EN,
)
# Apply settings delta
if settings is not None:
default_settings.apply_update(settings)
super().__init__(
sample_rate=sample_rate,
settings=TogetherTTSSettings(
model=model,
voice=voice,
language=language,
max_partial_length=max_partial_length,
),
push_start_frame=True,
settings=default_settings,
**kwargs,
)
@@ -105,6 +98,8 @@ class TogetherTTSService(WebsocketTTSService):
self._session_id = None
self._receive_task = None
self._context_id: Optional[str] = None
self._pending_commits = 0
self._flush_context_id: Optional[str] = None
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -114,6 +109,29 @@ class TogetherTTSService(WebsocketTTSService):
"""
return True
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
"""Apply a settings delta and reconnect to apply changes.
Together passes model/voice as URL query params, so a reconnect
is needed to apply changes.
Args:
delta: A settings delta with updated values.
Returns:
Dict mapping changed field names to their previous values.
"""
changed = await super()._update_settings(delta)
if not changed:
return changed
# Reconnect to apply updated settings (they become WS URL params)
await self._disconnect()
await self._connect()
return changed
def _build_websocket_url(self) -> str:
"""Build the WebSocket URL with query parameters."""
url = f"{self._url}?model={self._settings.model}&voice={self._settings.voice}"
@@ -153,7 +171,7 @@ class TogetherTTSService(WebsocketTTSService):
# ------------------------------------------------------------------
async def _connect(self):
"""Connect to the transcription endpoint and start receiving."""
"""Connect to the TTS endpoint and start receiving."""
await super()._connect()
await self._connect_websocket()
if self._websocket and not self._receive_task:
@@ -231,12 +249,22 @@ class TogetherTTSService(WebsocketTTSService):
async def flush_audio(self, context_id: Optional[str] = None):
"""Flush any pending audio and finalize the current context.
If all server-side commits have been completed, closes the audio context
immediately. Otherwise, marks the context for deferred closure so
``_handle_audio_done`` can close it when the last commit finishes.
Args:
context_id: Pipecat TTS context (unused for Together; required for
compatibility with :meth:`TTSService.on_turn_context_completed`).
context_id: Pipecat TTS context to flush.
"""
logger.trace(f"{self}: flushing audio (context_id={context_id})")
await self._ws_send({"type": "input_text_buffer.commit"})
ctx_id = context_id or self._context_id
if not ctx_id or not self.audio_context_available(ctx_id):
return
logger.trace(f"{self}: flushing audio (context_id={ctx_id})")
if self._pending_commits == 0:
await self.append_to_audio_context(ctx_id, TTSStoppedFrame(context_id=ctx_id))
await self.remove_audio_context(ctx_id)
else:
self._flush_context_id = ctx_id
# ------------------------------------------------------------------
# Server event handling
@@ -307,10 +335,11 @@ class TogetherTTSService(WebsocketTTSService):
Args:
evt: The delta event from the server.
"""
if not self._context_id or not self.audio_context_available(self._context_id):
return
delta = evt.get("delta")
if delta:
try:
await self.stop_ttfb_metrics()
audio_chunk = base64.b64decode(delta)
frame = TTSAudioRawFrame(
audio=audio_chunk,
@@ -318,19 +347,25 @@ class TogetherTTSService(WebsocketTTSService):
num_channels=1,
context_id=self._context_id,
)
await self.push_frame(frame)
await self.append_to_audio_context(self._context_id, frame)
except Exception as e:
logger.error(f"{self} error processing audio delta: {e}")
async def _handle_audio_done(self, evt: dict):
"""Handle audio output completion for a speech segment.
Decrements the pending commit counter and closes the audio context
if a flush was requested and this was the last pending commit.
Args:
evt: The done event from the server.
"""
if not self._context_id or not self.audio_context_available(self._context_id):
return
item_id = evt.get("item_id")
logger.debug(f"{self} audio generation complete for: {item_id}")
await self.push_frame(TTSStoppedFrame(context_id=self._context_id))
self._pending_commits = max(0, self._pending_commits - 1)
await self._maybe_close_context()
async def _handle_tts_failed(self, evt: dict):
"""Handle a TTS failure.
@@ -339,8 +374,9 @@ class TogetherTTSService(WebsocketTTSService):
evt: The failed event containing error details.
"""
error = evt.get("error", {})
self._pending_commits = max(0, self._pending_commits - 1)
await self.push_error(error_msg=f"TTS error: {error}")
await self.push_frame(TTSStoppedFrame(context_id=self._context_id))
await self._maybe_close_context()
async def _handle_error(self, evt: dict):
"""Handle a fatal error from the TTS session.
@@ -358,19 +394,27 @@ class TogetherTTSService(WebsocketTTSService):
await self.push_error(error_msg=msg)
raise Exception(msg)
async def _maybe_close_context(self):
"""Close the audio context if a flush was requested and no commits remain."""
if self._pending_commits == 0 and self._flush_context_id:
ctx_id = self._flush_context_id
self._flush_context_id = None
await self.append_to_audio_context(ctx_id, TTSStoppedFrame(context_id=ctx_id))
await self.remove_audio_context(ctx_id)
# ------------------------------------------------------------------
# Interruption handling
# ------------------------------------------------------------------
async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
"""Handle interruption by canceling current generation.
async def on_audio_context_interrupted(self, context_id: str):
"""Cancel current generation when the bot is interrupted.
Args:
frame: The interruption frame.
direction: Frame processing direction.
context_id: The ID of the audio context that was interrupted.
"""
await super()._handle_interruption(frame, direction)
await self.stop_all_metrics()
self._pending_commits = 0
self._flush_context_id = None
await self._ws_send({"type": "input_text_buffer.clear"})
# ------------------------------------------------------------------
@@ -381,12 +425,15 @@ class TogetherTTSService(WebsocketTTSService):
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Together AI's streaming API.
Audio frames are delivered asynchronously via the WebSocket receive
loop and routed through the audio context managed by the base class.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
Frame: None (audio arrives via WebSocket callbacks).
"""
logger.debug(f"{self}: Generating TTS [{text}]")
@@ -399,9 +446,7 @@ class TogetherTTSService(WebsocketTTSService):
return
self._context_id = context_id
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=context_id)
self._pending_commits += 1
try:
await self._ws_send({"type": "input_text_buffer.append", "text": text})
@@ -409,6 +454,7 @@ class TogetherTTSService(WebsocketTTSService):
await self.start_tts_usage_metrics(text)
except Exception as e:
logger.error(f"{self} error sending message: {e}")
self._pending_commits -= 1
yield TTSStoppedFrame(context_id=context_id)
await self._disconnect()
await self._connect()