メインコンテンツまでスキップ

ACT (Action Chunking Transformer) モデルのファインチューニング

概要

ACT(Action Chunking Transformer)は、精密な操作タスク向けに設計されたエンドツーエンドの模倣学習モデルです。このモデルは、アクションチャンク(action chunks)を予測することで、従来の模倣学習における複合誤差(compounding errors)の問題を克服し、低コストのハードウェア上で高い成功率のロボット操作を実現します。

主な特徴

  • アクションチャンキング予測:複数の連続するアクションを一度に予測し、複合誤差を減少させます。
  • Transformer アーキテクチャ:アテンションメカニズムを利用してシーケンスデータを処理します。
  • エンドツーエンドのトレーニング:生の観測データから直接アクションを予測します。
  • 高い成功率:精密な操作タスクにおいて優れたパフォーマンスを発揮します。
  • ハードウェアフレンドリー:コンシューマー向けのハードウェアで実行可能です。

先決条件

システム要件

  • OS:Linux(Ubuntu 20.04+ 推奨)または macOS
  • Python バージョン:3.8+
  • GPU:NVIDIA GPU(RTX 3070 以上推奨)、少なくとも 6GB のビデオメモリ
  • メモリ:少なくとも 16GB の RAM
  • ストレージ:少なくとも 30GB の空き容量

環境準備

1. LeRobot のインストール

# LeRobot リポジトリをクローン
git clone https://github.com/huggingface/lerobot.git
cd lerobot

# 仮想環境の作成(venv 推奨、conda でも可)
python -m venv .venv
source .venv/bin/activate
python -m pip install -U pip

# 依存関係のインストール
pip install -e .

2. ACT 特定の依存関係のインストール

# 追加のパッケージをインストール
pip install einops
pip install timm
pip install wandb

# Weights & Biases にログイン(任意)
wandb login

ACT モデルアーキテクチャ

コアコンポーネント

  1. 視覚エンコーダー:マルチビュー画像入力を処理します。
  2. 状態エンコーダー:ロボットの状態情報を処理します。
  3. Transformer デコーダー:アクションシーケンスを生成します。
  4. アクションヘッド:最終的なアクション予測を出力します。

主要パラメータ

  • Chunk Size:一度に予測されるアクション数(通常 50-100)。
  • Context Length:履歴観測の長さ。
  • Hidden Dimension:Transformer の隠れ層次元。
  • Number of Heads:アテンションヘッドの数。
  • Number of Layers:Transformer の層数。

データ準備

LeRobot 形式データ

ACT は LeRobot 形式のデータセットを必要とし、以下の構造を含みます:

your_dataset/
├── data/
│ ├── chunk-001/
│ │ ├── observation.images.cam_high.png
│ │ ├── observation.images.cam_low.png
│ │ ├── observation.images.cam_left_wrist.png
│ │ ├── observation.images.cam_right_wrist.png
│ │ ├── observation.state.npy
│ │ ├── action.npy
│ │ └── ...
│ └── chunk-002/
│ └── ...
├── meta.json
├── stats.safetensors
└── videos/
├── episode_000000.mp4
└── ...

データ品質要件

  • 基本的なトレーニングには最低 50 エピソードが必要。
  • 最適な効果を得るには 200 エピソード以上を推奨。
  • 各エピソードには完全なタスク実行が含まれている必要があります。
  • マルチビュー画像(少なくとも 2 台のカメラ)。
  • 高品質なアクションアノテーション。

ファインチューニングのトレーニング

重要なパラメータ制約

ACT モデルの n_action_steps chunk_size である必要があります。両方を同じ値(例:100)に設定することを推奨します。

  • chunk_size:モデルが一度に予測するアクションシーケンスの長さ
  • n_action_steps:実際に実行されるアクションステップ数

基本トレーニングコマンド

# 環境変数の設定
export HF_USER="your-huggingface-username"
export CUDA_VISIBLE_DEVICES=0

# ACT トレーニングの開始
lerobot-train \
--policy.type act \
--dataset.repo_id ${HF_USER}/your_dataset \
--batch_size 8 \
--steps 50000 \
--output_dir outputs/train/act_finetuned \
--job_name act_finetuning \
--policy.device cuda \
--policy.chunk_size 100 \
--policy.n_action_steps 100 \
--policy.n_obs_steps 1 \
--policy.optimizer_lr 1e-5 \
--policy.optimizer_weight_decay 1e-4 \
--policy.push_to_hub false \
--save_checkpoint true \
--save_freq 10000 \
--wandb.enable true

高度なトレーニング設定

マルチカメラ設定

# マルチカメラ設定向けの ACT トレーニング
lerobot-train \
--policy.type act \
--dataset.repo_id ${HF_USER}/your_dataset \
--batch_size 4 \
--steps 100000 \
--output_dir outputs/train/act_multicam \
--job_name act_multicam_training \
--policy.device cuda \
--policy.chunk_size 100 \
--policy.n_action_steps 100 \
--policy.n_obs_steps 2 \
--policy.vision_backbone resnet18 \
--policy.dim_model 512 \
--policy.dim_feedforward 3200 \
--policy.n_encoder_layers 4 \
--policy.n_decoder_layers 1 \
--policy.n_heads 8 \
--policy.optimizer_lr 1e-5 \
--policy.optimizer_weight_decay 1e-4 \
--policy.push_to_hub false \
--save_checkpoint true \
--wandb.enable true

メモリ最適化設定

# ビデオメモリの小さい GPU 向け
lerobot-train \
--policy.type act \
--dataset.repo_id ${HF_USER}/your_dataset \
--batch_size 2 \
--steps 75000 \
--output_dir outputs/train/act_memory_opt \
--job_name act_memory_optimized \
--policy.device cuda \
--policy.chunk_size 100 \
--policy.n_action_steps 100 \
--policy.n_obs_steps 1 \
--policy.vision_backbone resnet18 \
--policy.dim_model 256 \
--policy.optimizer_lr 1e-5 \
--policy.use_amp true \
--num_workers 2 \
--policy.push_to_hub false \
--save_checkpoint true \
--wandb.enable true

パラメータ詳細説明

コアパラメータ

パラメータ意味推奨値説明
--policy.typeポリシータイプactACT モデルタイプ
--policy.pretrained_path事前学習済みモデルパスlerobot/actLeRobot 公式 ACT モデル(任意)
--dataset.repo_idデータセットリポジトリ ID${HF_USER}/your_datasetあなたの HuggingFace データセット
--batch_sizeバッチサイズ8ビデオメモリに応じて調整。RTX 3070 なら 4-8 推奨
--stepsトレーニングステップ数50000精密タスクには 50,000-100,000 ステップ推奨
--output_dir出力ディレクトリoutputs/train/act_finetunedモデル保存パス
--job_nameジョブ名act_finetuningログおよび実験追跡用(任意)

ACT 特定パラメータ

パラメータ意味推奨値説明
--policy.chunk_sizeアクションチャンクサイズ100一度に予測するアクション数
--policy.n_action_steps実行アクションステップ数100実際に実行するアクション数
--policy.n_obs_steps観測ステップ数1履歴観測の数
--policy.vision_backbone視覚バックボーンresnet18画像特徴抽出ネットワーク
--policy.dim_modelモデル次元512Transformer の主次元
--policy.dim_feedforwardフィードフォワード次元3200Transformer FF 層の次元
--policy.n_encoder_layersエンコーダー層数4Transformer エンコーダー層数
--policy.n_decoder_layersデコーダー層数1Transformer デコーダー層数
--policy.n_headsアテンションヘッド数8マルチヘッドアテンションのヘッド数
--policy.use_vaeVAE の使用true変分目的の最適化

トレーニングパラメータ

パラメータ意味推奨値説明
--policy.optimizer_lr学習率1e-5ACT は小さめの学習率を推奨
--policy.optimizer_weight_decay重み減衰0.0正則化パラメータ
--policy.optimizer_lr_backboneバックボーン学習率1e-5視覚エンコーダーの学習率
--policy.use_amp混合精度trueビデオメモリの節約
--num_workersデータロードスレッド数4CPU コア数に応じて調整
--policy.push_to_hubHub へのプッシュfalseHuggingFace へのアップロード(repo_id が必要)
--save_checkpointチェックポイント保存trueトレーニングチェックポイントの保存
--save_freq保存頻度10000チェックポイント保存間隔

トレーニングの監視とデバッグ

Weights & Biases の統合

# 詳細な W&B 設定
lerobot-train \
--policy.type act \
--dataset.repo_id your-name/your-dataset \
--batch_size 8 \
--steps 50000 \
--policy.push_to_hub false \
--wandb.enable true \
--wandb.project act_experiments \
--wandb.notes "ACT training with 4 cameras" \
# ... 他のパラメータ

主要指標の監視

トレーニング中に注目すべき指標:

  • Total Loss:全体的な損失、安定して下降すべき。
  • Action Loss:アクション予測損失(L1/L2 損失)。
  • Learning Rate:学習率の変化曲線。
  • Gradient Norm:勾配ノルム、勾配爆発の監視。
  • GPU Memory:ビデオメモリの使用状況。
  • Training Speed:1秒あたりの処理サンプル数。

トレーニングログの分析

# log_analysis.py
import wandb
import matplotlib.pyplot as plt

def analyze_training_logs(project_name, run_name):
api = wandb.Api()
run = api.run(f"{project_name}/{run_name}")

# トレーニング指標の取得
history = run.history()

# 損失曲線のプロット
plt.figure(figsize=(12, 4))

plt.subplot(1, 3, 1)
plt.plot(history['step'], history['train/total_loss'])
plt.title('Total Loss')
plt.xlabel('Step')
plt.ylabel('Loss')

plt.subplot(1, 3, 2)
plt.plot(history['step'], history['train/action_loss'])
plt.title('Action Loss')
plt.xlabel('Step')
plt.ylabel('Loss')

plt.subplot(1, 3, 3)
plt.plot(history['step'], history['train/learning_rate'])
plt.title('Learning Rate')
plt.xlabel('Step')
plt.ylabel('LR')

plt.tight_layout()
plt.savefig('training_analysis.png')
plt.show()

if __name__ == "__main__":
analyze_training_logs("act_experiments", "act_v1_multicam")

モデルの評価

オフライン評価

# offline_evaluation.py
import torch
import numpy as np
from lerobot.policies.act.modeling_act import ACTPolicy
from lerobot.datasets.lerobot_dataset import LeRobotDataset

def evaluate_act_model(model_path, dataset_path):
# モデルのロード
policy = ACTPolicy.from_pretrained(model_path, device="cuda")
policy.eval()

# テストデータセットのロード
dataset = LeRobotDataset(dataset_path, split="test")

total_l1_loss = 0
total_l2_loss = 0
num_samples = 0

with torch.no_grad():
for batch in dataset:
# モデルの予測
prediction = policy(batch)

# 損失の計算
target_actions = batch['action']
predicted_actions = prediction['action']

l1_loss = torch.mean(torch.abs(predicted_actions - target_actions))
l2_loss = torch.mean((predicted_actions - target_actions) ** 2)

total_l1_loss += l1_loss.item()
total_l2_loss += l2_loss.item()
num_samples += 1

avg_l1_loss = total_l1_loss / num_samples
avg_l2_loss = total_l2_loss / num_samples

print(f"Average L1 Loss: {avg_l1_loss:.4f}")
print(f"Average L2 Loss: {avg_l2_loss:.4f}")

return avg_l1_loss, avg_l2_loss

if __name__ == "__main__":
model_path = "outputs/train/act_finetuned/checkpoints/last"
dataset_path = "path/to/your/test/dataset"
evaluate_act_model(model_path, dataset_path)

オンライン評価(ロボット環境)

# robot_evaluation.py
import torch
import numpy as np
from lerobot.policies.act.modeling_act import ACTPolicy

class ACTRobotController:
def __init__(self, model_path, camera_names):
self.policy = ACTPolicy.from_pretrained(model_path, device="cuda")
self.policy.eval()
self.camera_names = camera_names
self.action_queue = []

def get_action(self, observations):
# アクションキューが空の場合、新しいアクションチャンクを予測
if len(self.action_queue) == 0:
with torch.no_grad():
# 入力の構築
batch = self.prepare_observation(observations)

# アクションチャンクの予測
prediction = self.policy(batch)
actions = prediction['action'].cpu().numpy()[0] # [chunk_size, action_dim]

# アクションをキューに追加
self.action_queue = list(actions)

# キューから次のアクションを返す
return self.action_queue.pop(0)

def prepare_observation(self, observations):
batch = {}

# 画像観測の処理
for cam_name in self.camera_names:
image_key = f"observation.images.{cam_name}"
if image_key in observations:
image = observations[image_key]
# 画像の前処理(正規化、リサイズなど)
image_tensor = self.preprocess_image(image)
batch[image_key] = image_tensor.unsqueeze(0)

# 状態観測の処理
if "observation.state" in observations:
state = torch.tensor(observations["observation.state"], dtype=torch.float32)
batch["observation.state"] = state.unsqueeze(0)

return batch

def preprocess_image(self, image):
# 画像前処理ロジック
# トレーニング時と同じ処理を行う必要があります
image_tensor = torch.tensor(image).permute(2, 0, 1).float() / 255.0
return image_tensor

# 使用例
if __name__ == "__main__":
controller = ACTRobotController(
model_path="outputs/train/act_finetuned/checkpoints/last",
camera_names=["cam_high", "cam_low", "cam_left_wrist", "cam_right_wrist"]
)

# ロボット制御ループのシミュレーション
for step in range(100):
# 現在の観測を取得(ここでは実際のロボットから取得する必要があります)
observations = {
"observation.images.cam_high": np.random.randint(0, 255, (480, 640, 3)),
"observation.images.cam_low": np.random.randint(0, 255, (480, 640, 3)),
"observation.state": np.random.randn(7)
}

# アクションを取得
action = controller.get_action(observations)

# アクションを実行(ロボットへ送信)
print(f"Step {step}: Action = {action}")

# ここで実際のアクションをロボットに送信します
# robot.execute_action(action)

デプロイと最適化

モデルの量子化

# quantization.py
import torch
from lerobot.policies.act.modeling_act import ACTPolicy

def quantize_act_model(model_path, output_path):
# モデルのロード
policy = ACTPolicy.from_pretrained(model_path, device="cpu")
policy.eval()

# 動的量子化
quantized_policy = torch.quantization.quantize_dynamic(
policy,
{torch.nn.Linear},
dtype=torch.qint8
)

# 量子化モデルの保存
torch.save(quantized_policy.state_dict(), output_path)
print(f"Quantized model saved to {output_path}")

return quantized_policy

if __name__ == "__main__":
quantize_act_model(
"outputs/train/act_finetuned/checkpoints/last",
"outputs/act_quantized.pth"
)

推論の最適化

# optimized_inference.py
import torch
import torch.jit
from lerobot.policies.act.modeling_act import ACTPolicy

class OptimizedACTInference:
def __init__(self, model_path, use_jit=True):
self.policy = ACTPolicy.from_pretrained(model_path, device="cuda")
self.policy.eval()

if use_jit:
# TorchScript による最適化
self.policy = torch.jit.script(self.policy)

# モデルのウォームアップ
self.warmup()

def warmup(self):
# ダミーデータを使用してモデルをウォームアップ
dummy_batch = {
"observation.images.cam_high": torch.randn(1, 3, 224, 224, device="cuda"),
"observation.images.cam_low": torch.randn(1, 3, 224, 224, device="cuda"),
"observation.state": torch.randn(1, 7, device="cuda")
}

with torch.no_grad():
for _ in range(10):
_ = self.policy(dummy_batch)

@torch.no_grad()
def predict(self, observations):
# 高速推論
prediction = self.policy(observations)
return prediction['action'].cpu().numpy()

if __name__ == "__main__":
inference = OptimizedACTInference(
"outputs/train/act_finetuned/checkpoints/last"
)

# 推論速度のテスト
import time

observations = {
"observation.images.cam_high": torch.randn(1, 3, 224, 224, device="cuda"),
"observation.images.cam_low": torch.randn(1, 3, 224, 224, device="cuda"),
"observation.state": torch.randn(1, 7, device="cuda")
}

start_time = time.time()
for _ in range(100):
action = inference.predict(observations)
end_time = time.time()

avg_inference_time = (end_time - start_time) / 100
print(f"Average inference time: {avg_inference_time:.4f} seconds")
print(f"Inference frequency: {1/avg_inference_time:.2f} Hz")

ベストプラクティス

データ収集のアドバイス

  1. マルチビューデータ:複数のカメラを使用して豊富な視覚情報を取得します。
  2. 高品質なデモンストレーション:デモンストレーションデータの一貫性と正確性を確保します。
  3. タスクの多様性:異なる初期状態や目標構成を含めます。
  4. 失敗ケース:ロバスト性を高めるために、適切な失敗ケースを含めます。

トレーニング最適化のアドバイス

  1. アクションチャンクサイズ:タスクの複雑さに応じて chunk_size を調整します。
  2. 学習率スケジューラ:コサインアニーリングやステップ減衰を使用します。
  3. 正則化:重み減衰(Weight Decay)やドロップアウトを適切に使用します。
  4. データ拡張:画像に対して適切な拡張処理を行います。

デプロイ最適化のアドバイス

  1. モデル圧縮:量子化や剪定技術を使用してモデルサイズを削減します。
  2. 推論加速:TensorRT や ONNX を使用して推論を最適化します。
  3. メモリ管理:アクションキューと観測バッファを適切に管理します。
  4. リアルタイム性の保証:推論周波数が制御要件を満たしていることを確認します。

よくある質問 (FAQ)

Q: ACT は他の模倣学習手法と比較してどのような利点がありますか?

A: ACT の主な利点は以下の通りです:

  • 複合誤差の減少:アクションチャンクを予測することで誤差の累積を減少させます。
  • 成功率の向上:精密な操作タスクにおいて優れたパフォーマンスを発揮します。
  • エンドツーエンドのトレーニング:特徴量の手動設計が不要です。
  • マルチモーダル融合:視覚情報と状態情報を効果的に融合します。

Q: 適切な chunk_size をどのように選択すればよいですか?

A: chunk_size の選択はタスクの特性に依存します:

  • 高速なタスクchunk_size = 10-30
  • 中程度のタスクchunk_size = 50-100
  • 低速なタスクchunk_size = 100-200
  • 一般的には 50 から試すことをお勧めします。

Q: トレーニングにはどのくらい時間がかかりますか?

A: トレーニング時間は複数の要因に依存します:

  • データセットサイズ:100 エピソードで約 4-8 時間(RTX 3070)
  • モデルの複雑さ:モデルが大きいほど時間がかかります
  • ハードウェア構成:高性能な GPU はトレーニング時間を大幅に短縮します
  • 収束要件:通常 50,000-100,000 ステップ必要です

Q: マルチカメラデータをどのように処理しますか?

A: マルチカメラ処理のアドバイス:

  • カメラの選択:情報が補完的な視点を選択します。
  • 特徴融合:特徴量レベルで融合を行います。
  • アテンションメカニズム:モデルに重要な視点に注目することを学習させます。
  • 計算リソース:マルチカメラは計算負荷が増えることに注意してください。

Q: モデルの汎化能力をどのように向上させますか?

A: 汎化能力を向上させる方法:

  • データの多様性:異なる条件下でデータを収集します。
  • データ拡張:画像およびアクションの拡張技術を使用します。
  • 正則化:適切な重み減衰とドロップアウト。
  • ドメインランダム化:シミュレーションでドメインランダム化技術を使用します。
  • マルチタスク学習:複数の関連タスクで共同トレーニングを行います。

関連リソース

更新ログ

  • 2024-01: 初版リリース
  • 2024-02: マルチカメラサポートの追加
  • 2024-03: トレーニング効率と推論速度の最適化
  • 2024-04: モデル圧縮とデプロイ最適化の追加