This module contains the implementation of an agent loop using asyncio.
class AgentLoopException(AEAException)
Exception for agent loop runtime errors.
class AgentLoopStates(Enum)
Internal agent loop states.
class BaseAgentLoop(Runnable, WithLogger, ABC)
Base abstract agent loop class.
def __init__(agent: AbstractAgent,
loop: Optional[AbstractEventLoop] = None,
threaded: bool = False) -> None
Init loop.
Arguments:
agent
: Agent or AEA to run.loop
: optional asyncio event loop. if not specified a new loop will be created.threaded
: if True, run in threaded mode, else async
@property
def agent() -> AbstractAgent
Get agent.
@property
def state() -> AgentLoopStates
Get current main loop state.
async def wait_state(
state_or_states: Union[Any, Sequence[Any]]) -> Tuple[Any, Any]
Wait state to be set.
Arguments:
state_or_states
: state or list of states.
Returns:
tuple of previous state and new state.
@property
def is_running() -> bool
Get running state of the loop.
def set_loop(loop: AbstractEventLoop) -> None
Set event loop and all event loop related objects.
async def run() -> None
Run agent loop.
@abstractmethod
def send_to_skill(message_or_envelope: Union[Message, Envelope],
context: Optional[EnvelopeContext] = None) -> None
Send message or envelope to another skill.
If message passed it will be wrapped into envelope with optional envelope context.
Arguments:
message_or_envelope
: envelope to send to another skill.context
: envelope context
@property
@abstractmethod
def skill2skill_queue() -> Queue
Get skill to skill message queue.
class AsyncAgentLoop(BaseAgentLoop)
Asyncio based agent loop suitable only for AEA.
def __init__(agent: AbstractAgent,
loop: AbstractEventLoop = None,
threaded: bool = False) -> None
Init agent loop.
Arguments:
agent
: AEA instanceloop
: asyncio loop to use. optionalthreaded
: is a new thread to be started for the agent loop
@property
def skill2skill_queue() -> Queue
Get skill to skill message queue.
def send_to_skill(message_or_envelope: Union[Message, Envelope],
context: Optional[EnvelopeContext] = None) -> None
Send message or envelope to another skill.
If message passed it will be wrapped into envelope with optional envelope context.
Arguments:
message_or_envelope
: envelope to send to another skill.context
: envelope context