aea.protocols.dialogue.base
This module contains the classes required for dialogue management.
- DialogueLabel: The dialogue label class acts as an identifier for dialogues.
- Dialogue: The dialogue class maintains state of a dialogue and manages it.
- Dialogues: The dialogues class keeps track of all dialogues.
InvalidDialogueMessage Objects
class InvalidDialogueMessage(Exception)
Exception for adding invalid message to a dialogue.
DialogueLabel Objects
class DialogueLabel()
The dialogue label class acts as an identifier for dialogues.
__
init__
| __init__(dialogue_reference: Tuple[str, str], dialogue_opponent_addr: Address, dialogue_starter_addr: Address) -> None
Initialize a dialogue label.
Arguments:
dialogue_reference
: the reference of the dialogue.dialogue_opponent_addr
: the addr of the agent with which the dialogue is kept.dialogue_starter_addr
: the addr of the agent which started the dialogue.
Returns:
None
dialogue_
reference
| @property
| dialogue_reference() -> Tuple[str, str]
Get the dialogue reference.
dialogue_
starter_
reference
| @property
| dialogue_starter_reference() -> str
Get the dialogue starter reference.
dialogue_
responder_
reference
| @property
| dialogue_responder_reference() -> str
Get the dialogue responder reference.
dialogue_
opponent_
addr
| @property
| dialogue_opponent_addr() -> str
Get the address of the dialogue opponent.
dialogue_
starter_
addr
| @property
| dialogue_starter_addr() -> str
Get the address of the dialogue starter.
__
eq__
| __eq__(other) -> bool
Check for equality between two DialogueLabel objects.
__
hash__
| __hash__() -> int
Turn object into hash.
json
| @property
| json() -> Dict
Return the JSON representation.
from_
json
| @classmethod
| from_json(cls, obj: Dict[str, str]) -> "DialogueLabel"
Get dialogue label from json.
get_
incomplete_
version
| get_incomplete_version() -> "DialogueLabel"
Get the incomplete version of the label.
__
str__
| __str__()
Get the string representation.
from_
str
| @classmethod
| from_str(cls, obj: str) -> "DialogueLabel"
Get the dialogue label from string representation.
_
DialogueMeta Objects
class _DialogueMeta(type)
Metaclass for Dialogue.
Adds slot support forevery subclass Creates classlevvel Rules instance
__
new__
| __new__(cls, name: str, bases: Tuple[Type], dct: Dict)
Construct a new type.
Dialogue Objects
class Dialogue(, metaclass=_DialogueMeta)
The dialogue class maintains state of a dialogue and manages it.
Rules Objects
class Rules()
This class defines the rules for the dialogue.
__
init__
| __init__(initial_performatives: FrozenSet[Message.Performative], terminal_performatives: FrozenSet[Message.Performative], valid_replies: Dict[Message.Performative, FrozenSet[Message.Performative]]) -> None
Initialize a dialogue.
Arguments:
initial_performatives
: the set of all initial performatives.terminal_performatives
: the set of all terminal performatives.valid_replies
: the reply structure of speech-acts.
Returns:
None
initial_
performatives
| @property
| initial_performatives() -> FrozenSet[Message.Performative]
Get the performatives one of which the terminal message in the dialogue must have.
Returns:
the valid performatives of an terminal message
terminal_
performatives
| @property
| terminal_performatives() -> FrozenSet[Message.Performative]
Get the performatives one of which the terminal message in the dialogue must have.
Returns:
the valid performatives of an terminal message
valid_
replies
| @property
| valid_replies() -> Dict[Message.Performative, FrozenSet[Message.Performative]]
Get all the valid performatives which are a valid replies to performatives.
Returns:
the full valid reply structure.
get_
valid_
replies
| get_valid_replies(performative: Message.Performative) -> FrozenSet[Message.Performative]
Given a performative
, return the list of performatives which are its valid replies in a dialogue.
Arguments:
performative
: the performative in a message
Returns:
list of valid performative replies
Role Objects
class Role(Enum)
This class defines the agent's role in a dialogue.
__
str__
| __str__()
Get the string representation.
EndState Objects
class EndState(Enum)
This class defines the end states of a dialogue.
__
str__
| __str__()
Get the string representation.
__
init__
| __init__(dialogue_label: DialogueLabel, message_class: Type[Message], self_address: Address, role: Role) -> None
Initialize a dialogue.
Arguments:
dialogue_label
: the identifier of the dialogueself_address
: the address of the entity for whom this dialogue is maintainedrole
: the role of the agent this dialogue is maintained for
Returns:
None
add_
terminal_
state_
callback
| add_terminal_state_callback(fn: Callable[["Dialogue"], None]) -> None
Add callback to be called on dialogue reach terminal state.
Arguments:
fn
: callable to be called with one argument: Dialogue
Returns:
None
__
eq__
| __eq__(other) -> bool
Compare two dialogues.
json
| json() -> dict
Get json representation of the dialogue.
from_
json
| @classmethod
| from_json(cls, message_class: Type[Message], data: dict) -> "Dialogue"
Create a dialogue instance with all messages from json data.
Arguments:
message_class
: type of message used with this dialoguedata
: dict with data exported with Dialogue.to_json() method
Returns:
Dialogue instance
dialogue_
label
| @property
| dialogue_label() -> DialogueLabel
Get the dialogue label.
Returns:
The dialogue label
incomplete_
dialogue_
label
| @property
| incomplete_dialogue_label() -> DialogueLabel
Get the dialogue label.
Returns:
The incomplete dialogue label
dialogue_
labels
| @property
| dialogue_labels() -> Set[DialogueLabel]
Get the dialogue labels (incomplete and complete, if it exists)
Returns:
the dialogue labels
self_
address
| @property
| self_address() -> Address
Get the address of the entity for whom this dialogues is maintained.
Returns:
the address of this entity
role
| @property
| role() -> "Role"
Get the agent's role in the dialogue.
Returns:
the agent's role
rules
| @property
| rules() -> "Rules"
Get the dialogue rules.
Returns:
the rules
message_
class
| @property
| message_class() -> Type[Message]
Get the message class.
Returns:
the message class
is_
self_
initiated
| @property
| is_self_initiated() -> bool
Check whether the agent initiated the dialogue.
Returns:
True if the agent initiated the dialogue, False otherwise
last_
incoming_
message
| @property
| last_incoming_message() -> Optional[Message]
Get the last incoming message.
Returns:
the last incoming message if it exists, None otherwise
last_
outgoing_
message
| @property
| last_outgoing_message() -> Optional[Message]
Get the last outgoing message.
Returns:
the last outgoing message if it exists, None otherwise
last_
message
| @property
| last_message() -> Optional[Message]
Get the last message.
Returns:
the last message if it exists, None otherwise
is_
empty
| @property
| is_empty() -> bool
Check whether the dialogue is empty.
Returns:
True if empty, False otherwise
reply
| reply(performative: Message.Performative, target_message: Optional[Message] = None, target: Optional[int] = None, **kwargs, ,) -> Message
Reply to the 'target_message' in this dialogue with a message with 'performative', and contents from kwargs.
Note if no target_message is provided, the last message in the dialogue will be replied to.
Arguments:
target_message
: the message to reply to.target
: the id of the message to reply to.performative
: the performative of the reply message.kwargs
: the content of the reply message.
Returns:
the reply message if it was successfully added as a reply, None otherwise.
__
str__
| __str__() -> str
Get the string representation.
Returns:
The string representation of the dialogue
DialogueStats Objects
class DialogueStats()
Class to handle statistics on default dialogues.
__
init__
| __init__(end_states: FrozenSet[Dialogue.EndState]) -> None
Initialize a StatsManager.
Arguments:
end_states
: the list of dialogue endstates
self_
initiated
| @property
| self_initiated() -> Dict[Dialogue.EndState, int]
Get the stats dictionary on self initiated dialogues.
other_
initiated
| @property
| other_initiated() -> Dict[Dialogue.EndState, int]
Get the stats dictionary on other initiated dialogues.
add_
dialogue_
endstate
| add_dialogue_endstate(end_state: Dialogue.EndState, is_self_initiated: bool) -> None
Add dialogue endstate stats.
Arguments:
end_state
: the end state of the dialogueis_self_initiated
: whether the dialogue is initiated by the agent or the opponent
Returns:
None
find_
caller_
object
find_caller_object(object_type: Type)
Find caller object of certain type in the call stack.
BasicDialoguesStorage Objects
class BasicDialoguesStorage()
Dialogues state storage.
__
init__
| __init__(dialogues: "Dialogues") -> None
Init dialogues storage.
dialogues_
in_
terminal_
state
| @property
| dialogues_in_terminal_state() -> List["Dialogue"]
Get all dialogues in terminal state.
dialogues_
in_
active_
state
| @property
| dialogues_in_active_state() -> List["Dialogue"]
Get all dialogues in active state.
is_
terminal_
dialogues_
kept
| @property
| is_terminal_dialogues_kept() -> bool
Return True if dialogues should stay after terminal state.
dialogue_
terminal_
state_
callback
| dialogue_terminal_state_callback(dialogue: "Dialogue") -> None
Method to be called on dialogue terminal state reached.
setup
| setup() -> None
Set up dialogue storage.
teardown
| teardown() -> None
Tear down dialogue storage.
add
| add(dialogue: Dialogue) -> None
Add dialogue to storage.
Arguments:
dialogue
: dialogue to add.
Returns:
None
remove
| remove(dialogue_label: DialogueLabel) -> None
Remove dialogue from storage by it's label.
Arguments:
dialogue_label
: label of the dialogue to remove
Returns:
None
get
| get(dialogue_label: DialogueLabel) -> Optional[Dialogue]
Get dialogue stored by it's label.
Arguments:
dialogue_label
: label of the dialogue
Returns:
dialogue if presents or None
get_
dialogues_
with_
counterparty
| get_dialogues_with_counterparty(counterparty: Address) -> List[Dialogue]
Get the dialogues by address.
Arguments:
counterparty
: the counterparty
Returns:
The dialogues with the counterparty.
is_
in_
incomplete
| is_in_incomplete(dialogue_label: DialogueLabel) -> bool
Check dialogue label presents in list of incomplete.
set_
incomplete_
dialogue
| set_incomplete_dialogue(incomplete_dialogue_label, complete_dialogue_label) -> None
Set incomplete dialogue label.
is_
dialogue_
present
| is_dialogue_present(dialogue_label: DialogueLabel) -> bool
Check dialogue with label specified presents in storage.
get_
latest_
label
| get_latest_label(dialogue_label: DialogueLabel) -> DialogueLabel
Get latest label for dialogue.
PersistDialoguesStorage Objects
class PersistDialoguesStorage(BasicDialoguesStorage)
Persist dialogues storage.
Uses generic storage to load/save dialogues data on setup/teardown.
__
init__
| __init__(dialogues: "Dialogues") -> None
Init dialogues storage.
get_
skill_
component
| @staticmethod
| get_skill_component() -> Optional[SkillComponent]
Get skill component dialogues storage constructed for.
setup
| setup() -> None
Set up dialogue storage.
teardown
| teardown() -> None
Tear down dialogue storage.
remove
| remove(dialogue_label: DialogueLabel) -> None
Remove dialogue from memory and persistent storage.
PersistDialoguesStorageWithOffloading Objects
class PersistDialoguesStorageWithOffloading(PersistDialoguesStorage)
Dialogue Storage with dialogues offloading.
dialogue_
terminal_
state_
callback
| dialogue_terminal_state_callback(dialogue: "Dialogue") -> None
Call on dialogue reaches terminal staste.
get
| get(dialogue_label: DialogueLabel) -> Optional[Dialogue]
Try to get dialogue by label from memory or persists storage.
get_
dialogues_
with_
counterparty
| get_dialogues_with_counterparty(counterparty: Address) -> List[Dialogue]
Get the dialogues by address.
Arguments:
counterparty
: the counterparty
Returns:
The dialogues with the counterparty.
dialogues_
in_
terminal_
state
| @property
| dialogues_in_terminal_state() -> List["Dialogue"]
Get all dialogues in terminal state.
Dialogues Objects
class Dialogues()
The dialogues class keeps track of all dialogues for an agent.
__
init__
| __init__(self_address: Address, end_states: FrozenSet[Dialogue.EndState], message_class: Type[Message], dialogue_class: Type[Dialogue], role_from_first_message: Callable[[Message, Address], Dialogue.Role], keep_terminal_state_dialogues: Optional[bool] = None) -> None
Initialize dialogues.
Arguments:
self_address
: the address of the entity for whom dialogues are maintainedend_states
: the list of dialogue endstateskeep_terminal_state_dialogues
: specify do dialogues in terminal state should stay or not
Returns:
None
is_
keep_
dialogues_
in_
terminal_
state
| @property
| is_keep_dialogues_in_terminal_state() -> bool
Is requrired to keep dialogues in terminal state.
self_
address
| @property
| self_address() -> Address
Get the address of the agent for whom dialogues are maintained.
dialogue_
stats
| @property
| dialogue_stats() -> DialogueStats
Get the dialogue statistics.
Returns:
dialogue stats object
message_
class
| @property
| message_class() -> Type[Message]
Get the message class.
Returns:
the message class
dialogue_
class
| @property
| dialogue_class() -> Type[Dialogue]
Get the dialogue class.
Returns:
the dialogue class
get_
dialogues_
with_
counterparty
| get_dialogues_with_counterparty(counterparty: Address) -> List[Dialogue]
Get the dialogues by address.
Arguments:
counterparty
: the counterparty
Returns:
The dialogues with the counterparty.
new_
self_
initiated_
dialogue_
reference
| @classmethod
| new_self_initiated_dialogue_reference(cls) -> Tuple[str, str]
Return a dialogue label for a new self initiated dialogue.
Returns:
the next nonce
create
| create(counterparty: Address, performative: Message.Performative, **kwargs, ,) -> Tuple[Message, Dialogue]
Create a dialogue with 'counterparty', with an initial message whose performative is 'performative' and contents are from 'kwargs'.
Arguments:
counterparty
: the counterparty of the dialogue.performative
: the performative of the initial message.kwargs
: the content of the initial message.
Returns:
the initial message and the dialogue.
create_
with_
message
| create_with_message(counterparty: Address, initial_message: Message) -> Dialogue
Create a dialogue with 'counterparty', with an initial message provided.
Arguments:
counterparty
: the counterparty of the dialogue.initial_message
: the initial_message.
Returns:
the initial message and the dialogue.
update
| update(message: Message) -> Optional[Dialogue]
Update the state of dialogues with a new incoming message.
If the message is for a new dialogue, a new dialogue is created with 'message' as its first message, and returned. If the message is addressed to an existing dialogue, the dialogue is retrieved, extended with this message and returned. If there are any errors, e.g. the message dialogue reference does not exists or the message is invalid w.r.t. the dialogue, return None.
Arguments:
message
: a new incoming message
Returns:
the new or existing dialogue the message is intended for, or None in case of any errors.
get_
dialogue
| get_dialogue(message: Message) -> Optional[Dialogue]
Retrieve the dialogue 'message' belongs to.
Arguments:
message
: a message
Returns:
the dialogue, or None in case such a dialogue does not exist
setup
| setup() -> None
Set up.
teardown
| teardown() -> None
Tear down.