Skip to content

fix: report speech end times across other stt plugins #7988

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
chenghao-mou wants to merge 1 commit into from
+2,571 −216
Closed

fix: report speech end times across other stt plugins #7988

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 31 additions & 1 deletion livekit-agents/livekit/agents/inference/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,6 +321,15 @@ def _supports_agent_context_carryover(model: NotGivenOr[STTModels | str]) -> boo
}
)

_UTTERANCE_DURATION_MODELS = frozenset(
{
"soniox/stt-rt",
"soniox/stt-rt-v4",
"soniox/stt-rt-v5",
"speechmatics/linden-1",
}
)


def _aligned_transcript_for_model(
model: NotGivenOr[STTModels | str],
Expand Down Expand Up @@ -1444,7 +1453,28 @@ def _process_transcript(self, data: dict, is_final: bool) -> None:

if self._speaking:
self._speaking = False
end_event = stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)
words = data.get("words") or []
end_alternatives = []
if words and words[-1].get("end", 0) > 0:
end_alternatives = [
replace(speech_data, end_time=self.start_time_offset + words[-1]["end"])
]
elif (
data.get("duration", 0) > 0
and self._opts.model in _UTTERANCE_DURATION_MODELS
and (
not is_given(self._opts.fallback)
or all(
item["model"] in _UTTERANCE_DURATION_MODELS
for item in self._opts.fallback
)
)
):
end_alternatives = [speech_data]
end_event = stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=end_alternatives,
)
self._event_ch.send_nowait(end_event)
self._on_end_of_speech()
else:
Expand Down
7 changes: 7 additions & 0 deletions livekit-agents/livekit/agents/stt/multi_speaker_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,6 +281,13 @@ def push_audio(self, frame: rtc.AudioFrame) -> None:
self._rms_buffer = self._rms_buffer[-self._max_rms_size :]

def on_stt_event(self, ev: SpeechEvent) -> SpeechEvent | None:
if ev.type not in (
SpeechEventType.INTERIM_TRANSCRIPT,
SpeechEventType.PREFLIGHT_TRANSCRIPT,
SpeechEventType.FINAL_TRANSCRIPT,
):
return ev

if not ev.alternatives:
return ev

Expand Down
9 changes: 8 additions & 1 deletion livekit-plugins/livekit-plugins-assemblyai/livekit/plugins/assemblyai/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1113,7 +1113,14 @@ def _process_stream_event(self, data: dict) -> None:
self._session_id,
)

self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=final_event.alternatives
if words and words[-1].get("end", 0) > 0
else [],
)
)

if self._speech_duration > 0.0:
usage_event = stt.SpeechEvent(
Expand Down
13 changes: 10 additions & 3 deletions livekit-plugins/livekit-plugins-aws/livekit/plugins/aws/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -447,30 +447,37 @@ def _process_transcript_event(self, transcript_event: TranscriptEvent) -> None:

stream = transcript_event.transcript.results
for resp in stream:
speech_data: stt.SpeechData | None = None
if resp.start_time is not None and resp.start_time == 0.0:
self._event_ch.send_nowait(
stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH)
)

if resp.end_time is not None and resp.end_time > 0.0:
speech_data = self._streaming_recognize_response_to_speech_data(resp)
if resp.is_partial:
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.INTERIM_TRANSCRIPT,
alternatives=[self._streaming_recognize_response_to_speech_data(resp)],
alternatives=[speech_data],
)
)

else:
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.FINAL_TRANSCRIPT,
alternatives=[self._streaming_recognize_response_to_speech_data(resp)],
alternatives=[speech_data],
)
)

if not resp.is_partial:
self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=[speech_data] if speech_data is not None else [],
)
)

def _streaming_recognize_response_to_speech_data(self, resp: Result) -> stt.SpeechData:
confidence = 0.0
Expand Down
17 changes: 15 additions & 2 deletions livekit-plugins/livekit-plugins-azure/livekit/plugins/azure/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,8 @@ async def _run(self) -> None:
self._session_started_event.wait(), self._conn_options.timeout
)

self._input_ch.restart_provider_clock()

async def process_input() -> None:
async for input in self._input_ch:
if isinstance(input, rtc.AudioFrame):
Expand Down Expand Up @@ -424,7 +426,7 @@ def _on_speech_start(self, evt: speechsdk.SpeechRecognitionEventArgs) -> None:
stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH),
)

def _on_speech_end(self, evt: speechsdk.SpeechRecognitionEventArgs) -> None:
def _on_speech_end(self, evt: speechsdk.RecognitionEventArgs) -> None:
if not self._speaking:
return

Expand All @@ -433,7 +435,18 @@ def _on_speech_end(self, evt: speechsdk.SpeechRecognitionEventArgs) -> None:
with contextlib.suppress(RuntimeError):
self._loop.call_soon_threadsafe(
self._event_ch.send_nowait,
stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH),
stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=[
stt.SpeechData(
language=LanguageCode(""),
text="",
end_time=evt.offset / 10**7 + self.start_time_offset,
)
]
if evt.offset > 0
else [],
),
)

def _on_session_started(self, evt: speechsdk.SpeechRecognitionEventArgs) -> None:
Expand Down
7 changes: 6 additions & 1 deletion livekit-plugins/livekit-plugins-baseten/livekit/plugins/baseten/qwen3_stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -378,7 +378,12 @@ async def _recv() -> None:
if is_final:
if speaking:
self._event_ch.send_nowait(
stt_module.SpeechEvent(type=stt_module.SpeechEventType.END_OF_SPEECH)
stt_module.SpeechEvent(
type=stt_module.SpeechEventType.END_OF_SPEECH,
alternatives=[self._speech_data(event, segments, text)]
if segments and float(segments[-1].get("end_time", 0)) > 0
else [],
)
)
speaking = False
# Only the commit-triggered final ends the session; VAD
Expand Down
1 change: 1 addition & 0 deletions livekit-plugins/livekit-plugins-baseten/livekit/plugins/baseten/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -595,6 +595,7 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
while True:
try:
ws = await self._connect_ws()
self._input_ch.restart_provider_clock()
tasks = [
asyncio.create_task(send_task(ws)),
asyncio.create_task(recv_task(ws)),
Expand Down
21 changes: 10 additions & 11 deletions ...t-plugins-cartesia/livekit/plugins/cartesia/_recognize_streams/legacy_recognize_stream.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,6 @@ def __init__(
self._reconnect_event = asyncio.Event()
self._speaking = False
self._speech_duration: float = 0
self._last_speech_end_time: float = 0

async def _run(self) -> None:
if self._input_ch.closed:
Expand All @@ -177,7 +176,6 @@ async def _run(self) -> None:
# invocation by the base class) starts fresh.
self._speaking = False
self._speech_duration = 0
self._last_speech_end_time = 0

async def keepalive_task(ws: aiohttp.ClientWebSocketResponse) -> None:
try:
Expand Down Expand Up @@ -265,6 +263,7 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
while True:
try:
ws = await self._connect_ws()
self._input_ch.restart_provider_clock()

send = asyncio.create_task(send_task(ws))
recv = asyncio.create_task(recv_task(ws))
Expand Down Expand Up @@ -367,12 +366,7 @@ def _process_stream_event(self, data: STTEventMessage) -> None:
)
for word in words
]
# word timestamps are often within the audio window, so we track time separately
if self._last_speech_end_time == 0.0:
self._last_speech_end_time = self.start_time_offset
start_time = self._last_speech_end_time
end_time = start_time + data.get("duration", 0)
self._last_speech_end_time = end_time
end_time = words[-1].get("end", 0) if words else 0.0
is_final = data["is_final"]

if not text and not is_final:
Expand All @@ -388,8 +382,10 @@ def _process_stream_event(self, data: STTEventMessage) -> None:

speech_data = stt.SpeechData(
language=self._language or LanguageCode("en"),
start_time=start_time,
end_time=end_time,
start_time=words[0].get("start", 0) + self.start_time_offset
if end_time > 0
else 0.0,
end_time=end_time + self.start_time_offset if end_time > 0 else 0.0,
text=text,
words=timed_words,
)
Expand All @@ -416,7 +412,10 @@ def _process_stream_event(self, data: STTEventMessage) -> None:

if self._speaking:
self._speaking = False
end_event = stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)
end_event = stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=[speech_data] if end_time > 0 else [],
)
self._event_ch.send_nowait(end_event)
else:
event = stt.SpeechEvent(
Expand Down
33 changes: 31 additions & 2 deletions livekit-plugins/livekit-plugins-deepgram/livekit/plugins/deepgram/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -458,6 +458,7 @@ class SpeechStream(stt.SpeechStream):
_KEEPALIVE_MSG: str = json.dumps({"type": "KeepAlive"})
_CLOSE_MSG: str = json.dumps({"type": "CloseStream"})
_FINALIZE_MSG: str = json.dumps({"type": "Finalize"})
_final_alternatives: list[stt.SpeechData]

def __init__(
self,
Expand All @@ -481,6 +482,7 @@ def __init__(
self._session = http_session
self._opts.endpoint_url = base_url
self._speaking = False
self._final_alternatives = []
self._audio_duration_collector = PeriodicCollector(
callback=self._on_audio_duration_report,
duration=5.0,
Expand Down Expand Up @@ -584,6 +586,7 @@ def update_options(
self._reconnect_event.set()

def _on_end_of_speech(self) -> None:
self._final_alternatives = []
if self._pending_keyterm is not None:
self.update_options(keyterm=self._pending_keyterm)
self._pending_keyterm = None
Expand Down Expand Up @@ -701,6 +704,8 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
conn_start_time = 0.0
try:
ws = await self._connect_ws()
self._input_ch.restart_provider_clock()
self._final_alternatives = []
conn_start_time = time.perf_counter()
tasks = [
asyncio.create_task(send_task(ws)),
Expand Down Expand Up @@ -873,6 +878,8 @@ def _process_stream_event(self, data: dict) -> None:
self._event_ch.send_nowait(start_event)

if is_final_transcript:
words = data["channel"]["alternatives"][0]["words"]
self._final_alternatives = alts if words and words[-1].get("end", 0) > 0 else []
final_event = stt.SpeechEvent(
type=stt.SpeechEventType.FINAL_TRANSCRIPT,
request_id=request_id,
Expand All @@ -892,15 +899,37 @@ def _process_stream_event(self, data: dict) -> None:
# a non-empty transcript (deepgram doesn't have a SpeechEnded event)
if is_endpoint and self._speaking:
self._speaking = False
self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))
# An empty endpoint follows the last non-empty final on some turns.
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=self._final_alternatives,
)
)
self._on_end_of_speech()

elif data["type"] == "UtteranceEnd":
# Fired when utterance_end_ms is set and the configured silence duration has elapsed.
# https://developers.deepgram.com/docs/understand-endpointing-interim-results
last_word_end = data.get("last_word_end", 0)
if last_word_end < 0:
return # The utterance was already finalized.
if self._speaking:
self._speaking = False
self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=[
stt.SpeechData(
language=LanguageCode(self._opts.language),
text="",
end_time=self.start_time_offset + last_word_end,
)
]
if last_word_end > 0
else self._final_alternatives,
)
)
self._on_end_of_speech()
elif data["type"] == "Metadata":
pass # metadata is too noisy
Expand Down
24 changes: 19 additions & 5 deletions livekit-plugins/livekit-plugins-deepgram/livekit/plugins/deepgram/stt_v2.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
import os
import weakref
from collections.abc import Sequence
from dataclasses import dataclass
from dataclasses import dataclass, replace
from typing import Any, Literal

import aiohttp
Expand Down Expand Up @@ -563,6 +563,7 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
while True:
try:
ws = await self._connect_ws()
self._input_ch.restart_provider_clock()
# expose the connection for in-band Configure updates
self._ws = ws
tasks = [
Expand Down Expand Up @@ -678,7 +679,9 @@ def _on_audio_duration_report(self, duration: float) -> None:
)
self._event_ch.send_nowait(usage_event)

def _send_transcript_event(self, event_type: stt.SpeechEventType, data: dict) -> None:
def _send_transcript_event(
self, event_type: stt.SpeechEventType, data: dict
) -> list[stt.SpeechData]:
alts = _parse_transcription(self._opts.language, data, self.start_time_offset)
if alts:
event = stt.SpeechEvent(
Expand All @@ -687,6 +690,7 @@ def _send_transcript_event(self, event_type: stt.SpeechEventType, data: dict) ->
alternatives=alts,
)
self._event_ch.send_nowait(event)
return alts

def _process_stream_event(self, data: dict) -> None:
assert self._opts.language is not None
Expand Down Expand Up @@ -732,9 +736,19 @@ def _process_stream_event(self, data: dict) -> None:

self._speaking = False

self._send_transcript_event(stt.SpeechEventType.FINAL_TRANSCRIPT, data)

end_event = stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)
alts = self._send_transcript_event(stt.SpeechEventType.FINAL_TRANSCRIPT, data)
words = data.get("words") or []
last_word_end = words[-1].get("end", 0) if words else 0

end_event = stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=[
replace(alt, end_time=self.start_time_offset + last_word_end)
for alt in alts
]
if last_word_end > 0
else [],
)
self._event_ch.send_nowait(end_event)

elif data["type"] == "ConfigureSuccess":
Expand Down
1 change: 1 addition & 0 deletions livekit-plugins/livekit-plugins-fireworksai/livekit/plugins/fireworksai/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -386,6 +386,7 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
while True:
try:
ws = await self._connect_ws()
self._input_ch.restart_provider_clock()
tasks = [
asyncio.create_task(send_task(ws)),
asyncio.create_task(recv_task(ws)),
Expand Down
Loading
Loading