-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathHA7_Q1.py
342 lines (264 loc) · 10.5 KB
/
HA7_Q1.py
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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
# This code is proposed as a reference solution for various exercises of Home Assignements for the OReL course in 2023.
# This solution is tailored for simplicity of understanding and is in no way optimal, nor the only way to implement the different elements!
import numpy as np
####################################################################################################################################################
####################################################################################################################################################
####################################################################################################################################################
# ENVIRONMENTS
####################################################################################################################################################
####################################################################################################################################################
####################################################################################################################################################
# A simple 4-room gridworld implementation with a grid of 7x7 for a total of 20 states (the walls do not count!).
# We arbitrarily chose the actions '0' = 'go up', '1' = 'go right', '2' = 'go down' thus '3' = 'go left'
# Finally the state '0' is the top-left corner, 'nS - 1' is the down-right corner.
# The agent is teleported back to the the initial state '0' (top-left corner) , whenever performing any action in rewarding state '19' (down-right corner).
class four_room():
def __init__(self):
self.nS = 20
nS = self.nS
self.nA = 4
self.map = [[-1, -1, -1, -1, -1, -1, -1],
[-1, 0, 1, 2, 3, 4, -1],
[-1, 5, 6, -1, 7, 8, -1],
[-1, 9, -1, -1, 10, -1, -1],
[-1, 11, 12, 13, 14, 15, -1],
[-1, 16, 17, -1, 18, 19, -1],
[-1, -1, -1, -1, -1, -1, -1]]
map = np.array(self.map)
# We build the transitions matrix P using the map.
self.P = np.zeros((nS, 4, nS))
for s in range(nS):
temp = np.where(s == map)
x, y = temp[0][0], temp[1][0]
up = map[x-1, y]
right = map[x, y+1]
down = map[x+1, y]
left = map[x, y-1]
# Action 0: go up.
a = 0
self.P[s, a, s] += 0.1
# Up
if up == -1:
self.P[s, a, s] += 0.7
else:
self.P[s, a, up] += 0.7
# Right
if right == -1:
self.P[s, a, s] += 0.1
else:
self.P[s, a, right] += 0.1
# Left
if left == -1:
self.P[s, a, s] += 0.1
else:
self.P[s, a, left] += 0.1
# Action 1: go right.
a = 1
self.P[s, a, s] += 0.1
# Up
if up == -1:
self.P[s, a, s] += 0.1
else:
self.P[s, a, up] += 0.1
# Right
if right == -1:
self.P[s, a, s] += 0.7
else:
self.P[s, a, right] += 0.7
# Down
if down == -1:
self.P[s, a, s] += 0.1
else:
self.P[s, a, down] += 0.1
# Action 2: go down.
a = 2
self.P[s, a, s] += 0.1
# Right
if right == -1:
self.P[s, a, s] += 0.1
else:
self.P[s, a, right] += 0.1
# Down
if down == -1:
self.P[s, a, s] += 0.7
else:
self.P[s, a, down] += 0.7
# Left
if left == -1:
self.P[s, a, s] += 0.1
else:
self.P[s, a, left] += 0.1
# Action 3: go left.
a = 3
self.P[s, a, s] += 0.1
# Up
if up == -1:
self.P[s, a, s] += 0.1
else:
self.P[s, a, up] += 0.1
# Down
if down == -1:
self.P[s, a, s] += 0.1
else:
self.P[s, a, down] += 0.1
# Left
if left == -1:
self.P[s, a, s] += 0.7
else:
self.P[s, a, left] += 0.7
# Set to teleport back when in the rewarding state.
if s == self.nS - 1:
for a in range(4):
for ss in range(self.nS):
self.P[s, a, ss] = 0
if ss == 0:
self.P[s, a, ss] = 1
# We build the reward matrix R.
self.R = np.zeros((nS, 4))
for a in range(4):
self.R[nS - 1, a] = 1
# We (arbitrarily) set the initial state in the top-left corner.
self.s = 0
# To reset the environment in initial settings.
def reset(self):
self.s = 0
return self.s
# Perform a step in the environment for a given action. Return a couple state, reward (s_t, r_t).
def step(self, action):
new_s = np.random.choice(np.arange(self.nS), p=self.P[self.s, action])
reward = self.R[self.s, action]
self.s = new_s
return new_s, reward
####################################################################################################################################################
####################################################################################################################################################
####################################################################################################################################################
# VI and PI
####################################################################################################################################################
####################################################################################################################################################
####################################################################################################################################################
# An implementation of the Value Iteration algorithm for a given environment 'env' and discout 'gamma' < 1.
# An arbitrary 'max_iter' is a maximum number of iteration, usefull to catch any error in your code!
# Return the number of iterations, the final value and the optimal policy.
def VI(env, gamma = 0.9, max_iter = 10**3, epsilon = 10**(-2)):
# The variable containing the optimal policy estimate at the current iteration.
policy = np.zeros(env.nS, dtype=int)
niter = 0
# Initialise the value and epsilon as proposed in the course.
V0 = np.array([1/(1 - gamma) for _ in range(env.nS)])
V1 = np.zeros(env.nS)
epsilon = epsilon * (1 - gamma) / (2 * gamma)
# The main loop of the Value Iteration algorithm.
while True:
niter += 1
for s in range(env.nS):
for a in range(env.nA):
temp = env.R[s, a] + gamma * sum([V * p for (V, p) in zip(V0, env.P[s, a])])
if (a == 0) or (temp > V1[s]):
V1[s] = temp
policy[s] = a
# Testing the stopping criterion (+1 abitrary stop when 'max_iter' is reached).
if np.linalg.norm(V1 - V0) < epsilon:
return niter, V0, policy
else:
V0 = V1
V1 = np.array([1/(1 - gamma) for _ in range(env.nS)])
if niter > max_iter:
print("No convergence in VI after: ", max_iter, " steps!")
return niter, V0, policy
# A first implementation of the PI algorithms, using a matrix inversion to do the policy evaluation step.
def PI(env, gamma = 0.9):
# Initialisation of the variables.
policy0 = np.random.randint(env.nA, size = env.nS)
policy1 = np.zeros(env.nS, dtype = int)
niter = 0
# The main loop of the PI algorithm.
while True:
niter += 1
# Policy evaluation step.
P_pi = np.array([[env.P[s, policy0[s], ss] for ss in range(env.nS)] for s in range(env.nS)])
R_pi = np.array([env.R[s, policy0[s]] for s in range(env.nS)])
V0 = np.linalg.inv((np.eye(env.nS) - gamma * P_pi)) @ R_pi
V1 = np.zeros(env.nS)
# Updating the policy.
for s in range(env.nS):
for a in range(env.nA):
temp = env.R[s, a] + gamma * sum([u * p for (u, p) in zip(V0, env.P[s, a])])
if (a == 0) or (temp > V1[s]):
V1[s] = temp
policy1[s] = a
# Testing if the policy changed or not.
test = True
for s in range(env.nS):
if policy0[s] != policy1[s]:
test = False
break
if test:
return niter, policy1
else:
policy0 = policy1
policy1 = np.zeros(env.nS, dtype=int)
# An auxiliary function for the following algorithm.
# This an algorithmique alternative to the policy evaluation step of the PI.
# Nearly identical to the VI algorithm!
def policy_eval(policy, env, gamma, epsilon, max_iter):
# Initialisations.
niter = 0
V0 = np.array([1/(1 - gamma) for _ in range(env.nS)])
epsilon = epsilon * (1 - gamma) / (2 * gamma)
V1 = np.zeros(env.nS)
# A loop similar to the VI, but with a fixed policy.
while True:
niter += 1
for s in range(env.nS):
a = policy[s]
V1[s] = env.R[s, a] + gamma * sum([V * p for (V, p) in zip(V0, env.P[s, a])])
# The stopping criterion and an arbitrary limit on the number of iterations.
if np.linalg.norm(V1 - V0) < epsilon:
return V0
else:
V0 = V1
V1 = np.array([1/(1 - gamma) for _ in range(env.nS)])
if niter > max_iter:
print("No convergence in policy evaluation!")
return V0
# Another possible implementation of the PI, using an algorithmique policy evaluation instead of the matrix inversion.
# While it does introduce numerical estimation and thus some incertitude with epsilon too big, it can also prove computationally
# more efficient!
def PI_alternative(env, gamma = 0.9, max_iter = 10**5, epsilon = 10**(-2)):
# Initialisation of the variables.
policy0 = np.random.randint(env.nA, size = env.nS)
policy1 = np.zeros(env.nS, dtype = int)
niter = 0
# The main loop of the PI algorithm.
while True:
niter += 1
# Policy evaluation step.
V0 = policy_eval(policy0, env, gamma, epsilon, max_iter)
V1 = np.zeros(env.nS)
# Updating the policy.
for s in range(env.nS):
for a in range(env.nA):
temp = env.R[s, a] + gamma * sum([u * p for (u, p) in zip(V0, env.P[s, a])])
if (a == 0) or (temp > V1[s]):
V1[s] = temp
policy1[s] = a
# Testing if the policy changed or not.
test = True
for s in range(env.nS):
if policy0[s] != policy1[s]:
test = False
break
if test:
return niter, policy1
else:
policy0 = policy1
policy1 = np.zeros(env.nS, dtype=int)
####################################################################################################################################################
####################################################################################################################################################
####################################################################################################################################################
# Running experiments
####################################################################################################################################################
####################################################################################################################################################
####################################################################################################################################################
env = four_room()
print(VI(env))