核心定义与本质

行为克隆(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)

# -------------------- 经验缓冲区 --------------------
class RolloutBuffer:
    def __init__(self):
        self.states = []
        self.actions = []
        self.rewards = []
        self.log_probs = []
        self.values = []
        self.dones = []  # 终止标志（仅当 episode 因终止结束时为 True）

    def store(self, state, action, reward, log_prob, value, done):
        self.states.append(state)
        self.actions.append(action)
        self.rewards.append(reward)
        self.log_probs.append(log_prob)
        self.values.append(value)
        self.dones.append(done)

    def clear(self):
        self.states.clear()
        self.actions.clear()
        self.rewards.clear()
        self.log_probs.clear()
        self.values.clear()
        self.dones.clear()

# -------------------- GAE 优势计算 --------------------
def compute_gae(rewards, values, dones, next_value, gamma=0.99, lam=0.95):
    advantages = []
    gae = 0
    values = values + [next_value]  # 追加下一个状态的价值
    for t in reversed(range(len(rewards))):
        # 如果当前步是终止状态，则后续价值应为 0
        delta = rewards[t] + gamma * values[t+1] * (1 - dones[t]) - values[t]
        gae = delta + gamma * lam * (1 - dones[t]) * gae
        advantages.insert(0, gae)
    returns = [adv + val for adv, val in zip(advantages, values[:-1])]
    return advantages, returns

# -------------------- PPO 更新函数（修正版） --------------------
def ppo_update(policy, value, optimizer_p, optimizer_v,
               states, actions, old_log_probs, advantages, returns,
               clip_eps=0.2, epochs=10, mini_batch_size=64, entropy_coef=0.01):
    # 转换为 tensor
    states = paddle.to_tensor(states, dtype='float32')
    actions = paddle.to_tensor(actions, dtype='int64').reshape([-1, 1])  # [N, 1]
    old_log_probs = paddle.to_tensor(old_log_probs, dtype='float32')     # [N]
    advantages = paddle.to_tensor(advantages, dtype='float32')           # [N]
    returns = paddle.to_tensor(returns, dtype='float32')                 # [N]

    # 优势归一化
    advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)

    # 创建数据集
    dataset = paddle.io.TensorDataset([states, actions, old_log_probs, advantages, returns])
    data_loader = paddle.io.DataLoader(dataset, batch_size=mini_batch_size, shuffle=True)

    for _ in range(epochs):
        for batch in data_loader:
            s_batch, a_batch, old_log_batch, adv_batch, ret_batch = batch

            # 策略前向
            probs = policy(s_batch)                     # [batch_size, action_dim]
            dist = paddle.distribution.Categorical(probs)
            # 传入动作时保持 [batch_size, 1] 形状，然后压缩结果
            new_log_probs = dist.log_prob(a_batch)      # [batch_size, 1]
            new_log_probs = new_log_probs.squeeze(-1)   # [batch_size]
            entropy = dist.entropy().mean()

            # 比率
            ratio = paddle.exp(new_log_probs - old_log_batch)  # [batch_size]
            surr1 = ratio * adv_batch
            surr2 = paddle.clip(ratio, 1 - clip_eps, 1 + clip_eps) * adv_batch
            policy_loss = -paddle.mean(paddle.minimum(surr1, surr2))

            # 熵正则
            policy_loss -= entropy_coef * entropy

            # 价值损失
            value_pred = value(s_batch).squeeze()       # [batch_size]
            value_loss = F.mse_loss(value_pred, ret_batch)

            # 更新策略
            optimizer_p.clear_grad()
            policy_loss.backward()
            optimizer_p.step()

            # 更新价值网络
            optimizer_v.clear_grad()
            value_loss.backward()
            optimizer_v.step()

# -------------------- 收集固定步数的轨迹 --------------------
def collect_rollout(env, policy, value, buffer, num_steps=2048):
    state, _ = env.reset()
    done = False
    truncated = False

    for _ in range(num_steps):
        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()

        next_state, reward, terminated, truncated, _ = env.step(action)
        done = terminated or truncated

        buffer.store(state, action, reward, log_prob, v, terminated)  # 存储终止标志

        state = next_state
        if done:
            state, _ = env.reset()

    # 最后一个状态的价值（如果未结束）
    last_state_t = paddle.to_tensor(state, dtype='float32').unsqueeze(0)
    next_value = value(last_state_t).item() if not truncated else 0.0
    return next_value

# -------------------- 主训练函数 --------------------
def train_ppo(env, policy, value, optimizer_p, optimizer_v,
              total_timesteps=100000, update_timesteps=2048, **kwargs):
    buffer = RolloutBuffer()
    time_step = 0
    episode_returns = []
    progress_bar = tqdm(total=total_timesteps, desc="PPO训练")

    while time_step < total_timesteps:
        # 收集数据
        next_value = collect_rollout(env, policy, value, buffer, num_steps=update_timesteps)

        # 计算 GAE 和 returns
        advantages, returns = compute_gae(
            buffer.rewards, buffer.values, buffer.dones, next_value,
            gamma=kwargs.get('gamma', 0.99), lam=kwargs.get('lam', 0.95)
        )

        # PPO 更新
        ppo_update(policy, value, optimizer_p, optimizer_v,
                   buffer.states, buffer.actions, buffer.log_probs,
                   advantages, returns,
                   clip_eps=kwargs.get('clip_eps', 0.2),
                   epochs=kwargs.get('epochs', 10),
                   mini_batch_size=kwargs.get('mini_batch_size', 64),
                   entropy_coef=kwargs.get('entropy_coef', 0.01))

        time_step += len(buffer.rewards)
        progress_bar.update(len(buffer.rewards))

        # 可选：打印进度
        if len(episode_returns) > 0 and time_step % (update_timesteps * 5) == 0:
            avg_return = np.mean(episode_returns[-10:])
            tqdm.write(f"Step {time_step}, 最近10个episode平均回报: {avg_return:.2f}")

        buffer.clear()

    progress_bar.close()
    return episode_returns

# -------------------- 主程序 --------------------
if __name__ == '__main__':
    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())

    total_timesteps = 10000
    update_timesteps = 2048
    ppo_params = {
        'gamma': 0.99,
        'lam': 0.95,
        'clip_eps': 0.2,
        'epochs': 10,
        'mini_batch_size': 64,
        'entropy_coef': 0.01
    }

    print("开始改进版 PPO 训练...")
    returns = train_ppo(env, policy, value, optimizer_p, optimizer_v,
                        total_timesteps=total_timesteps,
                        update_timesteps=update_timesteps,
                        **ppo_params)

    paddle.save(policy.state_dict(), "net_ppo.pdparams")
    print("专家模型已保存为 net_ppo.pdparams")

    # 测试模型
    env_test = gym.make('CartPole-v1', render_mode='human')
    policy.eval()
    for i in range(5):
        state, _ = env_test.reset()
        done = False
        total_reward = 0
        while not done:
            state_t = paddle.to_tensor(state, dtype='float32').unsqueeze(0)
            probs = policy(state_t)
            action = paddle.argmax(probs).item()
            state, reward, terminated, truncated, _ = env_test.step(action)
            done = terminated or truncated
            total_reward += reward
        print(f"测试 episode {i+1} 回报: {total_reward}")
    env_test.close()

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()

3.绘制行为克隆的奖励曲线

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()