Does Robbyant/lingbot-map Support Batch Processing via stream.py? A Technical Deep Dive

Yes, Robbyant/lingbot-map fully supports batch processing through the stream.py module, which implements a streaming causal aggregator that processes batched sequences of frames while preserving the batch dimension throughout the entire pipeline.

The lingbot_map/aggregator/stream.py module in the Robbyant/lingbot-map repository provides a robust solution for processing multiple video sequences simultaneously. This technical analysis examines how the AggregatorStream class handles batch processing, from input tensors shaped [B, S, 3, H, W] through KV-cache management to final output generation.

How Batch Processing Works in stream.py

The batch processing capability centers on the AggregatorStream class defined in lingbot_map/aggregator/stream.py. This class inherits from AggregatorBase, which establishes the foundational batch-aware architecture.

The AggregatorStream Architecture

In lingbot_map/aggregator/base.py, the base class's forward method receives tensors shaped [B, S, 3, H, W]—representing B batches and S frames per batch—and processes them in a batch-aware manner (lines 63-66). The AggregatorStream constructor stores batch-related parameters including max_frame_num and kv_cache_* configurations, selecting between FlashInfer KV cache or PyTorch SDPA backends without collapsing the batch dimension.

Batch Dimension Preservation

Throughout the streaming pipeline, all tensors maintain the leading B axis. The method _process_global_attention (approximately line 370 in stream.py) forwards batch-aware arguments to _process_causal_stream, which reshapes tokens from [B·S_local, P, C] to [B, S_local·P, C] when necessary while preserving the batch axis (lines 354-360). This ensures that simultaneous streaming of multiple sequences occurs without interference between batch elements.

Key Implementation Details

Input Handling and Tensor Shapes

The pipeline expects input tensors with shape [B, S, 3, H, W] where:

• B is the batch size (number of video sequences)
• S is the number of frames per sequence
• 3 represents RGB channels
• H and W are height and width (typically 518×518)

The forward method in AggregatorBase (used unchanged by the streaming subclass) loops over block groups, invoking _process_frame_attention and the overridden _process_global_attention. Both functions accept B and return outputs with shape [B, S, P, 2C], preserving the batch dimension for downstream tasks (lines 399-406 in base.py).

KV-Cache Management for Batches

The KV-cache handling in stream.py maintains a per-batch cache through methods like _init_kv_cache, _get_flashinfer_manager, and clean_kv_cache. The FlashInfer manager stores caches for each block but indexes them per-frame, not per-batch, enabling simultaneous streaming of multiple sequences. This design allows the system to process batch element 0's frame t independently from batch element 1's frame t while sharing model weights.

Attention Mechanisms

The streaming logic relies on lingbot_map/layers/block.py for building blocks (Block, FlashInferBlock, SDPABlock) used in both frame and global attention paths. The lingbot_map/layers/rope.py module provides 2-D and 3-D RoPE position embeddings that the streaming aggregator optionally applies to batched sequences.

Practical Implementation Guide

Basic Batch Processing Example

The following code demonstrates processing a batch of 2 video clips, each containing 8 frames:

import torch
from lingbot_map.aggregator.stream import AggregatorStream

# Create a batch of 2 video clips, each with 8 frames of size 518×518

B, S, C, H, W = 2, 8, 3, 518, 518
images = torch.randn(B, S, C, H, W)  # values in [0, 1] expected by the model

# Initialise the streaming aggregator (default params)

agg = AggregatorStream(
    sliding_window_size=4,          # optional causal window

    num_frame_for_scale=2,          # scale-estimation frames

    enable_3d_rope=True,           # enable temporal RoPE

    use_sdpa=False,                 # use FlashInfer KV cache (default)

)

# Forward pass – returns a list of block outputs, each shaped [B, S, P, 2C]

outputs, patch_start_idx = agg(images)

print(f"Number of block outputs: {len(outputs)}")
print(f"Shape of first block output: {outputs[0].shape}")   # => torch.Size([2, 8, P, 2C])

Handling Variable Length Sequences

While the aggregator requires uniform frame counts within a batch (dimension S), you can process sequences of different lengths by padding or processing separately:

# Example with different frame counts per sample

batch_images = [
    torch.randn(5, 3, 518, 518),   # 5 frames

    torch.randn(10, 3, 518, 518),  # 10 frames

]

# Pad to the same length or process separately

# The aggregator requires uniform S for batched input

Batch size B can be any positive integer; the module streams each sequence independently while sharing model weights across the batch.

Summary

• AggregatorStream in lingbot_map/aggregator/stream.py inherits from AggregatorBase and fully supports batch processing via the B dimension.
• Input tensors use shape [B, S, 3, H, W] and outputs preserve [B, S, P, 2C] throughout the pipeline.
• KV-cache management indexes per-frame rather than per-batch, enabling efficient parallel streaming of multiple sequences.
• The architecture supports both FlashInfer and PyTorch SDPA backends while maintaining batch awareness in attention mechanisms.

Frequently Asked Questions

What input tensor shape does stream.py expect for batch processing?

The AggregatorStream class expects input tensors with shape [B, S, 3, H, W], where B is the batch size, S is the number of frames per sequence, 3 represents RGB channels, and H and W are typically 518×518 pixels. This shape is processed by the forward method in lingbot_map/aggregator/base.py (lines 63-66) and maintained throughout the streaming pipeline.

How does the KV-cache handle multiple batches simultaneously?

The KV-cache implementation in stream.py uses per-frame indexing rather than per-batch indexing. Methods like _init_kv_cache and _get_flashinfer_manager manage caches for each block but organize them by frame position, allowing the system to track temporal dependencies independently for each batch element while sharing the underlying model weights.

Can I process videos with different frame counts in the same batch?

The aggregator requires uniform sequence lengths within a single forward pass (constant S dimension). For videos with different frame counts, you must either pad sequences to the same length or process them in separate batches. The max_frame_num parameter helps manage memory allocation for variable-length content.

What is the difference between AggregatorStream and the base aggregator?

AggregatorStream specializes the base AggregatorBase class for causal streaming inference. While both support batch processing, AggregatorStream adds FlashInfer KV-cache support via lingbot_map/layers/flashinfer_cache.py, sliding window attention for temporal causality, and specific optimizations for frame-by-frame processing that preserve the batch dimension throughout streaming operations.

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