FluxDAgger: A Model-Decoupled DAgger Pipeline for Dual-Arm Robotic Manipulation

English | 简体中文

FluxDAgger is a model-decoupled DAgger pipeline for dual-arm robotic manipulation. It decouples the policy model, reward model, data collection controller, and dataset processing stack, making the system easier to adapt to different VLA models and reward models.

fluxdagger_demo.webm

The system is built around a modular ROS architecture with synchronized multi-camera observations, online human takeover, DAgger-style correction collection, reward visualization, and dataset processing.

🌟 Features

• Model-decoupled design: policy inference runs in an external project; this repository only consumes agreed ROS topics.
• Human-in-the-loop DAgger: supports autonomous rollout, online takeover, and corrective data collection.
• Synchronized multi-camera observations: timestamp-based synchronization aligns multi-camera streams and robot state.
• Four-arm Piper control: manages front/rear and master/slave arms through ROS nodes.
• Reward visualization: supports online and offline Qwen3-VL reward inference.
• Dataset processing pipeline: exports raw Parquet, segmented videos, NumPy arrays, and reward annotations.

📢 Latest News

[2026/05/29] 🔥 FluxDAgger has been open-sourced.

🛠️ Installation

1. Clone the repository

git clone https://github.com/FluxVLA/FluxDAgger.git
cd FluxDAgger

2. Set up the environment

bash setup_env.sh

setup_env.sh uses official PyPI by default. If PyPI is slow in mainland China, enable a mirror via DAGGER_DISABLE_PIP_MIRROR=0 — see Documentation. Manual installs are covered under Installation.

3. Build the ROS workspace

source /opt/ros/noetic/setup.bash
catkin_make
source devel/setup.bash

Usage

Launch FluxDAgger data collection

roslaunch dagger dagger.launch \
  scene_dir:=fold_towels \
  ckpt_dir:=/path/to/checkpoint \
  task_id:=policy_name \
  save_data_base_dir:=/home/agilex/dagger_data/data_buffer

Launch FluxDAgger with Qwen3 reward

roslaunch dagger dagger_with_reward.launch \
  scene_dir:=fold_towels \
  ckpt_dir:=/path/to/checkpoint \
  task_id:=policy_name \
  qwen3_model_path:=/path/to/qwen3_reward/checkpoint \
  save_data_base_dir:=/home/agilex/dagger_data/data_buffer

Framework

CAN Bus Control

For four-arm teleoperation and DAgger collection, FluxDAgger uses a modified CAN wiring and configuration. The original two USB-CAN setup is extended to four independent USB-CAN interfaces, so each Piper arm can be addressed through its own CAN channel instead of sharing a coupled control path. This allows the front/rear and master/slave arms to be enabled, switched, and commanded independently by the four arm_node instances.

Original CAN Bus Control	Modified CAN Bus Control

Timestamp Synchronization

Precise temporal alignment across multiple camera streams is critical for consistent multi-view observations. FluxDAgger follows the timestamp synchronization implementation from the official AgileX data collection system: each camera maintains a frame buffer, and frames are aligned to a common sync timestamp. Frames arriving before the sync time are discarded, ensuring all camera views correspond to the same physical moment during rollout and human takeover.

Repository Layout

FluxDAgger
├── src/
│   ├── dagger/
│   │   ├── config/      # default.yaml — single source of defaults
│   │   ├── dagger/      # Python package (nodes import dagger.* )
│   │   ├── launch/      # roslaunch files
│   │   └── scripts/     # ROS node entrypoints
│   └── dagger_msgs/     # Custom ROS messages
├── tools/
│   ├── data_processing/ # Parquet/video/NumPy data processing
│   ├── hardware/        # CAN and Piper initialization tools
│   ├── reward_inference/ # Offline reward inference scripts
│   └── visualize/       # Real-time and offline visualization tools
├── data_example/        # Example episode and tool output workspace
├── docs/                # Technical documentation
├── requirements.txt
└── setup_env.sh

Core Modules

Module	Description
sync_observation_node.py	Publishes synchronized multi-camera and robot-state observations.
dagger_controller_node.py	Manages episode state, keyboard commands, and arm mode switching.
dagger_collector_node.py	Collects synchronized observations, actions, and episode metadata.
qwen3_reward_node.py	Runs asynchronous Qwen3 reward inference and publishes reward signals.
arm_node.py	Controls per-arm mode, target subscription, and enable state.

Data Processing

FluxDAgger stores each collected episode as Parquet data with camera and robot states, then processes the data into multiple downstream formats:

• human/rollout parquet subsets
• full-trajectory MP4 videos and NumPy arrays
• videos segmented around human takeover periods
• offline reward annotations and visualization results

Related Projects

• FluxVLA: a full-stack engineering platform for VLA training, evaluation, acceleration, and real-robot inference.
• ARM: Advantage Reward Modeling for long-horizon robotic manipulation.

Support

If you encounter any issues while using this repository, feel free to contact us. You can reach us directly at ryan.hu@limxdynamics.com and wayne@limxdynamics.com, or open a GitHub issue for help.

🙏 Citation & Acknowledgements

If you find FluxDAgger useful, please consider citing or linking to the project page.

Acknowledgement: FluxDAgger builds on the Piper robot ecosystem and uses agilexrobotics/piper_sdk for Piper arm integration.