agentscope.environment package

Submodules

• agentscope.environment.env module
  • trigger_listener()
  • event_func()
  • EventListener
    • EventListener.__init__()
  • Env
    • Env.name
    • Env.get_children()
    • Env.add_child()
    • Env.remove_child()
    • Env.add_listener()
    • Env.remove_listener()
    • Env.get_listeners()
    • Env.describe()
    • Env.to_dist()
  • BasicEnv
    • BasicEnv.__init__()
    • BasicEnv.name
    • BasicEnv.get_children()
    • BasicEnv.add_child()
    • BasicEnv.remove_child()
    • BasicEnv.add_listener()
    • BasicEnv.remove_listener()
    • BasicEnv.get_listeners()
    • BasicEnv.describe()
    • BasicEnv.to_dist()
• agentscope.environment.event module
  • Event
    • Event.__init__()
    • Event.name
    • Event.args
    • Event.returns
  • Getable
    • Getable.get()
  • Setable
    • Setable.set()
  • Movable2D
    • Movable2D.move_by()
    • Movable2D.move_to()
    • Movable2D.get_position()
  • Holdable
    • Holdable.acquire()
    • Holdable.release()

Module contents

Import all environment related modules in the package.

class agentscope.environment.Event(name: str, args: dict | None = None, returns: Any | None = None)

Bases: object

A class representing the information of an event.

It contains the name of the event, the arguments of the event, and the returns of the event.

__init__(name: str, args: dict | None = None, returns: Any | None = None) → None

property name: str

Return the name of the event.

property args: dict

Return the arguments of the event.

property returns: Any

Return the returns of the event.

agentscope.environment.event_func(func: Callable) → Callable

A decorator to register an event function in Env and its subclasses.

Note

This decorator is only available in the subclasses of Env. If a function is decorated with @event_func, at the end of the function, all listeners bound to the function will be triggered automatically.

Parameters:

func (Callable) – The event function.

Returns:

The decorated event function.

Return type:

Callable

class agentscope.environment.Env(*args: tuple, **kwargs: dict)

Bases: ABC

The Env Interface. Env is a key concept of AgentScope, representing global data shared among agents.

Each env has its own name and value, and multiple envs can be organized into a tree structure, where each env can have multiple children envs.

Different implementations of envs may have different event functions, which are marked by @event_func. Users can bind EventListener to specific event functions, and the listener will be activated when the event function is called.

abstract property name: str

Name of the env.

Returns:

The name of the env.

Return type:

str

abstract get_children() → dict[str, Env]

Get the children envs of the current env.

Returns:

The children envs.

Return type:

dict[str, Env]

abstract add_child(child: Env) → bool

Add a child env to the current env.

Parameters:

child (Env) – The children envs.

Returns:

Whether the children were added successfully.

Return type:

bool

abstract remove_child(children_name: str) → bool

Remove a child env from the current env.

Parameters:

children_name (str) – The name of the children env.

Returns:

Whether the children were removed successfully.

Return type:

bool

abstract add_listener(target_event: str, listener: EventListener) → bool

Add a listener to the env.

Parameters:

• target_event (str) – The event function to listen.

• listener (EventListener) – The listener to add.

Returns:

Whether the listener was added successfully.

Return type:

bool

abstract remove_listener(target_event: str, listener_name: str) → bool

Remove a listener from the env.

Parameters:

• target_event (str) – The event function.

• listener_name (str) – The name of the listener to remove.

Returns:

Whether the listener was removed successfully.

Return type:

bool

abstract get_listeners(target_event: str) → List[EventListener]

Get the listeners of the specific event.

Parameters:

target_event (str) – The event name.

Returns:

The listeners of the specific event.

Return type:

List[EventListener]

abstract describe(**kwargs: Any) → str

Describe the current state of the environment.

to_dist(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_expire_time: int = 7200, max_timeout_seconds: int = 5, local_mode: bool = True, retry_strategy: ~agentscope.rpc.retry_strategy.RetryBase = <agentscope.rpc.retry_strategy.RetryFixedTimes object>) → Any

Convert current object into its distributed version.

Parameters:

• host (str, defaults to “localhost”) – Hostname of the rpc agent server.

• port (int, defaults to None) – Port of the rpc agent server.

• max_pool_size (int, defaults to 8192) – Only takes effect when host and port are not filled in. The max number of agent reply messages that the started agent server can accommodate. Note that the oldest message will be deleted after exceeding the pool size.

• max_expire_time (int, defaults to 7200) – Only takes effect when host and port are not filled in. Maximum time for reply messages to be cached in the launched agent server. Note that expired messages will be deleted.

• max_timeout_seconds (int, defaults to 5) – Max timeout seconds for the rpc call.

• local_mode (bool, defaults to True) – Only takes effect when host and port are not filled in. Whether the started agent server only listens to local requests.

• retry_strategy (RetryBase, defaults to _DEAFULT_RETRY_STRATEGY) – The retry strategy for the async rpc call.

Returns:

the wrapped agent instance with distributed functionality

Return type:

RpcObject

class agentscope.environment.BasicEnv(*args: tuple, **kwargs: dict)

Bases: Env

A basic implementation of Env, which has no event function and cannot get value.

Note

BasicEnv is used as the base class to implement other envs. Application developers should not use this class.

__init__(name: str, listeners: dict[str, List[EventListener]] | None = None, children: List[Env] | None = None) → None

Init an BasicEnv instance.

Parameters:

• name (str) – The name of the env.

• listeners (dict[str, List[EventListener]], optional) – The

• None. (envs. Defaults to)

• children (List[Env], optional) – A list of children

• None.

property name: str

Name of the env

get_children() → dict[str, Env]

Get the children envs of the current env.

Returns:

The children envs.

Return type:

dict[str, Env]

add_child(child: Env) → bool

Add a child env to the current env.

Parameters:

• child (Env) – The children

• envs.

Returns:

Whether the children were added successfully.

Return type:

bool

remove_child(children_name: str) → bool

Remove a child env from the current env.

Parameters:

children_name (str) – The name of the children env.

Returns:

Whether the children were removed successfully.

Return type:

bool

add_listener(target_event: str, listener: EventListener) → bool

Add a listener to the env.

Parameters:

• target_event (str) – The name of the event to listen.

• listener (EventListener) – The listener to add.

Returns:

Whether the listener was added successfully.

Return type:

bool

remove_listener(target_event: str, listener_name: str) → bool

Remove a listener from the env.

Parameters:

• target_event (str) – The event name.

• listener_name (str) – The name of the listener to remove.

Returns:

Whether the listener was removed successfully.

Return type:

bool

get_listeners(target_event: str) → List[EventListener]

Get the listeners of the specific event.

Parameters:

target_event (str) – The event name.

Returns:

The listeners of the specific event.

Return type:

List[EventListener]

describe(**kwargs: Any) → str

Describe the current state of the environment.

to_dist(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_expire_time: int = 7200, max_timeout_seconds: int = 5, local_mode: bool = True, retry_strategy: ~agentscope.rpc.retry_strategy.RetryBase = <agentscope.rpc.retry_strategy.RetryFixedTimes object>) → Any

Convert current object into its distributed version.

Parameters:

• host (str, defaults to “localhost”) – Hostname of the rpc agent server.

• port (int, defaults to None) – Port of the rpc agent server.

• max_pool_size (int, defaults to 8192) – Only takes effect when host and port are not filled in. The max number of agent reply messages that the started agent server can accommodate. Note that the oldest message will be deleted after exceeding the pool size.

• max_expire_time (int, defaults to 7200) – Only takes effect when host and port are not filled in. Maximum time for reply messages to be cached in the launched agent server. Note that expired messages will be deleted.

• max_timeout_seconds (int, defaults to 5) – Max timeout seconds for the rpc call.

• local_mode (bool, defaults to True) – Only takes effect when host and port are not filled in. Whether the started agent server only listens to local requests.

• retry_strategy (RetryBase, defaults to _DEAFULT_RETRY_STRATEGY) – The retry strategy for the async rpc call.

Returns:

the wrapped agent instance with distributed functionality

Return type:

RpcObject

class agentscope.environment.EventListener(name: str)

Bases: ABC

A base class representing a listener for listening the event of an environment. The actions of the listener should be implemented in the __call__ method.

Parameters:

• env (Env) – The environment instance who trigger the listener.

• event (Event) – The event information, which contains the event

• (name (function name) – str), the arguments of the event function

• (args – dict), and the return value of the event function

• (returns – Any).

Note

EventListener can only be bound to event functions (decorated with @event_func).

__init__(name: str) → None

Init a EventListener instance.

Parameters:

name (str) – The name of the listener.