Skip to content

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.

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: Any) -> 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__() -> 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.

Creates class level Rules instance to share among instances

__new__

 | __new__(cls, name: str, bases: Tuple[Type], dct: Dict) -> "_DialogueMeta"

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.

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__() -> str

Get the string representation.

EndState Objects

class EndState(Enum)

This class defines the end states of a dialogue.

__str__

 | __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 dialogue
  • message_class: the message class used
  • self_address: the address of the entity for whom this dialogue is maintained
  • role: the role of the agent this dialogue is maintained for

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

__eq__

 | __eq__(other: Any) -> 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 dialogue
  • data: 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: Any, ,) -> 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.

get_message_by_id

 | get_message_by_id(message_id: int) -> Optional[Message]

Get message by id, if not presents return None.

get_outgoing_next_message_id

 | get_outgoing_next_message_id() -> int

Get next outgoing message id.

get_incoming_next_message_id

 | get_incoming_next_message_id() -> int

Get next incoming message id.

__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 dialogue
  • is_self_initiated: whether the dialogue is initiated by the agent or the opponent

find_caller_object

find_caller_object(object_type: Type) -> Any

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.

remove

 | remove(dialogue_label: DialogueLabel) -> None

Remove dialogue from storage by it's label.

Arguments:

  • dialogue_label: label of the dialogue to remove

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: DialogueLabel, complete_dialogue_label: DialogueLabel) -> 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 state.

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 maintained
  • end_states: the list of dialogue endstates
  • message_class: the message class used
  • dialogue_class: the dialogue class used
  • role_from_first_message: the callable determining role from first message
  • keep_terminal_state_dialogues: specify do dialogues in terminal state should stay or not

is_keep_dialogues_in_terminal_state

 | @property
 | is_keep_dialogues_in_terminal_state() -> bool

Is required 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: Any, ,) -> 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

get_dialogue_from_label

 | get_dialogue_from_label(dialogue_label: DialogueLabel) -> Optional[Dialogue]

Retrieve a dialogue based on its label.

Arguments:

  • dialogue_label: the dialogue label

Returns:

the dialogue if present

setup

 | setup() -> None

Set up.

teardown

 | teardown() -> None

Tear down.