agentsociety.simulation.simulationengine

A clear version of the simulation.

Module Contents

Classes

SimulationEngine

Functions

_set_default_agent_config	Validates configuration options to ensure the user selects the correct combination.
_init_agent_class	Initialize the agent class.
evaluate_filter	Evaluate a filter string against a profile dictionary.

Data

__all__
MIN_ID
MAX_ID

API

agentsociety.simulation.simulationengine.__all__

[‘SimulationEngine’]

agentsociety.simulation.simulationengine.MIN_ID

1

agentsociety.simulation.simulationengine.MAX_ID

100000000

agentsociety.simulation.simulationengine._set_default_agent_config(self: agentsociety.configs.Config)

Validates configuration options to ensure the user selects the correct combination.

• Description:

  • If citizens contains at least one CITIZEN type agent, automatically fills empty institution agent lists with default configurations.

  • Sets default memory_config_func for citizen agents if not specified.

• Returns:

  • AgentsConfig: The validated configuration instance.

agentsociety.simulation.simulationengine._init_agent_class(agent_config: agentsociety.configs.AgentConfig, s3config: agentsociety.s3.S3Config)

Initialize the agent class.

• Args:

  • agent_config (AgentConfig): The agent configuration.

• Returns:

  • agents: A list of tuples, each containing an agent class, a memory config generator, and an index.

agentsociety.simulation.simulationengine.evaluate_filter(filter_str: str, profile: dict) → bool

Evaluate a filter string against a profile dictionary.

• Args:

  • filter_str (str): The filter string to evaluate, e.g. “${profile.age} > 0”

  • profile (dict): The profile dictionary to evaluate against

• Returns:

  • bool: True if the filter matches, False otherwise

• Note:

  • Returns False if profile is empty

  • Returns False if any key in filter_str is not in profile

class agentsociety.simulation.simulationengine.SimulationEngine(config: agentsociety.configs.Config, tenant_id: str = '')

Initialization

async _init_embedding()

Initialize embedding model with timeout.

async _init_embedding_task()

Actual embedding initialization task.

async init()

Initialize all the components

async close()

Close all the components

property name

property config

property llm

property enable_database

property database_writer

property environment

property messager

async _extract_target_agent_ids(target_agent: Optional[Union[list[int], agentsociety.configs.AgentFilterConfig]] = None) → list[int]

async gather(content: str, target_agent_ids: Optional[list[int]] = None, flatten: bool = False, keep_id: bool = False) → Union[dict[int, Any], list[Any]]

Collect specific information from agents.

• Description:

  • Asynchronously gathers specified content from targeted agents within all groups.

• Args:

  • content (str): The information to collect from the agents.

  • target_agent_ids (Optional[List[int]], optional): A list of agent IDs to target. Defaults to None, meaning all agents are targeted.

  • flatten (bool, optional): Whether to flatten the result. Defaults to False.

  • keep_id (bool, optional): Whether to keep the agent IDs in the result. Defaults to False.

• Returns:

  • Result of the gathering process as returned by each group’s gather method.

async filter(types: Optional[tuple[type[agentsociety.agent.Agent]]] = None, filter_str: Optional[str] = None) → list[int]

Filter out agents of specified types or with matching key-value pairs.

• Args:

  • types (Optional[Tuple[Type[Agent]]], optional): Types of agents to filter for. Defaults to None.

  • filter_str (Optional[str], optional): Filter string to match in agent attributes. Defaults to None.

• Raises:

  • ValueError: If neither types nor filter_str are provided.

• Returns:

  • List[int]: A list of filtered agent UUIDs.

async update_environment(key: str, value: str)

Update the environment variables for the simulation and all agent groups.

• Args:

  • key (str): The environment variable key to update.

  • value (str): The new value for the environment variable.

async update(target_agent_ids: list[int], target_key: str, content: Any, query: bool = False)

Update the memory of specified agents.

• Args:

  • target_agent_id (list[int]): The IDs of the target agents to update.

  • target_key (str): The key in the agent’s memory to update.

  • content (Any): The new content to set for the target key.

async economy_update(target_agent_id: int, target_key: str, content: Any, mode: Literal[replace, merge] = 'replace')

Update economic data for a specified agent.

• Args:

  • target_agent_id (int): The ID of the target agent whose economic data to update.

  • target_key (str): The key in the agent’s economic data to update.

  • content (Any): The new content to set for the target key.

  • mode (Literal[“replace”, “merge”], optional): Mode of updating the economic data. Defaults to “replace”.

async send_survey(survey: agentsociety.survey.models.Survey, agent_ids: list[int] = [], survey_day: Optional[int] = None, survey_t: Optional[float] = None, is_pending_survey: bool = False, pending_survey_id: Optional[int] = None) → dict[int, str]

Send a survey to specified agents.

• Args:

  • survey (Survey): The survey object to send.

  • agent_ids (List[int], optional): List of agent IDs to receive the survey. Defaults to an empty list.

  • survey_day (int, optional): The day of the survey. Defaults to None.

  • survey_t (float, optional): The time of the survey. Defaults to None.

  • is_pending_survey (bool, optional): Whether the survey is a pending survey. Defaults to False.

  • pending_survey_id (int, optional): The ID of the pending survey. Defaults to None.

• Returns:

  • dict[int, str]: A dictionary mapping agent IDs to their survey responses.

async send_interview_message(question: str, agent_ids: list[int])

Send an interview message to specified agents.

• Args:

  • question (str): The content of the message to send.

  • agent_ids (list[int]): A list of IDs for the agents to receive the message.

• Returns:

  • None

async send_intervention_message(intervention_message: str, agent_ids: list[int])

Send an intervention message to specified agents.

• Description:

  • Send an intervention message to specified agents.

• Args:

  • intervention_message (str): The content of the intervention message to send.

  • agent_ids (list[int]): A list of agent IDs to receive the intervention message.

async _save_exp_info() → None

Async save experiment info to YAML file and pgsql

async _save_global_prompt(prompt: str, day: int, t: float)

Save global prompt

async _gather_and_update_context(target_agent_ids: list[int], key: str, save_as: str)

Gather and update the context

_save_context()

async _message_dispatch()

Dispatches messages received via Message to the appropriate agents.

async _save(day: int, t: int)

Saves the current status of the agents at a given point in the simulation.

• Args:

  • day (int): The day number in the simulation time.

  • t (int): The tick or time unit in the simulation day.

async delete_agents(target_agent_ids: list[int])

Delete the specified agents.

• Args:

  • target_agent_ids (list[int]): The IDs of the agents to delete.

async next_round()

Proceed to the next round of the simulation.

async step(num_environment_ticks: int = 1) → agentsociety.simulation.type.Logs

Execute one step of the simulation where each agent performs its forward action.

• Description:

  • Checks if new agents need to be inserted based on the current day of the simulation. If so, it inserts them.

  • Executes the forward method for each agent group to advance the simulation by one step.

  • Saves the state of all agent groups after the step has been completed.

  • Optionally extracts metrics if the current step matches the interval specified for any metric extractors.

• Args:

  • num_environment_ticks (int): The number of ticks for the environment to step forward.

• Raises:

  • RuntimeError: If there is an error during the execution of the step, it logs the error and rethrows it as a RuntimeError.

• Returns:

  • Logs: The logs of the simulation.

async run_one_day(ticks_per_step: int)

Run the simulation for a day.

• Args:

  • ticks_per_step (int): The number of ticks per step.

• Description:

  • Updates the experiment status to running and sets up monitoring for the experiment’s status.

  • Runs the simulation loop until the end time, which is calculated based on the current time and the number of days to simulate.

  • After completing the simulation, updates the experiment status to finished, or to failed if an exception occurs.

• Raises:

  • RuntimeError: If there is an error during the simulation, it logs the error and updates the experiment status to failed before rethrowing the exception.

• Returns:

  • None

async run()

Run the simulation following the workflow in the config.