强化学习（一）：基础与核心概念

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

import numpy as np

class GridWorld:
    """智能体从起点走到终点；障碍格不可通过。"""

    def __init__(self, grid_size=(5, 5), start=(0, 0),
                 goal=(4, 4), obstacles=None):
        self.height, self.width = grid_size
        self.start = start
        self.goal = goal
        self.obstacles = set(obstacles or [])
        self.actions = [0, 1, 2, 3]                        # 上下左右
        self.action_names = ['U', 'D', 'L', 'R']
        self.action_deltas = {0: (-1, 0), 1: (1, 0),
                              2: (0, -1), 3: (0, 1)}
        self.reset()

    def reset(self):
        self.state = self.start
        return self.state

    def step(self, action):
        dr, dc = self.action_deltas[action]
        nr, nc = self.state[0] + dr, self.state[1] + dc
        if (0 <= nr < self.height and 0 <= nc < self.width
                and (nr, nc) not in self.obstacles):
            self.state = (nr, nc)
        if self.state == self.goal:
            return self.state, 10.0, True
        return self.state, -1.0, False

    def get_transitions(self, state, action):
        """确定性转移，返回 [(下一状态, 概率, 奖励)]。"""
        old = self.state
        self.state = state
        ns, r, _ = self.step(action)
        self.state = old
        return [(ns, 1.0, r)]

def value_iteration(env, gamma=0.9, theta=1e-6):
    """通过反复 max 备份求解 Bellman 最优方程。"""
    V = np.zeros((env.height, env.width))
    for iteration in range(1000):
        delta = 0
        for r in range(env.height):
            for c in range(env.width):
                s = (r, c)
                if s == env.goal or s in env.obstacles:
                    continue
                old_v = V[r, c]
                V[r, c] = max(
                    sum(p * (rew + gamma * V[ns])
                        for ns, p, rew in env.get_transitions(s, a))
                    for a in env.actions
                )
                delta = max(delta, abs(old_v - V[r, c]))
        if delta < theta:
            print(f"第 {iteration + 1} 次扫描后收敛")
            break

    policy = np.zeros((env.height, env.width), dtype=int)
    for r in range(env.height):
        for c in range(env.width):
            s = (r, c)
            if s == env.goal or s in env.obstacles:
                continue
            q = [sum(p * (rew + gamma * V[ns])
                     for ns, p, rew in env.get_transitions(s, a))
                 for a in env.actions]
            policy[r, c] = int(np.argmax(q))
    return V, policy

env = GridWorld(grid_size=(5, 5), start=(0, 0), goal=(4, 4),
                obstacles=[(1, 1), (2, 2), (3, 1)])
V, policy = value_iteration(env)
print(np.round(V, 1))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

import numpy as np
from collections import defaultdict

def mc_policy_evaluation(env, policy_fn, num_episodes=10000, gamma=0.9):
    """首次访问 MC 策略评估。"""
    returns = defaultdict(list)
    V = defaultdict(float)
    for _ in range(num_episodes):
        trajectory, state, done = [], env.reset(), False
        while not done:
            action = policy_fn(state)
            next_state, reward, done = env.step(action)
            trajectory.append((state, action, reward))
            state = next_state
        G, visited = 0.0, set()
        for state, _, reward in reversed(trajectory):
            G = reward + gamma * G
            if state not in visited:
                visited.add(state)
                returns[state].append(G)
                V[state] = float(np.mean(returns[state]))
    return V

def mc_control(env, num_episodes=10000, gamma=0.9, epsilon=0.1):
    """带 epsilon-贪心探索的同策略 MC 控制。"""
    Q = defaultdict(lambda: np.zeros(len(env.actions)))
    returns = defaultdict(list)
    for _ in range(num_episodes):
        trajectory, state, done = [], env.reset(), False
        while not done:
            if np.random.random() < epsilon:
                action = int(np.random.choice(env.actions))
            else:
                action = int(np.argmax(Q[state]))
            next_state, reward, done = env.step(action)
            trajectory.append((state, action, reward))
            state = next_state
        G, visited = 0.0, set()
        for state, action, reward in reversed(trajectory):
            G = reward + gamma * G
            if (state, action) not in visited:
                visited.add((state, action))
                returns[(state, action)].append(G)
                Q[state][action] = float(np.mean(returns[(state, action)]))
    return Q

1
2

S . . . . . . G
C C C C C C C C

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

import numpy as np

def _eps_greedy(Q, state, actions, epsilon):
    if np.random.random() < epsilon:
        return int(np.random.choice(actions))
    return int(np.argmax(Q[state]))

def sarsa(env, num_episodes=1000, alpha=0.1, gamma=0.9, epsilon=0.1):
    """Sarsa：同策略 TD 控制。"""
    Q = {(r, c): np.zeros(len(env.actions))
         for r in range(env.height) for c in range(env.width)}
    for _ in range(num_episodes):
        state = env.reset()
        action = _eps_greedy(Q, state, env.actions, epsilon)
        done = False
        while not done:
            next_state, reward, done = env.step(action)
            next_action = _eps_greedy(Q, next_state, env.actions, epsilon)
            target = reward + gamma * Q[next_state][next_action]
            Q[state][action] += alpha * (target - Q[state][action])
            state, action = next_state, next_action
    return Q

def q_learning(env, num_episodes=1000, alpha=0.1, gamma=0.9, epsilon=0.1):
    """Q-Learning：离策略 TD 控制。"""
    Q = {(r, c): np.zeros(len(env.actions))
         for r in range(env.height) for c in range(env.width)}
    for _ in range(num_episodes):
        state, done = env.reset(), False
        while not done:
            action = _eps_greedy(Q, state, env.actions, epsilon)
            next_state, reward, done = env.step(action)
            target = reward + gamma * np.max(Q[next_state])
            Q[state][action] += alpha * (target - Q[state][action])
            state = next_state
    return Q