Repository: RasaHQ/rasa
Pull request: Use a unique fingerprint for a resource that is consistent after caching.
Use a unique fingerprint for a resource that is consistent after caching. This ensures we always retrieve the correct resource for a node from the cache.

Fixes #10147

**Status (please check what you already did)**:
- [x] added some tests for the functionality
- [x] reformat files using `black` (please check [Readme](https://github.com/RasaHQ/rasa#code-style) for instructions)


Problem statement:
Missing intent in domain breaks UnexpecTED intent policy
### Rasa Open Source version

3.0.0rc1

### Rasa SDK version

_No response_

### Rasa X version

_No response_

### Python version

3.8

### What operating system are you using?

OSX

### What happened?

When I have an intent in my NLU data which is not in the domain, having a prediction with this intent in the tracker then breaks unexpecTED. Probably a similar a case as if we’d have a tracker with old intents.

```
Traceback (most recent call last):
  File "/Users/tayfun/.virtualenvs/rasa-rc-38/lib/python3.8/site-packages/rasa/engine/graph.py", line 455, in __call__
    output = self._fn(self._component, **run_kwargs)
  File "/Users/tayfun/.virtualenvs/rasa-rc-38/lib/python3.8/site-packages/rasa/core/policies/unexpected_intent_policy.py", line 615, in predict_action_probabilities
    is_unlikely_intent = self._check_unlikely_intent(
  File "/Users/tayfun/.virtualenvs/rasa-rc-38/lib/python3.8/site-packages/rasa/core/policies/unexpected_intent_policy.py", line 718, in _check_unlikely_intent
    predicted_intent_scores = {
  File "/Users/tayfun/.virtualenvs/rasa-rc-38/lib/python3.8/site-packages/rasa/core/policies/unexpected_intent_policy.py", line 719, in <dictcomp>
    index: similarities[0][index] for index, intent in enumerate(domain.intents)
IndexError: index 12 is out of bounds for axis 0 with size 12
```

### Command / Request

_No response_

### Relevant log output

_No response_

Hints from issue discussion:

Commit message: Merge remote-tracking branch 'origin/main' into 10147/use_unique_fingerprints_for_resource

Patch to review:
diff --git a/rasa/engine/caching.py b/rasa/engine/caching.py
index 771b030be59e..596235c86cad 100644
--- a/rasa/engine/caching.py
+++ b/rasa/engine/caching.py
@@ -124,7 +124,11 @@ def to_cache(self, directory: Path, model_storage: ModelStorage) -> None:
 
     @classmethod
     def from_cache(
-        cls, node_name: Text, directory: Path, model_storage: ModelStorage
+        cls,
+        node_name: Text,
+        directory: Path,
+        model_storage: ModelStorage,
+        output_fingerprint: Text,
     ) -> Cacheable:
         """Loads `Cacheable` from cache.
 
@@ -133,6 +137,9 @@ def from_cache(
             directory: Directory containing the persisted `Cacheable`.
             model_storage: The current model storage (e.g. used when restoring
                 `Resource` objects so that they can fill the model storage with data).
+            output_fingerprint: The fingerprint of the cached result (e.g. used when
+                restoring `Resource` objects as the fingerprint can not be easily
+                calculated from the object itself).
 
         Returns:
             Instantiated `Cacheable`.
@@ -407,7 +414,11 @@ def get_cached_result(
             return None
 
         return self._load_from_cache(
-            result_location, result_type, node_name, model_storage
+            result_location,
+            result_type,
+            node_name,
+            model_storage,
+            output_fingerprint_key,
         )
 
     def _get_cached_result(
@@ -434,6 +445,7 @@ def _load_from_cache(
         result_type: Text,
         node_name: Text,
         model_storage: ModelStorage,
+        output_fingerprint_key,
     ) -> Optional[Cacheable]:
         try:
             module = rasa.shared.utils.common.class_from_module_path(result_type)
@@ -446,7 +458,9 @@ def _load_from_cache(
                 )
                 return None
 
-            return module.from_cache(node_name, path_to_cached, model_storage)
+            return module.from_cache(
+                node_name, path_to_cached, model_storage, output_fingerprint_key
+            )
         except Exception as e:
             logger.warning(
                 f"Failed to restore cached output of type '{result_type}' from "
diff --git a/rasa/engine/storage/resource.py b/rasa/engine/storage/resource.py
index c604c0c1b2af..66f981f13f5a 100644
--- a/rasa/engine/storage/resource.py
+++ b/rasa/engine/storage/resource.py
@@ -1,9 +1,10 @@
 from __future__ import annotations
 import logging
 import typing
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Text
+from typing import Text, Optional
+import uuid
 
 import rasa.utils.common
 import rasa.utils.io
@@ -16,13 +17,32 @@
 
 @dataclass
 class Resource:
-    """Represents a persisted graph component in the graph."""
+    """Represents a persisted graph component in the graph.
+
+    Attributes:
+        name: The unique identifier for the `Resource`. Used to locate the associated
+            data from a `ModelStorage`. Normally matches the name of the node which
+            created it.
+        output_fingerprint: An unique identifier for a specific instantiation of a
+            `Resource`. Used to distinguish a specific persistence for the same
+            `Resource` when saving to the cache.
+
+    """
 
     name: Text
+    output_fingerprint: Optional[Text] = field(
+        default_factory=lambda: uuid.uuid4().hex,
+        # We do not use this for comparison as it is not consistent after serialization.
+        compare=False,
+    )
 
     @classmethod
     def from_cache(
-        cls, node_name: Text, directory: Path, model_storage: ModelStorage
+        cls,
+        node_name: Text,
+        directory: Path,
+        model_storage: ModelStorage,
+        output_fingerprint: Text,
     ) -> Resource:
         """Loads a `Resource` from the cache.
 
@@ -33,13 +53,14 @@ def from_cache(
             directory: The directory with the cached `Resource`.
             model_storage: The `ModelStorage` which the cached `Resource` will be added
                 to so that the `Resource` is accessible for other graph nodes.
+            output_fingerprint: The fingerprint of the cached `Resource`.
 
         Returns:
             The ready-to-use and accessible `Resource`.
         """
         logger.debug(f"Loading resource '{node_name}' from cache.")
 
-        resource = Resource(node_name)
+        resource = Resource(node_name, output_fingerprint=output_fingerprint)
         if not any(directory.glob("*")):
             logger.debug(f"Cached resource for '{node_name}' was empty.")
             return resource
@@ -79,11 +100,11 @@ def to_cache(self, directory: Path, model_storage: ModelStorage) -> None:
     def fingerprint(self) -> Text:
         """Provides fingerprint for `Resource`.
 
-        The fingerprint can be just the name as the persisted resource only changes
-        if the used training data (which is loaded in previous nodes) or the config
-        (which is fingerprinted separately) changes.
+        A unique fingerprint is created on initialization of a `Resource` however we
+        also allow a value to be provided for when we retrieve a `Resource` from the
+        cache (see `Resource.from_cache`).
 
         Returns:
             Fingerprint for `Resource`.
         """
-        return self.name
+        return self.output_fingerprint
diff --git a/rasa/engine/training/hooks.py b/rasa/engine/training/hooks.py
index c9435c9bc7a8..b399a0f904f5 100644
--- a/rasa/engine/training/hooks.py
+++ b/rasa/engine/training/hooks.py
@@ -42,6 +42,7 @@ def on_before_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,
diff --git a/tests/engine/graph_components_test_classes.py b/tests/engine/graph_components_test_classes.py
index 897f9b58138c..211a4f7a3943 100644
--- a/tests/engine/graph_components_test_classes.py
+++ b/tests/engine/graph_components_test_classes.py
@@ -226,7 +226,11 @@ def to_cache(self, directory: Path, model_storage: ModelStorage) -> None:
 
     @classmethod
     def from_cache(
-        cls, node_name: Text, directory: Path, model_storage: ModelStorage
+        cls,
+        node_name: Text,
+        directory: Path,
+        model_storage: ModelStorage,
+        output_fingerprint: Text,
     ) -> CacheableText:
         text = rasa.shared.utils.io.read_file(directory / "my_file.txt")
         return cls(text=text)
diff --git a/tests/engine/storage/test_resource.py b/tests/engine/storage/test_resource.py
index 097deb020d45..307b29c10e25 100644
--- a/tests/engine/storage/test_resource.py
+++ b/tests/engine/storage/test_resource.py
@@ -29,11 +29,13 @@ def test_resource_caching(tmp_path_factory: TempPathFactory):
     # Reload resource from cache and inspect
     new_model_storage = LocalModelStorage(tmp_path_factory.mktemp("new_model_storage"))
     reinstantiated_resource = Resource.from_cache(
-        resource.name, cache_dir, new_model_storage
+        resource.name, cache_dir, new_model_storage, resource.output_fingerprint
     )
 
     assert reinstantiated_resource == resource
 
+    assert reinstantiated_resource.fingerprint() == resource.fingerprint()
+
     # Read written resource data from model storage to see whether all expected
     # contents are there
     with new_model_storage.read_from(resource) as temporary_directory:
@@ -62,7 +64,10 @@ def test_resource_caching_if_already_restored(tmp_path_factory: TempPathFactory)
     rasa.utils.common.copy_directory(initial_storage_dir, new_storage_dir)
 
     reinstantiated_resource = Resource.from_cache(
-        resource.name, cache_dir, LocalModelStorage(new_storage_dir)
+        resource.name,
+        cache_dir,
+        LocalModelStorage(new_storage_dir),
+        resource.output_fingerprint,
     )
 
     assert reinstantiated_resource == resource
@@ -98,21 +103,29 @@ def test_resource_caching_if_already_restored_with_different_state(
 
     with pytest.raises(ValueError):
         Resource.from_cache(
-            resource.name, cache_dir, LocalModelStorage(new_storage_dir)
+            resource.name,
+            cache_dir,
+            LocalModelStorage(new_storage_dir),
+            resource.output_fingerprint,
         )
 
 
 def test_resource_fingerprinting():
-    resource1 = Resource("resource 1")
-    resource2 = Resource("resource 2")
+    resource1a = Resource("resource 1")
+    resource1b = Resource("resource 1")
+    resource2 = Resource("resource 3")
 
-    fingerprint1 = resource1.fingerprint()
-    fingerprint2 = resource2.fingerprint()
+    fingerprint1 = resource1a.fingerprint()
+    fingerprint2 = resource1b.fingerprint()
+    fingerprint3 = resource2.fingerprint()
 
     assert fingerprint1
     assert fingerprint2
+    assert fingerprint3
 
     assert fingerprint1 != fingerprint2
+    assert fingerprint2 != fingerprint3
+    assert fingerprint1 != fingerprint3
 
 
 def test_caching_empty_resource(
@@ -133,7 +146,9 @@ def test_caching_empty_resource(
     cache_dir = tmp_path_factory.mktemp("cache_dir")
 
     # this doesn't create an empty directory in `default_model_storage`
-    Resource.from_cache(resource_name, cache_dir, default_model_storage)
+    Resource.from_cache(
+        resource_name, cache_dir, default_model_storage, resource.output_fingerprint
+    )
 
     with pytest.raises(ValueError):
         with default_model_storage.read_from(resource) as _:
diff --git a/tests/engine/test_caching.py b/tests/engine/test_caching.py
index a862ba2f6950..5c0ad884b6ee 100644
--- a/tests/engine/test_caching.py
+++ b/tests/engine/test_caching.py
@@ -45,7 +45,11 @@ def to_cache(self, directory: Path, model_storage: ModelStorage) -> None:
 
     @classmethod
     def from_cache(
-        cls, node_name: Text, directory: Path, model_storage: ModelStorage
+        cls,
+        node_name: Text,
+        directory: Path,
+        model_storage: ModelStorage,
+        output_fingerprint: Text,
     ) -> "TestCacheableOutput":
 
         value = rasa.shared.utils.io.read_json_file(directory / "cached.json")
diff --git a/tests/engine/training/test_graph_trainer.py b/tests/engine/training/test_graph_trainer.py
index e4874a81df9c..d9a2ed4f0d91 100644
--- a/tests/engine/training/test_graph_trainer.py
+++ b/tests/engine/training/test_graph_trainer.py
@@ -8,7 +8,7 @@
 from _pytest.tmpdir import TempPathFactory
 import pytest
 
-from rasa.engine.caching import TrainingCache
+from rasa.engine.caching import LocalTrainingCache, TrainingCache
 from rasa.engine.exceptions import GraphComponentException
 from rasa.engine.graph import (
     GraphComponent,
@@ -494,6 +494,118 @@ def test_graph_trainer_train_logging_with_cached_components(
         }
 
 
+def test_resources_fingerprints_are_unique_when_cached(
+    tmp_path: Path,
+    temp_cache: LocalTrainingCache,
+    train_with_schema: Callable,
+    caplog: LogCaptureFixture,
+):
+    train_schema = GraphSchema(
+        {
+            "train": SchemaNode(
+                needs={},
+                uses=PersistableTestComponent,
+                fn="train",
+                constructor_name="create",
+                config={"test_value": "4"},
+                is_target=True,
+            ),
+            "process": SchemaNode(
+                needs={"resource": "train"},
+                uses=PersistableTestComponent,
+                fn="run_inference",
+                constructor_name="load",
+                config={},
+            ),
+            "assert_node": SchemaNode(
+                needs={"i": "process"},
+                uses=AssertComponent,
+                fn="run_assert",
+                constructor_name="create",
+                config={"value_to_assert": "4"},
+                is_target=True,
+            ),
+        }
+    )
+
+    # Train to cache
+    train_with_schema(train_schema, temp_cache)
+
+    train_schema.nodes["train"].config["test_value"] = "5"
+    train_schema.nodes["assert_node"].config["value_to_assert"] = "5"
+    train_with_schema(train_schema, temp_cache)
+
+    # This breaks when `Resource`s use the node name as a fingerprint.
+    # This is because the `Resource` for the first run is retrieved from the cache which
+    # returns 4 whereas it should be the second resource which returns 5, and the schema
+    # assert_node expects 5 now.
+    train_with_schema(train_schema, temp_cache)
+
+
+def test_resources_fingerprints_remain_after_being_cached(
+    tmp_path: Path,
+    temp_cache: LocalTrainingCache,
+    train_with_schema: Callable,
+    caplog: LogCaptureFixture,
+):
+    train_schema = GraphSchema(
+        {
+            "train": SchemaNode(
+                needs={},
+                uses=PersistableTestComponent,
+                fn="train",
+                constructor_name="create",
+                config={"test_value": "4"},
+                is_target=True,
+            ),
+            "process": SchemaNode(
+                needs={"resource": "train"},
+                uses=PersistableTestComponent,
+                fn="run_inference",
+                constructor_name="load",
+                config={},
+                is_target=True,
+            ),
+        }
+    )
+
+    # Train and cache.
+    train_with_schema(train_schema, temp_cache)
+
+    # We can determine if a cached `Resource` has a static fingerprint by comparing two
+    # subsequent cache entries of a child node.
+    import sqlalchemy as sa
+
+    with temp_cache._sessionmaker.begin() as session:
+        # This will get the cache entry for the "process" node.
+        query_for_most_recently_used_entry = sa.select(temp_cache.CacheEntry).order_by(
+            temp_cache.CacheEntry.last_used.desc()
+        )
+        entry = session.execute(query_for_most_recently_used_entry).scalars().first()
+        # The fingerprint key will incorporate the fingerprint of the `Resource`
+        # provided by the "train" node. We save this key to compare after the next run.
+        fingerprint_key = entry.fingerprint_key
+        # Deleting the entry will force it to be recreated next train.
+        delete_query = sa.delete(temp_cache.CacheEntry).where(
+            temp_cache.CacheEntry.fingerprint_key == fingerprint_key
+        )
+        session.execute(delete_query)
+
+    # In this second train, the Resource output of "train" will be retrieved from the
+    # cache.
+    train_with_schema(train_schema, temp_cache)
+
+    with temp_cache._sessionmaker.begin() as session:
+        # This will get the new cache entry for the "process" node.
+        query_for_most_recently_used_entry = sa.select(temp_cache.CacheEntry).order_by(
+            temp_cache.CacheEntry.last_used.desc()
+        )
+        entry = session.execute(query_for_most_recently_used_entry).scalars().first()
+        # Assert the fingerptint key of the new entry is the same. This confirms that
+        # the Resource from the cache has the same fingerprint.
+        assert entry.fingerprint_key == fingerprint_key
+
+
 @pytest.mark.parametrize(
     "on_before, on_after",
     [(lambda: True, lambda: 2 / 0), (lambda: 2 / 0, lambda: True)],


Code review comments:
diff_hunk: @@ -434,6 +445,7 @@ def _load_from_cache(
         result_type: Text,
         node_name: Text,
         model_storage: ModelStorage,
+        output_fingerprint_key,
path: rasa/engine/caching.py
text: @user1:
```suggestion
        output_fingerprint_key: Text,
```