Interaction Pipeline

Realtime Models

A realtime model is what actually conducts a conversation. It receives the rendered system prompt and a tool box, and returns a RealtimeSession that consumes input elements and produces realtime elements — transcripts, response text, audio, tool events and response boundaries.

Each provider is one implementation. InternalRealtimeModel is the exception: it emulates a realtime provider by composing separate speech recognition, chat and speech synthesis components.

Model Types and What They Compose

Realtime model types and the other pipeline components
Figure 1. Realtime model types, and the pipeline components around them

The upper half of the diagram is the distinction that matters most. A provider realtime model receives audio and returns audio: recognition, generation and synthesis all happen inside the provider’s own session, and the engine only feeds it elements and reads them back. InternalRealtimeModel instead composes a conversation out of three engine components — SpeechRecognitionManager for recognition, a streaming chat model resolved through the LLM factory, and SpeechService for synthesis — which is what allows a realtime conversation to run on providers that offer no speech-to-speech API, and to mix a recognizer, a language model and a voice from different vendors.

The lower half shows the other pipeline component categories and when they run: input normalization and block filters around each turn, the interaction and planning managers deciding what the agent does, extractors turning the turn into structured memory, and — at the end of the session — aggregators. Those are catalogued in Interaction Pipeline Components.

The Contract

RealtimeModel (common/src/main/kotlin/ai/promethist/ai/RealtimeModel.kt):

interface RealtimeModel<RP : RealtimeProperties> {
    val name: String get() = this::class.simpleName!!.removeSuffix("RealtimeModel")

    fun process(
        context: RealtimeContext,
        systemPrompt: SystemPrompt,
        toolBox: ToolBox,
        canDelegate: Boolean = false,
    ): RealtimeSession<RP>
}

Two things are worth noting:

  • name is derived from the class name by convention — OpenAIRealtimeModel is OpenAI, GeminiLiveRealtimeModel is GeminiLive, InternalRealtimeModel is Internal. This is the name matched against the provider of the resolved properties, so a model is registered simply by being a bean.

  • process is synchronous and opens nothing. It only constructs the session object; the provider connection is established lazily when the session’s element stream is first subscribed. The provider’s own properties are read from the context (context.realtimeProperties), and canDelegate only matters to a turnless model: it says whether the interaction manager answers the model’s delegations, see OpenAILiveRealtimeModel.

RealtimeSession

RealtimeSession (common/src/main/kotlin/ai/promethist/pipeline/RealtimeSession.kt) is the per-conversation provider session:

abstract class RealtimeSession(val systemPrompt: SystemPrompt) : LoggingContext {
    var lastActive: Instant
    var activeDuration: Long
    abstract val realtimeProperties: RealtimeProperties
    var tokenUsage = RealtimeTokenUsage()
    protected open val whenReady: Mono<Unit>

    fun activity(): Long
    fun process(context, transcriptRecognizer, elements, toolHandler): Flux<RealtimeElement>
    fun process(instruction: RealtimeInstruction): Mono<Void>

    protected abstract fun handle(elements: Flux<Element>, toolHandler: ToolHandler): Flux<RealtimeElement>
    protected abstract fun handle(instruction: RealtimeInstruction): Mono<Void>
}
Member Role

systemPrompt

Rendered once at session creation and fixed for the session’s lifetime.

realtimeProperties

The provider’s own properties — read across the pipeline for provider, sampleRate, temperature and sttProvider.

tokenUsage

Accumulated by the provider session as usage events arrive, and read at session end for the usage record.

activity() / activeDuration / lastActive

Active-time accounting for usage metering. Each gap contributes at most 30 seconds, so an idle connection does not inflate the total.

whenReady

Provider readiness gate. Input is held with delayUntil { whenReady }, which is how the engine avoids sending audio before the provider has acknowledged the session configuration. Providers that need no handshake are ready immediately.

What the Base Class Adds

The public process(context, transcriptRecognizer, elements, toolHandler) is the only entry point the pipeline uses, and it wraps the provider’s handle with behaviour every provider gets for free:

  1. Speech recognition routing. When a transcript recognizer is present and realtimeProperties.sttProvider is not None, raw audio is split out of the input and routed through SpeechRecognitionManager, while everything else passes through. Interim transcripts are forwarded to the client and into the model in parallel. When sttProvider is None the audio goes to the provider untouched and the provider does its own recognition and turn detection.

  2. Transcript normalization — guarantees at least one and at most one interim transcript before the final transcripts of a turn, so client-side barge-in logic has a stable signal.

  3. Turnless sentence release. A turnless model (isTurnless) has no response to end, so no sentence window ever closes - and the sentence assembly, which looks one token ahead to tell the end of a sentence from a dot inside "14. října", holds the last token back until the next one arrives. For such a model a sentence is therefore released once its closing token (., ? or !, with trailing whitespace and closing quotes ignored) has been followed by 300 ms without a further token, instead of when the model speaks again. No ResponseEnd is produced and there is no turn boundary. Only a token that closes a sentence starts the wait, so a pause in the middle of a sentence releases nothing and drops no text. A sentence that goes on after a dot more than 300 ms later - such as an ordinal at the end of a token - is split at that dot.

  4. Token to sentence assembly — groups streamed tokens into sentences per response window.

  5. Idle response end. If the stream falls silent for 20 seconds, a synthetic ResponseEnd is injected with a warning. Turn windowing downstream is driven by ResponseEnd, so this is what prevents a stalled provider from hanging a turn forever. Turnless models are excluded: they stream audio without a pause, so it would never fire.

  6. Error normalization. Everything leaving a session is mapped to a RealtimeException, whether it was thrown synchronously or signalled reactively.

process(instruction) is the injection leg used by planning. It wraps the provider’s handle(instruction) and deliberately swallows errors — a failed instruction injection must not kill a live conversation.

Realtime Elements

Realtime elements are the internal vocabulary between a session and the pipeline. They are translated to client-facing output elements centrally, in Context.outputElement (common/src/main/kotlin/ai/promethist/pipeline/Functions.kt); an element with no mapping is a programming error and fails loudly.

Element Meaning

Transcript

A user speech recognition result: isInterim, isSynthetic, isAsynchronous, provider. An interim transcript is the speech-start and barge-in signal; providers often emit a "." sentinel for it. Asynchronous transcripts arrive out of band from the provider’s own transcription.

Token

A streamed response text delta.

Text

A complete response sentence, either assembled from tokens or delivered whole by providers that do not stream.

SpeechItem

A chunk of output audio with provider, sampleRate, isLast and optional animation data (visemes, blendShapeWeights).

ResponseEnd

The end-of-response boundary. This is the token the whole pipeline windows on, and it reaches the client as the response-end command.

RealtimeInstruction

Steering text injected into the model: text plus an InstructionType of INSTRUCTION, COMMENTARY or SUPPORTING_CONTEXT. Only SUPPORTING_CONTEXT is contextual and merely enriches the model’s context; the other two are non-contextual and make the model respond — an INSTRUCTION as direction, a COMMENTARY as something to say. Produced by planning and by per-turn context injection.

RealtimeDelegation

A turnless model asking for help: an id, the text of what the user said (read back from the transcripts, since the model’s request carries none) and the offsetMs it refers to. Mapped onto the realtime-delegation command, and handed to the interaction manager when it takes delegations.

RealtimeCommand

An out-of-band control or telemetry event with a name and parameters, mapped one-to-one onto a client command. Names in use include tool-start, tool-success, tool-error, tool-call, response-instruction-end, realtime-planning and sessionShouldEnd.

RealtimeError

A localized, user-presentable error that does not terminate the stream.

MultimodalInteraction

A UI card or handover produced by a tool call. Also folded into the set of shown multimodals that planning receives.

EmotionalState

Prosody and emotion scores, rendered as visual state with userSpeaking / userNotSpeaking signals.

Properties and Provider Selection

RealtimeProperties is Jackson-polymorphic on its provider property, so a stored RealtimeConfiguration carries the provider’s own typed settings:

interface RealtimeProperties {
    val provider: String get() = this::class.simpleName!!.removeSuffix("RealtimeProperties")
    val isTurnless: Boolean get() = false
    val hasToolSupport: Boolean get() = true
    val gender: Gender?
    val sttProvider: String
    val sampleRate: Int
    val temperature: Double
}

isTurnless marks a model that speaks continuously and has no response that ends. It decides which pipeline manager drives the conversation (see Pipeline Managers), and today only OpenAI GPT-Live sets it. hasToolSupport is false for a model that calls no tool of its own.

sttProvider is the switch between the two operating modes described above — a provider name (for example Soniox) routes audio through the engine’s own recognition, None leaves recognition to the provider.

The configuration itself is resolved by RealtimeConfigurationService, which caches each lookup for promethist.realtime-configuration.ttl-seconds (900 by default) and fails with a not-found error for an unknown ID. Where the ID comes from depends on the setup:

  • AgentContextService uses the agent’s realtimeConfigurationId;

  • BasicContextService uses the connection’s realtimeConfigurationId parameter (or the X-Realtime-Configuration-ID header), which is what the bridge and preview flows set.

The model is then picked by string match, in RealtimeSessionFactory:

val model = realtimeModels.firstOrNull { it.name == context.realtimeProperties.provider } as RealtimeModel<RealtimeProperties>?

Providers

Model Behaviour

OpenAIRealtimeModel

OpenAI Realtime API over WebSocket. Built on OpenAIBaseRealtimeModel, which implements the protocol: a session update carrying the instructions, tools, audio formats, transcription model and turn detection; then conversation items and response requests for text input, audio append for raw audio, and a response cancel on an interim transcript (barge-in). Inbound events map to transcripts, tokens, speech items and tool calls.

OpenAILiveRealtimeModel

OpenAI GPT-Live, a speech-to-speech model with no turns, over its own WebSocket protocol. Selected when the model of the OpenAI properties starts with gpt-live. See OpenAILiveRealtimeModel.

AzureOpenAIRealtimeModel

The same protocol against Azure OpenAI, round-robining across the configured regions so load and quota spread over deployments.

GeminiLiveRealtimeModel

Google Gemini Live through the GenAI SDK rather than a raw socket. Configures response modality, system instruction, voice, input and output transcription with language hints, context-window compression and — only when sttProvider is None — automatic activity detection with sensitivity and silence thresholds. Besides raw audio it accepts image elements as video frames of the realtime input, so a client can stream camera or screen snapshots (Gemini samples them at about one frame per second) alongside speech. Tool calls run concurrently and can be cancelled by the server.

ElevenLabsRealtimeModel

ElevenLabs conversational API. The provider does its own turn detection (sttProvider is None) and sends no response-done event, so the session synthesizes ResponseEnd from a one-second idle timer.

HumeAIRealtimeModel

Hume AI EVI. In addition to transcripts and audio it emits emotional state from prosody scores, which is what drives the visual state for this provider.

InternalRealtimeModel

The engine’s own composition — see below.

Provider Quirks Worth Knowing

  • OpenAI and Azure sessions expire after the provider’s hard session limit; that is surfaced as a non-recoverable error with its own localized message. Two benign protocol codes (an already active response, and cancelling a response that is not active) are logged and ignored.

  • A recoverable provider error does not end the conversation: the pending input transcript is flushed, a recovery instruction ("you missed the user’s input, apologize and ask them to repeat") is injected into the model, and the error is emitted to the client.

  • Gemini Live handles the provider’s go-away by reconnecting with a stored resumption handle.

  • ElevenLabsRealtimeModel does not implement instruction injection, so planning instructions and per-turn context injection have no effect with that provider.

  • ElevenLabs and Hume AI do not report token usage, so the usage record of those sessions carries zero tokens.

OpenAILiveRealtimeModel

OpenAILiveRealtimeModel (engine/component/OpenAILiveRealtimeModel.kt) serves OpenAI GPT-Live, which is named OpenAILive because the provider of the OpenAI properties is OpenAILive whenever their model starts with gpt-live. The protocol differs from the Realtime API, hence the separate component.

The model owns the spoken conversation. It speaks continuously, decides for itself when to talk, and has no end-of-response event and no way to be told to respond — so the session synthesizes no turn boundary, and TurnlessPipelineManager drives it, see Turnless Conversations.

  • Connection. A WebSocket to /v1/live/sessions (promethist.realtime.openai-live.path). The handshake must not carry an Origin header, which Reactor Netty adds, or the API answers 403; the session removes it.

  • Startup. One session.start frame with the model, the instructions, the voice, PCM audio at the configured sample rate in both directions, and the delegation setting. Input transcription, turn detection and noise reduction are managed by the server and rejected as parameters. The session then sends 200 ms of silence, as the model ignores everything appended before it has received any input audio.

  • Input is audio only. An Action such as #intro is appended on the instructions channel followed by a directive to act on it now, since without response.create a vaguer wording reliably produces silence. Any other element has no channel and is dropped — a Commit included, as there are no turns to commit.

  • Input clock. The model’s timeline — what it hears, says and reasons about — advances only as input audio arrives. A client that stops sending audio, such as a push-to-talk client between presses, would freeze the model mid-sentence, so every second without a chunk of audio is answered with a second of silence (silentIfEmpty).

  • Output is the user’s transcript, always asynchronous so it never barges the client in and emitted once the transcription has been quiet for utteranceIdleTimeoutMs (700 ms by default); the model’s transcript as tokens, which the base session joins into sentences; and the model’s audio as speech items. Usage is reported in seconds of voice and no tokens.

  • Instructions reach the model on one of three append channels, by their type: INSTRUCTION on the instructions channel (it may interrupt the current speech), COMMENTARY on the commentary channel (what the model speaks, paraphrasing it) and SUPPORTING_CONTEXT on the thinking channel (absorbed silently). An append over maxAppendTokens (500 by default) is rejected, so longer text is split at sentence, line and word boundaries, and at last into windows of characters.

  • Delegation. The model decides when it needs help and asks with session.delegation.created, which carries only an id and an offset. The session fixes who answers for its whole lifetime:

    • client delegation — the engine does. The delegation, carrying the text read back from the transcripts, goes to the interaction manager, and the model has no tools of its own. Used when the interaction manager is a TurnlessDelegatingInteractionManager.

    • responses delegation — GPT-Live does, by calling a Responses API backend (delegationModel, gpt-5.1 by default) configured with the context part of the system prompt and the tool box’s tools. The session only executes the function calls that backend asks for, and once all are answered continues the response with response.create. The session instructions, without the context, tell the model that the backend exists and when to use it; without that it answers from its own knowledge.

  • Errors. A rejected append (invalid_request_error) or a transient server error costs that piece of context, not the conversation, and is reported to the client; any other error ends the session.

InternalRealtimeModel

InternalRealtimeModel (engine/component/InternalRealtimeModel.kt) satisfies the same contract by assembling the conversation from separate parts, which is how the platform runs a realtime conversation on providers that have no speech-to-speech API:

  • Recognition comes from the base class, because InternalRealtimeProperties.sttProvider defaults to a real provider — the session itself only ever sees transcripts and text.

  • Generation uses a streaming chat model resolved through SpringAiLLMFactory from InternalRealtimeProperties.modelConfiguration, so any configured chat provider can drive a realtime conversation.

  • Synthesis streams the generated tokens into SpeechService, which interleaves sentence text with speech items and can stream tokens directly into a token-streaming synthesizer.

Its turn loop is where barge-in is implemented explicitly: inputs are windowed on interim transcripts and the in-flight generation is dropped when a new interim transcript arrives, with a synthetic ResponseEnd closing the interrupted response. A final transcript starts a generation; within a window inputs are answered sequentially.

Each response is assembled as: the echoed input as a transcript, then the speech (or plain sentences when the connection is textOnly), then any elements produced by tool calls, then ResponseEnd. Contextual instructions are kept as a rolling window of the last five and injected as system messages; a non-contextual instruction is fed in as a new input, which triggers a response.

Failures are non-recoverable and typed: an unresolvable provider, a model stream error, or a speech synthesis error.

Where the Lifecycle Is Driven From

Nothing above decides when a session starts, when turns begin and end, or when planning runs — that is the orchestrator’s job. See Realtime Conversation Lifecycle.