diff --git a/README.md b/README.md index 15be8d3eb..5ea9fc4b9 100644 --- a/README.md +++ b/README.md @@ -91,7 +91,7 @@ Catch new features, interviews, and how-tos on our [Pipecat TV](https://www.yout | 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), [Cartesia](https://docs.pipecat.ai/server/services/stt/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/stt/elevenlabs), [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), [Gradium](https://docs.pipecat.ai/server/services/stt/gradium), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [Mistral](https://docs.pipecat.ai/server/services/stt/mistral), [NVIDIA Riva](https://docs.pipecat.ai/server/services/stt/riva), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Sarvam](https://docs.pipecat.ai/server/services/stt/sarvam), [Soniox](https://docs.pipecat.ai/server/services/stt/soniox), [Speechmatics](https://docs.pipecat.ai/server/services/stt/speechmatics), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | +| 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), [Cartesia](https://docs.pipecat.ai/server/services/stt/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/stt/elevenlabs), [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), [Gradium](https://docs.pipecat.ai/server/services/stt/gradium), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [Mistral](https://docs.pipecat.ai/server/services/stt/mistral), [NVIDIA Riva](https://docs.pipecat.ai/server/services/stt/riva), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Sarvam](https://docs.pipecat.ai/server/services/stt/sarvam), [Soniox](https://docs.pipecat.ai/server/services/stt/soniox), [Speechmatics](https://docs.pipecat.ai/server/services/stt/speechmatics), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper), [xAI](https://docs.pipecat.ai/server/services/stt/xai) | | 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), [Mistral](https://docs.pipecat.ai/server/services/llm/mistral), [Nebius](https://docs.pipecat.ai/server/services/llm/nebius), [Novita](https://docs.pipecat.ai/server/services/llm/novita), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nvidia), [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), [SambaNova](https://docs.pipecat.ai/server/services/llm/sambanova), [Sarvam](https://docs.pipecat.ai/server/services/llm/sarvam), [Together AI](https://docs.pipecat.ai/server/services/llm/together) | | Text-to-Speech | [Async](https://docs.pipecat.ai/server/services/tts/asyncai), [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Camb AI](https://docs.pipecat.ai/server/services/tts/camb), [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), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [Gradium](https://docs.pipecat.ai/server/services/tts/gradium), [Groq](https://docs.pipecat.ai/server/services/tts/groq), [Hume](https://docs.pipecat.ai/server/services/tts/hume), [Inworld](https://docs.pipecat.ai/server/services/tts/inworld), [Kokoro](https://docs.pipecat.ai/server/services/tts/kokoro), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [MiniMax](https://docs.pipecat.ai/server/services/tts/minimax), [Mistral](https://docs.pipecat.ai/server/services/tts/mistral), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [NVIDIA Riva](https://docs.pipecat.ai/server/services/tts/riva), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [Resemble](https://docs.pipecat.ai/server/services/tts/resemble), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [Sarvam](https://docs.pipecat.ai/server/services/tts/sarvam), [Smallest](https://docs.pipecat.ai/server/services/tts/smallest), [Speechmatics](https://docs.pipecat.ai/server/services/tts/speechmatics), [xAI](https://docs.pipecat.ai/server/services/tts/xai), [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), [Grok Voice Agent](https://docs.pipecat.ai/server/services/s2s/grok), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai), [Ultravox](https://docs.pipecat.ai/server/services/s2s/ultravox), | diff --git a/changelog/4340.added.md b/changelog/4340.added.md new file mode 100644 index 000000000..5abcc0f0e --- /dev/null +++ b/changelog/4340.added.md @@ -0,0 +1 @@ +- Added `XAISTTService` for real-time speech-to-text using xAI's voice STT WebSocket API (`wss://api.x.ai/v1/stt`). Streams raw audio (PCM, ยต-law, or A-law) and emits interim and final transcription frames driven by the server's `is_final` / `speech_final` flags. Settings expose `interim_results`, `endpointing`, `language`, `multichannel`, `channels`, and `diarize`. Requires the `xai` optional extra (`pip install "pipecat-ai[xai]"`). diff --git a/examples/transcription/transcription-xai.py b/examples/transcription/transcription-xai.py new file mode 100644 index 000000000..0cc667f9a --- /dev/null +++ b/examples/transcription/transcription-xai.py @@ -0,0 +1,86 @@ +# +# 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.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.frame_processor import FrameDirection, FrameProcessor +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.xai.stt import XAISTTService +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) + + +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 = XAISTTService( + api_key=os.getenv("XAI_API_KEY"), + ) + + tl = TranscriptionLogger() + + pipeline = Pipeline([transport.input(), 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/examples/voice/voice-xai.py b/examples/voice/voice-xai.py index 6d0d54e79..e7fa04350 100644 --- a/examples/voice/voice-xai.py +++ b/examples/voice/voice-xai.py @@ -22,8 +22,8 @@ from pipecat.processors.aggregators.llm_response_universal import ( ) from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport -from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.xai.llm import GrokLLMService +from pipecat.services.xai.stt import XAISTTService from pipecat.services.xai.tts import XAIHttpTTSService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams @@ -53,7 +53,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") async with aiohttp.ClientSession() as session: - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = XAISTTService(api_key=os.getenv("XAI_API_KEY")) tts = XAIHttpTTSService( api_key=os.getenv("XAI_API_KEY"), diff --git a/pyproject.toml b/pyproject.toml index b1ff2cbd2..205cf14c3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -123,7 +123,7 @@ webrtc = [ "aiortc>=1.14.0,<2", "opencv-python>=4.11.0.86,<5" ] websocket = [ "pipecat-ai[websockets-base]", "fastapi>=0.115.6,<1" ] websockets-base = [ "websockets>=13.1,<16.0" ] whisper = [ "faster-whisper~=1.2.1" ] -xai = [] +xai = [ "pipecat-ai[websockets-base]" ] [dependency-groups] dev = [ diff --git a/src/pipecat/services/stt_latency.py b/src/pipecat/services/stt_latency.py index 5ffd798de..7d285e353 100644 --- a/src/pipecat/services/stt_latency.py +++ b/src/pipecat/services/stt_latency.py @@ -57,3 +57,4 @@ WHISPER_TTFS_P99: float = DEFAULT_TTFS_P99 # No benchmark available yet; using conservative default MISTRAL_TTFS_P99: float = DEFAULT_TTFS_P99 SMALLEST_TTFS_P99: float = DEFAULT_TTFS_P99 +XAI_TTFS_P99: float = DEFAULT_TTFS_P99 diff --git a/src/pipecat/services/xai/stt.py b/src/pipecat/services/xai/stt.py new file mode 100644 index 000000000..df8406fc2 --- /dev/null +++ b/src/pipecat/services/xai/stt.py @@ -0,0 +1,411 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""xAI speech-to-text service implementation. + +This module provides integration with xAI's real-time speech-to-text WebSocket +API documented at https://docs.x.ai/developers/rest-api-reference/inference/voice. +""" + +import asyncio +import json +from collections.abc import AsyncGenerator +from dataclasses import dataclass, field +from typing import Any +from urllib.parse import urlencode + +from loguru import logger + +from pipecat import version as pipecat_version +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + Frame, + InterimTranscriptionFrame, + StartFrame, + TranscriptionFrame, +) +from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven +from pipecat.services.stt_latency import XAI_TTFS_P99 +from pipecat.services.stt_service import WebsocketSTTService +from pipecat.transcriptions.language import Language, resolve_language +from pipecat.utils.time import time_now_iso8601 +from pipecat.utils.tracing.service_decorators import traced_stt + +try: + from websockets.asyncio.client import connect as websocket_connect + from websockets.protocol import State +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error('In order to use xAI STT, you need to `pip install "pipecat-ai[xai]"`.') + raise Exception(f"Missing module: {e}") + + +def language_to_xai_stt_language(language: Language) -> str | None: + """Convert a Language enum to the xAI STT language code. + + xAI STT accepts two-letter language codes (e.g. ``en``, ``fr``, ``de``, + ``ja``). When set, the server applies Inverse Text Normalization. + + Args: + language: The Language enum value to convert. + + Returns: + The corresponding xAI STT language code, or None if not supported. + """ + LANGUAGE_MAP = { + Language.AR: "ar", + Language.BN: "bn", + Language.DE: "de", + Language.EN: "en", + Language.ES: "es", + Language.FR: "fr", + Language.HI: "hi", + Language.ID: "id", + Language.IT: "it", + Language.JA: "ja", + Language.KO: "ko", + Language.PT: "pt", + Language.RU: "ru", + Language.TR: "tr", + Language.VI: "vi", + Language.ZH: "zh", + } + + return resolve_language(language, LANGUAGE_MAP, use_base_code=True) + + +@dataclass +class XAISTTSettings(STTSettings): + """Settings for XAISTTService. + + Parameters: + interim_results: When True, partial transcripts are emitted + approximately every 500ms. + endpointing: Silence duration in milliseconds that triggers a + speech-final event. Range 0-5000. Server default is 10ms. + multichannel: When True, transcribes each interleaved channel + independently. Requires ``channels`` >= 2. + channels: Number of interleaved channels (2-8). Required when + ``multichannel`` is True. + diarize: When True, the server attaches a ``speaker`` field to each + word identifying the detected speaker. + """ + + interim_results: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + endpointing: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + multichannel: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + channels: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + diarize: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + + +class XAISTTService(WebsocketSTTService): + """xAI real-time speech-to-text service. + + Streams audio to xAI's WebSocket STT endpoint and emits interim and final + transcription frames. The ``XAI_API_KEY`` is passed directly as a Bearer + token on the WebSocket handshake. + + The connection is persistent: audio is streamed continuously and the + server emits ``transcript.partial`` events with ``is_final`` and + ``speech_final`` flags to mark utterance boundaries. If the connection + drops mid-session, the base class reconnects automatically. + """ + + Settings = XAISTTSettings + _settings: Settings + + def __init__( + self, + *, + api_key: str, + ws_url: str = "wss://api.x.ai/v1/stt", + sample_rate: int = 16000, + encoding: str = "pcm", + settings: Settings | None = None, + ttfs_p99_latency: float | None = XAI_TTFS_P99, + **kwargs, + ): + """Initialize the xAI STT service. + + Args: + api_key: xAI API key (used as Bearer for the WebSocket handshake). + ws_url: WebSocket endpoint URL. Defaults to ``wss://api.x.ai/v1/stt``. + sample_rate: Audio sample rate in Hz. Supported values: 8000, + 16000, 22050, 24000, 44100, 48000. Defaults to 16000. + encoding: Audio encoding. One of ``"pcm"`` (signed 16-bit LE), + ``"mulaw"``, or ``"alaw"``. Defaults to ``"pcm"``. + settings: Runtime-updatable settings overriding defaults. + ttfs_p99_latency: P99 latency from speech end to final transcript + in seconds. See https://github.com/pipecat-ai/stt-benchmark. + **kwargs: Additional arguments passed to WebsocketSTTService. + """ + default_settings = self.Settings( + model=None, + language=Language.EN, + interim_results=True, + endpointing=None, + multichannel=None, + channels=None, + diarize=None, + ) + if settings is not None: + default_settings.apply_update(settings) + + super().__init__( + sample_rate=sample_rate, + settings=default_settings, + ttfs_p99_latency=ttfs_p99_latency, + **kwargs, + ) + + self._api_key = api_key + self._ws_url = ws_url + self._encoding = encoding + + self._receive_task: asyncio.Task | None = None + self._session_ready = asyncio.Event() + + def can_generate_metrics(self) -> bool: + """Check if the service can generate metrics. + + Returns: + True if metrics generation is supported. + """ + return True + + def language_to_service_language(self, language: Language) -> str | None: + """Convert a Language enum to the xAI STT language code.""" + return language_to_xai_stt_language(language) + + async def _update_settings(self, delta: Settings) -> dict[str, Any]: + """Apply a settings delta and reconnect to apply changes. + + xAI STT configures the session via WebSocket query parameters, so any + change requires a fresh connection. + """ + changed = await super()._update_settings(delta) + if not changed: + return changed + await self._disconnect() + await self._connect() + return changed + + async def start(self, frame: StartFrame): + """Start the speech-to-text service.""" + await super().start(frame) + await self._connect() + + async def stop(self, frame: EndFrame): + """Stop the speech-to-text service.""" + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + """Cancel the speech-to-text service.""" + await super().cancel(frame) + await self._disconnect() + + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: + """Forward raw audio bytes to the xAI STT WebSocket. + + Transcription frames are pushed from the receive task, not yielded + from this coroutine. + """ + if self._websocket and self._websocket.state is State.OPEN and self._session_ready.is_set(): + try: + await self._websocket.send(audio) + except Exception as e: + await self.push_error(error_msg=f"xAI STT send failed: {e}", exception=e) + yield None + + def _build_ws_url(self) -> str: + """Build the WebSocket URL with session query parameters.""" + s = self._settings + params: dict[str, Any] = { + "sample_rate": self.sample_rate, + "encoding": self._encoding, + } + + if s.language is not None: + params["language"] = s.language + + optional_fields = { + "interim_results": s.interim_results, + "endpointing": s.endpointing, + "multichannel": s.multichannel, + "channels": s.channels, + "diarize": s.diarize, + } + for key, val in optional_fields.items(): + if val is None: + continue + if isinstance(val, bool): + params[key] = str(val).lower() + else: + params[key] = val + + return f"{self._ws_url}?{urlencode(params)}" + + async def _connect(self): + """Establish the WebSocket connection and start the receive task.""" + await super()._connect() + await self._connect_websocket() + if self._websocket and not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) + + async def _disconnect(self): + """Tear down the WebSocket connection and cancel the receive task.""" + await super()._disconnect() + try: + if self._websocket and self._websocket.state is State.OPEN: + await self._websocket.send(json.dumps({"type": "audio.done"})) + except Exception as e: + logger.debug(f"{self} error sending audio.done during disconnect: {e}") + + if self._receive_task: + await self.cancel_task(self._receive_task) + self._receive_task = None + + await self._disconnect_websocket() + + async def _connect_websocket(self): + """Open a WebSocket connection to the xAI STT endpoint.""" + try: + if self._websocket and self._websocket.state is State.OPEN: + return + + logger.debug("Connecting to xAI STT WebSocket") + self._session_ready.clear() + + ws_url = self._build_ws_url() + headers = { + "Authorization": f"Bearer {self._api_key}", + "User-Agent": f"xAI/1.0 (integration=Pipecat/{pipecat_version()})", + } + self._websocket = await websocket_connect(ws_url, additional_headers=headers) + await self._call_event_handler("on_connected") + logger.debug(f"{self} connected to xAI STT WebSocket") + except Exception as e: + await self.push_error(error_msg=f"Unable to connect to xAI STT: {e}", exception=e) + raise + + async def _disconnect_websocket(self): + """Close the WebSocket connection.""" + try: + if self._websocket: + logger.debug("Disconnecting from xAI STT WebSocket") + await self._websocket.close() + except Exception as e: + await self.push_error(error_msg=f"Error closing xAI STT websocket: {e}", exception=e) + finally: + self._websocket = None + self._session_ready.clear() + await self._call_event_handler("on_disconnected") + + async def _receive_messages(self): + """Receive and dispatch xAI STT WebSocket messages.""" + if not self._websocket: + raise Exception("Websocket not connected") + async for message in self._websocket: + try: + data = json.loads(message) + except json.JSONDecodeError: + logger.warning(f"{self} received non-JSON message: {message}") + continue + await self._handle_message(data) + + async def _handle_message(self, message: dict[str, Any]): + """Branch on xAI STT event type.""" + msg_type = message.get("type") + + if msg_type == "transcript.created": + self._session_ready.set() + logger.debug(f"{self} xAI STT session ready") + elif msg_type == "transcript.partial": + await self._handle_transcript(message) + elif msg_type == "transcript.done": + if message.get("text"): + await self._push_final_transcript(message, speech_final=True) + elif msg_type == "error": + await self.push_error( + error_msg=f"xAI STT error: {message.get('message', message)}", + exception=Exception(message), + ) + else: + logger.debug(f"{self} unhandled xAI STT message: {message}") + + async def _handle_transcript(self, message: dict[str, Any]): + text = message.get("text", "") + if not text: + return + + is_final = bool(message.get("is_final")) + speech_final = bool(message.get("speech_final")) + language = self._language_for_frame() + + if is_final: + await self._push_final_transcript( + message, speech_final=speech_final, language=language, text=text + ) + else: + await self.push_frame( + InterimTranscriptionFrame( + text, + self._user_id, + time_now_iso8601(), + language, + result=message, + ) + ) + + async def _push_final_transcript( + self, + message: dict[str, Any], + *, + speech_final: bool, + language: Language | None = None, + text: str | None = None, + ): + text = text if text is not None else message.get("text", "") + if not text: + return + language = language if language is not None else self._language_for_frame() + + await self.push_frame( + TranscriptionFrame( + text, + self._user_id, + time_now_iso8601(), + language, + result=message, + finalized=speech_final, + ) + ) + await self._trace_transcription(text, True, language) + if speech_final: + await self.stop_processing_metrics() + + def _language_for_frame(self) -> Language: + """Return a Language enum suitable for transcription frames. + + Settings stores the service-specific string (e.g. ``"en"``); frames + carry the enum value. + """ + lang = self._settings.language + if isinstance(lang, Language): + return lang + if isinstance(lang, str): + try: + return Language(lang) + except ValueError: + return Language.EN + return Language.EN + + @traced_stt + async def _trace_transcription(self, transcript: str, is_final: bool, language: Language): + """Record transcription event for tracing.""" + pass diff --git a/uv.lock b/uv.lock index 16e9e5e66..113fb056c 100644 --- a/uv.lock +++ b/uv.lock @@ -4403,6 +4403,9 @@ websockets-base = [ whisper = [ { name = "faster-whisper" }, ] +xai = [ + { name = "websockets" }, +] [package.dev-dependencies] dev = [ @@ -4507,6 +4510,7 @@ requires-dist = [ { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'soniox'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'ultravox'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'websocket'" }, + { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'xai'" }, { name = "pipecat-ai-small-webrtc-prebuilt", marker = "extra == 'runner'", specifier = ">=2.4.0" }, { name = "piper-tts", marker = "extra == 'piper'", specifier = ">=1.3.0,<2" }, { name = "protobuf", specifier = ">=6.31.1,<7" },