update directory structure
This commit is contained in:
@@ -0,0 +1,7 @@
|
||||
from .alphago import *
|
||||
from .base import *
|
||||
#from .pg import *
|
||||
from .predict import *
|
||||
#from .naive import *
|
||||
#from .naive_fast import *
|
||||
from .termination import *
|
||||
@@ -0,0 +1,161 @@
|
||||
import numpy as np
|
||||
from tugo.agents.base import Agent
|
||||
from game_logic.goboard_fast import Move
|
||||
|
||||
|
||||
# tag::init_alphago_node[]
|
||||
class AlphaGoNode:
|
||||
def __init__(self, parent=None, probability=1.0):
|
||||
self.parent = parent # <1>
|
||||
self.children = {} # <1>
|
||||
|
||||
self.visit_count = 0
|
||||
self.q_value = 0
|
||||
self.prior_value = probability # <2>
|
||||
self.u_value = probability # <3>
|
||||
# <1> Tree nodes have one parent and potentially many children.
|
||||
# <2> A node is initialized with a prior probability.
|
||||
# <3> The utility function will be updated during search.
|
||||
# end::init_alphago_node[]
|
||||
|
||||
# tag::select_node[]
|
||||
def select_child(self):
|
||||
return max(self.children.items(),
|
||||
key=lambda child: child[1].q_value + \
|
||||
child[1].u_value)
|
||||
# end::select_node[]
|
||||
|
||||
# tag::expand_children[]
|
||||
def expand_children(self, moves, probabilities):
|
||||
for move, prob in zip(moves, probabilities):
|
||||
if move not in self.children:
|
||||
self.children[move] = AlphaGoNode(parent=self, probability=prob)
|
||||
# end::expand_children[]
|
||||
|
||||
# tag::update_values[]
|
||||
def update_values(self, leaf_value):
|
||||
if self.parent is not None:
|
||||
self.parent.update_values(leaf_value) # <1>
|
||||
|
||||
self.visit_count += 1 # <2>
|
||||
|
||||
self.q_value += leaf_value / self.visit_count # <3>
|
||||
|
||||
if self.parent is not None:
|
||||
c_u = 5
|
||||
self.u_value = c_u * np.sqrt(self.parent.visit_count) \
|
||||
* self.prior_value / (1 + self.visit_count) # <4>
|
||||
|
||||
# <1> We update parents first to ensure we traverse the tree top to bottom.
|
||||
# <2> Increment the visit count for this node.
|
||||
# <3> Add the specified leaf value to the Q-value, normalized by visit count.
|
||||
# <4> Update utility with current visit counts.
|
||||
# end::update_values[]
|
||||
|
||||
|
||||
# tag::alphago_mcts_init[]
|
||||
class AlphaGoMCTS(Agent):
|
||||
# def __init__(self, policy_agent, fast_policy_agent, value_agent,
|
||||
# lambda_value=0.5, num_simulations=1000,
|
||||
# depth=50, rollout_limit=100):
|
||||
def __init__(self, policy_agent, fast_policy_agent, value_agent,
|
||||
lambda_value=0.5, num_simulations=100,
|
||||
depth=10, rollout_limit=10):
|
||||
self.policy = policy_agent
|
||||
self.rollout_policy = fast_policy_agent
|
||||
self.value = value_agent
|
||||
|
||||
self.lambda_value = lambda_value
|
||||
self.num_simulations = num_simulations
|
||||
self.depth = depth
|
||||
self.rollout_limit = rollout_limit
|
||||
self.root = AlphaGoNode()
|
||||
# end::alphago_mcts_init[]
|
||||
|
||||
# tag::alphago_mcts_rollout[]
|
||||
def select_move(self, game_state):
|
||||
for simulation in range(self.num_simulations): # <1>
|
||||
current_state = game_state
|
||||
node = self.root
|
||||
for depth in range(self.depth): # <2>
|
||||
if not node.children: # <3>
|
||||
if current_state.is_over():
|
||||
break
|
||||
moves, probabilities = self.policy_probabilities(current_state) # <4>
|
||||
node.expand_children(moves, probabilities) # <4>
|
||||
|
||||
move, node = node.select_child() # <5>
|
||||
current_state = current_state.apply_move(move) # <5>
|
||||
|
||||
value = self.value.predict(current_state) # <6>
|
||||
rollout = self.policy_rollout(current_state) # <6>
|
||||
|
||||
weighted_value = (1 - self.lambda_value) * value + \
|
||||
self.lambda_value * rollout # <7>
|
||||
|
||||
node.update_values(weighted_value) # <8>
|
||||
# <1> From current state play out a number of simulations
|
||||
# <2> Play moves until the specified depth is reached.
|
||||
# <3> If the current node doesn't have any children...
|
||||
# <4> ... expand them with probabilities from the strong policy.
|
||||
# <5> If there are children, we can select one and play the corresponding move.
|
||||
# <6> Compute output of value network and a rollout by the fast policy.
|
||||
# <7> Determine the combined value function.
|
||||
# <8> Update values for this node in the backup phase
|
||||
# end::alphago_mcts_rollout[]
|
||||
|
||||
# tag::alphago_mcts_selection[]
|
||||
move = max(self.root.children, key=lambda move: # <1>
|
||||
self.root.children.get(move).visit_count) # <1>
|
||||
|
||||
self.root = AlphaGoNode()
|
||||
if move in self.root.children: # <2>
|
||||
self.root = self.root.children[move]
|
||||
self.root.parent = None
|
||||
|
||||
return move
|
||||
# <1> Pick most visited child of the root as next move.
|
||||
# <2> If the picked move is a child, set new root to this child node.
|
||||
# end::alphago_mcts_selection[]
|
||||
|
||||
# tag::alphago_policy_probs[]
|
||||
def policy_probabilities(self, game_state):
|
||||
encoder = self.policy._encoder
|
||||
outputs = self.policy.predict(game_state)
|
||||
legal_moves = game_state.legal_moves()
|
||||
if not legal_moves:
|
||||
return [], []
|
||||
encoded_points = [encoder.encode_point(move.point) for move in legal_moves if move.point]
|
||||
legal_outputs = outputs[encoded_points]
|
||||
normalized_outputs = legal_outputs / np.sum(legal_outputs)
|
||||
return legal_moves, normalized_outputs
|
||||
# end::alphago_policy_probs[]
|
||||
|
||||
# tag::alphago_policy_rollout[]
|
||||
def policy_rollout(self, game_state):
|
||||
for step in range(self.rollout_limit):
|
||||
if game_state.is_over():
|
||||
break
|
||||
move_probabilities = self.rollout_policy.predict(game_state)
|
||||
encoder = self.rollout_policy.encoder
|
||||
for idx in np.argsort(move_probabilities)[::-1]:
|
||||
max_point = encoder.decode_point_index(idx)
|
||||
greedy_move = Move(max_point)
|
||||
if greedy_move in game_state.legal_moves():
|
||||
game_state = game_state.apply_move(greedy_move)
|
||||
break
|
||||
|
||||
next_player = game_state.next_player
|
||||
winner = game_state.winner()
|
||||
|
||||
if winner is not None:
|
||||
return 1 if winner == next_player else -1
|
||||
else:
|
||||
return 0
|
||||
# end::alphago_policy_rollout[]
|
||||
|
||||
|
||||
def serialize(self, h5file):
|
||||
raise IOError("AlphaGoMCTS agent can\'t be serialized" +
|
||||
"consider serializing the three underlying" +
|
||||
"neural networks instad.")
|
||||
@@ -0,0 +1,10 @@
|
||||
|
||||
class Agent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def select_move(self, game_state):
|
||||
raise NotImplementedError()
|
||||
|
||||
def diagnostics(self):
|
||||
return {}
|
||||
@@ -0,0 +1,38 @@
|
||||
from game_logic.gotypes import Point
|
||||
|
||||
|
||||
# tag::eye[]
|
||||
def is_point_an_eye(board, point, color):
|
||||
if board.get(point) is not None: # <1>
|
||||
return False
|
||||
for neighbor in point.neighbors(): # <2>
|
||||
if board.is_on_grid(neighbor):
|
||||
neighbor_color = board.get(neighbor)
|
||||
if neighbor_color != color:
|
||||
return False
|
||||
|
||||
friendly_corners = 0 # <3>
|
||||
off_board_corners = 0
|
||||
corners = [
|
||||
Point(point.row - 1, point.col - 1),
|
||||
Point(point.row - 1, point.col + 1),
|
||||
Point(point.row + 1, point.col - 1),
|
||||
Point(point.row + 1, point.col + 1),
|
||||
]
|
||||
for corner in corners:
|
||||
if board.is_on_grid(corner):
|
||||
corner_color = board.get(corner)
|
||||
if corner_color == color:
|
||||
friendly_corners += 1
|
||||
else:
|
||||
off_board_corners += 1
|
||||
if off_board_corners > 0:
|
||||
return off_board_corners + friendly_corners == 4 # <4>
|
||||
return friendly_corners >= 3 # <5>
|
||||
|
||||
# <1> An eye is an empty point.
|
||||
# <2> All adjacent points must contain friendly stones.
|
||||
# <3> We must control 3 out of 4 corners if the point is in the middle of the board; on the edge we must control all corners.
|
||||
# <4> Point is on the edge or corner.
|
||||
# <5> Point is in the middle.
|
||||
# end::eye[]
|
||||
@@ -0,0 +1,33 @@
|
||||
from game_logic.gotypes import Point
|
||||
|
||||
|
||||
def is_point_an_eye(board, point, color):
|
||||
if board.get(point) is not None:
|
||||
return False
|
||||
# All adjacent points must contain friendly stones.
|
||||
for neighbor in board.neighbors(point):
|
||||
neighbor_color = board.get(neighbor)
|
||||
if neighbor_color != color:
|
||||
return False
|
||||
# We must control 3 out of 4 corners if the point is in the middle
|
||||
# of the board; on the edge we must control all corners.
|
||||
friendly_corners = 0
|
||||
off_board_corners = 0
|
||||
corners = [
|
||||
Point(point.row - 1, point.col - 1),
|
||||
Point(point.row - 1, point.col + 1),
|
||||
Point(point.row + 1, point.col - 1),
|
||||
Point(point.row + 1, point.col + 1),
|
||||
]
|
||||
for corner in corners:
|
||||
if board.is_on_grid(corner):
|
||||
corner_color = board.get(corner)
|
||||
if corner_color == color:
|
||||
friendly_corners += 1
|
||||
else:
|
||||
off_board_corners += 1
|
||||
if off_board_corners > 0:
|
||||
# Point is on the edge or corner.
|
||||
return off_board_corners + friendly_corners == 4
|
||||
# Point is in the middle.
|
||||
return friendly_corners >= 3
|
||||
@@ -0,0 +1,22 @@
|
||||
import random
|
||||
from tugo.agents.base import Agent
|
||||
from tugo.agents.helpers import is_point_an_eye
|
||||
from tugo.goboard_slow import Move
|
||||
from game_logic.gotypes import Point
|
||||
|
||||
|
||||
class RandomBot(Agent):
|
||||
def select_move(self, game_state):
|
||||
"""Choose a random valid move that preserves our own eyes."""
|
||||
candidates = []
|
||||
for r in range(1, game_state.board.num_rows + 1):
|
||||
for c in range(1, game_state.board.num_cols + 1):
|
||||
candidate = Point(row=r, col=c)
|
||||
if game_state.is_valid_move(Move.play(candidate)) and \
|
||||
not is_point_an_eye(game_state.board,
|
||||
candidate,
|
||||
game_state.next_player):
|
||||
candidates.append(candidate)
|
||||
if not candidates:
|
||||
return Move.pass_turn()
|
||||
return Move.play(random.choice(candidates))
|
||||
@@ -0,0 +1,40 @@
|
||||
import numpy as np
|
||||
|
||||
from tugo.agents.base import Agent
|
||||
from tugo.agents.helpers_fast import is_point_an_eye
|
||||
from game_logic.goboard import Move
|
||||
from game_logic.gotypes import Point
|
||||
|
||||
|
||||
|
||||
|
||||
class FastRandomBot(Agent):
|
||||
def __init__(self):
|
||||
Agent.__init__(self)
|
||||
self.dim = None
|
||||
self.point_cache = []
|
||||
|
||||
def _update_cache(self, dim):
|
||||
self.dim = dim
|
||||
rows, cols = dim
|
||||
self.point_cache = []
|
||||
for r in range(1, rows + 1):
|
||||
for c in range(1, cols + 1):
|
||||
self.point_cache.append(Point(row=r, col=c))
|
||||
|
||||
def select_move(self, game_state):
|
||||
"""Choose a random valid move that preserves our own eyes."""
|
||||
dim = (game_state.board.num_rows, game_state.board.num_cols)
|
||||
if dim != self.dim:
|
||||
self._update_cache(dim)
|
||||
|
||||
idx = np.arange(len(self.point_cache))
|
||||
np.random.shuffle(idx)
|
||||
for i in idx:
|
||||
p = self.point_cache[i]
|
||||
if game_state.is_valid_move(Move.play(p)) and \
|
||||
not is_point_an_eye(game_state.board,
|
||||
p,
|
||||
game_state.next_player):
|
||||
return Move.play(p)
|
||||
return Move.pass_turn()
|
||||
@@ -0,0 +1,91 @@
|
||||
"""Policy gradient learning."""
|
||||
import numpy as np
|
||||
import torch
|
||||
import torch.nn.functional as F
|
||||
from torch.optim import SGD
|
||||
|
||||
from tugo.agents.base import Agent
|
||||
from tugo.agents.helpers import is_point_an_eye
|
||||
from game_logic import goboard
|
||||
|
||||
|
||||
def policy_gradient_loss(y_true, y_pred):
|
||||
clip_pred = torch.clamp(y_pred, 1e-10, 1 - 1e-10)
|
||||
loss = -1 * y_true * torch.log(clip_pred)
|
||||
return torch.mean(torch.sum(loss, dim=1))
|
||||
|
||||
|
||||
class PolicyAgent(Agent):
|
||||
"""An agent that uses a deep policy network to select moves."""
|
||||
def __init__(self, model, encoder):
|
||||
super().__init__()
|
||||
self._model = model
|
||||
self._encoder = encoder
|
||||
self._collector = None
|
||||
self._temperature = 0.0
|
||||
|
||||
def predict(self, game_state):
|
||||
encoded_state = self._encoder.encode(game_state)
|
||||
input_tensor = torch.tensor([encoded_state], dtype=torch.float32).to('cuda')
|
||||
with torch.no_grad():
|
||||
output_tensor = self._model(input_tensor)
|
||||
return output_tensor.cpu().numpy()[0]
|
||||
|
||||
def set_temperature(self, temperature):
|
||||
self._temperature = temperature
|
||||
|
||||
def set_collector(self, collector):
|
||||
self._collector = collector
|
||||
|
||||
def select_move(self, game_state):
|
||||
num_moves = self._encoder.board_width * self._encoder.board_height
|
||||
|
||||
move_probs = self.predict(game_state)
|
||||
|
||||
move_probs = move_probs ** 3
|
||||
eps = 1e-6
|
||||
move_probs = np.clip(move_probs, eps, 1 - eps)
|
||||
move_probs = move_probs / np.sum(move_probs)
|
||||
|
||||
candidates = np.arange(num_moves)
|
||||
ranked_moves = np.random.choice(candidates, num_moves, replace=False, p=move_probs)
|
||||
|
||||
for point_idx in ranked_moves:
|
||||
point = self._encoder.decode_point_index(point_idx)
|
||||
if game_state.is_valid_move(goboard.Move.play(point)) and \
|
||||
not is_point_an_eye(game_state.board, point, game_state.next_player):
|
||||
if self._collector is not None:
|
||||
self._collector.record_decision(state=board_tensor, action=point_idx)
|
||||
return goboard.Move.play(point)
|
||||
|
||||
return goboard.Move.pass_turn()
|
||||
|
||||
def train(self, experience, lr=1e-7, clipnorm=1.0, batch_size=512):
|
||||
opt = SGD(self._model.parameters(), lr=lr)
|
||||
n = experience.states.shape[0]
|
||||
num_moves = self._encoder.board_width * self._encoder.board_height
|
||||
y = torch.zeros((n, num_moves))
|
||||
for i in range(n):
|
||||
action = experience.actions[i]
|
||||
reward = experience.rewards[i]
|
||||
y[i][action] = reward
|
||||
|
||||
for epoch in range(1):
|
||||
permutation = torch.randperm(n)
|
||||
for i in range(0, n, batch_size):
|
||||
indices = permutation[i:i+batch_size]
|
||||
batch_x, batch_y = experience.states[indices], y[indices]
|
||||
opt.zero_grad()
|
||||
outputs = self._model(batch_x)
|
||||
loss = F.cross_entropy(outputs, batch_y)
|
||||
loss.backward()
|
||||
opt.step()
|
||||
|
||||
def save(self, path):
|
||||
torch.save(self._model.state_dict(), path)
|
||||
|
||||
@classmethod
|
||||
def load(cls, path, encoder):
|
||||
model = AlphaGoModel(encoder.get_input_shape(), is_policy_net=True).to('cuda')
|
||||
model.load_state_dict(torch.load(path))
|
||||
return cls(model, encoder)
|
||||
@@ -0,0 +1,48 @@
|
||||
from tugo.agents.base import Agent
|
||||
from tugo.agents.helpers import is_point_an_eye
|
||||
from game_logic import goboard
|
||||
from tugo.models import AlphaGoModel
|
||||
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
class DeepLearningAgent(Agent):
|
||||
def __init__(self, model, encoder):
|
||||
super().__init__()
|
||||
self.model = model
|
||||
self.encoder = encoder
|
||||
|
||||
def predict(self, game_state):
|
||||
encoded_state = self.encoder.encode(game_state)
|
||||
input_tensor = torch.tensor([encoded_state], dtype=torch.float32).to('cuda')
|
||||
with torch.no_grad():
|
||||
output_tensor = self.model(input_tensor)
|
||||
return output_tensor.cpu().numpy()[0]
|
||||
|
||||
def select_move(self, game_state):
|
||||
num_moves = self.encoder.board_width * self.encoder.board_height
|
||||
move_probs = self.predict(game_state)
|
||||
|
||||
move_probs = move_probs ** 3
|
||||
eps = 1e-6
|
||||
move_probs = np.clip(move_probs, eps, 1 - eps)
|
||||
move_probs = move_probs / np.sum(move_probs)
|
||||
|
||||
candidates = np.arange(num_moves)
|
||||
ranked_moves = np.random.choice(
|
||||
candidates, num_moves, replace=False, p=move_probs)
|
||||
for point_idx in ranked_moves:
|
||||
point = self.encoder.decode_point_index(point_idx)
|
||||
if game_state.is_valid_move(goboard.Move.play(point)) and \
|
||||
not is_point_an_eye(game_state.board, point, game_state.next_player):
|
||||
return goboard.Move.play(point)
|
||||
return goboard.Move.pass_turn()
|
||||
|
||||
def save(self, path):
|
||||
torch.save(self.model.state_dict(), path)
|
||||
|
||||
@classmethod
|
||||
def load(cls, path, encoder):
|
||||
model = AlphaGoModel(encoder.get_input_shape(), is_policy_net=True).to('cuda')
|
||||
model.load_state_dict(torch.load(path))
|
||||
return cls(model, encoder)
|
||||
@@ -0,0 +1,82 @@
|
||||
# tag::termination_imports[]
|
||||
from game_logic import goboard, scoring
|
||||
from tugo.agents.base import Agent
|
||||
|
||||
|
||||
# end::termination_imports[]
|
||||
|
||||
|
||||
# tag::termination_strategy[]
|
||||
class TerminationStrategy:
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def should_pass(self, game_state):
|
||||
return False
|
||||
|
||||
def should_resign(self, game_state):
|
||||
return False
|
||||
# end::termination_strategy[]
|
||||
|
||||
|
||||
# tag::opponent_passes[]
|
||||
class PassWhenOpponentPasses(TerminationStrategy):
|
||||
|
||||
def should_pass(self, game_state):
|
||||
if game_state.last_move is not None:
|
||||
return True if game_state.last_move.is_pass else False
|
||||
# end::opponent_passes[]
|
||||
|
||||
|
||||
# tag::resign_margin[]
|
||||
class ResignLargeMargin(TerminationStrategy):
|
||||
|
||||
def __init__(self, own_color, cut_off_move, margin):
|
||||
TerminationStrategy.__init__(self)
|
||||
self.own_color = own_color
|
||||
self.cut_off_move = cut_off_move
|
||||
self.margin = margin
|
||||
|
||||
self.moves_played = 0
|
||||
|
||||
def should_pass(self, game_state):
|
||||
return False
|
||||
|
||||
def should_resign(self, game_state):
|
||||
self.moves_played += 1
|
||||
if self.moves_played:
|
||||
game_result = scoring.compute_game_result(self)
|
||||
if game_result.winner != self.own_color and game_result.winning_margin >= self.margin:
|
||||
return True
|
||||
return False
|
||||
# end::resign_margin[]
|
||||
|
||||
|
||||
# tag::termination_agent[]
|
||||
class TerminationAgent(Agent):
|
||||
|
||||
def __init__(self, agent, strategy=None):
|
||||
Agent.__init__(self)
|
||||
self.agent = agent
|
||||
self.strategy = strategy if strategy is not None \
|
||||
else TerminationStrategy()
|
||||
|
||||
def select_move(self, game_state):
|
||||
if self.strategy.should_pass(game_state):
|
||||
return goboard.Move.pass_turn()
|
||||
elif self.strategy.should_resign(game_state):
|
||||
return goboard.Move.resign()
|
||||
else:
|
||||
return self.agent.select_move(game_state)
|
||||
# end::termination_agent[]
|
||||
|
||||
|
||||
# tag::get_termination[]
|
||||
def get(termination):
|
||||
if termination == 'opponent_passes':
|
||||
return PassWhenOpponentPasses()
|
||||
else:
|
||||
raise ValueError("Unsupported termination strategy: {}"
|
||||
.format(termination))
|
||||
# end::get_termination[]
|
||||
Reference in New Issue
Block a user