Skip to content

core.streaming

Streaming modules: bounded-queue producer/consumer for TTS pipeline with backpressure and interrupt support.

Streaming Coordinator

core.streaming.streaming_coordinator

Streaming coordinator — bounded-queue producer/consumer for TTS.

Replaces the inline unbounded asyncio.Queue + two ad-hoc coroutines in main.py's queue_agent_responses(). The coordinator provides

  • a bounded audio queue (default maxsize=8) with backpressure timeout and fallback drop policy,
  • a user-interrupt monitor (runs while the agent is speaking),
  • can_speak_event lifecycle management,
  • audio_playback_lock usage to prevent overlapping playback,
  • structured event publication for the entire streaming lifecycle,
  • telemetry recording via the MetricsCollector singleton,
  • a StreamingResult summary on completion.

The coordinator is constructed with callables for TTS synthesis and audio playback, making it fully testable without heavy imports.

StreamingCoordinator

Bounded-queue producer/consumer for agent → TTS → playback.

Parameters

synthesize_fn: Synchronous callable that returns an audio tensor for a given sentence. Will be dispatched to run_in_executor. playback_fn: Async callable that plays audio data (should use run_in_executor internally or be provided pre-wrapped). event_bus: EventBus instance for publishing lifecycle events. interrupt_fn: Optional external interrupt check (API/UI cancellation). max_queue_size: Bounded queue capacity. backpressure_timeout: Seconds to wait when the queue is full before dropping a sentence.

coordinate(sentences: list[str] | None = None, *, sentences_generator: AsyncGenerator[str, None] | None = None, sample_rate: int = 24000, correlation_id: str = '') -> StreamingResult async

Run the full producer/consumer pipeline for sentences.

Parameters

sentences: A list of sentences to process. Mutually exclusive with sentences_generator. If neither is provided, returns an empty StreamingResult. sentences_generator: An async generator yielding sentences one-at-a-time (enables true streaming where sentences enter the pipeline as they are produced by the agent). Mutually exclusive with sentences. sample_rate: Audio sample rate (default 24000). correlation_id: Optional correlation ID for event tracing.

Returns

StreamingResult with summary statistics.

interrupt(reason: InterruptReason = InterruptReason.USER_SPEECH) -> None

Signal the coordinator to stop early.

Streaming Events

core.streaming.streaming_events

Data classes for the streaming coordination pipeline.

Defines the result types returned by StreamingCoordinator and the set of reasons why a streaming run may be interrupted.

InterruptReason

Bases: Enum

Why a streaming run was interrupted (or NONE if it completed naturally).

CANCELLED = 'cancelled' class-attribute instance-attribute

Explicit cancellation request (e.g. from API or UI command).

ERROR = 'error' class-attribute instance-attribute

An unrecoverable error occurred mid-stream.

NONE = 'none' class-attribute instance-attribute

Run finished normally (all sentences processed, queue drained).

USER_SPEECH = 'user_speech' class-attribute instance-attribute

User speech detected during playback (VAD interrupt).

StreamingResult dataclass

Summary statistics for a streaming coordination run.

Returned by StreamingCoordinator.coordinate() when the producer and consumer coroutines complete.

to_dict() -> dict[str, Any]

Serialize to a plain dictionary (JSON-friendly).