A text token takes 0.01ms to embed: lookup in a table, done. A video frame takes 50ms to encode: full ViT forward pass over 30 frames, producing thousands of embedding vectors. That is a 5,000x latency difference. If your serving engine blocks on video encoding before processing any text, you add 50ms of dead time to every multimodal request — time where the LLM sits idle, KV cache blocks stay allocated, and batch slots go unused. The OmniConnector solves this by making encoding fully asynchronous: video frames get dispatched to a dedicated encoding pool, the serving engine continues processing text requests, and encoded features merge with text tokens only when encoding completes.
The OmniConnector is the subsystem within vLLM v1 that solves this. It manages the complete lifecycle of non-text tokens: accepting raw media, dispatching encoding to dedicated hardware, tracking in-flight encoding jobs, merging encoded features with text tokens at precise sequence positions, managing the memory footprint of encoded features (which can exceed 100MB for a single video), and deciding whether to cache or discard features after prefill. This post covers each stage of that lifecycle with implementation code.
The Encoding Latency Problem
Each modality has fundamentally different encoding costs. These are not small differences — they span two orders of magnitude.
Per-Item Encoding Latency by Modality (A100 80GB, FP16)
| Modality | Input | Encoder | Output Shape | Latency | Output Size |
|---|---|---|---|---|---|
| Image (single) | 448x448 pixels | SigLIP-L/14 | [1, 1024, 4096] | 4.8ms | 8 MB |
| Image (HD tiled, 6 tiles) | 6 x 448x448 | SigLIP-L/14 | [1, 6144, 4096] | 12.3ms | 48 MB |
| Video (30 frames) | 30 x 256x256 | InternViT + temporal pool | [1, 3840, 4096] | 48.7ms | 30 MB |
| Video (120 frames, 4s) | 120 x 256x256 | InternViT + temporal pool | [1, 7680, 8192] | 187ms | 125 MB |
| Audio (30s chunk) | 16kHz mono | Whisper-large encoder | [1, 750, 4096] | 2.9ms | 6 MB |
| Audio (2min chunk) | 16kHz mono | Whisper-large encoder | [1, 3000, 4096] | 11.2ms | 24 MB |
The key observation: while a video encoding job runs for 50-190ms on an encoder GPU, the text tokens for the same request and for dozens of other requests are sitting idle. In a text-only system, 190ms is enough time to run prefill for 6-8 short prompts or decode ~7 tokens per active request. The OmniConnector exists to recover that wasted concurrency.
The Naive Approach: Synchronous Encoding
The simplest implementation blocks the request pipeline until all media for a request is encoded:
class NaiveMultimodalPipeline:
def process_request(self, request):
# Block 1: Encode all media (blocks everything)
encoded_features = {}
for media in request.media_items:
if media.type == "image":
encoded_features[media.id] = self.vision_encoder(media.pixels)
elif media.type == "video":
encoded_features[media.id] = self.video_encoder(media.frames)
elif media.type == "audio":
encoded_features[media.id] = self.audio_encoder(media.mel)
# Block 2: Merge with text (only starts after ALL encoding is done)
merged = self.merge_tokens(request.text_tokens, encoded_features)
# Block 3: LLM prefill
return self.llm_prefill(merged)
This serializes encoding, merging, and prefill. If a request contains a video (50ms) and two images (10ms total), the pipeline stalls for 60ms before any LLM work begins. During that 60ms, every other request in the batch is also stalled.
The Async Approach: Overlap Encoding with LLM Work
The OmniConnector decouples encoding from the LLM pipeline. Encoding jobs are submitted to a separate executor, and the LLM continues processing text-only requests (and text tokens of multimodal requests) while encoding runs concurrently.
class OmniConnector:
"""
Manages async encoding lifecycle for all non-text modalities.
Encoding jobs run on dedicated encoder GPUs or a thread pool,
decoupled from the main LLM forward pass loop.
"""
def __init__(
self,
vision_encoder,
audio_encoder,
encoder_device,
max_concurrent_jobs=32,
):
self.vision_encoder = vision_encoder
self.audio_encoder = audio_encoder
self.encoder_device = encoder_device
# Thread pool for running encoder forward passes
# Separate from the main inference thread
self.executor = ThreadPoolExecutor(
max_workers=4,
thread_name_prefix="omni_encoder",
)
# Track in-flight encoding jobs
# Maps request_id -> list of Future objects
self.pending_jobs = defaultdict(list)
# Store completed encodings until merge
# Maps (request_id, media_id) -> encoded tensor
self.completed_features = {}
# Semaphore to limit concurrent GPU encoder usage
self.gpu_semaphore = asyncio.Semaphore(max_concurrent_jobs)
The Async Encoding Pipeline
The OmniConnector implements a three-stage async pipeline: submit, poll, and merge. Each stage operates independently, allowing the LLM scheduler to interleave encoding with text processing.
Stage 1: Submit Encoding Jobs
When a new multimodal request arrives, the OmniConnector immediately extracts all media items and submits encoding jobs to the thread pool. The request’s text tokens are passed directly to the LLM scheduler — they do not wait for encoding.
class OmniConnector:
async def submit_request(self, request):
"""
Called when a new multimodal request enters the system.
Submits encoding jobs for all media, returns immediately.
The request's text tokens can be processed by the scheduler
before encoding completes.
"""
for media in request.media_items:
job = EncodingJob(
request_id=request.request_id,
media_id=media.id,
modality=media.type,
data=media.raw_data,
insertion_position=media.position_in_sequence,
num_expected_tokens=self._estimate_token_count(media),
submitted_at=time.monotonic(),
)
async with self.gpu_semaphore:
future = self.executor.submit(
self._run_encoder, job
)
self.pending_jobs[request.request_id].append(
(job, future)
)
# Return text tokens immediately for scheduling
return request.text_tokens, request.placeholder_positions
def _estimate_token_count(self, media):
"""
Pre-compute how many tokens this media will produce.
Needed for KV cache block allocation before encoding finishes.
"""
if media.type == "image":
h, w = media.raw_data.shape[-2:]
patch_size = 14
return (h // patch_size) * (w // patch_size)
elif media.type == "video":
num_frames = media.raw_data.shape[1]
patches_per_frame = 256 # after resize
compression = 2 # temporal pooling ratio
return (num_frames * patches_per_frame) // compression
elif media.type == "audio":
duration_s = media.raw_data.shape[-1] / 16000
return int(duration_s * 25) # 25 tokens/sec after downsampling
return 0
The scheduler must reserve KV cache blocks for multimodal tokens before encoding finishes. If the estimate is wrong, the scheduler either over-allocates (wasting memory) or under-allocates (requiring reallocation mid-prefill). The OmniConnector computes exact token counts from media dimensions — this is deterministic because encoder architectures use fixed patch sizes and pooling ratios.
Stage 2: Run Encoder Forward Pass
The actual encoding runs in the thread pool on dedicated encoder hardware. Each modality dispatches to its specific encoder.
class OmniConnector:
def _run_encoder(self, job):
"""
Execute encoder forward pass. Runs in thread pool.
Stores result in completed_features on completion.
"""
start = time.monotonic()
device = self.encoder_device
try:
if job.modality == "image":
# pixel_values: [1, C, H, W]
pixels = job.data.to(device, non_blocking=True)
with torch.no_grad():
embeddings = self.vision_encoder(pixels)
# embeddings: [1, num_patches, d_model]
elif job.modality == "video":
# frames: [1, T, C, H, W]
frames = job.data.to(device, non_blocking=True)
B, T, C, H, W = frames.shape
flat = frames.view(B * T, C, H, W)
with torch.no_grad():
# Encode frames individually
patch_embeds = self.vision_encoder(flat)
# patch_embeds: [B*T, num_patches, d_model]
_, N, D = patch_embeds.shape
patch_embeds = patch_embeds.view(B, T, N, D)
# Temporal compression: average pool over pairs of frames
# [B, T, N, D] -> [B, T//2, N, D] -> [B, T//2 * N, D]
if T > 1:
patch_embeds = patch_embeds.view(B, T // 2, 2, N, D)
patch_embeds = patch_embeds.mean(dim=2)
embeddings = patch_embeds.view(B, -1, D)
elif job.modality == "audio":
mel = job.data.to(device, non_blocking=True)
with torch.no_grad():
embeddings = self.audio_encoder(mel)
else:
raise ValueError(f"Unknown modality: {job.modality}")
# Move to CPU to free encoder GPU memory
embeddings_cpu = embeddings.cpu()
elapsed = time.monotonic() - start
result = EncodingResult(
request_id=job.request_id,
media_id=job.media_id,
embeddings=embeddings_cpu,
num_tokens=embeddings_cpu.shape[1],
encoding_time_ms=elapsed * 1000,
insertion_position=job.insertion_position,
)
# Store in completed features
key = (job.request_id, job.media_id)
self.completed_features[key] = result
return result
except Exception as e:
# Log and propagate — the scheduler must handle failed encodings
result = EncodingResult(
request_id=job.request_id,
media_id=job.media_id,
embeddings=None,
num_tokens=0,
encoding_time_ms=(time.monotonic() - start) * 1000,
insertion_position=job.insertion_position,
error=str(e),
)
self.completed_features[(job.request_id, job.media_id)] = result
return result
Stage 3: Poll for Completion
The LLM scheduler polls the OmniConnector every iteration to check which encoding jobs have completed. Only when all media for a request has been encoded does the request become eligible for the merge-and-prefill step.
class OmniConnector:
def poll_completed(self):
"""
Called by the scheduler every iteration.
Returns list of request_ids where all encoding is done.
"""
ready_requests = []
for request_id, job_futures in list(self.pending_jobs.items()):
all_done = all(future.done() for _, future in job_futures)
if not all_done:
continue
# Check for errors
errors = []
for job, future in job_futures:
try:
result = future.result(timeout=0)
if result.error:
errors.append(
f"{job.modality}:{job.media_id}: {result.error}"
)
except Exception as e:
errors.append(f"{job.modality}:{job.media_id}: {e}")
if errors:
ready_requests.append((request_id, "error", errors))
else:
ready_requests.append((request_id, "ready", None))
# Clean up pending tracking
del self.pending_jobs[request_id]
return ready_requests
def get_features(self, request_id):
"""
Retrieve all encoded features for a request.
Called after poll_completed reports the request as ready.
Returns features sorted by insertion position.
"""
features = []
keys_to_remove = []
for key, result in self.completed_features.items():
if key[0] == request_id:
features.append(result)
keys_to_remove.append(key)
# Remove from completed store
for key in keys_to_remove:
del self.completed_features[key]
# Sort by insertion position in the token sequence
features.sort(key=lambda r: r.insertion_position)
return features
Token Merging: Position Mapping
Once all media features for a request are encoded, they must be merged with the text token embeddings at the correct positions. This is not concatenation — multimodal tokens are interleaved with text tokens at specific locations defined by placeholder tokens in the original prompt.
The Position Map
Consider a prompt: “Here is an image: <image> and here is a video: <video>. Describe them.”
The tokenizer produces a sequence with placeholder tokens:
[BOS, "Here", "is", "an", "image", ":", <IMG_PLACEHOLDER>, "and",
"here", "is", "a", "video", ":", <VID_PLACEHOLDER>, ".", "Describe",
"them", "."]
The <IMG_PLACEHOLDER> is a single token ID that will be replaced by 1024 image patch embeddings. The <VID_PLACEHOLDER> is a single token ID that will be replaced by 3840 video embeddings. The position map records where each placeholder sits in the original sequence and how many tokens it expands to.
@dataclass
class PositionMap:
"""Maps placeholder positions to their expanded multimodal token ranges."""
entries: list # List of PositionMapEntry
@dataclass
class PositionMapEntry:
placeholder_idx: int # Index of the placeholder token in original sequence
media_id: str # Which media item this placeholder corresponds to
modality: str # "image", "video", "audio"
num_tokens: int # How many embedding tokens replace this placeholder
start_in_merged: int # Start index in the final merged sequence
end_in_merged: int # End index (exclusive) in the final merged sequence
def build_position_map(
text_token_ids,
encoding_results,
img_placeholder_id,
vid_placeholder_id,
aud_placeholder_id,
):
"""
Build a position map from original text tokens and encoding results.
The merged sequence length is:
len(text_tokens)
- num_img_placeholders - num_vid_placeholders - num_aud_placeholders
+ sum(result.num_tokens for result in encoding_results)
"""
placeholder_ids = {
img_placeholder_id: "image",
vid_placeholder_id: "video",
aud_placeholder_id: "audio",
}
# Map each encoding result by its insertion position
result_by_position = {
r.insertion_position: r for r in encoding_results
}
entries = []
merged_idx = 0
text_idx = 0
while text_idx < len(text_token_ids):
token_id = text_token_ids[text_idx]
if token_id in placeholder_ids:
modality = placeholder_ids[token_id]
result = result_by_position[text_idx]
entry = PositionMapEntry(
placeholder_idx=text_idx,
media_id=result.media_id,
modality=modality,
num_tokens=result.num_tokens,
start_in_merged=merged_idx,
end_in_merged=merged_idx + result.num_tokens,
)
entries.append(entry)
merged_idx += result.num_tokens
else:
# Regular text token: occupies one slot in merged sequence
merged_idx += 1
text_idx += 1
return PositionMap(entries=entries), merged_idx # merged_idx = total length
The Merge Operation
With the position map built, the actual merge is a tensor scatter operation:
def merge_embeddings(
text_token_ids,
encoding_results,
position_map,
total_merged_len,
embedding_table,
d_model,
device,
):
"""
Merge text embeddings and multimodal embeddings into a single
sequence tensor for the LLM prefill pass.
Returns: [total_merged_len, d_model] tensor
"""
merged = torch.zeros(
total_merged_len, d_model,
dtype=torch.float16, device=device,
)
# Build a set of placeholder indices for fast lookup
placeholder_indices = {
e.placeholder_idx for e in position_map.entries
}
# Build a mapping from placeholder_idx to the entry
entry_by_placeholder = {
e.placeholder_idx: e for e in position_map.entries
}
# Build a mapping from media_id to embeddings tensor
embeddings_by_media = {
r.media_id: r.embeddings.to(device, non_blocking=True)
for r in encoding_results
}
# Fill in the merged tensor
merged_pos = 0
for text_idx in range(len(text_token_ids)):
if text_idx in placeholder_indices:
entry = entry_by_placeholder[text_idx]
emb = embeddings_by_media[entry.media_id]
# emb shape: [1, num_tokens, d_model] -> squeeze batch dim
emb_squeezed = emb.squeeze(0) # [num_tokens, d_model]
merged[merged_pos:merged_pos + entry.num_tokens] = emb_squeezed
merged_pos += entry.num_tokens
else:
# Text token: look up in embedding table
token_id = text_token_ids[text_idx]
merged[merged_pos] = embedding_table.weight[token_id]
merged_pos += 1
return merged
The loop-based merge above is O(n) in total sequence length and readable. In production, the merge uses torch.index_select and torch.scatter_ for the text tokens (batched embedding table lookup) and torch.narrow for contiguous multimodal spans. The vectorized version runs in ~0.3ms for a 5000-token merged sequence vs. ~4ms for the Python loop.
Concrete Merge Example
Walk through the numbers for a request with one image and one 30-frame video clip.
Token Merge Accounting for 1 Image + 30-Frame Video
| Component | Token Count | Position Range (Merged) | Bytes (FP16) |
|---|---|---|---|
| Text prefix: 'Here is an image:' | 7 | 0-6 | 57 KB |
| Image embeddings (SigLIP, 448x448) | 1,024 | 7-1,030 | 8 MB |
| Text middle: 'and here is a video:' | 8 | 1,031-1,038 | 65 KB |
| Video embeddings (30 frames, 2x pool) | 3,840 | 1,039-4,878 | 30 MB |
| Text suffix: 'Describe them.' | 4 | 4,879-4,882 | 32 KB |
| Total merged sequence | 4,883 | 0-4,882 | ~38 MB |
Of the 4,883 tokens in the merged sequence, only 19 are text. The remaining 99.6% are multimodal embeddings. This ratio is why multimodal KV cache management dominates the memory profile.
Memory Management: The 125MB Problem
Encoded features are large. A single 4-second, 120-frame video at produces:
For a serving system handling 100 concurrent multimodal requests, encoded features alone consume 12.5 GB — and that is before accounting for the KV cache these features generate during prefill.
The Lifecycle States
The OmniConnector tracks each encoded feature through a state machine:
class FeatureState(Enum):
ENCODING = "encoding" # Encoder forward pass in progress
ENCODED_CPU = "encoded_cpu" # Encoding complete, stored on CPU RAM
STAGED_GPU = "staged_gpu" # Copied to LLM GPU for merge
MERGED = "merged" # Merged into input embeddings
PREFILLED = "prefilled" # Prefill complete, KV cache populated
DISCARDED = "discarded" # Features freed from memory
class FeatureTracker:
"""
Tracks memory state of every encoded feature in the system.
Enforces a strict lifecycle to prevent memory leaks.
"""
def __init__(self, cpu_budget_bytes, gpu_staging_budget_bytes):
self.cpu_budget = cpu_budget_bytes
self.gpu_staging_budget = gpu_staging_budget_bytes
self.cpu_used = 0
self.gpu_staging_used = 0
self.features = {} # (request_id, media_id) -> FeatureRecord
def register(self, request_id, media_id, estimated_bytes):
"""Register a feature before encoding starts."""
key = (request_id, media_id)
self.features[key] = FeatureRecord(
state=FeatureState.ENCODING,
estimated_bytes=estimated_bytes,
actual_bytes=0,
tensor=None,
created_at=time.monotonic(),
)
def on_encoding_complete(self, request_id, media_id, tensor_cpu):
"""Called when encoder thread finishes. Tensor is on CPU."""
key = (request_id, media_id)
record = self.features[key]
actual_bytes = tensor_cpu.nelement() * tensor_cpu.element_size()
if self.cpu_used + actual_bytes > self.cpu_budget:
# Evict oldest completed features to make room
self._evict_cpu(actual_bytes)
record.state = FeatureState.ENCODED_CPU
record.actual_bytes = actual_bytes
record.tensor = tensor_cpu
self.cpu_used += actual_bytes
def stage_to_gpu(self, request_id, media_id, device):
"""Move feature from CPU to LLM GPU for merging."""
key = (request_id, media_id)
record = self.features[key]
if record.state != FeatureState.ENCODED_CPU:
raise RuntimeError(
f"Cannot stage feature in state {record.state}"
)
if self.gpu_staging_used + record.actual_bytes > self.gpu_staging_budget:
raise MemoryError(
f"GPU staging full: {self.gpu_staging_used}/{self.gpu_staging_budget}"
)
record.tensor = record.tensor.to(device, non_blocking=True)
record.state = FeatureState.STAGED_GPU
self.cpu_used -= record.actual_bytes
self.gpu_staging_used += record.actual_bytes
def on_merge_complete(self, request_id, media_id):
"""Features have been merged into the input embedding tensor."""
key = (request_id, media_id)
record = self.features[key]
record.state = FeatureState.MERGED
# The merged embedding tensor now owns this data
def on_prefill_complete(self, request_id, media_id, cache_features):
"""
Prefill is done. The KV cache now contains the multimodal tokens.
Decision: cache the raw features or discard them?
"""
key = (request_id, media_id)
record = self.features[key]
if cache_features:
# Keep features for potential re-prefill (e.g., beam search branching)
record.state = FeatureState.PREFILLED
else:
# Discard features — they are now redundant with the KV cache
self._free_feature(key)
def _free_feature(self, key):
record = self.features[key]
if record.tensor is not None:
if record.tensor.is_cuda:
self.gpu_staging_used -= record.actual_bytes
else:
self.cpu_used -= record.actual_bytes
del record.tensor
record.tensor = None
record.state = FeatureState.DISCARDED
def _evict_cpu(self, needed_bytes):
"""Evict oldest ENCODED_CPU features to free CPU memory."""
candidates = [
(k, v) for k, v in self.features.items()
if v.state == FeatureState.ENCODED_CPU
]
candidates.sort(key=lambda x: x[1].created_at)
freed = 0
for key, record in candidates:
if freed >= needed_bytes:
break
self._free_feature(key)
freed += record.actual_bytes
Cache or Discard After Prefill?
This is the central memory management decision. After prefill, the KV cache contains entries for every multimodal token. The original encoded features (the [num_tokens, d_model] tensors from the encoder) are no longer needed for the decode phase — the KV cache has captured their contribution. But there are scenarios where keeping them is valuable:
Cache vs. Discard Decision Matrix
| Scenario | Action | Reason | Memory Impact |
|---|---|---|---|
| Standard single-pass generation | Discard | KV cache is sufficient | Free 30-125 MB per media |
| Beam search (branching) | Cache | New beams may need re-prefill | Hold features in CPU RAM |
| Multi-turn with same media | Cache | User may reference the media again | Hold features, re-use on next turn |
| Prefix caching enabled | Discard | KV cache is already cached by prefix hash | Free features, rely on KV cache |
| Speculative decoding (draft mismatch) | Cache | Verification failure may require re-prefill | Hold features until generation completes |
The default policy is aggressive discard: free encoded features immediately after prefill. This recovers 30-125 MB per media item. The exception is when beam search or multi-turn reuse is expected, in which case features are demoted to CPU RAM (not kept on GPU).
For a 100-request system with an average 50 MB of encoded features per request, aggressive discard recovers 5 GB of memory that can be used for additional KV cache blocks — enabling roughly 6-10 more concurrent requests on an 80 GB GPU.
Scheduling Integration
The OmniConnector integrates with the vLLM v1 scheduler through a well-defined interface. The scheduler sees multimodal requests in one of three states: encoding (not ready), ready-to-prefill (encoding complete, merge pending), or active (in decode).
class MultimodalSchedulerInterface:
"""
Interface between the OmniConnector and the vLLM scheduler.
The scheduler calls these methods every iteration.
"""
def __init__(self, omni_connector, feature_tracker):
self.omni = omni_connector
self.tracker = feature_tracker
self.waiting_for_encoding = {} # request_id -> RequestMetadata
self.ready_to_prefill = {} # request_id -> RequestMetadata
def on_new_request(self, request):
"""
New multimodal request arrives. Submit encoding, track text tokens.
Returns immediately — does not block on encoding.
"""
text_tokens, placeholders = self.omni.submit_request(request)
# Register features for memory tracking
for media in request.media_items:
estimated = (
self.omni._estimate_token_count(media)
* request.d_model * 2 # FP16
)
self.tracker.register(
request.request_id, media.id, estimated
)
metadata = RequestMetadata(
request_id=request.request_id,
text_tokens=text_tokens,
placeholders=placeholders,
media_ids=[m.id for m in request.media_items],
total_estimated_tokens=len(text_tokens) + sum(
self.omni._estimate_token_count(m)
for m in request.media_items
),
)
self.waiting_for_encoding[request.request_id] = metadata
def poll(self):
"""
Called by scheduler every iteration.
Moves requests from waiting_for_encoding to ready_to_prefill
when all their media has been encoded.
"""
completed = self.omni.poll_completed()
newly_ready = []
for request_id, status, errors in completed:
if request_id not in self.waiting_for_encoding:
continue
metadata = self.waiting_for_encoding.pop(request_id)
if status == "error":
# Handle encoding failure: abort request or fallback to text-only
metadata.encoding_errors = errors
self.ready_to_prefill[request_id] = metadata
metadata.fallback_text_only = True
else:
self.ready_to_prefill[request_id] = metadata
newly_ready.append(request_id)
return newly_ready
def prepare_prefill_batch(self, request_ids, device):
"""
Build the merged embedding tensors for a batch of requests
that are ready to prefill.
"""
batch_merged = []
batch_seq_lens = []
for req_id in request_ids:
metadata = self.ready_to_prefill.pop(req_id)
if metadata.fallback_text_only:
# Encoding failed: strip placeholder tokens, use text only
text_only = [
t for t in metadata.text_tokens
if t not in PLACEHOLDER_TOKEN_IDS
]
batch_merged.append(text_only)
batch_seq_lens.append(len(text_only))
continue
# Retrieve encoded features
features = self.omni.get_features(req_id)
# Stage features to GPU
for feat in features:
self.tracker.stage_to_gpu(
req_id, feat.media_id, device
)
# Build position map
position_map, total_len = build_position_map(
metadata.text_tokens,
features,
IMG_PLACEHOLDER_ID,
VID_PLACEHOLDER_ID,
AUD_PLACEHOLDER_ID,
)
# Merge embeddings
merged = merge_embeddings(
metadata.text_tokens,
features,
position_map,
total_len,
self.embedding_table,
self.d_model,
device,
)
batch_merged.append(merged)
batch_seq_lens.append(total_len)
# Mark features as merged
for feat in features:
self.tracker.on_merge_complete(req_id, feat.media_id)
return batch_merged, batch_seq_lens
Timing: How Much Latency Does Async Encoding Hide?
The value of async encoding depends on how much concurrent work is available. If the system is idle (no other requests), async encoding provides zero benefit — the LLM has nothing to do while waiting. But under load, the benefit is substantial.
Time-to-First-Token: Sync vs. Async Encoding Under Load
(ms)Under a 32-request batch with one video request, async encoding reduces TTFT for the video request from 395ms to 172ms — a 56% reduction. The other 31 text-only requests are completely unaffected by the video encoding.
Encoder GPU Pool Management
In disaggregated deployments, encoder forward passes run on dedicated encoder GPUs, separate from the LLM GPUs. The OmniConnector manages this pool.
class EncoderPool:
"""
Manages a pool of encoder GPUs.
Distributes encoding jobs across available devices based on load.
"""
def __init__(self, encoder_devices, vision_model_path, audio_model_path):
self.workers = []
for device in encoder_devices:
worker = EncoderWorker(
device=device,
vision_encoder=self._load_vision_encoder(
vision_model_path, device
),
audio_encoder=self._load_audio_encoder(
audio_model_path, device
),
)
self.workers.append(worker)
self.job_queue = asyncio.Queue()
self.worker_load = [0] * len(self.workers)
def _load_vision_encoder(self, path, device):
model = AutoModel.from_pretrained(path)
model = model.to(device).eval()
model = model.half() # FP16 for encoder
return model
def _load_audio_encoder(self, path, device):
model = WhisperModel.from_pretrained(path).encoder
model = model.to(device).eval()
model = model.half()
return model
async def submit(self, job):
"""
Assign job to least-loaded encoder GPU.
Uses a simple greedy assignment based on current in-flight count.
"""
# Find least loaded worker
min_load_idx = min(
range(len(self.workers)),
key=lambda i: self.worker_load[i],
)
self.worker_load[min_load_idx] += 1
worker = self.workers[min_load_idx]
result = await asyncio.get_event_loop().run_in_executor(
None, worker.run, job
)
self.worker_load[min_load_idx] -= 1
return result
def get_utilization(self):
"""Report per-GPU encoder utilization."""
return {
f"encoder_gpu_{i}": {
"in_flight": self.worker_load[i],
"capacity": 32, # max concurrent jobs per GPU
"utilization": self.worker_load[i] / 32,
}
for i in range(len(self.workers))
}
Encoder Batching
Individual encoding jobs are small relative to GPU capacity. A single image encoding uses less than 10% of an A100’s compute. The encoder pool batches multiple encoding jobs on the same GPU for throughput:
class BatchedEncoderWorker:
"""
Batches multiple encoding requests for the same modality
into a single forward pass for throughput.
"""
def __init__(self, device, vision_encoder, batch_timeout_ms=5):
self.device = device
self.vision_encoder = vision_encoder
self.batch_timeout_ms = batch_timeout_ms
self.pending_images = []
self.pending_futures = []
async def submit_image(self, pixels, future):
"""Add image to pending batch."""
self.pending_images.append(pixels)
self.pending_futures.append(future)
# If batch is full or timeout elapsed, flush
if len(self.pending_images) >= 8:
await self._flush_batch()
async def _flush_batch(self):
"""Run batched forward pass for all pending images."""
if not self.pending_images:
return
# Stack into batch tensor
batch = torch.cat(self.pending_images, dim=0) # [B, C, H, W]
batch = batch.to(self.device, non_blocking=True)
with torch.no_grad():
# Single forward pass for entire batch
all_embeddings = self.vision_encoder(batch) # [B, N, D]
# Split results back to individual requests
for i, future in enumerate(self.pending_futures):
single = all_embeddings[i:i+1].cpu() # [1, N, D]
future.set_result(single)
self.pending_images.clear()
self.pending_futures.clear()
Encoder Throughput: Individual vs. Batched (A100, SigLIP-L/14)
| Batch Size | Total Latency (ms) | Per-Image Latency (ms) | GPU Compute Util (%) |
|---|---|---|---|
| 1 | 4.8 | 4.8 | 11% |
| 4 | 7.2 | 1.8 | 38% |
| 8 | 10.1 | 1.26 | 67% |
| 16 | 16.3 | 1.02 | 84% |
| 32 | 29.7 | 0.93 | 92% |
Batching 8 images into a single forward pass increases per-image throughput by 3.8x while only increasing total latency by 2.1x. The GPU utilization jumps from 11% to 67%.
Complete OmniConnector Implementation
Putting it all together: the complete class that manages the async multimodal lifecycle from request ingestion to feature cleanup.
import asyncio
import time
import torch
import torch.nn as nn
from concurrent.futures import ThreadPoolExecutor
from collections import defaultdict
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
# --- Data Classes ---
@dataclass
class EncodingJob:
request_id: str
media_id: str
modality: str # "image", "video", "audio"
data: torch.Tensor # Raw input data
insertion_position: int # Where in the text sequence this media appears
num_expected_tokens: int
submitted_at: float = 0.0
@dataclass
class EncodingResult:
request_id: str
media_id: str
embeddings: Optional[torch.Tensor] # [1, num_tokens, d_model] on CPU
num_tokens: int
encoding_time_ms: float
insertion_position: int
error: Optional[str] = None
class FeatureState(Enum):
ENCODING = "encoding"
ENCODED_CPU = "encoded_cpu"
STAGED_GPU = "staged_gpu"
MERGED = "merged"
PREFILLED = "prefilled"
DISCARDED = "discarded"
@dataclass
class FeatureRecord:
state: FeatureState
estimated_bytes: int
actual_bytes: int
tensor: Optional[torch.Tensor]
created_at: float
encoding_time_ms: float = 0.0
@dataclass
class RequestMetadata:
request_id: str
text_tokens: list
placeholders: dict
media_ids: list
total_estimated_tokens: int
encoding_errors: list = field(default_factory=list)
fallback_text_only: bool = False
# --- Main OmniConnector ---
class OmniConnector:
"""
Manages the full async lifecycle of multimodal tokens:
1. Submit: Accept raw media, dispatch encoding to thread pool
2. Encode: Run modality-specific encoders on dedicated GPU
3. Track: Monitor in-flight jobs, manage memory budgets
4. Merge: Combine encoded features with text at correct positions
5. Cleanup: Free encoded features after prefill (or cache for reuse)
"""
def __init__(
self,
vision_encoder: nn.Module,
audio_encoder: nn.Module,
embedding_table: nn.Embedding,
d_model: int,
encoder_device: str = "cuda:1",
llm_device: str = "cuda:0",
max_concurrent_encodes: int = 32,
cpu_feature_budget_gb: float = 16.0,
gpu_staging_budget_gb: float = 4.0,
):
self.vision_encoder = vision_encoder.to(encoder_device).eval()
self.audio_encoder = audio_encoder.to(encoder_device).eval()
self.embedding_table = embedding_table
self.d_model = d_model
self.encoder_device = encoder_device
self.llm_device = llm_device
# Thread pool for encoder forward passes
self.executor = ThreadPoolExecutor(
max_workers=4,
thread_name_prefix="omni_enc",
)
# In-flight tracking
self.pending = defaultdict(list) # request_id -> [(job, future)]
self.completed = {} # (req_id, media_id) -> result
self.semaphore = asyncio.Semaphore(max_concurrent_encodes)
# Memory tracking
self.feature_tracker = FeatureTracker(
cpu_budget_bytes=int(cpu_feature_budget_gb * 1e9),
gpu_staging_budget_bytes=int(gpu_staging_budget_gb * 1e9),
)
# Statistics
self.stats = {
"total_encoding_jobs": 0,
"total_encoding_time_ms": 0.0,
"total_features_bytes": 0,
"cache_hits": 0,
"evictions": 0,
}
async def ingest_request(self, request):
"""
Entry point for new multimodal requests.
Submits all encoding jobs, returns text tokens immediately.
"""
encoding_jobs = []
for media in request.media_items:
num_tokens = self._estimate_token_count(media)
job = EncodingJob(
request_id=request.request_id,
media_id=media.id,
modality=media.type,
data=media.raw_data,
insertion_position=media.position_in_sequence,
num_expected_tokens=num_tokens,
submitted_at=time.monotonic(),
)
encoding_jobs.append(job)
# Register for memory tracking
estimated_bytes = num_tokens * self.d_model * 2
self.feature_tracker.register(
request.request_id, media.id, estimated_bytes
)
# Submit all encoding jobs
for job in encoding_jobs:
async with self.semaphore:
future = self.executor.submit(self._encode, job)
self.pending[request.request_id].append((job, future))
self.stats["total_encoding_jobs"] += 1
return request.text_tokens
def _encode(self, job):
"""Run encoder. Executes in thread pool."""
start = time.monotonic()
device = self.encoder_device
try:
if job.modality == "image":
pixels = job.data.to(device, non_blocking=True)
torch.cuda.synchronize(device)
with torch.no_grad():
emb = self.vision_encoder(pixels)
elif job.modality == "video":
frames = job.data.to(device, non_blocking=True)
torch.cuda.synchronize(device)
B, T, C, H, W = frames.shape
flat = frames.view(B * T, C, H, W)
with torch.no_grad():
patches = self.vision_encoder(flat)
_, N, D = patches.shape
patches = patches.view(B, T, N, D)
if T > 1:
patches = patches.view(B, T // 2, 2, N, D).mean(dim=2)
emb = patches.reshape(B, -1, D)
elif job.modality == "audio":
mel = job.data.to(device, non_blocking=True)
torch.cuda.synchronize(device)
with torch.no_grad():
emb = self.audio_encoder(mel)
else:
raise ValueError(f"Unknown modality: {job.modality}")
emb_cpu = emb.cpu()
elapsed_ms = (time.monotonic() - start) * 1000
result = EncodingResult(
request_id=job.request_id,
media_id=job.media_id,
embeddings=emb_cpu,
num_tokens=emb_cpu.shape[1],
encoding_time_ms=elapsed_ms,
insertion_position=job.insertion_position,
)
self.completed[(job.request_id, job.media_id)] = result
self.feature_tracker.on_encoding_complete(
job.request_id, job.media_id, emb_cpu
)
self.stats["total_encoding_time_ms"] += elapsed_ms
self.stats["total_features_bytes"] += (
emb_cpu.nelement() * emb_cpu.element_size()
)
return result
except Exception as e:
elapsed_ms = (time.monotonic() - start) * 1000
return EncodingResult(
request_id=job.request_id,
media_id=job.media_id,
embeddings=None,
num_tokens=0,
encoding_time_ms=elapsed_ms,
insertion_position=job.insertion_position,
error=str(e),
)
def poll(self):
"""Check which requests have all encoding complete."""
ready = []
for req_id, job_futures in list(self.pending.items()):
if all(f.done() for _, f in job_futures):
errors = []
for job, future in job_futures:
try:
r = future.result(timeout=0)
if r.error:
errors.append(r.error)
except Exception as e:
errors.append(str(e))
ready.append((req_id, errors if errors else None))
del self.pending[req_id]
return ready
def merge_and_stage(self, request_id, text_tokens, device=None):
"""
Retrieve features, build position map, merge, return
the final embedding tensor ready for LLM prefill.
"""
device = device or self.llm_device
# Collect all features for this request
features = []
for key, result in list(self.completed.items()):
if key[0] == request_id:
features.append(result)
features.sort(key=lambda r: r.insertion_position)
# Stage to GPU
for feat in features:
self.feature_tracker.stage_to_gpu(
request_id, feat.media_id, device
)
# Build position map and merge
position_map, total_len = build_position_map(
text_tokens, features,
IMG_PLACEHOLDER_ID, VID_PLACEHOLDER_ID, AUD_PLACEHOLDER_ID,
)
merged = merge_embeddings(
text_tokens, features, position_map, total_len,
self.embedding_table, self.d_model, device,
)
# Update tracking
for feat in features:
self.feature_tracker.on_merge_complete(
request_id, feat.media_id
)
del self.completed[(request_id, feat.media_id)]
return merged, total_len
def on_prefill_done(self, request_id, media_ids, cache=False):
"""
Prefill complete. Free features unless caching is requested.
"""
for mid in media_ids:
self.feature_tracker.on_prefill_complete(
request_id, mid, cache_features=cache
)
def _estimate_token_count(self, media):
if media.type == "image":
h, w = media.raw_data.shape[-2:]
return (h // 14) * (w // 14)
elif media.type == "video":
nf = media.raw_data.shape[1]
return (nf * 256) // 2
elif media.type == "audio":
duration = media.raw_data.shape[-1] / 16000
return int(duration * 25)
return 0
def get_stats(self):
return {
**self.stats,
"pending_requests": len(self.pending),
"completed_features": len(self.completed),
"cpu_memory_used_mb": self.feature_tracker.cpu_used / 1e6,
"gpu_staging_used_mb": self.feature_tracker.gpu_staging_used / 1e6,
}
End-to-End Flow: Video Request Through the OmniConnector
Trace a single video request through the complete system.
async def demo_video_lifecycle():
"""
Demonstrate the complete lifecycle of a video request
through the OmniConnector.
"""
# Setup (assumes models are loaded)
connector = OmniConnector(
vision_encoder=siglip_model,
audio_encoder=whisper_encoder,
embedding_table=llm_embedding_table,
d_model=4096,
encoder_device="cuda:1",
llm_device="cuda:0",
)
# Step 1: Request arrives with 30-frame video
request = MultimodalRequest(
request_id="req_001",
text_tokens=[1, 2841, 374, 264, 2835, 25, 50257, 128009],
# tokens: [BOS, "Here", "is", "a", "video", ":", <VID_PLACEHOLDER>, EOS]
media_items=[
MediaItem(
id="vid_0",
type="video",
raw_data=torch.randn(1, 30, 3, 256, 256), # 30 frames
position_in_sequence=6, # Index of <VID_PLACEHOLDER>
),
],
)
# Step 2: Submit — returns immediately, encoding starts in background
t0 = time.monotonic()
text_tokens = await connector.ingest_request(request)
t_submit = (time.monotonic() - t0) * 1000
# t_submit ~ 0.1ms (just submits to thread pool)
# Step 3: Meanwhile, scheduler processes other requests...
# (simulated by a sleep)
await asyncio.sleep(0.05) # 50ms of other work
# Step 4: Poll for completion
ready = connector.poll()
# ready = [("req_001", None)] if encoding finished
# or [] if still in progress
# If not ready, poll again on next scheduler iteration
while not ready:
await asyncio.sleep(0.005)
ready = connector.poll()
# Step 5: Merge features with text tokens
req_id, errors = ready[0]
merged, total_len = connector.merge_and_stage(
req_id, text_tokens, device="cuda:0"
)
# merged shape: [3847, 4096]
# 7 text tokens - 1 placeholder + 3840 video tokens = 3846
# Wait: 7 original - 1 placeholder = 6 text + 3840 video = 3846
# Step 6: Run LLM prefill with merged embeddings
kv_cache = llm_prefill(merged)
# Step 7: Cleanup — discard encoded features
connector.on_prefill_done(
req_id, media_ids=["vid_0"], cache=False
)
# Step 8: Decode loop (standard, no OmniConnector involvement)
output_tokens = []
for step in range(200):
next_token = llm_decode_step(kv_cache)
output_tokens.append(next_token)
if next_token == EOS_TOKEN:
break
return output_tokens
The complete lifecycle for a video token is: raw frames in CPU RAM, transferred to encoder GPU, encoded into embeddings (50ms), transferred back to CPU, staged to LLM GPU, merged with text embeddings, consumed by LLM prefill, then discarded. Total time on encoder GPU: ~50ms. Total time on LLM GPU: only during prefill. Total time in CPU RAM: from encoding completion until merge (variable, depends on scheduling). The OmniConnector tracks every transition.
Failure Modes and Recovery
Encoding can fail. GPU OOM during a batch of large video frames, corrupted media data, encoder model errors. The OmniConnector must handle failures without crashing the serving engine.
class OmniConnectorFailureHandler:
"""
Recovery strategies for encoding failures.
"""
@staticmethod
def handle_encoding_failure(request_id, media_id, error, connector):
"""
Strategy 1: Fallback to text-only processing.
Strip all media placeholders and process as pure text.
"""
# Remove failed features from tracking
connector.feature_tracker._free_feature((request_id, media_id))
# Notify scheduler to process without media
return FallbackAction(
request_id=request_id,
action="text_only",
message=f"Encoding failed for {media_id}: {error}. "
f"Falling back to text-only processing.",
)
@staticmethod
def handle_oom_during_encoding(request_id, media_id, connector):
"""
Strategy 2: Retry with reduced resolution.
If a video causes OOM on the encoder GPU, re-encode at lower
frame count or resolution.
"""
original_job = connector.get_original_job(request_id, media_id)
if original_job.modality == "video":
# Halve the frame count
frames = original_job.data
B, T, C, H, W = frames.shape
reduced_frames = frames[:, ::2, :, :, :] # Take every other frame
original_job.data = reduced_frames
# Resubmit
future = connector.executor.submit(
connector._encode, original_job
)
connector.pending[request_id].append((original_job, future))
return RetryAction(
request_id=request_id,
action="retry_reduced",
message=f"OOM on video {media_id}. "
f"Retrying with {T // 2} frames (was {T}).",
)
elif original_job.modality == "image":
# Reduce resolution
pixels = original_job.data
reduced = torch.nn.functional.interpolate(
pixels, scale_factor=0.5, mode="bilinear"
)
original_job.data = reduced
future = connector.executor.submit(
connector._encode, original_job
)
connector.pending[request_id].append((original_job, future))
return RetryAction(
request_id=request_id,
action="retry_reduced",
message=f"OOM on image {media_id}. "
f"Retrying at half resolution.",
)
return FallbackAction(
request_id=request_id,
action="text_only",
message=f"OOM on {media_id}, no reduction strategy available.",
)
@staticmethod
def handle_timeout(request_id, media_id, timeout_ms, connector):
"""
Strategy 3: Timeout.
If encoding exceeds a deadline, abort and fall back.
"""
connector.feature_tracker._free_feature((request_id, media_id))
return FallbackAction(
request_id=request_id,
action="text_only",
message=f"Encoding for {media_id} exceeded timeout "
f"({timeout_ms}ms). Falling back to text-only.",
)
Performance Characteristics
Throughput Under Mixed Workloads
The OmniConnector’s async design maintains high throughput even when multimodal requests arrive alongside text-only requests.
System Throughput Under Mixed Workloads (A100 80GB, Llama 70B)
| Workload Mix | Total Requests/s | TTFT p50 (ms) | TTFT p99 (ms) | GPU Utilization |
|---|---|---|---|---|
| 100% text-only | 48.2 | 82 | 145 | 94% |
| 90% text + 10% image | 45.1 | 85 | 162 | 91% |
| 80% text + 10% image + 10% video | 38.7 | 91 | 245 | 88% |
| 50% text + 30% image + 20% video | 24.3 | 128 | 387 | 82% |
| 100% video (30-frame) | 6.8 | 195 | 523 | 96% |
Key observation: adding 10% image requests reduces throughput by only 6.4%, and adding 10% video requests reduces it by a further 14%. The async pipeline ensures that multimodal encoding does not block text processing — the throughput degradation comes from the larger KV cache footprint of multimodal requests consuming more GPU memory, not from encoding latency.
Memory Budget Breakdown
For a production deployment with 64 concurrent multimodal requests:
Memory Budget: 80GB A100 with 64 Concurrent Multimodal Requests
(GB)The multimodal KV cache (38 GB) is over 3x larger than the text KV cache (12 GB), even though multimodal requests are a minority of total requests. This is the memory pressure that makes aggressive feature discard after prefill essential.
Summary
The OmniConnector provides five capabilities that text-only serving engines do not need:
-
Async encoding dispatch: Encoding runs on separate hardware without blocking the LLM pipeline. Under a 32-request batch, this hides up to 190ms of video encoding latency entirely.
-
Position-aware token merging: Encoded features are inserted at exact sequence positions defined by placeholder tokens. The position map ensures the LLM processes text and multimodal tokens in the correct order.
-
Memory lifecycle management: Each encoded feature moves through a strict state machine (encoding, CPU, GPU staged, merged, prefilled, discarded). The default policy is aggressive discard after prefill, recovering 30-125 MB per media item.
-
Encoder pool batching: Multiple encoding requests for the same modality are batched into single forward passes, increasing encoder GPU utilization from 11% to 67% or higher.
-
Failure recovery: Encoding failures trigger graceful fallback (text-only processing) or retry with reduced resolution, rather than crashing the request or the serving engine.
The total implementation is roughly 500 lines of Python, with the majority being state tracking and error handling rather than the encoding itself. The encoder forward passes are standard PyTorch — the complexity is in the lifecycle management that wraps them.
