当前位置：首页 > news >正文

机械臂模仿学习2.1：行为克隆

news 2026/9/1 15:50:08

核心定义与本质

行为克隆(BC)是一种模仿学习方法，模仿人的动作，学出一个策略网络

本质是学习从状态到动作的映射函数，将模仿转化为监督学习（分类或者回归）

实现过程

1.数据收集

数据集由（状态，动作）二元组构成，记作

sj 为原始状态
aj为专家动作

2.模型选择

选择策略模型 πθ，以状态 s 为输入，输出动作 a 的概率分布或具体值

连续动作：线性模型线性层，输出连续值

离散动作：分类模型Softmax，输出动作概率分布

3.损失函数

连续动作：均方误差（MSE）或Huber 损失

离散动作：交叉熵

4.监督训练

使用梯度下降等算法最小化损失，调整参数 θ，使模型输出逼近专家动作

5.策略部署

训练完成后，模型 πθ 可直接用于新状态输入，独立预测并执行动作。

优化目标函数

痛点

分布偏移：模型仅在专家访问过的状态分布上训练。实际部署时，微小误差会导致模型进入数据集从未去过的“新状态”，失去参考依据。

复合误差：在“新状态”下，模型缺乏经验可能做出更错误的动作，导致状态进一步偏离。误差像滚雪球一样累积，最终导致任务彻底失败。

代码实践

1.训练PPO智能体

train_expert.py

# train_expert_ppo.py import paddle import paddle.nn.functional as F import gymnasium as gym import numpy as np from tqdm import tqdm class PolicyNet(paddle.nn.Layer): def __init__(self, state_dim, hidden_dim, action_dim): super().__init__() self.fc1 = paddle.nn.Linear(state_dim, hidden_dim) self.fc2 = paddle.nn.Linear(hidden_dim, action_dim) def forward(self, x): x = F.relu(self.fc1(x)) return F.softmax(self.fc2(x), axis=-1) # 输出概率 class ValueNet(paddle.nn.Layer): def __init__(self, state_dim, hidden_dim): super().__init__() self.fc1 = paddle.nn.Linear(state_dim, hidden_dim) self.fc2 = paddle.nn.Linear(hidden_dim, 1) def forward(self, x): x = F.relu(self.fc1(x)) return self.fc2(x) def compute_advantage(rewards, values, gamma=0.99, lam=0.95): """计算 GAE 优势函数""" advantages = [] gae = 0 values = values + [0] # 最后加一个0作为下一个状态的值 for t in reversed(range(len(rewards))): delta = rewards[t] + gamma * values[t+1] - values[t] gae = delta + gamma * lam * gae advantages.insert(0, gae) return advantages def ppo_update(policy, value, optimizer_p, optimizer_v, states, actions, old_log_probs, advantages, returns, clip_eps=0.2, epochs=10): for _ in range(epochs): # 计算新策略的对数概率 probs = policy(states) dist = paddle.distribution.Categorical(probs) # 调整 actions 形状以匹配 log_prob 要求 actions = actions.reshape([-1, 1]) # [batch_size, 1] new_log_probs = dist.log_prob(actions) # 结果应为 [batch_size, 1] new_log_probs = new_log_probs.squeeze(-1) # 压缩为 [batch_size] 以便与 old_log_probs 一致 # 计算比率 ratio = paddle.exp(new_log_probs - old_log_probs) # old_log_probs 是 [batch_size] surr1 = ratio * advantages surr2 = paddle.clip(ratio, 1 - clip_eps, 1 + clip_eps) * advantages policy_loss = -paddle.mean(paddle.minimum(surr1, surr2)) # 价值网络损失 value_pred = value(states).squeeze() # [batch_size] value_loss = F.mse_loss(value_pred, returns) # 更新 optimizer_p.clear_grad() policy_loss.backward() optimizer_p.step() optimizer_v.clear_grad() value_loss.backward() optimizer_v.step() def collect_trajectory(env, policy, value, max_steps=500): states, actions, rewards, log_probs, values = [], [], [], [], [] state = env.reset()[0] if isinstance(env.reset(), tuple) else env.reset() done = False while not done: state_t = paddle.to_tensor(state, dtype='float32').unsqueeze(0) probs = policy(state_t) v = value(state_t).item() dist = paddle.distribution.Categorical(probs) action = dist.sample([1]).item() log_prob = dist.log_prob(paddle.to_tensor([action])).item() step_result = env.step(action) if len(step_result) == 5: next_state, reward, terminated, truncated, _ = step_result done = terminated or truncated else: next_state, reward, done, _ = step_result states.append(state) actions.append(action) rewards.append(reward) log_probs.append(log_prob) values.append(v) state = next_state return states, actions, rewards, log_probs, values def train_ppo(env, policy, value, optimizer_p, optimizer_v, episodes=1000): return_list = [] for episode in tqdm(range(episodes), desc="PPO训练"): # 收集一条轨迹 states, actions, rewards, log_probs, values = collect_trajectory(env, policy, value) episode_return = sum(rewards) return_list.append(episode_return) # 计算优势与回报 advantages = compute_advantage(rewards, values) returns = [adv + v for adv, v in zip(advantages, values)] # 简化：return = advantage + value advantages = paddle.to_tensor(advantages, dtype='float32') returns = paddle.to_tensor(returns, dtype='float32') states = paddle.to_tensor(states, dtype='float32') actions = paddle.to_tensor(actions, dtype='int64') old_log_probs = paddle.to_tensor(log_probs, dtype='float32') # 归一化优势 advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8) # PPO更新 ppo_update(policy, value, optimizer_p, optimizer_v, states, actions, old_log_probs, advantages, returns) if (episode+1) % 100 == 0: avg = np.mean(return_list[-100:]) print(f"Episode {episode+1}, 平均回报: {avg:.2f}") return return_list # 主程序 env = gym.make('CartPole-v1') state_dim = env.observation_space.shape[0] action_dim = env.action_space.n hidden_dim = 128 policy = PolicyNet(state_dim, hidden_dim, action_dim) value = ValueNet(state_dim, hidden_dim) optimizer_p = paddle.optimizer.Adam(learning_rate=3e-4, parameters=policy.parameters()) optimizer_v = paddle.optimizer.Adam(learning_rate=1e-3, parameters=value.parameters()) print("开始PPO训练...") returns = train_ppo(env, policy, value, optimizer_p, optimizer_v, episodes=500) # PPO通常收敛更快 paddle.save(policy.state_dict(), "net_ppo.pdparams") print("专家模型已保存为 net_ppo.pdparams")

2.行为克隆

bc_cartpole.py

import paddle import paddle.nn.functional as F import gymnasium as gym import numpy as np import matplotlib.pyplot as plt from tqdm import tqdm # -------------------- 网络定义 -------------------- class PolicyNet(paddle.nn.Layer): def __init__(self, state_dim, hidden_dim, action_dim): super(PolicyNet, self).__init__() self.fc1 = paddle.nn.Linear(state_dim, hidden_dim) self.fc2 = paddle.nn.Linear(hidden_dim, action_dim) def forward(self, x): x = F.relu(self.fc1(x)) # 注意：这里返回 logits（不加 softmax），以便使用交叉熵 return self.fc2(x) # -------------------- 专家数据采样 -------------------- def sample_expert_data(n_episode, env, model): states = [] actions = [] for episode in range(n_episode): reset_result = env.reset() if isinstance(reset_result, tuple): state = reset_result[0] else: state = reset_result done = False while not done: state_tensor = paddle.to_tensor(state, dtype='float32').unsqueeze(0) logits = model(state_tensor) # 注意：模型返回 logits probs = F.softmax(logits, axis=-1) # 转换为概率以便取动作 action = paddle.argmax(probs, axis=-1).item() states.append(state) actions.append(action) step_result = env.step(action) if len(step_result) == 5: next_state, reward, terminated, truncated, _ = step_result done = terminated or truncated else: next_state, reward, done, _ = step_result state = next_state return np.array(states), np.array(actions) # -------------------- 行为克隆智能体 -------------------- class BehaviorClone: def __init__(self, state_dim, hidden_dim, action_dim, lr): self.policy = PolicyNet(state_dim, hidden_dim, action_dim) self.optimizer = paddle.optimizer.Adam(parameters=self.policy.parameters(), learning_rate=lr) def learn(self, states, actions): states = paddle.to_tensor(states, dtype="float32") actions = paddle.to_tensor(actions, dtype="int64") # 一维标签 logits = self.policy(states) # [batch, action_dim] loss = F.cross_entropy(logits, actions) # 交叉熵损失 self.optimizer.clear_grad() loss.backward() self.optimizer.step() return loss def take_action(self, state, deterministic=True): state = paddle.to_tensor([state], dtype="float32") logits = self.policy(state) probs = F.softmax(logits, axis=-1) if deterministic: return paddle.argmax(probs, axis=-1).item() else: return paddle.distribution.Categorical(probs).sample([1]).item() # -------------------- 测试函数 -------------------- def test_agent(agent, env, n_episode, deterministic=True): return_list = [] for _ in range(n_episode): reset_result = env.reset() state = reset_result[0] if isinstance(reset_result, tuple) else reset_result episode_return = 0 done = False while not done: action = agent.take_action(state, deterministic=deterministic) step_result = env.step(action) if len(step_result) == 5: next_state, reward, terminated, truncated, _ = step_result done = terminated or truncated else: next_state, reward, done, _ = step_result state = next_state episode_return += reward return_list.append(episode_return) return np.mean(return_list) def test_expert(env, model, n_episodes=10): returns = [] for _ in range(n_episodes): reset_result = env.reset() state = reset_result[0] if isinstance(reset_result, tuple) else reset_result episode_return = 0 done = False while not done: state_tensor = paddle.to_tensor(state, dtype='float32').unsqueeze(0) logits = model(state_tensor) probs = F.softmax(logits, axis=-1) action = paddle.argmax(probs, axis=-1).item() step_result = env.step(action) if len(step_result) == 5: state, reward, terminated, truncated, _ = step_result done = terminated or truncated else: state, reward, done, _ = step_result episode_return += reward returns.append(episode_return) return np.mean(returns) # -------------------- 主程序 -------------------- env_name = 'CartPole-v1' env = gym.make(env_name) state_dim = env.observation_space.shape[0] action_dim = env.action_space.n hidden_dim = 128 # 加载专家模型 actor = PolicyNet(state_dim, hidden_dim, action_dim) try: layer_state_dict = paddle.load("net_ppo.pdparams") actor.set_state_dict(layer_state_dict) print("专家模型加载成功") except Exception as e: print("加载失败，请检查文件路径:", e) exit() # 测试专家模型得分 expert_score = test_expert(env, actor) print(f"专家模型平均回报: {expert_score:.2f}") # 如果专家得分太低，建议重新训练专家（见后文） if expert_score < 400: print("警告：专家模型得分偏低，行为克隆效果可能不佳。建议先重新训练专家。") # 采样专家数据 n_episode = 100 expert_s, expert_a = sample_expert_data(n_episode, env, actor) print("专家数据量：", expert_s.shape[0]) # 训练行为克隆 lr = 1e-4 bc_agent = BehaviorClone(state_dim, hidden_dim, action_dim, lr) n_iterations = 3000 batch_size = 64 test_returns = [] with tqdm(total=n_iterations, desc="训练进度") as pbar: for i in range(n_iterations): sample_indices = np.random.randint(0, expert_s.shape[0], size=batch_size) bc_agent.learn(expert_s[sample_indices], expert_a[sample_indices]) if (i + 1) % 10 == 0: current_return = test_agent(bc_agent, env, 5, deterministic=True) test_returns.append(current_return) pbar.set_postfix({'return': '%.3f' % current_return}) pbar.update(1) # 绘制结果 plt.plot(range(len(test_returns)), test_returns) plt.xlabel('Iterations (x10)') plt.ylabel('Returns') plt.title('BC on {} (Expert Score: {:.1f})'.format(env_name, expert_score)) plt.show()

绘制行为克隆的奖励曲线

iteration_list = list(range(len(test_returns))) plt.plot(iteration_list, test_returns) plt.xlabel('Iterations') plt.ylabel('Returns') plt.title('BC on {}'.format(env_name)) plt.show()

查看全文

http://www.cnnetsun.cn/news/1256312.html