pacai.student.learning
1import logging 2import typing 3 4import edq.util.json 5 6import pacai.agents.mdp 7import pacai.agents.userinput 8import pacai.core.agent 9import pacai.core.agentaction 10import pacai.core.features 11import pacai.core.gamestate 12import pacai.core.mdp 13 14DEFAULT_VALUE_ITERATIONS: int = 100 15 16class ValueIterationAgent(pacai.agents.mdp.MDPAgent): 17 """ 18 An agent that performs value iteration on an MDP to compute values for each MDP state. 19 The agent then computes a policy based on those values and always follows the policy. 20 """ 21 22 def __init__(self, 23 mdp: pacai.core.mdp.MarkovDecisionProcess | None = None, 24 iterations: int = DEFAULT_VALUE_ITERATIONS, 25 **kwargs: typing.Any) -> None: 26 super().__init__(**kwargs) 27 28 if (mdp is None): 29 raise ValueError("ValueIterationAgent must be provided with an MDP.") 30 31 self.mdp = mdp 32 """ The MDP this agent will use. """ 33 34 self.mdp_state_values: dict[pacai.core.mdp.MDPStatePosition, float] = {} 35 """ The value for each MDP state. """ 36 37 self.iterations: int = int(iterations) 38 """ The number of value iterations to perform. """ 39 40 def game_start(self, initial_state: pacai.core.gamestate.GameState) -> None: 41 # Initialize the MDP. 42 self.mdp.game_start(initial_state) 43 44 # Perform value iteration and set self.mdp_state_values. 45 self.do_value_iteration(initial_state) 46 47 super().game_start(initial_state) 48 49 def do_value_iteration(self, game_state: pacai.core.gamestate.GameState) -> None: 50 """ 51 Perform value iteration (for self.iteration) iterations 52 and set self.mdp_state_values. 53 """ 54 55 # *** Your Code Here *** 56 57 def get_action(self, state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 58 mdp_state = self.mdp_state_class(position = state.get_agent_position(), game_state = state) 59 return self.get_policy_action(mdp_state, state) 60 61 def get_mdp_state_value(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> float: 62 return self.mdp_state_values.get(mdp_state, 0.0) 63 64 def get_qvalue(self, 65 mdp_state: pacai.core.mdp.MDPStatePosition, 66 game_state: pacai.core.gamestate.GameState, 67 action: pacai.core.action.Action, 68 ) -> float: 69 # *** Your Code Here *** 70 return 0.0 71 72 def get_policy_action(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 73 # *** Your Code Here *** 74 return pacai.core.action.STOP 75 76class QLearningAgent(pacai.agents.mdp.MDPAgent): 77 """ 78 An abstract value estimation agent that learns by estimating Q-values from experience. 79 """ 80 81 def __init__(self, 82 training_info: dict[str, typing.Any] | None = None, 83 **kwargs: typing.Any) -> None: 84 super().__init__(**kwargs) 85 86 self.last_state: pacai.core.gamestate.GameState | None = None 87 """ 88 The last seen state. 89 The difference between the last and current states will define the delta reward. 90 """ 91 92 self.total_rewards: float = 0.0 93 """ 94 The total rewards this agent as accumulated. 95 Purely for display/logging purposes. 96 """ 97 98 self.qvalues: dict[tuple[pacai.core.mdp.MDPStatePosition, pacai.core.action.Action], float] = {} 99 """ The Q-values for this agent. """ 100 101 # Load any training information. 102 if (training_info is not None): 103 self.unpack_training_info(training_info) 104 105 def pack_training_info(self) -> dict[str, typing.Any]: 106 """ 107 Return a dict that contains all the training information to pass onto future iterations of this agent. 108 This method will be used when this agent's epoch is complete to pass information to the next epoch's agent, 109 which will use unpack_training_info() to load this data. 110 The dict should be JSON serializable. 111 """ 112 113 return { 114 # [(raw mdp state, action, action), ...] 115 'qvalues': [(mdp_state.to_dict(), action, qvalue) for ((mdp_state, action), qvalue) in self.qvalues.items()] 116 } 117 118 def unpack_training_info(self, data: dict[str, typing.Any]) -> None: 119 """ 120 Load training information from the given dict, 121 which should have been created by pack_training_info(). 122 """ 123 124 for (raw_mdp_state, raw_action, qvalue) in data.get('qvalues', []): 125 mdp_state = self.mdp_state_class.from_dict(raw_mdp_state) 126 action = pacai.core.action.Action(raw_action) 127 self.qvalues[(mdp_state, action)] = qvalue 128 129 def game_start(self, initial_state: pacai.core.gamestate.GameState) -> None: 130 self.last_state = initial_state 131 132 def game_complete_full(self, 133 final_state: pacai.core.gamestate.GameState, 134 ) -> pacai.core.agentaction.AgentAction: 135 if (self.training): 136 logging.debug("Completed training epoch %d.", self.training_epoch) 137 138 self.update(final_state) 139 140 average_reward = 0.0 141 num_actions = len(final_state.get_agent_actions(self.agent_index)) 142 143 if (num_actions > 0): 144 average_reward = self.total_rewards / num_actions 145 146 logging.debug("Made %d moves for a total of %0.2f rewards (average: %0.2f).", 147 num_actions, self.total_rewards, average_reward) 148 149 # Store the training information for the next epoch's agent. 150 agent_action = super().game_complete_full(final_state) 151 agent_action.training_info['training_info'] = self.pack_training_info() 152 return agent_action 153 154 def update(self, new_state: pacai.core.gamestate.GameState) -> None: 155 """ 156 Update the agent based on the difference between the old state and new state. 157 """ 158 159 # Get the most recent action. 160 last_action = new_state.get_last_agent_action(self.agent_index) 161 if (last_action is None): 162 # No action has been taken yet, don't update. 163 return 164 165 # Update the last seen state. 166 old_state = self.last_state 167 self.last_state = new_state.copy() 168 169 if (old_state is None): 170 # We don't have an old state to compare against yet. 171 return 172 173 # Compute and store the score delta. 174 score_delta = new_state.score - old_state.score 175 self.total_rewards += score_delta 176 177 # Do not update if we are not training. 178 if (not self.training): 179 return 180 181 old_position = old_state.get_agent_position() 182 if (old_position is None): 183 # The agent was not on the board the last turn. Did they respawn? 184 return 185 186 new_position = self.last_positions[-1] 187 188 self.update_qvalue(score_delta, last_action, 189 old_state, new_state, 190 old_position, new_position) 191 192 def get_action(self, state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 193 # Update the agent by learning from the environment. 194 # This code should not change and anways be the first thing done in this method. 195 self.update(state) 196 197 # *** Your Code Here *** 198 return pacai.core.action.STOP 199 200 def get_mdp_state_value(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> float: 201 # *** Your Code Here *** 202 return 0.0 203 204 def get_policy_action(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 205 # *** Your Code Here *** 206 return pacai.core.action.STOP 207 208 def get_qvalue(self, 209 mdp_state: pacai.core.mdp.MDPStatePosition, 210 game_state: pacai.core.gamestate.GameState, 211 action: pacai.core.action.Action, 212 ) -> float: 213 return self.qvalues.get((mdp_state, action), 0.0) 214 215 def update_qvalue(self, 216 reward: float, 217 action: pacai.core.action.Action, 218 old_game_state: pacai.core.gamestate.GameState, new_game_state: pacai.core.gamestate.GameState, 219 old_position: pacai.core.board.Position | None, new_position: pacai.core.board.Position | None, 220 ) -> None: 221 """ 222 Update the Q-value for the specified transition. 223 This method will only be called when we are sure we want to update the Q-value 224 (i.e., we are training and all the required information is available). 225 """ 226 227 # *** Your Code Here *** 228 229class QLearningUserInputAgent(QLearningAgent): 230 """ 231 A Q-learning agent that learns from user actions. 232 In practical terms, this is not a very useful agent (we learn Q-values, but don't do anything with them). 233 However, this agent can be useful if you want to see how specific actions affect the learned Q-values. 234 """ 235 236 def __init__(self, **kwargs: typing.Any) -> None: 237 super().__init__(**kwargs) 238 239 kwargs['remember_last_action'] = False 240 self._user_input_agent: pacai.core.agent.Agent = pacai.agents.userinput.UserInputAgent(**kwargs) 241 """ Keep an agent that already knows how to work with user inputs. """ 242 243 def game_start_full(self, 244 agent_index: int, 245 suggested_seed: int, 246 initial_state: pacai.core.gamestate.GameState, 247 ) -> pacai.core.agentaction.AgentAction: 248 self._user_input_agent.game_start_full(agent_index, suggested_seed, initial_state) 249 return super().game_start_full(agent_index, suggested_seed, initial_state) 250 251 def game_complete_full(self, 252 final_state: pacai.core.gamestate.GameState, 253 ) -> pacai.core.agentaction.AgentAction: 254 self._user_input_agent.game_complete_full(final_state) 255 return super().game_complete_full(final_state) 256 257 def get_action_full(self, 258 state: pacai.core.gamestate.GameState, 259 user_inputs: list[pacai.core.action.Action], 260 ) -> pacai.core.agentaction.AgentAction: 261 # Get the action from the parent Q-learner. 262 agent_action = super().get_action_full(state, user_inputs) 263 264 # Just return if the action is an EXIT. 265 if (agent_action.action == pacai.core.mdp.ACTION_EXIT): 266 return agent_action 267 268 # If we are not exiting, then just ignore the Q-learning action. 269 return self._user_input_agent.get_action_full(state, user_inputs) 270 271class ApproximateQLearningAgent(QLearningAgent): 272 """ 273 A Q-learning agent that uses features and weights as Q-values instead of explicitly remembering each state. 274 """ 275 276 def __init__(self, 277 feature_extractor_func: pacai.core.features.FeatureExtractor | pacai.util.reflection.Reference | str = 278 pacai.core.features.score_feature_extractor, 279 **kwargs: typing.Any) -> None: 280 self.weights: pacai.core.features.WeightDict = pacai.core.features.WeightDict() 281 """ The feature weights learned by this agent. """ 282 283 clean_feature_extractor_func = pacai.util.reflection.resolve_and_fetch(pacai.core.features.FeatureExtractor, feature_extractor_func) 284 self.feature_extractor_func: pacai.core.features.FeatureExtractor = clean_feature_extractor_func 285 """ The feature extractor that will be used to get features from a state. """ 286 287 # Call super after ensuring that the weights exists so the training data can be unpacked into it. 288 super().__init__(**kwargs) 289 290 def pack_training_info(self) -> dict[str, typing.Any]: 291 return { 292 'weights': self.weights, 293 } 294 295 def unpack_training_info(self, data: dict[str, typing.Any]) -> None: 296 self.weights = pacai.core.features.WeightDict(data.get('weights', {})) 297 298 def game_complete(self, final_state: pacai.core.gamestate.GameState) -> None: 299 super().game_complete(final_state) 300 301 logging.debug("Weights: %s.", edq.util.json.dumps(self.weights)) 302 303 def get_qvalue(self, 304 mdp_state: pacai.core.mdp.MDPStatePosition, 305 game_state: pacai.core.gamestate.GameState, 306 action: pacai.core.action.Action, 307 ) -> float: 308 """ 309 Instead of using pre-computed Q-values for each state, 310 this should return $ weights ⋅ features $, 311 where `⋅` is the dot product operator. 312 """ 313 314 # *** Your Code Here *** 315 return 0.0 316 317 def update_qvalue(self, 318 reward: float, 319 action: pacai.core.action.Action, 320 old_game_state: pacai.core.gamestate.GameState, new_game_state: pacai.core.gamestate.GameState, 321 old_position: pacai.core.board.Position | None, new_position: pacai.core.board.Position | None, 322 ) -> None: 323 # *** Your Code Here *** 324 pass
DEFAULT_VALUE_ITERATIONS: int =
100
17class ValueIterationAgent(pacai.agents.mdp.MDPAgent): 18 """ 19 An agent that performs value iteration on an MDP to compute values for each MDP state. 20 The agent then computes a policy based on those values and always follows the policy. 21 """ 22 23 def __init__(self, 24 mdp: pacai.core.mdp.MarkovDecisionProcess | None = None, 25 iterations: int = DEFAULT_VALUE_ITERATIONS, 26 **kwargs: typing.Any) -> None: 27 super().__init__(**kwargs) 28 29 if (mdp is None): 30 raise ValueError("ValueIterationAgent must be provided with an MDP.") 31 32 self.mdp = mdp 33 """ The MDP this agent will use. """ 34 35 self.mdp_state_values: dict[pacai.core.mdp.MDPStatePosition, float] = {} 36 """ The value for each MDP state. """ 37 38 self.iterations: int = int(iterations) 39 """ The number of value iterations to perform. """ 40 41 def game_start(self, initial_state: pacai.core.gamestate.GameState) -> None: 42 # Initialize the MDP. 43 self.mdp.game_start(initial_state) 44 45 # Perform value iteration and set self.mdp_state_values. 46 self.do_value_iteration(initial_state) 47 48 super().game_start(initial_state) 49 50 def do_value_iteration(self, game_state: pacai.core.gamestate.GameState) -> None: 51 """ 52 Perform value iteration (for self.iteration) iterations 53 and set self.mdp_state_values. 54 """ 55 56 # *** Your Code Here *** 57 58 def get_action(self, state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 59 mdp_state = self.mdp_state_class(position = state.get_agent_position(), game_state = state) 60 return self.get_policy_action(mdp_state, state) 61 62 def get_mdp_state_value(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> float: 63 return self.mdp_state_values.get(mdp_state, 0.0) 64 65 def get_qvalue(self, 66 mdp_state: pacai.core.mdp.MDPStatePosition, 67 game_state: pacai.core.gamestate.GameState, 68 action: pacai.core.action.Action, 69 ) -> float: 70 # *** Your Code Here *** 71 return 0.0 72 73 def get_policy_action(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 74 # *** Your Code Here *** 75 return pacai.core.action.STOP
An agent that performs value iteration on an MDP to compute values for each MDP state. The agent then computes a policy based on those values and always follows the policy.
ValueIterationAgent( mdp: pacai.core.mdp.MarkovDecisionProcess | None = None, iterations: int = 100, **kwargs: Any)
23 def __init__(self, 24 mdp: pacai.core.mdp.MarkovDecisionProcess | None = None, 25 iterations: int = DEFAULT_VALUE_ITERATIONS, 26 **kwargs: typing.Any) -> None: 27 super().__init__(**kwargs) 28 29 if (mdp is None): 30 raise ValueError("ValueIterationAgent must be provided with an MDP.") 31 32 self.mdp = mdp 33 """ The MDP this agent will use. """ 34 35 self.mdp_state_values: dict[pacai.core.mdp.MDPStatePosition, float] = {} 36 """ The value for each MDP state. """ 37 38 self.iterations: int = int(iterations) 39 """ The number of value iterations to perform. """
41 def game_start(self, initial_state: pacai.core.gamestate.GameState) -> None: 42 # Initialize the MDP. 43 self.mdp.game_start(initial_state) 44 45 # Perform value iteration and set self.mdp_state_values. 46 self.do_value_iteration(initial_state) 47 48 super().game_start(initial_state)
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.
50 def do_value_iteration(self, game_state: pacai.core.gamestate.GameState) -> None: 51 """ 52 Perform value iteration (for self.iteration) iterations 53 and set self.mdp_state_values. 54 """ 55 56 # *** Your Code Here ***
Perform value iteration (for self.iteration) iterations and set self.mdp_state_values.
58 def get_action(self, state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 59 mdp_state = self.mdp_state_class(position = state.get_agent_position(), game_state = state) 60 return self.get_policy_action(mdp_state, state)
Get an action for this agent given the current state of the game. This is simplified version of get_action_full(), see that method for full details.
def
get_mdp_state_value( self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> float:
62 def get_mdp_state_value(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> float: 63 return self.mdp_state_values.get(mdp_state, 0.0)
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.
def
get_qvalue( self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState, action: pacai.core.action.Action) -> float:
65 def get_qvalue(self, 66 mdp_state: pacai.core.mdp.MDPStatePosition, 67 game_state: pacai.core.gamestate.GameState, 68 action: pacai.core.action.Action, 69 ) -> float: 70 # *** Your Code Here *** 71 return 0.0
Get the Q-value of the given MDP state and action pair. Pairs that have no known Q-value should return 0.0.
def
get_policy_action( self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> pacai.core.action.Action:
73 def get_policy_action(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 74 # *** Your Code Here *** 75 return pacai.core.action.STOP
Get the best action for this MDP state according to the current policy. If there are no legal action, return pacai.core.action.STOP.
Inherited Members
77class QLearningAgent(pacai.agents.mdp.MDPAgent): 78 """ 79 An abstract value estimation agent that learns by estimating Q-values from experience. 80 """ 81 82 def __init__(self, 83 training_info: dict[str, typing.Any] | None = None, 84 **kwargs: typing.Any) -> None: 85 super().__init__(**kwargs) 86 87 self.last_state: pacai.core.gamestate.GameState | None = None 88 """ 89 The last seen state. 90 The difference between the last and current states will define the delta reward. 91 """ 92 93 self.total_rewards: float = 0.0 94 """ 95 The total rewards this agent as accumulated. 96 Purely for display/logging purposes. 97 """ 98 99 self.qvalues: dict[tuple[pacai.core.mdp.MDPStatePosition, pacai.core.action.Action], float] = {} 100 """ The Q-values for this agent. """ 101 102 # Load any training information. 103 if (training_info is not None): 104 self.unpack_training_info(training_info) 105 106 def pack_training_info(self) -> dict[str, typing.Any]: 107 """ 108 Return a dict that contains all the training information to pass onto future iterations of this agent. 109 This method will be used when this agent's epoch is complete to pass information to the next epoch's agent, 110 which will use unpack_training_info() to load this data. 111 The dict should be JSON serializable. 112 """ 113 114 return { 115 # [(raw mdp state, action, action), ...] 116 'qvalues': [(mdp_state.to_dict(), action, qvalue) for ((mdp_state, action), qvalue) in self.qvalues.items()] 117 } 118 119 def unpack_training_info(self, data: dict[str, typing.Any]) -> None: 120 """ 121 Load training information from the given dict, 122 which should have been created by pack_training_info(). 123 """ 124 125 for (raw_mdp_state, raw_action, qvalue) in data.get('qvalues', []): 126 mdp_state = self.mdp_state_class.from_dict(raw_mdp_state) 127 action = pacai.core.action.Action(raw_action) 128 self.qvalues[(mdp_state, action)] = qvalue 129 130 def game_start(self, initial_state: pacai.core.gamestate.GameState) -> None: 131 self.last_state = initial_state 132 133 def game_complete_full(self, 134 final_state: pacai.core.gamestate.GameState, 135 ) -> pacai.core.agentaction.AgentAction: 136 if (self.training): 137 logging.debug("Completed training epoch %d.", self.training_epoch) 138 139 self.update(final_state) 140 141 average_reward = 0.0 142 num_actions = len(final_state.get_agent_actions(self.agent_index)) 143 144 if (num_actions > 0): 145 average_reward = self.total_rewards / num_actions 146 147 logging.debug("Made %d moves for a total of %0.2f rewards (average: %0.2f).", 148 num_actions, self.total_rewards, average_reward) 149 150 # Store the training information for the next epoch's agent. 151 agent_action = super().game_complete_full(final_state) 152 agent_action.training_info['training_info'] = self.pack_training_info() 153 return agent_action 154 155 def update(self, new_state: pacai.core.gamestate.GameState) -> None: 156 """ 157 Update the agent based on the difference between the old state and new state. 158 """ 159 160 # Get the most recent action. 161 last_action = new_state.get_last_agent_action(self.agent_index) 162 if (last_action is None): 163 # No action has been taken yet, don't update. 164 return 165 166 # Update the last seen state. 167 old_state = self.last_state 168 self.last_state = new_state.copy() 169 170 if (old_state is None): 171 # We don't have an old state to compare against yet. 172 return 173 174 # Compute and store the score delta. 175 score_delta = new_state.score - old_state.score 176 self.total_rewards += score_delta 177 178 # Do not update if we are not training. 179 if (not self.training): 180 return 181 182 old_position = old_state.get_agent_position() 183 if (old_position is None): 184 # The agent was not on the board the last turn. Did they respawn? 185 return 186 187 new_position = self.last_positions[-1] 188 189 self.update_qvalue(score_delta, last_action, 190 old_state, new_state, 191 old_position, new_position) 192 193 def get_action(self, state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 194 # Update the agent by learning from the environment. 195 # This code should not change and anways be the first thing done in this method. 196 self.update(state) 197 198 # *** Your Code Here *** 199 return pacai.core.action.STOP 200 201 def get_mdp_state_value(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> float: 202 # *** Your Code Here *** 203 return 0.0 204 205 def get_policy_action(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 206 # *** Your Code Here *** 207 return pacai.core.action.STOP 208 209 def get_qvalue(self, 210 mdp_state: pacai.core.mdp.MDPStatePosition, 211 game_state: pacai.core.gamestate.GameState, 212 action: pacai.core.action.Action, 213 ) -> float: 214 return self.qvalues.get((mdp_state, action), 0.0) 215 216 def update_qvalue(self, 217 reward: float, 218 action: pacai.core.action.Action, 219 old_game_state: pacai.core.gamestate.GameState, new_game_state: pacai.core.gamestate.GameState, 220 old_position: pacai.core.board.Position | None, new_position: pacai.core.board.Position | None, 221 ) -> None: 222 """ 223 Update the Q-value for the specified transition. 224 This method will only be called when we are sure we want to update the Q-value 225 (i.e., we are training and all the required information is available). 226 """ 227 228 # *** Your Code Here ***
An abstract value estimation agent that learns by estimating Q-values from experience.
QLearningAgent(training_info: dict[str, typing.Any] | None = None, **kwargs: Any)
82 def __init__(self, 83 training_info: dict[str, typing.Any] | None = None, 84 **kwargs: typing.Any) -> None: 85 super().__init__(**kwargs) 86 87 self.last_state: pacai.core.gamestate.GameState | None = None 88 """ 89 The last seen state. 90 The difference between the last and current states will define the delta reward. 91 """ 92 93 self.total_rewards: float = 0.0 94 """ 95 The total rewards this agent as accumulated. 96 Purely for display/logging purposes. 97 """ 98 99 self.qvalues: dict[tuple[pacai.core.mdp.MDPStatePosition, pacai.core.action.Action], float] = {} 100 """ The Q-values for this agent. """ 101 102 # Load any training information. 103 if (training_info is not None): 104 self.unpack_training_info(training_info)
last_state: pacai.core.gamestate.GameState | None
The last seen state. The difference between the last and current states will define the delta reward.
total_rewards: float
The total rewards this agent as accumulated. Purely for display/logging purposes.
qvalues: dict[tuple[pacai.core.mdp.MDPStatePosition, pacai.core.action.Action], float]
The Q-values for this agent.
def
pack_training_info(self) -> dict[str, typing.Any]:
106 def pack_training_info(self) -> dict[str, typing.Any]: 107 """ 108 Return a dict that contains all the training information to pass onto future iterations of this agent. 109 This method will be used when this agent's epoch is complete to pass information to the next epoch's agent, 110 which will use unpack_training_info() to load this data. 111 The dict should be JSON serializable. 112 """ 113 114 return { 115 # [(raw mdp state, action, action), ...] 116 'qvalues': [(mdp_state.to_dict(), action, qvalue) for ((mdp_state, action), qvalue) in self.qvalues.items()] 117 }
Return a dict that contains all the training information to pass onto future iterations of this agent. This method will be used when this agent's epoch is complete to pass information to the next epoch's agent, which will use unpack_training_info() to load this data. The dict should be JSON serializable.
def
unpack_training_info(self, data: dict[str, typing.Any]) -> None:
119 def unpack_training_info(self, data: dict[str, typing.Any]) -> None: 120 """ 121 Load training information from the given dict, 122 which should have been created by pack_training_info(). 123 """ 124 125 for (raw_mdp_state, raw_action, qvalue) in data.get('qvalues', []): 126 mdp_state = self.mdp_state_class.from_dict(raw_mdp_state) 127 action = pacai.core.action.Action(raw_action) 128 self.qvalues[(mdp_state, action)] = qvalue
Load training information from the given dict, which should have been created by pack_training_info().
130 def game_start(self, initial_state: pacai.core.gamestate.GameState) -> None: 131 self.last_state = initial_state
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.
def
game_complete_full( self, final_state: pacai.core.gamestate.GameState) -> pacai.core.agentaction.AgentAction:
133 def game_complete_full(self, 134 final_state: pacai.core.gamestate.GameState, 135 ) -> pacai.core.agentaction.AgentAction: 136 if (self.training): 137 logging.debug("Completed training epoch %d.", self.training_epoch) 138 139 self.update(final_state) 140 141 average_reward = 0.0 142 num_actions = len(final_state.get_agent_actions(self.agent_index)) 143 144 if (num_actions > 0): 145 average_reward = self.total_rewards / num_actions 146 147 logging.debug("Made %d moves for a total of %0.2f rewards (average: %0.2f).", 148 num_actions, self.total_rewards, average_reward) 149 150 # Store the training information for the next epoch's agent. 151 agent_action = super().game_complete_full(final_state) 152 agent_action.training_info['training_info'] = self.pack_training_info() 153 return agent_action
Notify this agent that the game has concluded. Agents should use this as an opportunity to make any final calculations and close any game-related resources.
155 def update(self, new_state: pacai.core.gamestate.GameState) -> None: 156 """ 157 Update the agent based on the difference between the old state and new state. 158 """ 159 160 # Get the most recent action. 161 last_action = new_state.get_last_agent_action(self.agent_index) 162 if (last_action is None): 163 # No action has been taken yet, don't update. 164 return 165 166 # Update the last seen state. 167 old_state = self.last_state 168 self.last_state = new_state.copy() 169 170 if (old_state is None): 171 # We don't have an old state to compare against yet. 172 return 173 174 # Compute and store the score delta. 175 score_delta = new_state.score - old_state.score 176 self.total_rewards += score_delta 177 178 # Do not update if we are not training. 179 if (not self.training): 180 return 181 182 old_position = old_state.get_agent_position() 183 if (old_position is None): 184 # The agent was not on the board the last turn. Did they respawn? 185 return 186 187 new_position = self.last_positions[-1] 188 189 self.update_qvalue(score_delta, last_action, 190 old_state, new_state, 191 old_position, new_position)
Update the agent based on the difference between the old state and new state.
193 def get_action(self, state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 194 # Update the agent by learning from the environment. 195 # This code should not change and anways be the first thing done in this method. 196 self.update(state) 197 198 # *** Your Code Here *** 199 return pacai.core.action.STOP
Get an action for this agent given the current state of the game. This is simplified version of get_action_full(), see that method for full details.
def
get_mdp_state_value( self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> float:
201 def get_mdp_state_value(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> float: 202 # *** Your Code Here *** 203 return 0.0
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.
def
get_policy_action( self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> pacai.core.action.Action:
205 def get_policy_action(self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState) -> pacai.core.action.Action: 206 # *** Your Code Here *** 207 return pacai.core.action.STOP
Get the best action for this MDP state according to the current policy. If there are no legal action, return pacai.core.action.STOP.
def
get_qvalue( self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState, action: pacai.core.action.Action) -> float:
209 def get_qvalue(self, 210 mdp_state: pacai.core.mdp.MDPStatePosition, 211 game_state: pacai.core.gamestate.GameState, 212 action: pacai.core.action.Action, 213 ) -> float: 214 return self.qvalues.get((mdp_state, action), 0.0)
Get the Q-value of the given MDP state and action pair. Pairs that have no known Q-value should return 0.0.
def
update_qvalue( self, reward: float, action: pacai.core.action.Action, old_game_state: pacai.core.gamestate.GameState, new_game_state: pacai.core.gamestate.GameState, old_position: pacai.core.board.Position | None, new_position: pacai.core.board.Position | None) -> None:
216 def update_qvalue(self, 217 reward: float, 218 action: pacai.core.action.Action, 219 old_game_state: pacai.core.gamestate.GameState, new_game_state: pacai.core.gamestate.GameState, 220 old_position: pacai.core.board.Position | None, new_position: pacai.core.board.Position | None, 221 ) -> None: 222 """ 223 Update the Q-value for the specified transition. 224 This method will only be called when we are sure we want to update the Q-value 225 (i.e., we are training and all the required information is available). 226 """ 227 228 # *** Your Code Here ***
Update the Q-value for the specified transition. This method will only be called when we are sure we want to update the Q-value (i.e., we are training and all the required information is available).
230class QLearningUserInputAgent(QLearningAgent): 231 """ 232 A Q-learning agent that learns from user actions. 233 In practical terms, this is not a very useful agent (we learn Q-values, but don't do anything with them). 234 However, this agent can be useful if you want to see how specific actions affect the learned Q-values. 235 """ 236 237 def __init__(self, **kwargs: typing.Any) -> None: 238 super().__init__(**kwargs) 239 240 kwargs['remember_last_action'] = False 241 self._user_input_agent: pacai.core.agent.Agent = pacai.agents.userinput.UserInputAgent(**kwargs) 242 """ Keep an agent that already knows how to work with user inputs. """ 243 244 def game_start_full(self, 245 agent_index: int, 246 suggested_seed: int, 247 initial_state: pacai.core.gamestate.GameState, 248 ) -> pacai.core.agentaction.AgentAction: 249 self._user_input_agent.game_start_full(agent_index, suggested_seed, initial_state) 250 return super().game_start_full(agent_index, suggested_seed, initial_state) 251 252 def game_complete_full(self, 253 final_state: pacai.core.gamestate.GameState, 254 ) -> pacai.core.agentaction.AgentAction: 255 self._user_input_agent.game_complete_full(final_state) 256 return super().game_complete_full(final_state) 257 258 def get_action_full(self, 259 state: pacai.core.gamestate.GameState, 260 user_inputs: list[pacai.core.action.Action], 261 ) -> pacai.core.agentaction.AgentAction: 262 # Get the action from the parent Q-learner. 263 agent_action = super().get_action_full(state, user_inputs) 264 265 # Just return if the action is an EXIT. 266 if (agent_action.action == pacai.core.mdp.ACTION_EXIT): 267 return agent_action 268 269 # If we are not exiting, then just ignore the Q-learning action. 270 return self._user_input_agent.get_action_full(state, user_inputs)
A Q-learning agent that learns from user actions. In practical terms, this is not a very useful agent (we learn Q-values, but don't do anything with them). However, this agent can be useful if you want to see how specific actions affect the learned Q-values.
QLearningUserInputAgent(**kwargs: Any)
237 def __init__(self, **kwargs: typing.Any) -> None: 238 super().__init__(**kwargs) 239 240 kwargs['remember_last_action'] = False 241 self._user_input_agent: pacai.core.agent.Agent = pacai.agents.userinput.UserInputAgent(**kwargs) 242 """ Keep an agent that already knows how to work with user inputs. """
def
game_start_full( self, agent_index: int, suggested_seed: int, initial_state: pacai.core.gamestate.GameState) -> pacai.core.agentaction.AgentAction:
244 def game_start_full(self, 245 agent_index: int, 246 suggested_seed: int, 247 initial_state: pacai.core.gamestate.GameState, 248 ) -> pacai.core.agentaction.AgentAction: 249 self._user_input_agent.game_start_full(agent_index, suggested_seed, initial_state) 250 return super().game_start_full(agent_index, suggested_seed, initial_state)
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.
def
game_complete_full( self, final_state: pacai.core.gamestate.GameState) -> pacai.core.agentaction.AgentAction:
252 def game_complete_full(self, 253 final_state: pacai.core.gamestate.GameState, 254 ) -> pacai.core.agentaction.AgentAction: 255 self._user_input_agent.game_complete_full(final_state) 256 return super().game_complete_full(final_state)
Notify this agent that the game has concluded. Agents should use this as an opportunity to make any final calculations and close any game-related resources.
def
get_action_full( self, state: pacai.core.gamestate.GameState, user_inputs: list[pacai.core.action.Action]) -> pacai.core.agentaction.AgentAction:
258 def get_action_full(self, 259 state: pacai.core.gamestate.GameState, 260 user_inputs: list[pacai.core.action.Action], 261 ) -> pacai.core.agentaction.AgentAction: 262 # Get the action from the parent Q-learner. 263 agent_action = super().get_action_full(state, user_inputs) 264 265 # Just return if the action is an EXIT. 266 if (agent_action.action == pacai.core.mdp.ACTION_EXIT): 267 return agent_action 268 269 # If we are not exiting, then just ignore the Q-learning action. 270 return self._user_input_agent.get_action_full(state, user_inputs)
Get an action for this agent given the current state of the game. Agents may keep internal state, but the given state should be considered the source of truth. Calls to this method may be subject to a timeout (enforced by the isolator).
By default, this method just calls get_action(). Agent classes should typically just implement get_action(), and only implement this if they need additional functionality.
Inherited Members
272class ApproximateQLearningAgent(QLearningAgent): 273 """ 274 A Q-learning agent that uses features and weights as Q-values instead of explicitly remembering each state. 275 """ 276 277 def __init__(self, 278 feature_extractor_func: pacai.core.features.FeatureExtractor | pacai.util.reflection.Reference | str = 279 pacai.core.features.score_feature_extractor, 280 **kwargs: typing.Any) -> None: 281 self.weights: pacai.core.features.WeightDict = pacai.core.features.WeightDict() 282 """ The feature weights learned by this agent. """ 283 284 clean_feature_extractor_func = pacai.util.reflection.resolve_and_fetch(pacai.core.features.FeatureExtractor, feature_extractor_func) 285 self.feature_extractor_func: pacai.core.features.FeatureExtractor = clean_feature_extractor_func 286 """ The feature extractor that will be used to get features from a state. """ 287 288 # Call super after ensuring that the weights exists so the training data can be unpacked into it. 289 super().__init__(**kwargs) 290 291 def pack_training_info(self) -> dict[str, typing.Any]: 292 return { 293 'weights': self.weights, 294 } 295 296 def unpack_training_info(self, data: dict[str, typing.Any]) -> None: 297 self.weights = pacai.core.features.WeightDict(data.get('weights', {})) 298 299 def game_complete(self, final_state: pacai.core.gamestate.GameState) -> None: 300 super().game_complete(final_state) 301 302 logging.debug("Weights: %s.", edq.util.json.dumps(self.weights)) 303 304 def get_qvalue(self, 305 mdp_state: pacai.core.mdp.MDPStatePosition, 306 game_state: pacai.core.gamestate.GameState, 307 action: pacai.core.action.Action, 308 ) -> float: 309 """ 310 Instead of using pre-computed Q-values for each state, 311 this should return $ weights ⋅ features $, 312 where `⋅` is the dot product operator. 313 """ 314 315 # *** Your Code Here *** 316 return 0.0 317 318 def update_qvalue(self, 319 reward: float, 320 action: pacai.core.action.Action, 321 old_game_state: pacai.core.gamestate.GameState, new_game_state: pacai.core.gamestate.GameState, 322 old_position: pacai.core.board.Position | None, new_position: pacai.core.board.Position | None, 323 ) -> None: 324 # *** Your Code Here *** 325 pass
A Q-learning agent that uses features and weights as Q-values instead of explicitly remembering each state.
ApproximateQLearningAgent( feature_extractor_func: pacai.core.features.FeatureExtractor | pacai.util.reflection.Reference | str = <function score_feature_extractor>, **kwargs: Any)
277 def __init__(self, 278 feature_extractor_func: pacai.core.features.FeatureExtractor | pacai.util.reflection.Reference | str = 279 pacai.core.features.score_feature_extractor, 280 **kwargs: typing.Any) -> None: 281 self.weights: pacai.core.features.WeightDict = pacai.core.features.WeightDict() 282 """ The feature weights learned by this agent. """ 283 284 clean_feature_extractor_func = pacai.util.reflection.resolve_and_fetch(pacai.core.features.FeatureExtractor, feature_extractor_func) 285 self.feature_extractor_func: pacai.core.features.FeatureExtractor = clean_feature_extractor_func 286 """ The feature extractor that will be used to get features from a state. """ 287 288 # Call super after ensuring that the weights exists so the training data can be unpacked into it. 289 super().__init__(**kwargs)
feature_extractor_func: pacai.core.features.FeatureExtractor
The feature extractor that will be used to get features from a state.
def
pack_training_info(self) -> dict[str, typing.Any]:
291 def pack_training_info(self) -> dict[str, typing.Any]: 292 return { 293 'weights': self.weights, 294 }
Return a dict that contains all the training information to pass onto future iterations of this agent. This method will be used when this agent's epoch is complete to pass information to the next epoch's agent, which will use unpack_training_info() to load this data. The dict should be JSON serializable.
def
unpack_training_info(self, data: dict[str, typing.Any]) -> None:
296 def unpack_training_info(self, data: dict[str, typing.Any]) -> None: 297 self.weights = pacai.core.features.WeightDict(data.get('weights', {}))
Load training information from the given dict, which should have been created by pack_training_info().
299 def game_complete(self, final_state: pacai.core.gamestate.GameState) -> None: 300 super().game_complete(final_state) 301 302 logging.debug("Weights: %s.", edq.util.json.dumps(self.weights))
Notify this agent that the game has concluded. Agents should use this as an opportunity to make any final calculations and close any game-related resources.
def
get_qvalue( self, mdp_state: pacai.core.mdp.MDPStatePosition, game_state: pacai.core.gamestate.GameState, action: pacai.core.action.Action) -> float:
304 def get_qvalue(self, 305 mdp_state: pacai.core.mdp.MDPStatePosition, 306 game_state: pacai.core.gamestate.GameState, 307 action: pacai.core.action.Action, 308 ) -> float: 309 """ 310 Instead of using pre-computed Q-values for each state, 311 this should return $ weights ⋅ features $, 312 where `⋅` is the dot product operator. 313 """ 314 315 # *** Your Code Here *** 316 return 0.0
Instead of using pre-computed Q-values for each state,
this should return $ weights ⋅ features $,
where ⋅ is the dot product operator.
def
update_qvalue( self, reward: float, action: pacai.core.action.Action, old_game_state: pacai.core.gamestate.GameState, new_game_state: pacai.core.gamestate.GameState, old_position: pacai.core.board.Position | None, new_position: pacai.core.board.Position | None) -> None:
318 def update_qvalue(self, 319 reward: float, 320 action: pacai.core.action.Action, 321 old_game_state: pacai.core.gamestate.GameState, new_game_state: pacai.core.gamestate.GameState, 322 old_position: pacai.core.board.Position | None, new_position: pacai.core.board.Position | None, 323 ) -> None: 324 # *** Your Code Here *** 325 pass
Update the Q-value for the specified transition. This method will only be called when we are sure we want to update the Q-value (i.e., we are training and all the required information is available).