rasahq--rasa/rasa/engine/training/hooks.py

import logging
from typing import Any, Dict, Text, Type

from rasa.engine.caching import TrainingCache
from rasa.engine.graph import ExecutionContext, GraphNodeHook, GraphSchema, SchemaNode
from rasa.engine.storage.storage import ModelStorage
from rasa.engine.training.components import PrecomputedValueProvider
import rasa.shared.utils.io
from rasa.engine.training import fingerprinting

logger = logging.getLogger(__name__)


class TrainingHook(GraphNodeHook):
    """Caches fingerprints and outputs of nodes during model training."""

    def __init__(
        self,
        cache: TrainingCache,
        model_storage: ModelStorage,
        pruned_schema: GraphSchema,
    ) -> None:
        """Initializes a `TrainingHook`.

        Args:
            cache: Cache used to store fingerprints and outputs.
            model_storage: Used to cache `Resource`s.
            pruned_schema: The pruned training schema.
        """
        self._cache = cache
        self._model_storage = model_storage
        self._pruned_schema = pruned_schema

    def on_before_node(
        self,
        node_name: Text,
        execution_context: ExecutionContext,
        config: Dict[Text, Any],
        received_inputs: Dict[Text, Any],
    ) -> Dict:
        """Calculates the run fingerprint for use in `on_after_node`."""
        graph_component_class = self._get_graph_component_class(
            execution_context, node_name
        )

        fingerprint_key = fingerprinting.calculate_fingerprint_key(
            graph_component_class=graph_component_class,
            config=config,
            inputs=received_inputs,
        )

        return {"fingerprint_key": fingerprint_key}

    def on_after_node(
        self,
        node_name: Text,
        execution_context: ExecutionContext,
        config: Dict[Text, Any],
        output: Any,
        input_hook_data: Dict,
    ) -> None:
        """Stores the fingerprints and caches the output of the node."""
        # We should not re-cache the output of a PrecomputedValueProvider.
        graph_component_class = self._pruned_schema.nodes[node_name].uses

        if graph_component_class == PrecomputedValueProvider:
            return None

        output_fingerprint = rasa.shared.utils.io.deep_container_fingerprint(output)
        fingerprint_key = input_hook_data["fingerprint_key"]

        logger.debug(
            f"Caching '{output.__class__.__name__}' with fingerprint_key: "
            f"'{fingerprint_key}' and output_fingerprint '{output_fingerprint}'."
        )

        self._cache.cache_output(
            fingerprint_key=fingerprint_key,
            output=output,
            output_fingerprint=output_fingerprint,
            model_storage=self._model_storage,
        )

    @staticmethod
    def _get_graph_component_class(
        execution_context: ExecutionContext, node_name: Text
    ) -> Type:
        graph_component_class = execution_context.graph_schema.nodes[node_name].uses
        return graph_component_class


class LoggingHook(GraphNodeHook):
    """Logs the training of components."""

    def __init__(self, pruned_schema: GraphSchema) -> None:
        """Creates hook.

        Args:
            pruned_schema: The pruned schema provides us with the information whether
                a component is cached or not.
        """
        self._pruned_schema = pruned_schema

    def on_before_node(
        self,
        node_name: Text,
        execution_context: ExecutionContext,
        config: Dict[Text, Any],
        received_inputs: Dict[Text, Any],
    ) -> Dict:
        """Logs the training start of a graph node."""
        node = self._pruned_schema.nodes[node_name]

        if not self._is_cached_node(node) and self._does_node_train(node):
            logger.info(f"Starting to train component '{node.uses.__name__}'.")

        return {}

    @staticmethod
    def _does_node_train(node: SchemaNode) -> bool:
        # Nodes which train are always targets so that they store their output in the
        # model storage. `is_input` filters out nodes which don't really train but e.g.
        # persist some training data.
        return node.is_target and not node.is_input

    @staticmethod
    def _is_cached_node(node: SchemaNode) -> bool:
        return node.uses == PrecomputedValueProvider

    def on_after_node(
        self,
        node_name: Text,
        execution_context: ExecutionContext,
        config: Dict[Text, Any],
        output: Any,
        input_hook_data: Dict,
    ) -> None:
        """Logs when a component finished its training."""
        node = self._pruned_schema.nodes[node_name]

        if not self._does_node_train(node):
            return

        if self._is_cached_node(node):
            actual_component = execution_context.graph_schema.nodes[node_name]
            logger.info(
                f"Restored component '{actual_component.uses.__name__}' from cache."
            )
        else:
            logger.info(f"Finished training component '{node.uses.__name__}'.")