vllm_omni.distributed.omni_connectors.adapter ¶

logger `module-attribute` ¶

logger = get_connector_logger(__name__)

compute_talker_prompt_ids_length ¶

compute_talker_prompt_ids_length(
    prompt_ids: list[int],
) -> int

Compute the length of the talker prompt ids.

Parameters:

Name	Type	Description	Default
`prompt_ids`	`list[int]`	The prompt ids tensor.	required

Returns:

Type	Description
`int`	The length of the talker prompt ids.

construct_next_stage_streaming_input_prompt ¶

construct_next_stage_streaming_input_prompt(
    payload_data: dict[str, Any],
    request: Request,
    *,
    max_model_len: int | None = None,
    previous_condition_len: int | None = None,
    previous_condition_seq: int | None = None,
    condition_seq: int | None = None,
    recompute_previous_chunks: int = 0,
    recompute_on_capacity: bool = False,
) -> bool

Update a downstream streaming request prompt from connector payload.

Async-chunk downstream stages are prewarmed before the real Talker prompt is known. When a Thinker payload carries ids.prompt or declares the placeholder length in meta.next_stage_prompt_len, this helper:

Preserves num_computed_tokens while extending a non-window prompt. Explicit replacements and one-condition window recomputes reset it. A full-attention Talker recomputes only when the next condition would exceed its context limit.
Moves already-computed output tokens into prompt_token_ids.
Appends a placeholder slice using the declared downstream prompt length, or derives its length from ids.prompt for legacy producers.
Refreshes block hashes so the scheduler allocates KV slots for the extended prompt without discarding prior computed state.

try_recv_via_connector ¶

try_recv_via_connector(
    task: dict[str, Any],
    connectors: dict[Any, Any],
    stage_id: int,
) -> tuple[Any, dict[str, Any] | None]

Attempts to resolve input data from either connector or IPC. Returns (engine_inputs, rx_metrics) or (None, None) if failed/skipped.

try_send_via_connector ¶

try_send_via_connector(
    connector: Any,
    stage_id: int,
    next_stage_id: int,
    req_id: str,
    next_inputs: Any,
    sampling_params: Any,
    original_prompt: Any,
    next_stage_queue_submit_fn: Callable[
        [dict[str, Any]], None
    ],
    metrics: OrchestratorAggregator,
) -> bool

Attempts to send data via OmniConnector. Returns True if successful, False otherwise. Encapsulates the logic of preparing payload, sending via connector, sending notification, and recording metrics.