livekit--agents
wehub-resource-sync
984 行
38 KiB
Python
984 行
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`"
|
|
)
|