| """ |
| 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""" |
| |
| 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 save_replay(self): |
| raise NotImplementedError |
|
|
| def get_simulation_info(self): |
| raise NotImplementedError |
|
|
