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

import copy
import logging
from pathlib import Path
from typing import Any, Dict, Text, Type, Union

from rasa.engine.caching import TrainingCache
from rasa.engine.graph import ExecutionContext, GraphSchema, GraphModelConfiguration
from rasa.engine.constants import PLACEHOLDER_IMPORTER
from rasa.engine.runner.interface import GraphRunner
from rasa.engine.storage.storage import ModelStorage, ModelMetadata
from rasa.engine.training.components import (
    PrecomputedValueProvider,
    FingerprintComponent,
    FingerprintStatus,
)
from rasa.engine.training.hooks import TrainingHook, LoggingHook
from rasa.shared.importers.importer import TrainingDataImporter

logger = logging.getLogger(__name__)


class GraphTrainer:
    """Trains a model using a graph schema."""

    def __init__(
        self,
        model_storage: ModelStorage,
        cache: TrainingCache,
        graph_runner_class: Type[GraphRunner],
    ) -> None:
        """Initializes a `GraphTrainer`.

        Args:
            model_storage: Storage which graph components can use to persist and load.
                Also used for packaging the trained model.
            cache: Cache used to store fingerprints and outputs.
            graph_runner_class: The class to instantiate the runner from.
        """
        self._model_storage = model_storage
        self._cache = cache
        self._graph_runner_class = graph_runner_class

    def train(
        self,
        model_configuration: GraphModelConfiguration,
        importer: TrainingDataImporter,
        output_filename: Path,
        force_retraining: bool = False,
        is_finetuning: bool = False,
    ) -> ModelMetadata:
        """Trains and packages a model and returns the prediction graph runner.

        Args:
            model_configuration: The model configuration (schemas, language, etc.)
            importer: The importer which provides the training data for the training.
            output_filename: The location to save the packaged model.
            force_retraining: If `True` then the cache is skipped and all components
                are retrained.

        Returns:
            The metadata describing the trained model.
        """
        logger.debug("Starting training.")

        # Retrieve the domain for the model metadata right at the start.
        # This avoids that something during the graph runs mutates it.
        domain = copy.deepcopy(importer.get_domain())

        if force_retraining:
            logger.debug(
                "Skip fingerprint run as a full training of the model was enforced."
            )
            pruned_training_schema = model_configuration.train_schema
        else:
            fingerprint_run_outputs = self.fingerprint(
                model_configuration.train_schema,
                importer=importer,
                is_finetuning=is_finetuning,
            )
            pruned_training_schema = self._prune_schema(
                model_configuration.train_schema, fingerprint_run_outputs
            )

        hooks = [
            LoggingHook(pruned_schema=pruned_training_schema),
            TrainingHook(
                cache=self._cache,
                model_storage=self._model_storage,
                pruned_schema=pruned_training_schema,
            ),
        ]

        graph_runner = self._graph_runner_class.create(
            graph_schema=pruned_training_schema,
            model_storage=self._model_storage,
            execution_context=ExecutionContext(
                graph_schema=model_configuration.train_schema,
                is_finetuning=is_finetuning,
            ),
            hooks=hooks,
        )

        logger.debug("Running the pruned train graph with real node execution.")

        graph_runner.run(inputs={PLACEHOLDER_IMPORTER: importer})

        return self._model_storage.create_model_package(
            output_filename, model_configuration, domain
        )

    def fingerprint(
        self,
        train_schema: GraphSchema,
        importer: TrainingDataImporter,
        is_finetuning: bool = False,
    ) -> Dict[Text, Union[FingerprintStatus, Any]]:
        """Runs the graph using fingerprints to determine which nodes need to re-run.

        Nodes which have a matching fingerprint key in the cache can either be removed
        entirely from the graph, or replaced with a cached value if their output is
        needed by descendent nodes.

        Args:
            train_schema: The train graph schema that will be run in fingerprint mode.
            importer: The importer which provides the training data for the training.
            is_finetuning: `True` if we want to finetune the model.

        Returns:
            Mapping of node names to fingerprint results.
        """
        fingerprint_schema = self._create_fingerprint_schema(train_schema)

        fingerprint_graph_runner = self._graph_runner_class.create(
            graph_schema=fingerprint_schema,
            model_storage=self._model_storage,
            execution_context=ExecutionContext(
                graph_schema=train_schema, is_finetuning=is_finetuning
            ),
        )

        logger.debug("Running the train graph in fingerprint mode.")
        return fingerprint_graph_runner.run(inputs={PLACEHOLDER_IMPORTER: importer})

    def _create_fingerprint_schema(self, train_schema: GraphSchema) -> GraphSchema:
        fingerprint_schema = copy.deepcopy(train_schema)
        for node_name, schema_node in fingerprint_schema.nodes.items():
            # We make every node a target so that `graph_runner.run(...)` returns
            # the output for each node. We need the output of each node
            # to decide which nodes we can prune.
            schema_node.is_target = True

            # We do not replace the input nodes as we need an up-to-date fingerprint of
            # any input data to the graph. This means we can prune according to what
            # has actually changed.
            if not schema_node.is_input:
                FingerprintComponent.replace_schema_node(schema_node, self._cache)
        return fingerprint_schema

    def _prune_schema(
        self,
        schema: GraphSchema,
        fingerprint_run_outputs: Dict[Text, Union[FingerprintStatus, Any]],
    ) -> GraphSchema:
        """Uses the fingerprint statuses to prune the graph schema.

        Walks the graph starting at each target node. If a node has a cache hit we
        replace it with a `PrecomputedValueProvider` and remove its input dependencies.
        At the end, any node that is not an ancestor of a target node will be pruned
        when we call `minimal_graph_schema()`.

        Args:
            schema: The graph to prune.
            fingerprint_run_outputs: Node outputs from the fingerprint run as a mapping
                from node name to output.

        Returns:
            The pruned schema.
        """
        pruned_schema = copy.deepcopy(schema)
        target_node_names = pruned_schema.target_names

        for target_node_name in target_node_names:
            self._walk_and_prune(
                pruned_schema, target_node_name, fingerprint_run_outputs
            )

        return pruned_schema.minimal_graph_schema()

    def _walk_and_prune(
        self,
        schema: GraphSchema,
        current_node_name: Text,
        fingerprint_run_outputs: Dict[Text, Union[FingerprintStatus, Any]],
    ) -> None:
        """Recursively walks backwards though a graph checking the status of each node.

        If node has a fingerprint key hit then we check if there is a cached output.
        If there is a cached output we will replace the node with a
        `PrecomputedValueProvider` and remove all its dependencies (`.needs`). If
        there is not a fingerprint key hit, or there is no cached output, the node is
        left untouched and will be executed again next run unless it is no longer the
        ancestor of a target node.

        Args:
            schema: The graph we are currently walking.
            current_node_name: The current node on the walk.
            fingerprint_run_outputs: The fingerprint statuses of every node as a mapping
                from node name to status.
        """
        fingerprint_run_output = fingerprint_run_outputs[current_node_name]
        node = schema.nodes[current_node_name]

        # If we have replaced this node with a `PrecomputedValueProvider` we have
        # already visited this node. A `PrecomputedValueProvider` is updated to have
        # no parent nodes, so
        # we can end the walk here.
        if node.uses == PrecomputedValueProvider:
            return

        # If the output was a `FingerprintStatus` we must check the cache and status.
        if isinstance(fingerprint_run_output, FingerprintStatus):
            # If there is a fingerprint key hit we can potentially use a cached output.
            if fingerprint_run_output.is_hit:
                output_result = self._cache.get_cached_result(
                    output_fingerprint_key=fingerprint_run_output.output_fingerprint,
                    node_name=current_node_name,
                    model_storage=self._model_storage,
                )
                if output_result:
                    logger.debug(
                        f"Updating '{current_node_name}' to use a "
                        f"'{PrecomputedValueProvider.__name__}'."
                    )
                    PrecomputedValueProvider.replace_schema_node(node, output_result)
                    # We remove all parent dependencies as the cached output value will
                    # be used.
                    node.needs = {}
                else:
                    # If there is no cached output the node must be re-run if it ends
                    # up as an ancestor of a target node.
                    fingerprint_run_output.is_hit = False

        # Else the node was an input node and the output is the actual node's output.
        else:
            # As fingerprint_run_output is just the node's output there is no need to
            # execute the node again. We can just return it from a
            # `PrecomputedValueProvider`.
            PrecomputedValueProvider.replace_schema_node(node, fingerprint_run_output)
            node.needs = {}

        # Continue walking for every parent node.
        for parent_node_name in node.needs.values():
            self._walk_and_prune(schema, parent_node_name, fingerprint_run_outputs)