"""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))]