10天速通强化学习-003--策略迭代、价值迭代

m0_72169557

1279人浏览 · 2025-02-26 16:29:09

m0_72169557 · 2025-02-26 16:29:09 发布

本节中需要明白的概念：

上一节中我们讲到最优状态价值函数和最优动作价值函数，我们要理解，最优状态价值函数是所有的动作价值函数与其概率相乘的结果，所以可以很自然的想到，最优状态价值函数和最优动作价值函数的大小关系，一般来说前者大，若是后者大，则前者取后者（有点废话了）

还有一点要明白：策略变化，状态价值函数和动作价值函数会跟着变化

思考：怎么找到每个状态的最大价值->借此来找到最优策略

本节介绍一个算法：动态规划算法

主要有两种：

第一种策略迭代（policy iteration）：包括策略评估 + 策略提升

也就是说先给定一个策略，然后计算这个策略的价值，然后根据价值改进策略（计算上一个策略的每个状态动作对的价值，选最大的作为新策略），得到新策略（这个新策略中的动作价值函数就不一定每个都最大了，因为这是根据旧策略推出来的最大，新策略中不一定最大，策略变化，价值会跟着变化），重复这个过程，结束条件就是新策略和旧策略一样，那就是最优策略了

第二种价值迭代（value iteration）：根据贝尔曼最优公式（状态价值函数那个），直接不断迭代（就好像给一个具体的函数表达式，你不停的朝一个方向变化自变量找到极值的做法一样）得到最优的状态价值，再根据价值反推策略

动态规划算法注意点：

基于动态规划的强化学习算法需要事先知道环境的状态转移函数和奖励函数。这通常是理想的，现实环境很少。并且该算法只适用于状态空间和动作空间是离散且有限的环境（连续空间--状态动作对无法取完，时间是无限的，下一步可能是0.1s 也可能是0.001s ，无限环境--计算量太大）

环境搭建

如图一：4*12网格，上下左右四个状态，碰到边界弹回原状态，进入灰色或者到重点会结束动作回到起点，进入灰色奖励-100，进入白色奖励 -1，进入原状态和终点奖励为0

代码：

主要就是创建状态转移矩阵和总共的环境

P = [[[] for j in range(4)] for i in range(self.nrow * self.ncol)]

这个P就是整个环境好好理解：注意：状态就是一维索引表示

注意理解：

环境有了之后创建状态转移矩阵

注意这个createP是一个从三位列表P中定位的一个具体元素，其实是一个一维数组，这个矩阵表示表示状态state采取动作action 之后的转移结果，：这个createP；里面有许多元素组成一维数组，这些元素是四元组形式（p,next_state,reward,done）--概率，下一个状态，奖励，是否终止

import copy


class CliffWalkingEnv:
    """ 悬崖漫步环境"""
    def __init__(self, ncol=12, nrow=4):
        self.ncol = ncol  # 定义网格世界的列
        self.nrow = nrow  # 定义网格世界的行
        # 转移矩阵P[state][action] = [(p, next_state, reward, done)]包含下一个状态和奖励
        self.P = self.createP()

    def createP(self):
        # 初始化
        P = [[[] for j in range(4)] for i in range(self.nrow * self.ncol)]
        # 4种动作, change[0]:上,change[1]:下, change[2]:左, change[3]:右。坐标系原点(0,0)
        # 定义在左上角
        change = [[0, -1], [0, 1], [-1, 0], [1, 0]]
        #通过三重循环遍历每个状态和每个动作，根据状态和动作计算下一个坐标next_x和next_y
        for i in range(self.nrow):
            for j in range(self.ncol):
                for a in range(4):
                    # 位置在悬崖或者目标状态,因为无法继续交互,任何动作奖励都为0,最后一行
                    if i == self.nrow - 1 and j > 0:
                        #p[state][action] = [（四元素）]
                        P[i * self.ncol + j][a] = [(1, i * self.ncol + j, 0,
                                                    True)]
                        continue
                    # 其他位置，坐标计算
                    #里面的min max 的意思是：max表示横坐标不能小于0，min表示纵坐标不能大于最大列宽，要不然出边界了
                    #碰到边界返回原状态，并且奖励为0
                    next_x = min(self.ncol - 1, max(0, j + change[a][0]))
                    next_y = min(self.nrow - 1, max(0, i + change[a][1]))
                    #二维坐标变换为一维索引
                    next_state = next_y * self.ncol + next_x
                    reward = -1
                    done = False
                    # 下一个位置在悬崖或者终点
                    if next_y == self.nrow - 1 and next_x > 0:
                        done = True
                        if next_x != self.ncol - 1:  # 下一个位置在悬崖
                            reward = -100
                    P[i * self.ncol + j][a] = [(1, next_state, reward, done)]
        return P

策略迭代算法 = 策略评估 + 策略提升（智能体）

策略评估

目的：

用来计算给定策略的状态价值函数（注意状态价值函数是一个分布）

方法:

对于一个给定的策略，计算状态价值函数的公式为

$V^\pi(s)=\sum_{a\in A}\pi(a|s)\left(r(s,a)+\gamma\sum_{s^{\prime}\in S}p(s^{\prime}|s,a)V^\pi(s^{\prime})\right)$

显然我们不能在使用求逆来计算了，规模量较大，（硬用也可以-这里要理解 $V^{\pi}(s)$ 和 $V^{\pi}(s')$ 是一样的）

现在变通一下式子：

$V^{k+1}(s)=\sum_{a\in A}\pi(a|s)\left(r(s,a)+\gamma\sum_{s^{\prime}\in S}P(s^{\prime}|s,a)V^k(s^{\prime})\right)$

有这个式子我们就可以得到一种动态规划的算法。给定一个任意的初始值 $V^{0}$ ，等式右边推左边，当 $k\rightarrow \oe$ 00 时， $V^{k}(s)$ j就和 $V^{\pi}(s)$ 一样了 (这个算法怎么理解呢，这个公式2只是一个计算方法或者计算技巧，不要带着实际问题去考虑，就好像给一个具体的函数表达式，你不停的朝一个方向变化自变量找到极值的做法一样)，实际不可能计算到无穷次，如果某一轮两个状态价值函数相差非常小，就可以结束了，算法证明自己了解

策略提升

目的：

根据上一步计算出的特定策略下的状态价值函数，来改进策略

方法：

明白一个道理：从一开始选取的策略计算出的状态价值函数，然后选取根据每一个状态的状态价值函数，选取状态动作对-（s,a）动作价值函数最大的这个作为新的该状态的状态价值函数，同时最大的作为新策略，这就是策略提升

公式表述：假设存在一个确定性策略 $\pi '$ ,在任意一个状态s下，都满足

$Q^{\pi}\left(s, \pi^{\prime}(s)\right) \geq V^{\pi}(s)$

于是在任意状态下，我们有（这里就是期望和极值的关系）

$V^{\pi'}\left(s\right) \geq V^{\pi}(s)$

这就是策略提升定理。

方案：贪心的在每一个状态选择动作价值最大的动作

$\pi^{\prime}(s)=\arg \max _{a} Q^{\pi}(s, a)=\arg \max _{a}\left\{r(s, a)+\gamma \sum_{s^{\prime}} P\left(s^{\prime} \mid s, a\right) V^{\pi}\left(s^{\prime}\right)\right\}$

结束条件：两个策略一样，即为最优策略，证明自己理解

小结

算法过程

$\pi ^{0 }\rightarrow$ 策略评估 $\rightarrow$ $V^{\pi^{0}}$ $\rightarrow$ 策略提升 $\rightarrow \pi^{1}$ 。。。。。

伪代码：

随机初始化策略 $\pi(s)$ 和价值函数 $V(s)$
while do: (策略评估循环，前者表示一轮中最大变化量，后者表示阈值）
- $\triangle \leftarrow 0$
- 对于每一个状态 s
  - $v\leftarrow V(s)$ (将当前价值赋值给 $v$ )
  - $V(s)\leftarrow r(s,\pi(s)) + \gamma \sum P(s'|s,\pi(s))V(s)$
  - $\triangle \leftarrow max(\triangle ,|v-V(s)|)$
end while
$\pi_{old}\leftarrow \pi$ （开始进行策略提升环节）
对于每一个状态s
- $\pi^{\prime}(s)=\arg \max _{a} Q^{\pi}(s, a)=\arg \max _{a}\left\{r(s, a)+\gamma \sum_{s^{\prime}} P\left(s^{\prime} \mid s, a\right) V^{\pi}\left(s^{\prime}\right)\right\}$ (argmax表示取极值的自变量，max表示取极值)
若 $\pi ^{old} = \pi$ ，停止算法并返回V和 $\pi$ ,否则返回策略评估循环

代码：

主要完成策略评估和策略提升环节

这里要明白深浅拷贝的区别：

为什么要用：要比较新旧策略
在类中
1. 浅拷贝：复制对象的一层结构，复制对象的引用，不是对象本身，会和原对象共享内部可变对象，一个变另一个也会变，两个策略一直相等
2. 深拷贝：完全独立，不会变

class PolicyIteration:
    """ 策略迭代算法 """
    def __init__(self, env, theta, gamma):
        self.env = env
        self.v = [0] * self.env.ncol * self.env.nrow  # 初始化价值为0
        self.pi = [[0.25, 0.25, 0.25, 0.25]
                   for i in range(self.env.ncol * self.env.nrow)]  # 初始化为均匀随机策略
        self.theta = theta  # 策略评估收敛阈值
        self.gamma = gamma  # 折扣因子

    def policy_evaluation(self):  # 策略评估
        cnt = 1  # 计数器
        while 1:
            max_diff = 0   #一轮迭代的最大变化量
            new_v = [0] * self.env.ncol * self.env.nrow   #新策略的状态价值函数，旧策略用类中自带的self.v
            #理解这种三维矩阵的形式的编写
            #遍历状态
            for s in range(self.env.ncol * self.env.nrow):
                qsa_list = []  # 开始计算状态s下的所有Q(s,a)价值
                #遍历每个状态的动作，计算状态动作对的价值
                for a in range(4):
                    qsa = 0
                    for res in self.env.P[s][a]:
                        p, next_state, r, done = res
                        #这里的v已经初始化了变成了0
                        qsa += p * (r + self.gamma * self.v[next_state] * (1 - done))
                        # 本章环境比较特殊,奖励和下一个状态有关,所以需要和状态转移概率相乘
                    qsa_list.append(self.pi[s][a] * qsa)
                # 状态价值函数和动作价值函数之间的关系,注意这里的qsa_list已经有概率在里面了
                #所以直接求和就行
                new_v[s] = sum(qsa_list)  
                #有可能会发生差值大小浮动情况
                max_diff = max(max_diff, abs(new_v[s] - self.v[s]))
            self.v = new_v
            if max_diff < self.theta: break  # 满足收敛条件,退出评估迭代
            cnt += 1
        print("策略评估进行%d轮后完成" % cnt)

    def policy_improvement(self):  # 策略提升
        for s in range(self.env.nrow * self.env.ncol):
            qsa_list = []
            for a in range(4):
                qsa = 0
                for res in self.env.P[s][a]:
                    p, next_state, r, done = res
                    qsa += p * (r + self.gamma * self.v[next_state] * (1 - done))
                qsa_list.append(qsa)
            #还在内层循环中，所以是寻找的某个状态的所有状态动作对价值函数
            maxq = max(qsa_list)
            cntq = qsa_list.count(maxq)  # 计算有几个动作得到了最大的Q值
            # 让这些动作均分概率，循环查找某个状态的所有状态价值对
            self.pi[s] = [1 / cntq if q == maxq else 0 for q in qsa_list]
        print("策略提升完成")
        return self.pi

    def policy_iteration(self):  # 策略迭代
        while 1:
            self.policy_evaluation()
            #这里要进行新旧策略的比较
            old_pi = copy.deepcopy(self.pi)  # 将列表进行深拷贝,方便接下来进行比较
            new_pi = self.policy_improvement()
            if old_pi == new_pi: break

环境代码和策略迭代代码已经完成。用>表示动作方向 o表示原地不动

def print_agent(agent, action_meaning, disaster=[], end=[]):
    print("状态价值：")
    #遍历每行每列，每个状态
    for i in range(agent.env.nrow):
        for j in range(agent.env.ncol):
            # 为了输出美观,保持输出6个字符
            print('%6.6s' % ('%.3f' % agent.v[i * agent.env.ncol + j]), end=' ')
        print()

    print("策略：")
    for i in range(agent.env.nrow):
        for j in range(agent.env.ncol):
            # 一些特殊的状态,例如悬崖漫步中的悬崖
            #if(状态位于悬崖或者终点)
            if (i * agent.env.ncol + j) in disaster:
                print('****', end=' ')
            elif (i * agent.env.ncol + j) in end:  # 目标状态
                print('EEEE', end=' ')
            #如果是正常状态
            #pi是一个二维列表，用来存储状态和动作概率
            else:
                #取出该状态的所有动作概率
                a = agent.pi[i * agent.env.ncol + j]
                pi_str = ''
                #标识动作
                for k in range(len(action_meaning)):
                    pi_str += action_meaning[k] if a[k] > 0 else 'o'
                print(pi_str, end=' ')
        print()


env = CliffWalkingEnv()
action_meaning = ['^', 'v', '<', '>']
theta = 0.001
gamma = 0.9
#创建实例
agent = PolicyIteration(env, theta, gamma)
agent.policy_iteration()
#在三维列表中的索引不是行乘列，只是一维列表，好好理解一下
#这个最后两个参数传进去的是一个是悬崖，一个是重点
print_agent(agent, action_meaning, list(range(37, 47)), [47])

由输出结果可以看出：经过了5轮的策略评估和策略提升的循环迭代（72，44等只是策略评估算法的迭代求值次数），策略收敛了，完成了。

价值迭代算法

就是先算价值，然后反推函数

有的人就问了没有策略就先算价值，怎么算，请看公式:

$V^{*}(s)=\max _{a \in \mathcal{A}}\left\{r(s, a)+\gamma \sum_{s^{\prime} \in \mathcal{S}} P\left(s^{\prime} \mid s, a\right) V^{*}\left(s^{\prime}\right)\right\}$

明白了吧，在环境搭建代码中

 # 转移矩阵P[state][action] = [(p, next_state, reward, done)]包含下一个状态和奖励
        self.P = self.createP()

这个P二维数组：给出了当前状态转移到下一个状态的概率和奖励（这不就是状态转移函数和奖励函数）

上述公式的变体：

$V^{k+1}(s)=\max _{a \in \mathcal{A}}\left\{r(s, a)+\gamma \sum_{s^{\prime} \in \mathcal{S}} P\left(s^{\prime} \mid s, a\right) V^{k}\left(s^{\prime}\right)\right\}$

是不是等到 $V^{k+1}(s)$ 和 $V^{k}(s)$ 相同的时候，就说明到达了贝尔曼最有方程的不动点，那这就是最优的状态价值函数，然后反推策略就可以了（其实这里相当于方法一中策略评估用了一次，直接使用策略提升进行不停循环--都是取动作价值函数最大的那个作为下一代）

恢复策略

$\pi(s)=\arg \max _{a}\left\{r(s, a)+\gamma \sum_{s^{\prime}} p\left(s^{\prime} \mid s, a\right) V^{k+1}\left(s^{\prime}\right)\right\}$

伪代码

随机初始化 $V(s)$
while
- $\triangle \leftarrow 0$
- 对于每一个状态
  - $v\leftarrow V(s)$ 当前值
  - $V(s)=\max _{a \in \mathcal{A}}\left\{r(s, a)+\gamma \sum_{s^{\prime} \in \mathcal{S}} P\left(s^{\prime} \mid s, a\right) V\left(s^{\prime}\right)\right\}$
  - $\triangle \leftarrow max(\triangle ,|v-V(s)|)$
enl while
返回一个确定性策略 $\pi(s)=\arg \max _{a}\left\{r(s, a)+\gamma \sum_{s^{\prime}} p\left(s^{\prime} \mid s, a\right) V^{k+1}\left(s^{\prime}\right)\right\}$

class ValueIteration:
    """ 价值迭代算法 """
    def __init__(self, env, theta, gamma):
        self.env = env
        self.v = [0] * self.env.ncol * self.env.nrow  # 初始化价值为0
        self.theta = theta  # 价值收敛阈值
        self.gamma = gamma
        # 价值迭代结束后得到的策略，初始化空
        self.pi = [None for i in range(self.env.ncol * self.env.nrow)]

    def value_iteration(self):
        cnt = 0
        while 1:
            max_diff = 0
            #新的状态价值函数
            new_v = [0] * self.env.ncol * self.env.nrow
            for s in range(self.env.ncol * self.env.nrow):
                qsa_list = []  # 开始计算状态s下的所有Q(s,a)价值
                for a in range(4):
                    qsa = 0
                    #解包
                    for res in self.env.P[s][a]:
                        p, next_state, r, done = res
                        qsa += p * (r + self.gamma * self.v[next_state] * (1 - done))
                    qsa_list.append(qsa)  # 这一行和下一行代码是价值迭代和策略迭代的主要区别
                new_v[s] = max(qsa_list)
                max_diff = max(max_diff, abs(new_v[s] - self.v[s]))
            self.v = new_v
            if max_diff < self.theta: break  # 满足收敛条件,退出评估迭代
            cnt += 1
        print("价值迭代一共进行%d轮" % cnt)
        self.get_policy()

    def get_policy(self):  # 根据价值函数导出一个贪婪策略
        for s in range(self.env.nrow * self.env.ncol):
            qsa_list = []
            for a in range(4):
                qsa = 0
                for res in self.env.P[s][a]:
                    p, next_state, r, done = res
                    qsa += p * (r + self.gamma * self.v[next_state] * (1 - done))
                qsa_list.append(qsa)
            maxq = max(qsa_list)
            cntq = qsa_list.count(maxq)  # 计算有几个动作得到了最大的Q值
            # 让这些动作均分概率
            self.pi[s] = [1 / cntq if q == maxq else 0 for q in qsa_list]


env = CliffWalkingEnv()
action_meaning = ['^', 'v', '<', '>']
theta = 0.001
gamma = 0.9
agent = ValueIteration(env, theta, gamma)
agent.value_iteration()
print_agent(agent, action_meaning, list(range(37, 47)), [47])

价值迭代一共才14轮，好

小结

可以看出价值迭代和策略迭代代码结构差不多

价值迭代有两个函数

价值迭代：找到最优价值

反推策略：找到最优策略

策略迭代有三个函数

策略评估：找到当前策略下的价值 --（价值迭代在这一步已经找到了最优价值）

策略提升：根据价值提升策略 -（策略反推和根据价值提升策略用的是一个方法，都是取最大动作价值函数）

合成函数：便于集成进行策略迭代

扩展

冰湖环境

可以用来练手

创建环境--其他的都一样，配环境比较麻烦，安装相应的库就行了，这里不写了

import gym
env = gym.make("FrozenLake-v0")  # 创建环境
env = env.unwrapped  # 解封装才能访问状态转移矩阵P
env.render()  # 环境渲染,通常是弹窗显示或打印出可视化的环境

holes = set()
ends = set()
for s in env.P:
    for a in env.P[s]:
        for s_ in env.P[s][a]:
           #元素s[2]是奖励值
            if s_[2] == 1.0:  # 获得奖励为1,代表是目标
                ends.add(s_[1])
            #元素s[3]是终止状态标志
            if s_[3] == True:
                holes.add(s_[1])
holes = holes - ends
print("冰洞的索引:", holes)
print("目标的索引:", ends)

for a in env.P[14]:  # 查看目标左边一格的状态转移信息
    print(env.P[14][a])

天工造物局

北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台

更多推荐

Understand Anything 多智能体知识图谱管线：从代码扫描到可交互 Dashboard 的实现解析

本文以 Understand Anything 的官方文档为主线，系统拆解它如何用一条「Tree-sitter 静态分析 + LLM 语义理解」的多智能体管线，把任意代码库 / 知识库 / 文档转换成可探索、可检索、可问答的知识图谱，并落到一个可交互的 Web Dashboard。读完你应能掌握：如何安装并在 Claude Code / Codex / Cursor / Copilot / Gem

天工造物局

graphify 视频音视频转写全流程解析：从 detect 结果到知识图谱文档的 Step 2.5

graphify 的核心能力是把任意语料（代码、文档、PDF、图片）转化为可查询的知识图谱，而视频与音频文件无法被直接阅读，必须先把它们转写成文本。本文围绕 `graphify` 运行手册中的转写参考文档 [transcribe.md](https://link.gitcode.com/i/d3285781588e0f1b84886dfdbc4fe7f6)，完整讲解 "Step 2.5 - Tra

天工造物局

ML-For-Beginners 终章：经典机器学习在金融、医疗、零售等八大行业中的真实落地场景

本篇技术指南基于开源课程仓库 ML-For-Beginners 的收尾课程《Postscript: Machine learning in the real world》（对应仓库目录 [9-Real-World/1-Applications](https://link.gitcode.com/i/6b23002df80861c20abf34d7db63ef41)）编写。该课此前 11 周的教学带