Skip to content

aea.runtime

This module contains the implementation of runtime for economic agent (AEA).

RuntimeStates Objects

class RuntimeStates(Enum)

Runtime states.

BaseRuntime Objects

class BaseRuntime(Runnable, WithLogger)

Abstract runtime class to create implementations.

__init__

def __init__(agent: AbstractAgent,
             multiplexer_options: Dict,
             loop_mode: Optional[str] = None,
             loop: Optional[AbstractEventLoop] = None,
             threaded: bool = False,
             task_manager_mode: Optional[str] = None) -> None

Init runtime.

Arguments:

  • agent: Agent to run.
  • multiplexer_options: options for the multiplexer.
  • loop_mode: agent main loop mode.
  • loop: optional event loop. if not provided a new one will be created.
  • threaded: if True, run in threaded mode, else async
  • task_manager_mode: mode of the task manager.

storage

@property
def storage() -> Optional[Storage]

Get optional storage.

loop_mode

@property
def loop_mode() -> str

Get current loop mode.

task_manager

@property
def task_manager() -> TaskManager

Get the task manager.

loop

@property
def loop() -> Optional[AbstractEventLoop]

Get event loop.

agent_loop

@property
def agent_loop() -> BaseAgentLoop

Get the agent loop.

multiplexer

@property
def multiplexer() -> AsyncMultiplexer

Get multiplexer.

is_running

@property
def is_running() -> bool

Get running state of the runtime.

is_stopped

@property
def is_stopped() -> bool

Get stopped state of the runtime.

state

@property
def state() -> RuntimeStates

Get runtime state.

Returns:

RuntimeStates

decision_maker

@property
def decision_maker() -> DecisionMaker

Return decision maker if set.

set_decision_maker

def set_decision_maker(decision_maker_handler: DecisionMakerHandler) -> None

Set decision maker with handler provided.

set_loop

def set_loop(loop: AbstractEventLoop) -> None

Set event loop to be used.

Arguments:

  • loop: event loop to use.

AsyncRuntime Objects

class AsyncRuntime(BaseRuntime)

Asynchronous runtime: uses asyncio loop for multiplexer and async agent main loop.

__init__

def __init__(agent: AbstractAgent,
             multiplexer_options: Dict,
             loop_mode: Optional[str] = None,
             loop: Optional[AbstractEventLoop] = None,
             threaded: bool = False,
             task_manager_mode: Optional[str] = None) -> None

Init runtime.

Arguments:

  • agent: Agent to run.
  • multiplexer_options: options for the multiplexer.
  • loop_mode: agent main loop mode.
  • loop: optional event loop. if not provided a new one will be created.
  • threaded: if True, run in threaded mode, else async
  • task_manager_mode: mode of the task manager.

set_loop

def set_loop(loop: AbstractEventLoop) -> None

Set event loop to be used.

Arguments:

  • loop: event loop to use.

run

async def run() -> None

Start runtime task.

Starts multiplexer and agent loop.

stop_runtime

async def stop_runtime() -> None

Stop runtime coroutine.

Stop main loop. Tear down the agent.. Disconnect multiplexer.

run_runtime

async def run_runtime() -> None

Run runtime which means start agent loop, multiplexer and storage.

ThreadedRuntime Objects

class ThreadedRuntime(AsyncRuntime)

Run agent and multiplexer in different threads with own asyncio loops.