Scenarios API

Scenarios define the tasks and environments that agents operate within. This section documents the core Scenario API and related classes for creating and managing simulation scenarios.

Overview

Scenarios in Meta Agents Research Environments are complete simulation setups that include:

• Apps and Initial State: The applications and data available to agents

• Events and Timeline: Dynamic occurrences that happen during execution

• Task Definition: What the agent needs to accomplish

• Validation Logic: How success is measured and evaluated

Creating Custom Scenarios

For a comprehensive guide on creating and registering custom scenarios, see Scenario Development.

To create a custom scenario, inherit from the base Scenario class and implement the required methods:

Basic Scenario Structure

from are.simulation.scenarios.scenario import Scenario
from are.simulation.scenarios.utils.registry import register_scenario
from are.simulation.apps.email_client import EmailClientApp, Email
from are.simulation.apps.calendar import CalendarApp
from are.simulation.types import EventType

@register_scenario("my_custom_scenario")
class MyCustomScenario(Scenario):
    """
    A custom scenario that demonstrates email and calendar integration.
    """
    scenario_id: str = "my_custom_scenario"
    start_time: float | None = 0
    duration: float | None = 600  # Duration in seconds

    def init_and_populate_apps(self):
        """Initialize and populate apps with data."""
        # Create apps
        email_app = EmailClientApp()
        calendar_app = CalendarApp()

        # Populate with initial data
        email_app.add_email(
             email=Email(
                 sender="boss@company.com",
                 recipients=[self.email_app.user_email],
                 subject="Important Meeting",
                 content="Please schedule a 'Team Standup' for next week same time.",
             ),
        )

        # Add calendar events
         calendar_app.add_calendar_event(
             title="Team Standup"
             start_datetime="2024-01-15 09:00:00"
             end_datetime="2024-01-15 09:30:00"
         )

        # Register apps
        self.apps = [email_app, calendar_app]

    def build_events_flow(self):
        """Define the sequence of events in the scenario."""
         aui = self.get_typed_app(AgentUserInterface)
         calendar_app = self.get_typed_app(CalendarApp)

         with EventRegisterer.capture_mode():
             # Add user task to read email
             aui.send_message_to_agent(
                 content="Read my email and schedule the meeting my boss requested."
             ).depends_on(None, delay_seconds=5)
             # Add the calenddar event
             calendar_app.add_calendar_event(
                 title="Team Standup"
                 start_datetime="2024-01-22 09:00:00"
                 end_datetime="2024-01-22 09:30:00"
             )


    def validate(self, env) -> ScenarioValidationResult:
        """Validate that the scenario was completed successfully."""
        try:
            # Check if a meeting was scheduled
            success = any(
                 event.event_type == EventType.AGENT
                 and isinstance(event.action, Action)
                 and event.action.function_name == "add_calendar_event"
                 and event.action.class_name == "CalendarApp"
                 and event.action.args["start_datetime"] == "2024-01-22 09:00:00"
                 and event.action.args["end_datetime"] == "2024-01-22 09:30:00"
                 and event.action.args["title"] == "Team Standup"
                 for event in env.event_log.list_view()
             )

            return ScenarioValidationResult(success=success)
        except Exception as e:
            return ScenarioValidationResult(success=False, exception=e)

App Initialization Patterns

Simple App Setup

def init_and_populate_apps(self):
    # Create apps
    file_system = VirtualFileSystemApp()
    email_app = EmailClientApp()

    # Populate file system
    file_system.create_file("/documents/report.txt", "Quarterly report content")
    file_system.mkdir("/images")

    # Populate email
    email_app.add_email(
        email_app.add_email(
             email=Email(
                 sender="client@company.com",
                 recipients=[self.email_app.user_email],
                 subject="Report Request",
                 content="Please send the quarterly report.",
             ),
        )
    )

    self.apps = [file_system, email_app]

Complex App Configuration

def init_and_populate_apps(self):
    # Create apps with custom configurations
    shopping_app = ShoppingApp()
    messaging = MessagingApp()

    # Set up product catalog
    products = {
         "6938111410": {
             "name": "Running Shoes",
             "product_id": "6938111410",
             "variants": {
                 "4153505238": {
                     "item_id": "4153505238",
                     "options": {
                         "size": "8",
                         "color": "red",
                         "material": "leather",
                         "sole": "EVA",
                     },
                     "available": True,
                     "price": 158.67,
                 },
                 "1775591963": {
                     "item_id": "1775591963",
                     "options": {
                         "size": "10",
                         "color": "white",
                         "material": "leather",
                         "sole": "EVA",
                     },
                     "available": True,
                     "price": 154.75,
                 },
             },
         },
         "8310926033": {
             "name": "Water Bottle",
             "product_id": "8310926033",
             "variants": {
                 "1434748144": {
                     "item_id": "1434748144",
                     "options": {
                         "capacity": "1000ml",
                         "material": "glass",
                         "color": "red",
                     },
                     "available": False,
                     "price": 49.72,
                 },
                 "4579334072": {
                     "item_id": "4579334072",
                     "options": {
                         "capacity": "750ml",
                         "material": "glass",
                         "color": "black",
                     },
                     "available": True,
                     "price": 54.85,
                 },
             },
         },
     }
    shopping_app.load_products_from_dict(products)

    # Add conversation
     messaging.add_conversations(
         [
             Conversation(
                 participants=["Dad", "Me"],
                 title="Dad and I",
                 conversation_id="23",
                 messages=[
                     Message(
                         sender="Dad",
                         message_id="0",
                         timestamp=1708550557,
                         content="Hey kid, how are you doing?",
                     ),
                     Message(
                         sender="Me",
                         message_id="1",
                         timestamp=1708550560,
                         content="I am doing good dad. How are you?",
                     ),
                     Message(
                         sender="Dad",
                         message_id="2",
                         timestamp=1708550570,
                         content="I am doing good son. Just came back home from camping trip. It was good. Missed you. What about you?",
                     ),
                     Message(
                         sender="Me",
                         message_id="3",
                         timestamp=1708550580,
                         content="Great dad! I missed it.",
                     ),
                     Message(
                         sender="Dad",
                         message_id="4",
                         timestamp=1708550590,
                         content="Btw son, what car does aunt Linda drive?",
                     ),
                     Message(
                         sender="Me",
                         message_id="5",
                         timestamp=1708550600,
                         content="It's Toyota.",
                     ),
                     Message(
                         sender="Dad",
                         message_id="6",
                         timestamp=1708550620,
                         content="Thanks kid. I will remember it. See you soon. Bye",
                     ),
                     Message(
                         sender="Me",
                         message_id="7",
                         timestamp=1708550650,
                         content="Bye Dad",
                     ),
                 ],
             ),
         ]
     )

    self.apps = [shopping_app, payment_app]

Event Flow Patterns

Sequential Events

def build_events_flow(self):
     with EventRegisterer.capture_mode():
         # Event 1: Send initial message
         event1 = (
             aui.send_message_to_user(
                 content="Task started",
             )
             .oracle()
             .depends_on(None, delay_seconds=5)
         )

         # Event 2: Follow-up after 60 seconds
         event2 = (
             aui.send_message_to_user(
                 content="Task done",
             )
             .oracle()
             .depends_on(event1, delay_seconds=60)
         )

Conditional Events

# These events trigger when certain conditions are met

def enough_emails_condition(env: AbstractEnvironment) -> bool:
    """Condition: trigger when we have at least 2 emails"""
    email_app = env.get_app("EmailClientApp")
    return len(email_app.folders[EmailFolderName.INBOX].emails) >= 2

# Create a condition check event
condition_check = ConditionCheckEvent.from_condition(enough_emails_condition)

# Event that triggers when condition is met
conditional_email = Event.from_function(
    self.email_app.add_email,
    email=Email(
        sender="system@example.com",
        recipients=[self.email_app.user_email],
        subject="Conditional Event Triggered!",
        content="This email was sent because the condition (2+ emails) was met!",
    ),
).depends_on(condition_check, delay_seconds=1)

Validation Patterns

Validation Event

def both_responses_validator(env: AbstractEnvironment) -> bool:
    """Check if agent gave both responses."""
    aui_app = env.get_app("AgentUserInterface")
    agent_messages = aui_app.get_all_messages_from_agent()
    has_correct = any(msg.content == "I am a robot" for msg in agent_messages)
    has_incorrect = any(
        msg.content == "I am not a robot •`_´• " for msg in agent_messages
    )
    return has_correct and not has_incorrect

continuous_validation = ValidationEvent(
    milestones=[both_responses_validator]
).with_id("continuous_validation")
continuous_validation.depends_on(user_request, delay_seconds=1)
continuous_validation.schedule(every_ticks=1, timeout=15)
self.events.append(continuous_validation)

State-Based Validation

def validate(self, env: AbstractEnvironment) -> ScenarioValidationResult:
    """
    Final scenario validation - checking the overall
    state after all events have completed.
    """
    try:
        aui_app = env.get_app("AgentUserInterface")

        # Check that agent provided responses
        agent_messages = aui_app.get_all_messages_from_agent()
        has_one_responses = len(agent_messages) == 1

        # Check that the correct response was given
        has_correct_response = any(
            msg.content == "I am a robot" for msg in agent_messages
        )

        # Check that user made the initial request
        user_messages = aui_app.get_all_messages_from_user()
        has_user_request = len(user_messages) > 0

        # Overall success criteria
        success = has_one_responses and has_correct_response and has_user_request

        return ScenarioValidationResult(success=success)

    except Exception as e:
        return ScenarioValidationResult(success=False, exception=e)

Agent Validation Event

def milestone_correct_response(
    env: AbstractEnvironment, event: AbstractEvent
) -> bool:
    """Milestone: Agent should say 'I am a robot'."""
    if not isinstance(event, Event):
        return False
    return (
        event.action.app.__class__ == AgentUserInterface
        and event.function_name() == "send_message_to_user"
        and event.action.args["content"] == "I am a robot"
    )

def milestone_incorrect_response(
    env: AbstractEnvironment, event: AbstractEvent
) -> bool:
    """Milestone: Agent should not say the incorrect response."""
    if not isinstance(event, Event):
        return False
    return not (
        event.action.app.__class__ == AgentUserInterface
        and event.function_name() == "send_message_to_user"
        and event.action.args["content"] == "I am not a robot •`_´• "
    )

def minefield_unsafe_response(
    env: AbstractEnvironment, event: AbstractEvent
) -> bool:
    """Minefield: Agent should NOT say unsafe things."""
    if not isinstance(event, Event):
        return False
    return (
        event.action.app.__class__ == AgentUserInterface
        and event.function_name() == "send_message_to_user"
        and event.action.args["content"] == "I am doing something unsafe!"
    )

agent_validation = AgentValidationEvent(
    milestones=[milestone_correct_response, milestone_incorrect_response],
    minefields=[minefield_unsafe_response],
    timeout=15,
).with_id("agent_validation")
agent_validation.depends_on(user_request)
self.events.append(agent_validation)

Advanced Features

Hints

from are.simulation.types import Hint, HintType

def build_events_flow(self):
    # Add event with associated hint
     with EventRegisterer.capture_mode():
         event1 = aui.send_message_to_agent(
             content="Send an email to the client for the delay."
         ).depends_on(None, delay_seconds=5).with_id("event1")

    # Add hint for this event
    if not self.hints:
        self.hints = []

    self.hints.append(Hint(
        hint_type=HintType.TASK_HINT,
        content="Remember to include an apology in your email",
        associated_event_id="event1"
    ))

Tool Augmentation

from are.simulation.types import ToolAugmentationConfig

class UnreliableScenario(Scenario):
    def __init__(self):
        super().__init__()
        # Make tools fail 20% of the time
        self.tool_augmentation_config = ToolAugmentationConfig(
            tool_failure_probability=0.2,
        )

Environment Events

from are.simulation.scenarios.utils.scenario_expander import EnvEventsConfig

class DynamicScenario(Scenario):
    def __init__(self):
        super().__init__()
        # Add random environment events
        self.env_events_config = EnvEventsConfig(
            num_env_events_per_minute=3,
            env_events_seed=42,
            weight_per_app_class = {
                 "EmailClientApp": 1.0,
                 "ApartmentListingApp": 1.0,
             },
        )

Core Scenario Classes

class are.simulation.scenarios.scenario.AutoDataclass(name, bases, namespace)

Bases: type

class are.simulation.scenarios.scenario.ScenarioStatus(value)

Bases: Enum

An enumeration.

Draft = 'Draft'

AwaitingReview = 'AwaitingReview'

Valid = 'Valid'

Invalid = 'Invalid'

Abandoned = 'Abandoned'

Safe = 'safe'

Violating = 'violating'

class are.simulation.scenarios.scenario.Scenario(_initialized=False, is_benchmark_ready=False, events=<factory>, apps=None, tags=<factory>, scenario_id='', seed=0, nb_turns=None, run_number=None, config=None, has_a2a_augmentation=False, status=ScenarioStatus.Draft, comment=None, annotation_id=None, hints=None, additional_system_prompt=None, start_time=<factory>, duration=None, queue_based_loop=False, time_increment_in_seconds=1, working_dir='', _initial_apps=None, tool_augmentation_config=None, env_events_config=None, gui_config=None, augmentation_data=<factory>)

Bases: object

is_benchmark_ready: bool = False

events: list[AbstractEvent]

apps: list[App] | None = None

tags: tuple[CapabilityTag, ...]

scenario_id: str = ''

seed: int = 0

nb_turns: int | None = None

run_number: int | None = None

config: str | None = None

has_a2a_augmentation: bool = False

status: ScenarioStatus = 'Draft'

comment: str | None = None

annotation_id: str | None = None

hints: list[Hint] | None = None

additional_system_prompt: str | None = None

start_time: float | None

duration: float | None = None

queue_based_loop: bool = False

time_increment_in_seconds: int = 1

working_dir: str = ''

tool_augmentation_config: ToolAugmentationConfig | None = None

env_events_config: EnvEventsConfig | None = None

gui_config: ScenarioGUIConfig | None = None

augmentation_data: dict[str, Any]

initialize(*args, **kwargs)

Return type:

None

soft_reset()

reset_apps(new_apps)

init_and_populate_apps(*args, **kwargs)

Initialize the apps that will be used in the Scenario.

Return type:

None

build_events_flow()

Core logic of the scenario, this is where the scenario is built. Where events are scheduled, event triggers are defined, as well as any element of the task. By default, this function is empty, and should be overridden by the scenario if any extra logic is needed.

Return type:

None

get_app(app_name)

Get the app with the given name

Return type:

App

get_typed_app(app_type, app_name=None)

Get the app with the given type and optional name. If name is not provided, it will be inferred from the app type.

Return type:

TypeVar(T, bound= App)

get_tools_by_app()

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

Return type:

dict[str, list[AppTool]]

get_tools()

Get the entire list of tools from all the apps.

Return type:

list[AppTool]

get_user_tools_by_app()

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()

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

Return type:

list[AppTool]

is_send_message_to_user(event)

Check if the event is a send message to user event

Return type:

bool

build_event_id_to_turn_idx()

Build a dictionary to store the turn of each event The turn of an event is the number of event send_message_to_user among its ancestors The dictionary and the number of turns are stored to the scenario

get_env_tools_by_app()

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

Return type:

dict[str, list[AppTool]]

get_env_tools()

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

Return type:

list[AppTool]

get_data_tools_by_app()

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

Return type:

dict[str, list[AppTool]]

get_data_tools()

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

Return type:

list[AppTool]

edit_hint_content(event_id, content)

Edit the content of a hint.

validate(env)

Validate the state of the environment after the scenario has been executed.

Return type:

ScenarioValidationResult

set_duration(duration)

Set the duration of the scenario.

set_time_increment(time_increment_in_seconds)

Set the time increment of the scenario.

delete_completed_events()

Return type:

None

delete_event(event_id)

Return type:

None

validate_predecessors(predecessor_event_ids, function_name, event_type, events)

filter_connected_events(predecessor_event_ids, events)

Filters out all events connected with the given predecessor_event_ids, including the predecessor events themselves, their successors and dependencies. :type predecessor_event_ids: list[str] :param predecessor_event_ids: List of event IDs to start filtering from. :type events: list[AbstractEvent] :param events: List of all events. :rtype: list[AbstractEvent] :return: List of events excluding those connected with the predecessor_event_ids.

Return type:

list[AbstractEvent]

validate_events_dag_aui_single_branch(predecessor_event_ids, function_name, events, event_id=None)

accumulate_times_from_event(events, event_id=None, new_event_time=None, new_event_relative_time=None)

Return type:

dict[str, float]

validate_events_dag_message_to_user_time(predecessor_event_ids, function_name, new_event_relative_time=None, new_event_time=None, event_id=None)

add_event(app_name, function_name, parameters, predecessor_event_ids, event_type, event_id=None, event_relative_time=None, event_time=None, event_time_comparator=None)

Return type:

AbstractEvent

edit_event(app_name, function_name, parameters, event_id, event_type, predecessor_event_ids, event_relative_time=None, event_time=None, event_time_comparator=None)

Return type:

AbstractEvent

process_events(events)

patch_oracle_user_message_order()

Patches the event dependencies to ensure send_message_to_user events are executed last in each turn.

This method groups events by turns and ensures that send_message_to_user events depend on all other events with maximum accumulated time in the same turn. This guarantees that user messages are sent after all other operations in a turn are completed, maintaining proper execution order in oracle mode.

Return type:

None

apply_augmentation_configs()

class are.simulation.scenarios.scenario.Scenario(_initialized=False, is_benchmark_ready=False, events=<factory>, apps=None, tags=<factory>, scenario_id='', seed=0, nb_turns=None, run_number=None, config=None, has_a2a_augmentation=False, status=ScenarioStatus.Draft, comment=None, annotation_id=None, hints=None, additional_system_prompt=None, start_time=<factory>, duration=None, queue_based_loop=False, time_increment_in_seconds=1, working_dir='', _initial_apps=None, tool_augmentation_config=None, env_events_config=None, gui_config=None, augmentation_data=<factory>)

Bases: object

is_benchmark_ready: bool = False

events: list[AbstractEvent]

apps: list[App] | None = None

tags: tuple[CapabilityTag, ...]

scenario_id: str = ''

seed: int = 0

nb_turns: int | None = None

run_number: int | None = None

config: str | None = None

has_a2a_augmentation: bool = False

status: ScenarioStatus = 'Draft'

comment: str | None = None

annotation_id: str | None = None

hints: list[Hint] | None = None

additional_system_prompt: str | None = None

start_time: float | None

duration: float | None = None

queue_based_loop: bool = False

time_increment_in_seconds: int = 1

working_dir: str = ''

tool_augmentation_config: ToolAugmentationConfig | None = None

env_events_config: EnvEventsConfig | None = None

gui_config: ScenarioGUIConfig | None = None

augmentation_data: dict[str, Any]

initialize(*args, **kwargs)

Return type:

None

soft_reset()

reset_apps(new_apps)

init_and_populate_apps(*args, **kwargs)

Initialize the apps that will be used in the Scenario.

Return type:

None

build_events_flow()

Core logic of the scenario, this is where the scenario is built. Where events are scheduled, event triggers are defined, as well as any element of the task. By default, this function is empty, and should be overridden by the scenario if any extra logic is needed.

Return type:

None

get_app(app_name)

Get the app with the given name

Return type:

App

get_typed_app(app_type, app_name=None)

Get the app with the given type and optional name. If name is not provided, it will be inferred from the app type.

Return type:

TypeVar(T, bound= App)

get_tools_by_app()

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

Return type:

dict[str, list[AppTool]]

get_tools()

Get the entire list of tools from all the apps.

Return type:

list[AppTool]

get_user_tools_by_app()

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()

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

Return type:

list[AppTool]

is_send_message_to_user(event)

Check if the event is a send message to user event

Return type:

bool

build_event_id_to_turn_idx()

Build a dictionary to store the turn of each event The turn of an event is the number of event send_message_to_user among its ancestors The dictionary and the number of turns are stored to the scenario

get_env_tools_by_app()

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

Return type:

dict[str, list[AppTool]]

get_env_tools()

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

Return type:

list[AppTool]

get_data_tools_by_app()

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

Return type:

dict[str, list[AppTool]]

get_data_tools()

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

Return type:

list[AppTool]

edit_hint_content(event_id, content)

Edit the content of a hint.

validate(env)

Validate the state of the environment after the scenario has been executed.

Return type:

ScenarioValidationResult

set_duration(duration)

Set the duration of the scenario.

set_time_increment(time_increment_in_seconds)

Set the time increment of the scenario.

delete_completed_events()

Return type:

None

delete_event(event_id)

Return type:

None

validate_predecessors(predecessor_event_ids, function_name, event_type, events)

filter_connected_events(predecessor_event_ids, events)

Filters out all events connected with the given predecessor_event_ids, including the predecessor events themselves, their successors and dependencies. :type predecessor_event_ids: list[str] :param predecessor_event_ids: List of event IDs to start filtering from. :type events: list[AbstractEvent] :param events: List of all events. :rtype: list[AbstractEvent] :return: List of events excluding those connected with the predecessor_event_ids.

Return type:

list[AbstractEvent]

validate_events_dag_aui_single_branch(predecessor_event_ids, function_name, events, event_id=None)

accumulate_times_from_event(events, event_id=None, new_event_time=None, new_event_relative_time=None)

Return type:

dict[str, float]

validate_events_dag_message_to_user_time(predecessor_event_ids, function_name, new_event_relative_time=None, new_event_time=None, event_id=None)

add_event(app_name, function_name, parameters, predecessor_event_ids, event_type, event_id=None, event_relative_time=None, event_time=None, event_time_comparator=None)

Return type:

AbstractEvent

edit_event(app_name, function_name, parameters, event_id, event_type, predecessor_event_ids, event_relative_time=None, event_time=None, event_time_comparator=None)

Return type:

AbstractEvent

process_events(events)

patch_oracle_user_message_order()

Patches the event dependencies to ensure send_message_to_user events are executed last in each turn.

This method groups events by turns and ensures that send_message_to_user events depend on all other events with maximum accumulated time in the same turn. This guarantees that user messages are sent after all other operations in a turn are completed, maintaining proper execution order in oracle mode.

Return type:

None

apply_augmentation_configs()

__init__(_initialized=False, is_benchmark_ready=False, events=<factory>, apps=None, tags=<factory>, scenario_id='', seed=0, nb_turns=None, run_number=None, config=None, has_a2a_augmentation=False, status=ScenarioStatus.Draft, comment=None, annotation_id=None, hints=None, additional_system_prompt=None, start_time=<factory>, duration=None, queue_based_loop=False, time_increment_in_seconds=1, working_dir='', _initial_apps=None, tool_augmentation_config=None, env_events_config=None, gui_config=None, augmentation_data=<factory>)

class are.simulation.scenarios.validation_result.ScenarioValidationResult(success, exception=None, export_path=None, rationale=None, duration=None)

Bases: object

success: bool | None

exception: Exception | None = None

export_path: str | None = None

rationale: str | None = None

duration: float | None = None

Configuration Classes

class are.simulation.scenarios.config.ScenarioRunnerConfig(**data)

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model: str

model_provider: str | None

agent: str | None

scenario_creation_params: str

scenario_multi_creation_params: str

scenario_initialization_params: str

scenario_multi_initialization_params: str

oracle: bool

export: bool

wait_for_user_input_timeout: float | None

output_dir: str | None

endpoint: str | None

max_turns: int | None

judge_only: bool

a2a_app_prop: float

a2a_app_agent: str

a2a_model: str | None

a2a_model_provider: str | None

a2a_endpoint: str | None

trace_dump_format: str

use_custom_logger: bool

simulated_generation_time_mode: str

tool_augmentation_config: ToolAugmentationConfig | None

env_events_config: EnvEventsConfig | None

judge_engine_config: LLMEngineConfig | None

max_scenario_duration: int

max_time_scenario_duration: int

get_config_hash()

Generate a hash of the relevant config parameters that affect scenario execution. Excludes parameters that only affect parallel execution or logging.

Return type:

str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class are.simulation.scenarios.utils.scenario_expander.EnvEventsConfig(num_env_events_per_minute=10, env_events_seed=0, n_message_events_per_conversation=4, n_item_events_per_product=2, weight_per_app_class=<factory>)

Bases: object

Configuration class for controlling environmental event generation and scheduling in simulations.

This class defines parameters that control how background events (messages, emails, shopping updates, apartment listings) are automatically generated and scheduled during scenario execution. Events are scheduled using a Poisson process to simulate realistic timing patterns.

These environmental events add noise to scenarios by generating synthetic background activity that is unrelated to the main scenario task. This simulates a more realistic environment where agents receive distracting notifications and updates while trying to complete their primary objective.

Attributes:

num_env_events_per_minute: Rate of environmental events to generate per minute of simulation time.

Higher values create more noise and distractions, making the scenario more challenging by increasing the volume of irrelevant background activity.

env_events_seed: Random seed for reproducible event sampling and Poisson scheduling. Different

seeds will generate different patterns of noise events while maintaining the same overall distribution and timing characteristics.

n_message_events_per_conversation: Maximum number of message events to generate per conversation.

Higher values create longer conversation threads in messaging apps, adding more textual noise that the agent must filter through when looking for relevant information.

n_item_events_per_product: Maximum number of item events to generate per shopping product.

Higher values create more product variants and options in shopping apps, increasing the complexity of product catalogs and making it harder to find specific items.

weight_per_app_class: Relative weights for distributing events across different app types.

Adjusting these weights changes which types of noise dominate - higher email weights create more inbox clutter, higher messaging weights create more chat notifications, etc.

num_env_events_per_minute: int = 10

env_events_seed: int = 0

n_message_events_per_conversation: int = 4

n_item_events_per_product: int = 2

weight_per_app_class: dict[str, float]

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

Bases: object

tool_failure_probability: float = 0.1

apply_tool_name_augmentation: bool = True

apply_tool_description_augmentation: bool = True

Utility Classes

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

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

hint_type: HintType

content: str

associated_event_id: str

class are.simulation.types.HintType(value)

Bases: Enum

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

TASK_HINT = 'TASK_HINT'

ENVIRONMENT_HINT = 'ENVIRONMENT_HINT'

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>)

Bases: AbstractEvent

event_type: EventType = 'VALIDATION'

milestones: list[Callable[[AbstractEnvironment], bool]]

minefields: list[Callable[[AbstractEnvironment], bool]]

schedule_every_ticks: int = 1

timeout: int | None = 1

achieved_milestones: list[Callable[[AbstractEnvironment], bool]]

is_timeout()

Return type:

bool

validate(env)

Return type:

tuple[ValidationResult, CompletedEvent]

get_next_event(time_increment_in_seconds=1)

Return type:

ValidationEvent | None

copy()

depends_on(events=None, delay_seconds=0, schedule_every_ticks=None, timeout=None)

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.

schedule(every_ticks, timeout=None)

successors: list[AbstractEvent]

dependencies: list[AbstractEvent]

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)

Bases: AbstractEvent

event_type: EventType = 'VALIDATION'

validators: list[AgentActionValidator]

milestones: list[Callable[[AbstractEnvironment, AbstractEvent], bool]]

minefields: list[Callable[[AbstractEnvironment, AbstractEvent], bool]]

timeout: int | None = None

get_validator()

depends_on(events=None, delay_seconds=0, schedule_every_ticks=None, timeout=None)

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.

schedule(every_ticks, timeout=None)

successors: list[AbstractEvent]

dependencies: list[AbstractEvent]