Stream.py Aggregator Input and Output Formats: A Complete Technical Guide

The AggregatorStream class in lingbot_map/aggregator/stream.py accepts token tensors of shape [B×S_local, P, C] and optional positional embeddings, returning updated tokens of the same shape along with a global index and intermediate representations.

The streaming aggregator in the Robbyant/lingbot-map repository processes patch embeddings for vision transformers using a causal attention mechanism with KV-caching. Understanding the exact tensor contracts for the AggregatorStream is essential for integrating this module into video understanding pipelines or custom vision architectures.

Input Tensor Specifications

The public forward-style interface inherited from AggregatorBase expects pre-computed patch embeddings that include special tokens (camera, register, or scale tokens).

Token Embeddings

The primary input tokens is a PyTorch tensor with the following layout:

• Shape: [B × S_local, P, C]
• B: Batch size
• S_local: Number of consecutive frames processed together (the local window)
• P: Tokens per frame (patch tokens plus special tokens)
• C: Embedding dimension (hidden size)

This flattened batch dimension allows the aggregator to process multiple local frames simultaneously while maintaining efficient memory layout.

Positional Embeddings

The optional pos parameter supports two distinct formats depending on the RoPE configuration:

2-D RoPE (Default):

• Shape: [B × S_global, P, 2]
• Standard 2-D positional encodings for spatial coordinates

3-D RoPE (when enable_3d_rope=True):

• Shape: [1, 1, S_global × P, head_dim/2]
• Complex-valued tensor following the format returned by WanRotaryPosEmbed in lingbot_map/layers/rope.py
• Used for temporal positional encoding in video sequences

Scalar Parameters

The _process_causal_stream method requires several scalar integers that define the tensor dimensions:

• B: Batch size (re-derived from the first dimension of tokens)
• S_local: Length of the local frame segment updated in-place
• S_global: Length of the global frame segment participating in causal attention (may exceed S_local when using sliding windows)
• P: Tokens per frame including special tokens
• C: Embedding dimension
• global_idx: Current index of the global block being processed
• Additional hyper-parameters: num_frame_for_scale, sliding_window_size, image_height, image_width

Output Tensor Specifications

The internal _process_causal_stream method returns a tuple of three values with precise tensor contracts.

Updated Tokens

tokens_out contains the transformed embeddings after streaming-causal global attention:

• Shape: [B × S_local, P, C]
• Matches the input layout exactly, enabling direct consumption by subsequent layers such as the patch-embedding head

Global Index and Intermediates

new_global_idx: An integer representing the updated global block index, used for stateful processing across multiple forward passes in a streaming context.

intermediates: A list of intermediate token representations with the following properties:

• Type: list[torch.Tensor]
• Length: Equal to the number of global blocks processed in the current call (typically matching depth)
• Shape of each tensor: [B, S_local, P, C]
• Useful for debugging, feature extraction, or skip connections

Processing Implementation Details

The aggregator maintains internal KV-cache state transparently to the caller. When use_flashinfer=True, the system utilizes FlashInferKVCacheManager from lingbot_map/layers/flashinfer_cache.py; otherwise, it falls back to native SDPA with dictionary-based caching.

The causal attention mechanism respects sliding-window constraints defined by sliding_window_size, ensuring that the streaming aggregator only attends to relevant historical frames. Scale-token frames are injected according to num_frame_for_scale, enabling multi-scale feature processing.

Code Examples

Standard 2D RoPE Usage

import torch
from lingbot_map.aggregator.stream import AggregatorStream

# Initialize the streaming aggregator

agg = AggregatorStream(
    sliding_window_size=32,
    num_frame_for_scale=2,
    enable_3d_rope=False,
    use_sdpa=True,
    embed_dim=1024,
    patch_size=16,
    img_size=224,
    depth=12,
    num_register_tokens=0,
)

# Configuration

B = 4
S_local = 2
S_global = 6
P = (224 // 16) ** 2 + 1  # patches + camera token

C = 1024

# Input tensors

tokens = torch.randn(B * S_local, P, C)
pos = torch.randn(B * S_global, P, 2)  # 2D RoPE

# Process through causal stream

tokens_out, new_idx, intermediates = agg._process_causal_stream(
    tokens,
    B=B,
    S_local=S_local,
    S_global=S_global,
    P=P,
    C=C,
    global_idx=0,
    pos=pos,
    num_frame_per_block=1,
    sliding_window_size=32,
    num_frame_for_scale=2,
    image_height=224,
    image_width=224,
)

print(f"Output shape: {tokens_out.shape}")  # [8, 197, 1024]

print(f"New global index: {new_idx}")      # 12

print(f"Intermediates count: {len(intermediates)}")  # 12

3D RoPE for Temporal Video Processing

# Configure for 3D temporal positional encoding

agg = AggregatorStream(
    enable_3d_rope=True,
    max_frame_num=1024,
    embed_dim=1024,
    patch_size=16,
    img_size=224,
    depth=12,
)

# When enable_3d_rope=True, pass pos=None

# The aggregator generates 3D positions internally via _get_3d_positions_streaming()

tokens_out, new_global_idx, intermediates = agg._process_causal_stream(
    tokens,
    B=B,
    S_local=S_local,
    S_global=S_global,
    P=P,
    C=C,
    global_idx=0,
    pos=None,  # 3D RoPE generated internally

    num_frame_per_block=1,
    sliding_window_size=32,
    num_frame_for_scale=2,
    image_height=224,
    image_width=224,
)

Summary

• Input format: Token tensor of shape [B×S_local, P, C] with optional positional embeddings (2D or 3D RoPE) and scalar dimension parameters.
• Output format: Updated tokens maintaining shape [B×S_local, P, C], incremented global index, and list of intermediate feature maps.
• Key files: lingbot_map/aggregator/stream.py implements the logic, inheriting from AggregatorBase in lingbot_map/aggregator/base.py.
• KV-cache: Managed internally via FlashInferKVCacheManager or native SDPA; transparent to callers.
• RoPE support: Configurable between 2D spatial and 3D temporal positional encodings via enable_3d_rope.

Frequently Asked Questions

What is the difference between S_local and S_global in the stream aggregator?

S_local represents the number of consecutive frames processed and updated in the current forward pass, while S_global represents the total length of the frame segment participating in causal attention. When using sliding windows, S_global may exceed S_local, allowing the aggregator to attend to historical context beyond the immediately local window without updating those older frames.

How does the aggregator handle positional embeddings for temporal video data?

When enable_3d_rope=True, the AggregatorStream automatically generates 3-D rotary positional embeddings via internal methods in lingbot_map/layers/rope.py. In this mode, callers should pass pos=None, and the system constructs complex-valued position tensors of shape [1, 1, S_global×P, head_dim/2] to encode temporal relationships across video frames.

What is the shape of the intermediates list returned by the stream aggregator?

The intermediates list contains one tensor per transformer block (typically equal to depth), where each tensor has shape [B, S_local, P, C]. Unlike the flattened input/output tokens, these intermediate representations explicitly separate the batch and local frame dimensions for easier feature extraction and debugging.

Does the caller need to manage the KV-cache state manually?

No. The KV-cache state is managed internally by the AggregatorStream class, either through FlashInferKVCacheManager when FlashInfer is enabled or through a dictionary-based cache for native SDPA. Callers only need to provide the token batch and positional data; the aggregator handles cache updates, sliding window constraints, and causal masking automatically across streaming calls.

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