# Copyright 2023 LiveKit, Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import annotations import asyncio import base64 import json import os import time import weakref from dataclasses import dataclass from typing import Any from urllib.parse import urlencode import aiohttp import httpx import openai from livekit import rtc from livekit.agents import ( DEFAULT_API_CONNECT_OPTIONS, APIConnectionError, APIConnectOptions, APIError, APIStatusError, APITimeoutError, LanguageCode, stt, utils, vad, ) from livekit.agents.types import ( NOT_GIVEN, NotGivenOr, ) from livekit.agents.utils import AudioBuffer, is_given from openai.types.audio import TranscriptionVerbose from openai.types.beta.realtime.transcription_session_update_param import ( SessionTurnDetection, ) from .log import logger from .models import STTModels from .utils import AsyncAzureADTokenProvider # OpenAI Realtime API has a timeout of 15 mins, we'll attempt to restart the session # before that timeout is reached _max_session_duration = 10 * 60 # emit interim transcriptions every 0.5 seconds _delta_transcript_interval = 0.5 SAMPLE_RATE = 24000 NUM_CHANNELS = 1 def _is_whisper_realtime(model: str) -> bool: # gpt-realtime-whisper rejects any turn_detection config; the client must # commit the audio buffer manually (e.g. driven by an external VAD). return model.startswith("gpt-realtime-whisper") @dataclass class _STTOptions: model: STTModels | str language: LanguageCode detect_language: bool turn_detection: SessionTurnDetection prompt: NotGivenOr[str] = NOT_GIVEN noise_reduction_type: NotGivenOr[str] = NOT_GIVEN temperature: NotGivenOr[float] = NOT_GIVEN class STT(stt.STT): def __init__( self, *, language: str = "en", detect_language: bool = False, model: STTModels | str = "gpt-4o-mini-transcribe", prompt: NotGivenOr[str] = NOT_GIVEN, turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN, noise_reduction_type: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, base_url: NotGivenOr[str] = NOT_GIVEN, api_key: NotGivenOr[str] = NOT_GIVEN, client: openai.AsyncClient | None = None, use_realtime: bool = False, vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN, ): """ Create a new instance of OpenAI STT. Args: language: The language code to use for transcription (e.g., "en" for English). detect_language: Whether to automatically detect the language. model: The OpenAI model to use for transcription. prompt: Optional text prompt to guide the transcription. Only supported for whisper-1. turn_detection: When using realtime transcription, this controls how model detects the user is done speaking. Final transcripts are generated only after the turn is over. See: https://platform.openai.com/docs/guides/realtime-vad Ignored for `gpt-realtime-whisper`, which does not support server-side turn detection. noise_reduction_type: Type of noise reduction to apply. "near_field" or "far_field" This isn't needed when using LiveKit's noise cancellation. temperature: Sampling temperature between 0 and 1. Lower values make the transcription more deterministic. Not supported for realtime transcription. base_url: Custom base URL for OpenAI API. api_key: Your OpenAI API key. If not provided, will use the OPENAI_API_KEY environment variable. client: Optional pre-configured OpenAI AsyncClient instance. use_realtime: Whether to use the realtime transcription API. (default: False) vad: Optional Voice Activity Detector used to commit the audio buffer when the model does not support server-side turn detection (e.g. `gpt-realtime-whisper`). When not provided and the model requires it, Silero VAD is auto-loaded with default settings. Pass `vad=None` to opt out of the auto-load and drive `input_audio_buffer.commit` yourself. """ # noqa: E501 if use_realtime and is_given(temperature): logger.warning( "temperature is not supported for realtime transcription; " "ignoring the provided value" ) temperature = NOT_GIVEN whisper_realtime = use_realtime and _is_whisper_realtime(model) if whisper_realtime: if is_given(turn_detection): logger.warning( "turn_detection is not supported for %s; ignoring the provided value", model ) turn_detection = NOT_GIVEN if not is_given(vad): try: from livekit.plugins.silero import VAD as SileroVAD except ImportError as e: raise ImportError( "livekit-plugins-silero is required for the gpt-realtime-whisper model " "(no server-side endpointing). Pass `vad=None` to opt out and drive " "`input_audio_buffer.commit` manually." ) from e vad = SileroVAD.load() super().__init__( capabilities=stt.STTCapabilities( streaming=use_realtime, interim_results=use_realtime, aligned_transcript=False ) ) if detect_language: language = "" if not is_given(turn_detection): turn_detection = { "type": "server_vad", "threshold": 0.5, "prefix_padding_ms": 600, "silence_duration_ms": 350, } self._opts = _STTOptions( language=LanguageCode(language), detect_language=detect_language, model=model, prompt=prompt, turn_detection=turn_detection, temperature=temperature, ) if is_given(noise_reduction_type): self._opts.noise_reduction_type = noise_reduction_type self._vad = vad if is_given(vad) else None if is_given(api_key) and not api_key: raise ValueError( "OpenAI API key is required, either as argument or set" " OPENAI_API_KEY environment variable" ) self._client = client or openai.AsyncClient( max_retries=0, api_key=api_key if is_given(api_key) else None, base_url=base_url if is_given(base_url) else None, http_client=httpx.AsyncClient( timeout=httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0), follow_redirects=True, limits=httpx.Limits( max_connections=50, max_keepalive_connections=50, keepalive_expiry=120, ), ), ) self._streams = weakref.WeakSet[SpeechStream]() self._session: aiohttp.ClientSession | None = None self._pool = utils.ConnectionPool[aiohttp.ClientWebSocketResponse]( max_session_duration=_max_session_duration, connect_cb=self._connect_ws, close_cb=self._close_ws, ) @property def model(self) -> str: return self._opts.model @property def provider(self) -> str: return self._client._base_url.netloc.decode("utf-8") @staticmethod def with_azure( *, language: str = "en", detect_language: bool = False, model: STTModels | str = "gpt-4o-mini-transcribe", prompt: NotGivenOr[str] = NOT_GIVEN, turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN, noise_reduction_type: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, azure_endpoint: str | None = None, azure_deployment: str | None = None, api_version: str | None = None, api_key: str | None = None, azure_ad_token: str | None = None, azure_ad_token_provider: AsyncAzureADTokenProvider | None = None, organization: str | None = None, project: str | None = None, base_url: str | None = None, use_realtime: bool = False, timeout: httpx.Timeout | None = None, vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN, ) -> STT: """ Create a new instance of Azure OpenAI STT. This automatically infers the following arguments from their corresponding environment variables if they are not provided: - `api_key` from `AZURE_OPENAI_API_KEY` - `organization` from `OPENAI_ORG_ID` - `project` from `OPENAI_PROJECT_ID` - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN` - `api_version` from `OPENAI_API_VERSION` - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT` """ # noqa: E501 azure_client = openai.AsyncAzureOpenAI( max_retries=0, azure_endpoint=azure_endpoint, azure_deployment=azure_deployment, api_version=api_version, api_key=api_key, azure_ad_token=azure_ad_token, azure_ad_token_provider=azure_ad_token_provider, organization=organization, project=project, base_url=base_url, timeout=timeout if timeout else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0), ) # type: ignore return STT( language=language, detect_language=detect_language, model=model, prompt=prompt, turn_detection=turn_detection, noise_reduction_type=noise_reduction_type, temperature=temperature, client=azure_client, use_realtime=use_realtime, vad=vad, ) @staticmethod def with_ovhcloud( *, model: str = "whisper-large-v3-turbo", api_key: NotGivenOr[str] = NOT_GIVEN, base_url: str = "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1", client: openai.AsyncClient | None = None, language: str = "en", detect_language: bool = False, prompt: NotGivenOr[str] = NOT_GIVEN, ) -> STT: """ Create a new instance of OVHcloud AI Endpoints STT. ``api_key`` must be set to your OVHcloud AI Endpoints API key, either using the argument or by setting the ``OVHCLOUD_API_KEY`` environmental variable. """ ovhcloud_api_key = api_key if is_given(api_key) else os.environ.get("OVHCLOUD_API_KEY") if not ovhcloud_api_key: raise ValueError("OVHcloud AI Endpoints API key is required") return STT( model=model, api_key=ovhcloud_api_key, base_url=base_url, client=client, language=language, detect_language=detect_language, prompt=prompt, use_realtime=False, ) def stream( self, *, language: NotGivenOr[str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> SpeechStream: if is_given(language): self._opts.language = LanguageCode(language) stream = SpeechStream( stt=self, pool=self._pool, conn_options=conn_options, vad_instance=self._vad, ) self._streams.add(stream) return stream def update_options( self, *, model: NotGivenOr[STTModels | str] = NOT_GIVEN, language: NotGivenOr[str] = NOT_GIVEN, detect_language: NotGivenOr[bool] = NOT_GIVEN, prompt: NotGivenOr[str] = NOT_GIVEN, turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN, noise_reduction_type: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, ) -> None: """ Update the options for the speech stream. Most options are updated at the connection level. SpeechStreams will be recreated when options are updated. Args: language: The language to transcribe in. detect_language: Whether to automatically detect the language. model: The model to use for transcription. prompt: Optional text prompt to guide the transcription. Only supported for whisper-1. turn_detection: When using realtime, this controls how model detects the user is done speaking. noise_reduction_type: Type of noise reduction to apply. "near_field" or "far_field" temperature: Sampling temperature between 0 and 1. Not supported for realtime transcription. """ # noqa: E501 if is_given(model): self._opts.model = model if is_given(language): self._opts.language = LanguageCode(language) if is_given(detect_language): self._opts.detect_language = detect_language self._opts.language = LanguageCode("") if is_given(prompt): self._opts.prompt = prompt if is_given(turn_detection): self._opts.turn_detection = turn_detection if is_given(noise_reduction_type): self._opts.noise_reduction_type = noise_reduction_type if is_given(temperature): if self.capabilities.streaming: logger.warning( "temperature is not supported for realtime transcription; " "ignoring the provided value" ) else: self._opts.temperature = temperature for stream in self._streams: if is_given(language): stream.update_options(language=language) async def _connect_ws(self, timeout: float) -> aiohttp.ClientWebSocketResponse: prompt = self._opts.prompt if is_given(self._opts.prompt) else "" transcription_config: dict[str, Any] = { "model": self._opts.model, } if prompt: transcription_config["prompt"] = prompt if self._opts.language: transcription_config["language"] = self._opts.language.language input_config: dict[str, Any] = { "format": { "type": "audio/pcm", "rate": SAMPLE_RATE, }, "transcription": transcription_config, } # gpt-realtime-whisper rejects any turn_detection config — omit the key entirely. # For other models, send the configured turn_detection (server-side VAD). if not _is_whisper_realtime(self._opts.model): input_config["turn_detection"] = self._opts.turn_detection if self._opts.noise_reduction_type: input_config["noise_reduction"] = {"type": self._opts.noise_reduction_type} realtime_config: dict[str, Any] = { "type": "session.update", "session": { "type": "transcription", "audio": { "input": input_config, }, }, } query_params: dict[str, str] = { "intent": "transcription", } headers = { "User-Agent": "LiveKit Agents", "Authorization": f"Bearer {self._client.api_key}", } url = f"{str(self._client.base_url).rstrip('/')}/realtime?{urlencode(query_params)}" if url.startswith("http"): url = url.replace("http", "ws", 1) session = self._ensure_session() ws = await asyncio.wait_for(session.ws_connect(url, headers=headers), timeout) await ws.send_json(realtime_config) return ws async def _close_ws(self, ws: aiohttp.ClientWebSocketResponse) -> None: await ws.close() def _ensure_session(self) -> aiohttp.ClientSession: if not self._session: self._session = utils.http_context.http_session() return self._session async def _recognize_impl( self, buffer: AudioBuffer, *, language: NotGivenOr[str] = NOT_GIVEN, conn_options: APIConnectOptions, ) -> stt.SpeechEvent: try: if is_given(language): self._opts.language = LanguageCode(language) data = rtc.combine_audio_frames(buffer).to_wav_bytes() prompt = self._opts.prompt if is_given(self._opts.prompt) else openai.omit format = "json" if self._opts.model == "whisper-1": # verbose_json returns language and other details, only supported for whisper-1 format = "verbose_json" resp = await self._client.audio.transcriptions.create( file=( "file.wav", data, "audio/wav", ), model=self._opts.model, # type: ignore language=self._opts.language.language if self._opts.language else "", prompt=prompt, response_format=format, temperature=self._opts.temperature if is_given(self._opts.temperature) else openai.omit, timeout=httpx.Timeout(30, connect=conn_options.timeout), ) sd = stt.SpeechData(text=resp.text, language=self._opts.language) if isinstance(resp, TranscriptionVerbose) and resp.language: sd.language = LanguageCode(resp.language) return stt.SpeechEvent( type=stt.SpeechEventType.FINAL_TRANSCRIPT, alternatives=[sd], ) except openai.APITimeoutError: raise APITimeoutError() from None except openai.APIStatusError as e: raise APIStatusError( e.message, status_code=e.status_code, request_id=e.request_id, body=e.body ) from None except Exception as e: raise APIConnectionError() from e class SpeechStream(stt.SpeechStream): def __init__( self, *, stt: STT, conn_options: APIConnectOptions, pool: utils.ConnectionPool[aiohttp.ClientWebSocketResponse], vad_instance: vad.VAD | None = None, ) -> None: super().__init__(stt=stt, conn_options=conn_options, sample_rate=SAMPLE_RATE) self._pool = pool self._language = stt._opts.language self._request_id = "" self._reconnect_event = asyncio.Event() self._vad = vad_instance self._speaking = False def update_options( self, *, language: str, ) -> None: self._language = LanguageCode(language) self._pool.invalidate() self._reconnect_event.set() def _start_speaking(self) -> None: if self._speaking: return self._speaking = True self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH)) def _stop_speaking(self) -> None: if not self._speaking: return self._speaking = False self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)) @utils.log_exceptions(logger=logger) async def _run(self) -> None: closing_ws = False @utils.log_exceptions(logger=logger) async def send_task( ws: aiohttp.ClientWebSocketResponse, vad_stream: vad.VADStream | None ) -> None: nonlocal closing_ws # forward audio to OAI in chunks of 50ms audio_bstream = utils.audio.AudioByteStream( sample_rate=SAMPLE_RATE, num_channels=NUM_CHANNELS, samples_per_channel=SAMPLE_RATE // 20, ) async for data in self._input_ch: frames: list[rtc.AudioFrame] = [] if isinstance(data, rtc.AudioFrame): if vad_stream is not None: vad_stream.push_frame(data) frames.extend(audio_bstream.write(data.data.tobytes())) elif isinstance(data, self._FlushSentinel): frames.extend(audio_bstream.flush()) for frame in frames: encoded_frame = { "type": "input_audio_buffer.append", "audio": base64.b64encode(frame.data.tobytes()).decode("utf-8"), } await ws.send_json(encoded_frame) if vad_stream is not None: vad_stream.end_input() closing_ws = True @utils.log_exceptions(logger=logger) async def vad_task(ws: aiohttp.ClientWebSocketResponse, vad_stream: vad.VADStream) -> None: async for ev in vad_stream: if ev.type == vad.VADEventType.START_OF_SPEECH: self._start_speaking() elif ev.type == vad.VADEventType.END_OF_SPEECH: self._stop_speaking() await ws.send_json({"type": "input_audio_buffer.commit"}) @utils.log_exceptions(logger=logger) async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None: nonlocal closing_ws current_text = "" current_item_id = "" last_interim_at: float = 0 connected_at = time.time() item_audio_timing: dict[str, dict[str, int]] = {} while True: msg = await ws.receive() if msg.type in ( aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSING, ): if closing_ws: # close is expected, see SpeechStream.aclose return # this will trigger a reconnection, see the _run loop raise APIStatusError( message="OpenAI Realtime STT connection closed unexpectedly", status_code=ws.close_code or -1, body=f"{msg.data=} {msg.extra=}", ) if msg.type != aiohttp.WSMsgType.TEXT: logger.warning("unexpected OpenAI message type %s", msg.type) continue try: data = json.loads(msg.data) msg_type = data.get("type") if msg_type == "input_audio_buffer.speech_started": item_id = data.get("item_id", "") current_item_id = item_id audio_start_ms = data.get("audio_start_ms", 0) item_audio_timing[item_id] = {"start_ms": audio_start_ms} if self._vad is None: self._start_speaking() elif msg_type == "input_audio_buffer.speech_stopped": item_id = data.get("item_id", "") audio_end_ms = data.get("audio_end_ms", 0) if item_id in item_audio_timing: item_audio_timing[item_id]["end_ms"] = audio_end_ms if self._vad is None: self._stop_speaking() elif msg_type == "conversation.item.input_audio_transcription.delta": delta = data.get("delta", "") item_id = data.get("item_id", "") or current_item_id if item_id: current_item_id = item_id if delta: current_text += delta if time.time() - last_interim_at > _delta_transcript_interval: self._event_ch.send_nowait( stt.SpeechEvent( type=stt.SpeechEventType.INTERIM_TRANSCRIPT, request_id=current_item_id, alternatives=[ stt.SpeechData( text=current_text, language=self._language, ) ], ) ) last_interim_at = time.time() elif msg_type == "conversation.item.input_audio_transcription.completed": current_text = "" transcript = data.get("transcript", "") item_id = data.get("item_id", "") if transcript: self._event_ch.send_nowait( stt.SpeechEvent( type=stt.SpeechEventType.FINAL_TRANSCRIPT, request_id=item_id, alternatives=[ stt.SpeechData( text=transcript, language=self._language, ) ], ) ) audio_duration = 0.0 if item_id in item_audio_timing: timing = item_audio_timing[item_id] start_ms = timing.get("start_ms", 0) end_ms = timing.get("end_ms", 0) if end_ms > start_ms: audio_duration = (end_ms - start_ms) / 1000.0 del item_audio_timing[item_id] # extract token usage if available usage = data.get("usage", {}) input_tokens = usage.get("input_tokens", 0) output_tokens = usage.get("output_tokens", 0) self._event_ch.send_nowait( stt.SpeechEvent( type=stt.SpeechEventType.RECOGNITION_USAGE, alternatives=[], recognition_usage=stt.RecognitionUsage( audio_duration=audio_duration, input_tokens=input_tokens, output_tokens=output_tokens, ), ) ) # restart session if needed if time.time() - connected_at > _max_session_duration: logger.info("resetting Realtime STT session due to timeout") self._pool.remove(ws) self._reconnect_event.set() return elif msg_type == "error": error_body = data.get("error", {}) raise APIError( message=f"OpenAI Realtime STT error: {error_body.get('message', 'Unknown error')}", body=error_body, retryable=False, ) except APIError: raise except Exception: logger.exception("failed to process OpenAI message") while True: closing_ws = False # reset the flag # a segment left open across the reconnect gap would fuse into the next utterance self._stop_speaking() async with self._pool.connection(timeout=self._conn_options.timeout) as ws: self._report_connection_acquired( self._pool.last_acquire_time, self._pool.last_connection_reused ) vad_stream = self._vad.stream() if self._vad is not None else None tasks = [ asyncio.create_task(send_task(ws, vad_stream)), asyncio.create_task(recv_task(ws)), ] if vad_stream is not None: tasks.append(asyncio.create_task(vad_task(ws, vad_stream))) tasks_group = asyncio.gather(*tasks) wait_reconnect_task = asyncio.create_task(self._reconnect_event.wait()) try: done, _ = await asyncio.wait( (tasks_group, wait_reconnect_task), return_when=asyncio.FIRST_COMPLETED, ) # propagate exceptions from completed tasks for task in done: if task != wait_reconnect_task: task.result() if wait_reconnect_task not in done: break self._reconnect_event.clear() finally: await utils.aio.gracefully_cancel(*tasks, wait_reconnect_task) tasks_group.cancel() tasks_group.exception() # retrieve the exception if vad_stream is not None: await vad_stream.aclose()