真香了！上手使用 DeepMind 分布式强化学习框架 Acme ，对开发者超友好

env = acme.wrappers.GymWrapper(gym.make('Blackjack-v0'))

class RandomAgent(acme.Actor):"""A random agent for the Black Jack environment."""
def __init__(self):
# init action values, will not be updated by random agentself.Q = np.zeros((32,11,2,2))
# specify the behavior policyself.behavior_policy = lambda q_values: np.random.choice(2)
# store timestep, action, next_timestepself.timestep = Noneself.action = Noneself.next_timestep = None
def select_action(self, observation):"Choose an action according to the behavior policy."return self.behavior_policy(self.Q[observation])def observe_first(self, timestep):"Observe the first timestep."self.timestep = timestepdef observe(self, action, next_timestep):"Observe the next timestep."self.action = actionself.next_timestep = next_timestep
def update(self, wait = False):"Update the policy."# no updates occur here, it's just a random policy

# first initialize env and agent# env = ...agent = RandomAgent()# repeat for a number of episodesfor episode in range(10):# make first observationtimestep = env.reset()agent.observe_first(timestep)# run an episodewhile not timestep.last():# generate an action from the agent's policyaction = agent.select_action(timestep.observation)
# step the environmenttimestep = env.step(action)# have the agent observe the next timestepagent.observe(action, next_timestep=timestep)
# let the agent perform updatesagent.update()

# init Acme's environment looploop = EnvironmentLoop(env, agent, logger=InMemoryLogger())# run a single episodeloop.run_episode()# or run multiple episodesloop.run(10

class SarsaAgent(acme.Actor):
def __init__(self, env_specs=None, epsilon=0.1, step_size=0.1):
# in Black Jack, we have the following dimensionsself.Q = np.zeros((32,11,2,2))
# epsilon for policy and step_size for TD learningself.epsilon = epsilonself.step_size = step_size
# set behavior policyself.behavior_policy = lambda q_values: epsilon_greedy(q_values, self.epsilon)
# store timestep, action, next_timestepself.timestep = Noneself.action = Noneself.next_timestep = Nonedef transform_state(self, state):# this is specifally required for the blackjack environmentstate = *map(int, state),return state
def select_action(self, observation):state = self.transform_state(observation)return self.behavior_policy(self.Q[state])def observe_first(self, timestep):self.timestep = timestepdef observe(self, action, next_timestep):self.action = actionself.next_timestep = next_timestep
def update(self):
# get variables for conveniencestate = self.timestep.observation_, reward, discount, next_state = self.next_timestepaction = self.action
# turn states into indicesstate = self.transform_state(state)next_state = self.transform_state(next_state)
# sample a next actionnext_action = self.behavior(self.Q[next_state])# compute and apply the TD errortd_error = reward + discount * self.Q[next_state][next_action] - self.Q[state][self.action]self.Q[state][action] += self.step_size * td_error
# finally, set timestep to next_timestepself.timestep = self.next_timestep

agent = SarsaAgent()loop = EnvironmentLoop(env, agent, logger=InMemoryLogger())loop.run(500000)

class QLearningAgent(acme.Actor):
def __init__(self, env_specs=None, step_size=0.1):
# Black Jack dimensionsself.Q = np.zeros((32,11,2,2))
# set step sizeself.step_size = step_size
# set behavior policy# self.policy = Noneself.behavior_policy = lambda q_values: epsilon_greedy(q_values, epsilon=0.1)
# store timestep, action, next_timestepself.timestep = Noneself.action = Noneself.next_timestep = Nonedef state_to_index(self, state):state = *map(int, state),return state
def transform_state(self, state):# this is specifally required for the blackjack environmentstate = *map(int, state),return state
def select_action(self, observation):state = self.transform_state(observation)return self.behavior_policy(self.Q[state])def observe_first(self, timestep):self.timestep = timestepdef observe(self, action, next_timestep):self.action = actionself.next_timestep = next_timestepdef update(self):# get variables for conveniencestate = self.timestep.observation_, reward, discount, next_state = self.next_timestepaction = self.action
# turn states into indicesstate = self.transform_state(state)next_state = self.transform_state(next_state)
# Q-value updatetd_error = reward + discount * np.max(self.Q[next_state]) - self.Q[state][action]self.Q[state][action] += self.step_size * td_error
# finally, set timestep to next_timestepself.timestep = self.next_timestep

agent = QLearningAgent()loop = EnvironmentLoop(env, agent, logger=InMemoryLogger())loop.run(500000)