wehub-resource-sync
751 lines
30 KiB
Python
751 lines
30 KiB
Python
# 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()
|