wehub-resource-sync
984 lines
38 KiB
Python
984 lines
38 KiB
Python
# Copyright 2025 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 dataclasses
|
||
import os
|
||
from enum import Enum
|
||
from typing import Any
|
||
|
||
from livekit.agents import (
|
||
DEFAULT_API_CONNECT_OPTIONS,
|
||
APIConnectOptions,
|
||
LanguageCode,
|
||
stt,
|
||
utils,
|
||
vad,
|
||
)
|
||
from livekit.agents.types import (
|
||
NOT_GIVEN,
|
||
NotGivenOr,
|
||
)
|
||
from livekit.agents.utils import AudioBuffer, is_given
|
||
from speechmatics.rt import ClientMessageType
|
||
from speechmatics.voice import (
|
||
AdditionalVocabEntry,
|
||
AgentServerMessageType,
|
||
AudioEncoding,
|
||
OperatingPoint,
|
||
SpeakerFocusConfig,
|
||
SpeakerFocusMode,
|
||
SpeakerIdentifier,
|
||
VoiceAgentClient,
|
||
VoiceAgentConfig,
|
||
VoiceAgentConfigPreset,
|
||
)
|
||
|
||
from .log import logger
|
||
from .version import __version__ as lk_version
|
||
|
||
|
||
class TurnDetectionMode(str, Enum):
|
||
"""Endpoint and turn detection handling mode.
|
||
|
||
How the STT engine handles the endpointing of speech. Use `TurnDetectionMode.EXTERNAL` when
|
||
turn boundaries are controlled manually, for example via an external VAD or the `finalize()`
|
||
method.
|
||
|
||
To use the STT engine's built-in endpointing, use `TurnDetectionMode.ADAPTIVE` for simple
|
||
voice activity detection or `TurnDetectionMode.SMART_TURN` for more advanced ML-based
|
||
endpointing.
|
||
|
||
The `TurnDetectionMode.FIXED` mode uses a fixed amount of silence, as determined by the
|
||
`end_of_utterance_silence_trigger` parameter.
|
||
|
||
The default is `TurnDetectionMode.EXTERNAL` which delegates endpointing to an external VAD
|
||
(Silero is auto-loaded if no `vad` is provided).
|
||
"""
|
||
|
||
EXTERNAL = "external"
|
||
FIXED = "fixed"
|
||
ADAPTIVE = "adaptive"
|
||
SMART_TURN = "smart_turn"
|
||
|
||
|
||
@dataclasses.dataclass
|
||
class STTOptions:
|
||
"""Configuration parameters for Speechmatics STT service."""
|
||
|
||
# Service configuration
|
||
language: LanguageCode = LanguageCode("en")
|
||
output_locale: str | None = None
|
||
domain: str | None = None
|
||
|
||
# Endpointing mode
|
||
turn_detection_mode: TurnDetectionMode = TurnDetectionMode.EXTERNAL
|
||
|
||
# Output formatting
|
||
speaker_active_format: str | None = None
|
||
speaker_passive_format: str | None = None
|
||
|
||
# Speakers
|
||
focus_speakers: list[str] = dataclasses.field(default_factory=list)
|
||
ignore_speakers: list[str] = dataclasses.field(default_factory=list)
|
||
focus_mode: SpeakerFocusMode = SpeakerFocusMode.RETAIN
|
||
known_speakers: list[SpeakerIdentifier] = dataclasses.field(default_factory=list)
|
||
|
||
# Custom dictionary
|
||
additional_vocab: list[AdditionalVocabEntry] = dataclasses.field(default_factory=list)
|
||
|
||
# -------------------
|
||
# Advanced features
|
||
# -------------------
|
||
|
||
# Features
|
||
operating_point: OperatingPoint | None = None
|
||
max_delay: float | None = None
|
||
end_of_utterance_silence_trigger: float | None = None
|
||
end_of_utterance_max_delay: float | None = None
|
||
punctuation_overrides: dict | None = None
|
||
include_partials: bool | None = None
|
||
|
||
# Diarization
|
||
enable_diarization: bool | None = None
|
||
speaker_sensitivity: float | None = None
|
||
max_speakers: int | None = None
|
||
prefer_current_speaker: bool | None = None
|
||
|
||
|
||
class STT(stt.STT):
|
||
def __init__(
|
||
self,
|
||
*,
|
||
api_key: NotGivenOr[str] = NOT_GIVEN,
|
||
base_url: NotGivenOr[str] = NOT_GIVEN,
|
||
turn_detection_mode: TurnDetectionMode = TurnDetectionMode.EXTERNAL,
|
||
operating_point: NotGivenOr[OperatingPoint] = NOT_GIVEN,
|
||
domain: NotGivenOr[str] = NOT_GIVEN,
|
||
language: str = "en",
|
||
output_locale: NotGivenOr[str] = NOT_GIVEN,
|
||
include_partials: NotGivenOr[bool] = NOT_GIVEN,
|
||
enable_diarization: NotGivenOr[bool] = NOT_GIVEN,
|
||
max_delay: NotGivenOr[float] = NOT_GIVEN,
|
||
end_of_utterance_silence_trigger: NotGivenOr[float] = NOT_GIVEN,
|
||
end_of_utterance_max_delay: NotGivenOr[float] = NOT_GIVEN,
|
||
additional_vocab: NotGivenOr[list[AdditionalVocabEntry]] = NOT_GIVEN,
|
||
punctuation_overrides: NotGivenOr[dict] = NOT_GIVEN,
|
||
speaker_sensitivity: NotGivenOr[float] = NOT_GIVEN,
|
||
max_speakers: NotGivenOr[int] = NOT_GIVEN,
|
||
speaker_active_format: NotGivenOr[str] = NOT_GIVEN,
|
||
speaker_passive_format: NotGivenOr[str] = NOT_GIVEN,
|
||
prefer_current_speaker: NotGivenOr[bool] = NOT_GIVEN,
|
||
focus_speakers: NotGivenOr[list[str]] = NOT_GIVEN,
|
||
ignore_speakers: NotGivenOr[list[str]] = NOT_GIVEN,
|
||
focus_mode: SpeakerFocusMode = SpeakerFocusMode.RETAIN,
|
||
known_speakers: NotGivenOr[list[SpeakerIdentifier]] = NOT_GIVEN,
|
||
sample_rate: int = 16000,
|
||
audio_encoding: AudioEncoding = AudioEncoding.PCM_S16LE,
|
||
vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
|
||
**kwargs: Any,
|
||
):
|
||
"""Create a new instance of Speechmatics STT using the Voice SDK.
|
||
|
||
Args:
|
||
api_key: Speechmatics API key. Can be set via `api_key` argument
|
||
or `SPEECHMATICS_API_KEY` environment variable.
|
||
|
||
base_url: Custom base URL for the API. Can be set via `base_url`
|
||
argument or `SPEECHMATICS_RT_URL` environment variable. Optional.
|
||
|
||
turn_detection_mode: Controls how the STT engine detects end of speech
|
||
turns. Use `EXTERNAL` when turn boundaries are controlled manually,
|
||
for example via an external VAD or the `finalize()` method. Use
|
||
`ADAPTIVE` for simple VAD or `SMART_TURN` for ML-based endpointing.
|
||
`FIXED` uses a fixed amount of silence, as determined by the
|
||
`end_of_utterance_silence_trigger` parameter.
|
||
Defaults to `TurnDetectionMode.EXTERNAL`.
|
||
|
||
operating_point: Operating point for transcription accuracy vs. latency
|
||
tradeoff. Overrides preset if provided. Optional.
|
||
|
||
domain: Domain to use. Optional.
|
||
|
||
language: Language code for the STT model. Defaults to `en`.
|
||
|
||
output_locale: Output locale for the STT model, e.g. `en-GB`. Optional.
|
||
|
||
include_partials: Include partial segment fragments (words) in the output
|
||
of AddPartialSegment messages. Partial fragments from the STT will
|
||
always be used for speaker activity detection. This setting is used
|
||
only for the formatted text output of individual segments. Optional.
|
||
|
||
enable_diarization: Enable speaker diarization. When enabled, the STT
|
||
engine will determine and attribute words to unique speakers.
|
||
Overrides preset if provided. Defaults to True.
|
||
|
||
max_delay: Maximum delay in seconds for transcription. This forces the
|
||
STT engine to speed up the processing of transcribed words and reduces
|
||
the interval between partial and final results. Lower values can have
|
||
an impact on accuracy. Overrides preset if provided. Optional.
|
||
|
||
end_of_utterance_silence_trigger: Silence duration in seconds that
|
||
triggers end of utterance. The delay is used to wait for any further
|
||
transcribed words before emitting the `FINAL_TRANSCRIPT` events.
|
||
Overrides preset if provided. Optional.
|
||
|
||
end_of_utterance_max_delay: Maximum delay in seconds for end of utterance.
|
||
Must be greater than `end_of_utterance_silence_trigger`.
|
||
Overrides preset if provided. Optional.
|
||
|
||
additional_vocab: List of additional vocabulary entries to increase the
|
||
weight of specific words in the transcription model. Defaults to [].
|
||
|
||
punctuation_overrides: Punctuation overrides. Allows overriding the
|
||
punctuation behaviour in the STT engine. Overrides preset if provided.
|
||
Optional.
|
||
|
||
speaker_sensitivity: Diarization sensitivity. A higher value increases the
|
||
sensitivity of diarization and helps when two or more speakers have
|
||
similar voices. Overrides preset if provided. Optional.
|
||
|
||
max_speakers: Maximum number of speakers to detect during diarization.
|
||
When set, the STT engine will limit the number of unique speakers
|
||
identified. Overrides preset if provided. Optional.
|
||
|
||
speaker_active_format: Formatter for active speaker output. The attributes
|
||
`text` and `speaker_id` are available. Example: `@{speaker_id}: {text}`.
|
||
Defaults to transcription output.
|
||
|
||
speaker_passive_format: Formatter for passive speaker output. The attributes
|
||
`text` and `speaker_id` are available. Example:
|
||
`@{speaker_id} [background]: {text}`. Defaults to transcription output.
|
||
|
||
prefer_current_speaker: When True, groups of words close together are given
|
||
extra weight to be identified as the same speaker. Overrides preset if
|
||
provided. Optional.
|
||
|
||
focus_speakers: List of speaker IDs to focus on. Only these speakers are
|
||
emitted as `FINAL_TRANSCRIPT` events; others are treated as passive.
|
||
Words from passive speakers are still processed but only emitted when a
|
||
focused speaker has also said new words. Defaults to [].
|
||
|
||
ignore_speakers: List of speaker IDs to ignore. These speakers are excluded
|
||
from transcription and their speech will not trigger VAD or end of
|
||
utterance detection. By default, any speaker with a label wrapped in
|
||
double underscores (e.g. `__ASSISTANT__`) is excluded. Defaults to [].
|
||
|
||
focus_mode: Controls what happens to words from non-focused speakers. When
|
||
`RETAIN`, non-ignored speakers are processed as passive frames. When
|
||
`IGNORE`, their words are discarded entirely. Defaults to
|
||
`SpeakerFocusMode.RETAIN`.
|
||
|
||
known_speakers: List of known speaker labels and identifiers. When supplied,
|
||
the STT engine uses them to attribute words to specific speakers across
|
||
sessions. Defaults to [].
|
||
|
||
sample_rate: Audio sample rate in Hz. Defaults to 16000.
|
||
|
||
audio_encoding: Audio encoding format. Defaults to `AudioEncoding.PCM_S16LE`.
|
||
|
||
vad: Optional external Voice Activity Detector. When provided, the STT
|
||
engine's endpointing is replaced by the VAD: each audio frame is
|
||
forwarded to the VAD, and `finalize()` is called whenever the VAD
|
||
reports end of speech. Providing a VAD implicitly sets
|
||
`turn_detection_mode` to `EXTERNAL`. When `turn_detection_mode` is
|
||
`EXTERNAL` and `vad` is not provided, Silero is auto-loaded to drive
|
||
finalize. Pass `vad=None` to opt out of the auto-load if you intend
|
||
to call `finalize()` from your own logic. Defaults to NOT_GIVEN.
|
||
|
||
**kwargs: Catches deprecated parameters. A warning is logged for any
|
||
recognised deprecated name.
|
||
"""
|
||
|
||
# Resolve final turn_detection_mode — a real `vad` forces EXTERNAL.
|
||
if is_given(vad) and vad is not None and turn_detection_mode != TurnDetectionMode.EXTERNAL:
|
||
logger.info(
|
||
"External `vad` provided; overriding turn_detection_mode "
|
||
f"{turn_detection_mode.value!r} -> 'external'"
|
||
)
|
||
turn_detection_mode = TurnDetectionMode.EXTERNAL
|
||
|
||
# In EXTERNAL mode the STT does not endpoint on its own. Auto-load Silero
|
||
# so finalize() is wired up, unless the caller explicitly passed `vad=None`
|
||
# to opt out (they'll drive finalize() themselves).
|
||
if turn_detection_mode == TurnDetectionMode.EXTERNAL and 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 Speechmatics with "
|
||
"turn_detection_mode=EXTERNAL (no server-side endpointing). "
|
||
"Pass `vad=None` to opt out and drive finalize() manually."
|
||
) from e
|
||
vad = SileroVAD.load()
|
||
|
||
# Normalize NOT_GIVEN -> None for downstream storage.
|
||
self._vad = vad if is_given(vad) else None
|
||
|
||
# Set default values for optional parameters
|
||
super().__init__(
|
||
capabilities=stt.STTCapabilities(
|
||
streaming=True,
|
||
interim_results=True,
|
||
diarization=enable_diarization if is_given(enable_diarization) else True,
|
||
aligned_transcript="chunk",
|
||
offline_recognize=False,
|
||
),
|
||
)
|
||
|
||
# Set STT options
|
||
def _set(value: Any) -> Any:
|
||
return value if is_given(value) else None
|
||
|
||
# Create STT options from parameters
|
||
self._stt_options = STTOptions(
|
||
language=LanguageCode(language),
|
||
output_locale=_set(output_locale),
|
||
domain=_set(domain),
|
||
turn_detection_mode=turn_detection_mode,
|
||
speaker_active_format=_set(speaker_active_format),
|
||
speaker_passive_format=_set(speaker_passive_format),
|
||
focus_speakers=_set(focus_speakers) or [],
|
||
ignore_speakers=_set(ignore_speakers) or [],
|
||
focus_mode=focus_mode,
|
||
known_speakers=_set(known_speakers) or [],
|
||
additional_vocab=_set(additional_vocab) or [],
|
||
operating_point=_set(operating_point),
|
||
max_delay=_set(max_delay),
|
||
end_of_utterance_silence_trigger=_set(end_of_utterance_silence_trigger),
|
||
end_of_utterance_max_delay=_set(end_of_utterance_max_delay),
|
||
punctuation_overrides=_set(punctuation_overrides),
|
||
include_partials=_set(include_partials),
|
||
enable_diarization=_set(enable_diarization),
|
||
speaker_sensitivity=_set(speaker_sensitivity),
|
||
max_speakers=_set(max_speakers),
|
||
prefer_current_speaker=_set(prefer_current_speaker),
|
||
)
|
||
|
||
# Migrate / warn about any deprecated kwargs
|
||
_check_deprecated_args(kwargs, self._stt_options)
|
||
|
||
# Validate config options
|
||
errors = self._validate_stt_options()
|
||
if errors:
|
||
raise ValueError("Invalid STT options: " + ", ".join(errors))
|
||
|
||
# Set API key
|
||
self._api_key: str = api_key if is_given(api_key) else os.getenv("SPEECHMATICS_API_KEY", "")
|
||
|
||
# Set base URL
|
||
self._base_url: str = (
|
||
base_url
|
||
if is_given(base_url)
|
||
else os.getenv("SPEECHMATICS_RT_URL", "wss://eu2.rt.speechmatics.com/v2")
|
||
)
|
||
|
||
# Validate API key and base URL
|
||
if not self._api_key:
|
||
raise ValueError("Missing Speechmatics API key")
|
||
if not self._base_url:
|
||
raise ValueError("Missing Speechmatics base URL")
|
||
|
||
# Set audio parameters
|
||
self._sample_rate = sample_rate
|
||
self._audio_encoding = audio_encoding
|
||
|
||
# Initialize list of streams
|
||
self._streams: list[SpeechStream] = []
|
||
|
||
# Show warning for external
|
||
if self._stt_options.turn_detection_mode == TurnDetectionMode.EXTERNAL:
|
||
logger.info("STT under external turn detection control")
|
||
|
||
@property
|
||
def provider(self) -> str:
|
||
return "Speechmatics"
|
||
|
||
@property
|
||
def model(self) -> str:
|
||
op = self._stt_options.operating_point
|
||
return str(op.value) if op is not None else "enhanced"
|
||
|
||
async def _recognize_impl(
|
||
self,
|
||
buffer: AudioBuffer,
|
||
*,
|
||
language: NotGivenOr[str] = NOT_GIVEN,
|
||
conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
|
||
) -> stt.SpeechEvent:
|
||
raise NotImplementedError("Not implemented")
|
||
|
||
def stream(
|
||
self,
|
||
*,
|
||
language: NotGivenOr[str] = NOT_GIVEN,
|
||
conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
|
||
) -> stt.RecognizeStream:
|
||
"""Create a new SpeechStream."""
|
||
|
||
# Create the stream
|
||
stream = SpeechStream(
|
||
stt=self,
|
||
conn_options=conn_options,
|
||
config=self._prepare_config(language),
|
||
id=len(self._streams),
|
||
vad_instance=self._vad,
|
||
)
|
||
|
||
# Add to the list of streams
|
||
self._streams.append(stream)
|
||
|
||
# Return the stream
|
||
return stream
|
||
|
||
def _validate_stt_options(self) -> list[str]:
|
||
"""Validate options in STTOptions."""
|
||
errors: list[str] = []
|
||
opts = self._stt_options
|
||
|
||
# end_of_utterance_silence_trigger must be between 0 and 2
|
||
if opts.end_of_utterance_silence_trigger is not None and not (
|
||
0 < opts.end_of_utterance_silence_trigger < 2
|
||
):
|
||
errors.append("end_of_utterance_silence_trigger must be between 0 and 2")
|
||
|
||
# end_of_utterance_max_delay must exceed end_of_utterance_silence_trigger so the engine has time to detect silence
|
||
if (
|
||
opts.end_of_utterance_max_delay is not None
|
||
and opts.end_of_utterance_silence_trigger is not None
|
||
and opts.end_of_utterance_max_delay <= opts.end_of_utterance_silence_trigger
|
||
):
|
||
errors.append(
|
||
"end_of_utterance_max_delay must be greater than end_of_utterance_silence_trigger"
|
||
)
|
||
|
||
# server rejects speaker counts outside 2–100
|
||
if opts.max_speakers is not None and not (1 < opts.max_speakers <= 100):
|
||
errors.append("max_speakers must be between 2 and 100")
|
||
|
||
# latency budget: below 0.7s is unsupported
|
||
if opts.max_delay is not None and not (0.7 <= opts.max_delay <= 4.0):
|
||
errors.append("max_delay must be between 0.7 and 4.0")
|
||
|
||
# diarization sensitivity range enforced by the engine
|
||
if opts.speaker_sensitivity is not None and not (0.0 < opts.speaker_sensitivity < 1.0):
|
||
errors.append("speaker_sensitivity must be between 0.0 and 1.0")
|
||
|
||
return errors
|
||
|
||
def _prepare_config(self, language: NotGivenOr[str] = NOT_GIVEN) -> VoiceAgentConfig:
|
||
"""Prepare VoiceAgentConfig from STTOptions."""
|
||
|
||
# Reference to STT options
|
||
opts = self._stt_options
|
||
|
||
# Preset taken from `FIXED`, `EXTERNAL`, `ADAPTIVE` or `SMART_TURN`
|
||
config = VoiceAgentConfigPreset.load(opts.turn_detection_mode.value)
|
||
|
||
# Set sample rate and encoding
|
||
config.sample_rate = self._sample_rate
|
||
config.audio_encoding = self._audio_encoding
|
||
|
||
# LanguageCode and domain
|
||
config.language = LanguageCode(language) if is_given(language) else opts.language
|
||
config.domain = opts.domain
|
||
config.output_locale = opts.output_locale
|
||
|
||
# Speaker configuration
|
||
config.speaker_config = SpeakerFocusConfig(
|
||
focus_speakers=opts.focus_speakers,
|
||
ignore_speakers=opts.ignore_speakers,
|
||
focus_mode=opts.focus_mode,
|
||
)
|
||
config.known_speakers = opts.known_speakers
|
||
|
||
# Additional vocabulary
|
||
config.additional_vocab = opts.additional_vocab
|
||
|
||
# Override preset parameters if provided
|
||
advanced_params = [
|
||
"enable_diarization",
|
||
"end_of_utterance_max_delay",
|
||
"end_of_utterance_silence_trigger",
|
||
"include_partials",
|
||
"max_delay",
|
||
"max_speakers",
|
||
"operating_point",
|
||
"prefer_current_speaker",
|
||
"punctuation_overrides",
|
||
"speaker_sensitivity",
|
||
]
|
||
|
||
# Override preset parameters if provided
|
||
for param in advanced_params:
|
||
value = getattr(opts, param)
|
||
if value is not None:
|
||
setattr(config, param, value)
|
||
|
||
# Return the config
|
||
return config
|
||
|
||
def update_speakers(
|
||
self,
|
||
focus_speakers: NotGivenOr[list[str]] = NOT_GIVEN,
|
||
ignore_speakers: NotGivenOr[list[str]] = NOT_GIVEN,
|
||
focus_mode: NotGivenOr[SpeakerFocusMode] = NOT_GIVEN,
|
||
) -> None:
|
||
"""Updates the speaker configuration.
|
||
|
||
This can update the speakers to listen to or ignore during an in-flight
|
||
transcription. Only available if diarization is enabled.
|
||
|
||
This will be applied to *all* streams (typically only one).
|
||
|
||
Args:
|
||
focus_speakers: List of speakers to focus on.
|
||
ignore_speakers: List of speakers to ignore.
|
||
focus_mode: Focus mode to use.
|
||
"""
|
||
# Do this for each stream
|
||
for stream in self._streams:
|
||
# Check if diarization is enabled
|
||
if not stream._config.enable_diarization:
|
||
raise ValueError("Diarization is not enabled")
|
||
|
||
# Update the configuration
|
||
if is_given(focus_speakers):
|
||
self._stt_options.focus_speakers = focus_speakers
|
||
stream._config.speaker_config.focus_speakers = focus_speakers
|
||
if is_given(ignore_speakers):
|
||
self._stt_options.ignore_speakers = ignore_speakers
|
||
stream._config.speaker_config.ignore_speakers = ignore_speakers
|
||
if is_given(focus_mode):
|
||
self._stt_options.focus_mode = focus_mode
|
||
stream._config.speaker_config.focus_mode = focus_mode
|
||
|
||
# Send update to client if stream is active
|
||
if stream._client and stream._client._is_connected:
|
||
stream._client.update_diarization_config(stream._config.speaker_config)
|
||
|
||
def finalize(self) -> None:
|
||
"""Finalize the turn (from external VAD).
|
||
|
||
When using an external VAD, such as Silero, this should be called
|
||
when the VAD detects the end of a speech turn. This will force the
|
||
finalization of the words in the STT buffer and emit them as final
|
||
segments.
|
||
"""
|
||
|
||
# Iterate over the streams
|
||
for stream in self._streams:
|
||
# Do not finalize if being handled by a client
|
||
if not stream._client or not stream._client._is_connected:
|
||
continue
|
||
|
||
# Check that VAD is not being handled by the client
|
||
if stream._config.vad_config is None or not stream._config.vad_config.enabled:
|
||
stream._client.finalize()
|
||
|
||
async def get_speaker_ids(
|
||
self,
|
||
) -> list[SpeakerIdentifier] | list[list[SpeakerIdentifier]]:
|
||
"""Get the list of speakers from the current STT session.
|
||
|
||
If diarization is enabled, then this will use the GET_SPEAKERS message
|
||
to retrieve the list of speakers for the current session. This should
|
||
be used once speakers have said at least 5 words to improve the results.
|
||
|
||
Returns:
|
||
list[SpeakerIdentifier]: List of speakers in the session.
|
||
"""
|
||
|
||
# Results
|
||
results: list[list[SpeakerIdentifier]] = []
|
||
|
||
# Iterate over all streams
|
||
for idx, stream in enumerate(self._streams):
|
||
# Skip streams that aren't actively connected
|
||
if stream._client is None or not stream._client._is_connected:
|
||
logger.warning(f"Not connected in stream {idx}")
|
||
results.append([])
|
||
continue
|
||
|
||
# Return if diarization is not enabled
|
||
if not stream._config.enable_diarization:
|
||
logger.warning(f"Diarization is not enabled in stream {idx}")
|
||
results.append([])
|
||
continue
|
||
|
||
# Clear the speaker result
|
||
stream._speaker_result_event.clear()
|
||
|
||
# Send message to client
|
||
await stream._client.send_message({"message": ClientMessageType.GET_SPEAKERS.value})
|
||
|
||
# Wait the result (5 second timeout)
|
||
try:
|
||
await asyncio.wait_for(
|
||
stream._speaker_result_event.wait(),
|
||
timeout=5.0,
|
||
)
|
||
except asyncio.TimeoutError:
|
||
logger.warning(f"GetSpeakers timed-out for stream {idx}")
|
||
results.append([])
|
||
continue
|
||
|
||
# Return the list of speakers
|
||
results.append(stream._speaker_result or [])
|
||
|
||
# Return the list of speakers
|
||
if len(results) == 1:
|
||
return results[0]
|
||
return results
|
||
|
||
|
||
class SpeechStream(stt.RecognizeStream):
|
||
def __init__(
|
||
self,
|
||
stt: STT,
|
||
conn_options: APIConnectOptions,
|
||
config: VoiceAgentConfig,
|
||
id: int,
|
||
vad_instance: vad.VAD | None = None,
|
||
) -> None:
|
||
super().__init__(
|
||
stt=stt,
|
||
conn_options=conn_options,
|
||
sample_rate=stt._sample_rate,
|
||
)
|
||
|
||
self._stt: STT = stt
|
||
self._id: int = id
|
||
self._config: VoiceAgentConfig = config
|
||
self._client: VoiceAgentClient | None = None
|
||
self._msg_queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
|
||
self._speech_duration: float = 0
|
||
|
||
self._vad: vad.VAD | None = vad_instance
|
||
self._vad_stream: vad.VADStream | None = None
|
||
|
||
self._tasks: list[asyncio.Task] = []
|
||
|
||
# Speaker result event
|
||
self._speaker_result_event: asyncio.Event = asyncio.Event()
|
||
self._speaker_result: list[SpeakerIdentifier] | None = None
|
||
|
||
async def _run(self) -> None:
|
||
"""Run the STT stream."""
|
||
logger.debug("Connecting to Speechmatics STT service")
|
||
|
||
# Config is required
|
||
if not self._config:
|
||
raise ValueError("Config is required")
|
||
|
||
# Create the Voice Agent client
|
||
self._client = VoiceAgentClient(
|
||
api_key=self._stt._api_key,
|
||
url=self._stt._base_url,
|
||
app=f"livekit/{lk_version}",
|
||
config=self._config,
|
||
)
|
||
|
||
# Add message handlers
|
||
def add_message(message: dict[str, Any]) -> None:
|
||
self._msg_queue.put_nowait(message)
|
||
|
||
# Default messages to listen to
|
||
messages: list[AgentServerMessageType] = [
|
||
AgentServerMessageType.RECOGNITION_STARTED,
|
||
AgentServerMessageType.INFO,
|
||
AgentServerMessageType.ERROR,
|
||
AgentServerMessageType.WARNING,
|
||
AgentServerMessageType.ADD_PARTIAL_SEGMENT,
|
||
AgentServerMessageType.ADD_SEGMENT,
|
||
AgentServerMessageType.START_OF_TURN,
|
||
AgentServerMessageType.END_OF_TURN,
|
||
]
|
||
|
||
# Speaker IDs message handler
|
||
if self._config.enable_diarization:
|
||
messages.append(AgentServerMessageType.SPEAKERS_RESULT)
|
||
|
||
# Optional debug messages to log
|
||
# messages.append(AgentServerMessageType.END_OF_UTTERANCE)
|
||
# messages.append(AgentServerMessageType.END_OF_TURN_PREDICTION)
|
||
# messages.append(AgentServerMessageType.DIAGNOSTICS)
|
||
|
||
# Add message handlers
|
||
for event in messages:
|
||
self._client.on(event, add_message) # type: ignore[arg-type]
|
||
|
||
# Connect to the service
|
||
await self._client.connect()
|
||
logger.debug("Connected to Speechmatics STT service")
|
||
|
||
# Open external VAD stream (if provided) before tasks start pushing frames
|
||
if self._vad is not None:
|
||
self._vad_stream = self._vad.stream()
|
||
|
||
# Audio and messaging tasks
|
||
audio_task = asyncio.create_task(self._process_audio())
|
||
message_task = asyncio.create_task(self._process_messages())
|
||
|
||
# Tasks
|
||
self._tasks = [audio_task, message_task]
|
||
|
||
# Optional VAD task: calls `client.finalize()` on end of speech
|
||
vad_task: asyncio.Task | None = None
|
||
if self._vad_stream is not None:
|
||
vad_task = asyncio.create_task(self._process_vad(self._vad_stream))
|
||
self._tasks.append(vad_task)
|
||
|
||
# Wait for tasks to complete
|
||
try:
|
||
done, pending = await asyncio.wait(self._tasks, return_when=asyncio.FIRST_COMPLETED)
|
||
for task in done:
|
||
task.result()
|
||
|
||
# Disconnect the client
|
||
finally:
|
||
# Cancel audio first — stops sending audio to the STT engine
|
||
audio_task.cancel()
|
||
try:
|
||
await audio_task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
|
||
# Close the VAD stream so its task drains and exits
|
||
if self._vad_stream is not None:
|
||
await self._vad_stream.aclose()
|
||
self._vad_stream = None
|
||
|
||
if vad_task is not None:
|
||
await utils.aio.cancel_and_wait(vad_task)
|
||
|
||
# Disconnect flushes final messages from the STT engine
|
||
await self._client.disconnect()
|
||
|
||
# Cancel message task after disconnect — final messages have been processed
|
||
message_task.cancel()
|
||
try:
|
||
await message_task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
|
||
# Remove from active streams so stale streams aren't iterated
|
||
if self in self._stt._streams:
|
||
self._stt._streams.remove(self)
|
||
|
||
async def _process_audio(self) -> None:
|
||
"""Process audio from the input channel."""
|
||
try:
|
||
# Input audio stream
|
||
audio_bstream = utils.audio.AudioByteStream(
|
||
sample_rate=self._stt._sample_rate,
|
||
num_channels=1,
|
||
)
|
||
|
||
# Process input audio
|
||
async for data in self._input_ch:
|
||
# Handle flush sentinel
|
||
if isinstance(data, self._FlushSentinel):
|
||
frames = audio_bstream.flush()
|
||
else:
|
||
# Forward the original frame to the VAD before resampling/repacking
|
||
if self._vad_stream is not None:
|
||
self._vad_stream.push_frame(data)
|
||
frames = audio_bstream.write(data.data.tobytes())
|
||
|
||
# Send audio frames
|
||
if self._client:
|
||
for frame in frames:
|
||
self._speech_duration += frame.duration
|
||
await self._client.send_audio(frame.data.tobytes())
|
||
|
||
# No more input — let the VAD flush any pending event
|
||
if self._vad_stream is not None:
|
||
self._vad_stream.end_input()
|
||
|
||
except asyncio.CancelledError:
|
||
pass
|
||
|
||
async def _process_vad(self, vad_stream: vad.VADStream) -> None:
|
||
"""Call `client.finalize()` whenever the external VAD reports end of speech."""
|
||
try:
|
||
async for ev in vad_stream:
|
||
if ev.type == vad.VADEventType.END_OF_SPEECH:
|
||
if self._client and self._client._is_connected:
|
||
self._client.finalize()
|
||
except asyncio.CancelledError:
|
||
pass
|
||
|
||
async def _process_messages(self) -> None:
|
||
"""Process messages from the STT client."""
|
||
try:
|
||
while True:
|
||
message = await self._msg_queue.get()
|
||
self._handle_message(message)
|
||
except asyncio.CancelledError:
|
||
pass
|
||
|
||
def _handle_message(self, message: dict[str, Any]) -> None:
|
||
"""Handle a message from the STT client."""
|
||
|
||
# Get the message type
|
||
event = message.get("message", None)
|
||
|
||
# Only handle valid messages
|
||
if event is None:
|
||
return
|
||
|
||
# Log info, error and warning messages
|
||
elif event in [
|
||
AgentServerMessageType.RECOGNITION_STARTED,
|
||
AgentServerMessageType.INFO,
|
||
]:
|
||
logger.info(f"{event} -> {message}")
|
||
elif event == AgentServerMessageType.WARNING:
|
||
logger.warning(f"{event} -> {message}")
|
||
elif event == AgentServerMessageType.ERROR:
|
||
logger.error(f"{event} -> {message}")
|
||
|
||
# Handle the messages
|
||
elif event == AgentServerMessageType.ADD_PARTIAL_SEGMENT:
|
||
self._handle_partial_segment(message)
|
||
elif event == AgentServerMessageType.ADD_SEGMENT:
|
||
self._handle_segment(message)
|
||
elif event == AgentServerMessageType.START_OF_TURN:
|
||
self._handle_start_of_turn(message)
|
||
elif event == AgentServerMessageType.END_OF_TURN:
|
||
self._handle_end_of_turn(message)
|
||
|
||
# Handle the speaker result message
|
||
elif event == AgentServerMessageType.SPEAKERS_RESULT:
|
||
self._handle_speakers_result(message)
|
||
|
||
# Log all other messages
|
||
else:
|
||
logger.debug(f"{event} -> {message}")
|
||
|
||
def _handle_partial_segment(self, message: dict[str, Any]) -> None:
|
||
"""Handle AddPartialSegment events."""
|
||
segments: list[dict[str, Any]] = message.get("segments", [])
|
||
if segments:
|
||
self._send_frames(segments, is_final=False)
|
||
|
||
def _handle_segment(self, message: dict[str, Any]) -> None:
|
||
"""Handle AddSegment events."""
|
||
segments: list[dict[str, Any]] = message.get("segments", [])
|
||
if segments:
|
||
self._send_frames(segments, is_final=True)
|
||
|
||
def _handle_start_of_turn(self, message: dict[str, Any]) -> None:
|
||
"""Handle StartOfTurn events."""
|
||
logger.debug("StartOfTurn received")
|
||
self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH))
|
||
|
||
def _handle_end_of_turn(self, message: dict[str, Any]) -> None:
|
||
"""Handle EndOfTurn events."""
|
||
logger.debug("EndOfTurn received")
|
||
self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))
|
||
|
||
if self._speech_duration > 0.0:
|
||
usage_event = stt.SpeechEvent(
|
||
type=stt.SpeechEventType.RECOGNITION_USAGE,
|
||
alternatives=[],
|
||
recognition_usage=stt.RecognitionUsage(audio_duration=self._speech_duration),
|
||
)
|
||
self._event_ch.send_nowait(usage_event)
|
||
self._speech_duration = 0
|
||
|
||
def _handle_speakers_result(self, message: dict[str, Any]) -> None:
|
||
"""Handle SpeakersResult events."""
|
||
logger.debug("SpeakersResult received")
|
||
self._speaker_result = message.get("speakers", [])
|
||
self._speaker_result_event.set()
|
||
|
||
def _send_frames(self, segments: list[dict[str, Any]], is_final: bool) -> None:
|
||
"""Send frames to the pipeline."""
|
||
|
||
# Check for empty segments
|
||
if not segments:
|
||
return
|
||
|
||
# Get the options
|
||
opts = self._stt._stt_options
|
||
|
||
# Determine the event type
|
||
event_type = (
|
||
stt.SpeechEventType.FINAL_TRANSCRIPT
|
||
if is_final
|
||
else stt.SpeechEventType.INTERIM_TRANSCRIPT
|
||
)
|
||
|
||
# Process each segment
|
||
for segment in segments:
|
||
# Format the text based on speaker activity
|
||
is_active = segment.get("is_active", True)
|
||
format_str = (
|
||
opts.speaker_active_format if is_active else opts.speaker_passive_format
|
||
) or "{text}"
|
||
text = format_str.format(
|
||
speaker_id=segment.get("speaker_id", "UU"),
|
||
text=segment.get("text", ""),
|
||
)
|
||
|
||
# Create speech event
|
||
speech_data = stt.SpeechData(
|
||
language=LanguageCode(segment.get("language", opts.language)),
|
||
text=text,
|
||
speaker_id=segment.get("speaker_id", "UU"),
|
||
start_time=segment.get("metadata", {}).get("start_time", 0)
|
||
+ self.start_time_offset,
|
||
end_time=segment.get("metadata", {}).get("end_time", 0) + self.start_time_offset,
|
||
)
|
||
|
||
# Create speech event
|
||
event = stt.SpeechEvent(
|
||
type=event_type,
|
||
alternatives=[speech_data],
|
||
)
|
||
|
||
# Send the event
|
||
self._event_ch.send_nowait(event)
|
||
|
||
async def aclose(self) -> None:
|
||
"""Close the STT stream."""
|
||
await super().aclose()
|
||
|
||
# Cancel message processing task
|
||
if self._tasks:
|
||
for task in self._tasks:
|
||
task.cancel()
|
||
try:
|
||
await task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
|
||
# Close the VAD stream if it's still open
|
||
if self._vad_stream is not None:
|
||
await self._vad_stream.aclose()
|
||
self._vad_stream = None
|
||
|
||
# Close the client
|
||
if self._client and self._client._is_connected:
|
||
await self._client.disconnect()
|
||
self._client = None
|
||
|
||
# Remove from active streams
|
||
if self in self._stt._streams:
|
||
self._stt._streams.remove(self)
|
||
|
||
|
||
def _check_deprecated_args(kwargs: dict[str, Any], opts: STTOptions) -> None:
|
||
"""Warn about deprecated kwargs and migrate values where possible."""
|
||
|
||
# Removed — no replacement
|
||
for name in (
|
||
"end_of_utterance_mode",
|
||
"chunk_size",
|
||
"transcription_config",
|
||
"audio_settings",
|
||
"http_session",
|
||
):
|
||
if name in kwargs:
|
||
logger.warning(f"`{name}` is deprecated and no longer used")
|
||
|
||
# Partials
|
||
if "enable_partials" in kwargs:
|
||
if opts.include_partials is None:
|
||
logger.warning("`enable_partials` is deprecated, migrated to `include_partials`")
|
||
opts.include_partials = bool(kwargs["enable_partials"])
|
||
else:
|
||
logger.warning(
|
||
"Both `enable_partials` and `include_partials` provided; using `include_partials`"
|
||
)
|
||
|
||
# Diarization
|
||
if "diarization_sensitivity" in kwargs and isinstance(
|
||
kwargs["diarization_sensitivity"], (int, float)
|
||
):
|
||
if opts.speaker_sensitivity is None:
|
||
logger.warning(
|
||
"`diarization_sensitivity` is deprecated, migrated to `speaker_sensitivity`"
|
||
)
|
||
opts.speaker_sensitivity = kwargs["diarization_sensitivity"]
|
||
else:
|
||
logger.warning(
|
||
"Both `diarization_sensitivity` and `speaker_sensitivity` provided;"
|
||
" using `speaker_sensitivity`"
|
||
)
|