In [1]:
%matplotlib inline
import gym
import matplotlib
import numpy as np
import sys
from collections import defaultdict
if "../" not in sys.path:
sys.path.append("../")
from lib.envs.blackjack import BlackjackEnv
from lib import plotting
matplotlib.style.use('ggplot')
In [2]:
env = BlackjackEnv()
In [3]:
def create_random_policy(nA):
"""
Creates a random policy function.
Args:
nA: Number of actions in the environment.
Returns:
A function that takes an observation as input and returns a vector
of action probabilities
"""
A = np.ones(nA, dtype=float) / nA
def policy_fn(observation):
return A
return policy_fn
In [4]:
def create_greedy_policy(Q):
"""
Creates a greedy policy based on Q values.
Args:
Q: A dictionary that maps from state -> action values
Returns:
A function that takes an observation as input and returns a vector
of action probabilities.
"""
def policy_fn(observation):
return np.eye(len(Q[observation]),
dtype=float)[np.argmax(Q[observation])]
return policy_fn
In [5]:
def mc_control_importance_sampling(env, num_episodes, behavior_policy, discount_factor=1.0):
"""
Monte Carlo Control Off-Policy Control using Weighted Importance Sampling.
Finds an optimal greedy policy.
Args:
env: OpenAI gym environment.
num_episodes: Nubmer of episodes to sample.
behavior_policy: The behavior to follow while generating episodes.
A function that given an observation returns a vector of probabilities for each action.
discount_factor: Lambda discount factor.
Returns:
A tuple (Q, policy).
Q is a dictionary mapping state -> action values.
policy is a function that takes an observation as an argument and returns
action probabilities. This is the optimal greedy policy.
"""
# The final action-value function.
# A dictionary that maps state -> action values
Q = defaultdict(lambda: np.zeros(env.action_space.n))
# The cumulative denominator of the weighted importance sampling formula
# (across all episodes)
C = defaultdict(lambda: np.zeros(env.action_space.n))
# Our greedily policy we want to learn
target_policy = create_greedy_policy(Q)
for _ in range(num_episodes):
done = False
episode = []
s = env.reset()
while not done:
probs = behavior_policy(s)
action = np.random.choice(np.arange(len(probs)), p=probs)
ns, reward, done, info = env.step(action)
episode.append((s, action, reward))
s = ns
# Sum of discounted returns
G = 0.0
# The importance sampling ratio (the weights of the returns)
W = 1.0
for s, a, r in episode[::-1]:
G = discount_factor * G + r
C[s][a] += W
Q[s][a] += W/C[s][a] * (G - Q[s][a])
if a != np.argmax(target_policy(s)):
break
W *= 1.0/behavior_policy(s)[a]
return Q, target_policy
In [6]:
random_policy = create_random_policy(env.action_space.n)
Q, policy = mc_control_importance_sampling(env, num_episodes=500000, behavior_policy=random_policy)
In [7]:
# For plotting: Create value function from action-value function
# by picking the best action at each state
V = defaultdict(float)
for state, action_values in Q.items():
action_value = np.max(action_values)
V[state] = action_value
plotting.plot_value_function(V, title="Optimal Value Function")
In [ ]: