本文继续深入探讨蒙特卡罗 (MC)方法。这些方法的特点是能够仅从经验中学习,不需要任何环境模型,这与动态规划(DP)方法形成对比。
这一特性极具吸引力 - 因为在实际应用中,环境模型往往是未知的,或者难以精确建模转移概率。以21点游戏为例:尽管我们完全理解游戏规则,但通过DP方法解决它将极为繁琐 - 因为需要计算各种条件概率,例如给定当前已发牌的情况下,"21点"出现的概率,再抽到一张7的概率等。而通过MC方法,可以绕过这些复杂计算,直接从游戏体验中学习。
由于不依赖模型,MC方法是无偏的。它们在概念上简单明了,易于理解,但表现出较高的方差,且不能采用迭代方式求解(即无法进行自举)。
本文结构如下:首先介绍MC方法和"预测"问题，接着我们讨论"控制"问题。将展示一个基于两个(不太实际的)假设的初始MC控制算法:我们将观察到无限多的情节,且每个状态-动作对将被访问无限多次(探索性启动)。
文章的后半部分将讨论如何移除这些假设:第一个假设相对容易处理,但后者需要更多考虑。我们首先介绍一种on-policy方法,其中最优策略保持ε-greedy,然后转向涉及重要性采样的off-policy方法。
我们首先讨论预测问题 - 即估计给定策略的价值函数。MC方法的核心思想是采样情节,收集并平均跟随状态的回报来计算价值函数 - 在极限情况下,这将收敛到期望值。这一原理是所有MC方法的共同特征,也是这类方法名称的由来(因Monte Carlo地区赌博盛行而得名)。
我们可以定义两种略有不同的计算平均值的方法:首次访问MC方法和每次访问MC方法。在生成情节时,某个状态可能被多次访问。首次访问MC方法仅对第一次出现后的回报进行平均来估计状态的值,而每次访问MC方法对所有访问后的回报进行平均。这两种方法具有略微不同的理论特性,但本文我们将主要关注首次访问MC。其伪代码如下:
这种方法在当前情境下并非最优:在基于模型的方法中,当估计价值函数时,可以对任何状态探索所有可能的动作,然后选择导向最高价值状态的动作。在无模型的情况下,估计状态-动作对的值(即动作-价值函数)更为合适。这可以按照上面为价值函数所示的方式进行,但存在一个关键问题:很可能有些状态-动作对从未被访问过,没有结果可以平均,导致其价值估计无法改善。
这会涉及到探索-利用权衡的问题,以及如何持续保持探索。一个保证这一点的假设是在随机状态/动作组合中开始每个情节,并要求所有动作都有非零概率被选择。这被称为探索性启动假设。我们将在下一节MC控制中结合其实现。
Monte Carlo控制与动态规划文章中一样,我们将使用广义策略迭代(GPI) - 策略评估和策略改进的迭代步骤。
这个过程本身在这里并无特别之处 - 除了我们必须做两个(不太实际的)假设才能使其工作:我们需要探索性启动的假设,并且要求策略评估的内部循环进行无限多步。
后一个假设可以相对容易地移除:允许提前终止策略评估循环是可以的,只需让策略的值更接近最优值而不必达到它 。
目前,我们将保留探索性启动假设,并展示一个基于此假设的控制算法 - 在文章的最后部分,我们将展示如何移除这个假设。
带有探索性启动的Monte Carlo控制以下是带有探索性启动(ES)的MC控制的伪代码:
现在让我们将其转换为Python代码:
这里遇到的第一个挑战是如何保证ES假设:gymnasium [2]和一般的RL环境并不设计用于跳转到任意状态。我们初始化环境,然后从初始状态开始行动(例如,一个问题是历史信息:如果跳转到随机状态,如何生成可能需要的历史帧?)。
也许可以修改gymnasium,或实现允许生成随机状态的自定义环境,但这里实现一个函数来生成环境的几个可能状态,然后从这个列表中采样。
这个函数会被多次调用(对每个生成的情节),所以我们希望缓存这个计算成本高的生成过程。我在测试时发现functools.lrucache装饰器在确定gymnasium环境的相等性时存在问题，所以自己实现了一个自定义缓存装饰器,忽略第一个(环境)参数(另一个选择是重写环境的equals方法):
def call_once(func): """自定义缓存装饰器 忽略第一个参数。 """ cache = {} def wrapper(*args, **kwargs): key = (func.__name__, args[1:]) assert not kwargs, "We don't support kwargs atm" if key not in cache: cache[key] = func(*args) return cache[key] return wrapper @call_once def generate_possible_states( env: ParametrizedEnv, num_runs: int = 100 ) -> list[tuple[int, ParametrizedEnv]]: """生成环境的可能状态。 为此,通过选择随机状态并从该状态开始 遵循随机策略,记录新状态, 迭代地增加已知状态集。 Args: env: 使用的环境 num_runs: 发现循环的次数 Returns: 包含发现状态的列表 - 这些是状态(观察) 和表示该状态的gym环境的元组 """ _, action_space = get_observation_action_space(env) observation, _ = env.env.reset() possible_states = [(observation, copy.deepcopy(env))] for _ in range(num_runs): observation, env = random.choice(possible_states) env = copy.deepcopy(env) terminated = truncated = False while not terminated and not truncated: action = np.random.choice([a for a in range(action_space.n)]) observation, _, terminated, truncated, _ = env.env.step(action) if observation in set([state for state, _ in possible_states]): break else: if not terminated and not truncated: possible_states.append((observation, env)) return possible_states def generate_random_start(env: ParametrizedEnv) -> tuple[int, ParametrizedEnv]: """选择一个随机起始状态。 为此,首先生成所有可能的状态(缓存), 然后从这些状态中随机选择一个。 """ possible_states = generate_possible_states(env) observation, env = random.choice(possible_states) return copy.deepcopy(observation), copy.deepcopy(env)
这个generate_possible_states函数迭代地构建一组可能的状态:它首先从已知状态中选择一个,然后遵循随机动作,同时记录所有新状态。
另外还实现了一个辅助函数来生成遵循特定策略的情节:
def generate_episode( env: ParametrizedEnv, pi: np.ndarray, exploring_starts: bool, max_episode_length: int = 20, ) -> list[tuple[Any, Any, Any]]: """生成一个遵循给定策略的情节。 Args: env: 使用的环境 pi: 要遵循的策略 exploring_starts: 当为True时遵循探索性启动假设(ES) Returns: 生成的情节 """ _, action_space = get_observation_action_space(env) episode = [] observation, _ = env.env.reset() if exploring_starts: # 如果是ES,选择随机起始状态 observation, env = generate_random_start(env) terminated = truncated = False initial_step = True while not terminated and not truncated: if initial_step and exploring_starts: # 如果是ES,初始时选择随机动作 action = np.random.choice([a for a in range(action_space.n)]) else: action = np.random.choice( [a for a in range(action_space.n)], p=pi[observation] ) initial_step = False observation_new, reward, terminated, truncated, _ = env.env.step(action) episode.append((observation, action, reward)) # 终止智能体陷入死循环的情节 if len(episode) > max_episode_length: break observation = observation_new return episode
当使用ES调用时,第一个动作选择是完全随机的 - 否则在每一步根据策略分布采样动作。提前终止"陷入死循环"的情节很重要:由于在MC方法中完整地执行情节并且只在之后更新策略,可能会发生策略陷入某个非理想动作导致循环访问序列的情况(例如,在左上角不断向左转 - 这个动作没有任何实际效果)。为了避免无限等待(或直到环境自行终止),我们需要在达到一定步数后主动中断。
有了这些辅助函数,就可以可以实现上面介绍的MC ES算法:
def mc_es(env: ParametrizedEnv) -> np.ndarray: """通过带有探索性启动的Monte Carlo方法 求解传入的Gymnasium环境。 Args: env: 包含问题的环境 Returns: 找到的策略 """ observation_space, action_space = get_observation_action_space(env) pi = ( np.ones((observation_space.n, action_space.n)).astype(np.int32) / action_space.n ) Q = np.zeros((observation_space.n, action_space.n)) returns = defaultdict(list) num_steps = 1000 for t in range(num_steps): episode = generate_episode(env, pi, True) G = 0.0 for t in range(len(episode) - 1, -1, -1): s, a, r = episode[t] G = env.gamma * G + r prev_s = [(s, a) for (s, a, _) in episode[:t]] if (s, a) not in prev_s: returns[s, a].append(G) Q[s, a] = statistics.fmean(returns[s, a]) if not all(Q[s, a] == Q[s, 0] for a in range(action_space.n)): for a in range(action_space.n): pi[s, a] = 1 if a == np.argmax(Q[s]) else 0 return np.argmax(pi, 1)
从Python代码到上面的伪代码的映射应该相对直观 - 但需要做一个重要的补充,即在更新策略之前进行相等性检查:在生成情节时可能没有遇到非零奖励,所以策略更新最初会在缺乏信息的情况下最大化某个动作。
无探索性启动的Monte Carlo控制上面我们已经看到了一个基于ES假设的初始MC控制方法。但这种方法在实际应用中存在局限性 - 因此在后面部分,我们将介绍两种不依赖这个假设的方法。
总的来说,为了使这些方法有效,必须确保所有状态-动作对都能被无限次探索(*),有两类方法可以帮助我们实现这一点:on-policy方法和off-policy方法。On-policy方法试图改进生成数据的策略,而off-policy方法可以改进目标策略,同时从不同的行为策略生成数据。
这里我们首先介绍一个on-policy方法,然后再介绍一个off-policy方法。到目前为止我们考虑的所有方法都是on-policy方法,因为只涉及一个策略 - 我们用它来生成数据,同时也直接优化它。为了在on-policy方法中满足条件(*),核心思想是保持策略"软性",而不是像我们在上面做的那样进行"硬性"更新。也就是说,当更新策略朝向最大Q值的动作时,我们不会直接转向确定性策略,而是仅仅朝着这样一个确定性策略移动,同时保持
对于所有状态和动作成立。
除此之外,与上面介绍的MC ES算法没有太大区别,结构上的相似性应该很明显:
Python实现:
def on_policy_mc(env: ParametrizedEnv) -> np.ndarray: """通过on-policy Monte Carlo控制方法 求解传入的Gymnasium环境。 Args: env: 包含问题的环境 Returns: 找到的策略 """ observation_space, action_space = get_observation_action_space(env) pi = ( np.ones((observation_space.n, action_space.n)).astype(np.int32) / action_space.n ) Q = np.zeros((observation_space.n, action_space.n)) returns = defaultdict(list) num_steps = 1000 for _ in range(num_steps): episode = generate_episode(env, pi, False) G = 0.0 for t in range(len(episode) - 1, -1, -1): s, a, r = episode[t] G = env.gamma * G + r prev_s = [(s, a) for (s, a, _) in episode[:t]] if (s, a) not in prev_s: returns[s, a].append(G) Q[s, a] = statistics.fmean(returns[s, a]) if not all(Q[s, a] == Q[s, 0] for a in range(action_space.n)): A_star = np.argmax(Q[s, :]) for a in range(action_space.n): pi[s, a] = ( 1 - env.eps + env.eps / action_space.n if a == A_star else env.eps / action_space.n ) return np.argmax(pi, 1)
Off-Policy Monte Carlo控制要介绍off-policy方法,我们需要更深入的讨论。这类方法的提出源于在on-policy方法中所做的折衷:希望学习一个最优策略,但同时还必须保持探索。到目前为止介绍的on-policy方法正是解决了这样一个折衷:它们学习一个近似最优的策略,该策略仍然保持一定程度的探索。使用两个策略似乎更为直观:优化一个目标策略,同时使用另一个策略来探索和生成数据。
正是由于这个原因,off-policy方法实际上更加通用和强大(一些最著名的RL算法,如Q-Learning,都是off-policy方法) - 但它们通常具有更高的方差,收敛速度也较慢。
大多数off-policy方法的共同特点是使用重要性采样:因为我们想计算目标策略的期望,但实际上是从不同的策略采样,所以直接计算的期望是有偏的。重要性采样通过适当地缩放观察到的值,使得得到的估计重新变得无偏。由于篇幅限制,这里不做更详细的介绍。
要将这应用于强化学习,我们再次从预测问题开始。首先看一下遵循策略π时状态-动作轨迹出现的概率:
现在我们引入两个策略:π是我们的目标策略,即我们想要优化的策略,b是我们的行为策略 - 用于生成数据的策略。我们需要满足覆盖性假设:如果在π下可能采取某个动作,那么在b下该动作也必须有非零概率 - 这意味着b需要在与π不同的状态下是一个随机策略。
接下来计算轨迹在π和b下出现的概率比:
值得注意的是,环境本身的任何属性,如状态转移概率都被约掉了,只剩下策略比率。
所以需要这个比率来进行off-policy预测,因为我们想从b收集的数据中估计策略π的值 - 这意味着直接计算的期望
会是有偏的。但是,通过重要性采样可以修正这个偏差:
具体应用到RL预测问题得到:
这个形式应该看起来很熟悉:一个状态的预测值只是(重要性采样修正的)平均回报(除以用于计算期望的状态数 - 这是典型的Monte Carlo期望估计)。
但对于我们的算法,实际上会使用加权重要性采样:
也就是说,它在分母中也包含了重要性采样权重,即计算加权平均值(注意分子和分母的权重通常不会相互抵消,因为它们出现在求和中)。
如果我们只观察到一个回报,加权重要性采样的权重会相互抵消。期望会是有偏的(它会得出v_b),但这似乎是合理的,因为我们只观察到这一个样本。对于普通重要性采样,期望总是无偏的,但它可能会非常极端 - 例如,想象重要性采样权重是10或100。总的来说,加权重要性采样是有偏的(尽管随着样本数增加,偏差趋向于0),但比普通重要性采样具有更低的方差,后者可能具有无界(无限)方差。
回到off-policy MC预测,我们想要形成的估计是:
理论上,可以在每一步单独计算这个测量值 - 也就是保留一个权重和回报的列表并计算相应的值。
但是我们可以引入一种增量方法来节省内存。在这种方法中只需要存储先前的估计,并使用新值来更新它
此外还需要保持对迄今为止累积权重的运行估计:
乍看之下,可能不太明显这些公式确实是加权重要性采样的增量版本。但它们是正确的。在这里省略了证明以保持文章的简洁,并且因为我认为它对理解核心概念并不是至关重要的。但是我们这里想象一个更简单的,非增量版本的这个算法可能有助于理解。
使用这个增量更新方案,MC off-policy预测算法如下:
转到off-policy MC控制算法:
我们看到完全相同的重要性采样更新模式,只是现在还更新了一个策略。注意上面的W != 0检查现在被"break"替代,只要选择的动作不对应于π(这意味着W = 0,因为π是确定性的)。出于同样的原因,更新规则也是1/b(…)而不是π(…)/b(…)。
这个算法可以直接翻译成Python代码:
def off_policy_mc(env: ParametrizedEnv) -> np.ndarray: """通过off-policy Monte Carlo控制方法 求解传入的Gymnasium环境。 Args: env: 包含问题的环境 Returns: 找到的策略 """ observation_space, action_space = get_observation_action_space(env) Q = np.zeros((observation_space.n, action_space.n)) C = np.zeros((observation_space.n, action_space.n)) pi = np.argmax(Q, 1) num_steps = 1000 for _ in range(num_steps): b = np.random.rand(int(observation_space.n), int(action_space.n)) b = b / np.expand_dims(np.sum(b, axis=1), -1) episode = generate_episode(env, b, False) G = 0.0 W = 1 for t in range(len(episode) - 1, -1, -1): s, a, r = episode[t] G = env.gamma * G + r C[s, a] += W Q[s, a] += W / C[s, a] * (G - Q[s, a]) pi = np.argmax(Q, 1) if a != np.argmax(Q[s]): break W *= 1 / b[s, a] return pi
我们将展示一个增量算法的"更简单"和更直观的版本 - 将实现"纯"重要性采样,而不使用增量更新。这有助于理解,它清晰地展示了重要性采样是如何真正使用的,一步一步地形成所有需要的项:
def off_policy_mc_non_inc(env: ParametrizedEnv) -> np.ndarray: """通过on-policy MonteCarlo控制方法求解传入的Gymnasium环境 - 但不使用Sutton的增量算法来更新重要性采样权重。 Args: env: 包含问题的环境 Returns: 找到的策略 """ observation_space, action_space = get_observation_action_space(env) Q = np.zeros((observation_space.n, action_space.n)) num_steps = 10000 returns = defaultdict(list) ratios = defaultdict(list) for _ in range(num_steps): b = np.random.rand(int(observation_space.n), int(action_space.n)) b = b / np.expand_dims(np.sum(b, axis=1), -1) episode = generate_episode(env, b, False) G = 0.0 ratio = 1 for t in range(len(episode) - 1, -1, -1): s, a, r = episode[t] G = env.gamma * G + r # 创建当前目标策略, # 它是Q函数的argmax, # 但对于Q值相等的情况给予相等的权重 pi = np.zeros_like(Q) pi[np.arange(Q.shape[0]), np.argmax(Q, 1)] = 1 uniform_rows = np.all(Q == Q[:, [0]], axis=1) pi[uniform_rows] = 1 / action_space.n ratio *= pi[s, a] / b[s, a] if ratio == 0: break returns[s, a].append(G) ratios[s, a].append(ratio) Q[s, a] = sum([r * s for r, s in zip(returns[s, a], ratios[s, a])]) / sum( [s for s in ratios[s, a]] ) Q = np.nan_to_num(Q, nan=0.0) return np.argmax(Q, 1)
更深入地分析这个实现，的目标是收集并平均所有状态-动作对的回报 - 不同之处在于这里我们使用every-visit MC而不是first-visit MC(因此没有检查该对是否已经在情节中出现过)。因此通过生成情节并平均收集的回报来构建Q函数 - 这里也不是简单地平均,而是使用(加权)重要性采样，所以需要计算的量是:
集合τ表示访问状态s的所有时间步 - 对于first-visit MC,它只包括首次访问s的所有时间步(使用状态-动作对)。忽略重要性采样权重,这个公式将得到与MC ES完全相同的结果 - 对于状态的每次出现,我们收集其相应的回报G_t,并取平均值。现在,还需要重要性采样权重s来修正我们的有偏估计。这些权重是:
因此,除了跟踪每个访问的状态-动作对的回报之外,还用这个量填充ratios。为了生成它,当反向遍历一个情节时,我们迭代地计算p_{T-1:T-1}、p_{T-2:T-1}等,并将这个运行计算存储在ratio中 - 在每一步都将其附加到ratios中。即时计算当前目标策略π以估计正确的权重 - 它只是从当前Q获得的确定性策略。
这种方法是有效的(就像所有其他方法一样),但它需要大约10倍的情节数才能保证收敛,相比增量off-policy MC控制算法而言 - 这是非常奇怪的。我的理解是在这里只是实现了一个具有完全相同功能的非增量版本。所以这个额外的实现能更清楚地说明off-policy MC控制是如何工作的，但是并不能在实际应用中保证快速的运行。如果您有任何问题、评论或建议/解决方案 - 特别是关于为什么结果会有差异 - 请随时让我知道!
总结MC方法的独特之处在于它们仅从经验中学习,不需要环境模型 - 这为解决复杂问题提供了令人兴奋的可能性。
我们首先介绍了一个基于探索性启动(ES)假设的MC控制算法 - 即每个状态-动作对都以非零概率被无限探索。之后,我们尝试移除ES假设 - 最后,我们还提供了一个非增量off-policy MC控制算法的实现 - 这可能有助于更深入地理解算法的工作原理,并展示了如何以直观的方式实现off-policy MC控制,而不需要任何优化技巧。
参考文献
[1] Sutton, R. S., & Barto, A. G. (2018). Reinforcement learning: An introduction. MIT press. http://incompleteideas.net/book/RLbook2020.pdf 注：公式都是从这里截图的
[2] Farama Foundation. (2022). Gymnasium: A Python Library for Reinforcement Learning Environments. https://github.com/Farama-Foundation/Gymnasium
https://avoid.overfit.cn/post/400bebe168ab407e95c73580c331f6da
作者：Oliver S
