Environment API¶

The Environment system is the core orchestration layer of Meta Agents Research Environments simulations. It manages the simulation lifecycle, event scheduling, time management, and coordination between apps and agents. This section documents the Environment API and configuration options.

Overview ¶

The Environment acts as the central controller for Meta Agents Research Environments simulations. It provides:

Time Management: Controls simulation time, including start time, duration, and time increments
Event Orchestration: Manages the event loop, scheduling, and execution
App Registration: Coordinates registration and interaction between apps
State Management: Maintains simulation state and provides persistence
Lifecycle Control: Start, pause, resume, and stop simulation execution

Key Components ¶

Environment: Main simulation controller
EnvironmentConfig: Configuration settings for environment behavior
Event Loop: Background thread managing event execution
Time Manager: Unified time management across the simulation with pause/resume capabilities
Notification System: Message queuing and notification handling for agent communication

Basic Usage ¶

Simple Environment Setup ¶

from are.simulation.environment import Environment, EnvironmentConfig

# Create configuration
config = EnvironmentConfig(
    start_time=0,                    # Start at time 0 (or use datetime.now().timestamp())
    duration=60,                     # Run for 60 seconds
    time_increment_in_seconds=1      # 1-second time steps
)

# Create and start environment
env = Environment(config=config)
env.start()  # Non-blocking, starts event loop in separate thread

Environment Configuration ¶

import datetime
from are.simulation.environment import EnvironmentConfig

config = EnvironmentConfig(
    # Absolute start time - use datetime for specific times
    start_time=datetime.datetime(2024, 1, 1, 9, 0, 0).timestamp(),

    # Duration in seconds (None = run until main thread exits)
    duration=3600,  # 1 hour simulation

    # Time increment between ticks
    time_increment_in_seconds=0.5,
)

Lifecycle Management ¶

env = Environment(config=config)

# Start the simulation (non-blocking)
env.start()

# Simulation is now running in background
# Main thread can continue with other work
time.sleep(10)

# Pause simulation
env.pause()

# Do other work while paused
time.sleep(5)

# Resume simulation
env.resume()

# Keep main thread alive until simulation completes
time.sleep(config.duration)

App Registration ¶

from are.simulation.apps.email_client import EmailClientApp
from are.simulation.apps.calendar import CalendarApp

# Create apps
email_app = EmailClientApp()
calendar_app = CalendarApp()

# Register apps with environment
env.register_apps([email_app, calendar_app])

# Get registered app by name
email_app = env.get_app("EmailClientApp")

Event Scheduling ¶

from are.simulation.types import Event, Action

# Create an action
action = Action(
    function=email_app.add_email,
    args={"email": email_object}
)

# Create and schedule event
event = Event(action=action, event_time=env.start_time + 30)
env.schedule(event)

# Schedule multiple events
events = [event1, event2, event3]
env.schedule(events)

Advanced Features ¶

Time Management ¶

# Get current simulation time
current_time = env.time_manager.time()

# Check if simulation is running
if env.is_running():
    print("Simulation is active")

# Get simulation duration
total_duration = env.config.duration

# Get elapsed time
elapsed = env.time_manager.time_passed()

Event Access and Monitoring ¶

# Access event history
completed_events = env.event_log.list_view()

# Monitor specific event types
agent_events = [e for e in env.event_log.list_view() if e.event_type == EventType.AGENT]

Threading and Concurrency ¶

The Environment runs its event loop in a separate background thread, allowing the main thread to continue other work:

env = Environment(config)
env.start()  # Returns immediately, event loop runs in background

# You can do other stuff while event loop is running e.g. do nothing for a few seconds
time.sleep(2)

# You can also pause the simulation here for 3 seconds
env.pause()
time.sleep(3)

# And then resume it
env.resume()

For practical examples, see the environment tutorial Tutorials.

Core Environment Classes ¶

class are.simulation.environment.EnvironmentConfig(start_time=None, duration=60, time_increment_in_seconds=1, oracle_mode=False, dump_dir=None, exit_when_no_events=False, queue_based_loop=False, wait_for_user_input_timeout=None, verbose=True)[source]

Bases: object

Configuration for the Environment class.

This class encapsulates all the parameters required to configure the behavior of the Environment, including time management, event loop behavior, and other settings.

Attributes: - start_time (float or None): The start time of the environment in seconds. Defaults to 0 if None. - duration (float or None): The duration of the environment in seconds. If None, the environment runs indefinitely. - time_increment_in_seconds (int): The time increment between each tick in seconds. Must be >= 1. - oracle_mode (bool): Indicates if the environment is in oracle mode. In oracle mode, OracleEvents are processed. - dump_dir (str): Directory to dump environment states. Used only in oracle mode. - exit_when_no_events (bool): Determines whether to exit the event loop when there are no more events to process. If True, the event loop exits once all events are processed, even if the duration hasn’t been reached. This should be set to True only when running without an agent, as it prevents the event loop from blocking indefinitely in env.join(). When running with an agent, it should typically be False to allow the agent to generate more events over time. - queue_based_loop (bool): Controls whether the event loop is based on the queue or time: Queue-based event loop: The event queue is polled for events to process, and time jumps to the next event time. Time-based event loop: Time is incremented in fixed increments, and events are processed when they are ready. - wait_for_user_input_timeout (float): Timeout for waiting for user input in seconds. - verbose (bool): Indicates whether to print event loop ticks to the terminal.

start_time: float | None = None

duration: float | None = 60

time_increment_in_seconds: int = 1

oracle_mode: bool = False

dump_dir: str | None = None

exit_when_no_events: bool = False

queue_based_loop: bool = False

wait_for_user_input_timeout: float | None = None

verbose: bool = True

class are.simulation.environment.Environment(config=None, environment_type=EnvironmentType.UNKNOWN, notification_system=None, add_event_to_agent_log=None)[source]

Bases: AbstractEnvironment

The Environment class is responsible for managing the simulation environment.

It handles the event loop, time management, app registration, and event processing. The environment can run with or without an agent, and its behavior can be configured through the EnvironmentConfig class.

The event loop can operate in two modes: 1. Time-based: Time is incremented in fixed intervals, and events are processed when ready 2. Queue-based: The event queue is polled for events, and time jumps to the next event time

The event loop will exit when one of the following conditions is met: - The duration is reached (if specified) - The stop event is set (via the stop() method) - There are no more events to process (if exit_when_no_events is True)

This last condition should only be enabled when running without an agent, as it prevents the event loop from blocking indefinitely when there are no more events to process. When running with an agent, exit_when_no_events should typically be False to allow the agent to generate more events over time.

state: EnvironmentState | None

set_is_replaying(is_replaying)[source]

start(debug=False)[source]

Starts internal clock from environment

stop(final_state=EnvironmentState.STOPPED)[source]

Stop the event loop

join()[source]

Wait for the event loop to finish.

This method blocks until the event loop thread completes. The event loop will complete when one of the following conditions is met: 1. The duration is reached (if specified) 2. The stop event is set (via the stop() method) 3. There are no more events to process (if exit_when_no_events is True)

When running without an agent, it’s important to set exit_when_no_events=True in the EnvironmentConfig to ensure this method doesn’t block indefinitely. When running with an agent, exit_when_no_events should typically be False to allow the agent to generate more events over time.

pause()[source]: Pause the event loop

resume()[source]: Resume the event loop

resume_with_offset(offset)[source]

Resume the event loop with an offset

prepare_events_for_start()[source]: Prepare the events for start by scheduling them

delete_all_completed_events()[source]

get_apps_state()[source]

Get the state of the apps

Return type:: Any

apps_state_json()[source]

Return type:: str

event_log_json()[source]

Return type:: str

get_state()[source]

Get a simple dict representation of the environment state. With the states of the Apps, event log, event queue, and event triggers.

Return type:: dict[str, Any]

reset_app_states()[source]

put_last_event_from_log_to_queue()[source]: Put the last event in the log back in the event queue with the same time it was executed. This is useful for replay.

tick()[source]

Execute one tick of the event loop

Get triggered events
Get events to process and process them

wait_for_next_notification()[source]

Wait for the next notification by processing events until a notification is triggered. :param timeout: The timeout timestamp in seconds.

Return type:: None

jump_to_next_event_time()[source]: Skip to the next tick where events should be processed Thus skipping all ticks where no events are to be processed

get_next_event_time()[source]

Get the next event time from the event queue.

Return type:: float | None

process_event(event)[source]

Return type:: list[AbstractEvent]

resolve_arg_placeholders(event, past_events)[source]

Resolve arg placeholders in the event. If an event has an arg that is a placeholder for a past event, we replace it with the return value of that event.

Return type:: Event

get_successors(event)[source]

Get the successors of an event. We also set the correct event time for the successors.

Return type:: list[AbstractEvent]

final_validation_checks()[source]: Final validation checks to fail there are any remaining ValidationEvents or AgentValidationEvents

register_apps(apps)[source]

Registers apps to the environment, this function initializes the add event to an app. It is also used to verify if a condition is possible to be met in the triggers

append_to_world_logs(world_log)[source]

add_to_log(events)[source]

Add an event to the event log.

get_world_logs()[source]

Return type:: list[BaseAgentLog]

set_world_logs(world_logs)[source]

append_world_logs(log)[source]

increment_validators_tick_count()[source]

validate_agent_action(event)[source]

Validate an agent action.

run(scenario, wait_for_end=True, schedule_events=True)[source]

Run the event loop

schedule(events)[source]

Add future event to the event queue.

get_latest_event()[source]: Get the last completed event that happened from the event log.

get_past_event_at_index(index)[source]

Get the past event at a specific index from the event log. WARNING: this function can be expensive as the log is stored in a priority queue for now.

has_app(app_name)[source]

Check if an app is registered to the environment.

Return type:: bool

get_app(app_name)[source]

Get an app by its name.

Return type:: App

get_app_with_class(app_class)[source]

Check if an app is registered to the environment.

Return type:: Optional[TypeVar(AppType, bound= App)]

get_tools_by_app()[source]

Get for each app, the list of tools it has.

Return type:: dict[str, list[AppTool]]

get_tools()[source]

Get the entire list of tools from all the apps.

Return type:: list[AppTool]

get_user_tools_by_app()[source]

Get for each app, the list of tools it has that are accessible to the User.

Return type:: dict[str, list[AppTool]]

get_user_tools()[source]

Get the entire list of User tools from all the apps.

Return type:: list[AppTool]

get_event_log_size()[source]

Return type:: int

get_event_queue_length()[source]

Return type:: int

get_currently_active_validation_events()[source]

Get the currently active validation events.

Return type:: list[ValidationEvent]

log_info(message)[source]

log_debug(message)[source]

log_warning(message)[source]

log_error(message)[source]

is_paused()[source]

Return type:: bool

is_running()[source]

Return type:: bool

Environment ¶

class are.simulation.environment.Environment(config=None, environment_type=EnvironmentType.UNKNOWN, notification_system=None, add_event_to_agent_log=None)[source]

Bases: AbstractEnvironment

The Environment class is responsible for managing the simulation environment.

__init__(config=None, environment_type=EnvironmentType.UNKNOWN, notification_system=None, add_event_to_agent_log=None)[source]

event_log: EventLog

event_queue: EventQueue

state: EnvironmentState | None

protocol_to_app: dict[Protocol, App]

agent_action_validators: list[AgentActionValidator]

graphql_cache: dict[str, Any]

add_event_to_agent_log: Optional[Callable[[CompletedEvent], None]]

set_is_replaying(is_replaying)[source]

start(debug=False)[source]

Starts internal clock from environment

stop(final_state=EnvironmentState.STOPPED)[source]

Stop the event loop

join()[source]

Wait for the event loop to finish.

pause()[source]: Pause the event loop

resume()[source]: Resume the event loop

resume_with_offset(offset)[source]

Resume the event loop with an offset

prepare_events_for_start()[source]: Prepare the events for start by scheduling them

delete_all_completed_events()[source]

get_apps_state()[source]

Get the state of the apps

Return type:: Any

apps_state_json()[source]

Return type:: str

event_log_json()[source]

Return type:: str

get_state()[source]

Get a simple dict representation of the environment state. With the states of the Apps, event log, event queue, and event triggers.

Return type:: dict[str, Any]

reset_app_states()[source]

put_last_event_from_log_to_queue()[source]: Put the last event in the log back in the event queue with the same time it was executed. This is useful for replay.

tick()[source]

Execute one tick of the event loop

Get triggered events
Get events to process and process them

wait_for_next_notification()[source]

Wait for the next notification by processing events until a notification is triggered. :param timeout: The timeout timestamp in seconds.

Return type:: None

jump_to_next_event_time()[source]: Skip to the next tick where events should be processed Thus skipping all ticks where no events are to be processed

get_next_event_time()[source]

Get the next event time from the event queue.

Return type:: float | None

process_event(event)[source]

Return type:: list[AbstractEvent]

resolve_arg_placeholders(event, past_events)[source]

Resolve arg placeholders in the event. If an event has an arg that is a placeholder for a past event, we replace it with the return value of that event.

Return type:: Event

get_successors(event)[source]

Get the successors of an event. We also set the correct event time for the successors.

Return type:: list[AbstractEvent]

final_validation_checks()[source]: Final validation checks to fail there are any remaining ValidationEvents or AgentValidationEvents

register_apps(apps)[source]

Registers apps to the environment, this function initializes the add event to an app. It is also used to verify if a condition is possible to be met in the triggers

append_to_world_logs(world_log)[source]

add_to_log(events)[source]

Add an event to the event log.

get_world_logs()[source]

Return type:: list[BaseAgentLog]

set_world_logs(world_logs)[source]

append_world_logs(log)[source]

increment_validators_tick_count()[source]

validate_agent_action(event)[source]

Validate an agent action.

run(scenario, wait_for_end=True, schedule_events=True)[source]

Run the event loop

schedule(events)[source]

Add future event to the event queue.

get_latest_event()[source]: Get the last completed event that happened from the event log.

get_past_event_at_index(index)[source]

Get the past event at a specific index from the event log. WARNING: this function can be expensive as the log is stored in a priority queue for now.

has_app(app_name)[source]

Check if an app is registered to the environment.

Return type:: bool

get_app(app_name)[source]

Get an app by its name.

Return type:: App

get_app_with_class(app_class)[source]

Check if an app is registered to the environment.

Return type:: Optional[TypeVar(AppType, bound= App)]

get_tools_by_app()[source]

Get for each app, the list of tools it has.

Return type:: dict[str, list[AppTool]]

get_tools()[source]

Get the entire list of tools from all the apps.

Return type:: list[AppTool]

get_user_tools_by_app()[source]

Get for each app, the list of tools it has that are accessible to the User.

Return type:: dict[str, list[AppTool]]

get_user_tools()[source]

Get the entire list of User tools from all the apps.

Return type:: list[AppTool]

get_event_log_size()[source]

Return type:: int

get_event_queue_length()[source]

Return type:: int

get_currently_active_validation_events()[source]

Get the currently active validation events.

Return type:: list[ValidationEvent]

log_info(message)[source]

log_debug(message)[source]

log_warning(message)[source]

log_error(message)[source]

is_paused()[source]

Return type:: bool

is_running()[source]

Return type:: bool

EnvironmentConfig ¶

Bases: object

Configuration for the Environment class.

This class encapsulates all the parameters required to configure the behavior of the Environment, including time management, event loop behavior, and other settings.

start_time: float | None = None

duration: float | None = 60

time_increment_in_seconds: int = 1

oracle_mode: bool = False

dump_dir: str | None = None

exit_when_no_events: bool = False

queue_based_loop: bool = False

wait_for_user_input_timeout: float | None = None

verbose: bool = True

__init__(start_time=None, duration=60, time_increment_in_seconds=1, oracle_mode=False, dump_dir=None, exit_when_no_events=False, queue_based_loop=False, wait_for_user_input_timeout=None, verbose=True)

Key Methods ¶

Lifecycle Management

Environment.run(scenario, wait_for_end=True, schedule_events=True)[source]¶

Run the event loop

Environment.start(debug=False)[source]¶

Starts internal clock from environment

Environment.pause()[source]¶: Pause the event loop

Environment.resume()[source]¶: Resume the event loop

Environment.stop(final_state=EnvironmentState.STOPPED)[source]¶

Stop the event loop

Environment.join()[source]¶

Wait for the event loop to finish.

App Management

Environment.register_apps(apps)[source]¶

Registers apps to the environment, this function initializes the add event to an app. It is also used to verify if a condition is possible to be met in the triggers

Environment.get_app(app_name)[source]¶

Get an app by its name.

Return type:: App

Environment.get_tools_by_app()[source]¶

Get for each app, the list of tools it has.

Return type:: dict[str, list[AppTool]]

Event Management

Environment.schedule(events)[source]¶

Add future event to the event queue.

Environment.get_currently_active_validation_events()[source]¶

Get the currently active validation events.

Return type:: list[ValidationEvent]

Event Log ¶

The EventLog is a crucial component of the Environment that stores all completed events during simulation execution. It maintains a chronological record of all events that have occurred, enabling monitoring, analysis, and debugging of simulation behavior.

The event log is automatically populated as events are processed by the environment. Each time an event completes (successfully or with errors), it’s added to the event log as a CompletedEvent instance.

EventLog Class ¶

class are.simulation.types.EnvironmentState(value)[source]

Bases: Enum

The state of the environment. - SETUP: the environment is being setup, this is the initial state of the environment - RUNNING: the environment event loop is running and events are being registered and logged - PAUSED: the environment is paused, and no events are being registered or logged, but can be restarted - STOPPED: the environment is completely stopped

class are.simulation.types.EventTimeComparator(value)[source]

Bases: Enum

Comparator for event time filtering. - LESS_THAN: Less than comparison - GREATER_THAN: Greater than comparison - EQUAL: Equal comparison

class are.simulation.types.AbstractEnvironment[source]

Bases: ABC

class are.simulation.types.HintType(value)[source]

Bases: Enum

Type of the hint, depends on the linked event - TASK_HINT: hints initiated by the send_message_to_agent

class are.simulation.types.Hint(*, hint_type, content, associated_event_id)[source]

Bases: object

Hint associated with an event - hint_type: Type of the hint, depends on the linked event - content: Content of the hint - associated_event_id: The id of the event that this hint is associated with

class are.simulation.types.EnvironmentType(value)[source]

Bases: Enum

The type of the environment.

class are.simulation.types.EventType(value)[source]

Bases: Enum

Type of the events, depends on who initiated them. - AGENT: events initiated by the agent - ENV: events initiated by the environment, or the scenario designer. - USER: events initiated by the user or user proxy (unused for now). - CONDITION: events that check a condition and trigger other events. - VALIDATION: events that validate the state of the environment. - STOP: events that stop the simulation.

class are.simulation.types.Action(function, args=<factory>, resolved_args=<factory>, app=None, action_id=None, operation_type=OperationType.READ, tool_metadata=None)[source]

Bases: object

Action associated with an event, this is a function that will be called when the event is executed.

function: Function to be called when the action is executed, it can be a class method, or a regular function
args: Dict mapping the argument names to values to call the function with at execution time
app: The actual App instance to use for the action execution
If not specified at creation time, it can be deducted from the function instance e.g. if function=email_app.add_emails then we can deduct that app=email_app
action_id: The unique id of the action, this is used to identify the event in the logs.
This is created automatically and does NOT need to be handled by the user
tool_metadata: Optional metadata from the AppTool that this action is associated with

class are.simulation.types.ActionDescription(app, function, args)[source]

Bases: object

class are.simulation.types.ConditionCheckAction(function, action_id=None)[source]

Bases: object

class are.simulation.types.EventMetadata(return_value=None, exception=None, exception_stack_trace=None, completed=True)[source]

Bases: object

Metadata for a completed event, which includes details such as the return value, exception, etc. This is particularly useful for completed events from App API calls, where: - return_value: the return value of the API call. - exception: any exception that occurred during the API call. - exception_stack_trace: the stack trace of the exception, if any. - completed: a flag indicating whether the execution was completed.

class are.simulation.types.AbstractEvent(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>)[source]

Bases: ABC

Abstract event class, that contains shared field between completed and future events.

event_type: the type of the event, either AGENT, ENV or USER.
action: the action that will be executed when the event happens, either directly a function or an Action obj.
event_time: the time at which the event will happen, this can get overridden in various placed for e.g. in case of conditional triggers.
event_relative_time: the relative time wrt the simulation start time
WARNING when the event is going to be added to the queue, this information is going to be used to set event_time
event_id: the unique id of the event, this is used to identify the event in the logs.

depends_on(events=None, delay_seconds=0)[source]

This function is used to add dependencies to the event. If e1 depends on e2 and e3, then e1 will only be executed after e2 and e3 are executed. If a delay is specified, then the event will be executed after the delay after the dependencies are executed.

followed_by(events, delay_seconds=0.0)[source]

This function is used to add successors to the event. If e1 is followed by e2 and e3, then e2 and e3 will only be executed after e1 is executed. If a delay is specified, then the event will be executed after the delay after the dependencies are executed.

is_ready()[source]

This function is used to check if the event is ready to be scheduled i.e. put into the event_queue. An event is ready to be executed if all its dependencies are executed. When an event has its event_time set, it means it is ready to be scheduled.

Return type:: bool

compute_absolute_time(start_time=0)[source]

Here we compute the absolute time of an event based on its relative time as well as the time of its dependencies.

class are.simulation.types.Event(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, action=None)[source]

Bases: AbstractEvent

Represents an event that will happen in the future. This is what we create often when populating a scenario, and what gets added to the event queue.

execute(fire_individual_events=False)[source]

Executes the action corresponding to the events and returns the completed event with its metadata. Here by default we make sure we only have ONE event, and that whatever happens inside the action is not registered as individual events. This is to guarantee 2 things: 1/ Events are transactional, either the whole Event (and associated action) happened or not 2/ Not duplicating what is registered in the event log (otherwise we will register BOTH the current event, and every underlying one)

Return type:: CompletedEvent

class are.simulation.types.StopEvent(*args, **kwargs)[source]: Bases: AbstractEvent

class are.simulation.types.CompletedEvent(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, action=None, metadata=None, _tool_name=None)[source]

Bases: AbstractEvent

Represents an event that already happened, and thus we have some additional metadata on it.

property tool_name: Tool name used for validation

class are.simulation.types.ConditionCheckEvent(event_type=EventType.CONDITION, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, action=None, schedule_every_ticks=1, timeout=None, _internal_check_count=0)[source]

Bases: AbstractEvent

depends_on(events=None, delay_seconds=0, schedule_every_ticks=None, timeout=None)[source]

class are.simulation.types.OracleEvent(event_type=EventType.AGENT, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, make_event=None, event_time_comparator=None, action_desc=None)[source]

Bases: AbstractEvent

class are.simulation.types.CompletedOracleEvent(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, action=None, metadata=None, _tool_name=None, absolute_event_time=None, event_time_comparator=None)[source]

Bases: CompletedEvent

A completed oracle event with timing information from the original oracle event.

classmethod from_completed_event_and_oracle_event(completed_event, oracle_event)[source]

Create a completed oracle event from a completed event and an oracle event. The absolute event time is taken from the oracle event, and the event time comparator is taken from the oracle event if it exists.

class are.simulation.types.ValidationResult(success, achieved_milestones=<factory>)[source]

Bases: object

Represents the result of a validation event. - success: whether the validation was successful or not. - message: a message describing the result of the validation. - failed_milestones: the list of milestones that failed during the validation. - triggered_minefields: the list of minefields that were triggered during the validation.

class are.simulation.types.AgentActionValidator(milestones=<factory>, minefields=<factory>, timeout=None, _start_tick=0, _internal_check_count=0, achieved_milestones=<factory>)[source]

Bases: object

class are.simulation.types.AgentValidationEvent(event_type=EventType.VALIDATION, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, validators=<factory>, milestones=<factory>, minefields=<factory>, timeout=None)[source]

Bases: AbstractEvent

depends_on(events=None, delay_seconds=0, schedule_every_ticks=None, timeout=None)[source]

exception are.simulation.types.ValidationException[source]: Bases: Exception

class are.simulation.types.ValidationEvent(event_type=EventType.VALIDATION, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, milestones=<factory>, minefields=<factory>, schedule_every_ticks=1, timeout=1, _internal_check_count=0, achieved_milestones=<factory>)[source]

Bases: AbstractEvent

depends_on(events=None, delay_seconds=0, schedule_every_ticks=None, timeout=None)[source]

class are.simulation.types.EventLog(past_events=<factory>)[source]

Bases: object

Event log, contains all the events that happened so far in the environment.

class are.simulation.types.EventQueue(future_events=<factory>, already_scheduled=<factory>)[source]

Bases: object

Event queue, contains all the events that will happen in the future.

class are.simulation.types.EventRegisterer[source]

Bases: object

Class that handles all the logic for registering events.

classmethod is_active()[source]: Checks whether we should fire events or not.

classmethod is_capture_mode()[source]: Capture mode makes sure a function call is not executed but only a “fictitious” CompletedEvent is returned. This is useful for debugging and testing, as well as easily creating CompletedEvent instances for validation

classmethod set_active(state)[source]: Sets whether we should fire events or not.

classmethod set_capture_mode(state)[source]: Sets whether capture mode is active or not.

classmethod event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT)[source]

This decorator is used to wrap API calls, so that we can register the event and add the appropriate CompletedEvent instance. The CompletedEvent instance is only added to the Event Log if the App is already registered in the environment.

This decorator is also used to capture fictitious CompletedEvent instances when capture mode is active. Capture mode allows to easily simulate and create CompletedEvent instances without actually executing the API call. This is useful for debugging and testing, as well as defining validation trajectories.

classmethod disable()[source]: Context manager to disable event firing, and just return the result of the function call as if decorator is not applied.

classmethod capture_mode()[source]: Context manager to replace a function call decorated with event_registered, by just a fictitious CompletedEvent instance.

are.simulation.types.event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT)

are.simulation.types.disable_events(): Context manager to disable event firing, and just return the result of the function call as if decorator is not applied.

class are.simulation.types.CapabilityTag(value)[source]

Bases: Enum

An enumeration.

class are.simulation.types.ScenarioGUIConfig(show_timestamps=False)[source]

Bases: object

class are.simulation.types.ToolAugmentationConfig(tool_failure_probability=0.1, apply_tool_name_augmentation=True, apply_tool_description_augmentation=True)[source]

Bases: object

class are.simulation.types.SimulatedGenerationTimeConfig(mode='measured', seconds=1.0)[source]

Bases: object

Configuration for simulating the LLM’s generation time in are.simulation.

Modes:

fixed: The simulated generation time is set to a fixed to seconds.
measured: The simulated generation time is measured from the first successful generation.

class are.simulation.types.PlaceholderMetadata(parent_tool_name, parent_turn_idx, parent_event_id, placeholder_turn_idx, placeholder_event_id)[source]

Bases: object

Metadata for a placeholder.

class are.simulation.types.ExecutionMetadata(has_placeholder_conflicts, placeholders)[source]

Bases: object

Data structure for execution data.

class are.simulation.types.EventLog(past_events=<factory>)[source]

Bases: object

Event log, contains all the events that happened so far in the environment.

past_events: PriorityQueue[CompletedEvent]

put(event)[source]

__len__()[source]

list_view()[source]

Return type:: list[CompletedEvent]

to_dict()[source]

static from_list_view(events)[source]

__init__(past_events=<factory>)

Event Log Usage ¶

# Access event log from environment
env = Environment(config)

# Get all completed events
completed_events = env.event_log.list_view()

# Get event count
event_count = len(env.event_log)

# Filter events by type
agent_events = [e for e in env.event_log.list_view() if e.event_type == EventType.AGENT]
env_events = [e for e in env.event_log.list_view() if e.event_type == EventType.ENV]

# Access specific event details
for event in completed_events:
    print(f"Event ID: {event.event_id}")
    print(f"Event Type: {event.event_type}")
    print(f"Event Time: {event.event_time}")
    print(f"App: {event.app_name()}")
    print(f"Function: {event.function_name()}")
    print(f"Return Value: {event.metadata.return_value}")
    if event.failed():
        print(f"Exception: {event.metadata.exception}")

Key Features ¶

Chronological Storage: Events are stored in order by their execution time using a priority queue
Immutable Records: Events are copied when added to prevent mutation of logged events
Rich Metadata: Each logged event includes execution metadata like return values, exceptions, and timing
Flexible Access: Provides both list view and dictionary serialization for different use cases
Memory Efficient: Uses priority queue for efficient storage and retrieval

The event log is essential for:

Monitoring: Track what events have occurred during simulation
Debugging: Examine event execution order and results
Analysis: Post-simulation analysis of agent behavior and environment state changes
Validation: Verify that expected events occurred during simulation

Time Manager ¶

The TimeManager class provides sophisticated time management for simulations, including pause/resume capabilities and time offset handling.

class are.simulation.time_manager.TimeManager[source]

Bases: object

A class for managing time in the simulation.

time()[source]

Returns the current time, which is the sum of the start time and the time passed.

Return type:: float

Returns:: float: The current time.

time_passed()[source]

Returns the time passed since the start time, taking into account any pauses.

Return type:: float

Returns:: float: The time passed.

real_time_passed()[source]

Returns the real time passed since the start time, without considering any pauses.

Return type:: float

Returns:: float: The real time passed.

reset(start_time=None)[source]

Resets the start time to the current time, or to the specified start_time if provided.

Return type:: None

Args:: start_time (float | None): The new start time. Defaults to current time.

pause()[source]

Pauses the time.

Return type:: None

resume()[source]

Resumes the time.

Return type:: None

set_offset(offset)[source]

Return type:: None

add_offset(offset)[source]

Return type:: None

TimeManager ¶

class are.simulation.time_manager.TimeManager[source]

Bases: object

A class for managing time in the simulation.

__init__()[source]

time()[source]

Returns the current time, which is the sum of the start time and the time passed.

Return type:: float

Returns:: float: The current time.

time_passed()[source]

Returns the time passed since the start time, taking into account any pauses.

Return type:: float

Returns:: float: The time passed.

real_time_passed()[source]

Returns the real time passed since the start time, without considering any pauses.

Return type:: float

Returns:: float: The real time passed.

reset(start_time=None)[source]

Resets the start time to the current time, or to the specified start_time if provided.

Return type:: None

Args:: start_time (float | None): The new start time. Defaults to current time.

pause()[source]

Pauses the time.

Return type:: None

resume()[source]

Resumes the time.

Return type:: None

set_offset(offset)[source]

Return type:: None

add_offset(offset)[source]

Return type:: None

Time Management Usage ¶

from are.simulation.time_manager import TimeManager

# Create time manager
time_manager = TimeManager()

# Get current simulation time
current_time = time_manager.time()

# Get time passed since start
elapsed = time_manager.time_passed()

# Pause simulation time
time_manager.pause()

# Resume simulation time
time_manager.resume()

# Reset time with custom start time
import time
time_manager.reset(time.time())

# Add time offset (e.g., for time jumps)
time_manager.add_offset(3600)  # Jump forward 1 hour

Notification System ¶

The Notification System provides message queuing and notification handling for agent communication and environment events.

class are.simulation.notification_system.MessageType(value)[source]

Bases: Enum

An enumeration.

USER_MESSAGE = 'USER_MESSAGE'

ENVIRONMENT_NOTIFICATION = 'ENVIRONMENT_NOTIFICATION'

ENVIRONMENT_STOP = 'ENVIRONMENT_STOP'

class are.simulation.notification_system.VerbosityLevel(value)[source]

Bases: IntEnum

Defines the verbosity levels for the notification system. LOW: Only user messages and system notifications (due reminders, wait for notification timeouts) are notified. MEDIUM: ENV events that are possible consequences of Agent actions are notified. HIGH: Most ENV events are notified, even independent ENV events that are not caused by Agent actions.

LOW = 1

MEDIUM = 2

HIGH = 3

class are.simulation.notification_system.Message(message_type, message, timestamp, attachments=<factory>)[source]

Bases: object

message_type: MessageType

message: str

timestamp: datetime

attachments: list[Attachment]

class are.simulation.notification_system.AbstractMessageQueue[source]

Bases: object

abstract put(message)[source]

Return type:: None

abstract get_by_timestamp(timestamp)[source]

Return type:: list[Message]

class are.simulation.notification_system.MessageQueue[source]

Bases: AbstractMessageQueue

put(message)[source]

Return type:: None

get_by_timestamp(timestamp)[source]

Return type:: list[Message]

has_environment_stop_message()[source]

Return type:: bool

list_view()[source]

Return type:: list[Message]

has_new_messages(timestamp)[source]

Return type:: bool

class are.simulation.notification_system.NotificationSystemConfig(notified_tools=<factory>)[source]

Bases: object

notified_tools: dict[str, list[str]]

are.simulation.notification_system.get_notification_tools(verbosity_level=VerbosityLevel.LOW)[source]

Returns the tools to be notified based on the specified verbosity level.

Return type:: dict[str, list[str]]

Args:: level: The verbosity level to determine which tools to notify
Returns:: A dictionary mapping app names to lists of function names to be notified

class are.simulation.notification_system.BaseNotificationSystem(config=NotificationSystemConfig(notified_tools={}))[source]

Bases: object

Base class for all notification systems.

initialize(time_manager)[source]

Return type:: None

setup_reminder_app(reminder_app)[source]

Return type:: None

setup_system_app(system_app)[source]

Return type:: None

clear()[source]

Return type:: None

get_current_time()[source]

Return type:: float

get_next_notification_time()[source]

Return type:: float | None

handle_time_based_notifications()[source]

Return type:: None

handle_timeout_after_events()[source]

Handle timeout notifications after all events in the current tick have been processed. This ensures that timeout notifications are only triggered if no other notifications were generated during the same tick.

Return type:: None

convert_reminders_to_message(due_reminders)[source]

Convert due reminders to a notification message.

Return type:: Message | None

convert_wait_for_notification_timeout_to_message(wait_for_notification_timeout)[source]

Return type:: Message | None

handle_event(event)[source]

Return type:: None

convert_to_message(event)[source]

Return type:: Message | None

are.simulation.notification_system.get_args(event)[source]

Return type:: dict[str, Any]

class are.simulation.notification_system.VerboseNotificationSystem(verbosity_level=VerbosityLevel.MEDIUM)[source]

Bases: BaseNotificationSystem

are.simulation.notification_system.get_content_for_message(event)[source]

Return type:: str | None

Core Classes ¶

BaseNotificationSystem¶

class are.simulation.notification_system.BaseNotificationSystem(config=NotificationSystemConfig(notified_tools={}))[source]

Bases: object

Base class for all notification systems.

__init__(config=NotificationSystemConfig(notified_tools={}))[source]

initialize(time_manager)[source]

Return type:: None

setup_reminder_app(reminder_app)[source]

Return type:: None

setup_system_app(system_app)[source]

Return type:: None

clear()[source]

Return type:: None

get_current_time()[source]

Return type:: float

get_next_notification_time()[source]

Return type:: float | None

handle_time_based_notifications()[source]

Return type:: None

handle_timeout_after_events()[source]

Return type:: None

convert_reminders_to_message(due_reminders)[source]

Convert due reminders to a notification message.

Return type:: Message | None

convert_wait_for_notification_timeout_to_message(wait_for_notification_timeout)[source]

Return type:: Message | None

handle_event(event)[source]

Return type:: None

convert_to_message(event)[source]

Return type:: Message | None

VerboseNotificationSystem¶

class are.simulation.notification_system.VerboseNotificationSystem(verbosity_level=VerbosityLevel.MEDIUM)[source]

Bases: BaseNotificationSystem

__init__(verbosity_level=VerbosityLevel.MEDIUM)[source]

Configuration Classes ¶

NotificationSystemConfig¶

class are.simulation.notification_system.NotificationSystemConfig(notified_tools=<factory>)[source]

Bases: object

notified_tools: dict[str, list[str]]

__init__(notified_tools=<factory>)

Message Classes ¶

Message¶

class are.simulation.notification_system.Message(message_type, message, timestamp, attachments=<factory>)[source]

Bases: object

message_type: MessageType

message: str

timestamp: datetime

attachments: list[Attachment]

__init__(message_type, message, timestamp, attachments=<factory>)

MessageQueue¶

class are.simulation.notification_system.MessageQueue[source]

Bases: AbstractMessageQueue

__init__()[source]

put(message)[source]

Return type:: None

get_by_timestamp(timestamp)[source]

Return type:: list[Message]

has_environment_stop_message()[source]

Return type:: bool

list_view()[source]

Return type:: list[Message]

has_new_messages(timestamp)[source]

Return type:: bool

Enumerations ¶

MessageType¶

class are.simulation.notification_system.MessageType(value)[source]

Bases: Enum

An enumeration.

USER_MESSAGE = 'USER_MESSAGE'

ENVIRONMENT_NOTIFICATION = 'ENVIRONMENT_NOTIFICATION'

ENVIRONMENT_STOP = 'ENVIRONMENT_STOP'

VerbosityLevel¶

class are.simulation.notification_system.VerbosityLevel(value)[source]

Bases: IntEnum

LOW = 1

MEDIUM = 2

HIGH = 3

Notification System Usage ¶

from are.simulation.notification_system import (
    BaseNotificationSystem,
    VerboseNotificationSystem,
    VerbosityLevel,
    NotificationSystemConfig
)

# Create a notification system with specific verbosity
notification_system = VerboseNotificationSystem(
    verbosity_level=VerbosityLevel.MEDIUM
)

# Or create with custom configuration
config = NotificationSystemConfig(
    notified_tools={
        "EmailClientApp": ["send_email", "receive_email"],
        "CalendarApp": ["add_event", "delete_event"]
    }
)
base_notification_system = BaseNotificationSystem(config)

# Initialize with time manager
from are.simulation.time_manager import TimeManager
time_manager = TimeManager()
notification_system.initialize(time_manager)

# Handle events
notification_system.handle_event(some_event)

# Get messages for current time
current_time = datetime.now(timezone.utc)
messages = notification_system.message_queue.get_by_timestamp(current_time)

Verbosity Levels¶

The notification system supports three verbosity levels:

LOW (VerbosityLevel.LOW): Only user messages and system notifications
MEDIUM (VerbosityLevel.MEDIUM): Environment events that are consequences of agent actions
HIGH (VerbosityLevel.HIGH): Most environment events, including independent events

Key Functions ¶

are.simulation.notification_system.get_notification_tools(verbosity_level=VerbosityLevel.LOW)[source]¶

Returns the tools to be notified based on the specified verbosity level.

Return type:: dict[str, list[str]]

Args:: level: The verbosity level to determine which tools to notify
Returns:: A dictionary mapping app names to lists of function names to be notified

are.simulation.notification_system.get_content_for_message(event)[source]¶

Return type:: str | None