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

> Discover the input and output formats for the stream.py aggregator in Robbyant/lingbot-map. Learn to process token tensors and positional embeddings for efficient aggregation. Get the technical details now.

- Repository: [Robbyant/lingbot-map](https://github.com/Robbyant/lingbot-map)
- Tags: api-reference
- Published: 2026-07-07

---

**The `AggregatorStream` class in [`lingbot_map/aggregator/stream.py`](https://github.com/Robbyant/lingbot-map/blob/main/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`](https://github.com/Robbyant/lingbot-map/blob/main/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`](https://github.com/Robbyant/lingbot-map/blob/main/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

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

```python

# 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`](https://github.com/Robbyant/lingbot-map/blob/main/lingbot_map/aggregator/stream.py) implements the logic, inheriting from `AggregatorBase` in [`lingbot_map/aggregator/base.py`](https://github.com/Robbyant/lingbot-map/blob/main/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`](https://github.com/Robbyant/lingbot-map/blob/main/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.