Universal Value Function Approximators 论文阅读（强化学习，迁移？）

前言

Universal Value Function Approximators 个人实现（请大佬指正）

*关于UVFA如何迁移的问题，这也是我为什么反复看这篇文章的原因，我觉值函数逼近的最大用法就是如何迁移，如果仅仅是更改值函数的结构，这没有太大意义。但是从面前理解来看并没有回答好如何迁移这个问题 我想把思路给各位学者分享也请给我学者为我指正

Universal Value Function Approximators这篇论文我分别两次去阅读它，以下是两次大致的理解：

第一次：文章中心意思是训练一个网络作为一个奖励生成器（有点像装配网络中的SVM）但是呢，这个生成器呢是用来代替价值网络的又不是显示的外部环境生成器（就不像装配网络中那个svm）。我觉着难点是如何去训练，输入它说了是环境s和目标g（这里以棋盘为例估计就是xy坐标s和当前距离g）（用有监督或者强化学习框架（强化学习框架获得与我想法类似）），怎么取获取这些数据，复杂的环境又该怎么办。还有这个训练好如何和强化学习嵌入也是的麻烦的事？

文章注重了价值函数的作用，但是如何获得是个麻烦事

获得之后，他是提取了这个UVF用于迁移和嵌入指导这和我论文想法类似

第二次：这篇论文确实想训练一个和svm类似的东西，其输入是状态g和伪奖励g（是由子目标提供的）。但是训练好后是想将UVF嵌入DQN这类强化学习主框架的，如何嵌入？目前看是生成和子目标相关的价值函数V来更新UVF本身，然后构建关于子目标的与V相关的Q来更新主网络DQN。(但是如何与DQN结合呢？）

   两次独立的阅读，我都把UVF看成了独立于经典（比如DQN）以外的模块，这是因为它提供了完整的监督学习的路子，和不完整的强化学习训练UVF的描述以及可以进行迁移学习的描述（但都没有给完整的介绍，所以很多细节靠猜），这就让我认为UVF是个独立的模块可以用监督学习和强化学习来训练它，然后再嵌入回强化学习以此来进行状态迁移。后面反应过来不是这样的,因为我在全文实在看不到除了文章新定义的Q(s, a, g)外的其它Q(s, a,),所以我决定跳出把他作为单独模块来看的思路。

思路的改变需要注意以下几条问题的澄清:

1、$Q(s,a,g)$就是UVFA为整个强化学习构建的而不是它自己，UVFA也在共享，我们全文以DQN为例：

$$DQ{N}_{UVFA}=(r+\gamma \cdot ma{x}_a^{\prime }Q(s^{\prime },a^{\prime },g)-Q(s,a,g){)}^2$$
传统的DQN更新:
$$DQN=(r+\gamma \cdot ma{x}_a^{\prime }Q(s^{\prime },a^{\prime })-Q(s,a){)}^2$$
   从第一条信息可以看出，如果将UVFA看做DQN的一部分，其实问题就简单了，只不过是引入了一个关于g的奖励来更新Q（后面我会根据个人想法来给出代码段）。

2、 澄清下文中提到的第一个是通过监督学习进行学习的classical grid-world with 4 room也就是图4来探讨的下文章的实验过程。图4中的左图和中图似乎是UVFA输出的可视化？按道理UVFA输出的是一个具体的值，我不太清楚这是如何可视化的（问题1）。中文提到UVFA的输入是状态s(在这里我觉是grid-world with 4 room的坐标）和目标g（应该是图像中的坐标），那么grid-world with 4 room这个实验中subgoal是在像素中随机取点吗？既然有subgoal那么全局的目标又是什么，两者什么关系，还有图4中提到了subgoal后文也提到了UVFA的迁移，在grid-world with 4 room怎么实现迁移，从哪里迁移到哪里（问题2）？

  问题1是对于实验设置的问题 $DQ{N}_{UVFA}$需要状态$s$和$g$，后者文中明确说明是网格坐标点，前者也提到是坐标点，但我很疑惑如果s是具体坐标如何是智能体知道图中的墙壁和门？所以我只能把他理解为用可以表示坐标的矩阵。可视化的问题，我有两个猜测可能是输出的V(s,g)是与grid-world同等尺寸的矩阵，矩阵中的值做成了可视化的热力图，然后最大化V转为Q(s,a,g) = r(s,a) + γ·（V(s’,g))最大（可能性小）；还有一种理解。在 4-room grid-world 里，状态 s 和目标 g 都对应网格中的一个方格坐标。UVFA 给出的输出是一个标量价值 (V(s,g))。在 4-room grid-world 里，状态 s 和目标 g 都对应网格中的一个方格坐标。可视化的做法就是：

1. 固定一个目标 g*（比如右上角的某个方格），
2. 对所有可能的状态 s （即每一个方格）依次输入网络，得到对应的$V_{\theta }(s,g^{\ast })$，
3. 把这些标量值填回到对应的方格里，用热力图（colormap）或灰度图显示。
   • 左图：真值函数$V_{\mathrm{true}}(s,g^{\ast })$
   • 中图：拟合出的$V_{\theta }(s,g^{\ast })$

  问题2 关于UVFA如何迁移的问题，这也是我为什么反复看这篇文章的原因，我觉值函数逼近的最大用法就是如何迁移，如果仅仅是更改值函数的结构，这没有太大意义。 但是从面前理解来看并没有回答好如何迁移这个问题。在 UVFA 里，「迁移（transfer）」其实就是利用在一组目标上学到的知识，来帮助对另一组目标（甚至是完全没见过的目标）做出合理的价值估计或更快地学习——这在强化学习里就对应于“零样本”或“少样本”地对新任务的泛化/适应。这一点与文中的泛化性测试（文中图8）很像，至少我目前看不到不同。这与我想象的迁移工作还是很有很大差距的（或许可以迁移到房间更多的地图上，如果仅仅是目标点的重新设定这个种似乎谈不上是迁移）。另外还有子目标和全局目标的问题，文中的正文中并没有提到两者的联系，甚至是混用的，比如图4中出现了subgoal但是正文中的描述还是g（目标），这其实和迁移的理解是有关联的。如果仅仅是泛化性测试（文中图8）的所谓的‘迁移’，那么subgoal与g就是一个东西所以应该作者的初衷是可以拓展的，subgoal是当前的目标点，全局目标就是更远的更复杂的坐标点（但是目前没有看到）。

3、论文中还提到了用强化学习来训练UVFA，这里我有个问题UVFA本身就是嵌入在强化学习算法中，比如说在论文附件中使用DQN来玩Ms Pacman，所以我的理解是Reinforcement Learning Experiments用强化学习来训练UVFA可以任务是用具有UVFA的DQN在MS Pacman这个游戏上做测试，不知道我的理解里否正确

  所以"用强化学习来训练UVFA"实际上是指：使用标准的强化学习算法（如DQN）来训练具有UVFA架构的价值网络，而不是先用监督学习训练UVFA然后再插入到强化学习算法中。
在这种情况下，UVFA不是独立于强化学习算法的组件，而是被整合到算法中，成为其核心部分。这使得智能体可以在与环境交互的过程中直接学习通用价值函数，而无需预先计算真实价值。（这里是对我前两次阅读的思路的澄清）

4、按我对MS Pacman游戏应的规则和具有UVFA的DQN的理解，这个实验的设置应该是应该是Ms. Pac-Man躲避敌人吃尽可能多的豆子设置一个关于豆子数量的得分相关的奖励，在每次执行一个动作后（这里应该是上下左右）计算一次这个奖励，最后是奖励换算成Q值求Q值最大。但是在用具有UVFA的DQN在MS Pacman这个游戏上测试的过程由于引入了子目标和目标gi，文中还额为提到When the subgoal is achieved, the environment is reset, Pacman’s position is set to a random location, and a new episode begins. 这让我有点无法与“我对MS Pacman游戏应的规则和具有UVFA的DQN的理解”相互对应。（用具有UVFA的DQN在MS Pacman这个游戏上做测试，这个实验过程我有一点疑问，MS Pacman游戏应的规则应该是Ms. Pac-Man躲避敌人吃尽可能多的豆子，但是在论文附件中提到Each demon uses a variant of Deep Q-Learning with … to learn the value functions with respect to its subgoal. 这里的demon 是指敌人吗还是Pac-man如果是pac-man不是一局游戏只有一个吗。而且这里又提到了子目标，请问这里的子目标和目标gi的关系是什么，在更新Q值时除了pseudo-reward还有其他reward吗）

   这也是我前两次无解的原因之一，因为我一直以为这个实验的最终目的是吃尽可能多的豆子（每个"demon"是一个负责学习特定子目标价值函数的独立DQN实例并不是游戏中的敌人，也不是多个Pac-Man (游戏中确实只有一个Pac-Man角色)），而伪奖励(Pseudo-reward)只是子任务，会有进行迁移然后完成最终任务，但是我自始至终没有看到相关描述，所以这里的迁移很可能只是测试验证泛化性的那种意思！
   MS Pacman的游戏的设定在这里也改了:
在标准Ms. Pacman游戏中：

• Pac-Man需要吃掉所有豆子同时避开敌人
• 游戏持续到Pac-Man失去所有生命或清空关卡
• 奖励基于吃豆子、能量豆和幽灵获得的分数

但在UVFA的实验设置中，研究者做了重要修改：

1. 单一子目标训练：

   • 每个训练回合只关注一个特定豆子位置(子目标)
   • 当Pac-Man达到该豆子位置时，回合结束并重置
   • 这与标准游戏规则不同，标准游戏中吃到一个豆子后游戏继续

2. 目标是学习到达特定位置的能力：

   • 实验不是为了学习完整的Ms. Pacman游戏策略
   • 而是学习"如何从任意位置到达任意豆子位置"的通用导航能力

3. 环境重置机制：

   • "When the subgoal is achieved, the environment is reset…"这句话表明这是一个修改过的训练框架
   • 每次达成子目标后就重置环境，放置Pac-Man在随机位置，开始新回合
   • 这样设计是为了收集大量不同起始位置到特定目标的轨迹

这种设置与你原本对Ms. Pacman的理解不同，因为：

• 不是在玩完整的游戏
• 每个回合只关注一个特定豆子
• 实验的核心是学习从任意位置到任意豆子位置的导航策略

这种实验设计允许UVFA学习一个通用映射：从任何状态s到任何目标位置g的价值函数。一旦学习完成，UVFA就可以用来指导Pac-Man从任何位置高效地导航到游戏中的任何豆子位置。

这是一个为了验证UVFA能力而设计的受控实验，而不是直接应用于标准游戏规则的完整实现。

澄清完这4点以后，最大的问题也可以回答了，这篇文章确实没有提到正儿八经的迁移问题（我理解的迁移以 MS Pacman为例是，获得任意区域的豆子是子任务，到完成整个吃豆子的任务才是迁移，或者是完成一个简单的动作，然后可课程学习一样迁移到完成一个复合动作），文章的迁移更像是 MS Pacman或grid-world with 4 rooms完成了A,B地图随机点的到达任务，然后再A.B地图中随机一个点，他也可以到达（泛化性测试），也许可能能迁移到不同地图的不同坐标点，但是文章没有明确说，只是提到unseen goals。

代码：

目前看来UVFA只是修改了值函的求解部分，将强化学习的表达进行了修改，那么用TD更新和Horde没啥差别，代码如下：

也可能我理解错误，因为这篇文章被很多博客在介绍解读技术细节，但我的理解只是改变了V值。

"""
UVFA-DQN for CartPole EnvironmentTask setup:
- State s: 4-dimensional vector [x, x_dot, theta, theta_dot] representing cart position, velocity, pole angle, and angular velocity.
- Goal g: a scalar desired cart horizontal position in the range [-2.4, 2.4]. Each episode trains the agent to reach a sampled subgoal.
- Reward function:r = +1 if |x' - g| < 0.1, else r = -|x' - g|; encourages the agent to move the cart toward g.
- Multi-task training: train_goals is a set of subgoals sampled during training; test_goals are held-out positions for zero-shot transfer evaluation.
"""
import random
import gym
import numpy as np
import collections
import torch
import torch.nn.functional as F
from tqdm import tqdmdevice = torch.device("cuda" if torch.cuda.is_available() else "cpu")class ReplayBufferUVFA:"""Replay buffer storing (state, action, reward, next_state, done, goal)"""def __init__(self, capacity):self.buffer = collections.deque(maxlen=capacity)def add(self, state, action, reward, next_state, done, goal):self.buffer.append((state, action, reward, next_state, done, goal))def sample(self, batch_size):batch = random.sample(self.buffer, batch_size)s, a, r, s2, d, g = zip(*batch)return (np.array(s),np.array(a),np.array(r, dtype=np.float32),np.array(s2),np.array(d, dtype=np.float32),np.array(g, dtype=np.float32))def size(self): return len(self.buffer)class QNetUVFA(torch.nn.Module):#单流结构"""UVFA Q-network: takes state and goal as input"""def __init__(self, state_dim, goal_dim, hidden_dim, action_dim):super().__init__()self.fc1 = torch.nn.Linear(state_dim + goal_dim, hidden_dim)self.fc2 = torch.nn.Linear(hidden_dim, action_dim)def forward(self, state, goal):# state: [B, state_dim], goal: [B, goal_dim]x = torch.cat([state, goal], dim=1)x = F.relu(self.fc1(x))return self.fc2(x)#svd分解那套，我没有加到网络中，感觉这个在网络结构中可有可无
# def svd_embedding(training_goals):
#        # Step 1: Compute value functions for training goals
#     M = np.zeros((env.num_states, len(training_goals)))
#     for g_idx, g in enumerate(training_goals):
#         V = value_iteration(env, g)
#         M[:, g_idx] = V#     # Step 2: Perform matrix factorization using SVD (rank=5)
#     rank = 5
#     U, S, Vt = np.linalg.svd(M, full_matrices=False)
#     phi = U[:, :rank]  # State embeddings
#     psi = (np.diag(S[:rank]) @ Vt[:rank, :]).T  # Goal embeddings#     # Step 3: Train goal embedding network
#     goal_features = np.array(training_goals) / (env.size - 1)  # Normalize to [0,1]
#     goal_features = torch.from_numpy(goal_features).float()
#     psi_targets = torch.from_numpy(psi).float()#     return psi_targets# # Value iteration to compute V(s) for a goal
# def value_iteration(env, g, theta=1e-6):
#     V = np.zeros(env.num_states)
#     while True:
#         delta = 0
#         for s_idx, s in enumerate(env.states):
#             v = V[s_idx]
#             max_q = -np.inf
#             for a in env.actions:
#                 s_next = env.get_next_state(s, a)
#                 s_next_idx = env.states.index(s_next)
#                 r = env.get_reward(s, a, s_next, g)
#                 q = r + env.gamma * V[s_next_idx]
#                 max_q = max(max_q, q)
#             V[s_idx] = max_q
#             delta = max(delta, abs(v - V[s_idx]))
#         if delta < theta:
#             break
#     return V# # UVFA Embedding Networks
# class StateEmbeddingNet(torch.nn.Module):
#     def __init__(self, state_dim, hidden_dim, embed_dim):
#         super(StateEmbeddingNet, self).__init__()
#         self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
#         self.fc2 = torch.nn.Linear(hidden_dim, embed_dim)#     def forward(self, x):
#         x = F.relu(self.fc1(x))
#         return self.fc2(x)# class GoalEmbeddingNet(torch.nn.Module):
#     def __init__(self, goal_dim, hidden_dim, embed_dim):
#         super(GoalEmbeddingNet, self).__init__()
#         self.fc1 = torch.nn.Linear(goal_dim, hidden_dim)
#         self.fc2 = torch.nn.Linear(hidden_dim, embed_dim)#     def forward(self, x):
#         x = F.relu(self.fc1(x))
#         return self.fc2(x)# class QNetUVFA(torch.nn.Module):#双流结构没调试
#     """UVFA Q-network: takes state and goal as input"""
#     def __init__(self, state_dim, goal_dim, hidden_dim, embed_dim):
#         super().__init__()
#         self.fc1 = GoalEmbeddingNet(goal_dim, hidden_dim, embed_dim)
#         self.fc2 = StateEmbeddingNet(goal_dim, hidden_dim, embed_dim)class UVFADQN:"""Conditional DQN (UVFA) agent that inputs state and goal"""def __init__(self, state_dim, goal_dim, hidden_dim, action_dim,lr, gamma, epsilon, target_update):self.gamma = gammaself.epsilon = epsilonself.count = 0self.target_update = target_updateself.q_net = QNetUVFA(state_dim, goal_dim, hidden_dim, action_dim).to(device)self.target_q = QNetUVFA(state_dim, goal_dim, hidden_dim, action_dim).to(device)self.target_q.load_state_dict(self.q_net.state_dict())self.opt = torch.optim.Adam(self.q_net.parameters(), lr=lr)def select_action(self, state, goal):if random.random() < self.epsilon:return random.randrange(self.q_net.fc2.out_features)s = torch.tensor(state, dtype=torch.float32, device=device).unsqueeze(0)g = torch.tensor([[goal]], dtype=torch.float32, device=device)q = self.q_net(s, g)return q.argmax().item()def update(self, buffer, batch_size):s, a, r, s2, done, g = buffer.sample(batch_size)states = torch.tensor(s, dtype=torch.float32, device=device)actions = torch.tensor(a, dtype=torch.int64, device=device).unsqueeze(1)rewards = torch.tensor(r, dtype=torch.float32, device=device).unsqueeze(1)next_states = torch.tensor(s2, dtype=torch.float32, device=device)dones = torch.tensor(done, dtype=torch.float32, device=device).unsqueeze(1)goals = torch.tensor(g.reshape(-1,1), dtype=torch.float32, device=device)# Q(s,g,a)q_values = self.q_net(states, goals).gather(1, actions)# target: r + gamma * max_a' Q_target(s',g,a')with torch.no_grad():next_q = self.target_q(next_states, goals).max(1)[0].unsqueeze(1)q_target = rewards + self.gamma * next_q * (1 - dones)loss = F.mse_loss(q_values, q_target)self.opt.zero_grad()loss.backward()self.opt.step()if self.count % self.target_update == 0:self.target_q.load_state_dict(self.q_net.state_dict())self.count += 1def train_uvfa(env, agent, buffer, train_goals, num_episodes,minimal_size, batch_size):"""Train UVFA-DQN: sample a subgoal each episode for multi-task learning"""returns = []for ep in tqdm(range(num_episodes)):state = env.reset()# sample a subgoal for this episodegoal = float(random.choice(train_goals))ep_ret = 0done = Falsewhile not done:action = agent.select_action(state, goal)next_state, _, done, _ = env.step(action)# define reward based on distance to goal (scalar comparison)dist = abs(next_state[0] - goal)reward = 1.0 if dist < 0.1 else -distbuffer.add(state, action, reward, next_state, done, goal)state = next_stateep_ret += rewardif buffer.size() > minimal_size:agent.update(buffer, batch_size)returns.append(ep_ret)return returnsdef zero_shot_eval(env, agent, test_goals, episodes=20):"""Zero-shot transfer evaluation: directly test trained UVFA-DQN on unseen goals (no parameter updates)"""results = {}for g in test_goals:total = 0for _ in range(episodes):s = env.reset()done = Falseep_r = 0goal = float(g)while not done:a = agent.select_action(s, goal)s, _, done, _ = env.step(a)dist = abs(s[0] - goal)r = 1.0 if dist < 0.1 else -distep_r += rtotal += ep_rresults[goal] = total / episodesreturn resultsif __name__ == '__main__':env = gym.make('CartPole-v0')state_dim = env.observation_space.shape[0]goal_dim = 1action_dim = env.action_space.n# UVFA hyperparameterslr = 2e-3gamma = 0.98epsilon = 0.05target_update = 10buffer_size = 10000minimal_size = 500batch_size = 64num_episodes = 500# define training and zero-shot test goalstrain_goals = np.linspace(-2.0, 2.0, 5)test_goals = np.linspace(-1.5, 1.5, 3)buffer = ReplayBufferUVFA(buffer_size)agent = UVFADQN(state_dim, goal_dim, hidden_dim=128,action_dim=action_dim,lr=lr, gamma=gamma,epsilon=epsilon, target_update=target_update)# Train on multi-task subgoalstrain_uvfa(env, agent, buffer, train_goals,num_episodes, minimal_size, batch_size)# Zero-shot transfer evaluation (unseen goals)zs_results = zero_shot_eval(env, agent, test_goals)print("Zero-shot transfer returns:", zs_results)