《用Python动手学习强化学习》【MDP（马尔可夫决策过程）】

糖逗

代码出处：《用Python动手学习强化学习》第一章：强化学习的问题设定：马尔可夫决策过程

1. import random
   
2. from enum import Enum
   
3. import numpy as np
   
4. 
   
5. 
   
6. class State():
   
7.     def __init__(self, row = -1, column = -1):
   
8.         self.row = row
   
9.         self.column = column
   
10.         
    
11.     def __repr__(self):
    
12.         return "<State:[{},{}]>".format(self.row, self.column)
    
13.    
    
14.     def clone(self):
    
15.         return State(self.row, self.column)
    
16. 
    
17.     def __hash__(self):
    
18.         return hash((self.row, self.column))
    
19.    
    
20.     def __eq__(self, other):
    
21.         return self.row == other.row and self.column == other.column
    
22.    
    
23.    
    
24. class Action(Enum):
    
25.     UP = 1
    
26.     DOWN = -1
    
27.     LEFT = 2
    
28.     RIGHT = -2
    
29.    
    
30. 
    
31. 
    
32. class Environment():#迁移函数和奖励函数
    
33.     def __init__(self, grid, move_prob = 0.8):
    
34.         '''
    
35.         0：普通格子
    
36.         -1:有危险的格子（游戏结束）
    
37.         1：有奖励的格子（游戏结束）
    
38.         9：被屏蔽的格子（无法放置智能体）
    
39.         '''
    
40.         self.grid = grid
    
41.         '''
    
42.         默认的奖励是负数，就像施加了初始位置
    
43.         '''
    
44.         self.default_reward = -0.04
    
45.         '''
    
46.         智能体能够以move_prob的概率向所选方向移动
    
47.         '''
    
48.         self.move_prob = move_prob
    
49.         self.reset()
    
50.         
    
51.     @property
    
52.     def row_length(self):
    
53.         return len(self.grid)
    
54.    
    
55.     @property
    
56.     def column_length(self):
    
57.         return len(self.grid[0])
    
58.    
    
59.     @property
    
60.     def actions(self):
    
61.         return [Action.UP, Action.DOWN, Action.LEFT, Action.RIGHT]
    
62.    
    
63.     @property
    
64.     def states(self):
    
65.         states = []
    
66.         for row in range(self.row_length):
    
67.             for column in range(self.column_length):
    
68.                 if self.grid[row][column] != 9:
    
69.                     states.append(State(row, column))
    
70.         return states
    
71. 
    
72.     def reset(self):
    
73.         #初始位置在左下角
    
74.         self.agent_state = State(self.row_length - 1, 0)
    
75.         return self.agent_state
    
76.    
    
77.     def can_action_at(self, state):
    
78.         if self.grid[state.row][state.column] == 0:
    
79.             return True
    
80.         else:
    
81.             return False
    
82.         
    
83.     def _move(self, state, action):
    
84.         if not self.can_action_at(state):
    
85.             raise Exception("Can't move from here!")
    
86.         
    
87.         next_state = state.clone()
    
88.         
    
89.         if action == Action.UP:
    
90.             next_state.row -= 1
    
91.         elif action == Action.DOWN:
    
92.             next_state.row += 1
    
93.         elif action == Action.LEFT:
    
94.             next_state.column -= 1
    
95.         elif action == Action.RIGHT:
    
96.             next_state.column += 1
    
97.             
    
98.         #检查状态是否在grid外
    
99.         if not (0 <= next_state.row < self.row_length):
    
100.             next_state = state
     
101.         if not (0 <= next_state.column < self.column_length):
     
102.             next_state = state
     
103.         
     
104.         #检查智能体是否到达了被屏蔽的格子
     
105.         if self.grid[next_state.row][next_state.column] == 9:
     
106.             next_state = state
     
107.             
     
108.         return next_state
     
109.         
     
110.     def transit_func(self, state, action):
     
111.         transition_probs = {}
     
112.         if not self.can_action_at(state):
     
113.             #游戏结束
     
114.             return transition_probs
     
115.         
     
116.         opposite_direction = Action(action.value * -1)
     
117.         
     
118.         for a in self.actions:
     
119.             prob = 0
     
120.             if a == action:
     
121.                 prob = self.move_prob
     
122.             elif a != opposite_direction:
     
123.                 prob = (1 - self.move_prob) / 2
     
124.             
     
125.             next_state = self._move(state, a)
     
126.             if next_state not in transition_probs:
     
127.                 #求期望
     
128.                 transition_probs[next_state] = prob
     
129.             else:
     
130.                 transition_probs[next_state] += prob
     
131.         return transition_probs
     
132.    
     
133.     def reward_func(self, state):
     
134.         reward = self.default_reward
     
135.         done = False
     
136.         #检查下一种状态的属性
     
137.         attribute = self.grid[state.row][state.column]
     
138.         if attribute == 1:
     
139.             #获得奖励，游戏结束
     
140.             reward = 1
     
141.             done = True
     
142.         elif attribute == -1:
     
143.             #遇到危险，游戏结束
     
144.             reward = -1
     
145.             done = True
     
146.         return reward, done
     
147.    
     
148.     def transit(self, state, action):
     
149.         transition_probs = self.transit_func(state, action)
     
150.         if len(transition_probs) == 0:
     
151.             return None, None, True
     
152.         
     
153.         next_states = []
     
154.         probs = []
     
155.         for s in transition_probs:
     
156.             next_states.append(s)
     
157.             probs.append(transition_probs[s])
     
158.         
     
159.         next_state = np.random.choice(next_states, p = probs)
     
160.         reward, done = self.reward_func(next_state)
     
161.         return next_state, reward, done
     
162.    
     
163.     def step(self, action):
     
164.         next_state, reward, done = self.transit(self.agent_state, action)
     
165.         if next_state is not None:
     
166.             self.agent_state = next_state
     
167.         return next_state, reward, done
     
168.    
     
169.    
     
170. class Agent():
     
171.     def __init__(self, env):
     
172.         self.actions = env.actions
     
173.    
     
174.     def policy(self, state):
     
175.         return random.choice(self.actions)
     
176.    
     
177. 
     
178. def main():
     
179.     grid = [[0, 0, 0, 1], [0, 9, 0, -1], [0, 0, 0, 0]]
     
180.     env = Environment(grid)#环境包括迁移函数和奖励函数
     
181.     agent = Agent(env)
     
182.    
     
183.     for i in range(10):
     
184.         state = env.reset()
     
185.         total_reward = 0
     
186.         done = False
     
187.         
     
188.         while not done:
     
189.             action = agent.policy(state)
     
190.             next_state, reward, done = env.step(action)
     
191.             total_reward += reward
     
192.             state = next_state
     
193.             
     
194.         print("Episode{}:Agent gets {} reward.".format(i, total_reward))
     
195. 
     
196. if __name__ == "__main__":
     
197.     main()
     
198.    

复制代码

糖逗