dive_into_reinforcement_learning

2022-04-19

前言

上海交大动手学强化学习：中文版强化学习教程，简洁易懂，但是还不完善，完整版看其出版的书
Reinforcement Learning: An Introduction：英文版原著，比较完整，偏理论，配套代码相比上一个来说更为一般化。例如，多臂老虎机的非稳定性问题，即每个杆的奖励不是固定的以及其他讨论更为详细。
深度强化学习：待学习跟进

多臂老虎机 (Multi-Armed Bandit, MAB)

Code

import numpy as np
import matplotlib.pyplot as plt

class BernoulliBandit:
    """ 伯努利多臂老虎机，输入K为拉杆个数 """
    def __init__(self, K):
        self.probs = np.random.uniform(size=K) # 随机生成K个0-1之间的数，作为每个拉杆的获奖概率
        self.best_idx = np.argmax(self.probs) # 获奖概率最大的拉杆
        self.best_prob = self.probs[self.best_idx] # 最大的获奖概率
        self.K = K

    def step(self, k):
        # 当玩家选择了k号拉杆后，根据该老虎机k号拉杆获得奖励的概率返回 1（获奖）或 0（未获奖）
        if np.random.rand() < self.probs[k]:
            return 1
        else:
            return 0

np.random.seed(1) # 设定随机种子，使实验具有可重复性
K = 10
bandit_10_arm = BernoulliBandit(K)
print("随机生成了一个 %d臂伯努利老虎机" % K)
print("获得奖励概率最大的拉杆为%d号，其获奖概率为%.4f" % (bandit_10_arm.best_idx, bandit_10_arm.best_prob))


class Solver:
    """多臂老虎机算法基本框架"""
    def __init__(self, bandit):
        self.bandit = bandit
        self.counts = np.zeros(self.bandit.K)  # 每个拉杆的尝试次数
        self.regret = 0.0  # 当前步的累积懊悔
        self.actions = []  # 维护一个列表，记录每一步的动作
        self.regrets = []  # 维护一个列表，记录每一步的累积懊悔

    def update_regret(self, k):
        # 计算累积懊悔并保存，k为本次行动选择的拉杆的编号
        self.regret += self.bandit.best_prob - self.bandit.probs[k]
        self.regrets.append(self.regret)

    def run_one_step(self):
        # 返回当前行动选择哪一个拉杆，由每个具体的策略实现
        return NotImplementedError

    def run(self, num_steps):
        # 运行一定次数，num_steps为总运行次数
        for _ in range(num_steps):
            k = self.run_one_step()
            self.counts[k] += 1
            self.actions.append(k)
            self.update_regret(k)


class EpsilonGreedy(Solver):
    """Epsilon贪心算法，继承Solver类"""
    def __init__(self, bandit, epsilon=0.01, init_prob=1.0):
        super(EpsilonGreedy, self).__init__(bandit)
        self.epsilon = epsilon
        self.estimates = np.array([init_prob] * self.bandit.K)  # 初始化对所有拉杆的奖励估值

    def run_one_step(self):
        if np.random.random() < self.epsilon:
            k = np.random.randint(0, self.bandit.K)
        else:
            k = np.argmax(self.estimates)  # 选择奖励估值最大的拉杆
        r = self.bandit.step(k)
        self.estimates[k] += 1.0 / (self.counts[k] + 1) * (r - self.estimates[k])
        return k


class DecayingEpsilonGreedy(Solver):
    """epsilon取值随时间衰减的Epsilon贪心算法，继承Solver类"""
    def __init__(self, bandit, init_prob=1.0):
        super(DecayingEpsilonGreedy, self).__init__(bandit)
        self.estimates = np.array([init_prob] * self.bandit.K)
        self.total_count = 0

    def run_one_step(self):
        self.total_count += 1
        if np.random.random() < 1 / self.total_count:  # epsilon随时间衰减
            k = np.random.randint(0, self.bandit.K)
        else:
            k = np.argmax(self.estimates)

        r = self.bandit.step(k)
        self.estimates[k] += 1.0 / (self.counts[k] + 1) * (r - self.estimates[k])
        return k


class UCB(Solver):
    """UCB算法，继承Solver类"""
    def __init__(self, bandit, coef, init_prob=1.0):
        super(UCB, self).__init__(bandit)
        self.total_count = 0
        self.estimates = np.array([init_prob] * self.bandit.K)
        self.coef = coef

    def run_one_step(self):
        self.total_count += 1
        ucb = self.estimates + self.coef * np.sqrt(np.log(self.total_count) / (2 * self.counts + 1))  # 计算置信上界
        k = np.argmax(ucb)  # 选出置信上界最大的拉杆
        r = self.bandit.step(k)
        self.estimates[k] += 1.0 / (self.counts[k] + 1) * (r - self.estimates[k])
        return k


class ThompsonSampling(Solver):
    """汤普森采样算法，继承Solver类"""
    def __init__(self, bandit):
        super(ThompsonSampling, self).__init__(bandit)
        self._a = np.ones(self.bandit.K)  # 列表，表示每个拉杆奖励为1的次数
        self._b = np.ones(self.bandit.K)  # 列表，表示每个拉杆奖励为0的次数

    def run_one_step(self):
        samples = np.random.beta(self._a, self._b)  # 按照Beta分布采样一组奖励
        k = np.argmax(samples)  # 选出采样数值最大的拉杆
        r = self.bandit.step(k)

        self._a[k] += r  # 更新Beta分布的第一个参数
        self._b[k] += (1 - r)  # 更新Beta分布的第二个参数

        return k


def plot_results(solvers, solver_names):
    """
    生成累积懊悔随时间变化的图像。
    输入solvers是一个列表，列表中每个元素是一种特定的策略。
    而solver_names也是一个列表，包含每个策略的名称
    """
    for idx, solver in enumerate(solvers):
        time_list = range(len(solver.regrets))
        plt.plot(time_list, solver.regrets, label=solver_names[idx])
    plt.xlabel("Time steps")
    plt.ylabel("Cumulative regrets")
    plt.title("%d-armed bandit" % solvers[0].bandit.K)
    plt.legend()
    plt.show()


# 1次epsilon贪心算法
np.random.seed(1)
epsilon_greedy_solver = EpsilonGreedy(bandit_10_arm, epsilon=0.01)
epsilon_greedy_solver.run(5000)
print('epsilon贪心算法的累积懊悔为：', epsilon_greedy_solver.regret)
plot_results([epsilon_greedy_solver], ['EpsilonGreedy'])

# 多次epsilon贪心算法
np.random.seed(0)
epsilons = [1e-4, 0.01, 0.1, 0.25, 0.5]
epsilon_greedy_solver_list = [EpsilonGreedy(bandit_10_arm, epsilon=e) for e in epsilons]
epsilon_greedy_solver_names = ["epsilon={}".format(e) for e in epsilons]
for solver in epsilon_greedy_solver_list:
    solver.run(5000)
plot_results(epsilon_greedy_solver_list, epsilon_greedy_solver_names)

# 1次epsilon衰减贪心算法
np.random.seed(1)
decaying_epsilon_greedy_solver = DecayingEpsilonGreedy(bandit_10_arm)
decaying_epsilon_greedy_solver.run(5000)
print('epsilon衰减贪心算法的累积懊悔：', decaying_epsilon_greedy_solver.regret)
plot_results([decaying_epsilon_greedy_solver], ['DecayingEpsilonGreedy'])

# Upper Confidence Bound算法
np.random.seed(1)
coef = 0.1  # 控制不确定性比重的系数
UCB_solver = UCB(bandit_10_arm, coef)
UCB_solver.run(5000)
print('置信上界算法的累积懊悔为：', UCB_solver.regret)
plot_results([UCB_solver], ['UCB'])

# 汤普森采样算法
np.random.seed(1)
thompson_sampling_solver = ThompsonSampling(bandit_10_arm)
thompson_sampling_solver.run(5000)
print('汤普森采样算法的累积懊悔为：',  thompson_sampling_solver.regret)
plot_results([thompson_sampling_solver], ['ThompsonSampling'])

Results

随机生成了一个 10臂伯努利老虎机
获得奖励概率最大的拉杆为1号，其获奖概率为0.7203
epsilon贪心算法的累积懊悔为： 25.526630933945313
epsilon衰减贪心算法的累积懊悔： 10.114334931260183
置信上界算法的累积懊悔为： 4.360254345326592
汤普森采样算法的累积懊悔为： 57.19161964443925

马尔科夫决策过程 (Markov Decision Process)

Code

import numpy as np

######################## 马尔科夫奖励过程 ########################
np.random.seed(0)
# 定义状态转移概率矩阵P
P = [[0.9, 0.1, 0.0, 0.0, 0.0, 0.0],
     [0.5, 0.0, 0.5, 0.0, 0.0, 0.0],
     [0.0, 0.0, 0.0, 0.6, 0.0, 0.4],
     [0.0, 0.0, 0.0, 0.0, 0.3, 0.7],
     [0.0, 0.2, 0.3, 0.5, 0.0, 0.0],
     [0.0, 0.0, 0.0, 0.0, 0.0, 1.0]]

P = np.array(P)

rewards = [-1, -2, -2, 10, 1, 0]  # 定义奖励函数
gamma = 0.5  # 定义折扣因子


# 给定一条序列，计算从某个索引开始到序列最后得到的回报
def compute_return(start_index, chain, gamma):
    G = 0
    for i in reversed(range(start_index, len(chain))):
        G = gamma * G + rewards[chain[i] - 1]
    return G


# 一个状态序列，s1 -> s2 -> s3 -> s6
chain = [1, 2, 3, 6]
start_index = 0
G = compute_return(start_index, chain, gamma)
print("根据本序列计算得到回报为：%s" % G)


def compute(P, rewards, gamma, states_num):
    """利用贝尔曼方程矩阵形式计算解析解，states_num是MRP的状态数"""
    rewards = np.array(rewards).reshape((-1, 1))  # rewards写成列向量形式
    value = np.dot(np.linalg.inv(np.eye(states_num, states_num) - gamma * P), rewards)
    return value


V = compute(P, rewards, gamma, 6)
print("MRP中每个状态价值分别为\n", V)

######################## 马尔科夫决策过程 ########################
S = ["S1", "S2", "S3", "S4", "S5"]  # 状态集合
A = ["保持S1", "前往S1", "前往S2", "前往S3", "前往S4", "前往S5", "概率前往"]  # 动作集合
# 状态转移函数
P = {
    "S1-保持S1-S1": 1.0, "S1-前往S2-S2": 1.0,
    "S2-前往S1-S1": 1.0, "S2-前往S3-S3": 1.0,
    "S3-前往S4-S4": 1.0, "S3-前往S5-S5": 1.0,
    "S4-前往S5-S5": 1.0, "S4-概率前往-S2": 0.2,
    "S4-概率前往-S3": 0.4, "S4-概率前往-S4": 0.4,
}
# 奖励函数
R = {
    "S1-保持S1": -1, "S1-前往S2": 0,
    "S2-前往S1": -1, "S2-前往S3": -2,
    "S3-前往S4": -2, "S3-前往S5": 0,
    "S4-前往S5": 10, "S4-概率前往": 1
}

gamma = 0.5  # 折扣因子
MDP = (S, A, P, R, gamma)

# 策略1，随机策略
Pi_1 = {
    "S1-保持S1": 0.5, "S1-前往S2": 0.5,
    "S2-前往S1": 0.5, "S2-前往S3": 0.5,
    "S3-前往S4": 0.5, "S3-前往S5": 0.5,
    "S4-前往S5": 0.5, "S4-概率前往": 0.5
}
# 策略2
Pi_2 = {
    "S1-保持S1": 0.6, "S1-前往S2": 0.4,
    "S2-前往S1": 0.3, "S2-前往S3": 0.7,
    "S3-前往S4": 0.5, "S3-前往S5": 0.5,
    "S4-前往S5": 0.1, "S4-概率前往": 0.9
}


# 把输入的两个字符串通过“-”连接，便于使用我们上述定义的P,R变量
def join(str1, str2):
    return str1 + '-' + str2


gamma = 0.5
# 转化后的MRP的状态转移矩阵
# 策略1
P_from_mdp_to_mrp = [
    [0.5, 0.5, 0.0, 0.0, 0.0],
    [0.5, 0.0, 0.5, 0.0, 0.0],
    [0.0, 0.0, 0.0, 0.5, 0.5],
    [0.0, 0.1, 0.2, 0.2, 0.5],
    [0.0, 0.0, 0.0, 0.0, 1.0]
]
P_from_mdp_to_mrp = np.array(P_from_mdp_to_mrp)
R_from_mdp_to_mrp = [-0.5, -1.5, -1.0, 5.5, 0]
V = compute(P_from_mdp_to_mrp, R_from_mdp_to_mrp, gamma, 5)
print("MDP中每个状态价值分别为\n", V)


def sample(MDP, Pi, timestep_max, number):
    """采样函数，策略Pi，限制最长时间步timestep_max，总共采样序列数number"""
    S, A, P, R, gamma = MDP
    episodes = []
    for _ in range(number):
        episode = []
        timestep = 0
        s = S[np.random.randint(4)]  # 随机选择一个除了s5以外的状态作为起点
        while s != "S5" and timestep <= timestep_max:  # 当前状态为终止状态或者时间步太长时，一次采样结束
            timestep += 1
            rand, temp = np.random.rand(), 0
            # 在状态s根据策略选择动作
            for a_opt in A:
                temp += Pi.get(join(s, a_opt), 0)
                if temp > rand:  # 根据概率选择动作
                    a = a_opt
                    r = R.get(join(s, a), 0)
                    break
            rand, temp = np.random.rand(), 0
            # 根据状态转移概率得到下一个状态s_next
            for s_opt in S:
                temp += P.get(join(join(s, a), s_opt), 0)
                if temp > rand:
                    s_next = s_opt
                    break
            episode.append((s, a, r, s_next))  # 把(s, a, r, s_next)元组放入序列中
            s = s_next  # s_next变成当前状态，开始接下来的循环
        episodes.append(episode)
    return episodes


# 采样5次，每个序列最长不超过1000步
episodes = sample(MDP, Pi_1, 20, 5)
print('第一条序列\n', episodes[0])
print('第二条序列\n', episodes[1])
print('第五条序列\n', episodes[4])


# 对所有采样序列计算所有状态的价值
def MC(episodes, V, N, gamma):
    for episode in episodes:
        G = 0
        for i in range(len(episode) - 1, - 1, -1):  ## 一个序列从后往前计算
            s, a, r, s_next = episode[i]
            G = r + gamma * G
            N[s] = N[s] + 1
            V[s] = V[s] + (G - V[s]) / N[s]


timestep_max = 20
# 采样1000次，可以自行修改
episodes = sample(MDP, Pi_1, timestep_max, 1000)
gamma = 0.5
V = {"S1": 0, "S2": 0, "S3": 0, "S4": 0, "S5": 0}
N = {"S1": 0, "S2": 0, "S3": 0, "S4": 0, "S5": 0}
MC(episodes, V, N, gamma)
print("使用蒙特卡洛法计算MDP状态价值为\n", V)


def occupancy(episodes, s, a, timestep_max, gamma):
    """计算状态-动作对(s, a)出现的概率，以此来估算策略的占用度量"""
    rho = 0
    total_times = np.zeros(timestep_max)  # 记录每个时间步t各被经历过几次
    occur_times = np.zeros(timestep_max)  # 记录(s_t, a_t)=(s,a)的次数
    for episode in episodes:
        for i in range(len(episode)):
            s_opt, a_opt, r, s_next = episode[i]
            total_times[i] += 1
            if s == s_opt and a == a_opt:
                occur_times[i] += 1

    for i in reversed(range(timestep_max)):
        if total_times[i]:
            rho += gamma ** i * occur_times[i] / total_times[i]

    return (1 - gamma) * rho


gamma = 0.5
timestep_max = 1000

episodes_1 = sample(MDP, Pi_1, timestep_max, 1000)
episodes_2 = sample(MDP, Pi_2, timestep_max, 1000)
rho_1 = occupancy(episodes_1, "S4", "概率前往", timestep_max, gamma)
rho_2 = occupancy(episodes_2, "S4", "概率前往", timestep_max, gamma)
print(rho_1, rho_2)

Results

根据本序列计算得到回报为：-2.5
MRP中每个状态价值分别为
 [[-2.01950168]
 [-2.21451846]
 [ 1.16142785]
 [10.53809283]
 [ 3.58728554]
 [ 0.        ]]
MDP中每个状态价值分别为
 [[-1.22555411]
 [-1.67666232]
 [ 0.51890482]
 [ 6.0756193 ]
 [ 0.        ]]
第一条序列
 [('S1', '前往S2', 0, 'S2'), ('S2', '前往S3', -2, 'S3'), ('S3', '前往S5', 0, 'S5')]
第二条序列
 [('S4', '概率前往', 1, 'S4'), ('S4', '前往S5', 10, 'S5')]
第五条序列
 [('S2', '前往S3', -2, 'S3'), ('S3', '前往S4', -2, 'S4'), ('S4', '前往S5', 10, 'S5')]
使用蒙特卡洛法计算MDP状态价值为
 {'S1': -1.228923788722258, 'S2': -1.6955696284402704, 'S3': 0.4823809701532294, 'S4': 5.967514743019431, 'S5': 0}
0.112567796310472 0.23199480615618912

动态规划算法 (Dynamic Programming)

Code

import copy


class CliffWalkingEnv:
    """ Cliff Walking环境"""

    def __init__(self, ncol=12, nrow=4):
        self.ncol = ncol  # 定义环境的宽
        self.nrow = nrow  # 定义环境的高
        self.P = self.createP()  # 转移矩阵P[state][action] = [(p, next_state, reward, done)]，包含下一个状态和奖励

    def createP(self):
        P = [[[] for j in range(4)] for i in range(self.nrow * self.ncol)]  # 初始化
        change = [[0, -1], [0, 1], [-1, 0], [1, 0]]  # 4 种动作, 0:上, 1:下, 2:左, 3:右。原点(0,0)定义在左上角
        for i in range(self.nrow):
            for j in range(self.ncol):
                for a in range(4):
                    if i == self.nrow - 1 and j > 0:  # 位置在悬崖或者终点，因为无法继续交互，任何动作奖励都为0
                        P[i * self.ncol + j][a] = [(1, i * self.ncol + j, 0, True)]
                        continue
                    # 其他位置
                    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


# 策略迭代
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 True:
            max_diff = 0
            new_v = [0] * self.env.ncol * self.env.nrow
            for s in range(self.env.ncol * self.env.nrow):
                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(self.pi[s][a] * qsa)
                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)
            self.pi[s] = [1 / cntq if q == maxq else 0 for q in qsa_list]  # 让相同的动作价值均分概率
        print("策略提升完成")
        return self.pi

    def policy_iteration(self):  # 策略迭代
        while True:
            self.policy_evaluation()
            old_pi = copy.deepcopy(self.pi)  # 将列表进行深拷贝，方便接下来进行比较
            new_pi = self.policy_improvement()
            if old_pi == new_pi:
                break


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

    print("策略：")
    for i in range(agent.env.nrow):
        for j in range(agent.env.ncol):
            if (i * agent.env.ncol + j) in disaster:  # 一些特殊的状态，例如Cliff Walking中的悬崖
                print("****", end=" ")
            elif(i * agent.env.ncol + j) in end:  # 终点
                print("EEEE", end=" ")
            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])


# 价值迭代
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 True:
            max_diff = 0
            new_v = [0] * self.env.ncol * self.env.nrow
            for s in range(self.env.ncol * self.env.nrow):
                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)   # 这一行和下一行是和策略迭代的主要区别
                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 += r + p * self.gamma * self.v[next_state] * (1 - done)
                qsa_list.append(qsa)
            maxq = max(qsa_list)
            cntq = qsa_list.count(maxq)
            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])


import gym

env = gym.make("FrozenLake-v1")  # 创建环境, gym==0.19.0版本已经更新了FrozenLake的版本
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]:
            if s_[2] == 1.0:  # 获得奖励为1，代表是终点
                ends.add(s_[1])
            if s_[3] is True:
                holes.add(s_[1])
holes = holes - ends
print("冰洞的索引：", holes)
print("终点的索引", ends)

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


action_meaning = ['<', 'v', '>', '^']  # 这个动作意义是gym库中对FrozenLake这个环境事先规定好的
theta = 1e-5
gamma = 0.9
agent = PolicyIteration(env, theta, gamma)
agent.policy_iteration()
print_agent(agent, action_meaning, [5, 7, 11, 12], [15])

agent = ValueIteration(env, theta, gamma)
agent.value_iteration()
print_agent(agent, action_meaning, [5, 7, 11, 12], [15])

Results

D:\Anaconda3\envs\torch\python.exe D:/PythonProject/reinforcement_learning/03_Dynamic_Programming.py
策略评估进行60轮后完成
策略提升完成
策略评估进行72轮后完成
策略提升完成
策略评估进行44轮后完成
策略提升完成
策略评估进行12轮后完成
策略提升完成
策略评估进行1轮后完成
策略提升完成
状态价值：
-7.712 -7.458 -7.176 -6.862 -6.513 -6.126 -5.695 -5.217 -4.686 -4.095 -3.439 -2.710 
-7.458 -7.176 -6.862 -6.513 -6.126 -5.695 -5.217 -4.686 -4.095 -3.439 -2.710 -1.900 
-7.176 -6.862 -6.513 -6.126 -5.695 -5.217 -4.686 -4.095 -3.439 -2.710 -1.900 -1.000 
-7.458  0.000  0.000  0.000  0.000  0.000  0.000  0.000  0.000  0.000  0.000  0.000 
策略：
ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovoo 
ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovoo 
ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ovoo 
^ooo **** **** **** **** **** **** **** **** **** **** EEEE 

价值迭代一共进行14轮
状态价值：
-7.712 -7.458 -7.176 -6.862 -6.513 -6.126 -5.695 -5.217 -4.686 -4.095 -3.439 -2.710 
-7.458 -7.176 -6.862 -6.513 -6.126 -5.695 -5.217 -4.686 -4.095 -3.439 -2.710 -1.900 
-7.176 -6.862 -6.513 -6.126 -5.695 -5.217 -4.686 -4.095 -3.439 -2.710 -1.900 -1.000 
-7.458  0.000  0.000  0.000  0.000  0.000  0.000  0.000  0.000  0.000  0.000  0.000 
策略：
ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovoo 
ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovo> ovoo 
ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ovoo 
^ooo **** **** **** **** **** **** **** **** **** **** EEEE 

SFFF
FHFH
FFFH
HFFG
冰洞的索引： {11, 12, 5, 7}
终点的索引 {15}
[(0.3333333333333333, 10, 0.0, False), (0.3333333333333333, 13, 0.0, False), (0.3333333333333333, 14, 0.0, False)]
[(0.3333333333333333, 13, 0.0, False), (0.3333333333333333, 14, 0.0, False), (0.3333333333333333, 15, 1.0, True)]
[(0.3333333333333333, 14, 0.0, False), (0.3333333333333333, 15, 1.0, True), (0.3333333333333333, 10, 0.0, False)]
[(0.3333333333333333, 15, 1.0, True), (0.3333333333333333, 10, 0.0, False), (0.3333333333333333, 13, 0.0, False)]

策略评估进行25轮后完成
策略提升完成
策略评估进行58轮后完成
策略提升完成
状态价值：
 0.069  0.061  0.074  0.056 
 0.092  0.000  0.112  0.000 
 0.145  0.247  0.300  0.000 
 0.000  0.380  0.639  0.000 
策略：
<ooo ooo^ <ooo ooo^ 
<ooo **** <o>o **** 
ooo^ ovoo <ooo **** 
**** oo>o ovoo EEEE 

价值迭代一共进行60轮
状态价值：
 0.069  0.061  0.074  0.056 
 0.092  0.000  0.112  0.000 
 0.145  0.247  0.300  0.000 
 0.000  0.380  0.639  0.000 
策略：
<ooo ooo^ <ooo ooo^ 
<ooo **** <o>o **** 
ooo^ ovoo <ooo **** 
**** oo>o ovoo EEEE 

Process finished with exit code 0

时序差分(Temporal Difference)算法

Code

import matplotlib.pyplot as plt
import numpy as np
from tqdm import tqdm


class CliffWalkingEnv:
    def __init__(self, ncol, nrow):
        self.nrow = nrow
        self.ncol = ncol
        self.x = 0  # 记录当前智能体位置的横坐标
        self.y = self.nrow - 1  # 记录当前智能体位置的纵坐标

    def step(self, action):  # 外部调用这个函数来让当前位置改变
        change = [[0, -1], [0, 1], [-1, 0], [1, 0]]  # 4种动作，上下左右。圆点(0,0)定义在左上角
        self.x = min(self.ncol - 1, max(0, self.x + change[action][0]))
        self.y = min(self.nrow - 1, max(0, self.y + change[action][1]))
        next_state = self.y * self.ncol + self.x
        reward = -1
        done = False
        if self.y == self.nrow - 1 and self.x > 0:  # 下一个位置在悬崖或者终点
            done = True
            if self.x != self.ncol - 1:
                reward = -100

        return next_state, reward, done

    def reset(self):  # 回归初试状态，坐标轴原点在左上角
        self.x = 0
        self.y = self.nrow - 1
        return self.y * self.ncol + self.x


class Sarsa:
    """Sarsa算法"""
    def __init__(self, ncol, nrow, epsilon, alpha, gamma, n_action=4):
        self.Q_table = np.zeros([nrow * ncol, n_action])  # 初始化Q(s, a)表格，默认终止状态时所有动作的价值都是0
        self.n_action = n_action  # 动作个数
        self.alpha = alpha  # 学习率
        self.gamma = gamma  # 折扣因子
        self.epsilon = epsilon  # epsilon-greedy策略中的参数

    def take_action(self, state):  # 选取下一步的操作
        if np.random.random() < self.epsilon:
            action = np.random.randint(self.n_action)
        else:
            action = np.argmax(self.Q_table[state])
        return action

    def best_action(self, state):  # 用于打印策略
        Q_max = np.max(self.Q_table[state])
        a = [0 for _ in range(self.n_action)]
        for i in range(self.n_action):  # 若两个动作的价值一样，都会记录下来
            if self.Q_table[state, i] == Q_max:
                a[i] = 1
        return a

    def update(self, s0, a0, r, s1, a1):
        td_error = r + self.gamma * self.Q_table[s1, a1] - self.Q_table[s0, a0]
        self.Q_table[s0, a0] += self.alpha * td_error


class nstep_Sarsa:
    """n步Sarsa算法"""
    def __init__(self, n, ncol, nrow, epsilon, alpha, gamma, n_action=4):
        self.Q_table = np.zeros([nrow * ncol, n_action])
        self.n_action = n_action
        self.alpha = alpha
        self.gamma = gamma
        self.epsilon = epsilon
        self.n = n  # 采用n步Sarsa
        self.state_list = []  # 保存之前的状态
        self.action_list = []  # 保存之前的动作
        self.reward_list = []  # 保存之前的奖励

    def take_action(self, state):
        if np.random.random() < self.epsilon:
            action = np.random.randint(self.n_action)
        else:
            action = np.argmax(self.Q_table[state])
        return action

    def best_action(self, state):  # 用于打印策略
        Q_max = np.max(self.Q_table[state])
        a = [0 for _ in range(self.n_action)]
        for i in range(self.n_action):
            if self.Q_table[state, i] == Q_max:
                a[i] = 1
        return a

    def update(self, s0, a0, r, s1, a1, done):
        self.state_list.append(s0)
        self.action_list.append(a0)
        self.reward_list.append(r)
        if len(self.state_list) == self.n:  # 若保存的数据可以进行n步更新
            G = self.Q_table[s1, a1]  # 得到Q(s_{t+n}, a_{t+n})
            for i in reversed(range(self.n)):
                G = self.gamma * G + self.reward_list[i]  # b不断向前计算每一步的回报
                if done and i > 0:  # 如果到达终止状态，最后几步虽然长度不够n步，我们也对其进行更新
                    s = self.state_list[i]
                    a = self.action_list[i]
                    self.Q_table[s, a] += self.alpha * (G - self.Q_table[s, a])
            s = self.state_list.pop(0)  # 需要更新的状态动作从列表中删除，下次不必更新
            a = self.action_list.pop(0)
            self.reward_list.pop(0)
            self.Q_table[s, a] += self.alpha * (G - self.Q_table[s, a])  # n-step sarsa的主要更新步骤

        if done:  # 如果到达终止状态，即将开始下一条序列，则将列表全清空
            self.state_list = []
            self.action_list = []
            self.reward_list = []


class QLearning:
    """Q-learning算法"""
    def __init__(self, ncol, nrow, epsilon, alpha, gamma, n_action=4):
        self.Q_table = np.zeros([nrow * ncol, n_action])  # 初始化Q(s, a)表格
        self.n_action = n_action  # 动作个数
        self.alpha = alpha  # 学习率
        self.gamma = gamma  # 折扣因子
        self.epsilon = epsilon  # epison-greedy策略中的参数

    def take_action(self, state):  # 选取下一步的 操作
        if np.random.random() < self.epsilon:
            action = np.random.randint(self.n_action)
        else:
            action = np.argmax(self.Q_table[state])
        return action

    def best_action(self, state):  # 用于打印策略
        Q_max = np.max(self.Q_table[state])
        a = [0 for _ in range(self.n_action)]
        for i in range(self.n_action):
            if self.Q_table[state, i] == Q_max:
                a[i] = 1
        return a

    def update(self, s0, a0, r, s1):
        td_error = r + self.gamma * self.Q_table[s1].max() - self.Q_table[s0, a0]
        self.Q_table[s0, a0] += self.alpha * td_error


def print_agent(agent, env, action_meaning, disaster=[], end=[]):
    for i in range(env.nrow):
        for j in range(env.ncol):
            if (i * env.ncol + j) in disaster:
                print('****', end=' ')
            elif (i * env.ncol + j) in end:
                print('EEEE', end=' ')
            else:
                a = agent.best_action(i * 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()


def print_return_and_action(env=None, agent=None, alg_name=None):
    num_episodes = 500  # 智能体在环境中运行多少条序列

    return_list = []  # 记录每一条序列的回报
    for i in range(10):  # 显示10个进度条
        with tqdm(total=int(num_episodes / 10), desc="Iteration %d" % i) as pbar:  # tqdm的进度条功能
            for i_episode in range(int(num_episodes / 10)):  # 每个进度条的序列数
                episode_return = 0
                state = env.reset()
                if alg_name != 'Q-learning':
                    action = agent.take_action(state)
                done = False
                while not done:
                    if alg_name != 'Q-learning':
                        next_state, reward, done = env.step(action)
                        next_action = agent.take_action(next_state)
                    else:
                        action = agent.take_action(state)
                        next_state, reward, done = env.step(action)
                    episode_return += reward  # 这里回报的计算不进行折扣因子衰减
                    if alg_name == 'Sarsa':
                        agent.update(state, action, reward, next_state, next_action)
                    elif alg_name == '5-step Sarsa':
                        agent.update(state, action, reward, next_state, next_action, done)
                    elif alg_name == 'Q-learning':
                        agent.update(state, action, reward, next_state)  # 注意这里跟Sarsa的区别，不再是用next_action
                    state = next_state
                    if alg_name != 'Q-learning':
                        action = next_action
                return_list.append(episode_return)
                if (i_episode + 1) % 10 == 0:  # 每10条序列打印一下这10条序列的平均回报
                    pbar.set_postfix({'episode': '%d' % (num_episodes / 10 * i + i_episode + 1),
                                      'return': '%.3f' % np.mean(return_list[-10:])})
                pbar.update(1)

    episodes_list = list(range(len(return_list)))
    plt.plot(episodes_list, return_list)
    plt.xlabel('Episodes')
    plt.ylabel('Returns')
    plt.title('{} on {}'.format(alg_name, 'Cliff Walking'))
    plt.show()

    action_meaning = ['^', 'v', '<', '>']
    print('%s算法最终收敛得到的策略为：' % alg_name)
    print_agent(agent, env, action_meaning, list(range(37, 47)), [47])


np.random.seed(0)
ncol = 12
nrow = 4
env = CliffWalkingEnv(ncol, nrow)
epsilon = 0.1
alpha = 0.1
gamma = 0.9
agent = Sarsa(ncol, nrow, epsilon, alpha, gamma)
print_return_and_action(env, agent, alg_name='Sarsa')

n_step = 5
agent = nstep_Sarsa(n_step, ncol, nrow, epsilon, alpha, gamma)
print_return_and_action(env, agent, alg_name='5-step Sarsa')

agent = QLearning(ncol, nrow, epsilon, alpha, gamma)
print_return_and_action(env, agent, alg_name='Q-learning')

Results

D:\Anaconda3\envs\torch\python.exe D:/PythonProject/reinforcement_learning/04_Temporal_Difference.py
Iteration 0: 100%|██████████| 50/50 [00:00<00:00, 1598.78it/s, episode=50, return=-119.400]
Iteration 1: 100%|██████████| 50/50 [00:00<00:00, 3207.24it/s, episode=100, return=-63.000]
Iteration 2: 100%|██████████| 50/50 [00:00<00:00, 2256.60it/s, episode=150, return=-51.200]
Iteration 3: 100%|██████████| 50/50 [00:00<?, ?it/s, episode=200, return=-48.100]
Iteration 4: 100%|██████████| 50/50 [00:00<00:00, 3194.54it/s, episode=250, return=-35.700]
Iteration 5: 100%|██████████| 50/50 [00:00<00:00, 3199.71it/s, episode=300, return=-29.900]
Iteration 6: 100%|██████████| 50/50 [00:00<?, ?it/s, episode=350, return=-28.300]
Iteration 7: 100%|██████████| 50/50 [00:00<00:00, 3200.78it/s, episode=400, return=-27.700]
Iteration 8: 100%|██████████| 50/50 [00:00<?, ?it/s, episode=450, return=-28.500]
Iteration 9: 100%|██████████| 50/50 [00:00<?, ?it/s, episode=500, return=-18.900]
Iteration 0:  18%|█▊        | 9/50 [00:00<00:00, 574.69it/s, episode=10, return=-238.900]Sarsa算法最终收敛得到的策略为：
ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ovoo 
ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ovoo 
^ooo ooo> ^ooo ooo> ooo> ooo> ooo> ^ooo ^ooo ooo> ooo> ovoo 
^ooo **** **** **** **** **** **** **** **** **** **** EEEE 
Iteration 0: 100%|██████████| 50/50 [00:00<00:00, 936.23it/s, episode=50, return=-23.800]
Iteration 1: 100%|██████████| 50/50 [00:00<00:00, 1600.12it/s, episode=100, return=-18.900]
Iteration 2: 100%|██████████| 50/50 [00:00<?, ?it/s, episode=150, return=-18.800]
Iteration 3: 100%|██████████| 50/50 [00:00<00:00, 3200.93it/s, episode=200, return=-19.600]
Iteration 4: 100%|██████████| 50/50 [00:00<00:00, 3194.64it/s, episode=250, return=-19.100]
Iteration 5: 100%|██████████| 50/50 [00:00<00:00, 3206.80it/s, episode=300, return=-28.200]
Iteration 6: 100%|██████████| 50/50 [00:00<00:00, 2256.19it/s, episode=350, return=-27.400]
Iteration 7: 100%|██████████| 50/50 [00:00<?, ?it/s, episode=400, return=-19.800]
Iteration 8: 100%|██████████| 50/50 [00:00<00:00, 2263.37it/s, episode=450, return=-27.400]
Iteration 9: 100%|██████████| 50/50 [00:00<00:00, 5013.75it/s, episode=500, return=-27.700]
Iteration 0:  58%|█████▊    | 29/50 [00:00<00:00, 1321.99it/s, episode=30, return=-165.400]5-step Sarsa算法最终收敛得到的策略为：
ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ovoo ovoo 
^ooo ^ooo ^ooo oo<o ^ooo ^ooo ^ooo ooo> ^ooo ^ooo ooo> ovoo 
^ooo ooo> ^ooo oo<o ^ooo oo<o oo<o ^ooo oo<o ^ooo ooo> ovoo 
^ooo **** **** **** **** **** **** **** **** **** **** EEEE 
Iteration 0: 100%|██████████| 50/50 [00:00<00:00, 1285.44it/s, episode=50, return=-114.400]
Iteration 1: 100%|██████████| 50/50 [00:00<00:00, 1790.43it/s, episode=100, return=-80.300]
Iteration 2: 100%|██████████| 50/50 [00:00<00:00, 2278.77it/s, episode=150, return=-67.100]
Iteration 3: 100%|██████████| 50/50 [00:00<00:00, 2949.21it/s, episode=200, return=-38.000]
Iteration 4: 100%|██████████| 50/50 [00:00<00:00, 3867.71it/s, episode=250, return=-43.900]
Iteration 5: 100%|██████████| 50/50 [00:00<00:00, 4164.49it/s, episode=300, return=-28.700]
Iteration 6: 100%|██████████| 50/50 [00:00<00:00, 3682.90it/s, episode=350, return=-36.200]
Iteration 7: 100%|██████████| 50/50 [00:00<00:00, 5548.76it/s, episode=400, return=-15.800]
Iteration 8: 100%|██████████| 50/50 [00:00<00:00, 7199.04it/s, episode=450, return=-51.500]
Iteration 9: 100%|██████████| 50/50 [00:00<00:00, 7162.65it/s, episode=500, return=-60.300]
Q-learning算法最终收敛得到的策略为：
^ooo oo<o oo<o ooo> ooo> ooo> ^ooo ooo> ooo> ooo> ovoo ovoo 
oo<o ^ooo ^ooo ooo> ooo> ooo> ^ooo ooo> ovoo ooo> ooo> ovoo 
ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ooo> ovoo 
^ooo **** **** **** **** **** **** **** **** **** **** EEEE 

Process finished with exit code 0

Dyna-Q算法

Code

import matplotlib.pyplot as plt
import numpy as np
from tqdm import tqdm
import random
import time


class CliffWalkingEnv:
    def __init__(self, ncol, nrow):
        self.nrow = nrow
        self.ncol = ncol
        self.x = 0              # 记录当前智能体位置的横坐标
        self.y = self.nrow - 1  # 记录当前智能体为孩子的纵坐标

    def step(self, action):  # 外部调用这个函数来让当前位置改变
        change = [[0, -1], [0, 1], [-1, 0], [1, 0]]  # 上下左右四个动作，原点左上角
        self.x = min(self.ncol - 1, max(0, self.x + change[action][0]))
        self.y = min(self.nrow - 1, max(0, self.y + change[action][1]))
        next_state = self.y * self.ncol + self.x
        reward = -1
        done = False
        if self.y == self.nrow - 1 and self.x > 0:  # 下一个位置在悬崖或者终点
            done = True
            if self.x != self.ncol - 1:
                reward = -100
        return next_state, reward, done

    def reset(self):  # 回归初始状态，起点在左上角
        self.x = 0
        self.y = self.nrow - 1
        return self.y * self.ncol + self.x


class DynaQ:
    """Dyna-Q算法"""
    def __init__(self, ncol, nrow, epsilon, alpha, gamma, n_planning, n_action=4):
        self.Q_table = np.zeros([nrow * ncol, n_action])  # 初始化Q(s, a)表格
        self.n_action = n_action  # 动作个数
        self.alpha = alpha  # 学习率
        self.gamma = gamma
        self.epsilon = epsilon  # epsilon-greedy策略中的参数
        self.n_planning = n_planning
        self.model = dict()

    def take_action(self, state):  # 选取下一步的操作
        if np.random.random() < self.epsilon:
            action = np.random.randint(self.n_action)
        else:
            action = np.argmax(self.Q_table[state])
        return action

    def q_learning(self, s0, a0, r, s1):
        td_error = r + self.gamma * self.Q_table[s1].max() - self.Q_table[s0, a0]
        self.Q_table[s0, a0] += self.alpha * td_error

    def update(self, s0, a0, r, s1):
        self.q_learning(s0, a0, r, s1)
        self.model[(s0, a0)] = r, s1  # 将数据添加到模型中
        for _ in range(self.n_planning):  # Q-planning循环
            (s, a), (r, s_) = random.choice(list(self.model.items()))  # 随机选择曾经遇到过的状态-动作对
            self.q_learning(s, a, r, s_)


def DynaQ_CliffWalking(n_planning):
    ncol = 12
    nrow = 4
    env = CliffWalkingEnv(ncol, nrow)
    epsilon = 0.01
    alpha = 0.1
    gamma = 0.9
    agent = DynaQ(ncol, nrow, epsilon, alpha, gamma, n_planning)
    num_episodes = 300  # 智能体在环境中运行多少条序列

    return_list = []  # 记录每一条序列的回报
    for i in range(10):  # 显示10个进度条
        with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:  # tqdm的进度条功能
            for i_episode in range(int(num_episodes / 10)):  # 每个进度条的序列数
                episode_return = 0
                state = env.reset()
                done = False
                while not done:
                    action = agent.take_action(state)
                    next_state, reward, done = env.step(action)
                    episode_return += reward  # 这里回报的计算不进行折扣因子衰减
                    agent.update(state, action, reward, next_state)
                    state = next_state
                return_list.append(episode_return)
                if (i_episode + 1) % 10 == 0:  # 每10条序列打印一下这10条序列的平均回报
                    pbar.set_postfix({'episode': '%d' % (num_episodes / 10 * i + i_episode+1), 'return': '%.3f' % np.mean(return_list[-10:])})
                pbar.update(1)
    return return_list


np.random.seed(0)
random.seed(0)
n_planning_list = [0, 2, 20]
for n_planning in n_planning_list:
    print('Q-planning步数为：%d' % n_planning)
    time.sleep(0.5)
    return_list = DynaQ_CliffWalking(n_planning)
    episodes_list = list(range(len(return_list)))
    plt.plot(episodes_list, return_list, label=str(n_planning) + ' planning steps')
plt.legend()
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('Dyna-Q on {}'.format('Cliff Walking'))
plt.show()

Results

D:\Anaconda3\envs\torch\python.exe D:/PythonProject/reinforcement_learning/05_Dyna-Q.py
Q-planning步数为：0
Iteration 0: 100%|██████████| 30/30 [00:00<00:00, 812.13it/s, episode=30, return=-138.400]
Iteration 1: 100%|██████████| 30/30 [00:00<00:00, 1307.93it/s, episode=60, return=-64.100]
Iteration 2: 100%|██████████| 30/30 [00:00<00:00, 1769.72it/s, episode=90, return=-46.000]
Iteration 3: 100%|██████████| 30/30 [00:00<00:00, 2153.54it/s, episode=120, return=-38.000]
Iteration 4: 100%|██████████| 30/30 [00:00<00:00, 2501.18it/s, episode=150, return=-28.600]
Iteration 5: 100%|██████████| 30/30 [00:00<00:00, 3015.68it/s, episode=180, return=-25.300]
Iteration 6: 100%|██████████| 30/30 [00:00<00:00, 3326.70it/s, episode=210, return=-23.600]
Iteration 7: 100%|██████████| 30/30 [00:00<00:00, 3764.18it/s, episode=240, return=-20.100]
Iteration 8: 100%|██████████| 30/30 [00:00<00:00, 5008.92it/s, episode=270, return=-17.100]
Iteration 9: 100%|██████████| 30/30 [00:00<00:00, 5023.12it/s, episode=300, return=-16.500]
Q-planning步数为：2
Iteration 0: 100%|██████████| 30/30 [00:00<00:00, 570.24it/s, episode=30, return=-53.800]
Iteration 1: 100%|██████████| 30/30 [00:00<00:00, 812.98it/s, episode=60, return=-37.100]
Iteration 2: 100%|██████████| 30/30 [00:00<00:00, 1157.06it/s, episode=90, return=-23.600]
Iteration 3: 100%|██████████| 30/30 [00:00<00:00, 1581.15it/s, episode=120, return=-18.500]
Iteration 4: 100%|██████████| 30/30 [00:00<00:00, 1582.62it/s, episode=150, return=-16.400]
Iteration 5: 100%|██████████| 30/30 [00:00<00:00, 1772.62it/s, episode=180, return=-16.400]
Iteration 6: 100%|██████████| 30/30 [00:00<00:00, 2144.03it/s, episode=210, return=-13.400]
Iteration 7: 100%|██████████| 30/30 [00:00<00:00, 2513.57it/s, episode=240, return=-13.200]
Iteration 8: 100%|██████████| 30/30 [00:00<00:00, 2307.86it/s, episode=270, return=-13.200]
Iteration 9: 100%|██████████| 30/30 [00:00<00:00, 2153.69it/s, episode=300, return=-13.500]
Q-planning步数为：20
Iteration 0: 100%|██████████| 30/30 [00:00<00:00, 205.97it/s, episode=30, return=-18.500]
Iteration 1: 100%|██████████| 30/30 [00:00<00:00, 323.45it/s, episode=60, return=-13.600]
Iteration 2: 100%|██████████| 30/30 [00:00<00:00, 376.01it/s, episode=90, return=-13.000]
Iteration 3: 100%|██████████| 30/30 [00:00<00:00, 376.15it/s, episode=120, return=-13.500]
Iteration 4: 100%|██████████| 30/30 [00:00<00:00, 349.67it/s, episode=150, return=-13.500]
Iteration 5: 100%|██████████| 30/30 [00:00<00:00, 347.17it/s, episode=180, return=-13.000]
Iteration 6: 100%|██████████| 30/30 [00:00<00:00, 376.00it/s, episode=210, return=-22.000]
Iteration 7: 100%|██████████| 30/30 [00:00<00:00, 362.56it/s, episode=240, return=-23.200]
Iteration 8: 100%|██████████| 30/30 [00:00<00:00, 385.51it/s, episode=270, return=-13.000]
Iteration 9: 100%|██████████| 30/30 [00:00<00:00, 362.52it/s, episode=300, return=-13.400]

Process finished with exit code 0

DQN算法

Code

import random
import gym
import numpy as np
import collections
from tqdm import tqdm
import torch
import torch.nn.functional as F
import matplotlib.pyplot as plt
import rl_utils


class ReplayBuffer:
    """经验回放池"""
    def __init__(self, capacity):
        self.buffer = collections.deque(maxlen=capacity)  # 队列，先进先出

    def add(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))  # 将数据加入buffer

    def sample(self, batch_size):  # 从buffer中采样数据，数量为batch_size
        transitions = random.sample(self.buffer, batch_size)
        state, action, reward, next_state, done = zip(*transitions)
        return np.array(state), action, reward, np.array(next_state), done

    def size(self):  # 目前buffer中数据的数量
        return len(self.buffer)


class Qnet(torch.nn.Module):
    """一层隐藏层的Q网络"""
    def __init__(self, state_dim, hidden_dim, action_dim):
        super(Qnet, self).__init__()
        self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
        self.fc2 = torch.nn.Linear(hidden_dim, action_dim)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        return self.fc2(x)


class DQN:
    """DQN算法"""
    def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma, epsilon, target_update, device):
        self.action_dim = action_dim
        self.q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device)  # Q网络
        self.target_q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device)  # 目标网络
        self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=learning_rate)  # 使用Adam优化器
        self.gamma = gamma  # 折扣因子
        self.epsilon = epsilon  # epsilon-greedy
        self.target_update = target_update  # 目标网络更新频率
        self.count = 0  # 计数器，记录更新次数

    def take_action(self, state):  # epsilon greedy策略采取动作
        if np.random.random() < self.epsilon:
            action = np.random.randint(self.action_dim)
        else:
            state = torch.tensor([state], dtype=torch.float)
            action = self.q_net(state).argmax().item()
        return action

    def update(self, transition_dict):
        states = torch.tensor(transition_dict['states'], dtype=torch.float)
        actions = torch.tensor(transition_dict['actions']).view(-1, 1)
        rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1)
        next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float)
        dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1)

        q_values = self.q_net(states).gather(1, actions)  # Q值
        max_next_q_values = self.target_q_net(next_states).max(1)[0].view(-1, 1)  # 下个状态的最大Q值
        q_targets = rewards + self.gamma * max_next_q_values * (1 - dones)  # TD目标
        dqn_loss = torch.mean(F.mse_loss(q_values, q_targets))  # 均方误差损失函数
        self.optimizer.zero_grad()
        dqn_loss.backward()  # 反向传播更新参数
        self.optimizer.step()

        if self.count % self.target_update == 0:
            self.target_q_net.load_state_dict(self.q_net.state_dict())  # 更新目标网络

        self.count += 1


lr = 2e-3
num_episodes = 500
hidden_dim = 128
gamma = 0.98
epsilon = 0.01
target_update = 10
buffer_size = 10000
minimal_size = 500
batch_size = 64
device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')

env_name = 'CartPole-v0'
env = gym.make(env_name)
random.seed(0)
np.random.seed(0)
env.seed(0)
torch.manual_seed(0)
replay_buffer = ReplayBuffer(buffer_size)
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
agent = DQN(state_dim, hidden_dim, action_dim, lr, gamma, epsilon, target_update, device)

return_list = []
for i in range(10):
    with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:
        for i_episode in range(int(num_episodes / 10)):
            episode_return = 0
            state = env.reset()
            done = False
            while not done:
                action = agent.take_action(state)
                next_state, reward, done, _ = env.step(action)
                replay_buffer.add(state, action, reward, next_state, done)
                state = next_state
                episode_return += reward
                if replay_buffer.size() > minimal_size:  # 当buffer数据数量超过一定值后，才进行Q网络训练
                    b_s, b_a, b_r, b_ns, b_d = replay_buffer.sample(batch_size)
                    transition_dict = {'states': b_s,
                                       'actions': b_a,
                                       'next_states': b_ns,
                                       'rewards': b_r,
                                       'dones': b_d}
                    agent.update(transition_dict)
            return_list.append(episode_return)
            if (i_episode + 1) % 10 == 0:
                pbar.set_postfix({'episode': '%d' % (num_episodes / 10 * i + i_episode + 1),
                                  'return': '%.3f' % np.mean(return_list[-10:])})
            pbar.update(1)


episodes_list = list(range(len(return_list)))
plt.plot(episodes_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('DQN on {}'.format(env_name))
plt.show()

mv_return = rl_utils.moving_average(return_list, 9)
plt.plot(episodes_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('DQN on {}'.format(env_name))
plt.show()

rl_utils.py

from tqdm import tqdm
import numpy as np
import torch
import collections
import random


class ReplayBuffer:
    def __init__(self, capacity):
        self.buffer = collections.deque(maxlen=capacity)

    def add(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))

    def sample(self, batch_size):
        transitions = random.sample(self.buffer, batch_size)
        state, action, reward, next_state, done = zip(*transitions)
        return np.array(state), action, reward, np.array(next_state), done

    def size(self):
        return len(self.buffer)


def moving_average(a, window_size):
    cumulative_sum = np.cumsum(np.insert(a, 0, 0))
    middle = (cumulative_sum[window_size:] - cumulative_sum[:-window_size]) / window_size
    r = np.arange(1, window_size-1, 2)
    begin = np.cumsum(a[:window_size-1])[::2] / r
    end = (np.cumsum(a[:-window_size:-1])[::2] / r)[::-1]
    return np.concatenate((begin, middle, end))


def train_on_policy_agent(env, agent, num_episodes):
    return_list = []
    for i in range(10):
        with tqdm(total=int(num_episodes/10), desc='Iteration %d' % i) as pbar:
            for i_episode in range(int(num_episodes/10)):
                episode_return = 0
                transition_dict = {'states': [], 'actions': [], 'next_states': [], 'rewards': [], 'dones': []}
                state = env.reset()
                done = False
                while not done:
                    action = agent.take_action(state)
                    next_state, reward, done, _ = env.step(action)
                    transition_dict['states'].append(state)
                    transition_dict['actions'].append(action)
                    transition_dict['next_states'].append(next_state)
                    transition_dict['rewards'].append(reward)
                    transition_dict['dones'].append(done)
                    state = next_state
                    episode_return += reward
                return_list.append(episode_return)
                agent.update(transition_dict)
                if (i_episode+1) % 10 == 0:
                    pbar.set_postfix({'episode': '%d' % (num_episodes/10 * i + i_episode+1), 'return': '%.3f' % np.mean(return_list[-10:])})
                pbar.update(1)
    return return_list


def train_off_policy_agent(env, agent, num_episodes, replay_buffer, minimal_size, batch_size):
    return_list = []
    for i in range(10):
        with tqdm(total=int(num_episodes/10), desc='Iteration %d' % i) as pbar:
            for i_episode in range(int(num_episodes/10)):
                episode_return = 0
                state = env.reset()
                done = False
                while not done:
                    action = agent.take_action(state)
                    next_state, reward, done, _ = env.step(action)
                    replay_buffer.add(state, action, reward, next_state, done)
                    state = next_state
                    episode_return += reward
                    if replay_buffer.size() > minimal_size:
                        b_s, b_a, b_r, b_ns, b_d = replay_buffer.sample(batch_size)
                        transition_dict = {'states': b_s, 'actions': b_a, 'next_states': b_ns, 'rewards': b_r, 'dones': b_d}
                        agent.update(transition_dict)
                return_list.append(episode_return)
                if (i_episode+1) % 10 == 0:
                    pbar.set_postfix({'episode': '%d' % (num_episodes/10 * i + i_episode+1), 'return': '%.3f' % np.mean(return_list[-10:])})
                pbar.update(1)
    return return_list


def compute_advantage(gamma, lmbda, td_delta):
    td_delta = td_delta.detach().numpy()
    advantage_list = []
    advantage = 0.0
    for delta in td_delta[::-1]:
        advantage = gamma * lmbda * advantage + delta
        advantage_list.append(advantage)
    advantage_list.reverse()
    return torch.tensor(advantage_list, dtype=torch.float)

Results

D:\Anaconda3\envs\torch\python.exe D:/PythonProject/reinforcement_learning/06_Deep_Q_Network.py
Iteration 0: 100%|██████████| 50/50 [00:00<00:00, 737.24it/s, episode=50, return=9.300]
Iteration 1: 100%|██████████| 50/50 [00:00<00:00, 71.66it/s, episode=100, return=12.300]
Iteration 2: 100%|██████████| 50/50 [00:03<00:00, 14.42it/s, episode=150, return=123.000]
Iteration 3: 100%|██████████| 50/50 [00:12<00:00,  4.16it/s, episode=200, return=148.200]
Iteration 4: 100%|██████████| 50/50 [00:12<00:00,  3.88it/s, episode=250, return=188.000]
Iteration 5: 100%|██████████| 50/50 [00:13<00:00,  3.75it/s, episode=300, return=189.200]
Iteration 6: 100%|██████████| 50/50 [00:13<00:00,  3.73it/s, episode=350, return=200.000]
Iteration 7: 100%|██████████| 50/50 [00:13<00:00,  3.73it/s, episode=400, return=200.000]
Iteration 8: 100%|██████████| 50/50 [00:13<00:00,  3.58it/s, episode=450, return=194.400]
Iteration 9: 100%|██████████| 50/50 [00:11<00:00,  4.52it/s, episode=500, return=152.400]

Process finished with exit code 0

DQN改进算法

Code

import random
import gym
import numpy as np
import torch
import torch.nn.functional as F
import matplotlib.pyplot as plt
import rl_utils
from tqdm import tqdm


class Qnet(torch.nn.Module):
    """一层隐藏层的Q网络"""

    def __init__(self, state_dim, hidden_dim, action_dim):
        super(Qnet, self).__init__()
        self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
        self.fc2 = torch.nn.Linear(hidden_dim, action_dim)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        return self.fc2(x)


class DQN:
    """DQN算法，包括DoubleDQN"""

    def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma, epsilon, target_update, device,
                 dqn_type='VanillaDQN'):
        self.action_dim = action_dim
        self.q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device)
        self.target_q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device)
        self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=learning_rate)
        self.gamma = gamma
        self.epsilon = epsilon
        self.target_update = target_update
        self.count = 0
        self.dqn_type = dqn_type

    def take_action(self, state):
        if np.random.random() < self.epsilon:
            action = np.random.randint(self.action_dim)
        else:
            state = torch.tensor([state], dtype=torch.float)
            action = self.q_net(state).argmax().item()
        return action

    def max_q_value(self, state):
        state = torch.tensor([state], dtype=torch.float)
        return self.q_net(state).max().item()

    def update(self, transition_dict):
        states = torch.tensor(transition_dict['states'], dtype=torch.float)
        actions = torch.tensor(transition_dict['actions']).view(-1, 1)
        rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1)
        next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float)
        dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1)

        q_values = self.q_net(states).gather(1, actions)
        if self.dqn_type == 'DoubleDQN':  # Double DQN的区别
            max_action = self.q_net(next_states).max(1)[1].view(-1, 1)
            max_next_q_values = self.target_q_net(next_states).gather(1, max_action)
        else:
            max_next_q_values = self.target_q_net(next_states).max(1)[0].view(-1, 1)
        q_targets = rewards + self.gamma * max_next_q_values * (1 - dones)
        dqn_loss = torch.mean(F.mse_loss(q_values, q_targets))
        self.optimizer.zero_grad()
        dqn_loss.backward()
        self.optimizer.step()

        if self.count % self.target_update == 0:
            self.target_q_net.load_state_dict(self.q_net.state_dict())
        self.count += 1


lr = 1e-2
num_episodes = 200
hidden_dim = 128
gamma = 0.98
epsilon = 0.01
target_update = 50
buffer_size = 5000
minimal_size = 1000
batch_size = 64
device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')

env_name = 'Pendulum-v0'
env = gym.make(env_name)
state_dim = env.observation_space.shape[0]
action_dim = 11  # 将连续动作分成11个离散动作


def dis_to_con(discrete_action, env, action_dim):  # 离散动作转回连续的函数
    action_lowbound = env.action_space.low[0]  # 连续动作的最小值
    action_upbound = env.action_space.high[0]  # 连续动作的最大值
    return action_lowbound + (discrete_action / (action_dim - 1)) * (action_upbound - action_lowbound)


def train_DQN(agent, env, num_episodes, replay_buffer, minimal_size, batch_size):
    return_list = []
    max_q_value_list = []
    max_q_value = 0
    for i in range(10):
        with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:
            for i_episode in range(int(num_episodes / 10)):
                episode_return = 0
                state = env.reset()
                done = False
                while not done:
                    action = agent.take_action(state)
                    max_q_value = agent.max_q_value(state) * 0.005 + max_q_value * 0.995  # 平滑处理
                    max_q_value_list.append(max_q_value)
                    action_continuous = dis_to_con(action, env, agent.action_dim)
                    next_state, reward, done, _ = env.step([action_continuous])
                    replay_buffer.add(state, action, reward, next_state, done)
                    state = next_state
                    episode_return += reward
                    if replay_buffer.size() > minimal_size:
                        b_s, b_a, b_r, b_ns, b_d = replay_buffer.sample(batch_size)
                        transition_dict = {'states': b_s,
                                           'actions': b_a,
                                           'next_states': b_ns,
                                           'rewards': b_r,
                                           'dones': b_d}
                        agent.update(transition_dict)
                return_list.append(episode_return)
                if (i_episode + 1) % 10 == 0:
                    pbar.set_postfix({'episode': '%d' % (num_episodes / 10 * i + i_episode + 1),
                                      'return': '%.3f' % np.mean(return_list[-10:])})
                pbar.update(1)
    return return_list, max_q_value_list


random.seed(0)
np.random.seed(0)
env.seed(0)
torch.manual_seed(0)

# DQN在Pendulum环境上的测试
replay_buffer = rl_utils.ReplayBuffer(buffer_size)
agent = DQN(state_dim, hidden_dim, action_dim, lr, gamma, epsilon, target_update, device)
return_list, max_q_value_list = train_DQN(agent, env, num_episodes, replay_buffer, minimal_size, batch_size)

episodes_list = list(range(len(return_list)))
mv_return = rl_utils.moving_average(return_list, 5)
plt.plot(episodes_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('DQN on {}'.format(env_name))
plt.show()

frames_list = list(range(len(max_q_value_list)))
plt.plot(frames_list, max_q_value_list)
plt.axhline(0, c='orange', ls='--')
plt.axhline(10, c='red', ls='--')
plt.xlabel('Frames')
plt.ylabel('Q value')
plt.title('DQN on {}'.format(env_name))
plt.show()

# Double DQN在Pendulum环境上的测试
replay_buffer = rl_utils.ReplayBuffer(buffer_size)
agent = DQN(state_dim, hidden_dim, action_dim, lr, gamma, epsilon, target_update, device, 'DoubleDQN')
return_list, max_q_value_list = train_DQN(agent, env, num_episodes, replay_buffer, minimal_size, batch_size)

episodes_list = list(range(len(return_list)))
mv_return = rl_utils.moving_average(return_list, 5)
plt.plot(episodes_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('Double DQN on {}'.format(env_name))
plt.show()

frames_list = list(range(len(max_q_value_list)))
plt.plot(frames_list, max_q_value_list)
plt.axhline(0, c='orange', ls='--')
plt.axhline(10, c='red', ls='--')
plt.xlabel('Frames')
plt.ylabel('Q value')
plt.title('Double DQN on {}'.format(env_name))
plt.show()


class VAnet(torch.nn.Module):
    """一层隐藏层的A网络和V网络"""
    def __init__(self, state_dim, hidden_dim, action_dim):
        super(VAnet, self).__init__()
        self.fc1 = torch.nn.Linear(state_dim, hidden_dim)  # 共享网络部分
        self.fc_A = torch.nn.Linear(hidden_dim, action_dim)
        self.fc_V = torch.nn.Linear(hidden_dim, 1)

    def forward(self, x):
        A = self.fc_A(F.relu(self.fc1(x)))
        V = self.fc_V(F.relu(self.fc1(x)))
        Q = V + A - A.mean(1).view(-1, 1)  # Q值由V值和A值计算得到
        return Q


class DQN_V2:
    """DQN算法，包括Double DQN和Dueling DQN"""
    def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma, epsilon, target_update, device, dqn_type='VanillanDQN'):
        self.action_dim = action_dim
        if dqn_type == 'DuelingDQN':  # Dueling DQN采取不一样的网络框架
            self.q_net = VAnet(state_dim, hidden_dim, self.action_dim).to(device)
            self.target_q_net = VAnet(state_dim, hidden_dim, self.action_dim).to(device)
        else:
            self.q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device)
            self.target_q_net = Qnet(state_dim, hidden_dim, self.action_dim).to(device)

        self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=learning_rate)
        self.gamma = gamma
        self.epsilon = epsilon
        self.target_update = target_update
        self.count = 0
        self.dqn_type = dqn_type

    def take_action(self, state):
        if np.random.random() < self.epsilon:
            action = np.random.randint(self.action_dim)
        else:
            state = torch.tensor([state], dtype=torch.float)
            action = self.q_net(state).argmax().item()
        return action

    def max_q_value(self, state):
        state = torch.tensor([state], dtype=torch.float)
        return self.q_net(state).max().item()

    def update(self, transition_dict):
        states = torch.tensor(transition_dict['states'], dtype=torch.float)
        actions = torch.tensor(transition_dict['actions']).view(-1, 1)
        rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1)
        next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float)
        dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1)

        q_values = self.q_net(states).gather(1, actions)
        if self.dqn_type == 'DoubleDQN':
            max_action = self.q_net(next_states).max(1)[1].view(-1, 1)
            max_next_q_values = self.target_q_net(next_states).gather(1, max_action)
        else:
            max_next_q_values = self.target_q_net(next_states).max(1)[0].view(-1, 1)
        q_targets = rewards + self.gamma * max_next_q_values * (1 - dones)
        dqn_loss = torch.mean(F.mse_loss(q_values, q_targets))
        self.optimizer.zero_grad()
        dqn_loss.backward()
        self.optimizer.step()

        if self.count % self.target_update == 0:
            self.target_q_net.load_state_dict(self.q_net.state_dict())
        self.count += 1


replay_buffer = rl_utils.ReplayBuffer(buffer_size)
agent = DQN(state_dim, hidden_dim, action_dim, lr, gamma, epsilon, target_update, device, 'DuelingDQN')
return_list, max_q_value_list = train_DQN(agent, env, num_episodes, replay_buffer, minimal_size, batch_size)

episodes_list = list(range(len(return_list)))
mv_return = rl_utils.moving_average(return_list, 5)
plt.plot(episodes_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('Dueling DQN on {}'.format(env_name))
plt.show()

frames_list = list(range(len(max_q_value_list)))
plt.plot(frames_list, max_q_value_list)
plt.axhline(0, c='orange', ls='--')
plt.axhline(10, c='red', ls='--')
plt.xlabel('Frames')
plt.ylabel('Q value')
plt.title('Dueling DQN on {}'.format(env_name))
plt.show()

Results

C:\Users\98383\Anaconda3\envs\torch\python.exe D:/reinforcement_learning/07_Double_DQN.py
Iteration 0:   0%|          | 0/20 [00:00<?, ?it/s]D:/reinforcement_learning/07_Double_DQN.py:43: UserWarning: Creating a tensor from a list of numpy.ndarrays is extremely slow. Please consider converting the list to a single numpy.ndarray with numpy.array() before converting to a tensor. (Triggered internally at  ..\torch\csrc\utils\tensor_new.cpp:201.)
  state = torch.tensor([state], dtype=torch.float)
Iteration 0: 100%|██████████| 20/20 [00:04<00:00,  4.06it/s, episode=20, return=-1082.336]
Iteration 1: 100%|██████████| 20/20 [00:06<00:00,  2.97it/s, episode=40, return=-403.747]
Iteration 2: 100%|██████████| 20/20 [00:06<00:00,  3.28it/s, episode=60, return=-208.177]
Iteration 3: 100%|██████████| 20/20 [00:06<00:00,  3.10it/s, episode=80, return=-247.167]
Iteration 4: 100%|██████████| 20/20 [00:06<00:00,  2.95it/s, episode=100, return=-204.798]
Iteration 5: 100%|██████████| 20/20 [00:06<00:00,  3.10it/s, episode=120, return=-199.216]
Iteration 6: 100%|██████████| 20/20 [00:06<00:00,  3.13it/s, episode=140, return=-187.849]
Iteration 7: 100%|██████████| 20/20 [00:06<00:00,  3.12it/s, episode=160, return=-290.403]
Iteration 8: 100%|██████████| 20/20 [00:06<00:00,  3.12it/s, episode=180, return=-265.666]
Iteration 9: 100%|██████████| 20/20 [00:06<00:00,  3.08it/s, episode=200, return=-211.823]
Iteration 0: 100%|██████████| 20/20 [00:05<00:00,  3.68it/s, episode=20, return=-896.746]
Iteration 1: 100%|██████████| 20/20 [00:06<00:00,  3.04it/s, episode=40, return=-227.288]
Iteration 2: 100%|██████████| 20/20 [00:06<00:00,  3.11it/s, episode=60, return=-251.530]
Iteration 3: 100%|██████████| 20/20 [00:06<00:00,  3.16it/s, episode=80, return=-247.396]
Iteration 4: 100%|██████████| 20/20 [00:06<00:00,  3.06it/s, episode=100, return=-273.071]
Iteration 5: 100%|██████████| 20/20 [00:06<00:00,  3.13it/s, episode=120, return=-296.870]
Iteration 6: 100%|██████████| 20/20 [00:06<00:00,  3.17it/s, episode=140, return=-250.661]
Iteration 7: 100%|██████████| 20/20 [00:06<00:00,  3.06it/s, episode=160, return=-395.462]
Iteration 8: 100%|██████████| 20/20 [00:06<00:00,  3.10it/s, episode=180, return=-239.515]
Iteration 9: 100%|██████████| 20/20 [00:06<00:00,  3.16it/s, episode=200, return=-188.763]
Iteration 0: 100%|██████████| 20/20 [00:04<00:00,  4.15it/s, episode=20, return=-599.272]
Iteration 1: 100%|██████████| 20/20 [00:06<00:00,  3.23it/s, episode=40, return=-172.765]
Iteration 2: 100%|██████████| 20/20 [00:06<00:00,  3.25it/s, episode=60, return=-260.010]
Iteration 3: 100%|██████████| 20/20 [00:06<00:00,  3.24it/s, episode=80, return=-376.028]
Iteration 4: 100%|██████████| 20/20 [00:05<00:00,  3.34it/s, episode=100, return=-143.622]
Iteration 5: 100%|██████████| 20/20 [00:06<00:00,  3.17it/s, episode=120, return=-152.848]
Iteration 6: 100%|██████████| 20/20 [00:06<00:00,  3.26it/s, episode=140, return=-207.692]
Iteration 7: 100%|██████████| 20/20 [00:06<00:00,  3.10it/s, episode=160, return=-220.573]
Iteration 8: 100%|██████████| 20/20 [00:06<00:00,  3.06it/s, episode=180, return=-173.121]
Iteration 9: 100%|██████████| 20/20 [00:06<00:00,  3.04it/s, episode=200, return=-182.890]

Process finished with exit code 0

策略梯度算法

Code

import gym
import torch
import torch.nn.functional as F
import numpy as np
import matplotlib.pyplot as plt
from tqdm import tqdm
import rl_utils


class PolicyNet(torch.nn.Module):
    def __init__(self, state_dim, hidden_dim, action_dim):
        super(PolicyNet, self).__init__()
        self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
        self.fc2 = torch.nn.Linear(hidden_dim, action_dim)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        return F.softmax(self.fc2(x), dim=1)


class Reinforce:
    def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma, device):
        self.policy_net = PolicyNet(state_dim, hidden_dim, action_dim).to(device)
        self.optimizer = torch.optim.Adam(self.policy_net.parameters(), lr=learning_rate)
        self.gamma = gamma  # 折扣因子

    def take_action(self, state):  # 根据动作概率分布随机采样
        state = torch.tensor([state], dtype=torch.float)
        probs = self.policy_net(state)
        action_dist = torch.distributions.Categorical(probs)
        action = action_dist.sample()
        return action.item()

    def update(self, transition_dict):
        reward_list = transition_dict['rewards']
        state_list = transition_dict['states']
        action_list = transition_dict['actions']

        G = 0
        self.optimizer.zero_grad()
        for i in reversed(range(len(reward_list))):  # 从最后一步算起
            reward = reward_list[i]
            state = torch.tensor([state_list[i]], dtype=torch.float)
            action = torch.tensor([action_list[i]]).view(-1, 1)
            log_prob = torch.log(self.policy_net(state)).gather(1, action)
            G = self.gamma * G + reward
            loss = - log_prob * G  # 每一步的损失函数
            loss.backward()  # 反向传播计算梯度
        self.optimizer.step()  # 梯度下降


learning_rate = 1e-3
num_episodes = 1000
hidden_dim = 128
gamma = 0.98
device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')

env_name = 'CartPole-v0'
env = gym.make(env_name)
env.seed(0)
torch.manual_seed(0)
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
agent = Reinforce(state_dim, hidden_dim, action_dim, learning_rate, gamma, device)

return_list = []
for i in range(10):
    with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:
        for i_episode in range(int(num_episodes / 10)):
            episode_return = 0
            transition_dict = {'states': [],
                               'actions': [],
                               'next_states': [],
                               'rewards': [],
                               'dones': []}
            state = env.reset()
            done = False
            while not done:
                action = agent.take_action(state)
                next_state, reward, done, _ = env.step(action)
                transition_dict['states'].append(state)
                transition_dict['actions'].append(action)
                transition_dict['next_states'].append(next_state)
                transition_dict['rewards'].append(reward)
                transition_dict['dones'].append(done)
                state = next_state
                episode_return += reward
            return_list.append(episode_return)
            agent.update(transition_dict)
            if (i_episode + 1) % 10 == 0:
                pbar.set_postfix({'episode': '%d' % (num_episodes / 10 * i + i_episode + 1),
                                  'return': '%.3f' % np.mean(return_list[-10:])})
            pbar.update(1)


episodes_list = list(range(len(return_list)))
plt.plot(episodes_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('Reinforce on {}'.format(env_name))
plt.show()

mv_return = rl_utils.moving_average(return_list, 9)
plt.plot(episodes_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('Reinforce on {}'.format(env_name))
plt.show()

Results

C:\Users\98383\Anaconda3\envs\torch\python.exe D:/reinforcement_learning/08_Reinforce.py
Iteration 0:   0%|          | 0/100 [00:00<?, ?it/s]D:/reinforcement_learning/08_Reinforce.py:28: UserWarning: Creating a tensor from a list of numpy.ndarrays is extremely slow. Please consider converting the list to a single numpy.ndarray with numpy.array() before converting to a tensor. (Triggered internally at  ..\torch\csrc\utils\tensor_new.cpp:201.)
  state = torch.tensor([state], dtype=torch.float)
Iteration 0: 100%|██████████| 100/100 [00:01<00:00, 57.85it/s, episode=100, return=55.500]
Iteration 1: 100%|██████████| 100/100 [00:03<00:00, 27.35it/s, episode=200, return=75.300]
Iteration 2: 100%|██████████| 100/100 [00:07<00:00, 13.95it/s, episode=300, return=178.800]
Iteration 3: 100%|██████████| 100/100 [00:09<00:00, 10.74it/s, episode=400, return=164.600]
Iteration 4: 100%|██████████| 100/100 [00:09<00:00, 10.62it/s, episode=500, return=156.500]
Iteration 5: 100%|██████████| 100/100 [00:09<00:00, 10.28it/s, episode=600, return=187.400]
Iteration 6: 100%|██████████| 100/100 [00:09<00:00, 10.25it/s, episode=700, return=194.500]
Iteration 7: 100%|██████████| 100/100 [00:10<00:00,  9.72it/s, episode=800, return=200.000]
Iteration 8: 100%|██████████| 100/100 [00:10<00:00,  9.56it/s, episode=900, return=200.000]
Iteration 9: 100%|██████████| 100/100 [00:10<00:00,  9.85it/s, episode=1000, return=186.100]

Process finished with exit code 0

Actor-Critic算法

Code

import gym
import torch
import torch.nn.functional as F
import numpy as np
import matplotlib.pyplot as plt
import rl_utils
from tqdm import tqdm


class PolicyNet(torch.nn.Module):
    def __init__(self, state_dim, hidden_dim, action_dim):
        super(PolicyNet, self).__init__()
        self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
        self.fc2 = torch.nn.Linear(hidden_dim, action_dim)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        return F.softmax(self.fc2(x), dim=1)


class ValueNet(torch.nn.Module):
    def __init__(self, state_dim, hidden_dim):
        super(ValueNet, self).__init__()
        self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
        self.fc2 = torch.nn.Linear(hidden_dim, 1)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        return self.fc2(x)


class ActorCritic:
    def __init__(self, state_dim, hidden_dim, action_dim, actor_lr, critic_lr, gamma, device):
        self.actor = PolicyNet(state_dim, hidden_dim, action_dim).to(device)
        self.critic = ValueNet(state_dim, hidden_dim).to(device)  # 价值网络
        self.actor_optimizer = torch.optim.Adam(self.actor.parameters(), lr=actor_lr)
        self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=critic_lr)  # 价值网络优化器
        self.gamma = gamma

    def take_action(self, state):
        state = torch.tensor([state], dtype=torch.float)
        probs = self.actor(state)
        action_dist = torch.distributions.Categorical(probs)
        action = action_dist.sample()
        return action.item()

    def update(self, transition_dict):
        states = torch.tensor(transition_dict['states'], dtype=torch.float)
        actions = torch.tensor(transition_dict['actions']).view(-1, 1)
        rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1)
        next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float)
        dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1)

        td_target = rewards + self.gamma * self.critic(next_states) * (1 - dones)  # 时序差分
        td_delta = td_target - self.critic(states)  # 时序差分误差
        log_probs = torch.log(self.actor(states).gather(1, actions))
        actor_loss = torch.mean(-log_probs * td_delta.detach())
        critic_loss = torch.mean(F.mse_loss(self.critic(states), td_target.detach()))  # 均方误差
        self.actor_optimizer.zero_grad()
        self.critic_optimizer.zero_grad()
        actor_loss.backward()  # 计算策略网络的梯度
        critic_loss.backward()  # 计算价值网络的梯度
        self.actor_optimizer.step()  # 更新策略网络参数
        self.critic_optimizer.step()  # 更新价值网络参数


actor_lr = 1e-3
critic_lr = 1e-2
num_episodes = 1000
hidden_dim = 128
gamma = 0.98
device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')

env_name = 'CartPole-v0'
env = gym.make(env_name)
env.seed(0)
torch.manual_seed(0)
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
agent = ActorCritic(state_dim, hidden_dim, action_dim, actor_lr, critic_lr, gamma, device)

return_list = rl_utils.train_on_policy_agent(env, agent, num_episodes)

episodes_list = list(range(len(return_list)))
plt.plot(episodes_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('Actor-Critic on {}'.format(env_name))
plt.show()

mv_return = rl_utils.moving_average(return_list, 9)
plt.plot(episodes_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('Actor-Critic on {}'.format(env_name))
plt.show()

Results

C:\Users\98383\Anaconda3\envs\torch\python.exe D:/reinforcement_learning/09_Actor_Critic.py
Iteration 0:   0%|          | 0/100 [00:00<?, ?it/s]D:/reinforcement_learning/09_Actor_Critic.py:41: UserWarning: Creating a tensor from a list of numpy.ndarrays is extremely slow. Please consider converting the list to a single numpy.ndarray with numpy.array() before converting to a tensor. (Triggered internally at  ..\torch\csrc\utils\tensor_new.cpp:201.)
  state = torch.tensor([state], dtype=torch.float)
Iteration 0: 100%|██████████| 100/100 [00:00<00:00, 168.71it/s, episode=100, return=21.100]
Iteration 1: 100%|██████████| 100/100 [00:01<00:00, 71.20it/s, episode=200, return=72.800]
Iteration 2: 100%|██████████| 100/100 [00:02<00:00, 42.75it/s, episode=300, return=109.300]
Iteration 3: 100%|██████████| 100/100 [00:04<00:00, 24.58it/s, episode=400, return=163.000]
Iteration 4: 100%|██████████| 100/100 [00:05<00:00, 17.37it/s, episode=500, return=193.600]
Iteration 5: 100%|██████████| 100/100 [00:06<00:00, 16.25it/s, episode=600, return=195.900]
Iteration 6: 100%|██████████| 100/100 [00:06<00:00, 16.35it/s, episode=700, return=199.100]
Iteration 7: 100%|██████████| 100/100 [00:06<00:00, 16.13it/s, episode=800, return=186.900]
Iteration 8: 100%|██████████| 100/100 [00:06<00:00, 16.66it/s, episode=900, return=200.000]
Iteration 9: 100%|██████████| 100/100 [00:06<00:00, 16.50it/s, episode=1000, return=200.000]

Process finished with exit code 0

TRPO算法

Code

import torch
import numpy as np
import gym
import matplotlib.pyplot as plt
import torch.nn.functional as F
import rl_utils
import copy


class PolicyNet(torch.nn.Module):
    def __init__(self, state_dim, hidden_dim, action_dim):
        super(PolicyNet, self).__init__()
        self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
        self.fc2 = torch.nn.Linear(hidden_dim, action_dim)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        return F.softmax(self.fc2(x), dim=1)


class ValueNet(torch.nn.Module):
    def __init__(self, state_dim, hidden_dim):
        super(ValueNet, self).__init__()
        self.fc1 = torch.nn.Linear(state_dim, hidden_dim)
        self.fc2 = torch.nn.Linear(hidden_dim, 1)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        return self.fc2(x)


class TRPO:
    """TRPO算法"""
    def __init__(self, hidden_dim, state_space, action_space, lmbda, kl_constraint,
                 alpha, critic_lr, gamma, device):
        state_dim = state_space.shape[0]
        action_dim = action_space.n
        # 策略网络参数不需要优化器更新
        self.actor = PolicyNet(state_dim, hidden_dim, action_dim).to(device)
        self.critic = ValueNet(state_dim, hidden_dim).to(device)
        self.critic_optimizer = torch.optim.Adam(self.critic.parameters(), lr=critic_lr)
        self.gamma = gamma
        self.lmbda = lmbda  # GAE参数
        self.kl_constraint = kl_constraint  # KL距离最大限制
        self.alpha = alpha  # 线性搜索参数
        self.device = device

    def take_action(self, state):
        state = torch.tensor([state], dtype=torch.float).to(self.device)
        probs = self.actor(state)
        action_dist = torch.distributions.Categorical(probs)
        action = action_dist.sample()
        return action.item()

    def hessian_matrix_vector_product(self, states, old_action_dists, vector):
        # 计算黑塞矩阵和一个向量的乘积
        new_action_dists = torch.distributions.Categorical(self.actor(states))
        kl = torch.mean(torch.distributions.kl.kl_divergence(old_action_dists, new_action_dists))  # 计算平均KL距离
        kl_grad = torch.autograd.grad(kl, self.actor.parameters(), create_graph=True)
        kl_grad_vector = torch.cat([grad.view(-1) for grad in kl_grad])
        # KL距离的梯度先和向量进行点积运算
        kl_grad_vector_product = torch.dot(kl_grad_vector, vector)
        grad2 = torch.autograd.grad(kl_grad_vector_product, self.actor.parameters())
        grad2_vector = torch.cat([grad.view(-1) for grad in grad2])
        return grad2_vector

    def conjugate_gradient(self, grad, states, old_action_dists):  # 共轭梯度法求解方程
        x = torch.zeros_like(grad)
        r = grad.clone()
        p = grad.clone()
        rdotr = torch.dot(r, r)
        for i in range(10):  # 共轭梯度主循环
            Hp = self.hessian_matrix_vector_product(states, old_action_dists, p)
            alpha = rdotr / torch.dot(p, Hp)
            x += alpha * p
            r -= alpha * Hp
            new_rdotr = torch.dot(r, r)
            if new_rdotr < 1e-10:
                break
            beta = new_rdotr / rdotr
            p = r + beta * p
            rdotr = new_rdotr
        return x

    def compute_surrogate_obj(self, states, actions, advantage, old_log_probs, actor):  # 计算策略目标
        log_probs = torch.log(actor(states).gather(1, actions))
        ratio = torch.exp(log_probs - old_log_probs)
        return torch.mean(ratio * advantage)

    def line_search(self, states, actions, advantage, old_log_probs, old_action_dists, max_vec):
        old_para = torch.nn.utils.convert_parameters.parameters_to_vector(self.actor.parameters())
        old_obj = self.compute_surrogate_obj(states, actions, advantage, old_log_probs, self.actor)
        for i in range(15):  # 线性搜索主循环
            coef = self.alpha ** i
            new_para = old_para + coef * max_vec
            new_actor = copy.deepcopy(self.actor)
            torch.nn.utils.convert_parameters.vector_to_parameters(new_para, new_actor.parameters())
            new_actions_dists = torch.distributions.Categorical(new_actor(states))
            kl_div = torch.mean(torch.distributions.kl.kl_divergence(old_action_dists, new_actions_dists))
            new_obj = self.compute_surrogate_obj(states, actions, advantage, old_log_probs, new_actor)
            if new_obj > old_obj and kl_div < self.kl_constraint:
                return new_para
        return old_para

    def policy_learn(self, states, actions, old_action_dists, old_log_probs, advantage):
        surrogate_obj = self.compute_surrogate_obj(states, actions, advantage, old_log_probs, self.actor)
        grads = torch.autograd.grad(surrogate_obj, self.actor.parameters())
        obj_grad = torch.cat([grad.view(-1) for grad in grads]).detach()
        # 用共轭梯度法计算x = H^(-1)g
        descent_direction = self.conjugate_gradient(obj_grad, states, old_action_dists)
        Hd = self.hessian_matrix_vector_product(states, old_action_dists, descent_direction)
        max_coef = torch.sqrt(2 * self.kl_constraint / (torch.dot(descent_direction, Hd) + 1e-8))
        new_para = self.line_search(states, actions, advantage, old_log_probs, old_action_dists, descent_direction * max_coef)
        torch.nn.utils.convert_parameters.vector_to_parameters(new_para, self.actor.parameters())  # 用线性搜索后的参数更新策略

    def update(self, transition_dict):
        states = torch.tensor(transition_dict['states'], dtype=torch.float).to(self.device)
        actions = torch.tensor(transition_dict['actions']).view(-1, 1).to(self.device)
        rewards = torch.tensor(transition_dict['rewards'], dtype=torch.float).view(-1, 1).to(self.device)
        next_states = torch.tensor(transition_dict['next_states'], dtype=torch.float).to(self.device)
        dones = torch.tensor(transition_dict['dones'], dtype=torch.float).view(-1, 1).to(self.device)

        td_target = rewards + self.gamma * self.critic(next_states) * (1 - dones)
        td_delta = td_target - self.critic(states)
        advantage = rl_utils.compute_advantage(self.gamma, self.lmbda, td_delta.cpu()).to(self.device)

        old_log_probs = torch.log(self.actor(states).gather(1, actions)).detach()
        old_action_dists = torch.distributions.Categorical(self.actor(states).detach())
        critic_loss = torch.mean(F.mse_loss(self.critic(states), td_target.detach()))
        self.critic_optimizer.zero_grad()
        critic_loss.backward()
        self.critic_optimizer.step()  # 更新价值函数
        # 更新策略函数
        self.policy_learn(states, actions, old_action_dists, old_log_probs, advantage)


num_episodes = 500
hidden_dim = 128
gamma = 0.98
lmbda = 0.95
critic_lr = 1e-2
kl_constraint = 0.00005
alpha = 0.5
device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')

env_name = 'CartPole-v0'
env = gym.make(env_name)
env.seed(0)
torch.manual_seed(0)
agent = TRPO(hidden_dim, env.observation_space, env.action_space, lmbda, kl_constraint, alpha, critic_lr, gamma, device)
return_list = rl_utils.train_on_policy_agent(env, agent, num_episodes)

episodes_list = list(range(len(return_list)))
plt.plot(episodes_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('TRPO on {}'.format(env_name))
plt.show()

mv_return = rl_utils.moving_average(return_list, 9)
plt.plot(episodes_list, mv_return)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('TRPO on {}'.format(env_name))
plt.show()

Results

C:\Users\98383\Anaconda3\envs\torch\python.exe D:/reinforcement_learning/10_True_Region_Policy_Optimization.py
Iteration 0:   0%|          | 0/50 [00:00<?, ?it/s]D:/reinforcement_learning/10_True_Region_Policy_Optimization.py:49: UserWarning: Creating a tensor from a list of numpy.ndarrays is extremely slow. Please consider converting the list to a single numpy.ndarray with numpy.array() before converting to a tensor. (Triggered internally at  ..\torch\csrc\utils\tensor_new.cpp:201.)
  state = torch.tensor([state], dtype=torch.float).to(self.device)
Iteration 0: 100%|██████████| 50/50 [00:01<00:00, 37.92it/s, episode=50, return=51.100]
Iteration 1: 100%|██████████| 50/50 [00:02<00:00, 22.40it/s, episode=100, return=71.700]
Iteration 2: 100%|██████████| 50/50 [00:03<00:00, 15.84it/s, episode=150, return=134.100]
Iteration 3: 100%|██████████| 50/50 [00:03<00:00, 14.31it/s, episode=200, return=169.400]
Iteration 4: 100%|██████████| 50/50 [00:04<00:00, 12.42it/s, episode=250, return=179.000]
Iteration 5: 100%|██████████| 50/50 [00:04<00:00, 12.48it/s, episode=300, return=175.100]
Iteration 6: 100%|██████████| 50/50 [00:04<00:00, 11.97it/s, episode=350, return=193.500]
Iteration 7: 100%|██████████| 50/50 [00:04<00:00, 11.51it/s, episode=400, return=190.000]
Iteration 8: 100%|██████████| 50/50 [00:04<00:00, 11.51it/s, episode=450, return=182.900]
Iteration 9: 100%|██████████| 50/50 [00:04<00:00, 11.80it/s, episode=500, return=187.500]

Process finished with exit code 0