Skip to content

vllm_omni.engine.duplex.config

Session configuration, capabilities and value types of the duplex session model.

These classes are the typed half of the public contract: DuplexOmni accepts a DuplexSessionConfig (or the equivalent OpenAI Realtime session object via :meth:DuplexSessionConfig.from_realtime) and reports DuplexCapabilities. They used to live in entrypoints/duplex/protocol.py.

DuplexAssistantAudioTextMark dataclass

audio_end_ms instance-attribute 

audio_end_ms: int

text_chars instance-attribute 

text_chars: int

DuplexAudioChunk dataclass

data instance-attribute 

data: str

format class-attribute instance-attribute 

format: str = 'wav'

sample_rate_hz class-attribute instance-attribute 

sample_rate_hz: int | None = None

DuplexCapabilities dataclass

Runtime/model capabilities exposed by the duplex serving protocol.

These are intentionally explicit so the serving layer does not assume all duplex models support the same input append, rollback, or turn policy. supports_core_kv_lease is reserved for scheduler-owned KV lifecycle; model-owned decoder/TTS state must use supports_model_internal_state. supports_core_resumable_request means the scheduler can resume the same request id across streaming updates, but it is not a KV lease by itself. Realtime support means this endpoint can speak the native Realtime event schema for the supported audio duplex paths while keeping model- or scheduler-specific limits explicit in the capability payload.

adapter_patterns class-attribute instance-attribute 

adapter_patterns: list[str] = field(
    default_factory=lambda: ["chunk_group_append"]
)

chunk_period_ms class-attribute instance-attribute 

chunk_period_ms: int | None = 1000

optional_input_modalities class-attribute instance-attribute 

optional_input_modalities: frozenset[str] = field(
    default_factory=lambda: frozenset({"video"})
)

required_input_modalities class-attribute instance-attribute 

required_input_modalities: frozenset[str] = field(
    default_factory=lambda: frozenset({"audio"})
)

requires_model_runner_kv class-attribute instance-attribute 

requires_model_runner_kv: bool = False

requires_native_stage_role class-attribute instance-attribute 

requires_native_stage_role: bool = False

session_admission_mode class-attribute instance-attribute 

session_admission_mode: str = 'serving_managed'

signal_sources class-attribute instance-attribute 

signal_sources: list[str] = field(
    default_factory=lambda: [
        "client_event",
        "server_policy",
        "model_native",
    ]
)

stage_handoff_transport class-attribute instance-attribute 

stage_handoff_transport: str | None = None

supports_audio_truncate class-attribute instance-attribute 

supports_audio_truncate: bool = False

supports_barge_in class-attribute instance-attribute 

supports_barge_in: bool = True

supports_chat_completions class-attribute instance-attribute 

supports_chat_completions: bool = False

supports_client_commit class-attribute instance-attribute 

supports_client_commit: bool = True

supports_concurrent_turn_requests class-attribute instance-attribute 

supports_concurrent_turn_requests: bool = False

supports_core_kv_lease class-attribute instance-attribute 

supports_core_kv_lease: bool = False

supports_core_resumable_request class-attribute instance-attribute 

supports_core_resumable_request: bool = False

supports_external_turn_signal class-attribute instance-attribute 

supports_external_turn_signal: bool = True

supports_image_input class-attribute instance-attribute 

supports_image_input: bool = False

supports_independent_io_streams class-attribute instance-attribute 

supports_independent_io_streams: bool = False

supports_input_append class-attribute instance-attribute 

supports_input_append: bool = False

supports_kv_lease class-attribute instance-attribute 

supports_kv_lease: bool = False

supports_model_internal_state class-attribute instance-attribute 

supports_model_internal_state: bool = False

supports_model_native_turn_policy class-attribute instance-attribute 

supports_model_native_turn_policy: bool = False

supports_multi_session class-attribute instance-attribute 

supports_multi_session: bool = False

supports_multi_session_same_replica class-attribute instance-attribute 

supports_multi_session_same_replica: bool = False

supports_playback_ack class-attribute instance-attribute 

supports_playback_ack: bool = True

supports_realtime_endpoint class-attribute instance-attribute 

supports_realtime_endpoint: bool = False

supports_reencode_context class-attribute instance-attribute 

supports_reencode_context: bool = True

supports_replace_latest_chunk class-attribute instance-attribute 

supports_replace_latest_chunk: bool = True

supports_rollback_to_checkpoint class-attribute instance-attribute 

supports_rollback_to_checkpoint: bool = False

supports_scheduler_native_append class-attribute instance-attribute 

supports_scheduler_native_append: bool = False

supports_session_adapter class-attribute instance-attribute 

supports_session_adapter: bool = True

supports_session_lease class-attribute instance-attribute 

supports_session_lease: bool = False

supports_session_resume class-attribute instance-attribute 

supports_session_resume: bool = False

supports_stage_connector_handoff class-attribute instance-attribute 

supports_stage_connector_handoff: bool = False

supports_stage_resumption class-attribute instance-attribute 

supports_stage_resumption: bool = False

supports_text_only_turn class-attribute instance-attribute 

supports_text_only_turn: bool = False

supports_turn_commit_only class-attribute instance-attribute 

supports_turn_commit_only: bool = True

target_barge_in_latency_ms class-attribute instance-attribute 

target_barge_in_latency_ms: int | None = 1000

text_turn_priming_units class-attribute instance-attribute 

text_turn_priming_units: int = 0

accepts_input_modality 

accepts_input_modality(modality: str) -> bool

allows_video_without_audio 

allows_video_without_audio() -> bool

Whether video without speech/audio is legal turn content (AURA video-compulsory).

as_dict 

as_dict() -> dict[str, object]

validate_append_modalities 

validate_append_modalities(
    *, has_audio: bool, has_video: bool
) -> str | None

Return an error message when the append modalities violate this contract.

DuplexCommittedInput dataclass

epoch instance-attribute 

epoch: int

input_commit_seq instance-attribute 

input_commit_seq: int

message instance-attribute 

message: dict[str, object]

turn_id instance-attribute 

turn_id: int

DuplexConfigError

Bases: ValueError

A Realtime session/response object was rejected while mapping it onto the duplex config.

code instance-attribute 

code = code

param instance-attribute 

param = param

DuplexOverlapPolicy

Bases: str, Enum

BARGE_IN_ON_SPEECH class-attribute instance-attribute 

BARGE_IN_ON_SPEECH = 'barge_in_on_speech'

LISTEN_ONLY class-attribute instance-attribute 

LISTEN_ONLY = 'listen_only'

DuplexPlaybackCommitPolicy

Bases: str, Enum

ACK_ONLY class-attribute instance-attribute 

ACK_ONLY = 'ack_only'

COMMIT_ALL_ON_DONE class-attribute instance-attribute 

COMMIT_ALL_ON_DONE = 'commit_all_on_done'

DuplexPlaybackCursor dataclass

audio_complete class-attribute instance-attribute 

audio_complete: bool = False

committed_ms class-attribute instance-attribute 

committed_ms: int = 0

generated_ms class-attribute instance-attribute 

generated_ms: int = 0

played_ms class-attribute instance-attribute 

played_ms: int = 0

sent_ms class-attribute instance-attribute 

sent_ms: int = 0

text_requires_complete_audio class-attribute instance-attribute 

text_requires_complete_audio: bool = False

acknowledge 

acknowledge(
    played_ms: int, committed_ms: int | None = None
) -> None

as_dict 

as_dict() -> dict[str, int]

snapshot 

snapshot() -> DuplexPlaybackView

truncate_committed 

truncate_committed(committed_ms: int) -> None

DuplexPlaybackView dataclass

audio_complete class-attribute instance-attribute 

audio_complete: bool = False

committed_ms class-attribute instance-attribute 

committed_ms: int = 0

generated_ms class-attribute instance-attribute 

generated_ms: int = 0

played_ms class-attribute instance-attribute 

played_ms: int = 0

sent_ms class-attribute instance-attribute 

sent_ms: int = 0

text_requires_complete_audio class-attribute instance-attribute 

text_requires_complete_audio: bool = False

as_dict 

as_dict() -> dict[str, int]

DuplexSessionConfig dataclass

extra_body class-attribute instance-attribute 

extra_body: dict[str, object] = field(default_factory=dict)

idle_timeout_s class-attribute instance-attribute 

idle_timeout_s: float = 300.0

initial_user_text class-attribute instance-attribute 

initial_user_text: str | None = None

instructions class-attribute instance-attribute 

instructions: str | None = None

max_tokens class-attribute instance-attribute 

max_tokens: int | None = None

modalities class-attribute instance-attribute 

modalities: list[str] = field(
    default_factory=lambda: ["text", "audio"]
)

model class-attribute instance-attribute 

model: str | None = None

overlap_barge_in_ms class-attribute instance-attribute 

overlap_barge_in_ms: int = 1200

overlap_policy class-attribute instance-attribute 

overlap_policy: str = DuplexOverlapPolicy.LISTEN_ONLY.value

overlap_short_ack_ms class-attribute instance-attribute 

overlap_short_ack_ms: int = 700

overlap_silence_rms class-attribute instance-attribute 

overlap_silence_rms: float = 0.003

playback_commit_policy class-attribute instance-attribute 

playback_commit_policy: str = (
    DuplexPlaybackCommitPolicy.COMMIT_ALL_ON_DONE.value
)

ref_audio class-attribute instance-attribute 

ref_audio: str | None = None

response_format class-attribute instance-attribute 

response_format: str = 'wav'

speed class-attribute instance-attribute 

speed: float | None = None

temperature class-attribute instance-attribute 

temperature: float | None = None

use_tts_template class-attribute instance-attribute 

use_tts_template: bool = True

voice class-attribute instance-attribute 

voice: str | None = None

apply_realtime_update 

apply_realtime_update(
    patch: Mapping[str, object],
    *,
    session_id: str | None = None,
    audio_started: bool = False,
) -> None

Apply a Realtime session.update patch in place.

Raises :class:DuplexConfigError (code in model_update_unsupported, voice_update_after_audio_unsupported, ref_audio_update_unsupported) when the patch changes something a live session cannot change. audio_started is playback.generated_ms > 0 or playback.sent_ms > 0.

as_dict 

as_dict() -> dict[str, object]

from_event classmethod 

from_event(event: dict[str, object]) -> DuplexSessionConfig

from_realtime classmethod 

from_realtime(
    session_payload: Mapping[str, object],
    *,
    model: str | None = None,
) -> DuplexSessionConfig

Build a config from an OpenAI Realtime session object (session.update / open_session).

Applies the wire defaults, validates audio formats and turn detection (raising :class:DuplexConfigError with the same codes the Realtime translator used), fills turn_detection defaults / derives overlap_policy, and stores the Realtime-only fields under the realtime_* keys of extra_body exactly like the old _session_create_from_realtime.

normalized 

normalized() -> DuplexSessionConfig

Apply the clamps / defaults from_event applies to a wire object (typed configs skip them).

DuplexSessionState

Bases: str, Enum

CLOSED class-attribute instance-attribute 

CLOSED = 'closed'

CLOSING class-attribute instance-attribute 

CLOSING = 'closing'

OPEN class-attribute instance-attribute 

OPEN = 'open'

DuplexTurnEventType

Bases: str, Enum

ASSISTANT_DONE class-attribute instance-attribute 

ASSISTANT_DONE = 'assistant_done'

ASSISTANT_STARTED class-attribute instance-attribute 

ASSISTANT_STARTED = 'assistant_started'

BARGE_IN class-attribute instance-attribute 

BARGE_IN = 'barge_in'

CLOSE class-attribute instance-attribute 

CLOSE = 'close'

PLAYBACK_ACK class-attribute instance-attribute 

PLAYBACK_ACK = 'playback_ack'

TIMEOUT class-attribute instance-attribute 

TIMEOUT = 'timeout'

USER_COMMITTED class-attribute instance-attribute 

USER_COMMITTED = 'user_committed'

USER_STARTED class-attribute instance-attribute 

USER_STARTED = 'user_started'

DuplexTurnState

Bases: str, Enum

ASSISTANT_GENERATING class-attribute instance-attribute 

ASSISTANT_GENERATING = 'assistant_generating'

ASSISTANT_PLAYING class-attribute instance-attribute 

ASSISTANT_PLAYING = 'assistant_playing'

BARGE_IN class-attribute instance-attribute 

BARGE_IN = 'barge_in'

IDLE class-attribute instance-attribute 

IDLE = 'idle'

USER_COMMITTED class-attribute instance-attribute 

USER_COMMITTED = 'user_committed'

USER_SPEAKING class-attribute instance-attribute 

USER_SPEAKING = 'user_speaking'

ResponseCreateOptions dataclass

extra_body class-attribute instance-attribute 

extra_body: Mapping[str, object] = field(
    default_factory=dict
)

instructions class-attribute instance-attribute 

instructions: str | None = None

max_tokens class-attribute instance-attribute 

max_tokens: int | None = None

modalities class-attribute instance-attribute 

modalities: tuple[str, ...] | None = None

response_format class-attribute instance-attribute 

response_format: str | None = None

speed class-attribute instance-attribute 

speed: float | None = None

temperature class-attribute instance-attribute 

temperature: float | None = None

voice class-attribute instance-attribute 

voice: str | None = None

apply_to 

apply_to(config: DuplexSessionConfig) -> None

from_realtime classmethod 

from_realtime(
    response_payload: Mapping[str, object],
    *,
    private_runtime_config_keys: frozenset[
        str
    ] = frozenset(),
) -> ResponseCreateOptions

Parse an OpenAI Realtime response object into response-scoped options.

Raises :class:DuplexConfigError (code="unsupported_native_response_options") for options a model-native duplex session cannot apply per response. Private runtime keys in extra_body are dropped.

input_audio_transcription_config 

input_audio_transcription_config(
    session_payload: Mapping[str, object],
) -> dict[str, object] | None

Return the input_audio_transcription object of a Realtime session payload.

realtime_item_to_history_message 

realtime_item_to_history_message(
    item: object,
) -> dict[str, object] | None

Convert a Realtime conversation item into a chat history message (None when empty).

realtime_max_output_tokens 

realtime_max_output_tokens(value: object) -> int | None

Normalize Realtime max output tokens ("inf" -> None).