------------恢复内容开始------------
这两天年总算把自己的游戏写完了,也通过了所有的测试。
我将自己的代码上传到了我的github上, 地址是 https://github.com/fraser-in-school/OpenSpiel-BattleChess/tree/master/open_spiel/games
游戏名称是 battle_chess 该游戏总共有黑白双色,三种棋子,双方各五个棋,总共10个棋子。
现在的任务是需要训练这个游戏的agent(智能体)
还是先通过example 入手。
我们自己写的游戏类型是 二人的回合制完美信息零和博弈游戏,这个类型与breakthrough是相同的。
于是我把breakthrough_dqn 的代码直接粘贴过去了生成了一个新的文件
我把游戏名称改了一下,然后游戏的命令行参数设置成自己游戏的,就可以尝试运行起来了。
不过遗憾的是,这个py文件没有什么注释,只能根据代码来解读。
首先这个一个两个函数,一个main函数,另外一个就是这个函数
这个函数名字大概意思就是 评估该算法和随机(这里的随机表示随机的算法)智能体
1 def eval_against_random_bots(env, trained_agents, random_agents, num_episodes): 2 """Evaluates `trained_agents` against `random_agents` for `num_episodes`.""" 3 num_players = len(trained_agents) 4 sum_episode_rewards = np.zeros(num_players) 5 for player_pos in range(num_players): 6 cur_agents = random_agents[:] 7 cur_agents[player_pos] = trained_agents[player_pos] 8 for _ in range(num_episodes): 9 time_step = env.reset() 10 episode_rewards = 0 11 while not time_step.last(): 12 player_id = time_step.observations["current_player"] 13 if env.is_turn_based: 14 agent_output = cur_agents[player_id].step( 15 time_step, is_evaluation=True) 16 action_list = [agent_output.action] 17 else: 18 agents_output = [ 19 agent.step(time_step, is_evaluation=True) for agent in cur_agents 20 ] 21 action_list = [agent_output.action for agent_output in agents_output] 22 time_step = env.step(action_list) 23 episode_rewards += time_step.rewards[player_pos] 24 sum_episode_rewards[player_pos] += episode_rewards 25 return sum_episode_rewards / num_episodes
我python也不是很好,那我们就一步一步来,首先需要搞清楚
trained_agents, random_agents 到底是什么,使用搜索我们很快就可以看到它们在main函数里面的定义
1 # random agents for evaluation 2 random_agents = [ 3 random_agent.RandomAgent(player_id=idx, num_actions=num_actions) 4 for idx in range(num_players) 5 ]
然后是 trained_agents
1 agents = [ 2 dqn.DQN( 3 session=sess, 4 player_id=idx, 5 state_representation_size=info_state_size, 6 num_actions=num_actions, 7 hidden_layers_sizes=hidden_layers_sizes, 8 replay_buffer_capacity=FLAGS.replay_buffer_capacity, 9 batch_size=FLAGS.batch_size) for idx in range(num_players) 10 ]
这个方括号是怎么用我也不知道,百度一下是列表解析 大概就是将 idx 循环放入左边的表达式/函数中,得到的每一个结果作为列表的一个元素。
也就是其实这个agents里面有两个agents, 一个是 player=0, 一个是player=1
这跟我原本以为的一个agents对战一个random agent有出入。所以还是需要看代码,看不懂一点一点的去百度,不要怕。
然后接着看这个函数
num_players = len(trained_agents)
sum_episode_rewards = np.zeros(num_players)
这两句就是简单的初始化,num_players = 2, 第二个变量是reward的累计。
然后是外层的for循环
for player_pos in range(num_players):
cur_agents = random_agents[:]
cur_agents[player_pos] = trained_agents[player_pos]
这里先是curagents全部初始化随机,然后如果player_pos = 0, curagents[0] = train_agents
这两句代码就实现了curagents 总有一个随机agent和一个train_agents,他们会进行对战。
然后是内层的for 循环
for _ in range(num_episodes):
time_step = env.reset()
episode_rewards = 0
这里我们需要去了解 env.reset()
下面是rl_environment.py 的代码
def reset(self): """Starts a new sequence and returns the first `TimeStep` of this sequence. Returns: A `TimeStep` namedtuple containing: observations: list of dicts containing one observations per player, each corresponding to `observation_spec()`. rewards: list of rewards at this timestep, or None if step_type is `StepType.FIRST`. discounts: list of discounts in the range [0, 1], or None if step_type is `StepType.FIRST`. step_type: A `StepType` value. """ self._should_reset = False self._state = self._game.new_initial_state() self._sample_external_events() observations = {"info_state": [], "legal_actions": [], "current_player": []} for player_id in range(self.num_players): observations["info_state"].append( self._state.observation_tensor(player_id) if self._use_observation else self._state.information_state_tensor(player_id)) observations["legal_actions"].append(self._state.legal_actions(player_id)) observations["current_player"] = self._state.current_player() return TimeStep( observations=observations, rewards=None, discounts=None, step_type=StepType.FIRST)
你只看return 也可以知道这个就是重置游戏的函数
然后看到再内层的while循环
while not time_step.last(): player_id = time_step.observations["current_player"] if env.is_turn_based: agent_output = cur_agents[player_id].step( time_step, is_evaluation=True) action_list = [agent_output.action] else: agents_output = [ agent.step(time_step, is_evaluation=True) for agent in cur_agents ] action_list = [agent_output.action for agent_output in agents_output] time_step = env.step(action_list) episode_rewards += time_step.rewards[player_pos]
如果不是最后一步,就继续,说明一个while循环就是一整把游戏。而上一层循环有多少次就说明每次调用该函数会进行多少把游戏,而上层循环的num_episodes是该函数的参数,可以看见为1000
if 的判断条件就是这个游戏是不是回合制游戏
首先agent_output 是由agent的step函数得到的
def step(self, time_step, is_evaluation=False): # If it is the end of the episode, don't select an action. if time_step.last(): return # Pick a random legal action. cur_legal_actions = time_step.observations["legal_actions"][self._player_id] action = np.random.choice(cur_legal_actions) probs = np.zeros(self._num_actions) probs[cur_legal_actions] = 1.0 / len(cur_legal_actions) return rl_agent.StepOutput(action=action, probs=probs)
上面是随机agent的代码,就是通过observation得到合法动作的列表。
至于最后的
rl_agent.StepOutput(action=action, probs=probs)
这个是random_agent的父类的一个函数
StepOutput = collections.namedtuple("step_output", ["action", "probs"])
nametuple 是一种类似字典的写法,去看看百度的用法。
我这里写了一个小示例
你大概可以看作一个有自己名字的字典。
所以这个random_agent 返回的是一个有自己名字的字典,字典里面两个值,一个是action,这个action是随机选取的,probs 是一个数组,这个数组的作用我现在还不清楚。
接着看我们的while循环,最后的
action_list = [agent_output.action]
相当于取出了字典里面的action值然后放入到了列表里
在这里的代码action_list 只有一个值,设置list应该是因为别的游戏有要求。
time_step = env.step(action_list)
episode_rewards += time_step.rewards[player_pos]
只剩这两行代码了,去看看 rl_environment的step函数,可以看到,这个就是使用action来更新环境的,这里的环境也可以理解为游戏状态。
整个函数的返回值就是reward值,这里是每一千次游戏获得的平均值,这个reward是一个数组,数组大小等于player人数,第一个为player=0获得的平均奖励,第二个为player=1获得的奖励。对于零和博弈,
一般设置为赢的一方得一分,输的一方得-1分,平局双方得零分。注意这里reward两个值加起来不等于0,因为这个都是dqn——agnet分别作为先手和后手的得分。
这篇博客就写到这里吧!
这里我提出一个问题供读者思考,一次函数调用共进行了多少局完整的游戏?这些游戏是怎么进行的?欢迎你们的回答!