File size: 2,451 Bytes
af1ec53 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | """
A Simulation is the environment of a Markov Game.
The Simulation is not responsible for properly checking / formatting the responses of LLM's.
This is the job of the `Agent` class.
Simulations expect clean actions, and are defined similarly to `gymnasium` environments, except that they are adapted for the Multi-agent setting.
"""
from abc import ABC, abstractmethod
from typing import Any, Tuple
from numpy.random import default_rng
from mllm.markov_games.rollout_tree import SimulationStepLog
class Simulation(ABC):
@abstractmethod
def __init__(self, seed: int, *args, **kwargs):
self.seed = seed
self.rng = default_rng(self.seed)
@abstractmethod
def step(self, actions: Any) -> Tuple[bool, SimulationStepLog]:
"""
Returns terminated, info
"""
raise NotImplementedError
def get_obs(self):
"""Returns all agent observations in dict
Returns:
observations
"""
raise NotImplementedError
def get_obs_agent(self, agent_id):
"""Returns observation for agent_id"""
raise NotImplementedError
def get_obs_size(self):
"""Returns the shape of the observation"""
raise NotImplementedError
def get_state(self):
raise NotImplementedError
def get_state_size(self):
"""Returns the shape of the state"""
raise NotImplementedError
def get_avail_actions(self):
raise NotImplementedError
def get_avail_agent_actions(self, agent_id):
"""Returns the available actions for agent_id"""
raise NotImplementedError
def get_total_actions(self):
"""Returns the total number of actions an agent could ever take"""
# TODO: This is only suitable for a discrete 1 dimensional action space for each agent
raise NotImplementedError
def get_safe_copy(self):
"""
Return copy of the agent object that is decorrelated from the original object.
"""
raise NotImplementedError
def reset(self):
"""Returns initial observations and states"""
raise NotImplementedError
def render(self):
raise NotImplementedError
def close(self):
raise NotImplementedError
# def seed(self):
# raise NotImplementedError
def save_replay(self):
raise NotImplementedError
def get_simulation_info(self):
raise NotImplementedError
|