Alpamayo 1.5

The TrajectoryFusionMixin provides the fuse_traj_tokens method, which integrates trajectory data (ego history and future) into the model's input token sequence. This is used to convert raw trajectory data into the specialized token format the model expects.

Requirements: To use this mixin, the model must have:

• hist_traj_tokenizer: A tokenizer for history trajectories.
• hist_token_start_idx: The starting index for history tokens in the vocabulary.
• config.traj_token_ids: A mapping containing the special token IDs for trajectories (e.g., "history", "future").
• traj_tokenizer and future_token_start_idx (if future trajectories are present in the data).

Args:

• input_ids: The current [B, n_token] tensor of input IDs.
• traj_data: A dictionary containing "ego_history_xyz" and "ego_history_rot". If "ego_future_xyz" is also present, future tokens will be fused.

Returns: input_ids of shape [B, n_token] with trajectory tokens fused into the sequence.

To amplify the effect of a navigation instruction, use Classifier-Free Guidance (CFG). This requires using the model.sample_trajectories_from_data_with_vlm_rollout_cfg_nav inference function and providing additional_nav_inference_kwargs.

Note: Running CFG requires 60GB+ GPU memory.

Guidance Weight ($\alpha$) values:

• \alpha = 1: No guidance amplification (default).
• \alpha > 1: Amplified guidance (more aggressive following of instructions, less diversity).
• \alpha = 0: Pure unconditioned prediction (ignores instruction).

with torch.autocast("cuda", dtype=torch.bfloat16):
    nav_result = nav_utils.compare_nav_conditions(
        model=model,
        processor=processor,
        data=data_scene1,
        nav_text=nav_text,
        num_traj_samples=16,
        top_p=0.98,
        temperature=0.6,
        max_generation_length=256,
        return_extra=False,
        nav_inference_fn=model.sample_trajectories_from_data_with_vlm_rollout_cfg_nav,
        additional_nav_inference_kwargs={
            "diffusion_kwargs": {
                "use_classifier_free_guidance": True,
                "inference_guidance_weight": 1.5,
                "temperature": 0.6,
            }
        },
    )

For fine-grained control, you can manually inject navigation instructions into the model's message template. Use helper.create_message with the nav_text argument to include the instruction in the prompt before passing it to the model's inference API.

# Define custom instruction
nav_text = "Turn left in 30m"

# Create message with nav_text
messages_nav = helper.create_message(
    frames,
    camera_indices=data_scene1["camera_indices"],
    nav_text=nav_text,
)

# Process and run inference as usual
inputs_nav = processor.apply_chat_template(
    messages_nav,
    tokenize=True,
    add_generation_prompt=False,
    continue_final_message=True,
    return_dict=True,
    return_tensors="pt",
)

model_inputs_nav = helper.to_device(
    {
        "tokenized_data": inputs_nav,
        "ego_history_xyz": data_scene1["ego_history_xyz"],
        "ego_history_rot": data_scene1["ego_history_rot"],
    },
    "cuda",
)

with torch.autocast("cuda", dtype=torch.bfloat16):
    pred_xyz_nav, pred_rot_nav, extra_nav = model.sample_trajectories_from_data_with_vlm_rollout(
        data=model_inputs_nav,
        top_p=0.98,
        temperature=0.6,
        num_traj_samples=16,
        max_generation_length=256,
        return_extra=True,
    )

Data preparation involves three main steps:

1. Load Dataset: Use load_physical_aiavdataset with a clip_id, a PhysicalAIAVDatasetInterface instance, and a list of camera_features.
2. Create Messages: Use helper.create_message to format image frames and camera indices into a chat-like structure.
3. Tokenize: Use processor.apply_chat_template to convert messages into tensors.

Finally, package the tokenized data with ego_history_xyz and ego_history_rot into a dictionary and move it to the device using helper.to_device.

# 1. Load data
data = load_physical_aiavdataset(
    clip_id,
    avdi=avdi,
    camera_features=cam_features,
)

# 2. Create messages
messages = helper.create_message(
    data["image_frames"].flatten(0, 1),
    camera_indices=data["camera_indices"],
)

# 3. Tokenize
inputs = processor.apply_chat_template(
    messages,
    tokenize=True,
    add_generation_prompt=False,
    continue_final_message=True,
    return_dict=True,
    return_tensors="pt",
)

# 4. Package for model
model_inputs = {
    "tokenized_data": inputs,
    "ego_history_xyz": data["ego_history_xyz"],
    "ego_history_rot": data["ego_history_rot"],
}
model_inputs = helper.to_device(model_inputs, "cuda")

To perform standard trajectory prediction, load the model using Alpamayo1_5.from_pretrained, process the scene images and ego history using a helper processor, and call sample_trajectories_from_data_with_vlm_rollout. The model generates a Chain-of-Causation (CoC) reasoning trace followed by predicted trajectories.

import torch
from alpamayo1_5.models.alpamayo1_5 import Alpamayo1_5
from alpamayo1_5 import helper

# Load model
model = Alpamayo1_5.from_pretrained("nvidia/Alpamayo-1.5-10B", dtype=torch.bfloat16).to("cuda")
processor = helper.get_processor(model.tokenizer)

# Prepare inputs (assuming data_scene1 is loaded via load_physical_aiavdataset)
messages = helper.create_message(
    data_scene1["image_frames"].flatten(0, 1),
    camera_indices=data_scene1["camera_indices"],
)
inputs = processor.apply_chat_template(
    messages,
    tokenize=True,
    add_generation_prompt=False,
    continue_final_message=True,
    return_dict=True,
    return_tensors="pt",
)

model_inputs = helper.to_device(
    {
        "tokenized_data": inputs,
        "ego_history_xyz": data_scene1["ego_history_xyz"],
        "ego_history_rot": data_scene1["ego_history_rot"],
    },
    "cuda",
)

# Run inference
with torch.autocast("cuda", dtype=torch.bfloat16):
    pred_xyz, pred_rot, extra = model.sample_trajectories_from_data_with_vlm_rollout(
        data=model_inputs,
        top_p=0.98,
        temperature=0.6,
        num_traj_samples=1,
        max_generation_length=256,
        return_extra=True,
    )

print("Chain-of-Causation:\n", extra["cot"][0])

Data preparation involves three main steps:

1. Load the dataset using load_physical_aiavdataset(clip_id).
2. Create chat messages using helper.create_message, passing the flattened image frames and camera indices.
3. Tokenize the messages using processor.apply_chat_template with tokenize=True and return_tensors="pt".

Finally, construct a model_inputs dictionary containing the tokenized_data, ego_history_xyz, and ego_history_rot, and move the entire dictionary to the target device using helper.to_device(model_inputs, "cuda").

from alpamayo1_5.load_physical_aiavdataset import load_physical_aiavdataset
from alpamayo1_5 import helper

data = load_physical_aiavdataset(clip_id)

messages = helper.create_message(
    data["image_frames"].flatten(0, 1),
    camera_indices=data["camera_indices"],
)

inputs = processor.apply_chat_template(
    messages,
    tokenize=True,
    add_generation_prompt=False,
    continue_final_message=True,
    return_dict=True,
    return_tensors="pt",
)

model_inputs = {
    "tokenized_data": inputs,
    "ego_history_xyz": data["ego_history_xyz"],
    "ego_history_rot": data["ego_history_rot"],
}
model_inputs = helper.to_device(model_inputs, "cuda")