Environment

Environment Base Class

class econsimulacra.envs.base.Environment(config, logger=None)[source]

Bases: Generic[ObsT]

Environment class.

The environment in EconSimulacra contains a grid space where agents are placed and can move around, a social network where agents can follow each other and interact, a set of items that agents can trade with each other, and services that support LLM-based agents in performing various actions and making decisions. The main implemented methods in this class include:

.reset(self): reset the environment to the initial state.
.get_observations(self, agent_id): get the observation for the given agent id.
.step(self, all_actions_dic): execute one step of the environment with the given actions from agents.

Parameters:

config (dict[str, Any])
logger (Logger | None)

get_service(service_type)[source]

Return the first service instance matching the given type.

Parameters:: service_type (Type[Any])
Return type:: Any | None

get_time_translator()[source]

Get the TimeTranslator service provider from the environment’s service dictionary, if it exists.

Return type:: TimeTranslator | None

get_memory_handler()[source]

Get the MemoryHandler service provider from the environment’s service dictionary, if it exists.

Return type:: MemoryHandler | None

get_persona_builder()[source]

Get the PersonaBuilder service provider from the environment’s service dictionary, if it exists.

Return type:: PersonaBuilder | None

get_time()[source]

Get the current time in the environment.

Note

If a TimeTranslator service provider is available, use it to convert the internal time step to a datetime string; otherwise, return the internal time step as an integer.

Return type:: int | str

get_time_step()[source]

Get the current time step in the environment as an integer.

Return type:: int

get_timedelta()[source]

Get the time delta for each step in the environment.

Note

If a TimeTranslator service provider is available, use it to get the time delta in a suitable format (e.g., “1 day”, “2 hours”, etc.); otherwise, return the default time delta of 1 (which can be interpreted as 1 step).

Return type:: int | str

register_classes(class_list)[source]

Register classes to the environment for dynamic instantiation from config.

Parameters:: class_list (list[Type]) – a list of classes to be registered. These classes can then be referred to by their class names in the environment config for dynamic instantiation.
Return type:: None

Note

Use this method to register your custom classes.

reset(seed)[source]

Reset the environment to the initial state.

Parameters:: seed (Optional[int]) – random seed for environment initialization. If None, no specific seed is set.
Return type:: None

Note

This method - resets the random seed, - generates the grid space and social network according to the config, - generates service providers, agents, and items according to the config,

remember_log(log)[source]

Call MemoryHandler.update(log) to update the memory with the given log, if MemoryHandler is available in the environment services.

Parameters:: log (Log) – the log to be remembered in memory.
Return type:: None

Note

Called when any kind of log (e.g., AgentGenerationLog, MoveLog, ConsumptionLog, OrderLog, ProposalLog, etc.) is generated in the environment.

get_persona(agent_id)[source]

Get the persona details for the given agent ID, if PersonaBuilder service provider is available in the environment.

Parameters:

agent_id (int) – the ID of the agent to get persona details for.

Returns:

the persona details dictionary for the given agent ID,: or None if PersonaBuilder service provider is not available in the environment.

Return type:

persona_dic (dict[str, Any], optional)

get_total_amount(item_name)[source]

Get the total amount of the specified item in the environment.

Parameters:: item_name (str) – the name of the item to get the total amount for.
Return type:: float | int

step(all_actions_dic)[source]

Execute one step of the environment.

Parameters:: all_actions_dic (dict[int, dict[str, Any]]) – a dictionary mapping agent IDs to their respective action dictionaries for this step.
Return type:: None

Note

See also: econsimulacra.envs.base.Environment.apply_action_to_env for the expected format of each agent’s action dictionary.

apply_action_to_env(agent_id, action_dic)[source]

Apply the action of a single agent to the environment.

Parameters:

agent_id (int) – the ID of the agent whose action is to be applied.
action_dic (dict[str, Any]) – the action dictionary of the agent for this step.

Return type:

None

Note

This method processes the action dictionary of a single agent by

checking the validity of each part of the action,
executing the valid part of the action

action_dic example:

{
    "move": tuple[int, ...] | str,
    "consumptions": [
        {"item_name": str, "item_amount": float | int}, ...
    ],
    "orders": [
        {"counterparty_id": int, "counterparty_name": str,
         "item_name": str, "item_amount": float | int, "ttl": int}, ...
    ],
    "proposals": [
        {"responder_agent_id": int, "responder_agent_name": str,
         "give_item_name": str, "give_item_amount": float | int,
         "get_item_name": str, "get_item_amount": float | int, "ttl": int}, ...
    ],
    "reactions": [
        {"kind": "order", "id": int, "accept_amount": float | int},
        {"kind": "proposal", "id": int, "accept": bool}, ...
    ],
    "set_prices": [{"item_name": str, "price": float}, ...],
    "tweet": str,
    "follow": int,
    "unfollow": int,
}

See also: econsimulacra.llm_services.constant.DEFAULT_ACTION_JSON_SCHEMA for the expected format of the action dictionary used to validate generated actions from LLM-based agents.

evaluate_agent_state(agent_id)[source]

Evaluate the state of the agent. Generate agent evaluation log.

Parameters:: agent_id (int) – agent id of the agent to evaluate.
Return type:: None

calculate_relative_wealth(agent_id)[source]

Calculate the relative wealth of the agent compared to other agents.

Parameters:

agent_id (int) – agent id of the agent to calculate the relative wealth.

Returns:

the calculated relative wealth.: Returns None if the agent is not a household agent.

Return type:

relative_wealth (float, optional)

Note

See also: econsimulacra.logs.StateEvaluationLog

calculate_buying_power(agent_id)[source]

Calculate the buying power of the agent based on its inventory and the item prices.

Parameters:: agent_id (int) – agent id of the agent to calculate the buying power.
Returns:: the calculated buying power.
Return type:: buying_power (float)

get_observations(agent_id)[source]

Get the observations for the agent with the given agent_id.

Parameters:: agent_id (int) – agent id of the agent to get the observations for.
Returns:: the observations for the agent.
Return type:: ObsT

Note

The observations are provided by the registered observation providers based on the agent’s request. There are three types of observations:

General observations provided to all agents, such as time and self position.
Additional observations provided only to agents with is_rich_info_allowed=True, such as item_name2price.
Additional observations provided only to co-located agents, such as others_inventory.

The agent can request the observations by specifying the keys of the desired observations. If the agent requests “all”, all available observations will be provided.

Order

class econsimulacra.envs.order.Order(agent_id, counterparty_id, item_name, item_amount, price, order_id, ttl=None)[source]

Bases: object

Order class.

In EconSimulacra, agents can submit an order by specifying the request in an dictionary format. Then, the environment will convert the dictionary into an Order object here and process it.

See also: econsimulacra.envs.base.Environment._add_new_orders_and_proposals

Parameters:

agent_id (int)
counterparty_id (int)
item_name (str)
item_amount (float | int)
price (float | None)
order_id (int)
ttl (int | None)

react(amount)[source]

React to the order by accepting a certain amount of the item.

Parameters:: amount (float | int) – The amount of the item accepted by the counterparty. Must be less than or equal to the remaining item amount in the order.
Return type:: None

Note

When the order is reacted by the counterparty, the environment with call this method to update the accepted amount. The accepted amount will be executed in the concurrent step. See also: econsimulacra.envs.base.Environment._process_reactions

execute()[source]

Execute the order by reducing the item amount by the accepted amount.

Note

This should be called in the concurrent step after the reactions are processed. See also: econsimulacra.envs.base.Environment._process_orders_and_proposals

Return type:: None

update_time()[source]

Update the time to live for the order.

Note

This should be called in each simulation step to decrement the time to live. See also: econsimulacra.envs.base.Environment._update_time

Return type:: None

is_fulfilled()[source]

Check if the order is fulfilled, which means either the item amount is fully accepted or the order has expired.

Note

This should be called to determine if the order can be removed from the simulation. See also: econsimulacra.envs.base.Environment._remove_expired_orders_and_proposals

Return type:: bool

is_expired()[source]

Check if the order is expired, which means the time to live has reached zero or below.

Note

This should be called to determine if the order has expired and should be removed from the simulation. See also: econsimulacra.envs.base.Environment._remove_expired_orders_and_proposals

Return type:: bool

class econsimulacra.envs.order.SwapProposal(proposer_agent_id, responder_agent_id, give_item_name, give_item_amount, get_item_name, get_item_amount, proposal_id, ttl=None)[source]

Bases: object

SwapProposal class.

In EconSimulacra, agents can also submit a swap proposal by specifying the request in an dictionary format. Then, the environment will convert the dictionary into a SwapProposal object here and process it.

See also: econsimulacra.envs.base.Environment._add_new_orders_and_proposals

Parameters:

proposer_agent_id (int)
responder_agent_id (int)
give_item_name (str)
give_item_amount (float | int)
get_item_name (str)
get_item_amount (float | int)
proposal_id (int)
ttl (int | None)

react(accept)[source]

React to the swap proposal by accepting or rejecting it.

Parameters:: accept (bool) – Whether the responder agent accepts the swap proposal.
Return type:: None

Note

Different from Order, SwapProposal can only be accepted or rejected as a whole, and there is no partial acceptance. When the swap proposal is reacted by the responder agent, the environment will call this method to update the acceptance status. The accepted swap proposal will be executed in the concurrent step, while the rejected swap proposal will be removed in the next step. See also: econsimulacra.envs.base.Environment._process_reactions

update_time()[source]

Update the time to live for the swap proposal.

Note

This should be called in each simulation step to decrement the time to live. See also: econsimulacra.envs.base.Environment._update_time

Return type:: None

is_fulfilled()[source]

Check if the swap proposal is fulfilled, which means either it is accepted or it has expired.

Returns:: whether the swap proposal is fulfilled.
Return type:: bool

is_expired()[source]

Check if the swap proposal is expired, which means the time to live has reached zero or below.

Note

This should be called to determine if the swap proposal has expired and should be removed from the simulation. See also: econsimulacra.envs.base.Environment._remove_expired_orders_and_proposals

Return type:: bool

Grid Space

class econsimulacra.envs.space.GridSpace(config)[source]

Bases: object

Grid Space class.

The grid space represents the spatial environment where agents are located. Each agent can be placed at a specific position in the grid, and multiple agents can occupy the same position. The GridSpace provides methods to manage agent placements and movements.

Parameters:: config (dict[str, Any])

get_space_size()[source]

Get the shape of the grid space.

Returns:: The dimensions of the grid space.
Return type:: tuple[int, …]

get_pos(agent_id)[source]

Get the position of the agent with the given ID.

Parameters:: agent_id (int) – The ID of the agent to get the position for.
Returns:: The position of the agent.
Return type:: tuple[int, …]

get_agents(pos)[source]

Get the set of agent IDs located at the given position.

Parameters:: pos (tuple[int, ...]) – The position to check.
Returns:: The set of agent IDs located at the given position.
Return type:: set[int]

place_agent(agent_id, pos)[source]

Place an agent with the given ID at the specified position in the grid space.

Parameters:

agent_id (int) – The ID of the agent to place.
pos (tuple[int, ...]) – The position to place the agent at.

Return type:

None

Note

See also: econsimulacra.envs.base.Environment._assign_agent_to_space

get_colocated_agents(agent_id)[source]

Get the set of agent IDs located at the same position as the given agent.

Parameters:: agent_id (int) – The ID of the agent to check.
Returns:: The set of agent IDs located at the same position as the given agent.
Return type:: set[int]

remove_agent(agent_id)[source]

Remove the agent with the given ID from the grid space.

Parameters:: agent_id (int) – The ID of the agent to remove.
Return type:: None

move_agent(agent_id, new_pos)[source]

Move the agent with the given ID to a new position in the grid space.

Parameters:

agent_id (int) – The ID of the agent to move.
new_pos (tuple[int, ...]) – The new position to move the agent to.

Return type:

None

Note

See also: econsimulacra.envs.base.Environment._move

move_many_agents(agent_id2new_pos)[source]

Move multiple agents to new positions in the grid space.

Parameters:: agent_id2new_pos (dict[int, tuple[int, ...]]) – A dictionary mapping agent IDs to their new positions.
Return type:: None

Observation Providers

class econsimulacra.envs.obs_providers.ObsProvider(env)[source]

Bases: ABC

Observation provider class (abstract class).

Each ObsProvider class must implement the get_obs method, which returns the observation for a given agent_id.

ObsProvider is used to build the observation space of the environment, and to provide observations to agents at each step. You can register your custom ObsProviders to the environment by adding them to obs_providers through either _build_observation_registry or _build_observation4allowed_agents_registry, depending on who can observe the information provided by the ObsProvider.

Example

>>> class MyCustomObsProvider(ObsProvider):
>>>     def get_obs(self, agent_id: int) -> Any:
>>>         # Return the custom observation for the given agent_id
>>>         return ...
>>>
>>> class MyCustomEnv(Environment):
>>>     def _build_observation_registry(self) -> dict[str, ObsProvider]:
>>>         obs_providers = super()._build_observation_registry()
>>>         obs_providers["my_custom_obs"] = MyCustomObsProvider(env=self)
>>>         return obs_providers

See also

econsimulacra.envs.base.Environment.get_observations(agent_id: int)

Parameters:: env (Environment)

abstractmethod get_obs(agent_id)[source]

Get the observation for the given agent_id.

Parameters:: agent_id (int) – The ID of the agent for which to get the observation.
Returns:: The observation for the given agent_id.
Return type:: Any

class econsimulacra.envs.obs_providers.ObsProviderFromCoLocatedAgents(env, co_located_agents)[source]

Bases: ABC

Observation provider class from co-located agents (abstract class).

This class is for fetching the information provided by co-located agents in the same grid cell. Each ObsProviderFromCoLocatedAgents class must implement the get_obs method, which returns the observation for a given agent_id.

Parameters:

env (Environment)
co_located_agents (set[int])

abstractmethod get_obs(agent_id)[source]

Get the observation for the given agent_id.

Parameters:: agent_id (int) – The ID of the agent for which to get the observation.
Returns:: The observation for the given agent_id.
Return type:: Any

class econsimulacra.envs.obs_providers.TimeProvider(env)[source]

Environment

Environment Base Class

Order

Grid Space

Observation Providers

Time Translator