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

> Explore how LingBot-Map's AggregatorStream processes temporal data with its KV cache. Learn about PyTorch SDPA, FlashInfer, and automatic eviction for efficient video analysis.

- Repository: [Robbyant/lingbot-map](https://github.com/Robbyant/lingbot-map)
- Tags: deep-dive
- Published: 2026-07-03

---

**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`](https://github.com/Robbyant/lingbot-map/blob/main/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:

```python

# 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:

```python
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:

```python

# 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:

```python
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

```python
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:

```python
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`](https://github.com/Robbyant/lingbot-map/blob/main/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.