006_决策模块的安全强化学习：防范策略中毒与保障自主安全

# 2025年差分隐私安全强化学习示例（基于PyTorch）
import torch
import numpy as np

class DPSafeRL:
    def __init__(self, model, privacy_budget=1.0, epsilon=0.1, delta=1e-5):
        self.model = model
        self.privacy_budget = privacy_budget
        self.epsilon = epsilon
        self.delta = delta
        self.optimizer = torch.optim.Adam(model.parameters())
    
    def add_noise_to_gradients(self, sensitivity=1.0):
        # 计算梯度敏感度
        for param in self.model.parameters():
            if param.grad is not None:
                # 添加拉普拉斯噪声
                noise = np.random.laplace(0, sensitivity/self.epsilon, size=param.grad.shape)
                param.grad += torch.from_numpy(noise).float().to(param.grad.device)
    
    def train_step(self, states, actions, rewards, next_states, dones):
        # 前向传播
        q_values = self.model(states)
        target_q = self._compute_targets(rewards, next_states, dones)
        
        # 计算损失
        loss = torch.nn.functional.mse_loss(q_values.gather(1, actions.unsqueeze(1)), target_q)
        
        # 反向传播
        self.optimizer.zero_grad()
        loss.backward()
        
        # 添加差分隐私噪声
        self.add_noise_to_gradients()
        
        # 梯度裁剪防止梯度爆炸
        torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0)
        
        # 参数更新
        self.optimizer.step()
        
        return loss.item()

# 2025年贝叶斯安全强化学习示例（基于PyTorch和Pyro）
import torch
import pyro
import pyro.distributions as dist
from pyro.nn import PyroModule, PyroSample

class BayesianSafeRLAgent:
    def __init__(self, state_dim, action_dim, config):
        self.state_dim = state_dim
        self.action_dim = action_dim
        self.config = config
        
        # 贝叶斯策略网络
        self.policy_network = BayesianPolicyNetwork(
            state_dim, 
            action_dim, 
            hidden_sizes=config['hidden_sizes'],
            prior_scale=config['prior_scale']
        )
        
        # 不确定性估计器
        self.uncertainty_estimator = UncertaintyEstimator(config['uncertainty_config'])
        
        # 安全约束模块
        self.safety_constraint = SafetyConstraint(
            threshold=config['safety_threshold'],
            penalty_weight=config['penalty_weight']
        )
        
        # 优化器
        self.optimizer = torch.optim.Adam(
            self.policy_network.parameters(),
            lr=config['learning_rate']
        )
    
    def select_action(self, state, exploration=True):
        """基于贝叶斯推断选择安全动作"""
        state_tensor = torch.FloatTensor(state).unsqueeze(0)
        
        # 获取后验分布
        with pyro.plate("model_sample"):
            action_dist = self.policy_network(state_tensor)
        
        # 不确定性估计
        uncertainty = self.uncertainty_estimator.estimate(
            state_tensor, 
            action_dist
        )
        
        # 安全约束过滤
        if exploration:
            # 探索模式：考虑不确定性
            actions = []
            uncertainties = []
            
            # 采样多个可能的动作
            for _ in range(self.config['num_action_samples']):
                action = action_dist.sample()
                action_uncertainty = self.uncertainty_estimator.estimate_action_uncertainty(
                    state_tensor, action
                )
                actions.append(action)
                uncertainties.append(action_uncertainty)
            
            # 选择安全且不确定性适当的动作
            safe_actions = []
            for action, unc in zip(actions, uncertainties):
                if self.safety_constraint.is_safe(state_tensor, action, unc):
                    safe_actions.append((action, unc))
            
            if safe_actions:
                # 在安全动作中，选择不确定性与探索参数匹配的动作
                exploration_factor = self.config.get('exploration_factor', 0.5)
                target_uncertainty = exploration_factor * max([unc for _, unc in safe_actions])
                
                # 找到最接近目标不确定性的动作
                selected_action, _ = min(
                    safe_actions, 
                    key=lambda x: abs(x[1] - target_uncertainty)
                )
                return selected_action.detach().numpy()[0]
        
        # 默认策略：选择均值动作并检查安全性
        mean_action = action_dist.mean
        if self.safety_constraint.is_safe(state_tensor, mean_action, uncertainty):
            return mean_action.detach().numpy()[0]
        else:
            # 安全回退动作
            return self.safety_constraint.get_safe_action(state_tensor).detach().numpy()[0]
    
    def train_step(self, batch):
        """训练贝叶斯安全策略网络"""
        states, actions, rewards, next_states, dones = batch
        
        # 重置Pyro参数存储
        pyro.clear_param_store()
        
        # 变分推断优化
        def model():
            # 先验分布
            pyro.sample("action_preds", self.policy_network(states))
        
        def guide():
            # 后验分布
            action_dist = self.policy_network(states)
            pyro.sample("action_preds", action_dist)
        
        # 计算变分下界损失
        elbo = pyro.infer.Trace_ELBO()
        loss = elbo.loss(model, guide)
        
        # 添加安全约束损失
        for i in range(len(states)):
            state = states[i].unsqueeze(0)
            action = actions[i].unsqueeze(0)
            
            # 估计不确定性
            action_dist = self.policy_network(state)
            uncertainty = self.uncertainty_estimator.estimate(state, action_dist)
            
            # 添加安全约束损失
            safety_loss = self.safety_constraint.compute_loss(state, action, uncertainty)
            loss += safety_loss
        
        # 优化
        self.optimizer.zero_grad()
        loss.backward()
        torch.nn.utils.clip_grad_norm_(self.policy_network.parameters(), 1.0)
        self.optimizer.step()
        
        return loss.item()

# 贝叶斯策略网络
class BayesianPolicyNetwork(PyroModule):
    def __init__(self, input_dim, output_dim, hidden_sizes=[64, 64], prior_scale=0.1):
        super().__init__()
        
        self.layers = PyroModule()
        self.layers.fc1 = PyroModule[torch.nn.Linear](input_dim, hidden_sizes[0])
        self.layers.fc2 = PyroModule[torch.nn.Linear](hidden_sizes[0], hidden_sizes[1])
        self.layers.fc_mean = PyroModule[torch.nn.Linear](hidden_sizes[1], output_dim)
        self.layers.fc_log_std = PyroModule[torch.nn.Linear](hidden_sizes[1], output_dim)
        
        # 权重的先验分布
        self.layers.fc1.weight = PyroSample(
            dist.Normal(0., prior_scale).expand([hidden_sizes[0], input_dim]).to_event(2)
        )
        self.layers.fc1.bias = PyroSample(
            dist.Normal(0., prior_scale).expand([hidden_sizes[0]]).to_event(1)
        )
        
        self.layers.fc2.weight = PyroSample(
            dist.Normal(0., prior_scale).expand([hidden_sizes[1], hidden_sizes[0]]).to_event(2)
        )
        self.layers.fc2.bias = PyroSample(
            dist.Normal(0., prior_scale).expand([hidden_sizes[1]]).to_event(1)
        )
        
        self.layers.fc_mean.weight = PyroSample(
            dist.Normal(0., prior_scale).expand([output_dim, hidden_sizes[1]]).to_event(2)
        )
        self.layers.fc_mean.bias = PyroSample(
            dist.Normal(0., prior_scale).expand([output_dim]).to_event(1)
        )
        
        self.layers.fc_log_std.weight = PyroSample(
            dist.Normal(0., prior_scale).expand([output_dim, hidden_sizes[1]]).to_event(2)
        )
        self.layers.fc_log_std.bias = PyroSample(
            dist.Normal(0., prior_scale).expand([output_dim]).to_event(1)
        )
        
        self.activation = torch.nn.ReLU()
    
    def forward(self, x):
        x = self.activation(self.layers.fc1(x))
        x = self.activation(self.layers.fc2(x))
        
        mean = self.layers.fc_mean(x)
        log_std = self.layers.fc_log_std(x)
        # 限制标准差范围
        log_std = torch.clamp(log_std, -20, 2)
        std = torch.exp(log_std)
        
        # 返回动作分布
        return dist.Normal(mean, std)

# 不确定性估计器
class UncertaintyEstimator:
    def __init__(self, config):
        self.config = config
    
    def estimate(self, state, action_dist):
        # 基于动作分布计算不确定性
        entropy = action_dist.entropy().mean()
        return entropy
    
    def estimate_action_uncertainty(self, state, action):
        # 估计特定动作的不确定性
        # 这是一个简化实现，实际中可能需要更复杂的方法
        return torch.tensor(0.5)  # 示例值

# 安全约束模块
class SafetyConstraint:
    def __init__(self, threshold=0.1, penalty_weight=10.0):
        self.threshold = threshold
        self.penalty_weight = penalty_weight
        # 安全区域定义（简化示例）
        self.safe_region = lambda state, action: torch.norm(action) < 1.0
    
    def is_safe(self, state, action, uncertainty):
        # 检查动作是否安全，同时考虑不确定性
        base_safety = self.safe_region(state, action)
        uncertainty_factor = torch.exp(-self.penalty_weight * uncertainty)
        return base_safety and uncertainty_factor > self.threshold
    
    def compute_loss(self, state, action, uncertainty):
        # 计算安全约束损失
        safety_violation = torch.max(torch.tensor(0.0), 1.0 - self.safe_region(state, action))
        uncertainty_penalty = uncertainty * self.penalty_weight
        return safety_violation + uncertainty_penalty
    
    def get_safe_action(self, state):
        # 返回安全的默认动作
        return torch.zeros(state.shape[0], 2)  # 示例：零动作

# 鲁棒强化学习示例
class RobustRLAgent:
    def __init__(self, state_dim, action_dim, config):
        self.state_dim = state_dim
        self.action_dim = action_dim
        self.config = config
        
        # 主策略网络
        self.policy_network = PolicyNetwork(state_dim, action_dim, config['hidden_sizes'])
        
        # 对手网络（用于生成对抗样本）
        self.adversary_network = AdversaryNetwork(state_dim, config['adv_config'])
        
        # 优化器
        self.policy_optimizer = torch.optim.Adam(
            self.policy_network.parameters(),
            lr=config['policy_lr']
        )
        self.adversary_optimizer = torch.optim.Adam(
            self.adversary_network.parameters(),
            lr=config['adv_lr']
        )
    
    def train_step(self, batch):
        states, actions, rewards, next_states, dones = batch
        
        # 1. 对手网络优化：生成最坏情况的状态扰动
        self.adversary_optimizer.zero_grad()
        
        # 生成对抗扰动
        adv_perturbations = self.adversary_network(states)
        # 限制扰动幅度
        adv_perturbations = torch.clamp(adv_perturbations, -self.config['epsilon'], self.config['epsilon'])
        
        # 生成对抗状态
        adv_states = states + adv_perturbations
        
        # 计算对手损失（最大化策略损失）
        adv_actions = self.policy_network(adv_states)
        adv_policy_loss = -self._compute_policy_loss(adv_states, adv_actions, rewards, next_states, dones)
        adv_policy_loss.backward()
        self.adversary_optimizer.step()
        
        # 2. 策略网络优化：在对抗状态上训练
        self.policy_optimizer.zero_grad()
        
        # 重新生成对抗扰动（使用更新后的对手网络）
        with torch.no_grad():
            adv_perturbations = self.adversary_network(states)
            adv_perturbations = torch.clamp(adv_perturbations, -self.config['epsilon'], self.config['epsilon'])
            adv_states = states + adv_perturbations
        
        # 在正常状态和对抗状态上训练
        normal_actions = self.policy_network(states)
        adv_actions = self.policy_network(adv_states)
        
        normal_loss = self._compute_policy_loss(states, normal_actions, rewards, next_states, dones)
        adv_loss = self._compute_policy_loss(adv_states, adv_actions, rewards, next_states, dones)
        
        # 组合损失
        total_loss = normal_loss + self.config['adv_weight'] * adv_loss
        total_loss.backward()
        self.policy_optimizer.step()
        
        return {
            'total_loss': total_loss.item(),
            'normal_loss': normal_loss.item(),
            'adv_loss': adv_loss.item()
        }
    
    def _compute_policy_loss(self, states, actions, rewards, next_states, dones):
        # 简化的策略损失计算
        # 实际应用中可能使用PPO、SAC等算法的损失函数
        return torch.tensor(0.0)  # 示例返回值

def robust_gradient_aggregation(gradients, threshold=2.0):
    """鲁棒梯度聚合，过滤异常梯度"""
    # 计算梯度均值
    mean_grad = torch.mean(torch.stack(gradients), dim=0)
    
    # 计算每个梯度与均值的距离
    distances = [torch.norm(g - mean_grad) for g in gradients]
    
    # 过滤异常梯度
    filtered_gradients = [g for g, d in zip(gradients, distances) if d < threshold * torch.mean(torch.tensor(distances))]
    
    # 返回过滤后的梯度均值
    return torch.mean(torch.stack(filtered_gradients), dim=0)

class ControlBarrierFunction:
    def __init__(self, robot_params, safety_margin=0.1):
        self.robot_params = robot_params
        self.safety_margin = safety_margin
        
    def h(self, state, obstacle_pos, obstacle_radius):
        """计算安全屏障函数值"""
        robot_pos = state[:2]
        distance = torch.norm(robot_pos - obstacle_pos)
        # 安全距离 = 机器人半径 + 障碍物半径 + 安全余量
        safe_distance = self.robot_params['radius'] + obstacle_radius + self.safety_margin
        return distance - safe_distance
    
    def grad_h(self, state, obstacle_pos):
        """计算安全屏障函数梯度"""
        robot_pos = state[:2]
        diff = robot_pos - obstacle_pos
        distance = torch.norm(diff)
        if distance > 1e-6:
            return torch.cat([diff / distance, torch.zeros_like(state[2:])])
        else:
            return torch.zeros_like(state)
    
    def get_safe_control(self, state, nominal_control, obstacle_pos, obstacle_radius):
        """计算满足安全约束的控制输入"""
        h_val = self.h(state, obstacle_pos, obstacle_radius)
        grad_h_val = self.grad_h(state, obstacle_pos)
        
        # 如果当前状态在安全区域，返回标称控制
        if h_val > 0 and grad_h_val @ nominal_control > -self.robot_params['gamma'] * h_val:
            return nominal_control
        
        # 否则，求解QP问题找到安全控制
        # 简化实现：投影到安全方向
        control_dim = nominal_control.shape[0]
        A = grad_h_val.unsqueeze(0)
        b = -self.robot_params['gamma'] * h_val
        
        # 计算投影矩阵
        I = torch.eye(control_dim)
        P = I - (A.T @ torch.inverse(A @ A.T) @ A) if (A @ A.T).det() > 1e-6 else I
        
        # 投影标称控制
        projected_control = P @ nominal_control
        
        # 计算补充项确保满足约束
        if A @ projected_control < b:
           补充项 = (b - A @ projected_control) / (A @ A.T + 1e-6) * A.T
            projected_control += 补充项
        
        return projected_control