CPT 强化学习(Cumulative Prospect Theory Reinforcement Learning)代码实现
✅ CPT 强化学习(Cumulative Prospect Theory Reinforcement Learning)代码实现
以下提供实用、可运行的 Python 实现,结合Cumulative Prospect Theory (CPT)与强化学习。
1. 核心概念回顾
在传统 RL 中,目标是最大化期望回报(Expected Return)。
在CPT-RL中,目标是最大化 CPT 值(考虑损失厌恶、概率扭曲、参照点),让 Agent 表现出人类-like 的风险偏好(风险规避/寻求)。
2. 完整代码实现(简单版:CPT-Q Learning)
importgymnasiumasgymimportnumpyasnpfromcollectionsimportdefaultdictimportrandom# ====================== CPT 核心函数 ======================defcpt_value(x,alpha=0.88,beta=0.88,lambda_loss=2.25):"""价值函数 v(x)"""x=np.array(x)returnnp.where(x>=0,x**alpha,-lambda_loss*(-x)**beta)defcpt_probability_weight(p,gamma_gain=0.61,gamma_loss=0.69):"""概率权重函数 w(p)"""p=np.array(p)# 简单实现:不同 gamma 用于收益和损失w_gain=p**gamma_gain/(p**gamma_gain+(1-p)**gamma_gain)**(1/gamma_gain)w_loss=p**gamma_loss/(p**gamma_loss+(1-p)**gamma_loss)**(1/gamma_loss)returnnp.where(p>=0,w_gain,w_loss)# 简化处理defestimate_cpt_value(rewards,alpha=0.88,beta=0.88,lambda_loss=2.25,gamma_gain=0.61,gamma_loss=0.69,reference=0.0):"""从多条 trajectory 的回报样本估计 CPT 值"""returns=np.array(rewards)-reference sorted_returns=np.sort(returns)n=len(sorted_returns)# 计算决策权重(累积形式简化版)probs=np.ones(n)/n weights=cpt_probability_weight(probs)# 累积权重(简化实现)cum_weights=np.cumsum(weights[::-1])[::-1]# 粗略近似values=cpt_value(sorted_returns)cpt_val=np.sum(cum_weights*values)/n# 归一化returncpt_val# ====================== CPT-Q Learning ======================classCPTQAgent:def__init__(self,state_size,action_size,alpha=0.1,gamma=0.99,epsilon=0.1,cpt_alpha=0.88,cpt_lambda=2.25):self.q_table=defaultdict(lambda:np.zeros(action_size))self.alpha=alpha# 学习率self.gamma=gamma# 折扣因子self.epsilon=epsilon# 探索率self.action_size=action_size self.cpt_alpha=cpt_alpha self.cpt_lambda=cpt_lambda self.memory=[]# 存储 trajectory returns 用于 CPT 估计defchoose_action(self,state):ifrandom.uniform(0,1)<self.epsilon:returnrandom.randint(0,self.action_size-1)returnnp.argmax(self.q_table[state])deflearn(self,state,action,reward,next_state,done):# 标准 Q-Learning 更新(可替换为 CPT 目标)old_value=self.q_table[state][action]next_max=np.max(self.q_table[next_state])# 传统 TD Targettd_target=reward+self.gamma*next_max*(1-done)self.q_table[state][action]=old_value+self.alpha*(td_target-old_value)ifdone:self.memory.append(reward)# 简化:记录 episode 回报# ====================== 训练示例 ======================deftrain_cpt_rl(env_name="CartPole-v1",episodes=500):env=gym.make(env_name)state_size=env.observation_space.shape[0]# 连续状态需离散化或用近似# 为简单起见,这里使用离散环境示例(如 FrozenLake)# 推荐使用简单离散环境演示env=gym.make("FrozenLake-v1",is_slippery=False)agent=CPTQAgent(state_size=env.observation_space.n,action_size=env.action_space.n)rewards_history=[]forepisodeinrange(episodes):state,_=env.reset()total_reward=0done=Falsewhilenotdone:action=agent.choose_action(state)next_state,reward,terminated,truncated,_=env.step(action)done=terminatedortruncated agent.learn(state,action,reward,next_state,done)state=next_state total_reward+=reward rewards_history.append(total_reward)ifepisode%100==0:print(f"Episode{episode}, Total Reward:{total_reward}")# 演示 CPT 值估计sample_returns=rewards_history[-50:]# 最近50个 episodecpt_val=estimate_cpt_value(sample_returns)print(f"\n最后50个episode的 CPT 值估计:{cpt_val:.4f}")returnagent,rewards_historyif__name__=="__main__":agent,history=train_cpt_rl(episodes=1000)3. 进阶:Policy Gradient + CPT(概念代码)
defcpt_policy_gradient_loss(returns,baseline=0.0):"""CPT 目标下的 Policy Gradient Loss"""advantages=np.array(returns)-baseline values=cpt_value(advantages)# 在实际 PG 中,用 values 替代 advantages 进行梯度更新return-np.mean(values)# 最大化 CPT 值4. 使用建议与扩展
- 离散 vs 连续:上面代码适合离散环境(如 FrozenLake、Taxi)。连续环境(如 CartPole)建议结合 Actor-Critic + CPT 价值估计。
- 更精确的 CPT 估计:参考 Prashanth et al. (2016) 的分位数方法或累积权重精确计算。
- 框架推荐:
- Stable-Baselines3 / CleanRL + 自定义 Reward Shaping
- PyTorch/TensorFlow 实现 Policy Gradient 版本(2024 论文有最新 PG 定理)
- 参数调优:
lambda_loss=2.25→ 更强的损失厌恶gamma_gain < 1→ 概率扭曲
想让我继续提供:
- 完整 PyTorch Actor-Critic CPT 版本(推荐生产级)
- 特定论文算法复现(如 CPT-SPSA 或 Policy Gradient)
- 在 Gymnasium 连续控制任务上的实现
- 与标准 RL 的对比实验代码
