简介

OpenSpiel 是一个用于通用强化学习研究以及游戏中的搜索 / 规划的环境和算法集合。OpenSpiel 支持 n 人（单智能体和多智能体）零和、合作以及一般和博弈的游戏，这些游戏可以是一次性的和序列性的、严格轮流行动的和同时行动的、完全信息和不完全信息的，它还支持如（部分可观测和完全可观测的）网格世界以及社会困境等传统的多智能体环境。OpenSpiel 还包含用于分析学习动态以及其他常见评估指标的工具。游戏被表示为程序性的扩展式博弈，并带有一些自然的扩展。其核心应用程序编程接口（API）和游戏是用 C++ 实现的，并提供了 Python 接口。算法和工具既有用 C++ 编写的，也有用 Python 编写的

对应的github地址为https://github.com/google-deepmind/open_spiel

安装

先clone前面的git工程https://github.com/google-deepmind/open_spiel

进入clone下来的open_spiel目录下

安装python的依赖

系统版本是Ubuntu 22.04及以上执行下面的指令

Python
python3 -m venv ./venv
source venv/bin/activate
python3 -m pip install -r requirements.txt

如果是更低的版本执行下面的指令

Python
virtualenv -p python3 venv
source venv/bin/activate
python3 -m pip install -r requirements.txt

安装open_spiel库

使用pip指令即可安装open_spiel库

Plain Text
python3 -m pip install open_spiel

项目结构

open_spiel 目录下的直接子目录大多是 C++ 代码（integration_tests 和 python 目录除外）。open_spiel/python 目录中有类似的结构，包含与之对应的 Python 代码。

一些顶级目录比较特殊：

  • open_spiel/integration_tests：针对所有游戏的通用（Python）测试。
  • open_spiel/tests：C++ 通用测试工具。
  • open_spiel/scripts：对开发有用的脚本（用于构建、运行测试等）。

例如，对于 C++ 代码而言：

  • open_spiel/：包含游戏抽象的 C++ 应用程序编程接口（API）。
  • open_spiel/games：包含游戏的 C++ 实现代码。
  • open_spiel/algorithms：在 OpenSpiel 中实现的 C++ 算法。
  • open_spiel/examples：C++ 示例代码。
  • open_spiel/tests：C++ 通用测试工具。

对于 Python 代码来说：

  • open_spiel/python/examples：Python 示例代码。
  • open_spiel/python/algorithms/：Python 算法代码。

基础例子

可以进入open_spiel\open_spiel\python\examples目录下，里面有许多官方例子

example.py

代码解析

下面是一个最简单例子的完整代码。该例子是创建了一个tic tac toe（井字棋）的玩法，使用随机选择的方式，进行对局。井字棋的玩法逻辑是两名玩家顺序在3*3的格子下棋，当其中一方的三个棋子形成连线时获胜。若格子填满时没有玩家形成连线则平局

python
# Copyright 2019 DeepMind Technologies Limited
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Python spiel example."""

import random
from absl import app
from absl import flags
import numpy as np

from open_spiel.python import games  # pylint: disable=unused-import
import pyspiel

FLAGS = flags.FLAGS

# Game strings can just contain the name or the name followed by parameters
# and arguments, e.g. "breakthrough(rows=6,columns=6)"
flags.DEFINE_string("game_string", "tic_tac_toe", "Game string")

def main(_):
  #显示所有的spiel注册的游戏
  games_list = pyspiel.registered_games()
  print("Registered games:")
  print(games_list)

  action_string = None

  print("Creating game: " + FLAGS.game_string)
  game = pyspiel.load_game(FLAGS.game_string)

  # Create the initial state
  state = game.new_initial_state()

  # Print the initial state
  print(str(state))

  while not state.is_terminal():
    # The state can be three different types: chance node,
    # simultaneous node, or decision node
    if state.is_chance_node():
      # Chance node: sample an outcome
      outcomes = state.chance_outcomes()
      num_actions = len(outcomes)
      print("Chance node, got " + str(num_actions) + " outcomes")
      action_list, prob_list = zip(*outcomes)
      action = np.random.choice(action_list, p=prob_list)
      print("Sampled outcome: ",
            state.action_to_string(state.current_player(), action))
      state.apply_action(action)
    elif state.is_simultaneous_node():
      # Simultaneous node: sample actions for all players.
      random_choice = lambda a: np.random.choice(a) if a else [0]
      chosen_actions = [
          random_choice(state.legal_actions(pid))
          for pid in range(game.num_players())
      ]
      print("Chosen actions: ", [
          state.action_to_string(pid, action)
          for pid, action in enumerate(chosen_actions)
      ])
      state.apply_actions(chosen_actions)
    else:
      # Decision node: sample action for the single current player
      action = random.choice(state.legal_actions(state.current_player()))
      action_string = state.action_to_string(state.current_player(), action)
      print("Player ", state.current_player(), ", randomly sampled action: ",
            action_string)
      state.apply_action(action)
    print(str(state))

  # Game is now done. Print utilities for each player
  returns = state.returns()
  for pid in range(game.num_players()):
    print("Utility for player {} is {}".format(pid, returns[pid]))

if __name__ == "__main__":
  app.run(main)
 

game = pyspiel.load_game(FLAGS.game_string) 是创建了一个新的游戏

state = game.new_initial_state()获取初始状态

state.is_terminal()是判断游戏是否结束

每个state有三种类型的节点 chance node,simultaneous node和decision node

chance节点是概率节点，返回的是一个action和概率的列表。这个是属于玩家不能控制的节点。一般是由游戏本身来决定的。比如玩家执行了一次roll骰子的操作。chance节点就是action为[1,2,3,4,5,6],概率为[1/6,1/6,1/6,1/6,1/6]。需要根据概率来选择节点。

simultaneous节点是一个同步节点，需要所有玩家都进行操作

decision节点是决策节点，需要当前玩家进行操作

在决定好相关的操作之后，执行state.apply_action(action)。游戏会进入下一个状态。

输出

下面是游戏输出:"."代表未填的格子，"x"代表玩家1的操作，"o"代表玩家2操作，当格子填满的时候，或者其中一个玩家填的格子形成三个的连线时，对局结束。形成3个连线的玩家得1分，对手得-1分。平局则双方得0分

yaml
Creating game: tic_tac_toe
...
...
...
Player  0 , randomly sampled action:  x(2,1)
...
...
.x.
Player  1 , randomly sampled action:  o(1,0)
...
o..
.x.
Player  0 , randomly sampled action:  x(0,2)
..x
o..
.x.
Player  1 , randomly sampled action:  o(2,2)
..x
o..
.xo
Player  0 , randomly sampled action:  x(0,1)
.xx
o..
.xo
Player  1 , randomly sampled action:  o(1,2)
.xx
o.o
.xo
Player  0 , randomly sampled action:  x(0,0)
xxx
o.o
.xo
Utility for player 0 is 1.0
Utility for player 1 is -1.0

rl_example.py

代码解析

下面是一个关于tic tac toe的强化学习的例子

python
# Copyright 2019 DeepMind Technologies Limited
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Python spiel example."""

import logging
from absl import app
from absl import flags
import numpy as np

from open_spiel.python import rl_environment

FLAGS = flags.FLAGS

flags.DEFINE_string("game", "tic_tac_toe", "Name of the game")
flags.DEFINE_integer("num_players", None, "Number of players")

def select_actions(observations, cur_player):
  cur_legal_actions = observations["legal_actions"][cur_player]
  actions = [np.random.choice(cur_legal_actions)]
  return actions

def print_iteration(time_step, actions, player_id):
  """Print TimeStep information."""
  obs = time_step.observations
  logging.info("Player: %s", player_id)
  if time_step.step_type.first():
    logging.info("Info state: %s, - - %s", obs["info_state"][player_id],
                 time_step.step_type)
  else:
    logging.info("Info state: %s, %s %s %s", obs["info_state"][player_id],
                 time_step.rewards[player_id], time_step.discounts[player_id],
                 time_step.step_type)
  logging.info("Action taken: %s", actions)
  logging.info("-" * 80)

def turn_based_example(unused_arg):
  """Example usage of the RL environment for turn-based games."""
  # `rl_main_loop.py` contains more details and simultaneous move examples.
  logging.info("Registered games: %s", rl_environment.registered_games())
  logging.info("Creating game %s", FLAGS.game)

  env_configs = {"players": FLAGS.num_players} if FLAGS.num_players else {}
  env = rl_environment.Environment(FLAGS.game, **env_configs)

  logging.info("Env specs: %s", env.observation_spec())
  logging.info("Action specs: %s", env.action_spec())

  time_step = env.reset()

  while not time_step.step_type.last():
    pid = time_step.observations["current_player"]
    actions = select_actions(time_step.observations, pid)
    print_iteration(time_step, actions, pid)
    time_step = env.step(actions)

  # Print final state of end game.
  for pid in range(env.num_players):
    print_iteration(time_step, actions, pid)

if __name__ == "__main__":
  app.run(turn_based_example)
 

env = rl_environment.Environment(FLAGS.game, **env_configs)由于强化学习，所以这里创建了一个代表环境的env，参数是游戏名

time_step = env.reset()重置环境

time_step.step_type.last()判断是否结束

time_step.observations代表当前状态

observations["legal_actions"][cur_player]为指定玩家可执行的操作

env.step(actions)执行动作，进行状态转移和给出奖励

输出

yaml
I0410 15:24:27.507118 139874665504896 rl_example.py:55] Creating game tic_tac_toe
I0410 15:24:27.538239 139874665504896 rl_environment.py:182] Using game string: tic_tac_toe
I0410 15:24:27.538874 139874665504896 rl_example.py:60] Env specs: {'info_state': (27,), 'legal_actions': (9,), 'current_player': (), 'serialized_state': ()}
I0410 15:24:27.538978 139874665504896 rl_example.py:61] Action specs: {'num_actions': 9, 'min': 0, 'max': 8, 'dtype': <class 'int'>}
I0410 15:24:27.540167 139874665504896 rl_example.py:39] Player: 0
I0410 15:24:27.540271 139874665504896 rl_example.py:41] Info state: [1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], - - StepType.FIRST
I0410 15:24:27.540327 139874665504896 rl_example.py:47] Action taken: [np.int64(1)]
I0410 15:24:27.540408 139874665504896 rl_example.py:48] --------------------------------------------------------------------------------
I0410 15:24:27.540894 139874665504896 rl_example.py:39] Player: 1
I0410 15:24:27.540965 139874665504896 rl_example.py:44] Info state: [1.0, 0.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], 0.0 1.0 StepType.MID
I0410 15:24:27.541048 139874665504896 rl_example.py:47] Action taken: [np.int64(7)]
I0410 15:24:27.541108 139874665504896 rl_example.py:48] --------------------------------------------------------------------------------
I0410 15:24:27.541267 139874665504896 rl_example.py:39] Player: 0
I0410 15:24:27.541320 139874665504896 rl_example.py:44] Info state: [1.0, 0.0, 1.0, 1.0, 1.0, 1.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], 0.0 1.0 StepType.MID
I0410 15:24:27.541370 139874665504896 rl_example.py:47] Action taken: [np.int64(3)]
I0410 15:24:27.541446 139874665504896 rl_example.py:48] --------------------------------------------------------------------------------
I0410 15:24:27.541556 139874665504896 rl_example.py:39] Player: 1
I0410 15:24:27.541639 139874665504896 rl_example.py:44] Info state: [1.0, 0.0, 1.0, 0.0, 1.0, 1.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0], 0.0 1.0 StepType.MID
I0410 15:24:27.541731 139874665504896 rl_example.py:47] Action taken: [np.int64(6)]
I0410 15:24:27.541793 139874665504896 rl_example.py:48] --------------------------------------------------------------------------------
I0410 15:24:27.541923 139874665504896 rl_example.py:39] Player: 0
I0410 15:24:27.542005 139874665504896 rl_example.py:44] Info state: [1.0, 0.0, 1.0, 0.0, 1.0, 1.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 1.0, 0.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0], 0.0 1.0 StepType.MID
I0410 15:24:27.542053 139874665504896 rl_example.py:47] Action taken: [np.int64(5)]
I0410 15:24:27.542098 139874665504896 rl_example.py:48] --------------------------------------------------------------------------------
I0410 15:24:27.542215 139874665504896 rl_example.py:39] Player: 1
I0410 15:24:27.542295 139874665504896 rl_example.py:44] Info state: [1.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 1.0, 0.0, 0.0, 1.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0], 0.0 1.0 StepType.MID
I0410 15:24:27.542353 139874665504896 rl_example.py:47] Action taken: [np.int64(0)]
I0410 15:24:27.542407 139874665504896 rl_example.py:48] --------------------------------------------------------------------------------
I0410 15:24:27.542520 139874665504896 rl_example.py:39] Player: 0
I0410 15:24:27.542574 139874665504896 rl_example.py:44] Info state: [0.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 1.0, 0.0, 0.0, 1.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0], 0.0 1.0 StepType.MID
I0410 15:24:27.542623 139874665504896 rl_example.py:47] Action taken: [np.int64(8)]
I0410 15:24:27.542680 139874665504896 rl_example.py:48] --------------------------------------------------------------------------------
I0410 15:24:27.542805 139874665504896 rl_example.py:39] Player: 1
I0410 15:24:27.542854 139874665504896 rl_example.py:44] Info state: [0.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 1.0, 0.0, 0.0, 1.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 1.0], 0.0 1.0 StepType.MID
I0410 15:24:27.542913 139874665504896 rl_example.py:47] Action taken: [np.int64(4)]
I0410 15:24:27.542972 139874665504896 rl_example.py:48] --------------------------------------------------------------------------------
I0410 15:24:27.543055 139874665504896 rl_example.py:39] Player: 0
I0410 15:24:27.543102 139874665504896 rl_example.py:44] Info state: [0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 1.0, 1.0, 0.0, 0.0, 1.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 1.0], 0.0 1.0 StepType.MID
I0410 15:24:27.543179 139874665504896 rl_example.py:47] Action taken: [np.int64(2)]
I0410 15:24:27.543238 139874665504896 rl_example.py:48] --------------------------------------------------------------------------------
I0410 15:24:27.543318 139874665504896 rl_example.py:39] Player: 0
I0410 15:24:27.543376 139874665504896 rl_example.py:44] Info state: [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 1.0, 1.0, 0.0, 0.0, 1.0, 1.0, 1.0, 0.0, 1.0, 0.0, 0.0, 1.0], 1.0 0.0 StepType.LAST
I0410 15:24:27.543421 139874665504896 rl_example.py:47] Action taken: [np.int64(2)]
I0410 15:24:27.543480 139874665504896 rl_example.py:48] --------------------------------------------------------------------------------
I0410 15:24:27.543520 139874665504896 rl_example.py:39] Player: 1
I0410 15:24:27.543589 139874665504896 rl_example.py:44] Info state: [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 1.0, 1.0, 0.0, 0.0, 1.0, 1.0, 1.0, 0.0, 1.0, 0.0, 0.0, 1.0], -1.0 0.0 StepType.LAST
I0410 15:24:27.543634 139874665504896 rl_example.py:47] Action taken: [np.int64(2)]
I0410 15:24:27.543707 139874665504896 rl_example.py:48] --------------------------------------------------------------------------------

mcts.py

代码解析

下面是一个蒙特卡洛搜索树算法的一个例子。让一个蒙特卡洛搜索的机器人和其他机器人对打。完成若干场tic tac toe的对局

python
# Copyright 2019 DeepMind Technologies Limited
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""MCTS example."""

import collections
import random
import sys

from absl import app
from absl import flags
import numpy as np

from open_spiel.python.algorithms import mcts
from open_spiel.python.algorithms.alpha_zero import evaluator as az_evaluator
from open_spiel.python.algorithms.alpha_zero import model as az_model
from open_spiel.python.bots import gtp
from open_spiel.python.bots import human
from open_spiel.python.bots import uniform_random
import pyspiel

_KNOWN_PLAYERS = [
    # A generic Monte Carlo Tree Search agent.
    "mcts",

    # A generic random agent.
    "random",

    # You'll be asked to provide the moves.
    "human",

    # Run an external program that speaks the Go Text Protocol.
    # Requires the gtp_path flag.
    "gtp",

    # Run an alpha_zero checkpoint with MCTS. Uses the specified UCT/sims.
    # Requires the az_path flag.
    "az"
]

flags.DEFINE_string("game", "tic_tac_toe", "Name of the game.")
flags.DEFINE_enum("player1", "mcts", _KNOWN_PLAYERS, "Who controls player 1.")
flags.DEFINE_enum("player2", "random", _KNOWN_PLAYERS, "Who controls player 2.")
flags.DEFINE_string("gtp_path", None, "Where to find a binary for gtp.")
flags.DEFINE_multi_string("gtp_cmd", [], "GTP commands to run at init.")
flags.DEFINE_string("az_path", None,
                    "Path to an alpha_zero checkpoint. Needed by an az player.")
flags.DEFINE_integer("uct_c", 2, "UCT's exploration constant.")
flags.DEFINE_integer("rollout_count", 1, "How many rollouts to do.")
flags.DEFINE_integer("max_simulations", 1000, "How many simulations to run.")
flags.DEFINE_integer("num_games", 1, "How many games to play.")
flags.DEFINE_integer("seed", None, "Seed for the random number generator.")
flags.DEFINE_bool("random_first", False, "Play the first move randomly.")
flags.DEFINE_bool("solve", True, "Whether to use MCTS-Solver.")
flags.DEFINE_bool("quiet", False, "Don't show the moves as they're played.")
flags.DEFINE_bool("verbose", False, "Show the MCTS stats of possible moves.")

FLAGS = flags.FLAGS

def _opt_print(*args, **kwargs):
  if not FLAGS.quiet:
    print(*args, **kwargs)

def _init_bot(bot_type, game, player_id):
  """Initializes a bot by type."""
  rng = np.random.RandomState(FLAGS.seed)
  if bot_type == "mcts":
    evaluator = mcts.RandomRolloutEvaluator(FLAGS.rollout_count, rng)
    return mcts.MCTSBot(
        game,
        FLAGS.uct_c,
        FLAGS.max_simulations,
        evaluator,
        random_state=rng,
        solve=FLAGS.solve,
        verbose=FLAGS.verbose)
  if bot_type == "az":
    model = az_model.Model.from_checkpoint(FLAGS.az_path)
    evaluator = az_evaluator.AlphaZeroEvaluator(game, model)
    return mcts.MCTSBot(
        game,
        FLAGS.uct_c,
        FLAGS.max_simulations,
        evaluator,
        random_state=rng,
        child_selection_fn=mcts.SearchNode.puct_value,
        solve=FLAGS.solve,
        verbose=FLAGS.verbose,
        dont_return_chance_node=True)
  if bot_type == "random":
    return uniform_random.UniformRandomBot(player_id, rng)
  if bot_type == "human":
    return human.HumanBot()
  if bot_type == "gtp":
    bot = gtp.GTPBot(game, FLAGS.gtp_path)
    for cmd in FLAGS.gtp_cmd:
      bot.gtp_cmd(cmd)
    return bot
  raise ValueError("Invalid bot type: %s" % bot_type)

def _get_action(state, action_str):
  for action in state.legal_actions():
    if action_str == state.action_to_string(state.current_player(), action):
      return action
  return None

def _play_game(game, bots, initial_actions):
  """Plays one game."""
  state = game.new_initial_state()
  _opt_print("Initial state:\n{}".format(state))

  history = []

  if FLAGS.random_first:
    assert not initial_actions
    initial_actions = [state.action_to_string(
        state.current_player(), random.choice(state.legal_actions()))]

  for action_str in initial_actions:
    action = _get_action(state, action_str)
    if action is None:
      sys.exit("Invalid action: {}".format(action_str))

    history.append(action_str)
    for bot in bots:
      bot.inform_action(state, state.current_player(), action)
    state.apply_action(action)
    _opt_print("Forced action", action_str)
    _opt_print("Next state:\n{}".format(state))

  while not state.is_terminal():
    current_player = state.current_player()
    # The state can be three different types: chance node,
    # simultaneous node, or decision node
    if state.is_chance_node():
      # Chance node: sample an outcome
      outcomes = state.chance_outcomes()
      num_actions = len(outcomes)
      _opt_print("Chance node, got " + str(num_actions) + " outcomes")
      action_list, prob_list = zip(*outcomes)
      action = np.random.choice(action_list, p=prob_list)
      action_str = state.action_to_string(current_player, action)
      _opt_print("Sampled action: ", action_str)
    elif state.is_simultaneous_node():
      raise ValueError("Game cannot have simultaneous nodes.")
    else:
      # Decision node: sample action for the single current player
      bot = bots[current_player]
      action = bot.step(state)
      action_str = state.action_to_string(current_player, action)
      _opt_print("Player {} sampled action: {}".format(current_player,
                                                       action_str))

    for i, bot in enumerate(bots):
      if i != current_player:
        bot.inform_action(state, current_player, action)
    history.append(action_str)
    state.apply_action(action)

    _opt_print("Next state:\n{}".format(state))

  # Game is now done. Print return for each player
  returns = state.returns()
  print("Returns:", " ".join(map(str, returns)), ", Game actions:",
        " ".join(history))

  for bot in bots:
    bot.restart()

  return returns, history

def main(argv):
  game = pyspiel.load_game(FLAGS.game)
  if game.num_players() > 2:
    sys.exit("This game requires more players than the example can handle.")
  bots = [
      _init_bot(FLAGS.player1, game, 0),
      _init_bot(FLAGS.player2, game, 1),
  ]
  histories = collections.defaultdict(int)
  overall_returns = [0, 0]
  overall_wins = [0, 0]
  game_num = 0
  try:
    for game_num in range(FLAGS.num_games):
      returns, history = _play_game(game, bots, argv[1:])
      histories[" ".join(history)] += 1
      for i, v in enumerate(returns):
        overall_returns[i] += v
        if v > 0:
          overall_wins[i] += 1
  except (KeyboardInterrupt, EOFError):
    game_num -= 1
    print("Caught a KeyboardInterrupt, stopping early.")
  print("Number of games played:", game_num + 1)
  print("Number of distinct games played:", len(histories))
  print("Players:", FLAGS.player1, FLAGS.player2)
  print("Overall wins", overall_wins)
  print("Overall returns", overall_returns)

if __name__ == "__main__":
  app.run(main)
 

bot=mcts.MCTSBot()就是根据参数创建一个mcts的机器人

action = bot.step(state)就是机器人根据当前的state获得一个操作
state.apply_action(action)然后state再根据操作更新状态

输出

下面是一个让一个蒙特卡洛树搜索机器人和一个随机机器人进行一百场对局的输出。可以看到最终的结果是mcts净胜95场

yaml
Initial state:
...
...
...
Player 0 sampled action: x(1,1)
Next state:
...
.x.
...
省略中间部分内容
Player 0 sampled action: x(0,1)
Next state:
xxx
ox.
oo.
Returns: 1.0 -1.0 , Game actions: x(1,1) o(2,0) x(0,2) o(1,0) x(0,0) o(2,1) x(0,1)
Number of games played: 100
Number of distinct games played: 91
Players: mcts random
Overall wins [95, 0]
Overall returns [95.0, -95.0]

部分关键逻辑解析

state的概率事件和clone相关逻辑解析

搜索类算法

下面是async_mcts.py的中随机评估器（RandomRolloutEvaluator）的部分代码

该函数的返回值是prior, value。

  • prior：动作的概率列表，每个元素是 (动作, 概率) 的元组
  • values：状态的价值估计。

如果现在是chance节点（is_chance_node为True），就会直接采用state的动作概率数组chance_outcomes。如果不是概率节点，则会对所有的可执行的操作，给予相同的概率

之后会复制当前的状态，作为working_state，然后在这个working_state下随机进行操作，直到对局结束。将结束时的得分作为当前的预估值。

python
  def prior_and_value(
      self, state: pyspiel.State
  ) -> tuple[list[tuple[int, float]], np.ndarray]:
    """Returns evaluation on given state."""
    # prior
    if state.is_chance_node():
      prior = state.chance_outcomes()#直接采用state给出的动作概率
    else:
      legal_actions = state.legal_actions(state.current_player())
      prior = [(action, 1.0 / len(legal_actions)) for action in legal_actions]#每个动作都给一个相同的概率
    # value
    working_state = state.clone()
    #按照随机选择的方式，一直执行到状态结束，获取当前状态价值的评估值
    while not working_state.is_terminal():
      if working_state.is_chance_node():
        outcomes = working_state.chance_outcomes()
        action_list, prob_list = zip(*outcomes)
        action = self._random_state.choice(action_list, p=prob_list)
      else:
        action = self._random_state.choice(working_state.legal_actions())
      working_state.apply_action(action)
    value = np.array(working_state.returns())
    return prior, value

RL_Environment

RL_Environment在step的时候，会执行一个对额外事件的采样_sample_external_events

python
def step(self, actions):
    assert len(actions) == self.num_actions_per_step, (
        "Invalid number of actions! Expected {}".format(
            self.num_actions_per_step))
    if self._should_reset:
      return self.reset()

    if self._enable_legality_check:
      self._check_legality(actions)

    if self.is_turn_based:
      self._state.apply_action(actions[0])
    else:
      self._state.apply_actions(actions)
    self._sample_external_events()

    return self.get_time_step()

采样的逻辑里，如果遇到的是chance_node，就会用当前的随机事件采样器进行采样

python
 def _sample_external_events(self):
    """Sample chance events until we get to a decision node."""
    while self._state.is_chance_node() or (self._state.current_player()
                                           == pyspiel.PlayerId.MEAN_FIELD):
      if self._state.is_chance_node():
        outcome = self._chance_event_sampler(self._state)
        self._state.apply_action(outcome)
      if self._state.current_player() == pyspiel.PlayerId.MEAN_FIELD:
        dist_to_register = self._state.distribution_support()
        dist = [
            self._mfg_distribution.value_str(str_state, default_value=0.0)
            for str_state in dist_to_register
        ]
        self._state.update_distribution(dist)
       
       

默认的概率事件的采样器ChanceEventSampler，会直接使用state.chance_outcomes的动作概率列表，随机进行操作

python
class ChanceEventSampler(object):
  """Default sampler for external chance events."""

  def __init__(self, seed=None):
    self.seed(seed)

  def seed(self, seed=None):
    self._rng = np.random.RandomState(seed)

  def __call__(self, state):
    """Sample a chance event in the given state."""
    actions, probs = zip(*state.chance_outcomes())
    return self._rng.choice(actions, p=probs)
 

因此在默认的RL环境下，不太需要关心change_node和chance_outcomes

自定义游戏的流程

如果想要直接使用open spiel中的一些智能体和算法，需要按照特定的规则自定义游戏，相关文档可以查看下方页面中的Add a game部分的内容

这里我们仅介绍添加新游戏最简单、最快捷的方法。理想情况下，你首先要了解通用的应用程序编程接口（API）（请参阅 open_spiel/spiel.h）。这些指南主要针对用 C++ 开发的游戏；对于用 Python 开发的游戏，过程类似，特殊注意事项会在步骤中注明。

1. 选择模板游戏

在 open_spiel/games/（或 open_spiel/python/games/）中选择一个游戏作为模板进行复制。推荐的模板游戏如下：

  • 对于无随机事件的完全信息游戏，可选择井字棋（Tic - Tac - Toe）和突破棋（Breakthrough）。
  • 对于有随机事件的完全信息游戏，可选择双陆棋（Backgammon）或猪游戏（Pig）。
  • 对于同时行动的游戏，可选择古夫棋（Goofspiel）和推角力棋（Oshi - Zumo）。
  • 对于不完全信息游戏，可选择勒杜克扑克（Leduc poker）和吹牛骰子（Liar’s dice）。

在接下来的步骤中，我们以井字棋为例。

2. 复制头文件和源文件

将 tic_tac_toe.h、tic_tac_toe.cc 和 tic_tac_toe_test.cc 复制为 new_game.h、new_game.cc 和 new_game_test.cc（如果是 Python 游戏，则复制 tic_tac_toe.py 和 tic_tac_toe_test.py）。

3. 配置 CMake

  • 使用 C++ 开发时：
  • 将新游戏的源文件添加到 open_spiel/games/CMakeLists.txt 中。
  • 将新游戏的测试目标添加到 open_spiel/games/CMakeLists.txt 中。
  • 使用 Python 开发时：
  • 将测试添加到 open_spiel/python/CMakeLists.txt 中。
  • 在 open_spiel/python/games/__init__.py 中导入新游戏。

4. 更新 C++/Python 样板代码

  • 在 new_game.h 中，重命名文件顶部和底部的头文件保护宏。
  • 在新文件中，将最内层的命名空间从 tic_tac_toe 重命名为 new_game。
  • 在新文件中，将 TicTacToeGame 和 TicTacToeState 重命名为 NewGameGame 和 NewGameState。
  • 在 new_game.cc 文件顶部，将短名称改为 new_game，并包含新游戏的头文件。

5. 更新 Python 集成测试

将新游戏的短名称添加到 open_spiel/python/tests/pyspiel_test.py 中的预期游戏列表里。

6. 验证复制的游戏

此时，你应该得到了一个换了名称的井字棋副本。它应该能够编译，测试也应该能够运行。你可以通过重新编译并运行示例程序 build/examples/example --game=new_game 来验证。注意：Python 游戏不能使用此示例程序运行，需使用 open_spiel/python/examples/example.py 代替。

7. 实现新游戏逻辑

现在，修改 NewGameGame 和 NewGameState 中函数的实现，以体现你新游戏的逻辑。大多数 API 函数从你所复制的游戏中应该很容易理解。如果不清楚，每个被重写的 API 函数在 open_spiel/spiel.h 的基类中都有完整的文档说明。

8. 测试新游戏

在构建游戏的过程中，若要测试游戏功能，你可以使用 open_spiel/tests/console_play_test.h 中的 ConsolePlayTest 进行交互式测试。至少，测试应该包括一些随机模拟测试（可参考其他游戏的测试示例）。注意：Python 游戏不能使用 ConsolePlayTest 进行测试，但 C++ 和 Python 游戏都可以使用 open_spiel/python/examples/mcts_example 在控制台与人类玩家一起进行测试。

9. 代码检查

使用代码检查工具检查你的代码，使其符合 Google 的风格指南。

  • 对于 C++ 代码，使用 cpplint。
  • 对于 Python 代码，使用 Google 风格指南中的 pylintrc 配置 pylint 进行检查，也可以使用 YAPF 对 Python 代码进行格式化。

10. 重新编译和测试

完成上述操作后，重新编译并再次运行测试，确保所有测试都能通过（包括你新游戏的测试！）。

11. 添加游戏流程文件以捕捉回归问题

运行 ./open_spiel/scripts/generate_new_playthrough.sh new_game 生成一个随机游戏流程文件，集成测试将使用该文件来防止出现回归问题。open_spiel/integration_tests/playthrough_test.py 会自动加载这些游戏流程文件，并将其与新生成的游戏流程进行比较。

如果你进行了影响游戏流程的更改，请运行 ./scripts/regenerate_playthroughs.sh 来更新这些文件。

支持的游戏和算法

支持的游戏

https://github.com/google-deepmind/open_spiel/blob/master/docs/games.md

支持的算法

https://github.com/google-deepmind/open_spiel/blob/master/docs/algorithms.md

OpenSpiel的核心接口

下面是openSpiel的一些核心接口以及其对应的描述

https://github.com/google-deepmind/open_spiel/blob/master/docs/api_reference.md

核心方法(Core）方法

方法

Python

C++

描述

deserialize_game_and_state(serialized_data: string)

Python

C++

从序列化的对象数据中重构出一个元组，包含（游戏对象，游戏状态对象）。

load_game(game_string: str)

Python

C++

根据指定的游戏字符串返回一个游戏对象。

load_game(game_string: str, parameters: Dict[str, Any])

Python

C++

根据指定的游戏字符串和参数值返回一个游戏对象。

registered_names()

Python

C++

返回库中所有游戏的短名称列表。

serialize_game_and_state(game: pyspiel.Game, state: pyspiel.State)

Python

C++

返回游戏状态以及创建该状态的游戏的字符串表示形式。

状态（State）方法

方法

Python

C++

描述

action_to_string(player: int, action: int)

Python

C++

返回指定玩家的动作的字符串表示形式。

apply_action(action: int)

Python

C++

将指定的动作应用到当前游戏状态。

apply_actions(actions: List[int])

Python

C++

将指定的联合动作（每个玩家的动作）应用到当前游戏状态。

chance_outcomes()

Python

C++

返回一个由（动作，概率）元组组成的列表，代表随机事件的结果分布。

current_player()

Python

C++

返回当前行动玩家的玩家 ID。

history()

Python

C++

返回从游戏开始以来所有玩家采取的动作序列。

information_state_string()

Python

C++

返回代表当前玩家信息状态的字符串。

information_state_string(player: int)

Python

C++

返回代表指定玩家信息状态的字符串。

information_state_tensor()

Python

C++

返回一个浮点数列表，代表当前玩家的信息状态。

information_state_tensor(player: int)

Python

C++

返回一个浮点数列表，代表指定玩家的信息状态。

is_chance_node()

Python

C++

如果当前状态是随机事件节点，则返回 True，否则返回 False。

is_simultaneous_node()

Python

C++

如果当前状态是同时行动玩家节点，则返回 True，否则返回 False。

is_terminal()

Python

C++

如果当前状态是终止状态（游戏已结束），则返回 True，否则返回 False。

legal_actions()

Python

C++

返回当前玩家的合法动作列表。

legal_actions(player: int)

Python

C++

返回指定玩家的合法动作列表。

observation_string()

Python

C++

返回代表当前玩家观察信息的字符串。

observation_string(player: int)

Python

C++

返回代表指定玩家观察信息的字符串。

observation_tensor()

Python

C++

返回一个浮点数列表，代表当前玩家的观察信息。

observation_tensor(player: int)

Python

C++

返回一个浮点数列表，代表指定玩家的观察信息。

returns()

Python

C++

返回回报列表（从游戏开始累计的奖励）：每个玩家对应一个值。

rewards()

Python

C++

返回中间奖励列表（自玩家上次行动以来获得的奖励）：每个玩家对应一个值。

serialize()

Python

C++

返回游戏状态的字符串表示形式，可用于从游戏中重构该状态。

游戏（Game）方法

方法

Python

C++

描述

action_to_string(player: int, action: int)

Python

C++

返回指定玩家动作的（与状态无关的）字符串表示形式。

deserialize_state(serialized_data: str)

Python

C++

从序列化的状态字符串中重构游戏状态。

information_state_tensor_shape()

Python

C++

信息状态张量应被视为的形状。

information_state_tensor_size()

Python

C++

状态的信息状态张量函数返回的列表大小（值的数量）。

max_chance_outcomes()

Python

C++

游戏中随机事件节点的不同随机结果的最大数量。

max_game_length()

Python

C++

任何一局游戏的最大长度（以游戏树中访问的决策节点数量衡量）。

max_utility()

Python

C++

游戏在任何一次游玩（回合）中可达到的最大效用（回报）。

min_utility()

Python

C++

游戏在任何一次游玩（回合）中可达到的最小效用（回报）。

new_initial_state()

Python

C++

返回游戏的一个新的初始状态（注意：可能是一个随机事件节点）。

num_distinct_actions()

Python

C++

返回游戏中（与状态无关的）不同动作的数量。

observation_tensor_shape()

Python

C++

观察张量应被视为的形状。

observation_tensor_size()

Python

C++

状态的观察张量函数返回的列表大小（值的数量）。

创作不易，如果觉得这篇文章对你有所帮助，可以动动小手，点个赞哈，ღ( ´･ᴗ･` )比心

# 机器学习 # python # 深度学习 # 人工智能
Logo
魔乐社区

魔乐社区（Modelers.cn) 是一个中立、公益的人工智能社区，提供人工智能工具、模型、数据的托管、展示与应用协同服务，为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作，由全产业链共同建设、共同运营、共同享有，推动国产AI生态繁荣发展。

更多推荐

cover

【计算机视觉】Pixel逐像素分类&Mask掩码分类理解摘要

avatar 魔乐社区

计算机视觉（opencv）实战三十二——CascadeClassifier 人脸微笑检测（摄像头）

本文从原理到实现，详细介绍了基于 OpenCV Haar 分类器的人脸与微笑检测：讲解了 Haar 特征和级联检测原理。对代码逐行拆解并解释参数含义。画出完整流程图，帮助理解执行过程。给出了常见问题和优化建议，甚至扩展到深度学习方法。这种方法简单、轻量、实时性好，非常适合入门和小型应用项目。但如果需要更高准确率和更强鲁棒性，建议使用深度学习检测器替代 Haar 分类器。

avatar 魔乐社区
cover

小杰机器学习（seven）——贝叶斯分类

avatar 魔乐社区