Common Library

Pipeline Abstractions

ai.promethist.pipeline defines how a conversation is executed, without saying anything about a concrete provider or agent type. The engine supplies the implementations — see Engine Pipeline.

Pipeline

Pipeline is the abstract runtime for one conversation. Its constructor takes the whole set of available PipelineComponent instances and sorts them by type:

abstract class Pipeline(components: Set<PipelineComponent>) {
    val extractors = components.filterIsInstance<Extractor<Context, *>>()
    val elementFilters = components.filterIsInstance<ElementFilter>()
    val blockElementFilters = components.filterIsInstance<BlockElementFilter>()
    val actionMappers = components.filterIsInstance<ActionMapper>().sortedBy { /* @Order */ }
    val speechSynthesizers = components.filterIsInstance<SpeechSynthesizer>()
    val interactionManagers = components.filterIsInstance<InteractionManager<Context>>()

    abstract val properties: PipelineProperties
    abstract val configuration: PipelineConfiguration
    abstract val authentication: Mono<Authentication>

    abstract fun getUser(session: Session, speaker: Mono<Speaker>?, anonymousMode: Boolean): Mono<User>
    abstract fun process(elements: Flux<Element>): Flux<OutputElement>
    abstract fun saveContext(context: Context): Mono<Boolean>

    fun createInput(turnElements: Flux<Element>, setupName: String): Input
}

That type-based sorting is the extension mechanism of the whole platform: a component is added by declaring a bean, and the pipeline discovers it.

PipelineConfiguration and PipelineKey

PipelineConfiguration is the immutable per-connection settings object: session ID, device ref, accepted locales and time zone, audio input mode, encoding and sample rates, output media and chunk size, input and output ElementFormat, protocol version, the feature flags (speechProcessing, textOnly, noSystemPrompt, anonymousMode, testMode, useAudioFromStart, useCaching), the audio suppression durations, the realtime configuration ID and the tested agent ref. Its locale(locales) function resolves the effective locale by matching the accepted locales against what an agent supports, falling back from language and country to language alone.

PipelineKey parses the {setupName}:{contentRef} key form:

data class PipelineKey(val setupName: String, val contentRef: Ref)

Context

Context is the per-turn state handed to every component:

interface Context : TtsContext {
    val pipeline: Pipeline
    val memory: Memory
    val turn: Turn
    val user: User
    val session: Session
    val locales: List<Locale>
    val animationDataType: AnimationDataType
    val isRecordingEnabled: Boolean
    val isAnonymous: Boolean
    val hasMemory: Boolean
    val currentTime: ZonedDateTime
    val locale: Locale
    var image: Binary?
    val promptInvocationCapture: PromptInvocationCapture
    fun addTranscript(transcript: Text): Mono<Void>
}
Type Role

ContextBase

Abstract base implementing newTurn() (creating the Turn and incrementing the session’s turn count) and addTranscript.

BasicContext

Context of an agent-less pipeline.

AgentContext<A : Agent>

Adds the project, the typed agent, its activeSkills and the sharedAgents, and implements RealtimeContext.

RealtimeContext

Adds the realtime session, the logging capture, the gender used for synthesis, and next() to advance to the following turn.

ContextService

The contract for building and persisting a context; implementations are selected per pipeline setup.

Input

Input wraps one turn’s element stream and exposes it as the several shared views components need, so the stream is consumed once regardless of how many components look at it:

class Input(
    elements: Flux<Element>,
    extractors: List<Extractor<Context, *>> = emptyList(),
    actionMappers: List<ActionMapper> = emptyList(),
) {
    val binaryElements: Flux<Binary>
    val textualElements: Flux<Textual>
    val interimTranscripts: Flux<Transcript>
    val transcripts: Flux<Transcript>
    val transcript: Mono<Text>     // collapses textual elements, resolving actions
    val bytes: Mono<ByteArray>     // concatenated audio

    val extractions: List<NamedValue<out Any>>
    fun extractionResult(name: String): Mono<out Any>
    fun extractionResults(logAttributes: Map<String, Any?>): Flux<out Any>

    lateinit var context: Context
}

The context is assigned by the pipeline once the turn’s context exists, which is why Input is created through Pipeline.createInput.

Component Interfaces

PipelineComponent is a marker interface; the sub-interfaces below are what implementations actually declare. Their engine implementations are listed in Interaction Pipeline Components.

Interface Contract

Extractor<C : Context, T : Any>

process(context: C, input: Input): ExtractionResult<T> — derive structured data from a turn.

ElementFilter

process(context: Context, elements: Flux<Element>): Flux<Element> — transform the element stream. The companion folds a list of filters into a chain.

BlockElementFilter

process(context: RealtimeContext, elements: Flux<Element>, block: (Flux<Element>) → Flux<RealtimeElement>): Flux<RealtimeElement> — a filter that wraps the rest of the processing rather than sitting in front of it, which is what recording and emotional interpolation need.

ActionMapper

matches(context, name) and process(context, name, parameters): Mono<Action> — turn a #name token into an Action carrying the wording for the model. The companion parses the #name:key=value syntax out of raw text.

SpeechSynthesizer

name, fileTypes, maxSampleRate(fileType) and animationDataTypes(locale). Whole-text synthesis extends TextSpeechSynthesizer (checkpointing and error mapping included); streaming synthesis implements TokenSpeechSynthesizer over a Flux<Textual>.

SpeechRecognizer

handle(session, config: SttStreamConfig, audioElements: Flux<Binary>): Flux<Transcript>, wrapped by process which maps provider errors to SpeechRecognitionException.

InteractionManager<C : RealtimeContext>

What every conversation needs: getSystemPrompt(context, model), getToolBox(context) and the hooks onSessionStart, onSessionDisconnect and onChatEvent (all but the prompt defaulting to no-ops).

TurnInteractionManager<C : RealtimeContext>

For a model whose conversation is delimited into turns: adds planningManager, process(context, input) and the hook turnEnd.

TurnlessInteractionManager<C : RealtimeContext>

For a model that speaks continuously: adds onChatMessage. A TurnlessDelegatingInteractionManager also answers the model’s delegations with onSessionDelegation.

RealtimePlanningManager<C : RealtimeContext>

process(context, input, toolHandler, multimodalShown): Flux<RealtimeElement> — the planning and tool execution step of a turn. BlankPlanningManager is the no-op implementation.

Supporting Types

RealtimeSession, RealtimeLoggingCapture, PromptInvocationCapture and TurnLogCapture collect what happened during a turn — the realtime provider’s session, the model invocations, the prompts and the timing — so it can be logged and persisted after the turn completes. TestPipeline is a pipeline implementation used in tests.