算法

一文通透動作分塊算法ACT：斯坦福ALOHA團隊推出的動作序列預測算法(Action Chunking with Transformers)
比較簡單，算法題目里就寫了：Action Chunking with Transformers，比較有特色的地方就是Action Chunking，核心就是不浪費之前做過的推理預測，統統拿過來加權一下，得到最終的答案。

源碼

逐行解讀ALOHA ACT的實現：機器人動作分塊算法ACT的代碼剖析、訓練部署(含真機上的智能分揀復現)
代碼寫得很優雅，讀起來很流暢

1.1.1 模仿學習及其挑戰：Action Chunking with Transformers(ACT)

預測動作中的小誤差會引起狀態的大差異，加劇模仿學習的“復合誤差”問題。為了解決這個問題，他們從動作分塊(action chunking)中獲得靈感，這是心理學中的一個概念，描述了如何將一系列動作組合在一起作為一個塊，最終作為一個單元執行

他們使用Transformers實現動作分塊策略，并將其訓練為條件VAE (CVAE)，以捕獲人類數據中的可變性。他們將該方法命名為Action Chunking with Transformers(ACT)，并發現它在一系列模擬和現實世界的精細操作任務上顯著優于以前的模仿學習算法

2.2.2 第二步 推斷z，以獲得CVAE解碼器輸入中的風格變量z

一文通透動作分塊算法ACT：斯坦福ALOHA團隊推出的動作序列預測算法(Action Chunking with Transformers)的這句話啥意思？

最后
只取第一個輸出，它對應于**[CLS]標記**，并使用另一個線性網絡來預測z分布的均值和方差，將其參數化為對角高斯分布
且使用重新參數化獲得z的樣本，這是一種允許在采樣過程中反向傳播的標準方法，以便編碼器和解碼器可以聯合優化[33]

看detr_vae.py的代碼就知道了：
在DETRVAE的if is_training頭上有個注釋：Obtain latent z from action sequence，
意思是風格變量z就是latent_input
[CLS]標記：encoder_output = encoder_output[0] # take cls output only
均值：mu = latent_info[:, :self.latent_dim]
方差：logvar = latent_info[:, self.latent_dim:]
使用重新參數化獲得z的樣本：latent_sample = reparametrize(mu, logvar)
最后：latent_input = self.latent_out_proj(latent_sample)

2.3 優勢特征：ACT與其他模仿學習方法的比較

一方面，transformer解碼器的“query”是第一層固定的正弦位置嵌入，即如上圖右下角所示的position embeddings(fixed)，其維度為k ×512
二方面，transformer解碼器的交叉注意力(cross-attention)層中的“keys”和“values”來自上述transformer編碼器的輸出

eval_bc（評估一個行為克隆(behavior cloning)模型）和train_bc（訓練行為克隆BC模型）的區別

我看到train_bc里頭有個eval的，但這個eval應該和eval_bc不一樣，雖然兩者都要用到policy.eval()
注：policy里頭就會調用

model, optimizer = build_ACT_model_and_optimizer(args_override)
self.model = model

1.8.3.2 根據觀察結果查詢策略、獲取動作

這里的train_bc的policy調用參數是(qpos_data, image_data, action_data, is_pad)
eval_bc的policy調用參數是(qpos, curr_image)
根據參數來判斷是訓練還是推理

在訓練模式下，會計算出一系列的損失并返回一個包含這些損失的字典
在推理模式下，會從模型中獲取預測的動作并返回

aloha act代碼里頭的qpos和action有什么區別？

https://metaso.cn/s/IOAGn1O

那mu, logvar是啥

https://metaso.cn/s/IOAGn1O
在變分自編碼器（VAE）中，mu 和 logvar 是兩個關鍵參數，它們分別代表潛在變量的均值和對數方差，用于生成潛在空間的樣本。
這段代碼是 變分自編碼器（VAE） 中的 重參數化技巧（Reparameterization Trick） 的實現，其作用是 從潛在變量的分布中采樣，同時保證 梯度可以連續傳播，從而實現端到端的訓練。

def reparametrize(mu, logvar):std = logvar.div(2).exp()eps = Variable(std.data.new(std.size()).normal_())return mu + std * eps

編碼器和編碼器的輸入與輸出

backbone + encoder 等等輸入到 self.transformer，其實self.transformer就是decoder部分
核心代碼是detr_vae.py的class DETRVAE(nn.Module):的def forward的if is_training:部分
前提：detr_vae.py的class DETRVAE(nn.Module):的def forward的參數：qpos, image, env_state, actions, is_pad，都來自于imitate_episodes.py的def forward_pass(data, policy)的data

編碼器的輸入與輸出

編碼器的核心調用語句：self.encoder(encoder_input, pos=pos_embed, src_key_padding_mask=is_pad)
參數的來源：

# project action sequence to embedding dim, and concat with a CLS token
action_embed = self.encoder_action_proj(actions) # (bs, seq, hidden_dim)
qpos_embed = self.encoder_joint_proj(qpos)  # (bs, hidden_dim) # qpos來自于forward_pass(data, policy):的image_data, qpos_data, action_data, is_pad = data
qpos_embed = torch.unsqueeze(qpos_embed, axis=1)  # (bs, 1, hidden_dim)
cls_embed = self.cls_embed.weight # (1, hidden_dim)
cls_embed = torch.unsqueeze(cls_embed, axis=0).repeat(bs, 1, 1) # (bs, 1, hidden_dim)
encoder_input = torch.cat([cls_embed, qpos_embed, action_embed], axis=1) # (bs, seq+1, hidden_dim)
encoder_input = encoder_input.permute(1, 0, 2) # (seq+1, bs, hidden_dim)
# do not mask cls token 輸出形狀為(bs, 2)的二維張量，里面元素全部填充為False
cls_joint_is_pad = torch.full((bs, 2), False).to(qpos.device) # False: not a padding
is_pad = torch.cat([cls_joint_is_pad, is_pad], axis=1)  # (bs, seq+1)
# obtain position embedding
pos_embed = self.pos_table.clone().detach()
pos_embed = pos_embed.permute(1, 0, 2)  # (seq+1, 1, hidden_dim)
# query model
encoder_output = self.encoder(encoder_input, pos=pos_embed, src_key_padding_mask=is_pad)
encoder_output = encoder_output[0] # take cls output only

編碼器的輸入與輸出

編碼器的的核心調用語句為
hs = self.transformer(src, None, self.query_embed.weight, pos, latent_input, proprio_input, self.additional_pos_embed.weight)[0]
其中：

1. src

all_cam_features = []
for cam_id, cam_name in enumerate(self.camera_names):features, pos = self.backbones[0](image[:, cam_id]) # HARDCODEDfeatures = features[0] # take the last layer featurepos = pos[0]all_cam_features.append(self.input_proj(features))all_cam_pos.append(pos)
# fold camera dimension into width dimension
src = torch.cat(all_cam_features, axis=3)

2. pos

for cam_id, cam_name in enumerate(self.camera_names):features, pos = self.backbones[0](image[:, cam_id]) # HARDCODEDfeatures = features[0] # take the last layer featurepos = pos[0]all_cam_features.append(self.input_proj(features))all_cam_pos.append(pos)
pos = torch.cat(all_cam_pos, axis=3)

3. latent_input 【Obtain latent z from action sequence】里的latent z

self.latent_dim = 32
latent_info = self.latent_proj(encoder_output) # 來自于編碼器的輸出
mu = latent_info[:, :self.latent_dim] # 潛在變量的均值
logvar = latent_info[:, self.latent_dim:] # 潛在變量的對數方差
latent_sample = reparametrize(mu, logvar) 
latent_input = self.latent_out_proj(latent_sample)

4. proprio_input = self.input_proj_robot_state(qpos) # qpos來自于forward_pass(data, policy):的image_data, qpos_data, action_data, is_pad = data

為什么env_max_reward 設成0 ？

可能真機不需要看模擬出來的精度？

# load environment
if real_robot:from aloha_scripts.robot_utils import move_grippers # requires alohafrom aloha_scripts.real_env import make_real_env # requires alohaenv = make_real_env(init_node=True)env_max_reward = 0 # 為什么設成0 ？
success_rate = np.mean(np.array(highest_rewards) == env_max_reward)
avg_return = np.mean(episode_returns)
summary_str = f'\nSuccess rate: {success_rate}\nAverage return: {avg_return}\n\n'
for r in range(env_max_reward+1):more_or_equal_r = (np.array(highest_rewards) >= r).sum()more_or_equal_r_rate = more_or_equal_r / num_rolloutssummary_str += f'Reward >= {r}: {more_or_equal_r}/{num_rollouts} = {more_or_equal_r_rate*100}%\n'print(summary_str)