How AggregatorStream in LingBot-Map Handles Temporal Processing with KV Cache

AggregatorStream maintains causal temporal attention across video frames using a dual-backend KV cache system that supports both PyTorch SDPA and FlashInfer, with automatic sliding-window eviction and scale-frame retention.

The AggregatorStream class in the Robbyant/lingbot-map repository implements a streaming-capable causal aggregator designed for video understanding. It processes video frames incrementally while preserving temporal context through a sophisticated key-value (KV) cache mechanism that enforces causality and manages memory efficiently via sliding-window eviction policies.

Architecture Overview

AggregatorStream extends the base aggregator functionality to handle streaming video inputs frame-by-frame. Unlike batch processing, this implementation maintains state across forward passes using a KV cache that stores key and value tensors from previous frames.

The architecture supports two distinct backends:

• PyTorch SDPA: Uses a Python dictionary to store KV tensors for each transformer block
• FlashInfer: Employs a paged KV cache manager optimized for fast inference

Both backends enforce causal temporal attention, ensuring each frame can only attend to itself and previous frames, never future ones.

KV Cache Initialization

When instantiating AggregatorStream, the constructor (__init__ in lingbot_map/aggregator/stream.py, lines 34-94) stores temporal hyperparameters and selects the appropriate backend based on self.use_sdpa and self.use_flashinfer flags.

The _init_kv_cache() method (lines 79-98) initializes the storage mechanism:

# SDPA path: Dictionary-based cache

self.kv_cache[f"k_{i}"] = None   # lines 88-93

self.kv_cache[f"v_{i}"] = None

For the FlashInfer path, the cache manager is lazy-initialized via _get_flashinfer_manager() (lines 198-230) upon first use:

self.kv_cache_manager = FlashInferKVCacheManager(
    max_num_frames=kv_cache_scale_frames + kv_cache_sliding_window + 16,
    # ... additional parameters

)  # lines 206-229

Temporal Processing Pipeline

The core temporal logic resides in _process_causal_stream() (lines 512-539). When processing new frames, the method:

1. Reshapes input tokens from [B*S_local, P, C] to [B, S_local*P, C] (lines 556-558)
2. Computes global frame indices using self.total_frames_processed to track position across stream calls (lines 668-672)
3. Generates 3-D positional encodings via _get_3d_positions_streaming() when enabled (lines 666-694)

The forward pass iterates through global transformer blocks, passing the appropriate cache reference:

# SDPA path (lines 887-925)

output = self.global_blocks[global_idx](
    tokens, 
    kv_cache=self.kv_cache,
    num_frames=S_global
)

# FlashInfer path (lines 928-954)

kv_manager = self._get_flashinfer_manager()
output = self.global_blocks[global_idx](
    tokens,
    kv_cache=kv_manager,
    num_frames=S_global
)

Each block reads the cached keys/values from previous frames, computes attention limited to the current and past frames, and writes updated KV tensors back to the cache.

Sliding-Window Management and Eviction

To prevent unbounded memory growth, the KV cache implements sliding-window eviction configured by kv_cache_sliding_window. The FlashInferKVCacheManager automatically drops the oldest frames once the window limit is exceeded.

Scale-frame handling ensures high-priority tokens persist longer. The manager reserves space for kv_cache_scale_frames beyond the sliding window, keeping critical contextual information available across longer temporal distances.

After processing each block group, the stream updates its temporal position:

self.total_frames_processed += S_global  # lines 1026-1027

This counter drives the correct frame indices for subsequent calls, ensuring accurate positional encoding and causal masking.

3-D Rotary Positional Encoding

When enable_3d_rope=True, the aggregator initializes temporal RoPE via _init_3d_rope(). During streaming, _get_3d_positions_streaming() generates position tensors that span the temporal dimension, allowing the model to understand both spatial and temporal relationships in video sequences.

Practical Implementation Example

import torch
from lingbot_map.aggregator.stream import AggregatorStream

# Configure the stream aggregator

stream = AggregatorStream(
    sliding_window_size=32,          # Attention limited to last 32 frames

    num_frame_for_scale=8,           # Retain 8 scale frames in cache

    kv_cache_sliding_window=64,      # Eviction window size

    enable_3d_rope=True,            # Enable temporal positional encoding

    max_frame_num=1024,              # Maximum absolute frame index

    use_flashinfer=True,             # Use FlashInfer backend (recommended)

    depth=12, 
    embed_dim=1024,
    num_heads=16, 
    mlp_ratio=4.0,
)

# Prepare streaming input: batch of 1 frame

B, S_local = 1, 1
P = (16 // stream.patch_size) ** 2 + stream.num_special_tokens
C = stream.embed_dim
tokens = torch.randn(B * S_local, P, C)  # Shape: [1, P, 1024]

# Process frame - KV cache updates automatically

output = stream(tokens)
print(output.shape)  # torch.Size([1, P, 1024])

# Continue with next frame - temporal context preserved

next_tokens = torch.randn(B * S_local, P, C)
output2 = stream(next_tokens)  # Attends to previous frame via KV cache

To reset the cache when starting a new video sequence:

stream.clean_kv_cache()  # lines 236-247

Summary

• AggregatorStream implements causal temporal processing through a dual-backend KV cache supporting both PyTorch SDPA and FlashInfer.
• Sliding-window eviction (kv_cache_sliding_window) prevents memory overflow while scale-frame retention preserves critical tokens longer.
• Automatic frame tracking via total_frames_processed ensures accurate positional encoding across streaming calls.
• 3-D RoPE optionally augments temporal understanding through rotary positional embeddings.
• Lazy initialization of the FlashInfer manager optimizes memory allocation until first forward pass.

Frequently Asked Questions

How does AggregatorStream enforce causal attention during streaming?

The stream enforces causality by passing num_frames=S_global to each transformer block while the KV cache already contains keys and values from past frames. The attention mechanism in both SDPABlock and FlashInferBlock respects the cache length, allowing attention only to cached (past) positions and the current frame, never future frames.

What is the difference between the SDPA and FlashInfer KV cache backends?

The SDPA backend uses a simple Python dictionary ({k_i, v_i, k_i_special, v_i_special}) stored per block, suitable for standard PyTorch operations. The FlashInfer backend uses a FlashInferKVCacheManager that implements paged memory management and optimized attention kernels, providing significantly faster inference for long sequences. The FlashInfer manager is lazy-initialized in _get_flashinfer_manager() to defer memory allocation until needed.

How does the sliding window mechanism handle scale frames?

The FlashInferKVCacheManager is constructed with max_num_frames = kv_cache_scale_frames + kv_cache_sliding_window + 16, reserving extra capacity beyond the standard sliding window. When eviction occurs, the manager preserves the most recent kv_cache_scale_frames tokens while dropping older regular frames, ensuring high-priority contextual information remains available for longer temporal dependencies.

Can I reset the KV cache between video sequences?

Yes, call clean_kv_cache() (defined in lines 236-247 of stream.py) to reset the internal cache state and zero out total_frames_processed. This prevents temporal leakage between unrelated videos and frees GPU memory for new sequences.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}