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__

def __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
def dialogue_reference() -> Tuple[str, str]

Get the dialogue reference.

dialogue_starter_reference

@property
def dialogue_starter_reference() -> str

Get the dialogue starter reference.

dialogue_responder_reference

@property
def dialogue_responder_reference() -> str

Get the dialogue responder reference.

dialogue_opponent_addr

@property
def dialogue_opponent_addr() -> str

Get the address of the dialogue opponent.

dialogue_starter_addr

@property
def dialogue_starter_addr() -> str

Get the address of the dialogue starter.

__eq__

def __eq__(other: Any) -> bool

Check for equality between two DialogueLabel objects.

__hash__

def __hash__() -> int

Turn object into hash.

json

@property
def json() -> Dict

Return the JSON representation.

from_json

@classmethod
def from_json(cls, obj: Dict[str, str]) -> "DialogueLabel"

Get dialogue label from json.

is_complete

def is_complete() -> bool

Check if the dialogue label is complete.

get_incomplete_version

def get_incomplete_version() -> "DialogueLabel"

Get the incomplete version of the label.

get_both_versions

def get_both_versions() -> Tuple["DialogueLabel", Optional["DialogueLabel"]]

Get the incomplete and complete versions of the label.

__str__

def __str__() -> str

Get the string representation.

from_str

@classmethod
def 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__

def __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.

INITIAL_PERFORMATIVES

type: FrozenSet[Message.Performative]

TERMINAL_PERFORMATIVES

type: FrozenSet[Message.Performative]

VALID_REPLIES

type: Dict[Message.Performative, FrozenSet[Message.Performative]]

Rules Objects

class Rules()

This class defines the rules for the dialogue.

__init__

def __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
def 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
def 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
def 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

def 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__

def __str__() -> str

Get the string representation.

EndState Objects

class EndState(Enum)

This class defines the end states of a dialogue.

__str__

def __str__() -> str

Get the string representation.

__init__

def __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

def 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__

def __eq__(other: Any) -> bool

Compare two dialogues.

json

def json() -> dict

Get json representation of the dialogue.

from_json

@classmethod
def 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
def dialogue_label() -> DialogueLabel

Get the dialogue label.

Returns:

The dialogue label

incomplete_dialogue_label

@property
def incomplete_dialogue_label() -> DialogueLabel

Get the dialogue label.

Returns:

The incomplete dialogue label

dialogue_labels

@property
def dialogue_labels() -> Set[DialogueLabel]

Get the dialogue labels (incomplete and complete, if it exists).

Returns:

the dialogue labels

self_address

@property
def self_address() -> Address

Get the address of the entity for whom this dialogues is maintained.

Returns:

the address of this entity

role

@property
def role() -> "Role"

Get the agent's role in the dialogue.

Returns:

the agent's role

rules

@property
def rules() -> "Rules"

Get the dialogue rules.

Returns:

the rules

message_class

@property
def message_class() -> Type[Message]

Get the message class.

Returns:

the message class

is_self_initiated

@property
def 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
def last_incoming_message() -> Optional[Message]

Get the last incoming message.

Returns:

the last incoming message if it exists, None otherwise

last_outgoing_message

@property
def last_outgoing_message() -> Optional[Message]

Get the last outgoing message.

Returns:

the last outgoing message if it exists, None otherwise

last_message

@property
def last_message() -> Optional[Message]

Get the last message.

Returns:

the last message if it exists, None otherwise

is_empty

@property
def is_empty() -> bool

Check whether the dialogue is empty.

Returns:

True if empty, False otherwise

reply

def 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

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

Get message by id, if not presents return None.

get_outgoing_next_message_id

def get_outgoing_next_message_id() -> int

Get next outgoing message id.

get_incoming_next_message_id

def get_incoming_next_message_id() -> int

Get next incoming message id.

__str__

def __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__

def __init__(end_states: FrozenSet[Dialogue.EndState]) -> None

Initialize a StatsManager.

Arguments:

  • end_states: the list of dialogue endstates

self_initiated

@property
def self_initiated() -> Dict[Dialogue.EndState, int]

Get the stats dictionary on self initiated dialogues.

other_initiated

@property
def other_initiated() -> Dict[Dialogue.EndState, int]

Get the stats dictionary on other initiated dialogues.

add_dialogue_endstate

def 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

def 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__

def __init__(dialogues: "Dialogues") -> None

Init dialogues storage.

cleanup

def cleanup() -> None

Clean up the dialogue storage

dialogues_in_terminal_state

@property
def dialogues_in_terminal_state() -> List["Dialogue"]

Get all dialogues in terminal state.

dialogues_in_active_state

@property
def dialogues_in_active_state() -> List["Dialogue"]

Get all dialogues in active state.

is_terminal_dialogues_kept

@property
def is_terminal_dialogues_kept() -> bool

Return True if dialogues should stay after terminal state.

dialogue_terminal_state_callback

def dialogue_terminal_state_callback(dialogue: "Dialogue") -> None

Method to be called on dialogue terminal state reached.

setup

def setup() -> None

Set up dialogue storage.

teardown

def teardown() -> None

Tear down dialogue storage.

add

def add(dialogue: Dialogue) -> None

Add dialogue to storage.

Label can be complete (if receiving a message, and add being called by Dialogue.update) or incomplete (if sending a message, and add being called by Dialogue.create).

Arguments:

  • dialogue: dialogue to add.

remove

def remove(dialogue_label: DialogueLabel) -> None

Remove dialogue from storage by it's label.

Arguments:

  • dialogue_label: label of the dialogue to remove

get

def 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

def 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

def is_in_incomplete(dialogue_label: DialogueLabel) -> bool

Check dialogue label presents in list of incomplete.

set_incomplete_dialogue

def set_incomplete_dialogue(incomplete_dialogue_label: DialogueLabel,
                            complete_dialogue_label: DialogueLabel) -> None

Set incomplete dialogue label.

is_dialogue_present

def is_dialogue_present(dialogue_label: DialogueLabel) -> bool

Check dialogue with label specified presents in storage.

get_latest_label

def 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__

def __init__(dialogues: "Dialogues") -> None

Init dialogues storage.

get_skill_component

@staticmethod
def get_skill_component() -> Optional[SkillComponent]

Get skill component dialogues storage constructed for.

setup

def setup() -> None

Set up dialogue storage.

teardown

def teardown() -> None

Tear down dialogue storage.

remove

def 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

def dialogue_terminal_state_callback(dialogue: "Dialogue") -> None

Call on dialogue reaches terminal state.

get

def get(dialogue_label: DialogueLabel) -> Optional[Dialogue]

Try to get dialogue by label from memory or persists storage.

get_dialogues_with_counterparty

def 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
def 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__

def __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

cleanup

def cleanup() -> None

Clean up the dialogue storage

is_keep_dialogues_in_terminal_state

@property
def is_keep_dialogues_in_terminal_state() -> bool

Is required to keep dialogues in terminal state.

self_address

@property
def self_address() -> Address

Get the address of the agent for whom dialogues are maintained.

dialogue_stats

@property
def dialogue_stats() -> DialogueStats

Get the dialogue statistics.

Returns:

dialogue stats object

message_class

@property
def message_class() -> Type[Message]

Get the message class.

Returns:

the message class

dialogue_class

@property
def dialogue_class() -> Type[Dialogue]

Get the dialogue class.

Returns:

the dialogue class

get_dialogues_with_counterparty

def 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
def new_self_initiated_dialogue_reference(cls) -> Tuple[str, str]

Return a dialogue label for a new self initiated dialogue.

Returns:

the next nonce

create

def 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

def 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

def 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

def 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

def 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

def setup() -> None

Set up.

teardown

def teardown() -> None

Tear down.