Rename transcript-gather to post-turn transcript wait
Switch the vocabulary for the timer-driven phase that runs when `wait_for_transcript_to_end_user_turn=False`. "Transcript gather" was too vague to be self-documenting; "post-turn transcript wait" names when it happens (after the user turn ends) and what it's for (waiting for late-arriving transcripts). Renames the internal property to `_wait_for_post_turn_transcripts` and the supporting state/method names to match (`_post_turn_transcript_wait_task`, `_complete_post_turn_transcript_wait`, etc.). Updates docstrings, comments, log messages, the example inline doc, and the test prose to use the new vocabulary consistently.
This commit is contained in:
@@ -80,10 +80,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
|
|||||||
#
|
#
|
||||||
# - Turn strategies do not consider user transcripts, so the user
|
# - Turn strategies do not consider user transcripts, so the user
|
||||||
# turn ends sooner.
|
# turn ends sooner.
|
||||||
# - User transcripts are handled by the aggregator: a simple timer
|
# - User transcripts are handled by the aggregator: a simple
|
||||||
# gives it time to gather them after the user turn ends, then
|
# post-turn transcript wait gives them time to arrive after the
|
||||||
# the aggregator emits `on_user_turn_message_finalized` with the
|
# user turn ends, then the aggregator emits
|
||||||
# new user context message.
|
# `on_user_turn_message_finalized` with the new user context
|
||||||
|
# message.
|
||||||
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
|
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
|
||||||
context,
|
context,
|
||||||
user_params=LLMUserAggregatorParams(
|
user_params=LLMUserAggregatorParams(
|
||||||
@@ -123,8 +124,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
|
|||||||
await task.cancel()
|
await task.cancel()
|
||||||
|
|
||||||
# `on_user_turn_stopped` fires at the end of the user turn. With
|
# `on_user_turn_stopped` fires at the end of the user turn. With
|
||||||
# `wait_for_transcript_to_end_user_turn=False`, the aggregator
|
# `wait_for_transcript_to_end_user_turn=False`, no user
|
||||||
# hasn't gathered user transcripts yet at this point, so
|
# transcripts have arrived yet at this point, so
|
||||||
# `message.content` is empty. Logged here to make the end-of-turn
|
# `message.content` is empty. Logged here to make the end-of-turn
|
||||||
# signal visible alongside the later finalization event.
|
# signal visible alongside the later finalization event.
|
||||||
@user_aggregator.event_handler("on_user_turn_stopped")
|
@user_aggregator.event_handler("on_user_turn_stopped")
|
||||||
@@ -133,8 +134,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
|
|||||||
|
|
||||||
# `on_user_turn_message_finalized` fires when the user message has
|
# `on_user_turn_message_finalized` fires when the user message has
|
||||||
# been finalized into the context. Here it fires later than
|
# been finalized into the context. Here it fires later than
|
||||||
# `on_user_turn_stopped`, after the aggregator has gathered the
|
# `on_user_turn_stopped`, after the aggregator's post-turn
|
||||||
# realtime service's user transcripts.
|
# transcript wait completes.
|
||||||
@user_aggregator.event_handler("on_user_turn_message_finalized")
|
@user_aggregator.event_handler("on_user_turn_message_finalized")
|
||||||
async def on_user_turn_message_finalized(
|
async def on_user_turn_message_finalized(
|
||||||
aggregator, strategy, message: UserMessageFinalizedMessage
|
aggregator, strategy, message: UserMessageFinalizedMessage
|
||||||
|
|||||||
@@ -138,16 +138,16 @@ class LLMUserAggregatorParams:
|
|||||||
|
|
||||||
- Turn strategies do not consider user transcripts, so the
|
- Turn strategies do not consider user transcripts, so the
|
||||||
user turn ends sooner. ``on_user_turn_stopped`` fires at
|
user turn ends sooner. ``on_user_turn_stopped`` fires at
|
||||||
the end of turn with empty content (transcripts haven't
|
the end of turn with empty content. To achieve this,
|
||||||
been gathered yet). To achieve this, the aggregator
|
the aggregator drops ``TranscriptionUserTurnStartStrategy``
|
||||||
drops ``TranscriptionUserTurnStartStrategy`` from start
|
from start strategies and flips
|
||||||
strategies and flips ``wait_for_transcript=False`` on
|
``wait_for_transcript=False`` on any stop strategy that
|
||||||
any stop strategy that supports it.
|
supports it.
|
||||||
- User transcripts are handled by the aggregator: a simple
|
- User transcripts are handled by the aggregator: a simple
|
||||||
timer gives it time to gather them after the user turn
|
post-turn transcript wait gives it time to receive them
|
||||||
ends, then the aggregator emits a new
|
after the user turn ends, then the aggregator emits a
|
||||||
``on_user_turn_message_finalized`` event with the new
|
new ``on_user_turn_message_finalized`` event with the
|
||||||
user context message.
|
new user context message.
|
||||||
filter_incomplete_user_turns: [DEPRECATED] Use
|
filter_incomplete_user_turns: [DEPRECATED] Use
|
||||||
``user_turn_strategies=FilterIncompleteUserTurnStrategies()``
|
``user_turn_strategies=FilterIncompleteUserTurnStrategies()``
|
||||||
instead. When enabled, the LLM outputs a turn-completion
|
instead. When enabled, the LLM outputs a turn-completion
|
||||||
@@ -286,14 +286,15 @@ class UserTurnStoppedMessage:
|
|||||||
With ``wait_for_transcript_to_end_user_turn=True`` (the default),
|
With ``wait_for_transcript_to_end_user_turn=True`` (the default),
|
||||||
the user message is finalized at the end of the turn, so
|
the user message is finalized at the end of the turn, so
|
||||||
``content`` carries the aggregated transcript. With it set to
|
``content`` carries the aggregated transcript. With it set to
|
||||||
False, the aggregator is still gathering user transcripts at this
|
False, the aggregator is still in its post-turn transcript wait
|
||||||
point, so ``content`` is ``None`` — subscribe to
|
at this point, so ``content`` is ``None`` — subscribe to
|
||||||
``on_user_turn_message_finalized`` for the assembled message.
|
``on_user_turn_message_finalized`` for the assembled message.
|
||||||
|
|
||||||
Parameters:
|
Parameters:
|
||||||
content: The aggregated user transcript, or ``None`` when
|
content: The aggregated user transcript, or ``None`` when
|
||||||
``wait_for_transcript_to_end_user_turn=False`` (the
|
``wait_for_transcript_to_end_user_turn=False`` (the
|
||||||
aggregator is still gathering transcripts at this point).
|
aggregator is still in its post-turn transcript wait at
|
||||||
|
this point).
|
||||||
timestamp: When the user turn started.
|
timestamp: When the user turn started.
|
||||||
user_id: Optional identifier for the user.
|
user_id: Optional identifier for the user.
|
||||||
|
|
||||||
@@ -311,9 +312,9 @@ class UserMessageFinalizedMessage:
|
|||||||
Fired when the user message has been finalized into the context.
|
Fired when the user message has been finalized into the context.
|
||||||
With ``wait_for_transcript_to_end_user_turn=True`` (the default)
|
With ``wait_for_transcript_to_end_user_turn=True`` (the default)
|
||||||
this coincides with ``on_user_turn_stopped``. With it set to
|
this coincides with ``on_user_turn_stopped``. With it set to
|
||||||
False, the aggregator first gathers user transcripts after the
|
False, the aggregator first runs a post-turn transcript wait, so
|
||||||
end of turn, so this event fires later than
|
this event fires later than ``on_user_turn_stopped``.
|
||||||
``on_user_turn_stopped``. ``content`` is always populated.
|
``content`` is always populated.
|
||||||
|
|
||||||
Parameters:
|
Parameters:
|
||||||
content: The aggregated user transcript.
|
content: The aggregated user transcript.
|
||||||
@@ -582,15 +583,15 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
``UserTurnStoppedMessage``. With
|
``UserTurnStoppedMessage``. With
|
||||||
``wait_for_transcript_to_end_user_turn=True`` (the default),
|
``wait_for_transcript_to_end_user_turn=True`` (the default),
|
||||||
``message.content`` carries the aggregated transcript. With it
|
``message.content`` carries the aggregated transcript. With it
|
||||||
set to False, the aggregator is still gathering user transcripts
|
set to False, the aggregator is still in its post-turn transcript
|
||||||
at this point, so ``message.content`` is ``None``; subscribe to
|
wait at this point, so ``message.content`` is ``None``; subscribe
|
||||||
``on_user_turn_message_finalized`` for the assembled message.
|
to ``on_user_turn_message_finalized`` for the assembled message.
|
||||||
- on_user_turn_message_finalized: Called when the user message
|
- on_user_turn_message_finalized: Called when the user message
|
||||||
has been finalized into the context, with a
|
has been finalized into the context, with a
|
||||||
``UserMessageFinalizedMessage``. With
|
``UserMessageFinalizedMessage``. With
|
||||||
``wait_for_transcript_to_end_user_turn=True`` this coincides
|
``wait_for_transcript_to_end_user_turn=True`` this coincides
|
||||||
with ``on_user_turn_stopped``; with it set to False it fires
|
with ``on_user_turn_stopped``; with it set to False it fires
|
||||||
later, after the aggregator's transcript-gather window
|
later, after the aggregator's post-turn transcript wait window
|
||||||
completes. ``message.content`` is always populated.
|
completes. ``message.content`` is always populated.
|
||||||
- on_user_turn_stop_timeout: Called when no user turn stop strategy triggers
|
- on_user_turn_stop_timeout: Called when no user turn stop strategy triggers
|
||||||
- on_user_turn_idle: Called when the user has been idle for the configured timeout
|
- on_user_turn_idle: Called when the user has been idle for the configured timeout
|
||||||
@@ -697,18 +698,18 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
# inferences fire before finalization.
|
# inferences fire before finalization.
|
||||||
self._full_user_turn_aggregation: str | None = None
|
self._full_user_turn_aggregation: str | None = None
|
||||||
|
|
||||||
# Transcript-gather state, used when the aggregator gathers
|
# Post-turn transcript wait state, used when the aggregator
|
||||||
# user transcripts after the user turn ends
|
# waits for transcripts after the user turn ends
|
||||||
# (`_aggregator_gathers_transcripts == True`):
|
# (`_wait_for_post_turn_transcripts == True`):
|
||||||
# `on_user_turn_stopped` has fired with empty content, and the
|
# `on_user_turn_stopped` has fired with empty content, and the
|
||||||
# aggregator is waiting on `_transcript_gather_task` before
|
# aggregator is waiting on `_post_turn_transcript_wait_task`
|
||||||
# finalizing the user message into context. The gather window
|
# before finalizing the user message into context. The wait
|
||||||
# duration is taken from the last `STTMetadataFrame` seen
|
# window duration is taken from the last `STTMetadataFrame`
|
||||||
# (`STTMetadataFrame.ttfs_p99_latency`), falling back to
|
# seen (`STTMetadataFrame.ttfs_p99_latency`), falling back to
|
||||||
# `DEFAULT_TTFS_P99` if no STT service has reported one.
|
# `DEFAULT_TTFS_P99` if no STT service has reported one.
|
||||||
self._gathering_for_strategy: BaseUserTurnStopStrategy | None = None
|
self._post_turn_transcript_wait_strategy: BaseUserTurnStopStrategy | None = None
|
||||||
self._inference_during_gather: bool = False
|
self._inference_during_post_turn_transcript_wait: bool = False
|
||||||
self._transcript_gather_task: asyncio.Task | None = None
|
self._post_turn_transcript_wait_task: asyncio.Task | None = None
|
||||||
self._stt_ttfs_p99_latency: float | None = None
|
self._stt_ttfs_p99_latency: float | None = None
|
||||||
|
|
||||||
self._user_turn_controller = UserTurnController(
|
self._user_turn_controller = UserTurnController(
|
||||||
@@ -754,16 +755,17 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
self._vad_controller.add_event_handler("on_broadcast_frame", self._on_broadcast_frame)
|
self._vad_controller.add_event_handler("on_broadcast_frame", self._on_broadcast_frame)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def _aggregator_gathers_transcripts(self) -> bool:
|
def _wait_for_post_turn_transcripts(self) -> bool:
|
||||||
"""True when the aggregator gathers user transcripts after the turn ends.
|
"""True when the aggregator runs a post-turn transcript wait.
|
||||||
|
|
||||||
Internal alias for ``wait_for_transcript_to_end_user_turn=False``.
|
Inverse of the public ``wait_for_transcript_to_end_user_turn``
|
||||||
In this mode, turn strategies don't consider user transcripts
|
param: when that's False, this is True. In this mode, turn
|
||||||
(so the user turn ends sooner), and the aggregator runs a
|
strategies don't consider user transcripts (so the user turn
|
||||||
simple timer after the end of turn to gather any transcripts
|
ends sooner), and the aggregator runs a simple timer after the
|
||||||
that arrive, then emits ``on_user_turn_message_finalized``
|
end of turn to receive any transcripts that arrive, then emits
|
||||||
with the assembled user context message. Always travels with
|
``on_user_turn_message_finalized`` with the assembled user
|
||||||
the strategy-mutation bundle applied at init.
|
context message. Always travels with the strategy-mutation
|
||||||
|
bundle applied at init.
|
||||||
"""
|
"""
|
||||||
return not self._params.wait_for_transcript_to_end_user_turn
|
return not self._params.wait_for_transcript_to_end_user_turn
|
||||||
|
|
||||||
@@ -868,7 +870,7 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
pass
|
pass
|
||||||
elif isinstance(frame, STTMetadataFrame):
|
elif isinstance(frame, STTMetadataFrame):
|
||||||
# Record the STT service's reported P99 TTFS so the
|
# Record the STT service's reported P99 TTFS so the
|
||||||
# transcript-gather timer can size itself to the real
|
# post-turn transcript wait timer can size itself to the real
|
||||||
# latency. Frame is also pushed downstream so other
|
# latency. Frame is also pushed downstream so other
|
||||||
# processors keep seeing it.
|
# processors keep seeing it.
|
||||||
self._stt_ttfs_p99_latency = frame.ttfs_p99_latency
|
self._stt_ttfs_p99_latency = frame.ttfs_p99_latency
|
||||||
@@ -933,15 +935,17 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
async def _finalize_on_session_end(self):
|
async def _finalize_on_session_end(self):
|
||||||
"""Flush any pending user message on session end.
|
"""Flush any pending user message on session end.
|
||||||
|
|
||||||
If a transcript-gather is in flight (the aggregator hasn't
|
If a post-turn transcript wait is in flight, complete it now so
|
||||||
finished gathering transcripts yet), complete it now so the
|
the user message is captured before the session shuts down.
|
||||||
user message is captured before the session shuts down.
|
|
||||||
Otherwise, run the mode-appropriate finalize path on whatever
|
Otherwise, run the mode-appropriate finalize path on whatever
|
||||||
is currently in the buffer.
|
is currently in the buffer.
|
||||||
"""
|
"""
|
||||||
if self._gathering_for_strategy is not None or self._inference_during_gather:
|
if (
|
||||||
await self._complete_transcript_gather(on_session_end=True)
|
self._post_turn_transcript_wait_strategy is not None
|
||||||
elif self._aggregator_gathers_transcripts:
|
or self._inference_during_post_turn_transcript_wait
|
||||||
|
):
|
||||||
|
await self._complete_post_turn_transcript_wait(on_session_end=True)
|
||||||
|
elif self._wait_for_post_turn_transcripts:
|
||||||
await self._finalize_user_message(on_session_end=True)
|
await self._finalize_user_message(on_session_end=True)
|
||||||
else:
|
else:
|
||||||
await self._finalize_user_turn(on_session_end=True)
|
await self._finalize_user_turn(on_session_end=True)
|
||||||
@@ -1076,17 +1080,20 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
):
|
):
|
||||||
logger.debug(f"{self}: User started speaking (strategy: {strategy})")
|
logger.debug(f"{self}: User started speaking (strategy: {strategy})")
|
||||||
|
|
||||||
# Precondition guard: if the previous turn's transcript-gather
|
# Precondition guard: if the previous turn's post-turn
|
||||||
# window is still active when the next turn starts, the
|
# transcript wait is still active when the next turn starts,
|
||||||
# assumption that transcripts arrive before the next turn
|
# the assumption that transcripts arrive before the next turn
|
||||||
# has been violated. Complete the previous turn's gather now
|
# has been violated. Complete the previous turn's wait now so
|
||||||
# so its user message is finalized before this turn proceeds.
|
# its user message is finalized before this turn proceeds.
|
||||||
if self._gathering_for_strategy is not None or self._inference_during_gather:
|
if (
|
||||||
|
self._post_turn_transcript_wait_strategy is not None
|
||||||
|
or self._inference_during_post_turn_transcript_wait
|
||||||
|
):
|
||||||
logger.warning(
|
logger.warning(
|
||||||
f"{self}: user turn started before previous turn's transcripts "
|
f"{self}: user turn started before previous turn's transcripts "
|
||||||
f"were gathered; flushing previous turn now"
|
f"arrived; flushing previous turn now"
|
||||||
)
|
)
|
||||||
await self._complete_transcript_gather()
|
await self._complete_post_turn_transcript_wait()
|
||||||
|
|
||||||
self._user_turn_start_timestamp = time_now_iso8601()
|
self._user_turn_start_timestamp = time_now_iso8601()
|
||||||
self._full_user_turn_aggregation = None
|
self._full_user_turn_aggregation = None
|
||||||
@@ -1108,12 +1115,12 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
):
|
):
|
||||||
logger.debug(f"{self}: User turn inference triggered (strategy: {strategy})")
|
logger.debug(f"{self}: User turn inference triggered (strategy: {strategy})")
|
||||||
|
|
||||||
if self._aggregator_gathers_transcripts:
|
if self._wait_for_post_turn_transcripts:
|
||||||
# The aggregator is gathering transcripts after the user
|
# The aggregator is in its post-turn transcript wait.
|
||||||
# turn end. Defer push_aggregation and event emission;
|
# Defer push_aggregation and event emission; they'll run
|
||||||
# they'll run alongside user message finalization when the
|
# alongside user message finalization when the wait window
|
||||||
# transcript-gather window completes.
|
# completes.
|
||||||
self._inference_during_gather = True
|
self._inference_during_post_turn_transcript_wait = True
|
||||||
return
|
return
|
||||||
|
|
||||||
# Push aggregation now: this writes the user message segment to
|
# Push aggregation now: this writes the user message segment to
|
||||||
@@ -1143,51 +1150,50 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
|
|
||||||
# End-of-turn side effects always fire on the strategy event,
|
# End-of-turn side effects always fire on the strategy event,
|
||||||
# regardless of whether user message finalization is deferred
|
# regardless of whether user message finalization is deferred
|
||||||
# to a transcript-gather window.
|
# to a post-turn transcript wait window.
|
||||||
if params.enable_user_speaking_frames:
|
if params.enable_user_speaking_frames:
|
||||||
await self.broadcast_frame(UserStoppedSpeakingFrame)
|
await self.broadcast_frame(UserStoppedSpeakingFrame)
|
||||||
|
|
||||||
await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame())
|
await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame())
|
||||||
|
|
||||||
if self._aggregator_gathers_transcripts:
|
if self._wait_for_post_turn_transcripts:
|
||||||
# Fire `on_user_turn_stopped` now for the end of turn —
|
# Fire `on_user_turn_stopped` now for the end of turn —
|
||||||
# content is `None` because the aggregator hasn't gathered
|
# content is `None` because no transcripts have arrived
|
||||||
# transcripts yet. Start the transcript-gather timer; when
|
# yet. Start the post-turn transcript wait timer; when it
|
||||||
# it completes, the aggregator finalizes the user message
|
# completes, the aggregator finalizes the user message and
|
||||||
# and emits `on_user_turn_message_finalized`. Consumers
|
# emits `on_user_turn_message_finalized`. Consumers wanting
|
||||||
# wanting the assembled message subscribe to
|
# the assembled message subscribe to
|
||||||
# `on_user_turn_message_finalized`.
|
# `on_user_turn_message_finalized`.
|
||||||
end_of_turn_message = UserTurnStoppedMessage(
|
end_of_turn_message = UserTurnStoppedMessage(
|
||||||
content=None, timestamp=self._user_turn_start_timestamp
|
content=None, timestamp=self._user_turn_start_timestamp
|
||||||
)
|
)
|
||||||
await self._call_event_handler("on_user_turn_stopped", strategy, end_of_turn_message)
|
await self._call_event_handler("on_user_turn_stopped", strategy, end_of_turn_message)
|
||||||
|
|
||||||
self._gathering_for_strategy = strategy
|
self._post_turn_transcript_wait_strategy = strategy
|
||||||
gather_timeout = (
|
wait_timeout = (
|
||||||
self._stt_ttfs_p99_latency
|
self._stt_ttfs_p99_latency
|
||||||
if self._stt_ttfs_p99_latency is not None
|
if self._stt_ttfs_p99_latency is not None
|
||||||
else DEFAULT_TTFS_P99
|
else DEFAULT_TTFS_P99
|
||||||
)
|
)
|
||||||
self._transcript_gather_task = self.create_task(
|
self._post_turn_transcript_wait_task = self.create_task(
|
||||||
self._transcript_gather_handler(gather_timeout),
|
self._post_turn_transcript_wait_handler(wait_timeout),
|
||||||
f"{self}::transcript_gather",
|
f"{self}::post_turn_transcript_wait",
|
||||||
)
|
)
|
||||||
return
|
return
|
||||||
|
|
||||||
await self._finalize_user_turn(strategy)
|
await self._finalize_user_turn(strategy)
|
||||||
|
|
||||||
async def _transcript_gather_handler(self, timeout: float):
|
async def _post_turn_transcript_wait_handler(self, timeout: float):
|
||||||
"""Transcript-gather timer.
|
"""Post-turn transcript wait timer.
|
||||||
|
|
||||||
Waits ``timeout`` seconds — giving transcripts time to arrive
|
Waits ``timeout`` seconds — giving transcripts time to arrive
|
||||||
after the end of turn — then completes the gather and
|
after the end of turn — then completes the wait and finalizes
|
||||||
finalizes the user message into context, with whatever
|
the user message into context, with whatever transcripts the
|
||||||
transcripts the aggregator has captured by then (possibly
|
aggregator has received by then (possibly none).
|
||||||
nothing).
|
|
||||||
|
|
||||||
The simple-timer approach relies on the assumptions that
|
The simple-timer approach relies on the assumptions that
|
||||||
transcripts don't arrive too late and that the bot response
|
transcripts don't arrive too late and that the assistant
|
||||||
won't finish before this timer.
|
response won't finish before this timer.
|
||||||
|
|
||||||
Cancelled by reset, the next-turn precondition guard, or
|
Cancelled by reset, the next-turn precondition guard, or
|
||||||
session end.
|
session end.
|
||||||
@@ -1197,30 +1203,30 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
return
|
return
|
||||||
finally:
|
finally:
|
||||||
self._transcript_gather_task = None
|
self._post_turn_transcript_wait_task = None
|
||||||
|
|
||||||
await self._complete_transcript_gather()
|
await self._complete_post_turn_transcript_wait()
|
||||||
|
|
||||||
async def _complete_transcript_gather(self, *, on_session_end: bool = False):
|
async def _complete_post_turn_transcript_wait(self, *, on_session_end: bool = False):
|
||||||
"""Complete the active transcript-gather window.
|
"""Complete the active post-turn transcript wait window.
|
||||||
|
|
||||||
``on_user_turn_stopped`` already fired at the end of turn (with
|
``on_user_turn_stopped`` already fired at the end of turn (with
|
||||||
empty content) and the aggregator has been gathering
|
empty content) and the aggregator has been receiving
|
||||||
transcripts since. This finalizes that work: flushes any
|
transcripts since. This finalizes that work: flushes any
|
||||||
inference-triggered segment whose push was deferred during the
|
inference-triggered segment whose push was deferred during the
|
||||||
gather, then emits ``on_user_turn_message_finalized`` with the
|
wait, then emits ``on_user_turn_message_finalized`` with the
|
||||||
assembled user context message. Called from the
|
assembled user context message. Called from the post-turn
|
||||||
transcript-gather timer (the normal path), the precondition
|
transcript wait timer (the normal path), the precondition guard
|
||||||
guard in ``_on_user_turn_started``, and the session-end paths.
|
in ``_on_user_turn_started``, and the session-end paths.
|
||||||
"""
|
"""
|
||||||
if self._transcript_gather_task:
|
if self._post_turn_transcript_wait_task:
|
||||||
await self.cancel_task(self._transcript_gather_task)
|
await self.cancel_task(self._post_turn_transcript_wait_task)
|
||||||
self._transcript_gather_task = None
|
self._post_turn_transcript_wait_task = None
|
||||||
|
|
||||||
gather_strategy = self._gathering_for_strategy
|
wait_strategy = self._post_turn_transcript_wait_strategy
|
||||||
had_pending_inference = self._inference_during_gather
|
had_pending_inference = self._inference_during_post_turn_transcript_wait
|
||||||
self._gathering_for_strategy = None
|
self._post_turn_transcript_wait_strategy = None
|
||||||
self._inference_during_gather = False
|
self._inference_during_post_turn_transcript_wait = False
|
||||||
|
|
||||||
if had_pending_inference:
|
if had_pending_inference:
|
||||||
segment = await self.push_aggregation()
|
segment = await self.push_aggregation()
|
||||||
@@ -1231,32 +1237,32 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
self._full_user_turn_aggregation = segment
|
self._full_user_turn_aggregation = segment
|
||||||
await self._call_event_handler("on_user_turn_inference_triggered", gather_strategy)
|
await self._call_event_handler("on_user_turn_inference_triggered", wait_strategy)
|
||||||
|
|
||||||
if gather_strategy is not None or on_session_end:
|
if wait_strategy is not None or on_session_end:
|
||||||
# `on_user_turn_stopped` already fired at the end of turn;
|
# `on_user_turn_stopped` already fired at the end of turn;
|
||||||
# this is the deferred user message finalization.
|
# this is the deferred user message finalization.
|
||||||
await self._finalize_user_message(gather_strategy, on_session_end=on_session_end)
|
await self._finalize_user_message(wait_strategy, on_session_end=on_session_end)
|
||||||
|
|
||||||
async def _on_reset_aggregation(
|
async def _on_reset_aggregation(
|
||||||
self, controller: UserTurnController, strategy: BaseUserTurnStartStrategy
|
self, controller: UserTurnController, strategy: BaseUserTurnStartStrategy
|
||||||
):
|
):
|
||||||
logger.debug(f"{self}: Resetting aggregation (strategy: {strategy})")
|
logger.debug(f"{self}: Resetting aggregation (strategy: {strategy})")
|
||||||
await self._cancel_transcript_gather()
|
await self._cancel_post_turn_transcript_wait()
|
||||||
await self.reset()
|
await self.reset()
|
||||||
|
|
||||||
async def _cancel_transcript_gather(self):
|
async def _cancel_post_turn_transcript_wait(self):
|
||||||
"""Cancel any active transcript-gather window without finalizing.
|
"""Cancel any active post-turn transcript wait window without finalizing.
|
||||||
|
|
||||||
Called from reset paths (interruption, explicit reset).
|
Called from reset paths (interruption, explicit reset).
|
||||||
"Reset" means "throw it away" — we don't flush a partial
|
"Reset" means "throw it away" — we don't flush a partial
|
||||||
transcript that was about to be invalidated anyway.
|
transcript that was about to be invalidated anyway.
|
||||||
"""
|
"""
|
||||||
if self._transcript_gather_task:
|
if self._post_turn_transcript_wait_task:
|
||||||
await self.cancel_task(self._transcript_gather_task)
|
await self.cancel_task(self._post_turn_transcript_wait_task)
|
||||||
self._transcript_gather_task = None
|
self._post_turn_transcript_wait_task = None
|
||||||
self._gathering_for_strategy = None
|
self._post_turn_transcript_wait_strategy = None
|
||||||
self._inference_during_gather = False
|
self._inference_during_post_turn_transcript_wait = False
|
||||||
|
|
||||||
async def _on_user_turn_stop_timeout(self, controller):
|
async def _on_user_turn_stop_timeout(self, controller):
|
||||||
await self._call_event_handler("on_user_turn_stop_timeout")
|
await self._call_event_handler("on_user_turn_stop_timeout")
|
||||||
@@ -1304,7 +1310,7 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
):
|
):
|
||||||
"""Finalize the user turn: flush the message, emit both events.
|
"""Finalize the user turn: flush the message, emit both events.
|
||||||
|
|
||||||
Used in the default mode (``_aggregator_gathers_transcripts ==
|
Used in the default mode (``_wait_for_post_turn_transcripts ==
|
||||||
False``), where end of turn and user message finalization
|
False``), where end of turn and user message finalization
|
||||||
coincide. Emits both ``on_user_turn_stopped`` and
|
coincide. Emits both ``on_user_turn_stopped`` and
|
||||||
``on_user_turn_message_finalized``.
|
``on_user_turn_message_finalized``.
|
||||||
@@ -1325,9 +1331,9 @@ class LLMUserAggregator(LLMContextAggregator):
|
|||||||
):
|
):
|
||||||
"""Finalize the user message: flush to context, emit one event.
|
"""Finalize the user message: flush to context, emit one event.
|
||||||
|
|
||||||
Used when the aggregator gathers transcripts after the user
|
Used when the aggregator runs a post-turn transcript wait
|
||||||
turn ends (``_aggregator_gathers_transcripts == True``), where
|
(``_wait_for_post_turn_transcripts == True``), where user
|
||||||
user message finalization fires after the end of turn. Emits
|
message finalization fires after the end of turn. Emits
|
||||||
``on_user_turn_message_finalized`` only; ``on_user_turn_stopped``
|
``on_user_turn_message_finalized`` only; ``on_user_turn_stopped``
|
||||||
was already emitted at the end of turn.
|
was already emitted at the end of turn.
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -767,10 +767,10 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
|
|||||||
"""``wait_for_transcript_to_end_user_turn=False`` splits the lifecycle:
|
"""``wait_for_transcript_to_end_user_turn=False`` splits the lifecycle:
|
||||||
|
|
||||||
- ``on_user_turn_stopped`` fires at the end of turn with empty
|
- ``on_user_turn_stopped`` fires at the end of turn with empty
|
||||||
content (the aggregator hasn't gathered transcripts yet).
|
content (no transcripts have arrived yet).
|
||||||
- Transcripts arriving after the end of turn are captured into
|
- Transcripts arriving after the end of turn are captured into
|
||||||
``_aggregation``.
|
``_aggregation``.
|
||||||
- When the transcript-gather timer fires,
|
- When the post-turn transcript wait timer fires,
|
||||||
``on_user_turn_message_finalized`` fires with the populated
|
``on_user_turn_message_finalized`` fires with the populated
|
||||||
user context message.
|
user context message.
|
||||||
"""
|
"""
|
||||||
@@ -817,7 +817,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
|
|||||||
# Transcripts arrive after the end of turn (just one
|
# Transcripts arrive after the end of turn (just one
|
||||||
# here for the basic case).
|
# here for the basic case).
|
||||||
TranscriptionFrame(text="Hello!", user_id="", timestamp="now"),
|
TranscriptionFrame(text="Hello!", user_id="", timestamp="now"),
|
||||||
# Wait for the transcript-gather timer to fire.
|
# Wait for the post-turn transcript wait timer to fire.
|
||||||
SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05),
|
SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05),
|
||||||
]
|
]
|
||||||
await run_test(pipeline, frames_to_send=frames_to_send)
|
await run_test(pipeline, frames_to_send=frames_to_send)
|
||||||
@@ -830,10 +830,10 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
|
|||||||
user_messages = [m for m in context.get_messages() if m.get("role") == "user"]
|
user_messages = [m for m in context.get_messages() if m.get("role") == "user"]
|
||||||
self.assertEqual([m["content"] for m in user_messages], ["Hello!"])
|
self.assertEqual([m["content"] for m in user_messages], ["Hello!"])
|
||||||
|
|
||||||
async def test_no_wait_for_transcript_uses_stt_metadata_for_gather_timer(self):
|
async def test_no_wait_for_transcript_uses_stt_metadata_for_wait_timer(self):
|
||||||
"""The transcript-gather timer prefers the STT-reported P99 TTFS
|
"""The post-turn transcript wait timer prefers the STT-reported P99 TTFS
|
||||||
over ``DEFAULT_TTFS_P99``. With a long ``DEFAULT_TTFS_P99`` and
|
over ``DEFAULT_TTFS_P99``. With a long ``DEFAULT_TTFS_P99`` and
|
||||||
a short STT-reported value, the gather completes by the shorter
|
a short STT-reported value, the wait completes by the shorter
|
||||||
time — if the timer fell back to ``DEFAULT_TTFS_P99``, this test
|
time — if the timer fell back to ``DEFAULT_TTFS_P99``, this test
|
||||||
would hang.
|
would hang.
|
||||||
"""
|
"""
|
||||||
@@ -880,7 +880,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
|
|||||||
# fires turn-stopped.
|
# fires turn-stopped.
|
||||||
SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05),
|
SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05),
|
||||||
TranscriptionFrame(text="Hello!", user_id="", timestamp="now"),
|
TranscriptionFrame(text="Hello!", user_id="", timestamp="now"),
|
||||||
# Wait for the transcript-gather timer to fire (sized
|
# Wait for the post-turn transcript wait timer to fire (sized
|
||||||
# to the STT-reported TTFS, not DEFAULT_TTFS_P99).
|
# to the STT-reported TTFS, not DEFAULT_TTFS_P99).
|
||||||
SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05),
|
SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05),
|
||||||
]
|
]
|
||||||
@@ -889,7 +889,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
|
|||||||
self.assertEqual(events, [("stopped", None), ("finalized", "Hello!")])
|
self.assertEqual(events, [("stopped", None), ("finalized", "Hello!")])
|
||||||
|
|
||||||
async def test_no_wait_for_transcript_no_transcripts_arrive(self):
|
async def test_no_wait_for_transcript_no_transcripts_arrive(self):
|
||||||
"""When no transcripts arrive, the transcript-gather timer still
|
"""When no transcripts arrive, the post-turn transcript wait timer still
|
||||||
runs — ``on_user_turn_message_finalized`` fires with empty
|
runs — ``on_user_turn_message_finalized`` fires with empty
|
||||||
content and nothing is written to context.
|
content and nothing is written to context.
|
||||||
"""
|
"""
|
||||||
@@ -999,7 +999,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
|
|||||||
# simplicity).
|
# simplicity).
|
||||||
TranscriptionFrame(text="Hello!", user_id="", timestamp="now"),
|
TranscriptionFrame(text="Hello!", user_id="", timestamp="now"),
|
||||||
SleepFrame(),
|
SleepFrame(),
|
||||||
# Turn 2 starts before turn 1's transcript-gather timer
|
# Turn 2 starts before turn 1's post-turn transcript wait timer
|
||||||
# fires — precondition violation. The aggregator should
|
# fires — precondition violation. The aggregator should
|
||||||
# force-flush turn 1 first.
|
# force-flush turn 1 first.
|
||||||
VADUserStartedSpeakingFrame(),
|
VADUserStartedSpeakingFrame(),
|
||||||
@@ -1030,7 +1030,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
|
|||||||
Correct ordering requires the user aggregator's deferred
|
Correct ordering requires the user aggregator's deferred
|
||||||
``push_aggregation`` to run before the assistant aggregator's
|
``push_aggregation`` to run before the assistant aggregator's
|
||||||
``push_aggregation`` (which fires on ``LLMFullResponseEndFrame``).
|
``push_aggregation`` (which fires on ``LLMFullResponseEndFrame``).
|
||||||
The patched-short transcript-gather timer plus the sleep
|
The patched-short post-turn transcript wait timer plus the sleep
|
||||||
between LLM start and end make that constraint hold here.
|
between LLM start and end make that constraint hold here.
|
||||||
"""
|
"""
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
@@ -1067,11 +1067,11 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
|
|||||||
# service has finally emitted them — just one here).
|
# service has finally emitted them — just one here).
|
||||||
TranscriptionFrame(text="What's the weather?", user_id="", timestamp="now"),
|
TranscriptionFrame(text="What's the weather?", user_id="", timestamp="now"),
|
||||||
# Bot starts responding. Ordering correctness depends on
|
# Bot starts responding. Ordering correctness depends on
|
||||||
# the user's transcript-gather timer firing before
|
# the user's post-turn transcript wait timer firing before
|
||||||
# LLMFullResponseEndFrame below.
|
# LLMFullResponseEndFrame below.
|
||||||
LLMFullResponseStartFrame(),
|
LLMFullResponseStartFrame(),
|
||||||
LLMTextFrame("It's sunny."),
|
LLMTextFrame("It's sunny."),
|
||||||
# Allow time for the user's transcript-gather timer to
|
# Allow time for the user's post-turn transcript wait timer to
|
||||||
# fire (flushing the user message to context) before
|
# fire (flushing the user message to context) before
|
||||||
# the assistant turn ends.
|
# the assistant turn ends.
|
||||||
SleepFrame(sleep=0.1),
|
SleepFrame(sleep=0.1),
|
||||||
@@ -1180,7 +1180,7 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
|
|||||||
At session end the aggregated text is flushed and
|
At session end the aggregated text is flushed and
|
||||||
``on_user_turn_message_finalized`` fires with the content.
|
``on_user_turn_message_finalized`` fires with the content.
|
||||||
``on_user_turn_stopped`` doesn't fire — when the aggregator
|
``on_user_turn_stopped`` doesn't fire — when the aggregator
|
||||||
gathers transcripts after the turn ends, it's reserved for
|
runs a post-turn transcript wait, that event is reserved for
|
||||||
the end-of-turn path.
|
the end-of-turn path.
|
||||||
"""
|
"""
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|||||||
Reference in New Issue
Block a user