✅ CPT 强化学习完整实现PyTorch 版 - Actor-Critic CPT以下是生产级友好的实现适合连续/离散控制任务结合Cumulative Prospect Theory修改优势函数Advantage。推荐配置默认使用环境LunarLander-v2连续动作空间难度适中能明显看出 CPT 的风险偏好差异框架PyTorch Gymnasium算法Actor-CriticA2C风格 CPT 价值重塑深度带经验回放 参考点自适应完整代码可直接运行importgymnasiumasgymimporttorchimporttorch.nnasnnimporttorch.optimasoptimimportnumpyasnpfromcollectionsimportdequeimportrandom# CPT 核心函数 classCPT:def__init__(self,alpha0.88,beta0.88,lambda_loss2.25,gamma_gain0.61,gamma_loss0.69,reference0.0):self.alphaalpha self.betabeta self.lambda_losslambda_loss self.gamma_gaingamma_gain self.gamma_lossgamma_loss self.referencereferencedefvalue(self,x):价值函数 v(x)xtorch.tensor(x,dtypetorch.float32)returntorch.where(x0,x**self.alpha,-self.lambda_loss*(-x)**self.beta)defprobability_weight(self,p):概率权重函数ptorch.tensor(p,dtypetorch.float32).clamp(1e-6,1-1e-6)w_gainp**self.gamma_gain/(p**self.gamma_gain(1-p)**self.gamma_gain)**(1/self.gamma_gain)w_lossp**self.gamma_loss/(p**self.gamma_loss(1-p)**self.gamma_loss)**(1/self.gamma_loss)returntorch.where(p0.5,w_gain,w_loss)# 简化defcompute_cpt_advantage(self,rewards,gamma0.99):从 trajectory 计算 CPT 优势returns[]R0forrinreversed(rewards):Rrgamma*R returns.insert(0,R)returnstorch.tensor(returns)cpt_returnsself.value(returns-self.reference)# 标准化cpt_advantages(cpt_returns-cpt_returns.mean())/(cpt_returns.std()1e-8)returncpt_advantages# Actor-Critic 网络 classActorCritic(nn.Module):def__init__(self,state_dim,action_dim,hidden_dim128):super().__init__()self.sharednn.Sequential(nn.Linear(state_dim,hidden_dim),nn.ReLU(),nn.Linear(hidden_dim,hidden_dim),nn.ReLU())# Actor (均值 标准差)self.actor_meannn.Linear(hidden_dim,action_dim)self.actor_logstdnn.Parameter(torch.zeros(action_dim))# Criticself.criticnn.Linear(hidden_dim,1)defforward(self,x):xself.shared(x)meanself.actor_mean(x)stdtorch.exp(self.actor_logstd)valueself.critic(x)returnmean,std,value# CPT Agent classCPTActorCriticAgent:def__init__(self,state_dim,action_dim,lr1e-3,gamma0.99,devicecpu):self.devicedevice self.gammagamma self.cptCPT()self.modelActorCritic(state_dim,action_dim).to(device)self.optimizeroptim.Adam(self.model.parameters(),lrlr)self.memorydeque(maxlen2048)# (state, action, reward, next_state, done)defselect_action(self,state):statetorch.FloatTensor(state).unsqueeze(0).to(self.device)mean,std,_self.model(state)disttorch.distributions.Normal(mean,std)actiondist.sample()actiontorch.clamp(action,-1.0,1.0)# LunarLander 动作范围returnaction.squeeze().cpu().numpy()defupdate(self,batch_size128):iflen(self.memory)batch_size:return0.0batchrandom.sample(self.memory,batch_size)states,actions,rewards,next_states,doneszip(*batch)statestorch.FloatTensor(np.array(states)).to(self.device)actionstorch.FloatTensor(np.array(actions)).to(self.device)rewardstorch.FloatTensor(rewards).to(self.device)next_statestorch.FloatTensor(np.array(next_states)).to(self.device)donestorch.FloatTensor(dones).to(self.device)# CPT Advantageadvantagesself.cpt.compute_cpt_advantage(rewards.tolist())# Forwardmeans,stds,valuesself.model(states)disttorch.distributions.Normal(means,stds.exp())log_probsdist.log_prob(actions).sum(dim-1)# Lossactor_loss-(log_probs*advantages.detach()).mean()critic_loss(values.squeeze()-advantages).pow(2).mean()lossactor_loss0.5*critic_loss self.optimizer.zero_grad()loss.backward()torch.nn.utils.clip_grad_norm_(self.model.parameters(),0.5)self.optimizer.step()returnloss.item()defstore(self,state,action,reward,next_state,done):self.memory.append((state,action,reward,next_state,done))# 训练主函数 deftrain_cpt_rl(episodes800,renderFalse):envgym.make(LunarLanderContinuous-v2,render_modehumanifrenderelseNone)state_dimenv.observation_space.shape[0]# 8action_dimenv.action_space.shape[0]# 2agentCPTActorCriticAgent(state_dim,action_dim,devicecudaiftorch.cuda.is_available()elsecpu)best_reward-float(inf)forepisodeinrange(episodes):state,_env.reset()episode_rewards[]doneFalsetotal_reward0whilenotdone:actionagent.select_action(state)next_state,reward,terminated,truncated,_env.step(action)doneterminatedortruncated agent.store(state,action,reward,next_state,done)statenext_state total_rewardreward episode_rewards.append(reward)iflen(agent.memory)256:lossagent.update()# 自适应参照点可选iftotal_rewardbest_reward:best_rewardtotal_reward agent.cpt.referencenp.mean(episode_rewards)*0.3ifepisode%500:print(fEpisode{episode:4d}| Reward:{total_reward:8.2f}| fBest:{best_reward:.2f}| Ref Point:{agent.cpt.reference:.2f})env.close()returnagentif__name____main__:agenttrain_cpt_rl(episodes1000)使用说明安装依赖pipinstallgymnasium[box2d]torch numpy运行直接执行即可看到 LunarLander 训练过程。CPT 关键影响损失厌恶λ2.25→ Agent 更倾向于避免大风险坠毁即使牺牲部分期望得分。可通过调整lambda_loss观察行为变化越大越保守。需要我调整吗切换到PPO版本更稳定使用Discrete 环境CartPole / FrozenLake增加经验回放 Prioritized Replay多环境并行训练VectorEnvCPT 在 DQN / SAC 中的实现
CPT 强化学习完整实现（PyTorch 版 - Actor-Critic + CPT）
发布时间：2026/5/24 22:10:07
✅ CPT 强化学习完整实现PyTorch 版 - Actor-Critic CPT以下是生产级友好的实现适合连续/离散控制任务结合Cumulative Prospect Theory修改优势函数Advantage。推荐配置默认使用环境LunarLander-v2连续动作空间难度适中能明显看出 CPT 的风险偏好差异框架PyTorch Gymnasium算法Actor-CriticA2C风格 CPT 价值重塑深度带经验回放 参考点自适应完整代码可直接运行importgymnasiumasgymimporttorchimporttorch.nnasnnimporttorch.optimasoptimimportnumpyasnpfromcollectionsimportdequeimportrandom# CPT 核心函数 classCPT:def__init__(self,alpha0.88,beta0.88,lambda_loss2.25,gamma_gain0.61,gamma_loss0.69,reference0.0):self.alphaalpha self.betabeta self.lambda_losslambda_loss self.gamma_gaingamma_gain self.gamma_lossgamma_loss self.referencereferencedefvalue(self,x):价值函数 v(x)xtorch.tensor(x,dtypetorch.float32)returntorch.where(x0,x**self.alpha,-self.lambda_loss*(-x)**self.beta)defprobability_weight(self,p):概率权重函数ptorch.tensor(p,dtypetorch.float32).clamp(1e-6,1-1e-6)w_gainp**self.gamma_gain/(p**self.gamma_gain(1-p)**self.gamma_gain)**(1/self.gamma_gain)w_lossp**self.gamma_loss/(p**self.gamma_loss(1-p)**self.gamma_loss)**(1/self.gamma_loss)returntorch.where(p0.5,w_gain,w_loss)# 简化defcompute_cpt_advantage(self,rewards,gamma0.99):从 trajectory 计算 CPT 优势returns[]R0forrinreversed(rewards):Rrgamma*R returns.insert(0,R)returnstorch.tensor(returns)cpt_returnsself.value(returns-self.reference)# 标准化cpt_advantages(cpt_returns-cpt_returns.mean())/(cpt_returns.std()1e-8)returncpt_advantages# Actor-Critic 网络 classActorCritic(nn.Module):def__init__(self,state_dim,action_dim,hidden_dim128):super().__init__()self.sharednn.Sequential(nn.Linear(state_dim,hidden_dim),nn.ReLU(),nn.Linear(hidden_dim,hidden_dim),nn.ReLU())# Actor (均值 标准差)self.actor_meannn.Linear(hidden_dim,action_dim)self.actor_logstdnn.Parameter(torch.zeros(action_dim))# Criticself.criticnn.Linear(hidden_dim,1)defforward(self,x):xself.shared(x)meanself.actor_mean(x)stdtorch.exp(self.actor_logstd)valueself.critic(x)returnmean,std,value# CPT Agent classCPTActorCriticAgent:def__init__(self,state_dim,action_dim,lr1e-3,gamma0.99,devicecpu):self.devicedevice self.gammagamma self.cptCPT()self.modelActorCritic(state_dim,action_dim).to(device)self.optimizeroptim.Adam(self.model.parameters(),lrlr)self.memorydeque(maxlen2048)# (state, action, reward, next_state, done)defselect_action(self,state):statetorch.FloatTensor(state).unsqueeze(0).to(self.device)mean,std,_self.model(state)disttorch.distributions.Normal(mean,std)actiondist.sample()actiontorch.clamp(action,-1.0,1.0)# LunarLander 动作范围returnaction.squeeze().cpu().numpy()defupdate(self,batch_size128):iflen(self.memory)batch_size:return0.0batchrandom.sample(self.memory,batch_size)states,actions,rewards,next_states,doneszip(*batch)statestorch.FloatTensor(np.array(states)).to(self.device)actionstorch.FloatTensor(np.array(actions)).to(self.device)rewardstorch.FloatTensor(rewards).to(self.device)next_statestorch.FloatTensor(np.array(next_states)).to(self.device)donestorch.FloatTensor(dones).to(self.device)# CPT Advantageadvantagesself.cpt.compute_cpt_advantage(rewards.tolist())# Forwardmeans,stds,valuesself.model(states)disttorch.distributions.Normal(means,stds.exp())log_probsdist.log_prob(actions).sum(dim-1)# Lossactor_loss-(log_probs*advantages.detach()).mean()critic_loss(values.squeeze()-advantages).pow(2).mean()lossactor_loss0.5*critic_loss self.optimizer.zero_grad()loss.backward()torch.nn.utils.clip_grad_norm_(self.model.parameters(),0.5)self.optimizer.step()returnloss.item()defstore(self,state,action,reward,next_state,done):self.memory.append((state,action,reward,next_state,done))# 训练主函数 deftrain_cpt_rl(episodes800,renderFalse):envgym.make(LunarLanderContinuous-v2,render_modehumanifrenderelseNone)state_dimenv.observation_space.shape[0]# 8action_dimenv.action_space.shape[0]# 2agentCPTActorCriticAgent(state_dim,action_dim,devicecudaiftorch.cuda.is_available()elsecpu)best_reward-float(inf)forepisodeinrange(episodes):state,_env.reset()episode_rewards[]doneFalsetotal_reward0whilenotdone:actionagent.select_action(state)next_state,reward,terminated,truncated,_env.step(action)doneterminatedortruncated agent.store(state,action,reward,next_state,done)statenext_state total_rewardreward episode_rewards.append(reward)iflen(agent.memory)256:lossagent.update()# 自适应参照点可选iftotal_rewardbest_reward:best_rewardtotal_reward agent.cpt.referencenp.mean(episode_rewards)*0.3ifepisode%500:print(fEpisode{episode:4d}| Reward:{total_reward:8.2f}| fBest:{best_reward:.2f}| Ref Point:{agent.cpt.reference:.2f})env.close()returnagentif__name____main__:agenttrain_cpt_rl(episodes1000)使用说明安装依赖pipinstallgymnasium[box2d]torch numpy运行直接执行即可看到 LunarLander 训练过程。CPT 关键影响损失厌恶λ2.25→ Agent 更倾向于避免大风险坠毁即使牺牲部分期望得分。可通过调整lambda_loss观察行为变化越大越保守。需要我调整吗切换到PPO版本更稳定使用Discrete 环境CartPole / FrozenLake增加经验回放 Prioritized Replay多环境并行训练VectorEnvCPT 在 DQN / SAC 中的实现
