【RL Latest Tech】离线强化学习：行为规范Actor Critic (BRAC) 算法

# BRAC算法的Python伪代码实现

def brac_algorithm(data, policy_network, value_network, beta_policy, num_iterations, lambda_reg):
    """
    data: 离线数据集，格式为 (s, a, r, s')
    policy_network: 策略网络，用于更新策略
    value_network: 价值网络，用于估计价值函数
    beta_policy: 行为策略
    num_iterations: 迭代次数
    lambda_reg: 正则化参数
    """
    
    for _ in range(num_iterations):
        # Step 1: 价值函数估计
        for s, a, r, s_next in data:
            v_next = max(value_network.predict(s_next))
            q_value = r + gamma * v_next
            value_network.update(s, a, q_value)

        # Step 2: 策略更新
        for s in data:
            # 获取当前策略动作分布
            action_distribution = policy_network.get_action_distribution(s)
            
            # 计算行为正则项
            behavior_regularization = compute_divergence(action_distribution, beta_policy.get_action_distribution(s))
            
            # 更新策略网络，最小化负的Q值减去正则项
            policy_network.update(s, lambda a: -value_network.predict(s, a) + lambda_reg * behavior_regularization)

    return policy_network

def compute_divergence(dist1, dist2):
    """
    计算两个分布之间的散度，如KL散度
    """
    divergence = kl_divergence(dist1, dist2)
    return divergence

def kl_divergence(p, q):
    """
    计算KL散度
    """
    return sum(p[i] * math.log(p[i] / q[i]) for i in range(len(p)))

"""《BRAC算法的Pytorch实现》
    时间：2024.09.25
    环境：No
    作者：不去幼儿园
"""
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F

class BRACPolicy(nn.Module):
    def __init__(self, state_dim, action_dim, beta_policy, lambda_reg, learning_rate=0.001):
        super(BRACPolicy, self).__init__()
        self.state_dim = state_dim
        self.action_dim = action_dim
        self.beta_policy = beta_policy
        self.lambda_reg = lambda_reg
        
        # 策略网络结构
        self.policy_network = nn.Sequential(
            nn.Linear(self.state_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 64),
            nn.ReLU(),
            nn.Linear(64, self.action_dim),
            nn.Softmax(dim=-1)
        )
        
        # 优化器
        self.optimizer = optim.Adam(self.policy_network.parameters(), lr=learning_rate)

    def forward(self, states):
        return self.policy_network(states)

    def update(self, states, value_network):
        # 将状态转换为张量
        states = torch.FloatTensor(states)

        # 计算策略的动作分布
        actions_probs = self.forward(states)
        
        # 从行为策略获取动作分布
        with torch.no_grad():
            beta_actions_probs = self.beta_policy.get_action_distribution(states)
        
        # 计算价值损失
        q_values = value_network.predict(states)
        value_loss = -torch.mean(torch.sum(q_values * actions_probs, dim=1))
        
        # 计算正则化项 (KL散度)
        kl_divergence = F.kl_div(actions_probs.log(), beta_actions_probs, reduction='batchmean')
        
        # 总损失
        loss = value_loss + self.lambda_reg * kl_divergence

        # 反向传播和优化
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

    def get_action_distribution(self, states):
        states = torch.FloatTensor(states)
        with torch.no_grad():
            return self.forward(states).numpy()

# 示例用法
state_dim = 10
action_dim = 5
lambda_reg = 0.1

# 假设有一个预训练的行为策略
class BetaPolicy:
    def get_action_distribution(self, states):
        # 返回一些伪随机的动作分布
        return torch.rand(states.shape[0], action_dim)

beta_policy = BetaPolicy()

# 创建BRAC-P策略对象
brac_policy = BRACPolicy(state_dim, action_dim, beta_policy, lambda_reg)

# 假设有一个简单的价值网络
class ValueNetwork:
    def predict(self, states):
        # 返回一些伪随机的Q值
        return torch.rand(states.shape[0], action_dim)

value_network = ValueNetwork()

# 更新策略
states = torch.rand(32, state_dim)  # 示例状态批次
brac_policy.update(states, value_network)

pip install gym torch numpy

"""《BRAC算法的CartPole环境实现》
    时间：2024.09.26
    环境：CartPole 
    作者：不去幼儿园
"""
import gym
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import numpy as np

# 创建 CartPole 环境
env = gym.make('CartPole-v1')

# 生成离线数据
def generate_offline_data(env, policy, num_episodes=100):
	data = []
	for _ in range(num_episodes):
		state, _ = env.reset()
		done = False
		while not done:
			action = policy(torch.FloatTensor(state)).argmax().item()
			next_state, reward, done, _, __ = env.step(action)
			data.append((state, action, reward, next_state, done))
			state = next_state
	return data

# 随机策略用于生成离线数据
class RandomPolicy(nn.Module):
	def forward(self, state):
		return torch.tensor([0.5, 0.5])  # 50% 概率选择任一动作


# 使用随机策略生成离线数据
print("\033[38;5;11mINFO:\033[0m 随机策略生成离线数据", flush=True)
random_policy = RandomPolicy()
offline_data = generate_offline_data(env, random_policy)

# BRAC Policy 网络
class BRACPolicy(nn.Module):
	def __init__(self, state_dim, action_dim, beta_policy, lambda_reg, learning_rate=0.001):
		super(BRACPolicy, self).__init__()
		self.state_dim = state_dim
		self.action_dim = action_dim
		self.beta_policy = beta_policy
		self.lambda_reg = lambda_reg

		# 策略网络结构
		self.policy_network = nn.Sequential(
			nn.Linear(self.state_dim, 64),
			nn.ReLU(),
			nn.Linear(64, 64),
			nn.ReLU(),
			nn.Linear(64, self.action_dim),
			nn.Softmax(dim=-1)
		)

		# 优化器
		self.optimizer = optim.Adam(self.policy_network.parameters(), lr=learning_rate)

	def forward(self, states):
		return self.policy_network(states)

	def update(self, states, value_network):
		# 将状态转换为张量
		states = torch.FloatTensor(states)

		# 计算策略的动作分布
		actions_probs = self.forward(states)

		# 从行为策略获取动作分布
		with torch.no_grad():
			beta_actions_probs = self.beta_policy.get_action_distribution(states)

		# 计算价值损失
		q_values = value_network.predict(states)
		value_loss = -torch.mean(torch.sum(q_values * actions_probs, dim=1))

		# 计算正则化项 (KL散度)
		kl_divergence = F.kl_div(actions_probs.log(), beta_actions_probs, reduction='batchmean')

		# 总损失
		loss = value_loss + self.lambda_reg * kl_divergence

		# 反向传播和优化
		self.optimizer.zero_grad()
		loss.backward()
		self.optimizer.step()

	# 更新BRACPolicy类的get_action_distribution方法，转换列表为numpy array
	def get_action_distribution(self, states):
		states = torch.FloatTensor(np.array(states))  # 转换为 numpy array 再转换为 tensor
		with torch.no_grad():
			return self.forward(states).numpy()


# Value 网络
class ValueNetwork(nn.Module):
	def __init__(self, state_dim, action_dim, learning_rate=0.001):
		super(ValueNetwork, self).__init__()
		self.network = nn.Sequential(
			nn.Linear(state_dim, 64),
			nn.ReLU(),
			nn.Linear(64, 64),
			nn.ReLU(),
			nn.Linear(64, action_dim)
		)
		self.optimizer = optim.Adam(self.network.parameters(), lr=learning_rate)

	def forward(self, states):
		return self.network(states)

	def predict(self, states):
		return self.forward(torch.FloatTensor(states))

	def update(self, states, actions, targets):
		q_values = self.forward(torch.FloatTensor(states))
		q_values = q_values.gather(1, torch.LongTensor(actions).unsqueeze(1)).squeeze(1)
		loss = F.mse_loss(q_values, torch.FloatTensor(targets))

		self.optimizer.zero_grad()
		loss.backward()
		self.optimizer.step()


# 预训练行为策略
class BetaPolicy:
	def get_action_distribution(self, states):
		return torch.rand(states.shape[0], action_dim)


# 初始化网络
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
beta_policy = BetaPolicy()
value_network = ValueNetwork(state_dim, action_dim)
brac_policy = BRACPolicy(state_dim, action_dim, beta_policy, lambda_reg=0.1)


# 训练
def train_brac_policy(data, policy, value_network, num_epochs=100):
	for _ in range(num_epochs):
		for batch in data:
			states, actions, rewards, next_states, dones = zip(*batch)

			# 将列表转换为 numpy arrays
			states = np.array(states)
			next_states = np.array(next_states)
			rewards = np.array(rewards, dtype=np.float32)  # 确保 rewards 为 float 类型
			dones = np.array(dones, dtype=np.float32)  # 确保 dones 为 float 类型

			# 更新价值网络
			q_next = value_network.predict(next_states).max(dim=1)[0].detach()  # 使用 detach() 防止梯度追踪
			targets = rewards + 0.99 * q_next.numpy() * (1 - dones)  # 使用 .numpy() 获取张量的值
			value_network.update(states, actions, targets)

			# 更新策略网络
			policy.update(states, value_network)


# 数据批次处理
batch_size = 32
batches = [offline_data[i:i + batch_size] for i in range(0, len(offline_data), batch_size)]

# 训练策略
print("\033[38;5;11mINFO:\033[0m BRAC算法策略训练中", flush=True)
train_brac_policy(batches, brac_policy, value_network)

# 测试BRAC策略
def test_policy(policy, env, num_episodes=10):
	total_rewards = []
	for _ in range(num_episodes):
		state, _ = env.reset()
		done = False
		episode_reward = 0
		while not done:
			action_probs = policy.get_action_distribution([state])

			# 将 numpy.ndarray 转换为 tensor
			action_probs = torch.FloatTensor(action_probs)

			# 选择具有最大概率的动作
			action = torch.argmax(action_probs).item()
			next_state, reward, done, _, __ = env.step(action)
			episode_reward += reward
			state = next_state
		total_rewards.append(episode_reward)
	return total_rewards


# 测试BRAC策略在 CartPole 上的表现
print("\033[38;5;11mINFO:\033[0m BRAC算法策略测试中", flush=True)
test_rewards = test_policy(brac_policy, env)
print(f"\u001b[38;5;2mSUCCESS:\u001b[0m BRAC策略的平均奖励: {np.mean(test_rewards)}", flush=True)