TensorRT-LLM 源码完全拆解（十九）：多模态输入——视觉编码器、Placeholder Token 与嵌入回填

1
{
2
    "mrope_config": {
3
        "mrope_rotary_cos_sin": torch.Tensor,    # Qwen2/2.5-VL
4
        "mrope_position_deltas": torch.Tensor,   # Qwen2/2.5-VL
5
    },
6
    "multimodal_embedding": torch.Tensor | List[SharedTensor handle dict],
7
    "image": {
8
        "pixel_values": torch.Tensor,
9
        "image_height": torch.Tensor | List[int],
10
        "image_width": torch.Tensor | List[int],
11
    },
12
    ...
13
}

1
_CPU_ONLY_MULTIMODAL_DATA_KEYS = frozenset({
2
    "multimodal_embed_mask_cumsum",
3
    "multimodal_embedding_lengths",
4
    MULTIMODAL_ENCODER_ITEM_METADATA_KEY,
5
})

1
cs = self.embed_mask_cumsum
2
# int(cs[idx]) below would D2H-sync if cs lived on CUDA.
3
assert cs.device.type == "cpu", f"embed_mask_cumsum must be CPU, got {cs.device}"

1
The encoding is canonical and self-describing: every value is
2
[1-byte type tag][typed metadata][length-prefixed payload] with all
3
multi-byte integers big-endian. This prevents cache-key hash collisions
4
between distinct values that happen to share a raw byte payload (for
5
example transposed image dimensions or reshaped arrays).

1
_VIDEO_HASH_METADATA_FIELDS = (
2
    "frames_indices",
3
    "fps",
4
    "duration",
5
    "total_num_frames",
6
)

1
- In-vocab fast path: when ``mm_token_ids`` is provided, the caller is
2
  declaring that mm placeholder IDs are real vocabulary entries. In that
3
  case the text path skips its ``index_select`` + ``torch.empty``
4
  + text scatter and embeds the full ``input_ids`` once, overwriting
5
  mm rows afterwards. This is the path taken by every VLM currently
6
  in tree EXCEPT Hyperclovax, which keeps the legacy ``vocab_size + 1``
7
  OOV remap in its input processor and therefore passes
8
  ``mm_token_ids=None`` so the OOV branch is taken — embedding an OOV
9
  id would be out-of-bounds.

1
@property
2
def multimodal_token_ids(self):
3
    return torch.tensor([self.image_token_id], dtype=torch.int32, device="cpu")

1
mm_mask[i]      True iff 位置 i 是任意 MM token（embed 槽位 或 special）
2
embed_mask[i]   True iff 位置 i 是 embed 槽位（排除 special）
3
special_mask[i] True iff 位置 i 是 special（未提供 mm_special_token_ids 时为 None）

1
input_ids=[1, 5, 5, 6, 5, 7, 2], mm_token_ids=[5], mm_special_token_ids=[6, 7]
2
→ embed_mask=[F, T, T, F, T, F, F]
3
→ mm_mask   =[F, T, T, T, T, T, F]

1
embedding_lengths: List[int] = []
2
offset = 0
3
for item_length in num_mm_tokens:
4
    item_positions = mm_positions[offset:offset + item_length]
5
    offset += item_length
6
    embedding_lengths.append(int(embed_mask[item_positions].sum().item()))

1
assert mm_positions.numel() == lengths_t.sum().item(), (
2
    f"Number of multimodal tokens ({mm_positions.numel()}) does not match "
3
    f"sum of per-unit lengths ({lengths_t.sum().item()}): "
4
    f"num_mm_tokens={num_mm_tokens}")

1
if mm_token_indices.shape[0] != mm_embed.shape[0]:
2
    raise ValueError(
3
        f"Multimodal token count mismatch: found {len(mm_token_indices)} image tokens in input_ids "
4
        f"but received {mm_embed.shape[0]} image embeddings.")

1
if features.shape[0] != rows * columns or length != rows * (columns + 1) + 2:
2
    raise ValueError("DeepSeek-V4.1 image grid does not match its placeholder span")

1
embeddings.append(
2
    torch.cat((
3
        self.image_start.to(features.dtype).unsqueeze(0),
4
        torch.cat((features, newline), dim=1).flatten(0, 1),
5
        self.image_end.to(features.dtype).unsqueeze(0),
6
    ))
7
)

1
for name in ("image_start", "image_end", "image_newline"):
2
    value = load_image_weight(name, (self.embedding_dim,), self.embedding_dtype)
3
    setattr(self, name, nn.Parameter(value))

1
Returns the total token count for each multimodal item, including any special tokens
2
(e.g., image_begin, image_end, image_break) that may be mixed with the actual
3
multimodal content tokens. This mm_token_lengths represents the full chunk from beginning
4
to end, not just pure image/video/audio tokens.

1
_MM_METADATA_ONLY_KEYS = frozenset({
2
    "mrope_config",
3
    "multimodal_embed_mask_cumsum",
4
    "multimodal_embedding_lengths",
5
    MULTIMODAL_ENCODER_ITEM_METADATA_KEY,
6
    "special_token_offsets",
7
    "layout_metadata",
8
})
9

10
def _has_mm_payload_keys(py_multimodal_data: Optional[dict]) -> bool:
11
    if not py_multimodal_data:
12
        return False
13
    return bool(set(py_multimodal_data.keys()) - _MM_METADATA_ONLY_KEYS)

1
Item scheduling bounds four distinct MM encoder resources. They are owned in
2
different places and measured in different units, so they are enumerated here once:
3

4
* (A) encoder batch cardinality -- ``encoder_batch_size``, counting atomic MM items
5
  rather than model-internal attention sequences.
6
* (B) encoder-forward workspace -- ``encoder_max_num_tokens``, in encoder attention
7
  tokens, clamped up to the largest atomic item; profiled by a direct full-budget
8
  encoder warmup.
9
* (C) resident output bytes -- ``output_budget_bytes``, the maximum post-encoder
10
  embeddings produced by one legal encoder iteration, converted to bytes. Enforced
11
  by the scheduler; any capacity not materialized by warmup is reserved in
12
  KV-capacity estimation.
13
* (D) reuse cache bytes -- ``encoder_cache_max_bytes`` may make the same
14
  ``TensorLRUCache`` larger than (C); it does not create a second pool.

1
def resolve_mm_encoder_token_budget(base_budget: int, model_max_atomic_item_tokens: int) -> int:
2
    """Keep the model's largest indivisible MM item schedulable."""
3
    return max(base_budget, model_max_atomic_item_tokens)

1
max_tokens_per_item = input_processor.get_mm_max_tokens_per_item()
2
model_max_atomic_item_tokens = max(max_tokens_per_item.values())
3
encoder_token_budget_base = encoder_max_num_tokens
4
effective_encoder_token_budget = resolve_mm_encoder_token_budget(
5
    encoder_token_budget_base, model_max_atomic_item_tokens
6
)
7
if effective_encoder_token_budget > encoder_max_num_tokens:
8
    logger.warning_once(...)
9
    encoder_max_num_tokens = effective_encoder_token_budget

1
Multimodal encoder token budget: configured=..., base=...,
2
effective=..., model_atomic_max=..., attention_capacity=...

1
attention_metadata_capacity = input_processor.get_mm_encoder_attention_metadata_capacity(
2
    encoder_max_num_tokens
3
)

1
Keys name this encoder's attention metadata objects, so they are
2
model-specific: this default declares a single `attention` object,
3
while a windowed encoder such as Qwen2.5-VL declares `full_attention`
4
and `window_attention`. There is no fixed superset to enumerate.

1
_ENCODER_FALLBACK_MAX_NUM_REQUESTS = 8192
2

3
def get_encoder_attention_metadata_capacity(self, max_num_tokens: int) -> dict[str, int]:
4
    return {"attention": max(max_num_tokens, _ENCODER_FALLBACK_MAX_NUM_REQUESTS)}

1
Legacy fallback for encoders that do not yet opt into atomic-item scheduling
2
or provide a model-specific item/token-to-segment mapping. Their encoder
3
forward is not runtime-budgeted, so configured item/token limits cannot
4
safely size fixed ``AttentionMetadata`` per-segment buffers.

1
# 1. 模型显式声明的 embedding_dim + embedding_dtype
2
embedding_dim = model.embedding_dim          # 可选契约
3
embedding_dtype = model.embedding_dtype      # 可选契约，缺省时用 model_config.torch_dtype
4
return embedding_dim * torch.empty((), dtype=embedding_dtype).element_size()
5

6
# 2. 文本 embedding 层的权重形状
7
weight = model.text_embedding_layer.weight
8
return weight.shape[-1] * weight.element_size()
9

10
# 3. 预训练配置里的 hidden_size（含 text_config.hidden_size 的嵌套情况）
11
hidden_size = getattr(pretrained, "hidden_size", None) or getattr(pretrained.text_config, "hidden_size", None)
12
return hidden_size * model.model_config.torch_dtype.itemsize
13

14
# 4. 都拿不到就报错，并且错误信息把三条路都列出来

1
def _encoder_cache_item_key(...):
2
    ...
3

4
def build_encoder_cache_item_keys(
5
    self, multimodal_hashes, item_refs, output_embedding_lengths, mm_processor_kwargs_hash,
6
):
7
    ...

1
A partially cached `MultimodalParams` is handled by encoding only its miss items
2
and interleaving cached items back in original per-item order. The default
3
`build_multimodal_encoder_input` handles stacked-on-dim-0 and packed-with-grid-thw
4
layouts; models with other layouts override that method.

1
for item_idx in item_indices:
2
    cache_key = state.item_cache_keys[item_idx]
3
    if cache_key is None:
4
        raise MultimodalEncoderRequestError(
5
            f"Scheduled MM item {item_idx} has no cache key",
6
            request_ids={request_id})
7
    scheduled_cache_keys.add(cache_key)
8
    encoder_items.append((multimodal_param, item_idx))
9
    output_targets.append((item_idx, cache_key, state.embedding_lengths[item_idx]))

1
for encoder_input, _, _ in encoder_inputs:
2
    encoder_input.to_device(
3
        "multimodal_data",
4
        "cuda",
5
        pin_memory=prefer_pinned(),
6
        target_keywords=getattr(self.model, "multimodal_data_device_paths", None),
7
    )

1
@property
2
def multimodal_data_device_paths(self) -> list[str]:
3
    return ["image.pixel_values", "multimodal_embedding"]

1
# Validate the entire batch before publishing any output. Otherwise a
2
# bad later item could leave earlier items READY and later requests
3
# admitted to prefill without committed outputs.
4
for output, (item_idx, cache_key, expected_rows) in zip(
5
    outputs, output_targets, strict=True
6
):
7
    if (not isinstance(output, torch.Tensor)
8
            or output.shape != (expected_rows, expected_width)
9
            or output.dtype != expected_dtype):
10
        ...
11
        raise MultimodalEncoderRequestError(...)

1
expected_width = self.bytes_per_embedding // expected_dtype.itemsize

1
for output, (_, cache_key, _) in zip(outputs, output_targets, strict=True):
2
    encoder_cache.commit(cache_key, output)
3

4
for request in requests:
5
    state = request.py_mm_encoder_state
6
    if state is None:
7
        continue
8
    state.mark_cache_keys_ready(scheduled_cache_keys)
9
    if state.progress is MultimodalEncoderProgress.READY:
10
        strip_mm_encoder_inputs(request.py_multimodal_data)

1
def strip_mm_encoder_inputs(mm_data: Dict[str, Any]) -> None:
2
    """Drop raw encoder payloads while preserving cached MM embeddings.
3

4
    Item-level encoder scheduling keeps the original request payload on CPU
5
    and transfers only selected items to GPU. Once every item is encoded, the
6
    raw modality dictionaries are no longer needed and must not be moved to
7
    GPU by the subsequent LLM input-preparation path.
8
    """
9
    for modality in ("image", "video", "audio"):
10
        mm_data.pop(modality, None)

1
# A batch-level error cannot be traced to one output, so fail only the
2
# scheduled producers. Requests that only share a scheduled entry keep
3
# their references, and the next scheduling pass picks a new producer.
4
producer_request_ids = scheduled_items.keys()
5
try:
6
    encoder_inputs = self.model.prepare_multimodal_encoder_inputs(encoder_items)
7
except MultimodalEncoderContractError as error:
8
    raise MultimodalEncoderRequestError(str(error), request_ids=producer_request_ids) from error

1
raise MultimodalEncoderRequestError(
2
    f"MM item {item_idx} must produce a {expected_dtype} tensor "
3
    f"with shape ({expected_rows}, {expected_width}), got {actual}",
4
    request_ids=requests_using_cache_keys({cache_key}),
5
)

1
Fuse text and multimodal embeddings. input_ids is [text_total_length + mm_total_length] and mm_embed is [mm_total_length, hidden_dim]. We just need to fuse them into [text_total_length + mm_total_length, hidden_dim] by slice-and-assign to the corresponding entries.

1
- Precedence: If kwargs provide indices (text_token_indices and mm_token_indices),
2
  those are used. If any one of them is not provided, fallback to filtering method.
3
  Sentinel-/OOV-based filtering (e.g., tokens >= vocab_size) is used only when
4
  neither index tensor and mm_token_ids is provided.

1
# NOTE: torch.where() enforces a host sync
2
text_token_indices = torch.where(~mm_token_mask)[0]
3
mm_token_indices = torch.where(mm_token_mask)[0]

1
- Sync-free contract: passing both ``text_token_indices`` and
2
  ``mm_token_indices`` skips the GPU ``torch.where`` host sync. The
3
  executor (``DecoderInputPreparer.prepare_inputs`` / ``NoKVCacheRunner.prepare_inputs``)
4
  precomputes them on a CPU ``input_ids`` copy via ``runners.prepare_multimodal_indices``
5
  (which uses ``filter_mm_token_from_input_ids`` against ``self.model.mm_token_ids``
6
  when present, else the OOV fallback ``>= vocab_size``) and ships them as
7
  pinned async H2D tensors in the inputs dict. VLM forwards are expected to
8
  forward these explicitly rather than relying on a ``**kwargs`` splat — if
9
  they don't, this function falls back to the host-syncing
10
  ``filter_mm_token_from_input_ids`` on GPU ``input_ids``.

1
def prepare_multimodal_indices(input_ids: list[int], *, model: Any) -> tuple[torch.Tensor, torch.Tensor]:
2
    input_ids = torch.tensor(input_ids, dtype=torch.int, device="cpu")
3
    vocab_size = model.config.vocab_size
4
    mm_token_ids = getattr(model, "multimodal_token_ids", None)
5
    if mm_token_ids is None:
6
        mm_token_ids = getattr(model, "mm_token_ids", None)
7
    text_token_indices, mm_token_indices = filter_mm_token_from_input_ids(
8
        input_ids, vocab_size=vocab_size, mm_token_ids=mm_token_ids)
9
    return text_token_indices, mm_token_indices

1
def ship_multimodal_indices(inputs, *, mm_token_indices_cpu, text_token_indices_cpu,
2
                            num_ctx_tokens, total_num_tokens) -> None:
3
    mm_token_indices_cpu = maybe_pin_memory(mm_token_indices_cpu)
4
    inputs["mm_token_indices"] = mm_token_indices_cpu.to("cuda", non_blocking=True)
5
    if total_num_tokens > num_ctx_tokens:
6
        extra_text = torch.arange(num_ctx_tokens, total_num_tokens, dtype=text_token_indices_cpu.dtype)
7
        text_token_indices_cpu = torch.cat([text_token_indices_cpu, extra_text])
8
    text_token_indices_cpu = maybe_pin_memory(text_token_indices_cpu)
9
    ...

1
Current speculative decode paths do not append multimodal placeholders after the context tokens.

1
input_embeds.index_copy_(
2
    0,
3
    _as_copy_index(mm_token_indices, input_embeds),
4
    mm_embed.to(dtype=input_embeds.dtype, device=input_embeds.device),
5
)

1
def _as_copy_index(index: torch.Tensor, destination: torch.Tensor) -> torch.Tensor:
2
    """Return ``index`` as the int64 index ``index_copy_`` needs on ``destination``.
3
    Explicit indices can come from callers with another dtype or device; for the
4
    executor's int64 device indices this is a no-op.
5
    """
6
    return index.to(device=destination.device, dtype=torch.long, non_blocking=True)

1
if mm_token_indices.shape[0] != mm_embed.shape[0]:
2
    raise ValueError(
3
        f"Multimodal token count mismatch: found {len(mm_token_indices)} image tokens in input_ids "
4
        f"but received {mm_embed.shape[0]} image embeddings.")

1
if extra_embeds is not None and len(extra_embeds) > 0:
2
    # only support single modality for deepstack features for now
3
    for i, extra_feature in enumerate(extra_embeds):
4
        extra_embed = torch.zeros(input_ids.shape[0], mm_embed.shape[-1],
5
                                  device=extra_feature.device, dtype=extra_feature.dtype)
6
        extra_embed.index_copy_(0, _as_copy_index(mm_token_indices, extra_embed), extra_feature)
7
        extra_embeds[i] = extra_embed

1
context_params = list(self.select_multimodal_params(multimodal_params or [], num_context_requests))
2
if not context_params:
3
    return PreparedLlmInputs(input_ids=input_ids, inputs_embeds=None)
4

5
embeddings = self._get_or_encode_multimodal_embeddings(context_params)
6
input_ids, embeddings = self.after_full_multimodal_embeddings(...)
7
active_embeddings = find_input_mm_embeds([embeddings], list(context_params))
8
active_embeddings, extra_embeds = self.after_active_multimodal_embeddings(...)
9
fused_input_ids, inputs_embeds, fused_extra_embeds = self._fuse_multimodal_embeddings(...)
10
return PreparedLlmInputs(input_ids=fused_input_ids, inputs_embeds=inputs_embeds,
11
                         extra_embeds=fused_extra_embeds)

1
def strip_mm_data_for_generation(mm_data: Dict[str, Any]) -> None:
2
    """Clear `mm_data` in place, retaining only `mrope_config.mrope_position_deltas`.
3
    ...
4
    """
5
    mrope_config = mm_data.get('mrope_config')
6
    mrope_deltas = None
7
    if isinstance(mrope_config, dict):
8
        mrope_deltas = mrope_config.get('mrope_position_deltas')
9
    mm_data.clear()
10
    if mrope_deltas is not None:
11
        mm_data['mrope_config'] = {'mrope_position_deltas': mrope_deltas}

1
[文本 120 token] [图像 576 行占位] [文本 80 token]

1
self.num_cached_mm_tokens = (int(cs[self.past_seen_token_num - 1])
2
                             if self.past_seen_token_num > 0 else 0)
3
self.num_mm_tokens_in_chunk = (int(cs[self.chunk_end_pos - 1]) - self.num_cached_mm_tokens
4
                               if self.chunk_end_pos > 0 else 0)
5
self.total_embeds_in_request = int(cs[-1])

1
Full-prefill, no-reuse iterations don't: `MultimodalRuntimeData` stays `None`
2
and `find_input_mm_embeds` handles the full payload.

1
if mm_token_ids is not None and vocab_size is not None:
2
    logger.debug("Both mm_token_ids and vocab_size are provided, using mm_token_ids and ignoring vocab_size")
3
if mm_token_ids is not None:
4
    embed_mask = torch.isin(input_ids, mm_token_ids)
5
else:
6
    embed_mask = input_ids >= vocab_size

1
if not (len(self.item_refs) == len(self.encoder_token_lengths) == len(self.output_embedding_lengths)):
2
    raise ValueError("Multimodal encoder item references and lengths must align")
3
if any(length <= 0 for length in self.encoder_token_lengths):
4
    raise ValueError("Multimodal encoder token lengths must be positive")
5
if any(length <= 0 for length in self.output_embedding_lengths):
6
    raise ValueError("Multimodal encoder output embedding lengths must be positive")

1
- Supports both individual batching (len(mm_embeds) == len(multimodal_params))
2
  and pre-concatenated batching (len(mm_embeds) == 1)
3
- Handles chunked prefill by considering chunk boundaries and current chunk tokens
4
- Example: if a request has 8 MM embed rows, 2 cached rows, and 3 rows
5
  in the current chunk, this keeps rows [2:5].

1
if len(mm_embeds) == 1:  # pre-concatenated; advance global cursor
2
    current_pos += runtime.total_embeds_in_request
3

4
if len(mm_embeds) == 1:
5
    sliced = [mm_embeds[0][start:end] for start, end in slices]
6
    return [_join_embeddings(sliced)]
7
return [mm_embeds[i][start:end] for i, (start, end) in enumerate(slices)]

1
if total_mm_tokens == sum(mm_embed.shape[0] for mm_embed in mm_embeds):
2
    # Item-scheduled batches were already sliced by the producer/gather
3
    # path. Do not apply the cached-prefix offset a second time.
4
    return mm_embeds

1
start = runtime.num_cached_mm_tokens if runtime is not None else 0
2
end = start + runtime.num_mm_tokens_in_chunk if runtime is not None else None
3
# Slice before joining: a long request must not copy all of its
4
# cached output for every small LLM chunk. One segment stays a view.
5
multimodal_data["multimodal_embedding"] = _join_embeddings(segments, start=start, end=end)
6
multimodal_data["multimodal_embedding_is_chunk"] = True

1
current_chunk_only = any(
2
    param.multimodal_data.get("multimodal_embedding_is_chunk", False)
3
    for param in multimodal_params)
4
...
5
if (current_chunk_only and runtime is not None
6
        and not param.multimodal_data.get("multimodal_embedding_is_chunk", False)):
7
    # Keep the full attached payload intact for later chunks/cache
8
    # writes; only this forward's gathered tensor is chunk-local.
9
    start = runtime.num_cached_mm_tokens
10
    end = start + runtime.num_mm_tokens_in_chunk
11
    if isinstance(embeds, list):
12
        embeds = _join_embeddings(embeds, start=start, end=end)
13
    else:
14
        embeds = embeds[start:end]

1
The runner is intentionally narrow: it does not know how to build attention metadata, what
2
model it is wrapping, or how to assemble the eager prelude (patch embedding, positional
3
embedding, etc.). It only owns the captured block loop, the static buffers that loop reads from,
4
and the mapping from real-input shapes to captured graph keys.
5

6
The runner's existence is itself the enable signal; callers should only construct one when CUDA
7
graph capture is wanted. Captures are taken at startup for every configured bucket, and the
8
captured set is fixed for the runner's lifetime.

1
class EncoderGraphKey(NamedTuple):
2
    """Internal key identifying an encoder graph workload size.
3

4
    * As a configured bucket, `total_tokens` is the maximum number of real (post-encode) tokens
5
      the bucket is meant to cover, and `num_contexts` is the number of real contexts.
6
      Neither includes any padding reservation; the runner adds that internally.
7
    * As a captured graph key, both fields reflect the padded state seen by the captured kernels.
8
    """
9
    total_tokens: int
10
    num_contexts: int

1
# Field order is (total_tokens, num_contexts), matching the desired bucket ordering for
2
# smallest-fit selection in `maybe_run()`.
3
self._buckets: List[EncoderGraphKey] = sorted(
4
    EncoderGraphKey(total_tokens=total_tokens, num_contexts=num_contexts)
5
    for total_tokens, num_contexts in config.buckets
6
)

1
# Single token kept aside for the dummy padding context when `enable_padding=True`.
2
# The attention backend rejects zero-length contexts, so the dummy must contain at least one token
3
# even when the workload exactly fills the bucket's real-token budget. This constant participates
4
# in both buffer sizing (`_padded_key_for_bucket`) and per-replay length assignment
5
# (`_padded_seq_lengths`, `_dummy_padded_seq_lengths`).
6
_PADDING_TOKEN_RESERVE: int = 1

1
* As a captured graph key, both fields reflect the padded state seen by the captured kernels.
2
  When `enable_padding=True` the key's `total_tokens` is `bucket.total_tokens + 1` and
3
  `num_contexts` is `bucket.num_contexts + 1`.

1
class _CapturedGraph(NamedTuple):
2
    ...
3
    # `data_ptr()` of each provider-declared graph-critical tensor on `metadata` after capture. The
4
    # runner asserts these stay stable across replays; the captured kernels bake these addresses in,
5
    # so a reallocation would silently read stale memory.
6
    # See `EncoderMetadataProvider.graph_critical_attrs` for more detail.
7
    metadata_tensor_ptrs: Mapping[str, int]

1
def _snapshot_metadata_tensor_ptrs(self) -> ...: ...
2
def _assert_metadata_buffers_stable(self) -> ...: ...

1
class _CapturedGraph(NamedTuple):
2
    ...
3
    # Each entry is a tensor view aliased to the captured output's memory via `make_weak_ref`; it
4
    # stays valid as long as the captured graph holds the underlying allocation alive.
5
    outputs: Dict[str, torch.Tensor]

1
`output_specs`: Token axis (`token_dim`) for each graph output, keyed by output name. Outputs
2
are not materialized by the runner - they are aliased from `encoder_fn`'s return - so
3
only the token axis is needed, to slice each output back to the real token extent.

1
`shape` describes the non-token dimensions only. The token-axis size is filled in from the
2
bucket and inserted at position `token_dim`. So for a ragged `(total_tokens, hidden_size)`
3
input, the spec is `EncoderGraphTensorSpec(shape=(hidden_size,), token_dim=0)`.

1
E（视觉编码） → P（预填充） → D（解码）

1
Use this on E/P prefill workers and cached-only paths. The encoder already ran somewhere else.

1
for param in multimodal_params:
2
    if param.encoder_event is not None:
3
        consumer_stream = torch.cuda.current_stream()
4
        consumer_stream.wait_event(param.encoder_event)
5
        embeds = param.multimodal_data.get("multimodal_embedding")
6
        ...
7
        for embed in embeds:
8
            if isinstance(embed, torch.Tensor) and embed.is_cuda:
9
                embed.record_stream(consumer_stream)

1
Register attached embeddings with the consumer stream as well: the request drops its Python
2
references after prefill, potentially before the asynchronous gather below has finished reading them.

1
class MultimodalEncoderRunner(NoKVCacheRunner):
2
    """Run an engine whose only job is multimodal encoding.
3

4
    This family is distinct from ``MultimodalItemScheduler``, which schedules vision items
5
    inside a decoder engine. Its graph phase is a no-op because
6
    ``MultimodalEncoderGraphRunner`` belongs to, and is owned by, the vision model layer.
7
    """

1
"multimodal_embedding": torch.Tensor | List[SharedTensor handle dict],
2
    # Pre-computed embeddings. In E/P handoff this may temporarily hold
3
    # SharedTensorContainer dicts; BaseWorker restores them to tensors with
4
    # to_tensor("multimodal_data") before PyTorch forward.

1
It is the minimum capacity of the unified encoder-output cache and is reserved during
2
KV-capacity estimation. Optional reuse may make the same cache larger. A request whose
3
total embedding exceeds this budget is rejected at admission.