Version: 3.x

rasa.core.evaluation.marker_base

MarkerRegistry Objects

class MarkerRegistry()

Keeps track of tags that can be used to configure markers.

register_builtin_markers

@classmethod
def register_builtin_markers(cls) -> None

Must import all modules containing markers.

configurable_marker

@classmethod
def configurable_marker(cls, marker_class: Type[Marker]) -> Type[Marker]

Decorator used to register a marker that can be used in config files.

Arguments:

  • marker_class - the marker class to be made available via config files

Returns:

the registered marker class

get_non_negated_tag

@classmethod
def get_non_negated_tag(cls, tag_or_negated_tag: Text) -> Tuple[Text, bool]

Returns the non-negated marker tag, given a (possible) negated marker tag.

Arguments:

  • tag_or_negated_tag - the tag for a possibly negated marker

Returns:

the tag itself if it was already positive, otherwise the positive version; and a boolean that represents whether the given tag was a negative one

InvalidMarkerConfig Objects

class InvalidMarkerConfig(RasaException)

Exception that can be raised when the config for a marker is not valid.

EventMetaData Objects

@dataclass
class EventMetaData()

Describes meta data per event in some session.

Marker Objects

class Marker(ABC)

A marker is a way of describing points in conversations you're interested in.

Here, markers are stateful objects because they track the events of a conversation. At each point in the conversation, one can observe whether a marker applies or does not apply to the conversation so far.

__init__

def __init__(name: Optional[Text] = None,
negated: bool = False,
description: Optional[Text] = None) -> None

Instantiates a marker.

Arguments:

  • name - a custom name that can be used to replace the default string conversion of this marker
  • negated - whether this marker should be negated (i.e. a negated marker applies if and only if the non-negated marker does not apply)
  • description - an optional description of the marker. It is not used internally but can be used to document the marker.

Raises:

InvalidMarkerConfig if the chosen name of the marker is the tag of a predefined marker.

get_tag

def get_tag() -> Text

Returns the tag describing this marker.

positive_tag

@staticmethod
@abstractmethod
def positive_tag() -> Text

Returns the tag to be used in a config file.

negated_tag

@staticmethod
def negated_tag() -> Optional[Text]

Returns the tag to be used in a config file for the negated version.

If this maps to None, then this indicates that we do not allow a short-cut for negating this marker. Hence, there is not a single tag to instantiate a negated version of this marker. One must use a "not" in the configuration file then.

track

def track(event: Event) -> None

Updates the marker according to the given event.

Arguments:

  • event - the next event of the conversation

reset

def reset() -> None

Clears the history of the marker.

flatten

@abstractmethod
def flatten() -> Iterator[Marker]

Returns an iterator over all conditions and operators used in this marker.

Returns:

an iterator over all conditions and operators that are part of this marker

validate_against_domain

@abstractmethod
def validate_against_domain(domain: Domain) -> bool

Checks that this marker (and its children) refer to entries in the domain.

Arguments:

  • domain - The domain to check against

max_depth

@abstractmethod
def max_depth() -> int

Gets the maximum depth from this point in the marker tree.

evaluate_events

def evaluate_events(events: List[Event]) -> List[SessionEvaluation]

Resets the marker, tracks all events, and collects some information.

The collected information includes:

  • the timestamp of each event where the marker applied and
  • the number of user turns that preceded that timestamp

If this marker is the special ANY_MARKER (identified by its name), then results will be collected for all (immediate) sub-markers.

Arguments:

  • events - a list of events describing a conversation

Returns:

a list that contains, for each session contained in the tracker, a dictionary mapping that maps marker names to meta data of relevant events

relevant_events

def relevant_events() -> List[int]

Returns the indices of those tracked events that are relevant for evaluation.

Note: Overwrite this method if you create a new marker class that should not contain meta data about each event where the marker applied in the final evaluation (see evaluate_events).

Returns:

indices of tracked events

from_path

@classmethod
def from_path(cls, path: Union[Path, Text]) -> "OrMarker"

Loads markers from one config file or all config files in a directory tree.

Each config file should contain a dictionary mapping marker names to the respective marker configuration. To avoid confusion, the marker names must not coincide with the tag of any pre-defined markers. Moreover, marker names must be unique. This means, if you you load the markers from multiple files, then you have to make sure that the names of the markers defined in these files are unique across all loaded files.

Note that all loaded markers will be combined into one marker via one artificial OR-operator. When evaluating the resulting marker, then the artificial OR-operator will be ignored and results will be reported for every sub-marker.

For more details how a single marker configuration looks like, have a look at Marker.from_config.

Arguments:

  • path - either the path to a single config file or the root of the directory tree that contains marker config files

Returns:

all configured markers, combined into one marker object

from_config

@staticmethod
def from_config(config: Any, name: Optional[Text] = None) -> Marker

Creates a marker from the given config.

A marker configuration is a dictionary mapping a marker tag (either a positive_tag or a negated_tag) to a sub-configuration. How that sub-configuration looks like, depends on whether the tag describes an operator (see OperatorMarker.from_tag_and_sub_config) or a condition (see ConditionMarker.from_tag_and_sub_config).

Arguments:

  • config - a marker configuration
  • name - a custom name that will be used for the top-level marker (if and only if there is only one top-level marker)

Returns:

the configured marker

evaluate_trackers

async def evaluate_trackers(trackers: AsyncIterator[
Optional[DialogueStateTracker]],
output_file: Path,
session_stats_file: Optional[Path] = None,
overall_stats_file: Optional[Path] = None) -> None

Collect markers for each dialogue in each tracker loaded.

Arguments:

  • trackers - An iterator over the trackers from which we want to extract markers.
  • output_file - Path to write out the extracted markers.
  • session_stats_file - (Optional) Path to write out statistics about the extracted markers for each session separately.
  • overall_stats_file - (Optional) Path to write out statistics about the markers extracted from all session data.

Raises:

FileExistsError if any of the specified files already exists NotADirectoryError if any of the specified files is supposed to be contained in a directory that does not exist

OperatorMarker Objects

class OperatorMarker(Marker, ABC)

Combines several markers into one.

__init__

def __init__(markers: List[Marker],
negated: bool = False,
name: Optional[Text] = None,
description: Optional[Text] = None) -> None

Instantiates a marker.

Arguments:

  • markers - the list of markers to combine
  • negated - whether this marker should be negated (i.e. a negated marker applies if and only if the non-negated marker does not apply)
  • name - a custom name that can be used to replace the default string conversion of this marker
  • description - an optional description of the marker. It is not used internally but can be used to document the marker.

Raises:

InvalidMarkerConfig if the given number of sub-markers does not match the expected number of sub-markers

expected_number_of_sub_markers

@staticmethod
def expected_number_of_sub_markers() -> Optional[int]

Returns the expected number of sub-markers (if there is any).

track

def track(event: Event) -> None

Updates the marker according to the given event.

All sub-markers will be updated before the compound marker itself is updated.

Arguments:

  • event - the next event of the conversation

flatten

def flatten() -> Iterator[Marker]

Returns an iterator over all included markers, plus this marker itself.

Returns:

an iterator over all markers that are part of this marker

reset

def reset() -> None

Resets the history of this marker and all its sub-markers.

validate_against_domain

def validate_against_domain(domain: Domain) -> bool

Checks that this marker (and its children) refer to entries in the domain.

Arguments:

  • domain - The domain to check against

max_depth

def max_depth() -> int

Gets the maximum depth from this point in the marker tree.

from_tag_and_sub_config

@staticmethod
def from_tag_and_sub_config(
tag: Text,
sub_config: Any,
name: Optional[Text] = None,
description: Optional[Text] = None) -> OperatorMarker

Creates an operator marker from the given config.

The configuration must consist of a list of marker configurations. See Marker.from_config for more details.

Arguments:

  • tag - the tag identifying an operator
  • sub_config - a list of marker configs
  • name - an optional custom name to be attached to the resulting marker
  • description - an optional description of the marker

Returns:

the configured operator marker

Raises:

InvalidMarkerConfig if the given config or the tag are not well-defined

ConditionMarker Objects

class ConditionMarker(Marker, ABC)

A marker that does not contain any sub-markers.

__init__

def __init__(text: Text,
negated: bool = False,
name: Optional[Text] = None,
description: Optional[Text] = None) -> None

Instantiates an atomic marker.

Arguments:

  • text - some text used to decide whether the marker applies
  • negated - whether this marker should be negated (i.e. a negated marker applies if and only if the non-negated marker does not apply)
  • name - a custom name that can be used to replace the default string conversion of this marker
  • description - an optional description of the marker. It is not used internally but can be used to document the marker.

flatten

def flatten() -> Iterator[ConditionMarker]

Returns an iterator that just returns this AtomicMarker.

Returns:

an iterator over all markers that are part of this marker, i.e. this marker

max_depth

def max_depth() -> int

Gets the maximum depth from this point in the marker tree.

from_tag_and_sub_config

@staticmethod
def from_tag_and_sub_config(
tag: Text,
sub_config: Any,
name: Optional[Text] = None,
description: Optional[Text] = None) -> ConditionMarker

Creates an atomic marker from the given config.

Arguments:

  • tag - the tag identifying a condition
  • sub_config - a single text parameter expected by all condition markers; e.g. if the tag is for an intent_detected marker then the config should contain an intent name
  • name - a custom name for this marker

Returns:

the configured ConditionMarker

Raises:

InvalidMarkerConfig if the given config or the tag are not well-defined