diff --git a/examples/foundational/07e-interruptible-together.py b/examples/foundational/07e-interruptible-together.py index e55d4dd1b..63254b350 100644 --- a/examples/foundational/07e-interruptible-together.py +++ b/examples/foundational/07e-interruptible-together.py @@ -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.", ), ) diff --git a/examples/foundational/13n-together-transcription.py b/examples/foundational/13n-together-transcription.py new file mode 100644 index 000000000..501d11035 --- /dev/null +++ b/examples/foundational/13n-together-transcription.py @@ -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() diff --git a/src/pipecat/services/together/llm.py b/src/pipecat/services/together/llm.py index 1e296769b..3faae5fff 100644 --- a/src/pipecat/services/together/llm.py +++ b/src/pipecat/services/together/llm.py @@ -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: diff --git a/src/pipecat/services/together/stt.py b/src/pipecat/services/together/stt.py index 0e20e4b8b..2a6907f90 100644 --- a/src/pipecat/services/together/stt.py +++ b/src/pipecat/services/together/stt.py @@ -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, ) diff --git a/src/pipecat/services/together/tts.py b/src/pipecat/services/together/tts.py index 171ba6b7b..494a6564e 100644 --- a/src/pipecat/services/together/tts.py +++ b/src/pipecat/services/together/tts.py @@ -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()