aea.skills.base
This module contains the base classes for the skills.
SkillContext Objects
class SkillContext()
This class implements the context of a skill.
__
init__
def __init__(agent_context: Optional[AgentContext] = None,
skill: Optional["Skill"] = None) -> None
Initialize a skill context.
Arguments:
agent_context
: the agent context.skill
: the skill.
is_
abstract_
component
@property
def is_abstract_component() -> bool
Get if the skill is abstract.
logger
@property
def logger() -> Logger
Get the logger.
logger
@logger.setter
def logger(logger_: Logger) -> None
Set the logger.
data_
dir
@property
def data_dir() -> str
Get the agent's data directory
set_
agent_
context
def set_agent_context(agent_context: AgentContext) -> None
Set the agent context.
shared_
state
@property
def shared_state() -> Dict[str, Any]
Get the shared state dictionary.
agent_
name
@property
def agent_name() -> str
Get agent name.
skill_
id
@property
def skill_id() -> PublicId
Get the skill id of the skill context.
is_
active
@property
def is_active() -> bool
Get the status of the skill (active/not active).
is_
active
@is_active.setter
def is_active(value: bool) -> None
Set the status of the skill (active/not active).
new_
behaviours
@property
def new_behaviours() -> "Queue[Behaviour]"
Queue for the new behaviours.
This queue can be used to send messages to the framework to request the registration of a behaviour.
Returns:
the queue of new behaviours.
new_
handlers
@property
def new_handlers() -> "Queue[Handler]"
Queue for the new handlers.
This queue can be used to send messages to the framework to request the registration of a handler.
Returns:
the queue of new handlers.
agent_
addresses
@property
def agent_addresses() -> Dict[str, str]
Get addresses.
agent_
address
@property
def agent_address() -> str
Get address.
public_
key
@property
def public_key() -> str
Get public key.
public_
keys
@property
def public_keys() -> Dict[str, str]
Get public keys.
connection_
status
@property
def connection_status() -> MultiplexerStatus
Get connection status.
outbox
@property
def outbox() -> OutBox
Get outbox.
storage
@property
def storage() -> Optional[Storage]
Get optional storage for agent.
message_
in_
queue
@property
def message_in_queue() -> Queue
Get message in queue.
decision_
maker_
message_
queue
@property
def decision_maker_message_queue() -> Queue
Get message queue of decision maker.
decision_
maker_
handler_
context
@property
def decision_maker_handler_context() -> SimpleNamespace
Get decision maker handler context.
task_
manager
@property
def task_manager() -> TaskManager
Get behaviours of the skill.
default_
ledger_
id
@property
def default_ledger_id() -> str
Get the default ledger id.
currency_
denominations
@property
def currency_denominations() -> Dict[str, str]
Get a dictionary mapping ledger ids to currency denominations.
search_
service_
address
@property
def search_service_address() -> Address
Get the address of the search service.
decision_
maker_
address
@property
def decision_maker_address() -> Address
Get the address of the decision maker.
handlers
@property
def handlers() -> SimpleNamespace
Get handlers of the skill.
behaviours
@property
def behaviours() -> SimpleNamespace
Get behaviours of the skill.
namespace
@property
def namespace() -> SimpleNamespace
Get the agent context namespace.
__
getattr__
def __getattr__(item: Any) -> Any
Get attribute.
send_
to_
skill
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
: the optional envelope context
SkillComponent Objects
class SkillComponent(ABC)
This class defines an abstract interface for skill component classes.
__
init__
def __init__(name: str,
skill_context: SkillContext,
configuration: Optional[SkillComponentConfiguration] = None,
**kwargs: Any) -> None
Initialize a skill component.
Arguments:
name
: the name of the component.configuration
: the configuration for the component.skill_context
: the skill context.kwargs
: the keyword arguments.
name
@property
def name() -> str
Get the name of the skill component.
context
@property
def context() -> SkillContext
Get the context of the skill component.
skill_
id
@property
def skill_id() -> PublicId
Get the skill id of the skill component.
configuration
@property
def configuration() -> SkillComponentConfiguration
Get the skill component configuration.
config
@property
def config() -> Dict[Any, Any]
Get the config of the skill component.
setup
@abstractmethod
def setup() -> None
Implement the setup.
teardown
@abstractmethod
def teardown() -> None
Implement the teardown.
parse_
module
@classmethod
@abstractmethod
def parse_module(cls, path: str, configs: Dict[str,
SkillComponentConfiguration],
skill_context: SkillContext) -> dict
Parse the component module.
AbstractBehaviour Objects
class AbstractBehaviour(SkillComponent, ABC)
Abstract behaviour for periodical calls.
tick_interval: float, interval to call behaviour's act. start_at: optional datetime, when to start periodical calls.
tick_
interval
@property
def tick_interval() -> float
Get the tick_interval in seconds.
start_
at
@property
def start_at() -> Optional[datetime.datetime]
Get the start time of the behaviour.
Behaviour Objects
class Behaviour(AbstractBehaviour, ABC)
This class implements an abstract behaviour.
In a subclass of Behaviour, the flag 'is_programmatically_defined' can be used by the developer to signal to the framework that the class is meant to be used programmatically; hence, in case the class is not declared in the configuration file but it is present in a skill module, the framework will just ignore this class instead of printing a warning message.
act
@abstractmethod
def act() -> None
Implement the behaviour.
Returns:
None
is_
done
def is_done() -> bool
Return True if the behaviour is terminated, False otherwise.
act_
wrapper
def act_wrapper() -> None
Wrap the call of the action. This method must be called only by the framework.
parse_
module
@classmethod
def parse_module(cls, path: str,
behaviour_configs: Dict[str, SkillComponentConfiguration],
skill_context: SkillContext) -> Dict[str, "Behaviour"]
Parse the behaviours module.
Arguments:
path
: path to the Python module containing the Behaviour classes.behaviour_configs
: a list of behaviour configurations.skill_context
: the skill context
Returns:
a list of Behaviour.
Handler Objects
class Handler(SkillComponent, ABC)
This class implements an abstract behaviour.
In a subclass of Handler, the flag 'is_programmatically_defined' can be used by the developer to signal to the framework that the component is meant to be used programmatically; hence, in case the class is not declared in the configuration file but it is present in a skill module, the framework will just ignore this class instead of printing a warning message.
SUPPORTED_PROTOCOL is read by the framework when the handlers are loaded to register them as 'listeners' to the protocol identified by the specified public id. Whenever a message of protocol 'SUPPORTED_PROTOCOL' is sent to the agent, the framework will call the 'handle' method.
handle
@abstractmethod
def handle(message: Message) -> None
Implement the reaction to a message.
Arguments:
message
: the message
Returns:
None
handle_
wrapper
def handle_wrapper(message: Message) -> None
Wrap the call of the handler. This method must be called only by the framework.
parse_
module
@classmethod
def parse_module(cls, path: str,
handler_configs: Dict[str, SkillComponentConfiguration],
skill_context: SkillContext) -> Dict[str, "Handler"]
Parse the handler module.
Arguments:
path
: path to the Python module containing the Handler class.handler_configs
: the list of handler configurations.skill_context
: the skill context
Returns:
an handler, or None if the parsing fails.
protocol_
dialogues
def protocol_dialogues(attribute: Optional[str] = None)
Protocol dialogues.
This method must NOT be called by the framework with exception handling. It assumes a user-behaviour whereby dialogues are stored under specifically named attributes.
Arguments:
attribute
: attribute under which dialogue is stored
Returns:
dialogue
Model Objects
class Model(SkillComponent, ABC)
This class implements an abstract model.
__
init__
def __init__(name: str,
skill_context: SkillContext,
configuration: Optional[SkillComponentConfiguration] = None,
keep_terminal_state_dialogues: Optional[bool] = None,
**kwargs: Any) -> None
Initialize a model.
Arguments:
name
: the name of the component.configuration
: the configuration for the component.skill_context
: the skill context.keep_terminal_state_dialogues
: specify do dialogues in terminal state should stay or notkwargs
: the keyword arguments.
setup
def setup() -> None
Set the class up.
teardown
def teardown() -> None
Tear the class down.
parse_
module
@classmethod
def parse_module(cls, path: str,
model_configs: Dict[str, SkillComponentConfiguration],
skill_context: SkillContext) -> Dict[str, "Model"]
Parse the model module.
Arguments:
path
: path to the Python skill module.model_configs
: a list of model configurations.skill_context
: the skill context
Returns:
a list of Model.
Skill Objects
class Skill(Component)
This class implements a skill.
__
init__
def __init__(configuration: SkillConfig,
skill_context: Optional[SkillContext] = None,
handlers: Optional[Dict[str, Handler]] = None,
behaviours: Optional[Dict[str, Behaviour]] = None,
models: Optional[Dict[str, Model]] = None,
**kwargs: Any)
Initialize a skill.
Arguments:
configuration
: the skill configuration.skill_context
: the skill context.handlers
: dictionary of handlers.behaviours
: dictionary of behaviours.models
: dictionary of models.kwargs
: the keyword arguments.
skill_
context
@property
def skill_context() -> SkillContext
Get the skill context.
handlers
@property
def handlers() -> Dict[str, Handler]
Get the handlers.
behaviours
@property
def behaviours() -> Dict[str, Behaviour]
Get the handlers.
models
@property
def models() -> Dict[str, Model]
Get the handlers.
from_
dir
@classmethod
def from_dir(cls, directory: str, agent_context: AgentContext,
**kwargs: Any) -> "Skill"
Load the skill from a directory.
Arguments:
directory
: the directory to the skill package.agent_context
: the skill context.kwargs
: the keyword arguments.
Returns:
the skill object.
logger
@property
def logger() -> Logger
Get the logger.
In the case of a skill, return the logger provided by the skill context.
Returns:
the logger
logger
@logger.setter
def logger(*args: str) -> None
Set the logger.
from_
config
@classmethod
def from_config(cls, configuration: SkillConfig, agent_context: AgentContext,
**kwargs: Any) -> "Skill"
Load the skill from configuration.
Arguments:
configuration
: a skill configuration. Must be associated with a directory.agent_context
: the agent context.kwargs
: the keyword arguments.
Returns:
the skill.
_
SkillComponentLoadingItem Objects
class _SkillComponentLoadingItem()
Class to represent a triple (component name, component configuration, component class).
__
init__
def __init__(name: str, config: SkillComponentConfiguration,
class_: Type[SkillComponent], type_: _SKILL_COMPONENT_TYPES)
Initialize the item.
_
SkillComponentLoader Objects
class _SkillComponentLoader()
This class implements the loading policy for skill components.
__
init__
def __init__(configuration: SkillConfig, skill_context: SkillContext,
**kwargs: Any)
Initialize the helper class.
load_
skill
def load_skill() -> Skill
Load the skill.