Skip to content
Navigation Menu
{{ message }}
forked from aimacode/aima-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathreinforcement_learning4e.py
More file actions
353 lines (299 loc) · 11.4 KB
/
Copy pathreinforcement_learning4e.py
File metadata and controls
353 lines (299 loc) · 11.4 KB
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
343
344
345
346
347
348
349
350
351
352
353
"""Reinforcement Learning (Chapter 21)"""
import random
from collections import defaultdict
from mdp4e import MDP, policy_evaluation
# _________________________________________
# 21.2 Passive Reinforcement Learning
# 21.2.1 Direct utility estimation
class PassiveDUEAgent:
"""
Passive (non-learning) agent that uses direct utility estimation
on a given MDP and policy.
import sys
from mdp import sequential_decision_environment
north = (0, 1)
south = (0,-1)
west = (-1, 0)
east = (1, 0)
policy = {(0, 2): east, (1, 2): east, (2, 2): east, (3, 2): None, (0, 1): north, (2, 1): north,
(3, 1): None, (0, 0): north, (1, 0): west, (2, 0): west, (3, 0): west,}
agent = PassiveDUEAgent(policy, sequential_decision_environment)
for i in range(200):
run_single_trial(agent,sequential_decision_environment)
agent.estimate_U()
agent.U[(0, 0)] > 0.2
True
"""
def __init__(self, pi, mdp):
self.pi = pi
self.mdp = mdp
self.U = {}
self.s = None
self.a = None
self.s_history = []
self.r_history = []
self.init = mdp.init
def __call__(self, percept):
s1, r1 = percept
self.s_history.append(s1)
self.r_history.append(r1)
##
##
if s1 in self.mdp.terminals:
self.s = self.a = None
else:
self.s, self.a = s1, self.pi[s1]
return self.a
def estimate_U(self):
# this function can be called only if the MDP has reached a terminal state
# it will also reset the mdp history
assert self.a is None, 'MDP is not in terminal state'
assert len(self.s_history) == len(self.r_history)
# calculating the utilities based on the current iteration
U2 = {s: [] for s in set(self.s_history)}
for i in range(len(self.s_history)):
s = self.s_history[i]
U2[s] += [sum(self.r_history[i:])]
U2 = {k: sum(v) / max(len(v), 1) for k, v in U2.items()}
# resetting history
self.s_history, self.r_history = [], []
# setting the new utilities to the average of the previous
# iteration and this one
for k in U2.keys():
if k in self.U.keys():
self.U[k] = (self.U[k] + U2[k]) / 2
else:
self.U[k] = U2[k]
return self.U
def update_state(self, percept):
"""To be overridden in most cases. The default case
assumes the percept to be of type (state, reward)"""
return percept
# 21.2.2 Adaptive dynamic programming
class PassiveADPAgent:
"""
[Figure 21.2]
Passive (non-learning) agent that uses adaptive dynamic programming
on a given MDP and policy.
import sys
from mdp import sequential_decision_environment
north = (0, 1)
south = (0,-1)
west = (-1, 0)
east = (1, 0)
policy = {(0, 2): east, (1, 2): east, (2, 2): east, (3, 2): None, (0, 1): north, (2, 1): north,
(3, 1): None, (0, 0): north, (1, 0): west, (2, 0): west, (3, 0): west,}
agent = PassiveADPAgent(policy, sequential_decision_environment)
for i in range(100):
run_single_trial(agent,sequential_decision_environment)
agent.U[(0, 0)] > 0.2
True
agent.U[(0, 1)] > 0.2
True
"""
class ModelMDP(MDP):
"""Class for implementing modified Version of input MDP with
an editable transition model P and a custom function T."""
def __init__(self, init, actlist, terminals, gamma, states):
super().__init__(init, actlist, terminals, states=states, gamma=gamma)
nested_dict = lambda: defaultdict(nested_dict)
# StackOverflow:whats-the-best-way-to-initialize-a-dict-of-dicts-in-python
self.P = nested_dict()
def T(self, s, a):
"""Return a list of tuples with probabilities for states
based on the learnt model P."""
return [(prob, res) for (res, prob) in self.P[(s, a)].items()]
def __init__(self, pi, mdp):
self.pi = pi
self.mdp = PassiveADPAgent.ModelMDP(mdp.init, mdp.actlist,
mdp.terminals, mdp.gamma, mdp.states)
self.U = {}
self.Nsa = defaultdict(int)
self.Ns1_sa = defaultdict(int)
self.s = None
self.a = None
self.visited = set() # keeping track of visited states
def __call__(self, percept):
s1, r1 = percept
mdp = self.mdp
R, P, terminals, pi = mdp.reward, mdp.P, mdp.terminals, self.pi
s, a, Nsa, Ns1_sa, U = self.s, self.a, self.Nsa, self.Ns1_sa, self.U
if s1 not in self.visited: # Reward is only known for visited state.
U[s1] = R[s1] = r1
self.visited.add(s1)
if s is not None:
Nsa[(s, a)] += 1
Ns1_sa[(s1, s, a)] += 1
# for each t such that Ns′|sa [t, s, a] is nonzero
for t in [res for (res, state, act), freq in Ns1_sa.items()
if (state, act) == (s, a) and freq != 0]:
P[(s, a)][t] = Ns1_sa[(t, s, a)] / Nsa[(s, a)]
self.U = policy_evaluation(pi, U, mdp)
##
##
self.Nsa, self.Ns1_sa = Nsa, Ns1_sa
if s1 in terminals:
self.s = self.a = None
else:
self.s, self.a = s1, self.pi[s1]
return self.a
def update_state(self, percept):
"""To be overridden in most cases. The default case
assumes the percept to be of type (state, reward)."""
return percept
# 21.2.3 Temporal-difference learning
class PassiveTDAgent:
"""
[Figure 21.4]
The abstract class for a Passive (non-learning) agent that uses
temporal differences to learn utility estimates. Override update_state
method to convert percept to state and reward. The mdp being provided
should be an instance of a subclass of the MDP Class.
import sys
from mdp import sequential_decision_environment
north = (0, 1)
south = (0,-1)
west = (-1, 0)
east = (1, 0)
policy = {(0, 2): east, (1, 2): east, (2, 2): east, (3, 2): None, (0, 1): north, (2, 1): north,
(3, 1): None, (0, 0): north, (1, 0): west, (2, 0): west, (3, 0): west,}
agent = PassiveTDAgent(policy, sequential_decision_environment, alpha=lambda n: 60./(59+n))
for i in range(200):
run_single_trial(agent,sequential_decision_environment)
agent.U[(0, 0)] > 0.2
True
agent.U[(0, 1)] > 0.2
True
"""
def __init__(self, pi, mdp, alpha=None):
self.pi = pi
self.U = {s: 0. for s in mdp.states}
self.Ns = {s: 0 for s in mdp.states}
self.s = None
self.a = None
self.r = None
self.gamma = mdp.gamma
self.terminals = mdp.terminals
if alpha:
self.alpha = alpha
else:
self.alpha = lambda n: 1 / (1 + n) # udacity video
def __call__(self, percept):
s1, r1 = self.update_state(percept)
pi, U, Ns, s, r = self.pi, self.U, self.Ns, self.s, self.r
alpha, gamma, terminals = self.alpha, self.gamma, self.terminals
if not Ns[s1]:
U[s1] = r1
if s is not None:
Ns[s] += 1
U[s] += alpha(Ns[s]) * (r + gamma * U[s1] - U[s])
if s1 in terminals:
self.s = self.a = self.r = None
else:
self.s, self.a, self.r = s1, pi[s1], r1
return self.a
def update_state(self, percept):
"""To be overridden in most cases. The default case
assumes the percept to be of type (state, reward)."""
return percept
# __________________________________________
# 21.3. Active Reinforcement Learning
# 21.3.2 Learning an action-utility function
class QLearningAgent:
"""
[Figure 21.8]
An exploratory Q-learning agent. It avoids having to learn the transition
model because the Q-value of a state can be related directly to those of
its neighbors.
import sys
from mdp import sequential_decision_environment
north = (0, 1)
south = (0,-1)
west = (-1, 0)
east = (1, 0)
policy = {(0, 2): east, (1, 2): east, (2, 2): east, (3, 2): None, (0, 1): north, (2, 1): north,
(3, 1): None, (0, 0): north, (1, 0): west, (2, 0): west, (3, 0): west,}
q_agent = QLearningAgent(sequential_decision_environment, Ne=5, Rplus=2, alpha=lambda n: 60./(59+n))
for i in range(200):
run_single_trial(q_agent,sequential_decision_environment)
q_agent.Q[((0, 1), (0, 1))] >= -0.5
True
q_agent.Q[((1, 0), (0, -1))] <= 0.5
True
"""
def __init__(self, mdp, Ne, Rplus, alpha=None):
self.gamma = mdp.gamma
self.terminals = mdp.terminals
self.all_act = mdp.actlist
self.Ne = Ne # iteration limit in exploration function
self.Rplus = Rplus # large value to assign before iteration limit
self.Q = defaultdict(float)
self.Nsa = defaultdict(float)
self.s = None
self.a = None
self.r = None
if alpha:
self.alpha = alpha
else:
self.alpha = lambda n: 1. / (1 + n) # udacity video
def f(self, u, n):
"""Exploration function. Returns fixed Rplus until
agent has visited state, action a Ne number of times.
Same as ADP agent in book."""
if n < self.Ne:
return self.Rplus
else:
return u
def actions_in_state(self, state):
"""Return actions possible in given state.
Useful for max and argmax."""
if state in self.terminals:
return [None]
else:
return self.all_act
def __call__(self, percept):
s1, r1 = self.update_state(percept)
Q, Nsa, s, a, r = self.Q, self.Nsa, self.s, self.a, self.r
alpha, gamma, terminals = self.alpha, self.gamma, self.terminals,
actions_in_state = self.actions_in_state
if s in terminals:
Q[s, None] = r1
if s is not None:
Nsa[s, a] += 1
Q[s, a] += alpha(Nsa[s, a]) * (r + gamma * max(Q[s1, a1]
for a1 in actions_in_state(s1)) - Q[s, a])
if s in terminals:
self.s = self.a = self.r = None
else:
self.s, self.r = s1, r1
self.a = max(actions_in_state(s1), key=lambda a1: self.f(Q[s1, a1], Nsa[s1, a1]))
return self.a
def update_state(self, percept):
"""To be overridden in most cases. The default case
assumes the percept to be of type (state, reward)."""
return percept
def run_single_trial(agent_program, mdp):
"""Execute trial for given agent_program
and mdp. mdp should be an instance of subclass
of mdp.MDP """
def take_single_action(mdp, s, a):
"""
Select outcome of taking action a
in state s. Weighted Sampling.
"""
x = random.uniform(0, 1)
cumulative_probability = 0.0
for probability_state in mdp.T(s, a):
probability, state = probability_state
cumulative_probability += probability
if x < cumulative_probability:
break
return state
current_state = mdp.init
while True:
current_reward = mdp.R(current_state)
percept = (current_state, current_reward)
next_action = agent_program(percept)
if next_action is None:
break
current_state = take_single_action(mdp, current_state, next_action)
You can’t perform that action at this time.