pacai.agents.mdp
1import abc 2import logging 3import typing 4 5import pacai.core.agent 6import pacai.core.agentaction 7import pacai.core.gamestate 8import pacai.core.mdp 9 10DEFAULT_LEARNING_RATE: float = 0.5 11DEFAULT_DISCOUNT_RATE: float = 0.9 12DEFAULT_EXPLORATION_RATE: float = 0.3 13 14class MDPAgent(pacai.core.agent.Agent): 15 """ 16 An agent that conceptually builds on top of a Markov Decision Process (MDP). 17 The agent does not need an actual MDP, it may just model/approximate one. 18 """ 19 20 def __init__(self, 21 learning_rate: float = DEFAULT_LEARNING_RATE, 22 discount_rate: float = DEFAULT_DISCOUNT_RATE, 23 exploration_rate: float = DEFAULT_EXPLORATION_RATE, 24 mdp_state_class: typing.Type | pacai.util.reflection.Reference | str | None = pacai.core.mdp.MDPStateBoard, 25 **kwargs: typing.Any) -> None: 26 super().__init__(**kwargs) 27 28 if (mdp_state_class is None): 29 raise ValueError("ValueIterationAgent must be provided with an MDP state class.") 30 31 clean_mdp_state_class = pacai.util.reflection.resolve_and_fetch(type, mdp_state_class) 32 if (not issubclass(clean_mdp_state_class, pacai.core.mdp.MDPStatePosition)): 33 name = pacai.util.reflection.get_qualified_name(clean_mdp_state_class) 34 raise ValueError(f"Did not get a subclass of 'pacai.core.mdp.MDPStatePosition' for `mdp_state_class`, got '{name}'.") 35 36 self.mdp_state_class: type[pacai.core.mdp.MDPStatePosition] = clean_mdp_state_class 37 """ The function used to create MDP states from game states. """ 38 39 self.learning_rate: float = float(learning_rate) 40 """ 41 The learning rate (alpha). 42 43 May not be used by all MDP agents. 44 """ 45 46 self.discount_rate: float = float(discount_rate) 47 """ 48 The discount rate (gamma). 49 A "discount" is a reduction to some score (usually a past or future score). 50 1.0 is no discount, and 0.0 is a full discount. 51 52 May not be used by all MDP agents. 53 """ 54 55 self.exploration_rate: float = float(exploration_rate) 56 """ 57 The exploration rate (epsilon). 58 0.0 means that the agent will never explore, and always follow the computed policy. 59 1.0 means that the agent will always explore, and never follow the computed policy. 60 61 May not be used by all MDP agents. 62 """ 63 64 logging.debug("Created an MDP agent with learning rate %0.2f, discount rate %0.2f, and exploration rate %0.2f.", 65 self.learning_rate, self.discount_rate, self.exploration_rate) 66 67 @abc.abstractmethod 68 def get_mdp_state_value(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> float: 69 """ 70 Get the value of the given MDP state. 71 The provided game state will be what the MDP state was generated from. 72 Unknown/unrecognized states should return 0.0. 73 """ 74 75 @abc.abstractmethod 76 def get_policy_action(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 77 """ 78 Get the best action for this MDP state according to the current policy. 79 If there are no legal action, return pacai.core.action.STOP. 80 """ 81 82 @abc.abstractmethod 83 def get_qvalue(self, 84 mdp_state: pacai.core.mdp.MDPStatePosition, 85 game_state: pacai.core.gamestate.GameState, 86 action: pacai.core.action.Action, 87 ) -> float: 88 """ 89 Get the Q-value of the given MDP state and action pair. 90 Pairs that have no known Q-value should return 0.0. 91 """ 92 93 def game_start_full(self, 94 agent_index: int, 95 suggested_seed: int, 96 initial_state: pacai.core.gamestate.GameState, 97 ) -> pacai.core.agentaction.AgentAction: 98 agent_action = super().game_start_full(agent_index, suggested_seed, initial_state) 99 100 # Pass the mdp state values and Q-values back to the game 101 # so they can be displayed in the UI. 102 103 serial_mdp_state_values = [] # [(MDP state, MDP state value), ...] 104 serial_policy = [] # [(MDP state, action), ...] 105 serial_qvalues = [] # [(MDP state, action, qvalue), ...] 106 107 for row in range(initial_state.board.height): 108 for col in range(initial_state.board.width): 109 position = pacai.core.board.Position(row, col) 110 111 mdp_state = self.mdp_state_class(position = position, game_state = initial_state) 112 raw_mdp_state = mdp_state.to_dict() 113 114 mdp_state_value = self.get_mdp_state_value(mdp_state, initial_state) 115 serial_mdp_state_values.append((raw_mdp_state, mdp_state_value)) 116 117 serial_policy.append((raw_mdp_state, self.get_policy_action(mdp_state, initial_state))) 118 119 for action in pacai.core.action.CARDINAL_DIRECTIONS: 120 qvalue = self.get_qvalue(mdp_state, initial_state, action) 121 serial_qvalues.append((raw_mdp_state, action, qvalue)) 122 123 agent_action.other_info['mdp_state_values'] = serial_mdp_state_values 124 agent_action.other_info['policy'] = serial_policy 125 agent_action.other_info['qvalues'] = serial_qvalues 126 127 return agent_action
15class MDPAgent(pacai.core.agent.Agent): 16 """ 17 An agent that conceptually builds on top of a Markov Decision Process (MDP). 18 The agent does not need an actual MDP, it may just model/approximate one. 19 """ 20 21 def __init__(self, 22 learning_rate: float = DEFAULT_LEARNING_RATE, 23 discount_rate: float = DEFAULT_DISCOUNT_RATE, 24 exploration_rate: float = DEFAULT_EXPLORATION_RATE, 25 mdp_state_class: typing.Type | pacai.util.reflection.Reference | str | None = pacai.core.mdp.MDPStateBoard, 26 **kwargs: typing.Any) -> None: 27 super().__init__(**kwargs) 28 29 if (mdp_state_class is None): 30 raise ValueError("ValueIterationAgent must be provided with an MDP state class.") 31 32 clean_mdp_state_class = pacai.util.reflection.resolve_and_fetch(type, mdp_state_class) 33 if (not issubclass(clean_mdp_state_class, pacai.core.mdp.MDPStatePosition)): 34 name = pacai.util.reflection.get_qualified_name(clean_mdp_state_class) 35 raise ValueError(f"Did not get a subclass of 'pacai.core.mdp.MDPStatePosition' for `mdp_state_class`, got '{name}'.") 36 37 self.mdp_state_class: type[pacai.core.mdp.MDPStatePosition] = clean_mdp_state_class 38 """ The function used to create MDP states from game states. """ 39 40 self.learning_rate: float = float(learning_rate) 41 """ 42 The learning rate (alpha). 43 44 May not be used by all MDP agents. 45 """ 46 47 self.discount_rate: float = float(discount_rate) 48 """ 49 The discount rate (gamma). 50 A "discount" is a reduction to some score (usually a past or future score). 51 1.0 is no discount, and 0.0 is a full discount. 52 53 May not be used by all MDP agents. 54 """ 55 56 self.exploration_rate: float = float(exploration_rate) 57 """ 58 The exploration rate (epsilon). 59 0.0 means that the agent will never explore, and always follow the computed policy. 60 1.0 means that the agent will always explore, and never follow the computed policy. 61 62 May not be used by all MDP agents. 63 """ 64 65 logging.debug("Created an MDP agent with learning rate %0.2f, discount rate %0.2f, and exploration rate %0.2f.", 66 self.learning_rate, self.discount_rate, self.exploration_rate) 67 68 @abc.abstractmethod 69 def get_mdp_state_value(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> float: 70 """ 71 Get the value of the given MDP state. 72 The provided game state will be what the MDP state was generated from. 73 Unknown/unrecognized states should return 0.0. 74 """ 75 76 @abc.abstractmethod 77 def get_policy_action(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 78 """ 79 Get the best action for this MDP state according to the current policy. 80 If there are no legal action, return pacai.core.action.STOP. 81 """ 82 83 @abc.abstractmethod 84 def get_qvalue(self, 85 mdp_state: pacai.core.mdp.MDPStatePosition, 86 game_state: pacai.core.gamestate.GameState, 87 action: pacai.core.action.Action, 88 ) -> float: 89 """ 90 Get the Q-value of the given MDP state and action pair. 91 Pairs that have no known Q-value should return 0.0. 92 """ 93 94 def game_start_full(self, 95 agent_index: int, 96 suggested_seed: int, 97 initial_state: pacai.core.gamestate.GameState, 98 ) -> pacai.core.agentaction.AgentAction: 99 agent_action = super().game_start_full(agent_index, suggested_seed, initial_state) 100 101 # Pass the mdp state values and Q-values back to the game 102 # so they can be displayed in the UI. 103 104 serial_mdp_state_values = [] # [(MDP state, MDP state value), ...] 105 serial_policy = [] # [(MDP state, action), ...] 106 serial_qvalues = [] # [(MDP state, action, qvalue), ...] 107 108 for row in range(initial_state.board.height): 109 for col in range(initial_state.board.width): 110 position = pacai.core.board.Position(row, col) 111 112 mdp_state = self.mdp_state_class(position = position, game_state = initial_state) 113 raw_mdp_state = mdp_state.to_dict() 114 115 mdp_state_value = self.get_mdp_state_value(mdp_state, initial_state) 116 serial_mdp_state_values.append((raw_mdp_state, mdp_state_value)) 117 118 serial_policy.append((raw_mdp_state, self.get_policy_action(mdp_state, initial_state))) 119 120 for action in pacai.core.action.CARDINAL_DIRECTIONS: 121 qvalue = self.get_qvalue(mdp_state, initial_state, action) 122 serial_qvalues.append((raw_mdp_state, action, qvalue)) 123 124 agent_action.other_info['mdp_state_values'] = serial_mdp_state_values 125 agent_action.other_info['policy'] = serial_policy 126 agent_action.other_info['qvalues'] = serial_qvalues 127 128 return agent_action
An agent that conceptually builds on top of a Markov Decision Process (MDP). The agent does not need an actual MDP, it may just model/approximate one.
The function used to create MDP states from game states.
The discount rate (gamma). A "discount" is a reduction to some score (usually a past or future score). 1.0 is no discount, and 0.0 is a full discount.
May not be used by all MDP agents.
The exploration rate (epsilon). 0.0 means that the agent will never explore, and always follow the computed policy. 1.0 means that the agent will always explore, and never follow the computed policy.
May not be used by all MDP agents.
68 @abc.abstractmethod 69 def get_mdp_state_value(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> float: 70 """ 71 Get the value of the given MDP state. 72 The provided game state will be what the MDP state was generated from. 73 Unknown/unrecognized states should return 0.0. 74 """
Get the value of the given MDP state. The provided game state will be what the MDP state was generated from. Unknown/unrecognized states should return 0.0.
76 @abc.abstractmethod 77 def get_policy_action(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 78 """ 79 Get the best action for this MDP state according to the current policy. 80 If there are no legal action, return pacai.core.action.STOP. 81 """
Get the best action for this MDP state according to the current policy. If there are no legal action, return pacai.core.action.STOP.
83 @abc.abstractmethod 84 def get_qvalue(self, 85 mdp_state: pacai.core.mdp.MDPStatePosition, 86 game_state: pacai.core.gamestate.GameState, 87 action: pacai.core.action.Action, 88 ) -> float: 89 """ 90 Get the Q-value of the given MDP state and action pair. 91 Pairs that have no known Q-value should return 0.0. 92 """
Get the Q-value of the given MDP state and action pair. Pairs that have no known Q-value should return 0.0.
94 def game_start_full(self, 95 agent_index: int, 96 suggested_seed: int, 97 initial_state: pacai.core.gamestate.GameState, 98 ) -> pacai.core.agentaction.AgentAction: 99 agent_action = super().game_start_full(agent_index, suggested_seed, initial_state) 100 101 # Pass the mdp state values and Q-values back to the game 102 # so they can be displayed in the UI. 103 104 serial_mdp_state_values = [] # [(MDP state, MDP state value), ...] 105 serial_policy = [] # [(MDP state, action), ...] 106 serial_qvalues = [] # [(MDP state, action, qvalue), ...] 107 108 for row in range(initial_state.board.height): 109 for col in range(initial_state.board.width): 110 position = pacai.core.board.Position(row, col) 111 112 mdp_state = self.mdp_state_class(position = position, game_state = initial_state) 113 raw_mdp_state = mdp_state.to_dict() 114 115 mdp_state_value = self.get_mdp_state_value(mdp_state, initial_state) 116 serial_mdp_state_values.append((raw_mdp_state, mdp_state_value)) 117 118 serial_policy.append((raw_mdp_state, self.get_policy_action(mdp_state, initial_state))) 119 120 for action in pacai.core.action.CARDINAL_DIRECTIONS: 121 qvalue = self.get_qvalue(mdp_state, initial_state, action) 122 serial_qvalues.append((raw_mdp_state, action, qvalue)) 123 124 agent_action.other_info['mdp_state_values'] = serial_mdp_state_values 125 agent_action.other_info['policy'] = serial_policy 126 agent_action.other_info['qvalues'] = serial_qvalues 127 128 return agent_action
Notify this agent that the game is about to start. The provided agent index is the game's index/id for this agent. The state represents the initial state of the game. Any precomputation for this game should be done in this method. Calls to this method may be subject to a timeout.