Low success rate in humanoid robot configuration

[continuity3]

We tested various algorithms—including pi0, pi0.5, wall-x, and LingBot-VLA—on robotic arms with different humanoid configurations. The success rates for many basic tasks were relatively low. To control for variables, we used standardized camera hardware and maintained consistent environmental lighting conditions. However, the success rate for humanoid robots was significantly lower than that for single-arm robotic arms. I would like to know if the experimental results for LingBot are consistent with this trend. If not, does the LingBot lab have any specific training methods or parameters for humanoid robots that they can share?

[YangS03]

Please consider the following debugging process:

1. First, check the open-loop results using the script attached.
2. Record the observations during the inference, then compare them with the dataset.

For your situation, regarding the statement, “However, the success rate for humanoid robots was significantly lower than that for single-arm robotic arms,” I strongly suggest that you check the deployment policy used in the open-loop inference script. For example, in [lingbot_robotwin_policy_rep.py](https://github.com/Robbyant/lingbot-vla/blob/main/deploy/lingbot_robotwin_policy_rep.py), the action and state order is changed compared with our training datasets. As a result, you will not get correct results with this policy. Instead, you should use lingbot_robotwin_policy.py in the same folder.

Alternatively, you may observe that one arm performs well while the other behaves randomly. In this case, you should check whether the action_dim in the training config matches your dataset. This parameter is used to mask unnecessary loss dimensions.

The following image shows potential results when the wrong policy is used. As shown in the image, the action results are very poor, especially after dimension 5 (the gripper of the left arm). This problem stems from the fact that training uses a dimension configuration of [arm, gripper, arm, gripper], while the model is inferred with a dimension configuration of [arm, arm, gripper, gripper].

Wrong policy result:

The following image shows the correct results on a dataset with 1,000 trajectory episodes collected from a Leju humanoid. The model is trained with a global batch size of 128, a learning rate of 1e-4, and runs for 5k steps (you will get much better results with more training steps).

Correct result:

from copy import deepcopy
from dataclasses import dataclass, field
import logging
from pathlib import Path
import re
from typing import Any
import warnings
import argparse

import json
import os
import time
from collections import deque
import torchvision
import yaml
from types import SimpleNamespace

from glob import glob
from tqdm import tqdm
from safetensors import safe_open
from pathlib import Path
from typing import Callable, Dict, List, Optional, Type, Union, Tuple, Any, Sequence

from transformers.models.auto.tokenization_auto import AutoTokenizer
from transformers import (
    AutoConfig,
    PretrainedConfig,
    PreTrainedModel,
    AutoProcessor,
)
from lerobot.configs.policies import PreTrainedConfig
from deploy.lingbot_leju_policy import QwenPi_LejuServer # TODO: replace with your policy

from typing import Callable, Dict, List, Optional, Type, Union, Tuple, Any, Sequence
import numpy as np
from matplotlib import pyplot as plt

import pandas as pd

from safetensors.torch import load_file
import torch
from torch import Tensor, nn
from PIL import Image
import torch.nn.functional as F
import transformers
from packaging.version import Version

from lerobot.common.datasets.lerobot_dataset import LeRobotDataset, LeRobotDatasetMetadata
from lingbotvla.data.vla_data.base_dataset import LejuDataset # TODO: replace with your dataset

def plot_trajectory_results(
    state_joints_across_time: np.ndarray,
    gt_action_across_time: np.ndarray,
    pred_action_across_time: np.ndarray,
    traj_id: int,
    state_keys: list[str],
    action_keys: list[str],
    action_horizon: int,
    save_plot_path: str,
) -> None:
    """
    Plot and save trajectory results comparing ground truth and predicted actions.

    Args:
        state_joints_across_time: Array of state joints over time
        gt_action_across_time: Ground truth actions over time
        pred_action_across_time: Predicted actions over time
        traj_id: Trajectory ID
        state_keys: List of state modality keys
        action_keys: List of action modality keys
        action_horizon: Action horizon used for inference
        save_plot_path: Path to save the plot
    """
    actual_steps = len(gt_action_across_time)
    action_dim = gt_action_across_time.shape[1]

    indices_to_plot = list(range(action_dim))

    num_plots = len(indices_to_plot)
    if num_plots == 0:
        logging.warning("No valid indices to plot")
        return

    # Always plot and save
    fig, axes = plt.subplots(nrows=num_plots, ncols=1, figsize=(8, 4 * num_plots))

    # Handle case where there's only one subplot
    if num_plots == 1:
        axes = [axes]

    # Add a global title showing the modality keys
    fig.suptitle(
        f"Trajectory {traj_id} - State: {', '.join(state_keys)} | Action: {', '.join(action_keys)}",
        fontsize=16,
        color="blue",
    )

    for plot_idx, action_idx in enumerate(indices_to_plot):
        ax = axes[plot_idx]
        # The dimensions of state_joints and action are the same
        # only when the robot uses actions directly as joint commands.
        # Therefore, do not plot them if this is not the case.
        if state_joints_across_time.shape == gt_action_across_time.shape:
            ax.plot(state_joints_across_time[:, action_idx], label="state joints")
        ax.plot(gt_action_across_time[:, action_idx], label="gt action")
        ax.plot(pred_action_across_time[:, action_idx], label="pred action")

        # put a dot every ACTION_HORIZON
        for j in range(0, actual_steps, action_horizon):
            if j == 0:
                ax.plot(j, gt_action_across_time[j, action_idx], "ro", label="inference point")
            else:
                ax.plot(j, gt_action_across_time[j, action_idx], "ro")

        ax.set_title(f"Action {action_idx}")
        ax.legend()

    plt.tight_layout()

    # Create filename with trajectory ID
    Path(save_plot_path).parent.mkdir(parents=True, exist_ok=True)
    plt.savefig(save_plot_path)

    plt.close()  # Close the figure to free memory


def evaluate_single_trajectory(
    policy,
    loader,
    traj_id: int,
    modality_keys: list[str] | None = None,
    steps=300,
    action_horizon=16,
    save_plot_path=None,
    prompt = None
):
    # Ensure steps doesn't exceed trajectory length

    start_id, end_id = loader.dataset.episode_data_index['from'][traj_id], loader.dataset.episode_data_index['to'][traj_id]

    arm_indices = loader.arm_indices
    effector_indices = loader.effector_indices
    new_indices = loader.new_indices
    origin_indices = torch.argsort(new_indices)

    
    gt_action_across_time = []
    state_joints_across_time = []
    pred_action_across_time  = []

    count = 0
    for data_id in range(start_id, end_id, 50):
        traj = loader[data_id]
        count += 1
        observation = {'observation.state': traj['observation.state'], 'task': traj['task']} 

        if prompt is not None:
            observation['task'] = prompt

        for image_key in image_keys:
            image = (traj[image_key]* 255).to(torch.uint8).permute(1, 2, 0).cpu().numpy()
            observation[image_key] = image
        preds = policy.infer(observation)

        gt_action_across_time += [traj['action'][:, origin_indices].numpy()]
        state_joints_across_time += [traj['observation.state'][origin_indices].numpy()]
        pred_action_across_time += [preds['action'][:, origin_indices]]
        
        # if count >=10: break
    
    gt_action_across_time = np.concatenate(gt_action_across_time, axis=0)
    state_joints_across_time = np.concatenate(state_joints_across_time, axis=0)
    pred_action_across_time = np.concatenate(pred_action_across_time, axis=0)
    
    pred_action_across_time = np.array(pred_action_across_time)
    assert gt_action_across_time.shape == pred_action_across_time.shape, (
        f"gt_action: {gt_action_across_time.shape}, pred_action: {pred_action_across_time.shape}"
    )

    # calc MSE and MAE across time
    mse = np.mean((gt_action_across_time - pred_action_across_time) ** 2)
    mae = np.mean(np.abs(gt_action_across_time - pred_action_across_time))
    logging.info(f"Unnormalized Action MSE across single traj: {mse}")
    logging.info(f"Unnormalized Action MAE across single traj: {mae}")

    logging.info(f"gt_action_joints vs time {gt_action_across_time.shape}")
    logging.info(f"pred_action_joints vs time {pred_action_across_time.shape}")

    # Plot trajectory results
    plot_trajectory_results(
        state_joints_across_time=state_joints_across_time,
        gt_action_across_time=gt_action_across_time,
        pred_action_across_time=pred_action_across_time,
        traj_id=traj_id,
        state_keys=['observation.state'],
        action_keys=['action'],
        action_horizon=action_horizon,
        save_plot_path=save_plot_path or f"/tmp/open_loop_eval/traj_{traj_id}.jpeg",
    )

    return mse, mae

def main(policy, robo_name, data_root, traj_ids, chunk_size, prompt, save_plot_path):

    policy.data_config.num_episode = None
    policy.data_config.chunk_size = policy.config.chunk_size

    policy.data_config.train_path = data_root
    policy.data_config.data_name = robo_name

    dataset_meta = LeRobotDatasetMetadata(data_root)
    delta_timestamps = {}
    for action_feature in ['action']:
        delta_timestamps[action_feature] = [t / dataset_meta.fps for t in range(policy.config.chunk_size)]
    dataset = LejuDataset(repo_id=data_root, open_loop_eval=True)
    print(f"Dataset length: {len(dataset)}")
    logging.info(f"Running evaluation on trajectories: {traj_ids}")

    all_mse = []
    all_mae = []

    for traj_id in traj_ids:
        if traj_id not in dataset.dataset.meta.episodes.keys():
            logging.warning(f"Trajectory ID {traj_id} is out of range. Skipping.")
            continue

        print(f"Running trajectory: {traj_id}")
        mse, mae = evaluate_single_trajectory(
            policy,
            dataset,
            traj_id,
            save_plot_path=os.path.join(save_plot_path,f'{traj_id}.png'),
            action_horizon=chunk_size,
            prompt = prompt
        )
        print(f"MSE for trajectory {traj_id}: {mse}, MAE: {mae}")
        all_mse.append(mse)
        all_mae.append(mae)

    if all_mse:
        avg_mse = np.mean(np.array(all_mse))
        avg_mae = np.mean(np.array(all_mae))
        print(f"Average MSE across all trajs: {avg_mse}")
        print(f"Average MAE across all trajs: {avg_mae}")
    else:
        logging.info("No valid trajectories were evaluated.")
    logging.info("Done")


if __name__ == "__main__":

    parser = argparse.ArgumentParser(description="open loop test")

    parser.add_argument(
        "--robo_name",
        type=str,
        default="leju",
    )

    parser.add_argument(
        "--model_path",
        type=str,
        default="/path/to/checkpoints/global_step_5000/hf_ckpt",
    )

    parser.add_argument(
        "--data_path",
        type=str,
        default="/path/to/data/TASK1-ToySorting/TASK1-ToySorting"
    )

    parser.add_argument(
        "--use_length",
        type=int,
        default=50,
        help="chunk的使用长度"
    )

    parser.add_argument(
        "--norm",
        type=str,
        help="norm file",
        default="/path/to/lingbot-vla/assets/norm_stats/leju_task_1_delta.json"
    )

    parser.add_argument(
        "--prompt",
        type=str,
        default="pick and place"
    )

    parser.add_argument(
        "--save_plot_path",
        type=str,
        default='./open_loop_test/leju_5k',
    )

    args = parser.parse_args()


    image_keys = [
        "observation.images.head_cam_h",
        "observation.images.wrist_cam_l",
        "observation.images.wrist_cam_r",
    ]


    robo_name = args.robo_name
    data_root = args.data_path
    model_path = args.model_path
    robot_norm_path = args.norm

    save_plot_path = args.save_plot_path
    os.makedirs(save_plot_path, exist_ok=True)
    traj_ids = list(range(0, 160, 10))
    model = QwenPi_LejuServer(
                path_to_pi_model=model_path,
                use_length=50,
                use_bf16=True,
                chunk_ret=True
            )
    model.reset(robo_name)
    main(model, robo_name, data_root, traj_ids, args.use_length, args.prompt, save_plot_path)

[YangS03]

Dataset example with dimensions matching those of our pretraining checkpoint [arm arm gripper gripper]. Please pay attention to the action-dimension order in policy.py.

class LejuDataset(Dataset):
    def __init__(
        self,
        repo_id="leju",
        config=PI0Config,
        tokenizer=AutoTokenizer,
        data_config=None,
        image_processor=None,
        use_depth_align=False,
        compute_norm=False,
        open_loop_eval=False,
    ):
        if not (compute_norm or open_loop_eval):
            image_transforms = Resize((data_config.img_size, data_config.img_size))
            self.image_processor = image_processor
            self.norm_stats_file = data_config.norm_stats_file
        else:
            image_transforms = None
        self.state_key = 'observation.state'
        self.action_key = 'action'
        self.top_img_key = 'observation.images.head_cam_h'
        self.left_img_key = 'observation.images.wrist_cam_l'
        self.right_img_key = 'observation.images.wrist_cam_r'
        # [i / 30 for i in range(50)] represents action chunks in 50 steps at 30 FPS.
        # The timestamps are set to 0 for the images and state, as we only use current obs.
        self.config = config
        self.tokenizer = tokenizer
        self.dataset_meta = LeRobotDatasetMetadata(repo_id)
        delta_timestamps = {
            self.action_key: [t / self.dataset_meta.fps for t in range(50)],
        }
        self.dataset = LeRobotDataset(
            repo_id=repo_id,
            image_transforms=image_transforms,
            delta_timestamps=delta_timestamps,
        )
        if not (compute_norm or open_loop_eval):
            with open(self.norm_stats_file) as f:
                self.norm_stats = json.load(f)
            self.normalizer = Normalizer(
                # norm_stats=self.dataset.meta.stats,
                norm_stats=self.norm_stats['norm_stats'],
                from_file=True,
                data_type='customized', # TODO
                norm_type={
                    self.top_img_key: "identity",
                    self.left_img_key: "identity",
                    self.right_img_key: "identity",
                    self.state_key: data_config.norm_type,
                    self.action_key: data_config.norm_type,
                },
            )
        self.use_depth_align = use_depth_align
        self.compute_norm = compute_norm
        self.arm_indices = torch.tensor([0, 1, 2, 3, 4, 5, 6, 8, 9, 10, 11, 12, 13, 14])
        self.effector_indices = torch.tensor([7, 15])
        self.new_indices = torch.cat([self.arm_indices, self.effector_indices])

        # get delta arm action
        self.delta_dim_indices = torch.tensor(np.arange(self.arm_indices.shape[0]))

        self.open_loop_eval = open_loop_eval

    def __len__(self):
        return len(self.dataset)

    def getdata(self, idx):
        item = self.dataset[idx]
        task = self.dataset_meta.tasks[int(item['task_index'])]
        assert task == item['task']
        item[self.action_key] = item[self.action_key][:, self.new_indices]
        item[self.state_key] = item[self.state_key][self.new_indices]

        if self.open_loop_eval: return item

        item[self.action_key][:, self.delta_dim_indices] -= item[self.state_key][None, self.delta_dim_indices]

        if self.compute_norm:
            return item
        else:
            normalized_item = self.normalizer.normalize(item)
            base_image = (normalized_item[self.top_img_key] * 255).to(torch.uint8)
            left_wrist_image = (normalized_item[self.left_img_key] * 255).to(
                torch.uint8
            )
            right_wrist_image = (normalized_item[self.right_img_key] * 255).to(
                torch.uint8
            )
            batch_dict =  {
                "image": {"base_0_rgb": base_image, "left_wrist_0_rgb": left_wrist_image, "right_wrist_0_rgb": right_wrist_image},
                "state": normalized_item[self.state_key].to(torch.float32),
                "action": normalized_item[self.action_key].to(torch.float32),
                "action_is_pad": normalized_item["action_is_pad"],
                "prompt": [item["task"]],
            }
            state = prepare_state(self.config, batch_dict) # bs,8 -> bs,32
            lang_tokens, lang_masks = prepare_language(self.config, self.tokenizer, batch_dict) # bs, seq_len
            actions = prepare_action(self.config, batch_dict) # bs,50,7 -> bs,50,32 , 7
            images, img_masks, pil_images = prepare_images(self.config, self.image_processor, batch_dict, use_depth_align=self.use_depth_align)

            batch_dict = {
                'images': images,
                'img_masks': img_masks,
                'state': state,
                'lang_tokens': lang_tokens,
                'lang_masks': lang_masks,
                'actions': actions,
                'action_is_pad': batch_dict['action_is_pad'],
            }
            if self.use_depth_align: batch_dict['pil_images'] = pil_images

            return batch_dict

    def __getitem__(self, idx: int) -> dict[str, torch.Tensor]:
        if idx < 0 or idx >= len(self):
            raise IndexError(f"Index {idx} out of bounds.")
        max_retries = 200
        attempts = 0
        cur = idx
        last_err = None
        while attempts < max_retries:
            try:
                return self.getdata(cur)
            except Exception as e:
                last_err = e
                attempts += 1
                cur = np.random.randint(0, len(self))
                if cur >= len(self):
                    cur = 0
                continue

        raise RuntimeError(
            f"Failed to fetch a valid item starting from idx={idx} after {attempts} attempts. "
            f"Last error: {repr(last_err)}"
        )

Script for computing the norm.json

import json
import numpy as np
import os
import re
import time
from pathlib import Path
from dataclasses import asdict, dataclass, field
from typing import TYPE_CHECKING

import torch
from tqdm import trange, tqdm
from torch.utils.data import DataLoader
from lingbotvla.utils import helper
from lingbotvla.utils.arguments import DataArguments, ModelArguments, TrainingArguments, parse_args
import lingbotvla.utils.normalize as normalize
from lingbotvla.data.vla_data.base_dataset import LejuDataset


if TYPE_CHECKING:
    from transformers import ProcessorMixin

    from lingbotvla.data.chat_template import ChatTemplate

logger = helper.create_logger(__name__)


@dataclass
class MyDataArguments(DataArguments):
    norm_path: str = field(
        default=None,
        metadata={"help": "Path to save norm stats."},
    )
    chunk_size: int = field(
        default=50,
        metadata={"help": "Chunk size of action."},
    )


@dataclass
class Arguments:
    model: "ModelArguments" = field(default_factory=ModelArguments)
    data: "MyDataArguments" = field(default_factory=MyDataArguments)
    train: "TrainingArguments" = field(default_factory=TrainingArguments)

def compute_norm(dataset, task_id, batch_size, stats, state_norm_keys, acton_norm_keys, delta_norm):
    data_loader = DataLoader(dataset, batch_size=batch_size, num_workers=16, shuffle=False, drop_last=True)
    success = False
    delta_dim_indices = dataset.delta_dim_indices
    for batch in tqdm(data_loader, desc=f"Computing stats of {task_id}"):
        for key in state_norm_keys:
            values = np.asarray(batch[key])
            # values = batch[key]
            stats[key].update(values.reshape(-1, values.shape[-1]))
        for key in acton_norm_keys:
            # print(batch[key].shape) # bs,50,16
            delta_action = batch[key][:, :, delta_dim_indices] # bs,50,14
            no_delta_indices = torch.tensor(list(set(range(batch[key].shape[-1])) - set(delta_dim_indices.tolist())))
            no_delta_action = batch[key][:, :, no_delta_indices][:,0].unsqueeze(1).repeat(1, 50, 1) # bs,50,2 -> bs,2 -> bs,50,2
            batch[key] = torch.cat([delta_action, no_delta_action], dim=-1) # bs,50,16
            values = np.asarray(batch[key][:,0]) if not delta_norm[key] else np.asarray(batch[key].reshape(batch[key].shape[0], -1)) # bs,50*16
            stats[key].update(values.reshape(-1, values.shape[-1]))
        success = True
    return success


def main():
    args = parse_args(Arguments)
    logger.info(f"Process rank: {args.train.global_rank}, world size: {args.train.world_size}")
    logger.info_rank0(json.dumps(asdict(args), indent=2))
    
    logger.info_rank0("Prepare data")
    stats = None
    
    assert args.data.datasets_type == 'vla'
    dataset = LejuDataset(repo_id=args.data.train_path, compute_norm=True)
    # print(dataset.new_indices)

    state_norm_keys = ['observation.state']
    acton_norm_keys = ['action']
    delta_norm = {'action': True}
    stats = {key: normalize.RunningStats() for key in acton_norm_keys+state_norm_keys}
    
    chunk_size = args.data.chunk_size

    try:
        success = compute_norm(dataset, args.data.train_path, args.train.global_batch_size, stats, state_norm_keys, acton_norm_keys, delta_norm)
    except Exception as e:
        fail_info = f"{args.data.train_path} {e}"
        print(fail_info)
        


    if success:
        norm_stats = {key: stats.get_statistics() for key, stats in stats.items()}
        norm_stats = {}
        for key, stats in stats.items():
            if key in delta_norm and delta_norm[key]==True:
                norm_stats[key] = stats.get_statistics(chunk_size=chunk_size)
            else:
                norm_stats[key] = stats.get_statistics()

        output_path = Path(args.data.norm_path)
        print(f"Writing stats to: {output_path}")
        normalize.save(output_path, norm_stats, stats._count)
        


if __name__ == "__main__":
    main()

[continuity3]

Thank you for your reply. I wanted to clarify that I used EEF for training. Whether it's the LingBotVLA, pi0, or GR00T model, the fitting effect on the training data is very good, and the performance in external tests is also excellent. However, during real-device inference, the robot's bias becomes somewhat large. Each VLA model is trained with around 200 data points over 50,000+ steps, which I believe is sufficient for overfitting. I'd like to know if all the data you collected for the real-machine experiments were joint angle data, and whether you used end-effector EEF for training. If so, could you provide your training curves for reference? Below are the training curves we've generated.

[YangS03]

The purple curve corresponds to the second figure in my previous response. Using the default L1 loss, the checkpoint at step 5k is already sufficient to achieve a reasonable success rate. Based on our experiments on Robotwin, we confirm that both end-effector (EEF) space and joint space lead to similar results; they are not the key factors determining whether the model successfully converges. This is also supported by recent work from Physical Intelligence.

What I want to emphasize is that the loss does not have a direct correlation with open-loop performance or success rate. Whether the model can infer correct actions still primarily depends on the consistency between your deployment setup and the training configuration.

[funguspooh]

Thank you for your reply. I wanted to clarify that I used EEF for training. Whether it's the LingBotVLA, pi0, or GR00T model, the fitting effect on the training data is very good, and the performance in external tests is also excellent. However, during real-device inference, the robot's bias becomes somewhat large. Each VLA model is trained with around 200 data points over 50,000+ steps, which I believe is sufficient for overfitting. I'd like to know if all the data you collected for the real-machine experiments were joint angle data, and whether you used end-effector EEF for training. If so, could you provide your training curves for reference? Below are the training curves we've generated.

Hello, could you explain how you use the End-Effectuator (EEF) for training? The official repository doesn't seem to contain any relevant code for this component. Would you be willing to provide the training code as a reference? I'm currently attempting to train using the EEF, but my code consistently fails to run properly. Thank you in advance!