Repository: RasaHQ/rasa
Pull request: Incremental Training inside Rasa Open Source
**Proposed changes**:
- Closes https://github.com/RasaHQ/rasa/issues/6971
- Adds the ability to load NLU components and Core policies in finetune mode which means they are initialized with a previously trained model and can be further trained on a mix of old and new training data.

Individual PRs:
- Changes to CLI - https://github.com/RasaHQ/rasa/issues/6971
- Applying constraints for finetuning - https://github.com/RasaHQ/rasa/pull/7427
   - Configuration of base model is exactly the same as configuration for finetuning except the number of epochs.
   - No new labels(intents, actions, entities and intent_response_keys) are added to the training data. New training examples for each of these can still be added.
   - Base model is trained with `MINIMUM_COMPATIBLE_VERSION` of currently installed Rasa Open Source version.
- Load NLU(https://github.com/RasaHQ/rasa/pull/7456) and Core(https://github.com/RasaHQ/rasa/pull/7458) pipelines in finetune mode.
- Loading NLU components and Core policies in finetune mode - https://github.com/RasaHQ/rasa/pull/7419 This PR includes running model regression tests to verify that performance of current models are not hurt by the extra featurization we do.
- Docs (https://github.com/RasaHQ/rasa/pull/7469)

**Status (please check what you already did)**:
- [x] added some tests for the functionality
- [x] updated the documentation
- [x] updated the changelog (please check [changelog](https://github.com/RasaHQ/rasa/tree/master/changelog) for instructions)
- [x] reformat files using `black` (please check [Readme](https://github.com/RasaHQ/rasa#code-style) for instructions)


Problem statement:
Incremental training
**Description of Problem**:
Once a model is trained it cannot be updated. It is not possible to continue training the model on new data that came in. Instead the model needs to be retrained from scratch, which takes up a lot of time.

**Overview of the Solution**:
It should be possible to load a model from a previous checkpoint and continue training with new data added.

Hints from issue discussion:
@tabergma how urgent is this one? 
@evgeniiaraz We want to tackle this issue this quarter. @dakshvar22 is leading this topic. Why do you ask?
@tabergma I wanted to work on it not to lose shape :)  but if it is urgent, I'll pick something non-essential

Based on the discussion in the [document](https://docs.google.com/document/d/1B7iVPRQONPoFssJfxhLMm5o23KWTPYNPQw_7XRc38Ao/edit), here are more fine-grained implementation tasks that are needed - 

**Changes to CLI and `rasa/train.py`**
1. Add a parameter to `rasa train` called `finetune_previous_model` which starts training in finetuning mode.
2. Add a parameter to `rasa train` called `finetune_model_path` which lets you specify the path to a previous model which should be finetuned.
3. `rasa.train_async_internal` should be refactored to check if training should proceed in finetuning mode(if `finetune_previous_model` is set to True). If yes, then it should check if it is possible to do so.(Check the doc for constraints for finetuning to be possible).
4. `rasa.nlu.train` should be refactored to create the `Trainer` object in fine-tune mode which means each component should be loaded with the model to be finetuned from. This will involve building the pipeline in a way similar to how the pipeline is built during inference, i.e. when `rasa shell` is run or `rasa test` is run.
5. `rasa.core.train` should also be refactored similarly as above.
6. Add telemetry event.

**Changes to ML components**

*CountVectorsFeaturizer*(CVF)

1. Add a parameter `max_additional_vocabulary_size` which lets users specify the additional buffer size that CVF should keep to accommodate new vocabulary tokens during fine-tuning.
2. `_train_with_independent_vocab` should be refactored to construct the vocabulary with the additional buffer specified above. Things to keep in mind here -

- When a new training cycle is triggered, the ordering of existing vocabulary tokens should not be changed and the new vocabulary tokens should only occupy the empty slots in the vocabulary.
- If the vocabulary size of CVF is exhausted, we should continue training, but warn the user that the vocabulary is exhausted and treat the new tokens which overflow as OOV tokens. At this point, the user should also be informed about the total vocabulary size of their dataset and should be prompted to retrain with full vocabulary.

*DIETClassifier, ResponseSelector and TEDPolicy*

1. `load()` should be refactored to load the models with weights in training mode and not in prediction mode. Currently, `_load_model()` builds the TF graph in predict mode which should be changed if the classifier is being loaded for finetuning. So instead of calling `_get_tf_call_model_function()`, `_get_tf_train_functions()` should be reused to build the graph for training.
2. Make sure the signature of `RasaModelData` in finetune mode is the same as what is constructed during training from scratch.
A working version(very draft) of the above steps is implemented on this [branch](https://github.com/RasaHQ/rasa/tree/incremental_training). From early observations, what needs to be improved/additionally done to make this mergeable as a feature -  

1. Ability to specify a model path to fine-tune from in the CLI.
2. Implement checks [here](https://github.com/RasaHQ/rasa/compare/incremental_training#diff-575b39b990e8be8f2064a83f51249b554d0887135090b2736fd975b95981d49dR197) to see if the previous model is compatible to be fine-tuned with the current configuration specified, for e.g., all parameters for the two configurations should be the same except number of epochs for training.
3. The above working version loads up the pipeline in fine-tune mode only for the NLU pipeline. Still needs to be done for the Core pipeline. The refactoring needed inside `TEDPolicy` is straightforward and identical to what is done for `DIETClassifier`. Loading up the instance of `Agent` class with the old model in fine-tune mode is what needs to be implemented.
4. While loading the NLU pipeline, currently the config of the loaded model is passed to the components, which means if I change the number of epochs in my new configuration, it is not used by the component. Will need to refactor that.
5. Make sure fine-tuning is possible for `rasa train nlu` and `rasa train core` as well. Currently it works for `rasa train`.

Of course docs, code quality and tests also need to be added.
Next steps based on the call with @dakshvar22 @joejuzl 
1. Create engineering issues from this (should be around 2-3 issues 🤔 )
2. Get started with the engineering issues in the week of November 23rd

Other things to keep in mind:
* Can we branch off `master` or does it make more sense to branch off the `e2e` branch?
I ran some initial experiments using the working version on this [branch](https://github.com/RasaHQ/rasa/tree/incremental_training) - 

**Setup**

**Data**: Financial Bot [NLU data](https://github.com/RasaHQ/financial-demo/blob/rasa-2-0/data/nlu/nlu.yml) split into 80:20 train test split. The train split is further divided into 2 sets - split 80:20. The first set is used for training an initial model from scratch. The second set is used for finetuning the first model trained. Consider the second set as new annotations that a user added to their training data.

Size of Set 1: 233
Size of Set 2: 59
Size of held-out test set: 73

**Training**: We train the first model from scratch for 100 epochs. Then add the second set to the training data and further train the first model for 30 more epochs.

[Config](https://github.com/RasaHQ/financial-demo/blob/rasa-2-0/config.yml)

*Note*: Finetuning is done by mixing the new data with the old data and then training on batches from the combined data.

Results:

**Initial Model**|**Training data**|**Number of epochs**|**Intent F1(held out test set)**|**Entity F1(held out test set)**|**Time for training**
:-----:|:-----:|:-----:|:-----:|:-----:|:-----:
Randomly initialized| Set 1| 100| 0.753| 0.9| 48s
Model trained on set 1| Set 1 + Set 2| 30| 0.861| 0.927| 16s
Randomly initialized| Set 1 + Set 2| 130| 0.876| 0.911| 1 min 16s
Experiments on Sara data - 

Size of Set 1: 3166
Size of Set 2: 792
Size of held-out test set: 990

[Config](https://github.com/RasaHQ/rasa-demo/blob/master/config.yml)

*Note*: `additional_vocabulary_size` was set to `1000` for char based CVF and `100` for word based CVF.

Results:

**Initial Model**|**Training data**|**Number of epochs**|**Intent F1**|**Entity F1**|**Response F1**|**Time for training**
:-----:|:-----:|:-----:|:-----:|:-----:|:-----:|:-----:
Randomly initialized| Set 1| 40| 0.789| 0.832| 0.927| 4m 10s
Model trained on set 1| Set 1 + Set 2| 10| 0.823| 0.861| 0.935| 1m 39s
Randomly initialized| Set 1 + Set 2| 50| 0.818| 0.854| 0.938| 6m 2s
@dakshvar22 Do I understand it correctly that the incremental training is in total faster than training everything at once? This seems somewhat counterintuitive for me as I'd expect overhead from loading training data / pipelines etc.
@wochinge The time mentioned above are the time to train `DIETClassifier` alone and does not include the pipeline and training data loading time. We should measure that too, but it would be much smaller in comparison to the amount of time required to train `DIETClassifier` for an additional 40/70 epochs as shown in the examples above.

Commit message: Add migration guide for policies (#7522)

* Add migration guide for policies

* spelling fix

* changelog

Patch to review:
diff --git a/changelog/6971.feature.md b/changelog/6971.feature.md
new file mode 100644
index 000000000000..dc1532850d48
--- /dev/null
+++ b/changelog/6971.feature.md
@@ -0,0 +1,14 @@
+Incremental training of models in a pipeline is now supported.
+
+If you have added new NLU training examples or new stories/rules for  
+dialogue manager, you don't need to train the pipeline from scratch.
+Instead, you can initialize the pipeline with a previously trained model
+and continue finetuning the model on the complete dataset consisting of
+new training examples. To do so, use `rasa train --finetune`. For more
+detailed explanation of the command, check out the docs on [incremental  
+training](./command-line-interface.mdx#incremental-training).
+
+Added a configuration parameter `additional_vocabulary_size` to  
+[`CountVectorsFeaturizer`](./components.mdx#countvectorsfeaturizer)  
+and `number_additional_patterns` to [`RegexFeaturizer`](./components.mdx#regexfeaturizer).
+These parameters are useful to configure when using incremental training for your pipelines.
\ No newline at end of file
diff --git a/changelog/7458.removal.md b/changelog/7458.removal.md
new file mode 100644
index 000000000000..0d52dcce642b
--- /dev/null
+++ b/changelog/7458.removal.md
@@ -0,0 +1,2 @@
+Interfaces for `Policy.__init__` and `Policy.load` have changed.
+See [migration guide](./migration-guide.mdx#rasa-21-to-rasa-22) for details.
diff --git a/data/test_domains/default_with_slots_and_no_actions.yml b/data/test_domains/default_with_slots_and_no_actions.yml
index 3bf7fc9aa31e..d7234be0f866 100644
--- a/data/test_domains/default_with_slots_and_no_actions.yml
+++ b/data/test_domains/default_with_slots_and_no_actions.yml
@@ -1,3 +1,5 @@
+version: "2.0"
+
 # all hashtags are comments :)
 intents:
  - greet
diff --git a/docs/docs/command-line-interface.mdx b/docs/docs/command-line-interface.mdx
index fe326a35cc24..e6c19b7b0bb2 100644
--- a/docs/docs/command-line-interface.mdx
+++ b/docs/docs/command-line-interface.mdx
@@ -85,6 +85,51 @@ The following arguments can be used to configure the training process:
 ```text [rasa train --help]
 ```
 
+### Incremental training
+
+In order to improve the performance of an assistant, it's helpful to practice [CDD](./conversation-driven-development.mdx)
+and add new training examples based on how your users have talked to your assistant. You can use `rasa train --finetune`
+to initialize the pipeline with an already trained model and further finetune it on the
+new training dataset that includes the additional training examples. This will help reduce the
+training time of the new model.
+
+By default, the command picks up the latest model in the `models/` directory. If you have a specific model
+which you want to improve, you may specify the path to this by
+running `rasa train --finetune <path to model to finetune>`. Finetuning a model usually
+requires fewer epochs to train machine learning components like `DIETClassifier`, `ResponseSelector` and `TEDPolicy` compared to training from scratch.
+Either use a model configuration for finetuning
+which defines fewer epochs than before or use the flag
+`--epoch-fraction`. `--epoch-fraction` will use a fraction of the epochs specified for each machine learning component
+in the model configuration file. For example, if `DIETClassifier` is configured to use 100 epochs,
+specifying `--epoch-fraction 0.5` will only use 50 epochs for finetuning.
+
+You can also finetune an NLU-only or dialogue management-only model by using
+`rasa train nlu --finetune` and `rasa train core --finetune` respectively.
+
+To be able to fine tune a model, the following conditions must be met:
+
+1. The configuration supplied should be exactly the same as the
+configuration used to train the model which is being finetuned.
+The only parameter that you can change is `epochs` for the individual machine learning components and policies.
+
+2. The set of labels(intents, actions, entities and slots) for which the base model is trained
+should be exactly the same as the ones present in the training data used for finetuning. This
+means that you cannot add new intent, action, entity or slot labels to your training data
+during incremental training. You can still add new training examples for each of the existing
+labels. If you have added/removed labels in the training data, the pipeline needs to be trained
+from scratch.
+
+3. The model to be finetuned is trained with `MINIMUM_COMPATIBLE_VERSION` of the currently installed rasa version.
+
+Checkout the docs for [`CountVectorsFeaturizer`](./components.mdx#countvectorsfeaturizer) and
+[`RegexFeaturizer`](./components.mdx#regexfeaturizer) to understand how to configure them appropriately for incremental training.
+
+:::note
+Finetuned models are expected to be on-par with performance of models trained from scratch. However,
+make sure to train your pipelines from scratch frequently to avoid running out of additional
+vocabulary slots for the models.
+:::
+
 ## rasa interactive
 
 You can [use Rasa X in local mode](https://rasa.com/docs/rasa-x) to do interactive learning in a UI,
diff --git a/docs/docs/components.mdx b/docs/docs/components.mdx
index e7870aaf229e..427f6590f8c6 100644
--- a/docs/docs/components.mdx
+++ b/docs/docs/components.mdx
@@ -836,6 +836,28 @@ Note: The `feature-dimension` for sequence and sentence features does not have t
     "use_word_boundaries": True
   ```
 
+  **Configuring for incremental training**
+
+  To ensure that `sparse_features` are of fixed size during
+  [incremental training](./command-line-interface.mdx#incremental-training), the
+  component should be configured to account for additional patterns that may be
+  added to the training data in future. To do so, configure the `number_additional_patterns`
+  parameter while training the base model from scratch:
+
+  ```yaml-rasa {3}
+  pipeline:
+  - name: RegexFeaturizer
+    number_additional_patterns: 10
+  ```
+
+  If not configured by the user, the component will account for a minimum of 10 additional
+  patterns and a maximum of twice the number of patterns currently
+  present in the training data (including lookup tables and regex patterns).
+  Once the component runs out of additional pattern slots
+  during incremental training, the new patterns are dropped
+  and not considered during featurization. At this point, it is advisable
+  to retrain a new model from scratch.
+
 
 ### CountVectorsFeaturizer
 
@@ -960,58 +982,91 @@ Note: The `feature-dimension` for sequence and sentence features does not have t
     "use_shared_vocab": False
   ```
 
+  **Configuring for incremental training**
+
+  To ensure that `sparse_features` are of fixed size during
+  [incremental training](./command-line-interface.mdx#incremental-training), the
+  component should be configured to account for additional vocabulary tokens
+  that may be added as part of new training examples in the future.
+  To do so, configure the `additional_vocabulary_size` parameter while training the base model from scratch:
+
+  ```yaml-rasa {3-6}
+  pipeline:
+  - name: CountVectorsFeaturizer
+    additional_vocabulary_size:
+      text: 1000
+      response: 1000
+      action_text: 1000
+  ```
+
+  As in the above example, you can define additional vocabulary size for each of
+  `text` (user messages), `response` (bot responses used by `ResponseSelector`) and
+  `action_text` (bot responses not used by `ResponseSelector`). If you are building a shared
+  vocabulary (`use_shared_vocab=True`), you only need to define a value for the `text` attribute.
+  If any of the attributes is not configured by the user, the component will
+  account for a minimum of 1000 additional vocabulary slots and a maximum of half the current vocabulary size.
+  Once the component runs out of additional available vocabulary slots
+  during incremental training, the new vocabulary tokens are dropped
+  and not considered during featurization. At this point, it is advisable
+  to retrain a new model from scratch.
+
+
 The above configuration parameters are the ones you should configure to fit your model to your data.
 However, additional parameters exist that can be adapted.
 
 <details><summary>More configurable parameters</summary>
 
 ```
-+-------------------+-------------------------+--------------------------------------------------------------+
-| Parameter         | Default Value           | Description                                                  |
-+===================+=========================+==============================================================+
-| use_shared_vocab  | False                   | If set to 'True' a common vocabulary is used for labels      |
-|                   |                         | and user message.                                            |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| analyzer          | word                    | Whether the features should be made of word n-gram or        |
-|                   |                         | character n-grams. Option 'char_wb' creates character        |
-|                   |                         | n-grams only from text inside word boundaries;               |
-|                   |                         | n-grams at the edges of words are padded with space.         |
-|                   |                         | Valid values: 'word', 'char', 'char_wb'.                     |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| strip_accents     | None                    | Remove accents during the pre-processing step.               |
-|                   |                         | Valid values: 'ascii', 'unicode', 'None'.                    |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| stop_words        | None                    | A list of stop words to use.                                 |
-|                   |                         | Valid values: 'english' (uses an internal list of            |
-|                   |                         | English stop words), a list of custom stop words, or         |
-|                   |                         | 'None'.                                                      |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| min_df            | 1                       | When building the vocabulary ignore terms that have a        |
-|                   |                         | document frequency strictly lower than the given threshold.  |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| max_df            | 1                       | When building the vocabulary ignore terms that have a        |
-|                   |                         | document frequency strictly higher than the given threshold  |
-|                   |                         | (corpus-specific stop words).                                |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| min_ngram         | 1                       | The lower boundary of the range of n-values for different    |
-|                   |                         | word n-grams or char n-grams to be extracted.                |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| max_ngram         | 1                       | The upper boundary of the range of n-values for different    |
-|                   |                         | word n-grams or char n-grams to be extracted.                |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| max_features      | None                    | If not 'None', build a vocabulary that only consider the top |
-|                   |                         | max_features ordered by term frequency across the corpus.    |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| lowercase         | True                    | Convert all characters to lowercase before tokenizing.       |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| OOV_token         | None                    | Keyword for unseen words.                                    |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| OOV_words         | []                      | List of words to be treated as 'OOV_token' during training.  |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| alias             | CountVectorFeaturizer   | Alias name of featurizer.                                    |
-+-------------------+-------------------------+--------------------------------------------------------------+
-| use_lemma         | True                    | Use the lemma of words for featurization.                    |
-+-------------------+-------------------------+--------------------------------------------------------------+
++---------------------------+-------------------------+--------------------------------------------------------------+
+| Parameter                 | Default Value           | Description                                                  |
++===========================+=========================+==============================================================+
+| use_shared_vocab          | False                   | If set to 'True' a common vocabulary is used for labels      |
+|                           |                         | and user message.                                            |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| analyzer                  | word                    | Whether the features should be made of word n-gram or        |
+|                           |                         | character n-grams. Option 'char_wb' creates character        |
+|                           |                         | n-grams only from text inside word boundaries;               |
+|                           |                         | n-grams at the edges of words are padded with space.         |
+|                           |                         | Valid values: 'word', 'char', 'char_wb'.                     |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| strip_accents             | None                    | Remove accents during the pre-processing step.               |
+|                           |                         | Valid values: 'ascii', 'unicode', 'None'.                    |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| stop_words                | None                    | A list of stop words to use.                                 |
+|                           |                         | Valid values: 'english' (uses an internal list of            |
+|                           |                         | English stop words), a list of custom stop words, or         |
+|                           |                         | 'None'.                                                      |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| min_df                    | 1                       | When building the vocabulary ignore terms that have a        |
+|                           |                         | document frequency strictly lower than the given threshold.  |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| max_df                    | 1                       | When building the vocabulary ignore terms that have a        |
+|                           |                         | document frequency strictly higher than the given threshold  |
+|                           |                         | (corpus-specific stop words).                                |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| min_ngram                 | 1                       | The lower boundary of the range of n-values for different    |
+|                           |                         | word n-grams or char n-grams to be extracted.                |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| max_ngram                 | 1                       | The upper boundary of the range of n-values for different    |
+|                           |                         | word n-grams or char n-grams to be extracted.                |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| max_features              | None                    | If not 'None', build a vocabulary that only consider the top |
+|                           |                         | max_features ordered by term frequency across the corpus.    |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| lowercase                 | True                    | Convert all characters to lowercase before tokenizing.       |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| OOV_token                 | None                    | Keyword for unseen words.                                    |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| OOV_words                 | []                      | List of words to be treated as 'OOV_token' during training.  |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| alias                     | CountVectorFeaturizer   | Alias name of featurizer.                                    |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| use_lemma                 | True                    | Use the lemma of words for featurization.                    |
++---------------------------+-------------------------+--------------------------------------------------------------+
+| additional_vocabulary_size| text: 1000              | Size of additional vocabulary to account for incremental     |
+|                           | response: 1000          | training while training a model from scratch                 |
+|                           | action_text: 1000       |                                                              |
++---------------------------+-------------------------+--------------------------------------------------------------+
 ```
 
 </details>
diff --git a/docs/docs/migration-guide.mdx b/docs/docs/migration-guide.mdx
index b851feb34772..c373a3302413 100644
--- a/docs/docs/migration-guide.mdx
+++ b/docs/docs/migration-guide.mdx
@@ -10,6 +10,15 @@ description: |
 This page contains information about changes between major versions and
 how you can migrate from one version to another.
 
+## Rasa 2.1 to Rasa 2.2
+
+### Policies
+
+[Policies](./policies.mdx) now require a `**kwargs` argument in their constructor and `load` method.
+Policies without `**kwargs` will be supported until Rasa version `3.0.0`.
+However when using [incremental training](./command-line-interface.mdx#incremental-training)
+`**kwargs` **must** be included.
+
 ## Rasa 2.0 to Rasa 2.1
 
 ### Deprecations
diff --git a/docs/docs/telemetry/events.json b/docs/docs/telemetry/events.json
index 3a0fcf9b4cf0..92ca49cd82d0 100644
--- a/docs/docs/telemetry/events.json
+++ b/docs/docs/telemetry/events.json
@@ -88,6 +88,10 @@
         "num_regexes": {
           "type": "integer",
           "description": "Total number of regexes defined."
+        },
+        "is_finetuning": {
+          "type": "boolean",
+          "description": "True if a model is trained by finetuning an existing model."
         }
       },
       "additionalProperties": false,
diff --git a/rasa/cli/arguments/interactive.py b/rasa/cli/arguments/interactive.py
index c4ed2a6ac17e..564421b87b39 100644
--- a/rasa/cli/arguments/interactive.py
+++ b/rasa/cli/arguments/interactive.py
@@ -1,50 +1,41 @@
 import argparse
 import uuid
 
-from rasa.cli.arguments.default_arguments import (
-    add_domain_param,
-    add_stories_param,
-    add_model_param,
-    add_endpoint_param,
-)
-from rasa.cli.arguments.train import (
-    add_force_param,
-    add_data_param,
-    add_config_param,
-    add_out_param,
-    add_debug_plots_param,
-    add_augmentation_param,
-    add_persist_nlu_data_param,
-)
-from rasa.cli.arguments.run import add_port_argument
+import rasa.cli.arguments.default_arguments
+import rasa.cli.arguments.train
+import rasa.cli.arguments.run
 
 
 def set_interactive_arguments(parser: argparse.ArgumentParser) -> None:
+    """Specifies arguments for `rasa interactive`."""
     parser.add_argument(
         "--e2e",
         action="store_true",
         help="Save story files in e2e format. In this format user messages "
         "will be included in the stories.",
     )
-    add_port_argument(parser)
+    rasa.cli.arguments.run.add_port_argument(parser)
 
-    add_model_param(parser, default=None)
-    add_data_param(parser)
+    rasa.cli.arguments.default_arguments.add_model_param(parser, default=None)
+    rasa.cli.arguments.train.add_data_param(parser)
 
     _add_common_params(parser)
     train_arguments = _add_training_arguments(parser)
 
-    add_force_param(train_arguments)
-    add_persist_nlu_data_param(train_arguments)
+    rasa.cli.arguments.train.add_force_param(train_arguments)
+    rasa.cli.arguments.train.add_persist_nlu_data_param(train_arguments)
 
 
 def set_interactive_core_arguments(parser: argparse.ArgumentParser) -> None:
-    add_model_param(parser, model_name="Rasa Core", default=None)
-    add_stories_param(parser)
+    """Specifies arguments for `rasa interactive core`."""
+    rasa.cli.arguments.default_arguments.add_model_param(
+        parser, model_name="Rasa Core", default=None
+    )
+    rasa.cli.arguments.default_arguments.add_stories_param(parser)
 
     _add_common_params(parser)
     _add_training_arguments(parser)
-    add_port_argument(parser)
+    rasa.cli.arguments.run.add_port_argument(parser)
 
 
 def _add_common_params(parser: argparse.ArgumentParser) -> None:
@@ -62,7 +53,7 @@ def _add_common_params(parser: argparse.ArgumentParser) -> None:
         "UUID that will be randomly generated.",
     )
 
-    add_endpoint_param(
+    rasa.cli.arguments.default_arguments.add_endpoint_param(
         parser,
         help_text="Configuration file for the model server and the connectors as a yml file.",
     )
@@ -71,12 +62,12 @@ def _add_common_params(parser: argparse.ArgumentParser) -> None:
 # noinspection PyProtectedMember
 def _add_training_arguments(parser: argparse.ArgumentParser) -> argparse._ArgumentGroup:
     train_arguments = parser.add_argument_group("Train Arguments")
-    add_config_param(train_arguments)
-    add_domain_param(train_arguments)
-    add_out_param(
+    rasa.cli.arguments.train.add_config_param(train_arguments)
+    rasa.cli.arguments.default_arguments.add_domain_param(train_arguments)
+    rasa.cli.arguments.train.add_out_param(
         train_arguments, help_text="Directory where your models should be stored."
     )
-    add_augmentation_param(train_arguments)
-    add_debug_plots_param(train_arguments)
-
+    rasa.cli.arguments.train.add_augmentation_param(train_arguments)
+    rasa.cli.arguments.train.add_debug_plots_param(train_arguments)
+    rasa.cli.arguments.train.add_finetune_params(train_arguments)
     return train_arguments
diff --git a/rasa/cli/arguments/train.py b/rasa/cli/arguments/train.py
index 55c96cb85ec9..9344f19720b6 100644
--- a/rasa/cli/arguments/train.py
+++ b/rasa/cli/arguments/train.py
@@ -10,8 +10,11 @@
 )
 from rasa.shared.constants import DEFAULT_CONFIG_PATH, DEFAULT_DATA_PATH
 
+USE_LATEST_MODEL_FOR_FINE_TUNING = True
 
-def set_train_arguments(parser: argparse.ArgumentParser):
+
+def set_train_arguments(parser: argparse.ArgumentParser) -> None:
+    """Specifies CLI arguments for `rasa train`."""
     add_data_param(parser)
     add_config_param(parser)
     add_domain_param(parser)
@@ -21,17 +24,19 @@ def set_train_arguments(parser: argparse.ArgumentParser):
     add_augmentation_param(parser)
     add_debug_plots_param(parser)
 
-    add_num_threads_param(parser)
+    _add_num_threads_param(parser)
 
-    add_model_name_param(parser)
+    _add_model_name_param(parser)
     add_persist_nlu_data_param(parser)
     add_force_param(parser)
+    add_finetune_params(parser)
 
 
-def set_train_core_arguments(parser: argparse.ArgumentParser):
+def set_train_core_arguments(parser: argparse.ArgumentParser) -> None:
+    """Specifies CLI arguments for `rasa train core`."""
     add_stories_param(parser)
     add_domain_param(parser)
-    add_core_config_param(parser)
+    _add_core_config_param(parser)
     add_out_param(parser, help_text="Directory where your models should be stored.")
 
     add_augmentation_param(parser)
@@ -39,26 +44,32 @@ def set_train_core_arguments(parser: argparse.ArgumentParser):
 
     add_force_param(parser)
 
-    add_model_name_param(parser)
+    _add_model_name_param(parser)
 
     compare_arguments = parser.add_argument_group("Comparison Arguments")
-    add_compare_params(compare_arguments)
+    _add_compare_params(compare_arguments)
+    add_finetune_params(parser)
 
 
-def set_train_nlu_arguments(parser: argparse.ArgumentParser):
+def set_train_nlu_arguments(parser: argparse.ArgumentParser) -> None:
+    """Specifies CLI arguments for `rasa train nlu`."""
     add_config_param(parser)
     add_domain_param(parser, default=None)
     add_out_param(parser, help_text="Directory where your models should be stored.")
 
     add_nlu_data_param(parser, help_text="File or folder containing your NLU data.")
 
-    add_num_threads_param(parser)
+    _add_num_threads_param(parser)
 
-    add_model_name_param(parser)
+    _add_model_name_param(parser)
     add_persist_nlu_data_param(parser)
+    add_finetune_params(parser)
 
 
-def add_force_param(parser: Union[argparse.ArgumentParser, argparse._ActionsContainer]):
+def add_force_param(
+    parser: Union[argparse.ArgumentParser, argparse._ActionsContainer]
+) -> None:
+    """Specifies if the model should be trained from scratch."""
     parser.add_argument(
         "--force",
         action="store_true",
@@ -66,7 +77,10 @@ def add_force_param(parser: Union[argparse.ArgumentParser, argparse._ActionsCont
     )
 
 
-def add_data_param(parser: Union[argparse.ArgumentParser, argparse._ActionsContainer]):
+def add_data_param(
+    parser: Union[argparse.ArgumentParser, argparse._ActionsContainer]
+) -> None:
+    """Specifies path to training data."""
     parser.add_argument(
         "--data",
         default=[DEFAULT_DATA_PATH],
@@ -75,7 +89,7 @@ def add_data_param(parser: Union[argparse.ArgumentParser, argparse._ActionsConta
     )
 
 
-def add_core_config_param(parser: argparse.ArgumentParser):
+def _add_core_config_param(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "-c",
         "--config",
@@ -87,9 +101,9 @@ def add_core_config_param(parser: argparse.ArgumentParser):
     )
 
 
-def add_compare_params(
+def _add_compare_params(
     parser: Union[argparse.ArgumentParser, argparse._ActionsContainer]
-):
+) -> None:
     parser.add_argument(
         "--percentages",
         nargs="*",
@@ -128,8 +142,8 @@ def add_dry_run_param(
 
 def add_augmentation_param(
     parser: Union[argparse.ArgumentParser, argparse._ActionsContainer]
-):
-    """Adds `--augmentation` argument to a specified `parser`.
+) -> None:
+    """Sets the augmentation factor for the Core training.
 
     Args:
         parser: An instance of `ArgumentParser` or `_ActionsContainer`.
@@ -144,7 +158,8 @@ def add_augmentation_param(
 
 def add_debug_plots_param(
     parser: Union[argparse.ArgumentParser, argparse._ActionsContainer]
-):
+) -> None:
+    """Specifies if conversation flow should be visualized."""
     parser.add_argument(
         "--debug-plots",
         default=False,
@@ -155,9 +170,9 @@ def add_debug_plots_param(
     )
 
 
-def add_num_threads_param(
+def _add_num_threads_param(
     parser: Union[argparse.ArgumentParser, argparse._ActionsContainer]
-):
+) -> None:
     parser.add_argument(
         "--num-threads",
         type=int,
@@ -166,7 +181,7 @@ def add_num_threads_param(
     )
 
 
-def add_model_name_param(parser: argparse.ArgumentParser):
+def _add_model_name_param(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "--fixed-model-name",
         type=str,
@@ -177,9 +192,35 @@ def add_model_name_param(parser: argparse.ArgumentParser):
 
 def add_persist_nlu_data_param(
     parser: Union[argparse.ArgumentParser, argparse._ActionsContainer]
-):
+) -> None:
+    """Adds parameters for persisting the NLU training data with the model."""
     parser.add_argument(
         "--persist-nlu-data",
         action="store_true",
-        help="Persist the nlu training data in the saved model.",
+        help="Persist the NLU training data in the saved model.",
+    )
+
+
+def add_finetune_params(
+    parser: Union[argparse.ArgumentParser, argparse._ActionsContainer]
+) -> None:
+    """Adds parameters for model finetuning."""
+    parser.add_argument(
+        "--finetune",
+        nargs="?",
+        # If the user doesn't specify `--finetune` at all
+        default=None,
+        # If the user only specifies `--finetune` without an additional path
+        const=USE_LATEST_MODEL_FOR_FINE_TUNING,
+        help="Fine-tune a previously trained model. If no model path is provided, Rasa "
+        "Open Source will try to finetune the latest trained model from the "
+        "model directory specified via '--out'.",
+    )
+
+    parser.add_argument(
+        "--epoch-fraction",
+        type=float,
+        default=1.0,
+        help="Fraction of epochs which are currently specified in the model "
+        "configuration which should be used when finetuning a model.",
     )
diff --git a/rasa/cli/train.py b/rasa/cli/train.py
index e9b735f6a749..b7b58428fc43 100644
--- a/rasa/cli/train.py
+++ b/rasa/cli/train.py
@@ -98,6 +98,8 @@ def train(args: argparse.Namespace, can_exit: bool = False) -> Optional[Text]:
         persist_nlu_training_data=args.persist_nlu_data,
         core_additional_arguments=extract_core_additional_arguments(args),
         nlu_additional_arguments=extract_nlu_additional_arguments(args),
+        model_to_finetune=_model_for_finetuning(args),
+        finetuning_epoch_fraction=args.epoch_fraction,
     )
     if training_result.code != 0 and can_exit:
         sys.exit(training_result.code)
@@ -105,14 +107,22 @@ def train(args: argparse.Namespace, can_exit: bool = False) -> Optional[Text]:
     return training_result.model
 
 
+def _model_for_finetuning(args: argparse.Namespace) -> Optional[Text]:
+    if args.finetune == train_arguments.USE_LATEST_MODEL_FOR_FINE_TUNING:
+        # We use this constant to signal that the user specified `--finetune` but
+        # didn't provide a path to a model. In this case we try to load the latest
+        # model from the output directory (that's usually models/).
+        return args.out
+
+
 def train_core(
     args: argparse.Namespace, train_path: Optional[Text] = None
 ) -> Optional[Text]:
-    """Trains a Core model.
+    """Trains a Rasa Core model only.
 
     Args:
-        args: Namespace arguments.
-        train_path: Directory where models should be stored.
+        args: Command-line arguments to configure training.
+        train_path: Path where trained model but not unzipped model should be stored.
 
     Returns:
         Path to a trained model or `None` if training was not successful.
@@ -145,6 +155,8 @@ def train_core(
             train_path=train_path,
             fixed_model_name=args.fixed_model_name,
             additional_arguments=additional_arguments,
+            model_to_finetune=_model_for_finetuning(args),
+            finetuning_epoch_fraction=args.epoch_fraction,
         )
     else:
         from rasa.core.train import do_compare_training
@@ -189,6 +201,8 @@ def train_nlu(
         persist_nlu_training_data=args.persist_nlu_data,
         additional_arguments=extract_nlu_additional_arguments(args),
         domain=args.domain,
+        model_to_finetune=_model_for_finetuning(args),
+        finetuning_epoch_fraction=args.epoch_fraction,
     )
 
 
diff --git a/rasa/core/agent.py b/rasa/core/agent.py
index 1ac70508d22b..28e5873e7d9f 100644
--- a/rasa/core/agent.py
+++ b/rasa/core/agent.py
@@ -22,7 +22,7 @@
     DEFAULT_DOMAIN_PATH,
     DEFAULT_CORE_SUBDIRECTORY_NAME,
 )
-from rasa.shared.exceptions import InvalidParameterException, RasaException
+from rasa.shared.exceptions import InvalidParameterException
 from rasa.shared.nlu.interpreter import NaturalLanguageInterpreter, RegexInterpreter
 from rasa.core.lock_store import InMemoryLockStore, LockStore
 from rasa.core.nlg import NaturalLanguageGenerator
@@ -411,6 +411,8 @@ def load(
         model_server: Optional[EndpointConfig] = None,
         remote_storage: Optional[Text] = None,
         path_to_model_archive: Optional[Text] = None,
+        new_config: Optional[Dict] = None,
+        finetuning_epoch_fraction: float = 1.0,
     ) -> "Agent":
         """Load a persisted model from the passed path."""
         try:
@@ -441,7 +443,15 @@ def load(
 
         if core_model:
             domain = Domain.load(os.path.join(core_model, DEFAULT_DOMAIN_PATH))
-            ensemble = PolicyEnsemble.load(core_model) if core_model else None
+            ensemble = (
+                PolicyEnsemble.load(
+                    core_model,
+                    new_config=new_config,
+                    finetuning_epoch_fraction=finetuning_epoch_fraction,
+                )
+                if core_model
+                else None
+            )
 
             # ensures the domain hasn't changed between test and train
             domain.compare_with_specification(core_model)
diff --git a/rasa/core/policies/ensemble.py b/rasa/core/policies/ensemble.py
index e7b72d8778c1..dfbce47c3b9e 100644
--- a/rasa/core/policies/ensemble.py
+++ b/rasa/core/policies/ensemble.py
@@ -1,12 +1,13 @@
 import importlib
 import json
 import logging
+import math
 import os
 import sys
 from collections import defaultdict
 from datetime import datetime
 from pathlib import Path
-from typing import Text, Optional, Any, List, Dict, Tuple, NamedTuple, Union
+from typing import Text, Optional, Any, List, Dict, Tuple, Type, Union
 
 import rasa.core
 import rasa.core.training.training
@@ -41,6 +42,7 @@
 from rasa.shared.core.trackers import DialogueStateTracker
 from rasa.shared.core.generator import TrackerWithCachedStates
 from rasa.core import registry
+from rasa.utils.tensorflow.constants import EPOCHS
 
 logger = logging.getLogger(__name__)
 
@@ -302,10 +304,29 @@ def _ensure_loaded_policy(cls, policy, policy_cls, policy_name: Text):
                 "".format(policy_name)
             )
 
-    @classmethod
-    def load(cls, path: Union[Text, Path]) -> "PolicyEnsemble":
-        """Loads policy and domain specification from storage"""
+    @staticmethod
+    def _get_updated_epochs(
+        policy_cls: Type[Policy],
+        config_for_policy: Dict[Text, Any],
+        finetuning_epoch_fraction: float,
+    ) -> Optional[int]:
+        if EPOCHS in config_for_policy:
+            epochs = config_for_policy[EPOCHS]
+        else:
+            try:
+                epochs = policy_cls.defaults[EPOCHS]
+            except (KeyError, AttributeError):
+                return None
+        return math.ceil(epochs * finetuning_epoch_fraction)
 
+    @classmethod
+    def load(
+        cls,
+        path: Union[Text, Path],
+        new_config: Optional[Dict] = None,
+        finetuning_epoch_fraction: float = 1.0,
+    ) -> "PolicyEnsemble":
+        """Loads policy and domain specification from disk."""
         metadata = cls.load_metadata(path)
         cls.ensure_model_compatibility(metadata)
         policies = []
@@ -313,9 +334,38 @@ def load(cls, path: Union[Text, Path]) -> "PolicyEnsemble":
             policy_cls = registry.policy_from_module_path(policy_name)
             dir_name = f"policy_{i}_{policy_cls.__name__}"
             policy_path = os.path.join(path, dir_name)
-            policy = policy_cls.load(policy_path)
+
+            context = {}
+            if new_config:
+                context["should_finetune"] = True
+
+                config_for_policy = new_config["policies"][i]
+                epochs = cls._get_updated_epochs(
+                    policy_cls, config_for_policy, finetuning_epoch_fraction
+                )
+                if epochs:
+                    context["epoch_override"] = epochs
+
+            if "kwargs" not in rasa.shared.utils.common.arguments_of(policy_cls.load):
+                if context:
+                    raise UnsupportedDialogueModelError(
+                        f"`{policy_cls.__name__}.{policy_cls.load.__name__}` does not "
+                        f"accept `**kwargs`. Attempting to pass {context} to the "
+                        f"policy. `**kwargs` should be added to all policies by "
+                        f"Rasa Open Source 3.0.0."
+                    )
+                else:
+                    rasa.shared.utils.io.raise_deprecation_warning(
+                        f"`{policy_cls.__name__}.{policy_cls.load.__name__}` does not "
+                        f"accept `**kwargs`. `**kwargs` are required for contextual "
+                        f"information e.g. the flag `should_finetune`.",
+                        warn_until_version="3.0.0",
+                    )
+
+            policy = policy_cls.load(policy_path, **context)
             cls._ensure_loaded_policy(policy, policy_cls, policy_name)
             policies.append(policy)
+
         ensemble_cls = rasa.shared.utils.common.class_from_module_path(
             metadata["ensemble_name"]
         )
diff --git a/rasa/core/policies/fallback.py b/rasa/core/policies/fallback.py
index aeef4f834fe4..9ddb96b4ff84 100644
--- a/rasa/core/policies/fallback.py
+++ b/rasa/core/policies/fallback.py
@@ -39,10 +39,12 @@ def __init__(
         ambiguity_threshold: float = DEFAULT_NLU_FALLBACK_AMBIGUITY_THRESHOLD,
         core_threshold: float = DEFAULT_CORE_FALLBACK_THRESHOLD,
         fallback_action_name: Text = ACTION_DEFAULT_FALLBACK_NAME,
+        **kwargs: Any,
     ) -> None:
         """Create a new Fallback policy.
 
         Args:
+            priority: Fallback policy priority.
             core_threshold: if NLU confidence threshold is met,
                 predict fallback action with confidence `core_threshold`.
                 If this is the highest confidence in the ensemble,
@@ -54,7 +56,7 @@ def __init__(
                 between confidences of the top two predictions
             fallback_action_name: name of the action to execute as a fallback
         """
-        super().__init__(priority=priority)
+        super().__init__(priority=priority, **kwargs)
 
         self.nlu_threshold = nlu_threshold
         self.ambiguity_threshold = ambiguity_threshold
diff --git a/rasa/core/policies/form_policy.py b/rasa/core/policies/form_policy.py
index ba75d8e20421..5fcb2f7a2db1 100644
--- a/rasa/core/policies/form_policy.py
+++ b/rasa/core/policies/form_policy.py
@@ -35,12 +35,17 @@ def __init__(
         featurizer: Optional[TrackerFeaturizer] = None,
         priority: int = FORM_POLICY_PRIORITY,
         lookup: Optional[Dict] = None,
+        **kwargs: Any,
     ) -> None:
 
         # max history is set to 2 in order to capture
         # previous meaningful action before action listen
         super().__init__(
-            featurizer=featurizer, priority=priority, max_history=2, lookup=lookup
+            featurizer=featurizer,
+            priority=priority,
+            max_history=2,
+            lookup=lookup,
+            **kwargs,
         )
 
         rasa.shared.utils.io.raise_deprecation_warning(
diff --git a/rasa/core/policies/mapping_policy.py b/rasa/core/policies/mapping_policy.py
index 3bfa83c60d42..fac4492a0df7 100644
--- a/rasa/core/policies/mapping_policy.py
+++ b/rasa/core/policies/mapping_policy.py
@@ -43,10 +43,9 @@ class MappingPolicy(Policy):
     def _standard_featurizer() -> None:
         return None
 
-    def __init__(self, priority: int = MAPPING_POLICY_PRIORITY) -> None:
+    def __init__(self, priority: int = MAPPING_POLICY_PRIORITY, **kwargs: Any) -> None:
         """Create a new Mapping policy."""
-
-        super().__init__(priority=priority)
+        super().__init__(priority=priority, **kwargs)
 
         rasa.shared.utils.io.raise_deprecation_warning(
             f"'{MappingPolicy.__name__}' is deprecated and will be removed in "
diff --git a/rasa/core/policies/memoization.py b/rasa/core/policies/memoization.py
index 55876c4a07af..849678e3f4a1 100644
--- a/rasa/core/policies/memoization.py
+++ b/rasa/core/policies/memoization.py
@@ -71,6 +71,7 @@ def __init__(
         priority: int = MEMOIZATION_POLICY_PRIORITY,
         max_history: Optional[int] = MAX_HISTORY_NOT_SET,
         lookup: Optional[Dict] = None,
+        **kwargs: Any,
     ) -> None:
         """Initialize the policy.
 
@@ -81,7 +82,6 @@ def __init__(
             lookup: a dictionary that stores featurized tracker states and
                 predicted actions for them
         """
-
         if max_history == MAX_HISTORY_NOT_SET:
             max_history = OLD_DEFAULT_MAX_HISTORY  # old default value
             rasa.shared.utils.io.raise_warning(
@@ -97,7 +97,7 @@ def __init__(
         if not featurizer:
             featurizer = self._standard_featurizer(max_history)
 
-        super().__init__(featurizer, priority)
+        super().__init__(featurizer, priority, **kwargs)
 
         self.max_history = self.featurizer.max_history
         self.lookup = lookup if lookup is not None else {}
diff --git a/rasa/core/policies/policy.py b/rasa/core/policies/policy.py
index 8b54d84b9184..bfe3a37af921 100644
--- a/rasa/core/policies/policy.py
+++ b/rasa/core/policies/policy.py
@@ -16,6 +16,8 @@
     TYPE_CHECKING,
 )
 import numpy as np
+
+from rasa.core.exceptions import UnsupportedDialogueModelError
 from rasa.shared.core.events import Event
 
 import rasa.shared.utils.common
@@ -34,6 +36,7 @@
 from rasa.core.constants import DEFAULT_POLICY_PRIORITY
 from rasa.shared.core.constants import USER, SLOTS, PREVIOUS_ACTION, ACTIVE_LOOP
 from rasa.shared.nlu.constants import ENTITIES, INTENT, TEXT, ACTION_TEXT, ACTION_NAME
+from rasa.utils.tensorflow.constants import EPOCHS
 
 if TYPE_CHECKING:
     from rasa.shared.nlu.training_data.features import Features
@@ -110,12 +113,17 @@ def __init__(
         self,
         featurizer: Optional[TrackerFeaturizer] = None,
         priority: int = DEFAULT_POLICY_PRIORITY,
+        should_finetune: bool = False,
+        **kwargs: Any,
     ) -> None:
+        """Constructs a new Policy object."""
         self.__featurizer = self._create_featurizer(featurizer)
         self.priority = priority
+        self.finetune_mode = should_finetune
 
     @property
     def featurizer(self):
+        """Returns the policy's featurizer."""
         return self.__featurizer
 
     @staticmethod
@@ -272,7 +280,7 @@ def persist(self, path: Union[Text, Path]) -> None:
         rasa.shared.utils.io.dump_obj_as_json_to_file(file, self._metadata())
 
     @classmethod
-    def load(cls, path: Union[Text, Path]) -> "Policy":
+    def load(cls, path: Union[Text, Path], **kwargs: Any) -> "Policy":
         """Loads a policy from path.
 
         Args:
@@ -290,6 +298,25 @@ def load(cls, path: Union[Text, Path]) -> "Policy":
                 featurizer = TrackerFeaturizer.load(path)
                 data["featurizer"] = featurizer
 
+            data.update(kwargs)
+
+            constructor_args = rasa.shared.utils.common.arguments_of(cls)
+            if "kwargs" not in constructor_args:
+                if set(data.keys()).issubset(set(constructor_args)):
+                    rasa.shared.utils.io.raise_deprecation_warning(
+                        f"`{cls.__name__}.__init__` does not accept `**kwargs` "
+                        f"This is required for contextual information e.g. the flag "
+                        f"`should_finetune`.",
+                        warn_until_version="3.0.0",
+                    )
+                else:
+                    raise UnsupportedDialogueModelError(
+                        f"`{cls.__name__}.__init__` does not accept `**kwargs`. "
+                        f"Attempting to pass {data} to the policy. "
+                        f"This argument should be added to all policies by "
+                        f"Rasa Open Source 3.0.0."
+                    )
+
             return cls(**data)
 
         logger.info(
diff --git a/rasa/core/policies/rule_policy.py b/rasa/core/policies/rule_policy.py
index c13086ec44af..033ace294237 100644
--- a/rasa/core/policies/rule_policy.py
+++ b/rasa/core/policies/rule_policy.py
@@ -108,6 +108,7 @@ def __init__(
         enable_fallback_prediction: bool = True,
         restrict_rules: bool = True,
         check_for_contradictions: bool = True,
+        **kwargs: Any,
     ) -> None:
         """Create a `RulePolicy` object.
 
@@ -124,6 +125,10 @@ def __init__(
                 if no rule matched.
             enable_fallback_prediction: If `True` `core_fallback_action_name` is
                 predicted in case no rule matched.
+            restrict_rules: If `True` rules are restricted to contain a maximum of 1
+                user message. This is used to avoid that users build a state machine
+                using the rules.
+            check_for_contradictions: Check for contradictions.
         """
         self._core_fallback_threshold = core_fallback_threshold
         self._fallback_action_name = core_fallback_action_name
@@ -136,7 +141,11 @@ def __init__(
 
         # max history is set to `None` in order to capture any lengths of rule stories
         super().__init__(
-            featurizer=featurizer, priority=priority, max_history=None, lookup=lookup
+            featurizer=featurizer,
+            priority=priority,
+            max_history=None,
+            lookup=lookup,
+            **kwargs,
         )
 
     @classmethod
diff --git a/rasa/core/policies/sklearn_policy.py b/rasa/core/policies/sklearn_policy.py
index ef616fcdc208..3c1340d3e4e2 100644
--- a/rasa/core/policies/sklearn_policy.py
+++ b/rasa/core/policies/sklearn_policy.py
@@ -27,7 +27,7 @@
 from sklearn.preprocessing import LabelEncoder
 from rasa.shared.nlu.constants import ACTION_TEXT, TEXT
 from rasa.shared.nlu.training_data.features import Features
-from rasa.utils.tensorflow.constants import SENTENCE
+from rasa.utils.tensorflow.constants import EPOCHS, SENTENCE
 from rasa.utils.tensorflow.model_data import Data
 
 # noinspection PyProtectedMember
@@ -72,6 +72,8 @@ def __init__(
         Args:
             featurizer: Featurizer used to convert the training data into
                 vector format.
+            priority: Policy priority
+            max_history: Maximum history of the dialogs.
             model: The sklearn model or model pipeline.
             param_grid: If *param_grid* is not None and *cv* is given,
                 a grid search on the given *param_grid* is performed
@@ -85,7 +87,6 @@ def __init__(
             shuffle: Whether to shuffle training data.
             zero_state_features: Contains default feature values for attributes
         """
-
         if featurizer:
             if not isinstance(featurizer, MaxHistoryTrackerFeaturizer):
                 raise TypeError(
@@ -104,7 +105,7 @@ def __init__(
                 )
             featurizer = self._standard_featurizer(max_history)
 
-        super().__init__(featurizer, priority)
+        super().__init__(featurizer, priority, **kwargs)
 
         self.model = model or self._default_model()
         self.cv = cv
@@ -302,7 +303,10 @@ def persist(self, path: Union[Text, Path]) -> None:
             )
 
     @classmethod
-    def load(cls, path: Union[Text, Path]) -> Policy:
+    def load(
+        cls, path: Union[Text, Path], should_finetune: bool = False, **kwargs: Any
+    ) -> Policy:
+        """See the docstring for `Policy.load`."""
         filename = Path(path) / "sklearn_model.pkl"
         zero_features_filename = Path(path) / "zero_state_features.pkl"
         if not Path(path).exists():
@@ -325,6 +329,7 @@ def load(cls, path: Union[Text, Path]) -> Policy:
             featurizer=featurizer,
             priority=meta["priority"],
             zero_state_features=zero_state_features,
+            should_finetune=should_finetune,
         )
 
         state = io_utils.pickle_load(filename)
diff --git a/rasa/core/policies/ted_policy.py b/rasa/core/policies/ted_policy.py
index 7a7b60352327..eaa725ef32ff 100644
--- a/rasa/core/policies/ted_policy.py
+++ b/rasa/core/policies/ted_policy.py
@@ -217,14 +217,16 @@ def __init__(
         max_history: Optional[int] = None,
         model: Optional[RasaModel] = None,
         zero_state_features: Optional[Dict[Text, List["Features"]]] = None,
+        should_finetune: bool = False,
         **kwargs: Any,
     ) -> None:
         """Declare instance variables with default values."""
-
         if not featurizer:
             featurizer = self._standard_featurizer(max_history)
 
-        super().__init__(featurizer, priority)
+        super().__init__(
+            featurizer, priority, should_finetune=should_finetune, **kwargs
+        )
         if isinstance(featurizer, FullDialogueTrackerFeaturizer):
             self.is_full_dialogue_featurizer_used = True
         else:
@@ -350,12 +352,16 @@ def train(
         # keep one example for persisting and loading
         self.data_example = model_data.first_data_example()
 
-        self.model = TED(
-            model_data.get_signature(),
-            self.config,
-            isinstance(self.featurizer, MaxHistoryTrackerFeaturizer),
-            self._label_data,
-        )
+        if not self.finetune_mode:
+            # This means the model wasn't loaded from a
+            # previously trained model and hence needs
+            # to be instantiated.
+            self.model = TED(
+                model_data.get_signature(),
+                self.config,
+                isinstance(self.featurizer, MaxHistoryTrackerFeaturizer),
+                self._label_data,
+            )
 
         self.model.fit(
             model_data,
@@ -437,8 +443,15 @@ def persist(self, path: Union[Text, Path]) -> None:
         )
 
     @classmethod
-    def load(cls, path: Union[Text, Path]) -> "TEDPolicy":
+    def load(
+        cls,
+        path: Union[Text, Path],
+        should_finetune: bool = False,
+        epoch_override: int = defaults[EPOCHS],
+        **kwargs: Any,
+    ) -> "TEDPolicy":
         """Loads a policy from the storage.
+
         **Needs to load its featurizer**
         """
         model_path = Path(path)
@@ -476,6 +489,8 @@ def load(cls, path: Union[Text, Path]) -> "TEDPolicy":
         )
         meta = train_utils.update_similarity_type(meta)
 
+        meta[EPOCHS] = epoch_override
+
         model = TED.load(
             str(tf_model_file),
             model_data_example,
@@ -485,26 +500,30 @@ def load(cls, path: Union[Text, Path]) -> "TEDPolicy":
                 featurizer, MaxHistoryTrackerFeaturizer
             ),
             label_data=label_data,
+            finetune_mode=should_finetune,
         )
 
-        # build the graph for prediction
-        predict_data_example = RasaModelData(
-            label_key=LABEL_KEY,
-            label_sub_key=LABEL_SUB_KEY,
-            data={
-                feature_name: features
-                for feature_name, features in model_data_example.items()
-                if feature_name
-                in STATE_LEVEL_FEATURES + FEATURES_TO_ENCODE + [DIALOGUE]
-            },
-        )
-        model.build_for_predict(predict_data_example)
+        if not should_finetune:
+            # build the graph for prediction
+
+            features_to_select = STATE_LEVEL_FEATURES + FEATURES_TO_ENCODE + [DIALOGUE]
+            predict_data_example = RasaModelData(
+                label_key=LABEL_KEY,
+                label_sub_key=LABEL_SUB_KEY,
+                data={
+                    feature_name: features
+                    for feature_name, features in model_data_example.items()
+                    if feature_name in features_to_select
+                },
+            )
+            model.build_for_predict(predict_data_example)
 
         return cls(
             featurizer=featurizer,
             priority=priority,
             model=model,
             zero_state_features=zero_state_features,
+            should_finetune=should_finetune,
             **meta,
         )
 
diff --git a/rasa/core/policies/two_stage_fallback.py b/rasa/core/policies/two_stage_fallback.py
index 92b2a5b43445..ed3db549dc6f 100644
--- a/rasa/core/policies/two_stage_fallback.py
+++ b/rasa/core/policies/two_stage_fallback.py
@@ -62,10 +62,12 @@ def __init__(
         fallback_core_action_name: Text = ACTION_DEFAULT_FALLBACK_NAME,
         fallback_nlu_action_name: Text = ACTION_DEFAULT_FALLBACK_NAME,
         deny_suggestion_intent_name: Text = USER_INTENT_OUT_OF_SCOPE,
+        **kwargs: Any,
     ) -> None:
         """Create a new Two-stage Fallback policy.
 
         Args:
+            priority: The fallback policy priority.
             nlu_threshold: minimum threshold for NLU confidence.
                 If intent prediction confidence is lower than this,
                 predict fallback action with confidence 1.0.
@@ -88,6 +90,7 @@ def __init__(
             ambiguity_threshold,
             core_threshold,
             fallback_core_action_name,
+            **kwargs,
         )
 
         self.fallback_nlu_action_name = fallback_nlu_action_name
diff --git a/rasa/core/train.py b/rasa/core/train.py
index 4074884ac7d8..44137f6bbd00 100644
--- a/rasa/core/train.py
+++ b/rasa/core/train.py
@@ -16,7 +16,7 @@
 if typing.TYPE_CHECKING:
     from rasa.shared.nlu.interpreter import NaturalLanguageInterpreter
     from rasa.core.utils import AvailableEndpoints
-
+    from rasa.core.agent import Agent
 
 logger = logging.getLogger(__name__)
 
@@ -30,10 +30,11 @@ async def train(
     policy_config: Optional[Union[Text, Dict]] = None,
     exclusion_percentage: Optional[int] = None,
     additional_arguments: Optional[Dict] = None,
-):
-    from rasa.core.agent import Agent
+    model_to_finetune: Optional["Agent"] = None,
+) -> "Agent":
     from rasa.core import config, utils
     from rasa.core.utils import AvailableEndpoints
+    from rasa.core.agent import Agent
 
     if not endpoints:
         endpoints = AvailableEndpoints()
@@ -64,6 +65,8 @@ async def train(
     training_data = await agent.load_data(
         training_resource, exclusion_percentage=exclusion_percentage, **data_load_args
     )
+    if model_to_finetune:
+        agent.policy_ensemble = model_to_finetune.policy_ensemble
     agent.train(training_data, **additional_arguments)
     agent.persist(output_path)
 
diff --git a/rasa/model.py b/rasa/model.py
index 43f74ccfcfee..7da06f2bc3cb 100644
--- a/rasa/model.py
+++ b/rasa/model.py
@@ -10,6 +10,9 @@
 from pathlib import Path
 from typing import Any, Text, Tuple, Union, Optional, List, Dict, NamedTuple
 
+from packaging import version
+
+from rasa.constants import MINIMUM_COMPATIBLE_VERSION
 import rasa.shared.utils.io
 import rasa.utils.io
 from rasa.cli.utils import create_output_path
@@ -30,7 +33,6 @@
 if typing.TYPE_CHECKING:
     from rasa.shared.importers.importer import TrainingDataImporter
 
-
 logger = logging.getLogger(__name__)
 
 
@@ -42,18 +44,19 @@
 FINGERPRINT_CONFIG_KEY = "config"
 FINGERPRINT_CONFIG_CORE_KEY = "core-config"
 FINGERPRINT_CONFIG_NLU_KEY = "nlu-config"
+FINGERPRINT_CONFIG_WITHOUT_EPOCHS_KEY = "config-without-epochs"
 FINGERPRINT_DOMAIN_WITHOUT_NLG_KEY = "domain"
 FINGERPRINT_NLG_KEY = "nlg"
 FINGERPRINT_RASA_VERSION_KEY = "version"
 FINGERPRINT_STORIES_KEY = "stories"
 FINGERPRINT_NLU_DATA_KEY = "messages"
+FINGERPRINT_NLU_LABELS_KEY = "nlu_labels"
 FINGERPRINT_PROJECT = "project"
 FINGERPRINT_TRAINED_AT_KEY = "trained_at"
 
 
 class Section(NamedTuple):
-    """Defines relevant fingerprint sections which are used to decide whether a model
-    should be retrained."""
+    """Specifies which fingerprint keys decide whether this sub-model is retrained."""
 
     name: Text
     relevant_keys: List[Text]
@@ -82,6 +85,8 @@ class Section(NamedTuple):
 
 
 class FingerprintComparisonResult:
+    """Container for the results of a fingerprint comparison."""
+
     def __init__(
         self,
         nlu: bool = True,
@@ -327,10 +332,14 @@ async def model_fingerprint(file_importer: "TrainingDataImporter") -> Fingerprin
         FINGERPRINT_CONFIG_NLU_KEY: _get_fingerprint_of_config(
             config, include_keys=CONFIG_KEYS_NLU
         ),
+        FINGERPRINT_CONFIG_WITHOUT_EPOCHS_KEY: _get_fingerprint_of_config_without_epochs(
+            config
+        ),
         FINGERPRINT_DOMAIN_WITHOUT_NLG_KEY: domain.fingerprint(),
         FINGERPRINT_NLG_KEY: rasa.shared.utils.io.deep_container_fingerprint(responses),
         FINGERPRINT_PROJECT: project_fingerprint(),
         FINGERPRINT_NLU_DATA_KEY: nlu_data.fingerprint(),
+        FINGERPRINT_NLU_LABELS_KEY: nlu_data.label_fingerprint(),
         FINGERPRINT_STORIES_KEY: stories.fingerprint(),
         FINGERPRINT_TRAINED_AT_KEY: time.time(),
         FINGERPRINT_RASA_VERSION_KEY: rasa.__version__,
@@ -352,6 +361,23 @@ def _get_fingerprint_of_config(
     return rasa.shared.utils.io.deep_container_fingerprint(sub_config)
 
 
+def _get_fingerprint_of_config_without_epochs(
+    config: Optional[Dict[Text, Any]],
+) -> Text:
+    if not config:
+        return ""
+
+    copied_config = copy.deepcopy(config)
+
+    for key in ["pipeline", "policies"]:
+        if copied_config.get(key):
+            for p in copied_config[key]:
+                if "epochs" in p:
+                    del p["epochs"]
+
+    return rasa.shared.utils.io.deep_container_fingerprint(copied_config)
+
+
 def fingerprint_from_path(model_path: Text) -> Fingerprint:
     """Load a persisted fingerprint.
 
@@ -473,6 +499,43 @@ def should_retrain(
         return fingerprint_comparison
 
 
+def can_finetune(
+    last_fingerprint: Fingerprint,
+    new_fingerprint: Fingerprint,
+    core: bool = False,
+    nlu: bool = False,
+) -> bool:
+    """Checks if components of a model can be finetuned with incremental training.
+
+    Args:
+        last_fingerprint: The fingerprint of the old model to potentially be fine-tuned.
+        new_fingerprint: The fingerprint of the new model.
+        core: Check sections for finetuning a core model.
+        nlu: Check sections for finetuning an nlu model.
+
+    Returns:
+        `True` if the old model can be finetuned, `False` otherwise.
+    """
+    section_keys = [
+        FINGERPRINT_CONFIG_WITHOUT_EPOCHS_KEY,
+    ]
+    if core:
+        section_keys.append(FINGERPRINT_DOMAIN_WITHOUT_NLG_KEY)
+    if nlu:
+        section_keys.append(FINGERPRINT_NLU_LABELS_KEY)
+
+    fingerprint_changed = did_section_fingerprint_change(
+        last_fingerprint,
+        new_fingerprint,
+        Section(name="finetune", relevant_keys=section_keys),
+    )
+
+    old_model_above_min_version = version.parse(
+        last_fingerprint.get(FINGERPRINT_RASA_VERSION_KEY)
+    ) >= version.parse(MINIMUM_COMPATIBLE_VERSION)
+    return old_model_above_min_version and not fingerprint_changed
+
+
 def package_model(
     fingerprint: Fingerprint,
     output_directory: Text,
@@ -515,8 +578,36 @@ async def update_model_with_new_domain(
         importer: Importer which provides the new domain.
         unpacked_model_path: Path to the unpacked model.
     """
-
     model_path = Path(unpacked_model_path) / DEFAULT_CORE_SUBDIRECTORY_NAME
     domain = await importer.get_domain()
 
     domain.persist(model_path / DEFAULT_DOMAIN_PATH)
+
+
+def get_model_for_finetuning(
+    previous_model_file: Optional[Union[Path, Text]]
+) -> Optional[Text]:
+    """Gets validated path for model to finetune.
+
+    Args:
+        previous_model_file: Path to model file which should be used for finetuning or
+            a directory in case the latest trained model should be used.
+
+    Returns:
+        Path to model archive. `None` if there is no model.
+    """
+    if Path(previous_model_file).is_dir():
+        logger.debug(
+            f"Trying to load latest model from '{previous_model_file}' for "
+            f"finetuning."
+        )
+        return get_latest_model(previous_model_file)
+
+    if Path(previous_model_file).is_file():
+        return previous_model_file
+
+    logger.debug(
+        "No valid model for finetuning found as directory either "
+        "contains no model or model file cannot be found."
+    )
+    return None
diff --git a/rasa/nlu/classifiers/diet_classifier.py b/rasa/nlu/classifiers/diet_classifier.py
index d16361f332ce..9ffa2b477d0d 100644
--- a/rasa/nlu/classifiers/diet_classifier.py
+++ b/rasa/nlu/classifiers/diet_classifier.py
@@ -308,9 +308,9 @@ def __init__(
         index_label_id_mapping: Optional[Dict[int, Text]] = None,
         entity_tag_specs: Optional[List[EntityTagSpec]] = None,
         model: Optional[RasaModel] = None,
+        finetune_mode: bool = False,
     ) -> None:
         """Declare instance variables with default values."""
-
         if component_config is not None and EPOCHS not in component_config:
             rasa.shared.utils.io.raise_warning(
                 f"Please configure the number of '{EPOCHS}' in your configuration file."
@@ -333,6 +333,17 @@ def __init__(
 
         self.split_entities_config = self.init_split_entities()
 
+        self.finetune_mode = finetune_mode
+
+        if not self.model and self.finetune_mode:
+            raise rasa.shared.exceptions.InvalidParameterException(
+                f"{self.__class__.__name__} was instantiated "
+                f"with `model=None` and `finetune_mode=True`. "
+                f"This is not a valid combination as the component "
+                f"needs an already instantiated and trained model "
+                f"to continue training in finetune mode."
+            )
+
     @property
     def label_key(self) -> Optional[Text]:
         """Return key if intent classification is activated."""
@@ -766,7 +777,9 @@ def train(
         # keep one example for persisting and loading
         self._data_example = model_data.first_data_example()
 
-        self.model = self._instantiate_model_class(model_data)
+        if not self.finetune_mode:
+            # No pre-trained model to load from. Create a new instance of the model.
+            self.model = self._instantiate_model_class(model_data)
 
         self.model.fit(
             model_data,
@@ -956,10 +969,10 @@ def load(
         model_dir: Text = None,
         model_metadata: Metadata = None,
         cached_component: Optional["DIETClassifier"] = None,
+        should_finetune: bool = False,
         **kwargs: Any,
     ) -> "DIETClassifier":
         """Loads the trained model from the provided directory."""
-
         if not model_dir or not meta.get("file"):
             logger.debug(
                 f"Failed to load model for '{cls.__name__}'. "
@@ -979,7 +992,12 @@ def load(
         meta = train_utils.update_similarity_type(meta)
 
         model = cls._load_model(
-            entity_tag_specs, label_data, meta, data_example, model_dir
+            entity_tag_specs,
+            label_data,
+            meta,
+            data_example,
+            model_dir,
+            finetune_mode=should_finetune,
         )
 
         return cls(
@@ -987,6 +1005,7 @@ def load(
             index_label_id_mapping=index_label_id_mapping,
             entity_tag_specs=entity_tag_specs,
             model=model,
+            finetune_mode=should_finetune,
         )
 
     @classmethod
@@ -1039,6 +1058,7 @@ def _load_model(
         meta: Dict[Text, Any],
         data_example: Dict[Text, Dict[Text, List[np.ndarray]]],
         model_dir: Text,
+        finetune_mode: bool = False,
     ) -> "RasaModel":
         file_name = meta.get("file")
         tf_model_file = os.path.join(model_dir, file_name + ".tf_model")
@@ -1051,20 +1071,27 @@ def _load_model(
         )
 
         model = cls._load_model_class(
-            tf_model_file, model_data_example, label_data, entity_tag_specs, meta
+            tf_model_file,
+            model_data_example,
+            label_data,
+            entity_tag_specs,
+            meta,
+            finetune_mode=finetune_mode,
         )
 
-        # build the graph for prediction
-        predict_data_example = RasaModelData(
-            label_key=label_key,
-            data={
-                feature_name: features
-                for feature_name, features in model_data_example.items()
-                if TEXT in feature_name
-            },
-        )
+        if not finetune_mode:
+
+            # build the graph for prediction
+            predict_data_example = RasaModelData(
+                label_key=label_key,
+                data={
+                    feature_name: features
+                    for feature_name, features in model_data_example.items()
+                    if TEXT in feature_name
+                },
+            )
 
-        model.build_for_predict(predict_data_example)
+            model.build_for_predict(predict_data_example)
 
         return model
 
@@ -1076,6 +1103,7 @@ def _load_model_class(
         label_data: RasaModelData,
         entity_tag_specs: List[EntityTagSpec],
         meta: Dict[Text, Any],
+        finetune_mode: bool,
     ) -> "RasaModel":
 
         return cls.model_class().load(
@@ -1085,6 +1113,7 @@ def _load_model_class(
             label_data=label_data,
             entity_tag_specs=entity_tag_specs,
             config=copy.deepcopy(meta),
+            finetune_mode=finetune_mode,
         )
 
     def _instantiate_model_class(self, model_data: RasaModelData) -> "RasaModel":
diff --git a/rasa/nlu/constants.py b/rasa/nlu/constants.py
index 14297822acb3..90dc50e07520 100644
--- a/rasa/nlu/constants.py
+++ b/rasa/nlu/constants.py
@@ -78,3 +78,5 @@
 FEATURIZER_CLASS_ALIAS = "alias"
 
 NO_LENGTH_RESTRICTION = -1
+MIN_ADDITIONAL_REGEX_PATTERNS = 10
+MIN_ADDITIONAL_CVF_VOCABULARY = 1000
diff --git a/rasa/nlu/featurizers/sparse_featurizer/count_vectors_featurizer.py b/rasa/nlu/featurizers/sparse_featurizer/count_vectors_featurizer.py
index 29978e3018a9..17ca655e0f33 100644
--- a/rasa/nlu/featurizers/sparse_featurizer/count_vectors_featurizer.py
+++ b/rasa/nlu/featurizers/sparse_featurizer/count_vectors_featurizer.py
@@ -2,7 +2,7 @@
 import os
 import re
 import scipy.sparse
-from typing import Any, Dict, List, Optional, Text, Type, Tuple
+from typing import Any, Dict, List, Optional, Text, Type, Tuple, Set
 
 import rasa.shared.utils.io
 from rasa.shared.constants import DOCS_URL_COMPONENTS
@@ -21,16 +21,21 @@
     MESSAGE_ATTRIBUTES,
     DENSE_FEATURIZABLE_ATTRIBUTES,
     FEATURIZER_CLASS_ALIAS,
+    MIN_ADDITIONAL_CVF_VOCABULARY,
 )
 from rasa.shared.nlu.constants import (
     TEXT,
     INTENT,
     INTENT_RESPONSE_KEY,
+    RESPONSE,
+    ACTION_TEXT,
     FEATURE_TYPE_SENTENCE,
     FEATURE_TYPE_SEQUENCE,
     ACTION_NAME,
 )
 
+BUFFER_SLOTS_PREFIX = "buf_"
+
 logger = logging.getLogger(__name__)
 
 
@@ -84,6 +89,8 @@ def required_components(cls) -> List[Type[Component]]:
         # indicates whether the featurizer should use the lemma of a word for
         # counting (if available) or not
         "use_lemma": True,
+        # Additional vocabulary size to be kept reserved for finetuning
+        "additional_vocabulary_size": {TEXT: None, RESPONSE: None, ACTION_TEXT: None},
     }
 
     @classmethod
@@ -124,8 +131,7 @@ def _load_count_vect_params(self) -> None:
         # use the lemma of the words or not
         self.use_lemma = self.component_config["use_lemma"]
 
-    # noinspection PyPep8Naming
-    def _load_OOV_params(self) -> None:
+    def _load_vocabulary_params(self) -> None:
         self.OOV_token = self.component_config["OOV_token"]
 
         self.OOV_words = self.component_config["OOV_words"]
@@ -143,8 +149,20 @@ def _load_OOV_params(self) -> None:
             if self.OOV_words:
                 self.OOV_words = [w.lower() for w in self.OOV_words]
 
+        # Additional vocabulary size to be kept reserved
+        self.additional_vocabulary_size = self.component_config[
+            "additional_vocabulary_size"
+        ]
+        if isinstance(self.additional_vocabulary_size, int):
+            # User defined a common value for all attributes
+            user_defined_additional_size = self.additional_vocabulary_size
+            self.additional_vocabulary_size = {
+                attribute: user_defined_additional_size
+                for attribute in DENSE_FEATURIZABLE_ATTRIBUTES
+            }
+
     def _check_attribute_vocabulary(self, attribute: Text) -> bool:
-        """Check if trained vocabulary exists in attribute's count vectorizer"""
+        """Checks if trained vocabulary exists in attribute's count vectorizer."""
         try:
             return hasattr(self.vectorizers[attribute], "vocabulary_")
         except (AttributeError, TypeError):
@@ -200,16 +218,16 @@ def __init__(
         self,
         component_config: Optional[Dict[Text, Any]] = None,
         vectorizers: Optional[Dict[Text, "CountVectorizer"]] = None,
+        finetune_mode: bool = False,
     ) -> None:
         """Construct a new count vectorizer using the sklearn framework."""
-
         super().__init__(component_config)
 
         # parameters for sklearn's CountVectorizer
         self._load_count_vect_params()
 
         # handling Out-Of-Vocabulary (OOV) words
-        self._load_OOV_params()
+        self._load_vocabulary_params()
 
         # warn that some of config parameters might be ignored
         self._check_analyzer()
@@ -220,6 +238,8 @@ def __init__(
         # declare class instance for CountVectorizer
         self.vectorizers = vectorizers
 
+        self.finetune_mode = finetune_mode
+
     def _get_message_tokens_by_attribute(
         self, message: "Message", attribute: Text
     ) -> List[Text]:
@@ -341,72 +361,257 @@ def _convert_attribute_tokens_to_texts(
 
         return attribute_texts
 
-    def _train_with_shared_vocab(self, attribute_texts: Dict[Text, List[Text]]):
-        """Construct the vectorizers and train them with a shared vocab"""
-
-        self.vectorizers = self._create_shared_vocab_vectorizers(
-            {
-                "strip_accents": self.strip_accents,
-                "lowercase": self.lowercase,
-                "stop_words": self.stop_words,
-                "min_ngram": self.min_ngram,
-                "max_ngram": self.max_ngram,
-                "max_df": self.max_df,
-                "min_df": self.min_df,
-                "max_features": self.max_features,
-                "analyzer": self.analyzer,
-            }
-        )
+    @staticmethod
+    def _get_starting_empty_index(vocabulary: Dict[Text, int]) -> int:
+        for key in vocabulary.keys():
+            if key.startswith(BUFFER_SLOTS_PREFIX):
+                return int(key.split(BUFFER_SLOTS_PREFIX)[1])
+        return len(vocabulary)
+
+    def _update_vectorizer_vocabulary(
+        self, attribute: Text, new_vocabulary: Set[Text]
+    ) -> None:
+        """Updates the existing vocabulary of the vectorizer with new unseen words.
 
-        combined_cleaned_texts = []
-        for attribute in self._attributes:
-            combined_cleaned_texts += attribute_texts[attribute]
+        These unseen words should only occupy the empty buffer slots.
 
-        try:
-            self.vectorizers[TEXT].fit(combined_cleaned_texts)
-        except ValueError:
-            logger.warning(
-                "Unable to train a shared CountVectorizer. "
-                "Leaving an untrained CountVectorizer"
+        Args:
+            attribute: Message attribute for which vocabulary should be updated.
+            new_vocabulary: Set of words to expand the vocabulary with if they are unseen.
+        """
+        existing_vocabulary: Dict[Text, int] = self.vectorizers[attribute].vocabulary
+        if len(new_vocabulary) > len(existing_vocabulary):
+            rasa.shared.utils.io.raise_warning(
+                f"New data contains vocabulary of size {len(new_vocabulary)} for attribute {attribute} "
+                f"which is larger than the maximum vocabulary size({len(existing_vocabulary)}) "
+                f"of the original model. Some tokens will have to be dropped "
+                f"in order to continue training. It is advised to re-train the "
+                f"model from scratch on the complete data."
             )
+        self._merge_new_vocabulary_tokens(existing_vocabulary, new_vocabulary)
+        self._set_vocabulary(attribute, existing_vocabulary)
+
+    def _merge_new_vocabulary_tokens(
+        self, existing_vocabulary: Dict[Text, int], vocabulary: Set[Text]
+    ) -> None:
+        available_empty_index = self._get_starting_empty_index(existing_vocabulary)
+        for token in vocabulary:
+            if token not in existing_vocabulary:
+                existing_vocabulary[token] = available_empty_index
+                del existing_vocabulary[f"{BUFFER_SLOTS_PREFIX}{available_empty_index}"]
+                available_empty_index += 1
+                if available_empty_index == len(existing_vocabulary):
+                    # We have exhausted all available vocabulary slots.
+                    # Drop the remaining vocabulary.
+                    return
+
+    def _get_additional_vocabulary_size(
+        self, attribute: Text, existing_vocabulary_size: int
+    ) -> int:
+        """Gets additional vocabulary size to be saved for incremental training.
+
+        If `self.additional_vocabulary_size` is not `None`,
+        we return that as the user should have specified
+        this number. If not then we take the default
+        additional vocabulary size which is 1/2 of the
+        current vocabulary size.
+
+        Args:
+            attribute: Message attribute for which additional vocabulary size should be computed.
+            existing_vocabulary_size: Current size of vocabulary learnt from the training data.
+
+        Returns:
+            Size of additional vocabulary that should be set aside for incremental training.
+        """
+        # Vocabulary expansion for INTENTS, ACTION_NAME
+        # and INTENT_RESPONSE_KEY is currently not supported as
+        # incremental training does not support creation/deletion
+        # of new/existing labels(intents, actions, etc.)
+        if attribute not in DENSE_FEATURIZABLE_ATTRIBUTES:
+            return 0
+
+        configured_additional_size = self.additional_vocabulary_size.get(attribute)
+        if configured_additional_size is not None:
+            return configured_additional_size
+
+        # If the user hasn't defined additional vocabulary size,
+        # then we increase it by 1000 minimum. If the current
+        # vocabulary size is greater than 2000, we take half of
+        # that number as additional vocabulary size.
+        return max(MIN_ADDITIONAL_CVF_VOCABULARY, int(existing_vocabulary_size * 0.5))
+
+    def _add_buffer_to_vocabulary(self, attribute: Text) -> None:
+        """Adds extra tokens to vocabulary for incremental training.
+
+        These extra tokens act as buffer slots which are used up sequentially
+        when more data is received as part of incremental training. Each of
+        these tokens start with a prefix `buf_` followed by the extra slot index.
+        So for example - buf_1, buf_2, buf_3... and so on.
+
+        Args:
+            attribute: Name of the attribute for which the vocabulary should be expanded.
+        """
+        original_vocabulary = self.vectorizers[attribute].vocabulary_
+        current_vocabulary_size = len(original_vocabulary)
+        for index in range(
+            current_vocabulary_size,
+            current_vocabulary_size
+            + self._get_additional_vocabulary_size(attribute, current_vocabulary_size),
+        ):
+            original_vocabulary[f"{BUFFER_SLOTS_PREFIX}{index}"] = index
+        self._set_vocabulary(attribute, original_vocabulary)
+
+    def _set_vocabulary(
+        self, attribute: Text, original_vocabulary: Dict[Text, int]
+    ) -> None:
+        """Sets the vocabulary of the vectorizer of attribute.
+
+        Args:
+            attribute: Message attribute for which vocabulary should be set
+            original_vocabulary: Vocabulary for the attribute to be set.
+        """
+        self.vectorizers[attribute].vocabulary_ = original_vocabulary
+        self.vectorizers[attribute]._validate_vocabulary()
+
+    @staticmethod
+    def _construct_vocabulary_from_texts(
+        vectorizer: CountVectorizer, texts: List[Text]
+    ) -> Set:
+        """Applies vectorizer's preprocessor on texts to get the vocabulary from texts.
+
+        Args:
+            vectorizer: Sklearn's count vectorizer which has been pre-configured.
+            texts: Examples from which the vocabulary should be constructed
+
+        Returns:
+            Unique vocabulary words extracted.
+        """
+        analyzer = vectorizer.build_analyzer()
+        vocabulary_words = set()
+        for example in texts:
+            example_vocabulary: List[Text] = analyzer(example)
+            vocabulary_words.update(example_vocabulary)
+        return vocabulary_words
 
     @staticmethod
     def _attribute_texts_is_non_empty(attribute_texts: List[Text]) -> bool:
         return any(attribute_texts)
 
-    def _train_with_independent_vocab(self, attribute_texts: Dict[Text, List[Text]]):
-        """Construct the vectorizers and train them with an independent vocab"""
-
-        self.vectorizers = self._create_independent_vocab_vectorizers(
-            {
-                "strip_accents": self.strip_accents,
-                "lowercase": self.lowercase,
-                "stop_words": self.stop_words,
-                "min_ngram": self.min_ngram,
-                "max_ngram": self.max_ngram,
-                "max_df": self.max_df,
-                "min_df": self.min_df,
-                "max_features": self.max_features,
-                "analyzer": self.analyzer,
-            }
-        )
+    def _train_with_shared_vocab(self, attribute_texts: Dict[Text, List[Text]]) -> None:
+        """Constructs the vectorizers and train them with a shared vocab."""
+        combined_cleaned_texts = []
+        for attribute in self._attributes:
+            combined_cleaned_texts += attribute_texts[attribute]
 
+        if not self.finetune_mode:
+            self.vectorizers = self._create_shared_vocab_vectorizers(
+                {
+                    "strip_accents": self.strip_accents,
+                    "lowercase": self.lowercase,
+                    "stop_words": self.stop_words,
+                    "min_ngram": self.min_ngram,
+                    "max_ngram": self.max_ngram,
+                    "max_df": self.max_df,
+                    "min_df": self.min_df,
+                    "max_features": self.max_features,
+                    "analyzer": self.analyzer,
+                }
+            )
+            self._fit_vectorizer_from_scratch(TEXT, combined_cleaned_texts)
+        else:
+            self._fit_loaded_vectorizer(TEXT, combined_cleaned_texts)
+        self._log_vocabulary_stats(TEXT)
+
+    def _train_with_independent_vocab(
+        self, attribute_texts: Dict[Text, List[Text]]
+    ) -> None:
+        """Constructs the vectorizers and train them with an independent vocab."""
+        if not self.finetune_mode:
+            self.vectorizers = self._create_independent_vocab_vectorizers(
+                {
+                    "strip_accents": self.strip_accents,
+                    "lowercase": self.lowercase,
+                    "stop_words": self.stop_words,
+                    "min_ngram": self.min_ngram,
+                    "max_ngram": self.max_ngram,
+                    "max_df": self.max_df,
+                    "min_df": self.min_df,
+                    "max_features": self.max_features,
+                    "analyzer": self.analyzer,
+                }
+            )
         for attribute in self._attributes:
             if self._attribute_texts_is_non_empty(attribute_texts[attribute]):
-                try:
-                    self.vectorizers[attribute].fit(attribute_texts[attribute])
-                except ValueError:
-                    logger.warning(
-                        f"Unable to train CountVectorizer for message "
-                        f"attribute {attribute}. Leaving an untrained "
-                        f"CountVectorizer for it."
+                if not self.finetune_mode:
+                    self._fit_vectorizer_from_scratch(
+                        attribute, attribute_texts[attribute]
                     )
+                else:
+                    self._fit_loaded_vectorizer(attribute, attribute_texts[attribute])
+
+                self._log_vocabulary_stats(attribute)
             else:
                 logger.debug(
                     f"No text provided for {attribute} attribute in any messages of "
                     f"training data. Skipping training a CountVectorizer for it."
                 )
 
+    def _log_vocabulary_stats(self, attribute: Text) -> None:
+        """Logs number of vocabulary slots filled out of the total number of available slots.
+
+        Args:
+            attribute: Message attribute for which vocabulary stats are logged.
+        """
+        if attribute in DENSE_FEATURIZABLE_ATTRIBUTES:
+            attribute_vocabulary = self.vectorizers[attribute].vocabulary_
+            first_empty_index = self._get_starting_empty_index(attribute_vocabulary)
+            logger.info(
+                f"{first_empty_index} vocabulary slots "
+                f"consumed out of {len(attribute_vocabulary)} "
+                f"slots configured for {attribute} attribute."
+            )
+
+    def _fit_loaded_vectorizer(
+        self, attribute: Text, attribute_texts: List[Text]
+    ) -> None:
+        """Fits training texts to a previously trained count vectorizer.
+
+        We do not use the `.fit()` method because the new unseen
+        words should occupy the buffer slots of the vocabulary.
+
+        Args:
+            attribute: Message attribute for which the vectorizer is to be trained.
+            attribute_texts: Training texts for the attribute
+        """
+        # Get vocabulary words by the preprocessor
+        new_vocabulary = self._construct_vocabulary_from_texts(
+            self.vectorizers[attribute], attribute_texts
+        )
+        # update the vocabulary of vectorizer with new vocabulary
+        self._update_vectorizer_vocabulary(attribute, new_vocabulary)
+
+    def _fit_vectorizer_from_scratch(
+        self, attribute: Text, attribute_texts: List[Text]
+    ) -> None:
+        """Fits training texts to an untrained count vectorizer.
+
+        Args:
+            attribute: Message attribute for which the vectorizer is to be trained.
+            attribute_texts: Training texts for the attribute
+        """
+        try:
+            self.vectorizers[attribute].fit(attribute_texts)
+        except ValueError:
+            logger.warning(
+                f"Unable to train CountVectorizer for message "
+                f"attribute {attribute} since the call to sklearn's "
+                f"`.fit()` method failed. Leaving an untrained "
+                f"CountVectorizer for it."
+            )
+        # Add buffer for extra vocabulary tokens
+        # that come in during incremental training.
+        self._add_buffer_to_vocabulary(attribute)
+
     def _create_features(
         self, attribute: Text, all_tokens: List[List[Text]]
     ) -> Tuple[
@@ -671,9 +876,9 @@ def load(
         model_dir: Optional[Text] = None,
         model_metadata: Optional[Metadata] = None,
         cached_component: Optional["CountVectorsFeaturizer"] = None,
+        should_finetune: bool = False,
         **kwargs: Any,
     ) -> "CountVectorsFeaturizer":
-
         file_name = meta.get("file")
         featurizer_file = os.path.join(model_dir, file_name)
 
@@ -693,7 +898,7 @@ def load(
                 meta, vocabulary=vocabulary
             )
 
-        ftr = cls(meta, vectorizers)
+        ftr = cls(meta, vectorizers, should_finetune)
 
         # make sure the vocabulary has been loaded correctly
         for attribute in vectorizers:
diff --git a/rasa/nlu/featurizers/sparse_featurizer/regex_featurizer.py b/rasa/nlu/featurizers/sparse_featurizer/regex_featurizer.py
index 134e4ca8c94a..2e360fda38c0 100644
--- a/rasa/nlu/featurizers/sparse_featurizer/regex_featurizer.py
+++ b/rasa/nlu/featurizers/sparse_featurizer/regex_featurizer.py
@@ -1,8 +1,7 @@
 import logging
-import os
 import re
 from typing import Any, Dict, List, Optional, Text, Type, Tuple
-
+from pathlib import Path
 import numpy as np
 import scipy.sparse
 
@@ -12,7 +11,11 @@
 from rasa.nlu import utils
 from rasa.nlu.components import Component
 from rasa.nlu.config import RasaNLUModelConfig
-from rasa.nlu.constants import TOKENS_NAMES, FEATURIZER_CLASS_ALIAS
+from rasa.nlu.constants import (
+    TOKENS_NAMES,
+    FEATURIZER_CLASS_ALIAS,
+    MIN_ADDITIONAL_REGEX_PATTERNS,
+)
 from rasa.shared.nlu.constants import (
     TEXT,
     RESPONSE,
@@ -26,6 +29,7 @@
 from rasa.nlu.tokenizers.tokenizer import Tokenizer
 from rasa.shared.nlu.training_data.training_data import TrainingData
 from rasa.shared.nlu.training_data.message import Message
+from rasa.shared.utils.common import lazy_property
 
 logger = logging.getLogger(__name__)
 
@@ -42,6 +46,9 @@ def required_components(cls) -> List[Type[Component]]:
         "use_lookup_tables": True,
         # use regexes to generate features
         "use_regexes": True,
+        # Additional number of patterns to consider
+        # for incremental training
+        "number_additional_patterns": None,
         # use match word boundaries for lookup table
         "use_word_boundaries": True,
     }
@@ -50,12 +57,109 @@ def __init__(
         self,
         component_config: Optional[Dict[Text, Any]] = None,
         known_patterns: Optional[List[Dict[Text, Text]]] = None,
+        pattern_vocabulary_stats: Optional[Dict[Text, int]] = None,
+        finetune_mode: bool = False,
     ) -> None:
-        """Construct new features for regexes and lookup table using regex expressions."""
+        """Constructs new features for regexes and lookup table using regex expressions.
+
+        Args:
+            component_config: Configuration for the component
+            known_patterns: Regex Patterns the component should pre-load itself with.
+            pattern_vocabulary_stats: Statistics about number of pattern slots filled and total number available.
+            finetune_mode: Load component in finetune mode.
+        """
         super().__init__(component_config)
 
         self.known_patterns = known_patterns if known_patterns else []
         self.case_sensitive = self.component_config["case_sensitive"]
+        self.number_additional_patterns = self.component_config[
+            "number_additional_patterns"
+        ]
+        self.finetune_mode = finetune_mode
+        self.pattern_vocabulary_stats = pattern_vocabulary_stats
+
+        if self.finetune_mode and not self.pattern_vocabulary_stats:
+            # If the featurizer is instantiated in finetune mode,
+            # the vocabulary stats for it should be known.
+            raise rasa.shared.exceptions.InvalidParameterException(
+                f"{self.__class__.__name__} was instantiated with"
+                f" `finetune_mode=True` but `pattern_vocabulary_stats`"
+                f" was left to `None`. This is invalid since the featurizer"
+                f" needs vocabulary statistics to featurize in finetune mode."
+            )
+
+    @lazy_property
+    def vocabulary_stats(self) -> Dict[Text, int]:
+        """Computes total vocabulary size and how much of it is consumed.
+
+        Returns:
+            Computed vocabulary size and number of filled vocabulary slots.
+        """
+        if not self.finetune_mode:
+            max_number_patterns = (
+                len(self.known_patterns) + self._get_num_additional_slots()
+            )
+            return {
+                "pattern_slots_filled": len(self.known_patterns),
+                "max_number_patterns": max_number_patterns,
+            }
+        else:
+            self.pattern_vocabulary_stats["pattern_slots_filled"] = len(
+                self.known_patterns
+            )
+            return self.pattern_vocabulary_stats
+
+    def _merge_new_patterns(self, new_patterns: List[Dict[Text, Text]]) -> None:
+        """Updates already known patterns with new patterns extracted from data.
+
+        Args:
+            new_patterns: Patterns extracted from training data and to be merged with known patterns.
+        """
+        max_number_patterns = self.pattern_vocabulary_stats["max_number_patterns"]
+        pattern_name_index_map = {
+            pattern["name"]: index for index, pattern in enumerate(self.known_patterns)
+        }
+        patterns_dropped = False
+
+        for extra_pattern in new_patterns:
+            new_pattern_name = extra_pattern["name"]
+
+            # Some patterns may have just new examples added
+            # to them. These do not count as additional pattern.
+            if new_pattern_name in pattern_name_index_map:
+                self.known_patterns[pattern_name_index_map[new_pattern_name]][
+                    "pattern"
+                ] = extra_pattern["pattern"]
+            else:
+                if len(self.known_patterns) == max_number_patterns:
+                    patterns_dropped = True
+                    continue
+                self.known_patterns.append(extra_pattern)
+        if patterns_dropped:
+            rasa.shared.utils.io.raise_warning(
+                f"The originally trained model was configured to "
+                f"handle a maximum number of {max_number_patterns} patterns. "
+                f"The current training data exceeds this number as "
+                f"there are {len(new_patterns)} patterns in total. "
+                f"Some patterns will be dropped and not used for "
+                f"featurization. It is advisable to re-train the "
+                f"model from scratch."
+            )
+
+    def _get_num_additional_slots(self) -> int:
+        """Computes number of additional pattern slots available in vocabulary on top of known patterns."""
+        if self.number_additional_patterns is None:
+            # We take twice the number of currently defined
+            # regex patterns as the number of additional
+            # vocabulary slots to support if this parameter
+            # is not configured by the user. Also, to avoid having
+            # to retrain from scratch very often, the default number
+            # of additional slots is kept to MIN_ADDITIONAL_SLOTS.
+            # This is an empirically tuned number.
+            self.number_additional_patterns = max(
+                MIN_ADDITIONAL_REGEX_PATTERNS, len(self.known_patterns) * 2
+            )
+        return self.number_additional_patterns
 
     def train(
         self,
@@ -63,13 +167,24 @@ def train(
         config: Optional[RasaNLUModelConfig] = None,
         **kwargs: Any,
     ) -> None:
-
-        self.known_patterns = pattern_utils.extract_patterns(
+        """Trains the component with all patterns extracted from training data.
+
+        Args:
+            training_data: Training data consisting of training examples and patterns available.
+            config: NLU Pipeline config
+            **kwargs: Any other arguments
+        """
+        patterns_from_data = pattern_utils.extract_patterns(
             training_data,
             use_lookup_tables=self.component_config["use_lookup_tables"],
             use_regexes=self.component_config["use_regexes"],
             use_word_boundaries=self.component_config["use_word_boundaries"],
         )
+        if self.finetune_mode:
+            # Merge patterns extracted from data with known patterns
+            self._merge_new_patterns(patterns_from_data)
+        else:
+            self.known_patterns = patterns_from_data
 
         for example in training_data.training_examples:
             for attribute in [TEXT, RESPONSE, ACTION_TEXT]:
@@ -79,6 +194,12 @@ def process(self, message: Message, **kwargs: Any) -> None:
         self._text_features_with_regex(message, TEXT)
 
     def _text_features_with_regex(self, message: Message, attribute: Text) -> None:
+        """Helper method to extract features and set them appropriately in the message object.
+
+        Args:
+            message: Message to be featurized.
+            attribute: Attribute of message to be featurized.
+        """
         if self.known_patterns:
             sequence_features, sentence_features = self._features_for_patterns(
                 message, attribute
@@ -106,11 +227,19 @@ def _features_for_patterns(
         self, message: Message, attribute: Text
     ) -> Tuple[Optional[scipy.sparse.coo_matrix], Optional[scipy.sparse.coo_matrix]]:
         """Checks which known patterns match the message.
+
         Given a sentence, returns a vector of {1,0} values indicating which
         regexes did match. Furthermore, if the
         message is tokenized, the function will mark all tokens with a dict
-        relating the name of the regex to whether it was matched."""
+        relating the name of the regex to whether it was matched.
+
+        Args:
+            message: Message to be featurized.
+            attribute: Attribute of message to be featurized.
 
+        Returns:
+           Token and sentence level features of message attribute.
+        """
         # Attribute not set (e.g. response not present)
         if not message.get(attribute):
             return None, None
@@ -127,8 +256,10 @@ def _features_for_patterns(
 
         sequence_length = len(tokens)
 
-        sequence_features = np.zeros([sequence_length, len(self.known_patterns)])
-        sentence_features = np.zeros([1, len(self.known_patterns)])
+        max_number_patterns = self.vocabulary_stats["max_number_patterns"]
+
+        sequence_features = np.zeros([sequence_length, max_number_patterns])
+        sentence_features = np.zeros([1, max_number_patterns])
 
         for pattern_index, pattern in enumerate(self.known_patterns):
             matches = re.finditer(pattern["pattern"], message.get(TEXT), flags=flags)
@@ -160,23 +291,58 @@ def load(
         model_dir: Optional[Text] = None,
         model_metadata: Optional[Metadata] = None,
         cached_component: Optional["RegexFeaturizer"] = None,
+        should_finetune: bool = False,
         **kwargs: Any,
     ) -> "RegexFeaturizer":
-
+        """Loads a previously trained component.
+
+        Args:
+            meta: Configuration of trained component.
+            model_dir: Path where trained pipeline is stored.
+            model_metadata: Metadata for the trained pipeline.
+            cached_component: Previously cached component(if any).
+            should_finetune: Indicates whether to load the component for further finetuning.
+            **kwargs:
+        """
         file_name = meta.get("file")
-        regex_file = os.path.join(model_dir, file_name)
 
-        if os.path.exists(regex_file):
-            known_patterns = rasa.shared.utils.io.read_json_file(regex_file)
-            return RegexFeaturizer(meta, known_patterns=known_patterns)
-        else:
-            return RegexFeaturizer(meta)
+        patterns_file_name = Path(model_dir) / (file_name + ".patterns.pkl")
+
+        vocabulary_stats_file_name = Path(model_dir) / (
+            file_name + ".vocabulary_stats.pkl"
+        )
+
+        known_patterns = None
+        vocabulary_stats = None
+        if patterns_file_name.exists():
+            known_patterns = rasa.shared.utils.io.read_json_file(patterns_file_name)
+        if vocabulary_stats_file_name.exists():
+            vocabulary_stats = rasa.shared.utils.io.read_json_file(
+                vocabulary_stats_file_name
+            )
+
+        return RegexFeaturizer(
+            meta,
+            known_patterns=known_patterns,
+            pattern_vocabulary_stats=vocabulary_stats,
+            finetune_mode=should_finetune,
+        )
 
     def persist(self, file_name: Text, model_dir: Text) -> Optional[Dict[Text, Any]]:
         """Persist this model into the passed directory.
-        Return the metadata necessary to load the model again."""
-        file_name = file_name + ".pkl"
-        regex_file = os.path.join(model_dir, file_name)
+
+        Args:
+            file_name: Prefix to add to all files stored as part of this component.
+            model_dir: Path where files should be stored.
+
+        Returns:
+            Metadata necessary to load the model again.
+        """
+        patterns_file_name = file_name + ".patterns.pkl"
+        regex_file = Path(model_dir) / patterns_file_name
         utils.write_json_to_file(regex_file, self.known_patterns, indent=4)
+        vocabulary_stats_file_name = file_name + ".vocabulary_stats.pkl"
+        vocabulary_file = Path(model_dir) / vocabulary_stats_file_name
+        utils.write_json_to_file(vocabulary_file, self.vocabulary_stats, indent=4)
 
         return {"file": file_name}
diff --git a/rasa/nlu/model.py b/rasa/nlu/model.py
index 241863aeb7da..2765b4c01a83 100644
--- a/rasa/nlu/model.py
+++ b/rasa/nlu/model.py
@@ -1,6 +1,7 @@
 import copy
 import datetime
 import logging
+from math import ceil
 import os
 from typing import Any, Dict, List, Optional, Text
 
@@ -26,6 +27,7 @@
 from rasa.shared.nlu.training_data.training_data import TrainingData
 from rasa.shared.nlu.training_data.message import Message
 from rasa.nlu.utils import write_json_to_file
+from rasa.utils.tensorflow.constants import EPOCHS
 
 logger = logging.getLogger(__name__)
 
@@ -138,7 +140,8 @@ def __init__(
         cfg: RasaNLUModelConfig,
         component_builder: Optional[ComponentBuilder] = None,
         skip_validation: bool = False,
-    ):
+        model_to_finetune: Optional["Interpreter"] = None,
+    ) -> None:
 
         self.config = cfg
         self.skip_validation = skip_validation
@@ -154,8 +157,10 @@ def __init__(
         if not self.skip_validation:
             components.validate_requirements(cfg.component_names)
 
-        # build pipeline
-        self.pipeline = self._build_pipeline(cfg, component_builder)
+        if model_to_finetune:
+            self.pipeline = model_to_finetune.pipeline
+        else:
+            self.pipeline = self._build_pipeline(cfg, component_builder)
 
     def _build_pipeline(
         self, cfg: RasaNLUModelConfig, component_builder: ComponentBuilder
@@ -297,6 +302,8 @@ def load(
         model_dir: Text,
         component_builder: Optional[ComponentBuilder] = None,
         skip_validation: bool = False,
+        new_config: Optional[Dict] = None,
+        finetuning_epoch_fraction: float = 1.0,
     ) -> "Interpreter":
         """Create an interpreter based on a persisted model.
 
@@ -307,25 +314,76 @@ def load(
             model_dir: The path of the model to load
             component_builder: The
                 :class:`rasa.nlu.components.ComponentBuilder` to use.
+            new_config: Optional new config to use for the new epochs.
+            finetuning_epoch_fraction: Value to multiply all epochs by.
 
         Returns:
             An interpreter that uses the loaded model.
         """
-
         model_metadata = Metadata.load(model_dir)
 
+        if new_config:
+            Interpreter._update_metadata_epochs(
+                model_metadata, new_config, finetuning_epoch_fraction
+            )
+
         Interpreter.ensure_model_compatibility(model_metadata)
-        return Interpreter.create(model_metadata, component_builder, skip_validation)
+        return Interpreter.create(
+            model_metadata,
+            component_builder,
+            skip_validation,
+            should_finetune=new_config is not None,
+        )
+
+    @staticmethod
+    def _get_default_value_for_component(name: Text, key: Text) -> Any:
+        from rasa.nlu.registry import get_component_class
+
+        return get_component_class(name).defaults[key]
+
+    @staticmethod
+    def _update_metadata_epochs(
+        model_metadata: Metadata,
+        new_config: Optional[Dict] = None,
+        finetuning_epoch_fraction: float = 1.0,
+    ) -> Metadata:
+        for old_component_config, new_component_config in zip(
+            model_metadata.metadata["pipeline"], new_config["pipeline"]
+        ):
+            if EPOCHS in old_component_config:
+                new_epochs = new_component_config.get(
+                    EPOCHS,
+                    Interpreter._get_default_value_for_component(
+                        old_component_config["class"], EPOCHS
+                    ),
+                )
+                old_component_config[EPOCHS] = ceil(
+                    new_epochs * finetuning_epoch_fraction
+                )
+        return model_metadata
 
     @staticmethod
     def create(
         model_metadata: Metadata,
         component_builder: Optional[ComponentBuilder] = None,
         skip_validation: bool = False,
+        should_finetune: bool = False,
     ) -> "Interpreter":
-        """Load stored model and components defined by the provided metadata."""
+        """Create model and components defined by the provided metadata.
 
-        context = {}
+        Args:
+            model_metadata: The metadata describing each component.
+            component_builder: The
+                :class:`rasa.nlu.components.ComponentBuilder` to use.
+            skip_validation: If set to `True`, does not check that all
+                required packages for the components are installed
+                before loading them.
+            should_finetune: Indicates if the model components will be fine-tuned.
+
+        Returns:
+            An interpreter that uses the created model.
+        """
+        context = {"should_finetune": should_finetune}
 
         if component_builder is None:
             # If no builder is passed, every interpreter creation will result
diff --git a/rasa/nlu/selectors/response_selector.py b/rasa/nlu/selectors/response_selector.py
index 59b8d3b29276..f468449b658b 100644
--- a/rasa/nlu/selectors/response_selector.py
+++ b/rasa/nlu/selectors/response_selector.py
@@ -240,6 +240,7 @@ def __init__(
         model: Optional[RasaModel] = None,
         all_retrieval_intents: Optional[List[Text]] = None,
         responses: Optional[Dict[Text, List[Dict[Text, Any]]]] = None,
+        finetune_mode: bool = False,
     ) -> None:
 
         component_config = component_config or {}
@@ -256,7 +257,11 @@ def __init__(
         self.use_text_as_label = False
 
         super().__init__(
-            component_config, index_label_id_mapping, entity_tag_specs, model
+            component_config,
+            index_label_id_mapping,
+            entity_tag_specs,
+            model,
+            finetune_mode=finetune_mode,
         )
 
     @property
@@ -458,6 +463,7 @@ def _load_model_class(
         label_data: RasaModelData,
         entity_tag_specs: List[EntityTagSpec],
         meta: Dict[Text, Any],
+        finetune_mode: bool = False,
     ) -> "RasaModel":
 
         return cls.model_class(meta[USE_TEXT_AS_LABEL]).load(
@@ -467,6 +473,7 @@ def _load_model_class(
             label_data=label_data,
             entity_tag_specs=entity_tag_specs,
             config=copy.deepcopy(meta),
+            finetune_mode=finetune_mode,
         )
 
     def _instantiate_model_class(self, model_data: RasaModelData) -> "RasaModel":
diff --git a/rasa/nlu/train.py b/rasa/nlu/train.py
index 1a7ea7c2520f..4a7ae63151d4 100644
--- a/rasa/nlu/train.py
+++ b/rasa/nlu/train.py
@@ -3,7 +3,6 @@
 from typing import Any, Optional, Text, Tuple, Union, Dict
 
 import rasa.shared.utils.common
-import rasa.utils.common as common_utils
 from rasa.nlu import config, utils
 from rasa.nlu.components import ComponentBuilder
 from rasa.nlu.config import RasaNLUModelConfig
@@ -83,6 +82,7 @@ async def train(
     component_builder: Optional[ComponentBuilder] = None,
     training_data_endpoint: Optional[EndpointConfig] = None,
     persist_nlu_training_data: bool = False,
+    model_to_finetune: Optional[Interpreter] = None,
     **kwargs: Any,
 ) -> Tuple[Trainer, Interpreter, Optional[Text]]:
     """Loads the trainer and the data and runs the training of the model."""
@@ -94,7 +94,9 @@ async def train(
     # Ensure we are training a model that we can save in the end
     # WARN: there is still a race condition if a model with the same name is
     # trained in another subprocess
-    trainer = Trainer(nlu_config, component_builder)
+    trainer = Trainer(
+        nlu_config, component_builder, model_to_finetune=model_to_finetune
+    )
     persistor = create_persistor(storage)
     if training_data_endpoint is not None:
         training_data = await load_data_from_endpoint(
diff --git a/rasa/shared/nlu/training_data/training_data.py b/rasa/shared/nlu/training_data/training_data.py
index 268e05340fc5..55352a410924 100644
--- a/rasa/shared/nlu/training_data/training_data.py
+++ b/rasa/shared/nlu/training_data/training_data.py
@@ -79,6 +79,21 @@ def fingerprint(self) -> Text:
 
         return rasa.shared.utils.io.deep_container_fingerprint(relevant_attributes)
 
+    def label_fingerprint(self) -> Text:
+        """Fingerprints the labels in the training data.
+
+        Returns:
+            hex string as a fingerprint of the training data labels.
+        """
+        labels = {
+            "intents": sorted(self.intents),
+            "entities": sorted(self.entities),
+            "entity_groups": sorted(self.entity_groups),
+            "entity_roles": sorted(self.entity_roles),
+            "actions": sorted(self.action_names),
+        }
+        return rasa.shared.utils.io.deep_container_fingerprint(labels)
+
     def merge(self, *others: Optional["TrainingData"]) -> "TrainingData":
         """Return merged instance of this data with other training data.
 
@@ -181,9 +196,14 @@ def intents(self) -> Set[Text]:
         """Returns the set of intents in the training data."""
         return {ex.get(INTENT) for ex in self.training_examples} - {None}
 
+    @lazy_property
+    def action_names(self) -> Set[Text]:
+        """Returns the set of action names in the training data."""
+        return {ex.get(ACTION_NAME) for ex in self.training_examples} - {None}
+
     @lazy_property
     def retrieval_intents(self) -> Set[Text]:
-        """Returns the total number of response types in the training data"""
+        """Returns the total number of response types in the training data."""
         return {
             ex.get(INTENT)
             for ex in self.training_examples
diff --git a/rasa/telemetry.py b/rasa/telemetry.py
index c3573ff51702..fd6e410fcb3d 100644
--- a/rasa/telemetry.py
+++ b/rasa/telemetry.py
@@ -658,7 +658,7 @@ def initialize_error_reporting() -> None:
 
 @async_generator.asynccontextmanager
 async def track_model_training(
-    training_data: "TrainingDataImporter", model_type: Text
+    training_data: "TrainingDataImporter", model_type: Text, is_finetuning: bool = False
 ) -> typing.AsyncGenerator[None, None]:
     """Track a model training started.
 
@@ -670,6 +670,7 @@ async def track_model_training(
         training_data: Training data used for the training.
         model_type: Specifies the type of training, should be either "rasa", "core"
             or "nlu".
+        is_finetuning: `True` if the model is trained by finetuning another model.
     """
     if not initialize_telemetry():
         # telemetry reporting is disabled. we won't do any reporting
@@ -705,6 +706,7 @@ async def track_model_training(
             "num_lookup_tables": len(nlu_data.lookup_tables),
             "num_synonyms": len(nlu_data.entity_synonyms),
             "num_regexes": len(nlu_data.regex_features),
+            "is_finetuning": is_finetuning,
         },
     )
     start = datetime.now()
diff --git a/rasa/train.py b/rasa/train.py
index 43b3a9062d66..af7e1aa3f1a7 100644
--- a/rasa/train.py
+++ b/rasa/train.py
@@ -2,13 +2,23 @@
 import os
 import tempfile
 from contextlib import ExitStack
-from typing import Text, NamedTuple, Tuple, Optional, List, Union, Dict
+from typing import (
+    Any,
+    Text,
+    NamedTuple,
+    Tuple,
+    Optional,
+    List,
+    Union,
+    Dict,
+    TYPE_CHECKING,
+)
 
 import rasa.core.interpreter
 from rasa.shared.nlu.interpreter import NaturalLanguageInterpreter
 from rasa.shared.importers.importer import TrainingDataImporter
 from rasa import model, telemetry
-from rasa.model import FingerprintComparisonResult
+from rasa.model import Fingerprint, FingerprintComparisonResult
 from rasa.shared.core.domain import Domain
 from rasa.nlu.model import Interpreter
 import rasa.utils.common
@@ -28,6 +38,8 @@
     DEFAULT_NLU_SUBDIRECTORY_NAME,
 )
 
+from rasa.core.agent import Agent
+
 CODE_CORE_NEEDS_TO_BE_RETRAINED = 0b0001
 CODE_NLU_NEEDS_TO_BE_RETRAINED = 0b0010
 CODE_NLG_NEEDS_TO_BE_RETRAINED = 0b0100
@@ -53,6 +65,8 @@ def train(
     core_additional_arguments: Optional[Dict] = None,
     nlu_additional_arguments: Optional[Dict] = None,
     loop: Optional[asyncio.AbstractEventLoop] = None,
+    model_to_finetune: Optional[Text] = None,
+    finetuning_epoch_fraction: float = 1.0,
 ) -> TrainingResult:
     """Runs Rasa Core and NLU training in `async` loop.
 
@@ -70,7 +84,11 @@ def train(
         core_additional_arguments: Additional training parameters for core training.
         nlu_additional_arguments: Additional training parameters forwarded to training
             method of each NLU component.
-        loop: The event loop which will be used to run `async` functions.
+        loop: Optional EventLoop for running coroutines.
+        model_to_finetune: Optional path to a model which should be finetuned or
+            a directory in case the latest trained model should be used.
+        finetuning_epoch_fraction: The fraction currently specified training epochs
+            in the model configuration which should be used for finetuning.
 
     Returns:
         An instance of `TrainingResult`.
@@ -87,6 +105,8 @@ def train(
             persist_nlu_training_data=persist_nlu_training_data,
             core_additional_arguments=core_additional_arguments,
             nlu_additional_arguments=nlu_additional_arguments,
+            model_to_finetune=model_to_finetune,
+            finetuning_epoch_fraction=finetuning_epoch_fraction,
         ),
         loop,
     )
@@ -103,6 +123,8 @@ async def train_async(
     persist_nlu_training_data: bool = False,
     core_additional_arguments: Optional[Dict] = None,
     nlu_additional_arguments: Optional[Dict] = None,
+    model_to_finetune: Optional[Text] = None,
+    finetuning_epoch_fraction: float = 1.0,
 ) -> TrainingResult:
     """Trains a Rasa model (Core and NLU).
 
@@ -120,6 +142,10 @@ async def train_async(
         core_additional_arguments: Additional training parameters for core training.
         nlu_additional_arguments: Additional training parameters forwarded to training
             method of each NLU component.
+        model_to_finetune: Optional path to a model which should be finetuned or
+            a directory in case the latest trained model should be used.
+        finetuning_epoch_fraction: The fraction currently specified training epochs
+            in the model configuration which should be used for finetuning.
 
     Returns:
         An instance of `TrainingResult`.
@@ -127,9 +153,7 @@ async def train_async(
     file_importer = TrainingDataImporter.load_from_config(
         config, domain, training_files
     )
-    with ExitStack() as stack:
-        train_path = stack.enter_context(TempDirectoryPath(tempfile.mkdtemp()))
-
+    with TempDirectoryPath(tempfile.mkdtemp()) as train_path:
         domain = await file_importer.get_domain()
 
         if domain.is_empty():
@@ -148,16 +172,19 @@ async def train_async(
             persist_nlu_training_data,
             core_additional_arguments=core_additional_arguments,
             nlu_additional_arguments=nlu_additional_arguments,
+            model_to_finetune=model_to_finetune,
+            finetuning_epoch_fraction=finetuning_epoch_fraction,
         )
 
 
 async def handle_domain_if_not_exists(
     file_importer: TrainingDataImporter, output_path, fixed_model_name
 ):
+    """Trains only the nlu model and prints a warning about missing domain."""
     nlu_model_only = await _train_nlu_with_validated_data(
         file_importer, output=output_path, fixed_model_name=fixed_model_name
     )
-    print_warning(
+    rasa.shared.utils.cli.print_warning(
         "Core training was skipped because no valid domain file was found. "
         "Only an NLU-model was created. Please specify a valid domain using "
         "the '--domain' argument or check if the provided domain file exists."
@@ -213,6 +240,8 @@ async def _train_async_internal(
     persist_nlu_training_data: bool,
     core_additional_arguments: Optional[Dict] = None,
     nlu_additional_arguments: Optional[Dict] = None,
+    model_to_finetune: Optional[Text] = None,
+    finetuning_epoch_fraction: float = 1.0,
 ) -> TrainingResult:
     """Trains a Rasa model (Core and NLU). Use only from `train_async`.
 
@@ -229,6 +258,10 @@ async def _train_async_internal(
         core_additional_arguments: Additional training parameters for core training.
         nlu_additional_arguments: Additional training parameters forwarded to training
             method of each NLU component.
+        model_to_finetune: Optional path to a model which should be finetuned or
+            a directory in case the latest trained model should be used.
+        finetuning_epoch_fraction: The fraction currently specified training epochs
+            in the model configuration which should be used for finetuning.
 
     Returns:
         An instance of `TrainingResult`.
@@ -251,35 +284,45 @@ async def _train_async_internal(
         return TrainingResult(code=code)
 
     if stories.is_empty() and nlu_data.can_train_nlu_model():
-        print_error(
+        rasa.shared.utils.cli.print_error(
             "No training data given. Please provide stories and NLU data in "
             "order to train a Rasa model using the '--data' argument."
         )
         return TrainingResult()
 
     if stories.is_empty():
-        print_warning("No stories present. Just a Rasa NLU model will be trained.")
+        rasa.shared.utils.cli.print_warning(
+            "No stories present. Just a Rasa NLU model will be trained."
+        )
         trained_model = await _train_nlu_with_validated_data(
             file_importer,
             output=output_path,
             fixed_model_name=fixed_model_name,
             persist_nlu_training_data=persist_nlu_training_data,
             additional_arguments=nlu_additional_arguments,
+            model_to_finetune=model_to_finetune,
+            finetuning_epoch_fraction=finetuning_epoch_fraction,
         )
         return TrainingResult(model=trained_model)
 
     if nlu_data.can_train_nlu_model():
-        print_warning("No NLU data present. Just a Rasa Core model will be trained.")
+        rasa.shared.utils.cli.print_warning(
+            "No NLU data present. Just a Rasa Core model will be trained."
+        )
         trained_model = await _train_core_with_validated_data(
             file_importer,
             output=output_path,
             fixed_model_name=fixed_model_name,
             additional_arguments=core_additional_arguments,
+            model_to_finetune=model_to_finetune,
+            finetuning_epoch_fraction=finetuning_epoch_fraction,
         )
         return TrainingResult(model=trained_model)
 
     if fingerprint_comparison.is_training_required():
-        async with telemetry.track_model_training(file_importer, model_type="rasa"):
+        async with telemetry.track_model_training(
+            file_importer, model_type="rasa",
+        ):
             await _do_training(
                 file_importer,
                 output_path=output_path,
@@ -290,6 +333,8 @@ async def _train_async_internal(
                 core_additional_arguments=core_additional_arguments,
                 nlu_additional_arguments=nlu_additional_arguments,
                 old_model_zip_path=old_model,
+                model_to_finetune=model_to_finetune,
+                finetuning_epoch_fraction=finetuning_epoch_fraction,
             )
         trained_model = model.package_model(
             fingerprint=new_fingerprint,
@@ -299,7 +344,7 @@ async def _train_async_internal(
         )
         return TrainingResult(model=trained_model)
 
-    print_success(
+    rasa.shared.utils.cli.print_success(
         "Nothing changed. You can use the old model stored at '{}'."
         "".format(os.path.abspath(old_model))
     )
@@ -316,6 +361,8 @@ async def _do_training(
     core_additional_arguments: Optional[Dict] = None,
     nlu_additional_arguments: Optional[Dict] = None,
     old_model_zip_path: Optional[Text] = None,
+    model_to_finetune: Optional["Text"] = None,
+    finetuning_epoch_fraction: float = 1.0,
 ):
     if not fingerprint_comparison_result:
         fingerprint_comparison_result = FingerprintComparisonResult()
@@ -329,10 +376,12 @@ async def _do_training(
             fixed_model_name=fixed_model_name,
             persist_nlu_training_data=persist_nlu_training_data,
             additional_arguments=nlu_additional_arguments,
+            model_to_finetune=model_to_finetune,
+            finetuning_epoch_fraction=finetuning_epoch_fraction,
         )
         interpreter_path = os.path.join(model_path, DEFAULT_NLU_SUBDIRECTORY_NAME)
     else:
-        print_color(
+        rasa.shared.utils.cli.print_color(
             "NLU data/configuration did not change. No need to retrain NLU model.",
             color=rasa.shared.utils.io.bcolors.OKBLUE,
         )
@@ -346,9 +395,11 @@ async def _do_training(
             additional_arguments=core_additional_arguments,
             interpreter=_load_interpreter(interpreter_path)
             or _interpreter_from_previous_model(old_model_zip_path),
+            model_to_finetune=model_to_finetune,
+            finetuning_epoch_fraction=finetuning_epoch_fraction,
         )
     elif fingerprint_comparison_result.should_retrain_nlg():
-        print_color(
+        rasa.shared.utils.cli.print_color(
             "Core stories/configuration did not change. "
             "Only the templates section has been changed. A new model with "
             "the updated templates will be created.",
@@ -356,7 +407,7 @@ async def _do_training(
         )
         await model.update_model_with_new_domain(file_importer, train_path)
     else:
-        print_color(
+        rasa.shared.utils.cli.print_color(
             "Core stories/configuration did not change. No need to retrain Core model.",
             color=rasa.shared.utils.io.bcolors.OKBLUE,
         )
@@ -390,6 +441,8 @@ def train_core(
     train_path: Optional[Text] = None,
     fixed_model_name: Optional[Text] = None,
     additional_arguments: Optional[Dict] = None,
+    model_to_finetune: Optional[Text] = None,
+    finetuning_epoch_fraction: float = 1.0,
 ) -> Optional[Text]:
     return rasa.utils.common.run_in_loop(
         train_core_async(
@@ -400,6 +453,8 @@ def train_core(
             train_path=train_path,
             fixed_model_name=fixed_model_name,
             additional_arguments=additional_arguments,
+            model_to_finetune=model_to_finetune,
+            finetuning_epoch_fraction=finetuning_epoch_fraction,
         )
     )
 
@@ -412,6 +467,8 @@ async def train_core_async(
     train_path: Optional[Text] = None,
     fixed_model_name: Optional[Text] = None,
     additional_arguments: Optional[Dict] = None,
+    model_to_finetune: Optional[Text] = None,
+    finetuning_epoch_fraction: float = 1.0,
 ) -> Optional[Text]:
     """Trains a Core model.
 
@@ -424,19 +481,22 @@ async def train_core_async(
             directory, otherwise in the provided directory.
         fixed_model_name: Name of model to be stored.
         additional_arguments: Additional training parameters.
+        model_to_finetune: Optional path to a model which should be finetuned or
+            a directory in case the latest trained model should be used.
+        finetuning_epoch_fraction: The fraction currently specified training epochs
+            in the model configuration which should be used for finetuning.
 
     Returns:
         If `train_path` is given it returns the path to the model archive,
         otherwise the path to the directory with the trained model files.
 
     """
-
     file_importer = TrainingDataImporter.load_core_importer_from_config(
         config, domain, [stories]
     )
     domain = await file_importer.get_domain()
     if domain.is_empty():
-        print_error(
+        rasa.shared.utils.cli.print_error(
             "Core training was skipped because no valid domain file was found. "
             "Please specify a valid domain using '--domain' argument or check "
             "if the provided domain file exists."
@@ -444,7 +504,7 @@ async def train_core_async(
         return None
 
     if not await file_importer.get_stories():
-        print_error(
+        rasa.shared.utils.cli.print_error(
             "No stories given. Please provide stories in order to "
             "train a Rasa Core model using the '--stories' argument."
         )
@@ -456,6 +516,8 @@ async def train_core_async(
         train_path=train_path,
         fixed_model_name=fixed_model_name,
         additional_arguments=additional_arguments,
+        model_to_finetune=model_to_finetune,
+        finetuning_epoch_fraction=finetuning_epoch_fraction,
     )
 
 
@@ -466,9 +528,10 @@ async def _train_core_with_validated_data(
     fixed_model_name: Optional[Text] = None,
     additional_arguments: Optional[Dict] = None,
     interpreter: Optional[Interpreter] = None,
+    model_to_finetune: Optional["Text"] = None,
+    finetuning_epoch_fraction: float = 1.0,
 ) -> Optional[Text]:
     """Train Core with validated training and config data."""
-
     import rasa.core.train
 
     with ExitStack() as stack:
@@ -480,11 +543,32 @@ async def _train_core_with_validated_data(
             _train_path = stack.enter_context(TempDirectoryPath(tempfile.mkdtemp()))
 
         # normal (not compare) training
-        print_color("Training Core model...", color=rasa.shared.utils.io.bcolors.OKBLUE)
+        rasa.shared.utils.cli.print_color(
+            "Training Core model...", color=rasa.shared.utils.io.bcolors.OKBLUE
+        )
         domain, config = await asyncio.gather(
             file_importer.get_domain(), file_importer.get_config()
         )
-        async with telemetry.track_model_training(file_importer, model_type="core"):
+
+        if model_to_finetune:
+            model_to_finetune = await _core_model_for_finetuning(
+                model_to_finetune,
+                file_importer=file_importer,
+                finetuning_epoch_fraction=finetuning_epoch_fraction,
+            )
+
+            if not model_to_finetune:
+                rasa.shared.utils.cli.print_error_and_exit(
+                    f"No Core model for finetuning found. Please make sure to either "
+                    f"specify a path to a previous model or to have a finetunable "
+                    f"model within the directory '{output}'."
+                )
+
+        async with telemetry.track_model_training(
+            file_importer,
+            model_type="core",
+            is_finetuning=model_to_finetune is not None,
+        ):
             await rasa.core.train(
                 domain_file=domain,
                 training_resource=file_importer,
@@ -492,8 +576,9 @@ async def _train_core_with_validated_data(
                 policy_config=config,
                 additional_arguments=additional_arguments,
                 interpreter=interpreter,
+                model_to_finetune=model_to_finetune,
             )
-        print_color(
+        rasa.shared.utils.cli.print_color(
             "Core model training completed.", color=rasa.shared.utils.io.bcolors.OKBLUE
         )
 
@@ -511,6 +596,40 @@ async def _train_core_with_validated_data(
         return _train_path
 
 
+async def _core_model_for_finetuning(
+    model_to_finetune: Text,
+    file_importer: TrainingDataImporter,
+    finetuning_epoch_fraction: float = 1.0,
+) -> Optional[Agent]:
+    path_to_archive = model.get_model_for_finetuning(model_to_finetune)
+    if not path_to_archive:
+        return None
+
+    rasa.shared.utils.cli.print_info(
+        f"Loading Core model from {path_to_archive} for finetuning...",
+    )
+
+    with model.unpack_model(path_to_archive) as unpacked:
+        new_fingerprint = await model.model_fingerprint(file_importer)
+        old_fingerprint = model.fingerprint_from_path(unpacked)
+        if not model.can_finetune(old_fingerprint, new_fingerprint, core=True):
+            rasa.shared.utils.cli.print_error_and_exit(
+                "Core model can not be finetuned."
+            )
+
+        config = await file_importer.get_config()
+        agent = Agent.load(
+            unpacked,
+            new_config=config,
+            finetuning_epoch_fraction=finetuning_epoch_fraction,
+        )
+        # Agent might be empty if no underlying Core model was found.
+        if agent.domain is not None and agent.policy_ensemble is not None:
+            return agent
+
+        return None
+
+
 def train_nlu(
     config: Text,
     nlu_data: Text,
@@ -520,6 +639,8 @@ def train_nlu(
     persist_nlu_training_data: bool = False,
     additional_arguments: Optional[Dict] = None,
     domain: Optional[Union[Domain, Text]] = None,
+    model_to_finetune: Optional[Text] = None,
+    finetuning_epoch_fraction: float = 1.0,
 ) -> Optional[Text]:
     """Trains an NLU model.
 
@@ -535,14 +656,16 @@ def train_nlu(
         additional_arguments: Additional training parameters which will be passed to
                               the `train` method of each component.
         domain: Path to the optional domain file/Domain object.
-
+        model_to_finetune: Optional path to a model which should be finetuned or
+            a directory in case the latest trained model should be used.
+        finetuning_epoch_fraction: The fraction currently specified training epochs
+            in the model configuration which should be used for finetuning.
 
     Returns:
         If `train_path` is given it returns the path to the model archive,
         otherwise the path to the directory with the trained model files.
 
     """
-
     return rasa.utils.common.run_in_loop(
         _train_nlu_async(
             config,
@@ -553,6 +676,8 @@ def train_nlu(
             persist_nlu_training_data,
             additional_arguments,
             domain=domain,
+            model_to_finetune=model_to_finetune,
+            finetuning_epoch_fraction=finetuning_epoch_fraction,
         )
     )
 
@@ -566,9 +691,11 @@ async def _train_nlu_async(
     persist_nlu_training_data: bool = False,
     additional_arguments: Optional[Dict] = None,
     domain: Optional[Union[Domain, Text]] = None,
+    model_to_finetune: Optional[Text] = None,
+    finetuning_epoch_fraction: float = 1.0,
 ) -> Optional[Text]:
     if not nlu_data:
-        print_error(
+        rasa.shared.utils.cli.print_error(
             "No NLU data given. Please provide NLU data in order to train "
             "a Rasa NLU model using the '--nlu' argument."
         )
@@ -581,7 +708,7 @@ async def _train_nlu_async(
 
     training_data = await file_importer.get_nlu_data()
     if training_data.can_train_nlu_model():
-        print_error(
+        rasa.shared.utils.cli.print_error(
             f"Path '{nlu_data}' doesn't contain valid NLU data in it. "
             f"Please verify the data format. "
             f"The NLU model training will be skipped now."
@@ -595,6 +722,8 @@ async def _train_nlu_async(
         fixed_model_name=fixed_model_name,
         persist_nlu_training_data=persist_nlu_training_data,
         additional_arguments=additional_arguments,
+        model_to_finetune=model_to_finetune,
+        finetuning_epoch_fraction=finetuning_epoch_fraction,
     )
 
 
@@ -605,9 +734,10 @@ async def _train_nlu_with_validated_data(
     fixed_model_name: Optional[Text] = None,
     persist_nlu_training_data: bool = False,
     additional_arguments: Optional[Dict] = None,
+    model_to_finetune: Optional["Text"] = None,
+    finetuning_epoch_fraction: float = 1.0,
 ) -> Optional[Text]:
     """Train NLU with validated training and config data."""
-
     import rasa.nlu.train
 
     if additional_arguments is None:
@@ -621,17 +751,39 @@ async def _train_nlu_with_validated_data(
             # Otherwise, create a temp train path and clean it up on exit.
             _train_path = stack.enter_context(TempDirectoryPath(tempfile.mkdtemp()))
         config = await file_importer.get_config()
-        print_color("Training NLU model...", color=rasa.shared.utils.io.bcolors.OKBLUE)
-        async with telemetry.track_model_training(file_importer, model_type="nlu"):
+        rasa.shared.utils.cli.print_color(
+            "Training NLU model...", color=rasa.shared.utils.io.bcolors.OKBLUE
+        )
+
+        if model_to_finetune:
+            model_to_finetune = await _nlu_model_for_finetuning(
+                model_to_finetune,
+                file_importer,
+                finetuning_epoch_fraction,
+                called_from_combined_training=train_path is not None,
+            )
+            if not model_to_finetune:
+                rasa.shared.utils.cli.print_error_and_exit(
+                    f"No NLU model for finetuning found. Please make sure to either "
+                    f"specify a path to a previous model or to have a finetunable "
+                    f"model within the directory '{output}'."
+                )
+
+        async with telemetry.track_model_training(
+            file_importer,
+            model_type="nlu",
+            is_finetuning=model_to_finetune is not None,
+        ):
             await rasa.nlu.train(
                 config,
                 file_importer,
                 _train_path,
                 fixed_model_name="nlu",
                 persist_nlu_training_data=persist_nlu_training_data,
+                model_to_finetune=model_to_finetune,
                 **additional_arguments,
             )
-        print_color(
+        rasa.shared.utils.cli.print_color(
             "NLU model training completed.", color=rasa.shared.utils.io.bcolors.OKBLUE
         )
 
@@ -648,3 +800,42 @@ async def _train_nlu_with_validated_data(
             )
 
         return _train_path
+
+
+async def _nlu_model_for_finetuning(
+    model_to_finetune: Text,
+    file_importer: TrainingDataImporter,
+    finetuning_epoch_fraction: float = 1.0,
+    called_from_combined_training: bool = False,
+) -> Optional[Interpreter]:
+
+    path_to_archive = model.get_model_for_finetuning(model_to_finetune)
+    if not path_to_archive:
+        return None
+
+    rasa.shared.utils.cli.print_info(
+        f"Loading NLU model from {path_to_archive} for finetuning...",
+    )
+    with model.unpack_model(path_to_archive) as unpacked:
+        _, old_nlu = model.get_model_subdirectories(unpacked)
+        new_fingerprint = await model.model_fingerprint(file_importer)
+        old_fingerprint = model.fingerprint_from_path(unpacked)
+        if not model.can_finetune(
+            old_fingerprint,
+            new_fingerprint,
+            nlu=True,
+            core=called_from_combined_training,
+        ):
+            rasa.shared.utils.cli.print_error_and_exit(
+                "NLU model can not be finetuned."
+            )
+
+        config = await file_importer.get_config()
+        model_to_finetune = Interpreter.load(
+            old_nlu,
+            new_config=config,
+            finetuning_epoch_fraction=finetuning_epoch_fraction,
+        )
+        if not model_to_finetune:
+            return None
+    return model_to_finetune
diff --git a/rasa/utils/tensorflow/models.py b/rasa/utils/tensorflow/models.py
index 54e4148b6f87..3fa10c590a1e 100644
--- a/rasa/utils/tensorflow/models.py
+++ b/rasa/utils/tensorflow/models.py
@@ -327,9 +327,28 @@ def copy_best(self, model_file_name: Text) -> None:
 
     @classmethod
     def load(
-        cls, model_file_name: Text, model_data_example: RasaModelData, *args, **kwargs
+        cls,
+        model_file_name: Text,
+        model_data_example: RasaModelData,
+        finetune_mode: bool = False,
+        *args,
+        **kwargs,
     ) -> "RasaModel":
-        logger.debug("Loading the model ...")
+        """Loads a model from the given weights.
+
+        Args:
+            model_file_name: Path to file containing model weights.
+            model_data_example: Example data point to construct the model architecture.
+            finetune_mode: Indicates whether to load the model for further finetuning.
+            *args: Any other non key-worded arguments.
+            **kwargs: Any other key-worded arguments.
+
+        Returns:
+            Loaded model with weights appropriately set.
+        """
+        logger.debug(
+            f"Loading the model from {model_file_name} with finetune_mode={finetune_mode}..."
+        )
         # create empty model
         model = cls(*args, **kwargs)
         # need to train on 1 example to build weights of the correct size
@@ -341,8 +360,10 @@ def load(
             evaluate_on_num_examples=0,
             batch_strategy=SEQUENCE,
             silent=True,  # don't confuse users with training output
-            loading=True,  # don't use tensorboard while loading
-            eager=True,  # no need to build tf graph, eager is faster here
+            loading=True,  # don't use tensorboard when doing a dummy fit run
+            eager=False
+            if finetune_mode
+            else True,  # load in eager mode only for prediction phase
         )
         # load trained weights
         model.load_weights(model_file_name)
diff --git a/tests/cli/test_rasa_interactive.py b/tests/cli/test_rasa_interactive.py
index 7afdb91558cf..92ee20ae9405 100644
--- a/tests/cli/test_rasa_interactive.py
+++ b/tests/cli/test_rasa_interactive.py
@@ -20,7 +20,9 @@ def test_interactive_help(run: Callable[..., RunResult]):
                         [--conversation-id CONVERSATION_ID]
                         [--endpoints ENDPOINTS] [-c CONFIG] [-d DOMAIN]
                         [--out OUT] [--augmentation AUGMENTATION]
-                        [--debug-plots] [--force] [--persist-nlu-data]
+                        [--debug-plots] [--finetune [FINETUNE]]
+                        [--epoch-fraction EPOCH_FRACTION] [--force]
+                        [--persist-nlu-data]
                         {core} ... [model-as-positional-argument]"""
 
     lines = help_text.split("\n")
@@ -38,7 +40,8 @@ def test_interactive_core_help(run: Callable[..., RunResult]):
                              [--conversation-id CONVERSATION_ID]
                              [--endpoints ENDPOINTS] [-c CONFIG] [-d DOMAIN]
                              [--out OUT] [--augmentation AUGMENTATION]
-                             [--debug-plots] [-p PORT]
+                             [--debug-plots] [--finetune [FINETUNE]]
+                             [--epoch-fraction EPOCH_FRACTION] [-p PORT]
                              [model-as-positional-argument]"""
 
     lines = help_text.split("\n")
diff --git a/tests/cli/test_rasa_train.py b/tests/cli/test_rasa_train.py
index 72ea57e6789f..a17fdfca360d 100644
--- a/tests/cli/test_rasa_train.py
+++ b/tests/cli/test_rasa_train.py
@@ -378,7 +378,7 @@ def test_train_nlu_persist_nlu_data(
     )
 
 
-def test_train_help(run):
+def test_train_help(run: Callable[..., RunResult]):
     output = run("train", "--help")
 
     help_text = """usage: rasa train [-h] [-v] [-vv] [--quiet] [--data DATA [DATA ...]]
@@ -386,12 +386,14 @@ def test_train_help(run):
                   [--augmentation AUGMENTATION] [--debug-plots]
                   [--num-threads NUM_THREADS]
                   [--fixed-model-name FIXED_MODEL_NAME] [--persist-nlu-data]
-                  [--force]
+                  [--force] [--finetune [FINETUNE]]
+                  [--epoch-fraction EPOCH_FRACTION]
                   {core,nlu} ..."""
 
     lines = help_text.split("\n")
     # expected help text lines should appear somewhere in the output
     printed_help = set(output.outlines)
+
     for line in lines:
         assert line in printed_help
 
@@ -402,7 +404,8 @@ def test_train_nlu_help(run: Callable[..., RunResult]):
     help_text = """usage: rasa train nlu [-h] [-v] [-vv] [--quiet] [-c CONFIG] [-d DOMAIN]
                       [--out OUT] [-u NLU] [--num-threads NUM_THREADS]
                       [--fixed-model-name FIXED_MODEL_NAME]
-                      [--persist-nlu-data]"""
+                      [--persist-nlu-data] [--finetune [FINETUNE]]
+                      [--epoch-fraction EPOCH_FRACTION]"""
 
     lines = help_text.split("\n")
     # expected help text lines should appear somewhere in the output
@@ -419,7 +422,8 @@ def test_train_core_help(run: Callable[..., RunResult]):
                        [--augmentation AUGMENTATION] [--debug-plots] [--force]
                        [--fixed-model-name FIXED_MODEL_NAME]
                        [--percentages [PERCENTAGES [PERCENTAGES ...]]]
-                       [--runs RUNS]"""
+                       [--runs RUNS] [--finetune [FINETUNE]]
+                       [--epoch-fraction EPOCH_FRACTION]"""
 
     lines = help_text.split("\n")
     # expected help text lines should appear somewhere in the output
diff --git a/tests/conftest.py b/tests/conftest.py
index 2fab94fa07a6..082e82f4e8df 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,13 +1,16 @@
 import asyncio
+import copy
 import os
 import random
+
+import mock
 import pytest
 import sys
 import uuid
 
 from sanic.request import Request
 
-from typing import Iterator, Callable
+from typing import Iterator, Callable, Generator
 
 from _pytest.tmpdir import TempdirFactory
 from pathlib import Path
@@ -29,7 +32,7 @@
 import rasa.core.run
 from rasa.core.tracker_store import InMemoryTrackerStore, TrackerStore
 from rasa.model import get_model
-from rasa.train import TrainingResult, train_async
+from rasa.train import TrainingResult, train_async, _train_nlu_async
 from rasa.utils.common import TempDirectoryPath
 from tests.core.conftest import (
     DEFAULT_DOMAIN_PATH_WITH_SLOTS,
@@ -90,7 +93,7 @@ async def default_agent(_trained_default_agent: Agent) -> Agent:
 
 
 @pytest.fixture(scope="session")
-async def trained_moodbot_path(trained_async) -> Text:
+async def trained_moodbot_path(trained_async: Callable) -> Text:
     return await trained_async(
         domain="examples/moodbot/domain.yml",
         config="examples/moodbot/config.yml",
@@ -98,6 +101,15 @@ async def trained_moodbot_path(trained_async) -> Text:
     )
 
 
+@pytest.fixture(scope="session")
+async def trained_nlu_moodbot_path(trained_nlu_async: Callable) -> Text:
+    return await trained_nlu_async(
+        domain="examples/moodbot/domain.yml",
+        config="examples/moodbot/config.yml",
+        nlu_data="examples/moodbot/data/nlu.yml",
+    )
+
+
 @pytest.fixture(scope="session")
 async def unpacked_trained_moodbot_path(
     trained_moodbot_path: Text,
@@ -126,10 +138,15 @@ def default_domain_path() -> Text:
 
 
 @pytest.fixture(scope="session")
-def default_domain() -> Domain:
+def _default_domain() -> Domain:
     return Domain.load(DEFAULT_DOMAIN_PATH_WITH_SLOTS)
 
 
+@pytest.fixture()
+def default_domain(_default_domain: Domain) -> Domain:
+    return copy.deepcopy(_default_domain)
+
+
 @pytest.fixture(scope="session")
 def default_stories_file() -> Text:
     return DEFAULT_STORIES_FILE
@@ -156,8 +173,13 @@ def end_to_end_story_file() -> Text:
 
 
 @pytest.fixture(scope="session")
-def default_config() -> List[Policy]:
-    return config.load(DEFAULT_CONFIG_PATH)
+def default_config_path() -> Text:
+    return DEFAULT_CONFIG_PATH
+
+
+@pytest.fixture(scope="session")
+def default_config(default_config_path) -> List[Policy]:
+    return config.load(default_config_path)
 
 
 @pytest.fixture(scope="session")
@@ -174,6 +196,19 @@ async def _train(
     return _train
 
 
+@pytest.fixture(scope="session")
+def trained_nlu_async(tmpdir_factory: TempdirFactory) -> Callable:
+    async def _train_nlu(
+        *args: Any, output_path: Optional[Text] = None, **kwargs: Any
+    ) -> Optional[Text]:
+        if output_path is None:
+            output_path = str(tmpdir_factory.mktemp("models"))
+
+        return await _train_nlu_async(*args, output=output_path, **kwargs)
+
+    return _train_nlu
+
+
 @pytest.fixture(scope="session")
 async def trained_rasa_model(
     trained_async: Callable,
@@ -190,6 +225,14 @@ async def trained_rasa_model(
     return trained_stack_model_path
 
 
+@pytest.fixture(scope="session")
+async def unpacked_trained_rasa_model(
+    trained_rasa_model: Text,
+) -> Generator[Text, None, None]:
+    with get_model(trained_rasa_model) as path:
+        yield path
+
+
 @pytest.fixture(scope="session")
 async def trained_core_model(
     trained_async: Callable,
@@ -310,3 +353,10 @@ def __init__(
         endpoints_path: Text = "",
     ) -> None:
         super().__init__(tracker_store, event_broker, endpoints_path)
+
+
+class AsyncMock(Mock):
+    """Helper class to mock async functions and methods."""
+
+    async def __call__(self, *args: Any, **kwargs: Any) -> Any:
+        return super().__call__(*args, **kwargs)
diff --git a/tests/core/conftest.py b/tests/core/conftest.py
index 805884bc166e..095342e288de 100644
--- a/tests/core/conftest.py
+++ b/tests/core/conftest.py
@@ -72,8 +72,8 @@ def as_feature(self):
 
 # noinspection PyAbstractClass,PyUnusedLocal,PyMissingConstructor
 class ExamplePolicy(Policy):
-    def __init__(self, example_arg):
-        super(ExamplePolicy, self).__init__()
+    def __init__(self, *args, **kwargs):
+        super(ExamplePolicy, self).__init__(*args, **kwargs)
 
 
 class MockedMongoTrackerStore(MongoTrackerStore):
@@ -190,13 +190,6 @@ def tracker_with_six_scheduled_reminders(
     return tracker
 
 
-@pytest.fixture(scope="session")
-def moodbot_metadata(unpacked_trained_moodbot_path: Text) -> PolicyEnsemble:
-    return PolicyEnsemble.load_metadata(
-        os.path.join(unpacked_trained_moodbot_path, "core")
-    )
-
-
 @pytest.fixture
 def default_nlg(default_domain: Domain) -> NaturalLanguageGenerator:
     return TemplatedNaturalLanguageGenerator(default_domain.templates)
diff --git a/tests/core/policies/test_rule_policy.py b/tests/core/policies/test_rule_policy.py
index b8b86a4d67df..98116619dfda 100644
--- a/tests/core/policies/test_rule_policy.py
+++ b/tests/core/policies/test_rule_policy.py
@@ -656,6 +656,48 @@ def test_all_policy_attributes_are_persisted(tmpdir: Path):
     assert persisted_policy._enable_fallback_prediction == enable_fallback_prediction
 
 
+async def test_rule_policy_finetune(
+    tmp_path: Path, trained_rule_policy: RulePolicy, trained_rule_policy_domain: Domain
+):
+
+    trained_rule_policy.persist(tmp_path)
+
+    loaded_policy = RulePolicy.load(tmp_path, should_finetune=True)
+
+    assert loaded_policy.finetune_mode
+
+    new_rule = TrackerWithCachedStates.from_events(
+        "stop story",
+        domain=trained_rule_policy_domain,
+        slots=trained_rule_policy_domain.slots,
+        evts=[
+            ActionExecuted(RULE_SNIPPET_ACTION_NAME),
+            ActionExecuted(ACTION_LISTEN_NAME),
+            UserUttered(intent={"name": "stopp"}),
+            ActionExecuted("utter_stop"),
+            ActionExecuted(ACTION_LISTEN_NAME),
+        ],
+        is_rule_tracker=True,
+    )
+
+    original_data = await training.load_data(
+        "examples/rules/data/rules.yml", trained_rule_policy_domain
+    )
+
+    loaded_policy.train(
+        original_data + [new_rule], trained_rule_policy_domain, RegexInterpreter()
+    )
+
+    assert (
+        len(loaded_policy.lookup["rules"])
+        == len(trained_rule_policy.lookup["rules"]) + 1
+    )
+    assert (
+        """[{"prev_action": {"action_name": "action_listen"}, "user": {"intent": "stopp"}}]"""
+        in loaded_policy.lookup["rules"]
+    )
+
+
 def test_faq_rule():
     domain = Domain.from_yaml(
         f"""
diff --git a/tests/core/test_actions.py b/tests/core/test_actions.py
index cbfd29eb567f..bfce0a5bdc91 100644
--- a/tests/core/test_actions.py
+++ b/tests/core/test_actions.py
@@ -75,8 +75,8 @@ def template_nlg():
 
 
 @pytest.fixture(scope="module")
-def template_sender_tracker(default_domain):
-    return DialogueStateTracker("template-sender", default_domain.slots)
+def template_sender_tracker(_default_domain: Domain):
+    return DialogueStateTracker("template-sender", _default_domain.slots)
 
 
 def test_text_format():
diff --git a/tests/core/test_agent.py b/tests/core/test_agent.py
index 2144f25cc38b..4972c4300103 100644
--- a/tests/core/test_agent.py
+++ b/tests/core/test_agent.py
@@ -1,4 +1,5 @@
 import asyncio
+from pathlib import Path
 from typing import Any, Dict, Text, List, Callable, Optional
 from unittest.mock import Mock
 
@@ -52,9 +53,9 @@ async def model(request: Request) -> StreamingHTTPResponse:
 
 @pytest.fixture()
 def model_server(
-    loop: asyncio.AbstractEventLoop, sanic_client: Callable, trained_moodbot_path: Text
+    loop: asyncio.AbstractEventLoop, sanic_client: Callable, trained_rasa_model: Text
 ) -> TestClient:
-    app = model_server_app(trained_moodbot_path, model_hash="somehash")
+    app = model_server_app(trained_rasa_model, model_hash="somehash")
     return loop.run_until_complete(sanic_client(app))
 
 
@@ -157,7 +158,7 @@ def test_agent_wrong_use_of_load():
 
 
 async def test_agent_with_model_server_in_thread(
-    model_server: TestClient, moodbot_domain: Domain, moodbot_metadata: Any
+    model_server: TestClient, default_domain: Domain, unpacked_trained_rasa_model: Text
 ):
     model_endpoint_config = EndpointConfig.from_dict(
         {"url": model_server.make_url("/model"), "wait_time_between_pulls": 2}
@@ -171,14 +172,17 @@ async def test_agent_with_model_server_in_thread(
     await asyncio.sleep(5)
 
     assert agent.fingerprint == "somehash"
-    assert hash(agent.domain) == hash(moodbot_domain)
+    assert agent.domain.as_dict() == default_domain.as_dict()
+
+    expected_policies = PolicyEnsemble.load_metadata(
+        str(Path(unpacked_trained_rasa_model, "core"))
+    )["policy_names"]
 
     agent_policies = {
         rasa.shared.utils.common.module_path_from_instance(p)
         for p in agent.policy_ensemble.policies
     }
-    moodbot_policies = set(moodbot_metadata["policy_names"])
-    assert agent_policies == moodbot_policies
+    assert agent_policies == set(expected_policies)
     assert model_server.app.number_of_model_requests == 1
     jobs.kill_scheduler()
 
diff --git a/tests/core/test_ensemble.py b/tests/core/test_ensemble.py
index 7f90f2591e93..050ad361c45f 100644
--- a/tests/core/test_ensemble.py
+++ b/tests/core/test_ensemble.py
@@ -1,9 +1,13 @@
 from pathlib import Path
-from typing import List, Any, Text, Optional
+from typing import List, Any, Text, Optional, Union
+from unittest.mock import Mock
 
+from _pytest.capture import CaptureFixture
+from _pytest.monkeypatch import MonkeyPatch
 import pytest
 import copy
 
+from rasa.core.exceptions import UnsupportedDialogueModelError
 from rasa.core.policies.memoization import MemoizationPolicy, AugmentedMemoizationPolicy
 from rasa.shared.nlu.interpreter import NaturalLanguageInterpreter, RegexInterpreter
 
@@ -37,7 +41,7 @@
 
 class WorkingPolicy(Policy):
     @classmethod
-    def load(cls, _) -> Policy:
+    def load(cls, *args: Any, **kwargs: Any) -> Policy:
         return WorkingPolicy()
 
     def persist(self, _) -> None:
@@ -74,6 +78,37 @@ def test_policy_loading_simple(tmp_path: Path):
     assert original_policy_ensemble.policies == loaded_policy_ensemble.policies
 
 
+class PolicyWithoutLoadKwargs(Policy):
+    @classmethod
+    def load(cls, path: Union[Text, Path]) -> Policy:
+        return PolicyWithoutLoadKwargs()
+
+    def persist(self, _) -> None:
+        pass
+
+
+def test_policy_loading_no_kwargs_with_context(tmp_path: Path):
+    original_policy_ensemble = PolicyEnsemble([PolicyWithoutLoadKwargs()])
+    original_policy_ensemble.train([], None, RegexInterpreter())
+    original_policy_ensemble.persist(str(tmp_path))
+
+    with pytest.raises(UnsupportedDialogueModelError) as execinfo:
+        PolicyEnsemble.load(str(tmp_path), new_config={"policies": [{}]})
+    assert "`PolicyWithoutLoadKwargs.load` does not accept `**kwargs`" in str(
+        execinfo.value
+    )
+
+
+def test_policy_loading_no_kwargs_with_no_context(
+    tmp_path: Path, capsys: CaptureFixture
+):
+    original_policy_ensemble = PolicyEnsemble([PolicyWithoutLoadKwargs()])
+    original_policy_ensemble.train([], None, RegexInterpreter())
+    original_policy_ensemble.persist(str(tmp_path))
+    with pytest.warns(FutureWarning):
+        PolicyEnsemble.load(str(tmp_path))
+
+
 class ConstantPolicy(Policy):
     def __init__(
         self,
@@ -83,8 +118,9 @@ def __init__(
         is_end_to_end_prediction: bool = False,
         events: Optional[List[Event]] = None,
         optional_events: Optional[List[Event]] = None,
+        **kwargs: Any,
     ) -> None:
-        super().__init__(priority=priority)
+        super().__init__(priority=priority, **kwargs)
         self.predict_index = predict_index
         self.confidence = confidence
         self.is_end_to_end_prediction = is_end_to_end_prediction
@@ -92,7 +128,7 @@ def __init__(
         self.optional_events = optional_events or []
 
     @classmethod
-    def load(cls, _) -> Policy:
+    def load(cls, args, **kwargs) -> Policy:
         pass
 
     def persist(self, _) -> None:
@@ -304,7 +340,7 @@ def test_fallback_wins_over_mapping():
 
 class LoadReturnsNonePolicy(Policy):
     @classmethod
-    def load(cls, _) -> None:
+    def load(cls, *args, **kwargs) -> None:
         return None
 
     def persist(self, _) -> None:
@@ -340,7 +376,7 @@ def test_policy_loading_load_returns_none(tmp_path: Path):
 
 class LoadReturnsWrongTypePolicy(Policy):
     @classmethod
-    def load(cls, _) -> Text:
+    def load(cls, *args, **kwargs) -> Text:
         return ""
 
     def persist(self, _) -> None:
diff --git a/tests/core/test_policies.py b/tests/core/test_policies.py
index b1a65f7e5d0f..3aa109bd896d 100644
--- a/tests/core/test_policies.py
+++ b/tests/core/test_policies.py
@@ -7,8 +7,10 @@
 from _pytest.monkeypatch import MonkeyPatch
 
 from rasa.core.channels import OutputChannel
+from rasa.core.exceptions import UnsupportedDialogueModelError
 from rasa.core.nlg import NaturalLanguageGenerator
 from rasa.shared.core.generator import TrackerWithCachedStates
+import rasa.shared.utils.io
 
 from rasa.core import training
 import rasa.core.actions.action
@@ -30,7 +32,12 @@
     USER,
 )
 from rasa.shared.core.domain import State, Domain
-from rasa.shared.core.events import ActionExecuted, ConversationPaused, Event
+from rasa.shared.core.events import (
+    ActionExecuted,
+    ConversationPaused,
+    Event,
+    UserUttered,
+)
 from rasa.core.featurizers.single_state_featurizer import SingleStateFeaturizer
 from rasa.core.featurizers.tracker_featurizers import (
     MaxHistoryTrackerFeaturizer,
@@ -136,11 +143,20 @@ def test_featurizer(self, trained_policy: Policy, tmp_path: Path):
         assert loaded.featurizer.max_history == self.max_history
         assert isinstance(loaded.featurizer.state_featurizer, SingleStateFeaturizer)
 
+    @pytest.mark.parametrize("should_finetune", [False, True])
     async def test_persist_and_load(
-        self, trained_policy: Policy, default_domain: Domain, tmp_path: Path
+        self,
+        trained_policy: Policy,
+        default_domain: Domain,
+        tmp_path: Path,
+        should_finetune: bool,
     ):
         trained_policy.persist(str(tmp_path))
-        loaded = trained_policy.__class__.load(str(tmp_path))
+        loaded = trained_policy.__class__.load(
+            str(tmp_path), should_finetune=should_finetune
+        )
+        assert loaded.finetune_mode == should_finetune
+
         trackers = await train_trackers(default_domain, augmentation_factor=20)
 
         for tracker in trackers:
@@ -335,6 +351,24 @@ def test_train_with_shuffle_false(
         # does not raise
         policy.train(trackers, domain=default_domain, interpreter=RegexInterpreter())
 
+    def test_finetune_after_load(
+        self,
+        trained_policy: SklearnPolicy,
+        trackers: List[TrackerWithCachedStates],
+        default_domain: Domain,
+        tmp_path: Path,
+    ):
+
+        trained_policy.persist(tmp_path)
+
+        loaded_policy = SklearnPolicy.load(tmp_path, should_finetune=True)
+
+        assert loaded_policy.finetune_mode
+
+        loaded_policy.train(trackers, default_domain, RegexInterpreter())
+
+        assert loaded_policy.model
+
 
 class TestTEDPolicy(PolicyTestCollection):
     def test_train_model_checkpointing(self, tmp_path: Path):
@@ -720,6 +754,44 @@ def test_memorise_with_nlu(
         recalled = trained_policy.recall(states, tracker, default_domain)
         assert recalled is not None
 
+    async def test_finetune_after_load(
+        self, trained_policy: MemoizationPolicy, default_domain: Domain, tmp_path: Path
+    ):
+
+        trained_policy.persist(tmp_path)
+
+        loaded_policy = MemoizationPolicy.load(tmp_path, should_finetune=True)
+
+        assert loaded_policy.finetune_mode
+
+        new_story = TrackerWithCachedStates.from_events(
+            "channel",
+            domain=default_domain,
+            slots=default_domain.slots,
+            evts=[
+                ActionExecuted(ACTION_LISTEN_NAME),
+                UserUttered(intent={"name": "why"}),
+                ActionExecuted("utter_channel"),
+                ActionExecuted(ACTION_LISTEN_NAME),
+            ],
+        )
+        original_train_data = await train_trackers(
+            default_domain, augmentation_factor=20
+        )
+        loaded_policy.train(
+            original_train_data + [new_story], default_domain, RegexInterpreter()
+        )
+
+        # Get the hash of the tracker state of new story
+        new_story_states, _ = loaded_policy.featurizer.training_states_and_actions(
+            [new_story], default_domain
+        )
+
+        # Feature keys for each new state should be present in the lookup
+        for states in new_story_states:
+            state_key = loaded_policy._create_feature_key(states)
+            assert state_key in loaded_policy.lookup
+
 
 class TestAugmentedMemoizationPolicy(TestMemoizationPolicy):
     def create_policy(
@@ -1229,3 +1301,34 @@ def test_get_training_trackers_for_policy(
 def test_deprecation_warnings_for_old_rule_like_policies(policy: Type[Policy]):
     with pytest.warns(FutureWarning):
         policy(None)
+
+
+class PolicyWithoutInitKwargs(Policy):
+    def __init__(self, *args: Any) -> None:
+        pass
+
+    def persist(self, _) -> None:
+        pass
+
+    @classmethod
+    def _metadata_filename(cls) -> Text:
+        return "no_finetune_policy"
+
+
+def test_loading_policy_with_no_constructor_kwargs(tmp_path: Path):
+    rasa.shared.utils.io.write_text_file(
+        "{}", tmp_path / PolicyWithoutInitKwargs._metadata_filename()
+    )
+    with pytest.raises(UnsupportedDialogueModelError) as execinfo:
+        PolicyWithoutInitKwargs.load(str(tmp_path), should_finetune=True)
+    assert "`PolicyWithoutInitKwargs.__init__` does not accept `**kwargs`." in str(
+        execinfo.value
+    )
+
+
+def test_loading_policy_with_no_constructor_kwargs_but_required_args(tmp_path: Path):
+    rasa.shared.utils.io.write_text_file(
+        "{}", tmp_path / PolicyWithoutInitKwargs._metadata_filename()
+    )
+    with pytest.warns(FutureWarning):
+        PolicyWithoutInitKwargs.load(str(tmp_path))
diff --git a/tests/nlu/classifiers/test_diet_classifier.py b/tests/nlu/classifiers/test_diet_classifier.py
index dbf6e3ab11f0..6f23f613fe59 100644
--- a/tests/nlu/classifiers/test_diet_classifier.py
+++ b/tests/nlu/classifiers/test_diet_classifier.py
@@ -3,6 +3,7 @@
 import numpy as np
 import pytest
 from unittest.mock import Mock
+from typing import List, Tuple, Text, Dict, Any, Optional
 
 from rasa.shared.nlu.training_data.features import Features
 from rasa.nlu import train
@@ -108,7 +109,10 @@ def test_check_labels_features_exist(messages, expected):
 
 
 async def _train_persist_load_with_different_settings(
-    pipeline, component_builder, tmp_path
+    pipeline: List[Dict[Text, Any]],
+    component_builder: ComponentBuilder,
+    tmp_path: Path,
+    should_finetune: bool,
 ):
     _config = RasaNLUModelConfig({"pipeline": pipeline, "language": "en"})
 
@@ -122,7 +126,11 @@ async def _train_persist_load_with_different_settings(
     assert trainer.pipeline
     assert trained.pipeline
 
-    loaded = Interpreter.load(persisted_path, component_builder)
+    loaded = Interpreter.load(
+        persisted_path,
+        component_builder,
+        new_config=_config if should_finetune else None,
+    )
 
     assert loaded.pipeline
     assert loaded.parse("Rasa is great!") == trained.parse("Rasa is great!")
@@ -130,7 +138,7 @@ async def _train_persist_load_with_different_settings(
 
 @pytest.mark.skip_on_windows
 async def test_train_persist_load_with_different_settings_non_windows(
-    component_builder, tmpdir
+    component_builder: ComponentBuilder, tmp_path: Path
 ):
     pipeline = [
         {
@@ -142,7 +150,10 @@ async def test_train_persist_load_with_different_settings_non_windows(
         {"name": "DIETClassifier", MASKED_LM: True, EPOCHS: 1},
     ]
     await _train_persist_load_with_different_settings(
-        pipeline, component_builder, tmpdir
+        pipeline, component_builder, tmp_path, should_finetune=False
+    )
+    await _train_persist_load_with_different_settings(
+        pipeline, component_builder, tmp_path, should_finetune=True
     )
 
 
@@ -153,7 +164,10 @@ async def test_train_persist_load_with_different_settings(component_builder, tmp
         {"name": "DIETClassifier", LOSS_TYPE: "margin", EPOCHS: 1},
     ]
     await _train_persist_load_with_different_settings(
-        pipeline, component_builder, tmpdir
+        pipeline, component_builder, tmpdir, should_finetune=False
+    )
+    await _train_persist_load_with_different_settings(
+        pipeline, component_builder, tmpdir, should_finetune=True
     )
 
 
diff --git a/tests/nlu/featurizers/test_count_vectors_featurizer.py b/tests/nlu/featurizers/test_count_vectors_featurizer.py
index a31cc054489f..17fc80224860 100644
--- a/tests/nlu/featurizers/test_count_vectors_featurizer.py
+++ b/tests/nlu/featurizers/test_count_vectors_featurizer.py
@@ -1,9 +1,10 @@
-from typing import List, Any
-
+from typing import List, Any, Text, Optional
 import numpy as np
 import pytest
 import scipy.sparse
-from typing import Text
+from pathlib import Path
+from _pytest.logging import LogCaptureFixture
+import logging
 
 from rasa.nlu.tokenizers.spacy_tokenizer import SpacyTokenizer
 from rasa.nlu.config import RasaNLUModelConfig
@@ -28,7 +29,7 @@
     ],
 )
 def test_count_vector_featurizer(sentence, expected, expected_cls):
-    ftr = CountVectorsFeaturizer()
+    ftr = CountVectorsFeaturizer({"additional_vocabulary_size": {"text": 0}})
 
     train_message = Message(data={TEXT: sentence})
     test_message = Message(data={TEXT: sentence})
@@ -63,7 +64,9 @@ def test_count_vector_featurizer(sentence, expected, expected_cls):
 def test_count_vector_featurizer_response_attribute_featurization(
     sentence, intent, response, intent_features, response_features
 ):
-    ftr = CountVectorsFeaturizer()
+    ftr = CountVectorsFeaturizer(
+        {"additional_vocabulary_size": {"text": 0, "response": 0}}
+    )
     tk = WhitespaceTokenizer()
 
     train_message = Message(data={TEXT: sentence})
@@ -121,7 +124,9 @@ def test_count_vector_featurizer_response_attribute_featurization(
 def test_count_vector_featurizer_attribute_featurization(
     sentence, intent, response, intent_features, response_features
 ):
-    ftr = CountVectorsFeaturizer()
+    ftr = CountVectorsFeaturizer(
+        {"additional_vocabulary_size": {"text": 0, "response": 0}}
+    )
     tk = WhitespaceTokenizer()
 
     train_message = Message(data={TEXT: sentence})
@@ -178,7 +183,12 @@ def test_count_vector_featurizer_attribute_featurization(
 def test_count_vector_featurizer_shared_vocab(
     sentence, intent, response, text_features, intent_features, response_features
 ):
-    ftr = CountVectorsFeaturizer({"use_shared_vocab": True})
+    ftr = CountVectorsFeaturizer(
+        {
+            "use_shared_vocab": True,
+            "additional_vocabulary_size": {"text": 0, "response": 0},
+        }
+    )
     tk = WhitespaceTokenizer()
 
     train_message = Message(data={TEXT: sentence})
@@ -223,7 +233,9 @@ def test_count_vector_featurizer_shared_vocab(
     ],
 )
 def test_count_vector_featurizer_oov_token(sentence, expected):
-    ftr = CountVectorsFeaturizer({"OOV_token": "__oov__"})
+    ftr = CountVectorsFeaturizer(
+        {"OOV_token": "__oov__", "additional_vocabulary_size": {"text": 0}}
+    )
     train_message = Message(data={TEXT: sentence})
     WhitespaceTokenizer().process(train_message)
 
@@ -254,7 +266,11 @@ def test_count_vector_featurizer_oov_token(sentence, expected):
 def test_count_vector_featurizer_oov_words(sentence, expected):
 
     ftr = CountVectorsFeaturizer(
-        {"OOV_token": "__oov__", "OOV_words": ["oov_word0", "OOV_word1"]}
+        {
+            "OOV_token": "__oov__",
+            "OOV_words": ["oov_word0", "OOV_word1"],
+            "additional_vocabulary_size": {"text": 0},
+        }
     )
     train_message = Message(data={TEXT: sentence})
     WhitespaceTokenizer().process(train_message)
@@ -288,7 +304,7 @@ def test_count_vector_featurizer_oov_words(sentence, expected):
 )
 def test_count_vector_featurizer_using_tokens(tokens, expected):
 
-    ftr = CountVectorsFeaturizer()
+    ftr = CountVectorsFeaturizer({"additional_vocabulary_size": {"text": 0}})
 
     # using empty string instead of real text string to make sure
     # count vector only can come from `tokens` feature.
@@ -326,7 +342,14 @@ def test_count_vector_featurizer_using_tokens(tokens, expected):
     ],
 )
 def test_count_vector_featurizer_char(sentence, expected):
-    ftr = CountVectorsFeaturizer({"min_ngram": 1, "max_ngram": 2, "analyzer": "char"})
+    ftr = CountVectorsFeaturizer(
+        {
+            "min_ngram": 1,
+            "max_ngram": 2,
+            "analyzer": "char",
+            "additional_vocabulary_size": {"text": 0},
+        }
+    )
 
     train_message = Message(data={TEXT: sentence})
     WhitespaceTokenizer().process(train_message)
@@ -347,7 +370,7 @@ def test_count_vector_featurizer_char(sentence, expected):
     assert sen_vec is not None
 
 
-def test_count_vector_featurizer_persist_load(tmp_path):
+def test_count_vector_featurizer_persist_load(tmp_path: Path):
 
     # set non default values to config
     config = {
@@ -360,6 +383,7 @@ def test_count_vector_featurizer_persist_load(tmp_path):
         "max_ngram": 3,
         "max_features": 10,
         "lowercase": False,
+        "additional_vocabulary_size": {"text": 0},
     }
     train_ftr = CountVectorsFeaturizer(config)
 
@@ -391,7 +415,7 @@ def test_count_vector_featurizer_persist_load(tmp_path):
     # load featurizer
     meta = train_ftr.component_config.copy()
     meta.update(file_dict)
-    test_ftr = CountVectorsFeaturizer.load(meta, str(tmp_path))
+    test_ftr = CountVectorsFeaturizer.load(meta, str(tmp_path), finetune_mode=False)
     test_vect_params = {
         attribute: vectorizer.get_params()
         for attribute, vectorizer in test_ftr.vectorizers.items()
@@ -439,7 +463,9 @@ def test_count_vector_featurizer_persist_load(tmp_path):
 
 def test_count_vectors_featurizer_train():
 
-    featurizer = CountVectorsFeaturizer.create({}, RasaNLUModelConfig())
+    featurizer = CountVectorsFeaturizer.create(
+        {"additional_vocabulary_size": {"text": 0, "response": 0}}, RasaNLUModelConfig()
+    )
 
     sentence = "Hey how are you today ?"
     message = Message(data={TEXT: sentence})
@@ -499,7 +525,9 @@ def test_count_vector_featurizer_use_lemma(
     sentence_features: List[List[int]],
     use_lemma: bool,
 ):
-    ftr = CountVectorsFeaturizer({"use_lemma": use_lemma})
+    ftr = CountVectorsFeaturizer(
+        {"use_lemma": use_lemma, "additional_vocabulary_size": {"text": 0}}
+    )
 
     train_message = Message(data={TEXT: sentence})
     train_message.set(SPACY_DOCS[TEXT], spacy_nlp(sentence))
@@ -540,7 +568,12 @@ def test_count_vector_featurizer_action_attribute_featurization(
     action_name_features: np.ndarray,
     response_features: np.ndarray,
 ):
-    ftr = CountVectorsFeaturizer({"token_pattern": r"(?u)\b\w+\b"})
+    ftr = CountVectorsFeaturizer(
+        {
+            "token_pattern": r"(?u)\b\w+\b",
+            "additional_vocabulary_size": {"text": 0, "response": 0, "action_text": 0},
+        }
+    )
     tk = WhitespaceTokenizer()
 
     train_message = Message(data={TEXT: sentence})
@@ -604,7 +637,12 @@ def test_count_vector_featurizer_process_by_attribute(
     action_name_features: np.ndarray,
     response_features: np.ndarray,
 ):
-    ftr = CountVectorsFeaturizer({"token_pattern": r"(?u)\b\w+\b"})
+    ftr = CountVectorsFeaturizer(
+        {
+            "token_pattern": r"(?u)\b\w+\b",
+            "additional_vocabulary_size": {"text": 0, "response": 0, "action_text": 0},
+        }
+    )
     tk = WhitespaceTokenizer()
 
     # add a second example that has some response, so that the vocabulary for
@@ -637,3 +675,195 @@ def test_count_vector_featurizer_process_by_attribute(
 
     assert action_name_seq_vecs.toarray()[0] == action_name_features
     assert action_name_sen_vecs is None
+
+
+@pytest.mark.parametrize(
+    "additional_size, text, real_vocabulary_size, total_vocabulary_size",
+    [(None, "hello my name is John.", 5, 1005), (10, "hello my name is John.", 5, 15)],
+)
+def test_cvf_independent_train_vocabulary_expand(
+    additional_size: Optional[int],
+    text: Text,
+    real_vocabulary_size: int,
+    total_vocabulary_size: int,
+):
+
+    tokenizer = WhitespaceTokenizer()
+    featurizer = CountVectorsFeaturizer(
+        {
+            "additional_vocabulary_size": {
+                TEXT: additional_size,
+                RESPONSE: additional_size,
+                ACTION_TEXT: additional_size,
+            }
+        },
+        finetune_mode=False,
+    )
+
+    train_message = Message(
+        data={
+            TEXT: text,
+            INTENT: "intent_1",
+            RESPONSE: text,
+            ACTION_TEXT: text,
+            ACTION_NAME: "action_1",
+        }
+    )
+    data = TrainingData([train_message])
+
+    tokenizer.train(data)
+    featurizer.train(data)
+
+    for attribute in [TEXT, RESPONSE, ACTION_TEXT]:
+        attribute_vocabulary = featurizer.vectorizers[attribute].vocabulary_
+        assert len(attribute_vocabulary) == total_vocabulary_size
+        assert (
+            featurizer._get_starting_empty_index(attribute_vocabulary)
+            == real_vocabulary_size
+        )
+
+    for attribute in [INTENT, ACTION_NAME]:
+        attribute_vocabulary = featurizer.vectorizers[attribute].vocabulary_
+        assert len(attribute_vocabulary) == 1
+
+
+@pytest.mark.parametrize(
+    "additional_size, text, real_vocabulary_size, total_vocabulary_size",
+    [(None, "hello my name is John.", 7, 1007), (10, "hello my name is John.", 7, 17)],
+)
+def test_cvf_shared_train_vocabulary_expand(
+    additional_size: Optional[int],
+    text: Text,
+    real_vocabulary_size: int,
+    total_vocabulary_size: int,
+):
+
+    tokenizer = WhitespaceTokenizer()
+    featurizer = CountVectorsFeaturizer(
+        {
+            "additional_vocabulary_size": {
+                "text": additional_size,
+                "response": additional_size,
+                "action_text": additional_size,
+            },
+            "use_shared_vocab": True,
+        },
+        finetune_mode=False,
+    )
+
+    train_message = Message(
+        data={
+            TEXT: text,
+            INTENT: "intent_1",
+            RESPONSE: text,
+            ACTION_TEXT: text,
+            ACTION_NAME: "action_1",
+        }
+    )
+    data = TrainingData([train_message])
+
+    tokenizer.train(data)
+    featurizer.train(data)
+
+    shared_vocabulary = featurizer.vectorizers["text"].vocabulary_
+    assert len(shared_vocabulary) == total_vocabulary_size
+    assert (
+        featurizer._get_starting_empty_index(shared_vocabulary) == real_vocabulary_size
+    )
+
+
+@pytest.mark.parametrize(
+    "additional_size, original_train_text, additional_train_text, total_vocabulary_size, remaining_buffer_size",
+    [
+        (10, "hello my name is John.", "I am also new.", 15, 6),
+        (None, "hello my name is John.", "I am also new.", 1005, 996),
+    ],
+)
+def test_cvf_incremental_train_vocabulary(
+    additional_size: Optional[int],
+    original_train_text: Text,
+    additional_train_text: Text,
+    total_vocabulary_size: int,
+    remaining_buffer_size: int,
+    tmp_path: Path,
+):
+
+    tokenizer = WhitespaceTokenizer()
+    original_featurizer = CountVectorsFeaturizer(
+        {"additional_vocabulary_size": {"text": additional_size}}, finetune_mode=False,
+    )
+    train_message = Message(data={"text": original_train_text})
+    data = TrainingData([train_message])
+
+    tokenizer.train(data)
+    original_featurizer.train(data)
+
+    # Check total vocabulary size with buffer slots before finetuning
+    original_vocabulary = original_featurizer.vectorizers["text"].vocabulary_
+    assert len(original_vocabulary) == total_vocabulary_size
+
+    file_dict = original_featurizer.persist("ftr", str(tmp_path))
+
+    # load original_featurizer
+    meta = original_featurizer.component_config.copy()
+    meta.update(file_dict)
+    new_featurizer = CountVectorsFeaturizer.load(
+        meta, str(tmp_path), should_finetune=True
+    )
+
+    # Check total vocabulary size with buffer slots before finetuning
+    assert len(new_featurizer.vectorizers["text"].vocabulary_) == total_vocabulary_size
+
+    additional_train_message = Message(data={"text": additional_train_text})
+    data = TrainingData([train_message, additional_train_message])
+    tokenizer.train(data)
+    new_featurizer.train(data)
+
+    new_vocabulary = new_featurizer.vectorizers["text"].vocabulary_
+
+    # Check total vocabulary size with buffer slots after finetuning
+    assert len(new_vocabulary) == total_vocabulary_size
+
+    # Check remaining buffer slots after finetuning
+    assert (
+        len(new_vocabulary) - new_featurizer._get_starting_empty_index(new_vocabulary)
+        == remaining_buffer_size
+    )
+
+    # Check indices of original vocabulary haven't changed in the new vocabulary
+    for vocab_token, vocab_index in original_vocabulary.items():
+        if not vocab_token.startswith("buf_"):
+            assert vocab_token in new_vocabulary
+            assert new_vocabulary.get(vocab_token) == vocab_index
+
+
+def test_cvf_incremental_train_vocabulary_overflow(tmp_path: Path,):
+    additional_size = 3
+    original_train_text = "hello my name is John."
+    additional_train_text = "I am also new."
+    tokenizer = WhitespaceTokenizer()
+    original_featurizer = CountVectorsFeaturizer(
+        {"additional_vocabulary_size": {"text": additional_size}}, finetune_mode=False,
+    )
+    train_message = Message(data={"text": original_train_text})
+    data = TrainingData([train_message])
+
+    tokenizer.train(data)
+    original_featurizer.train(data)
+
+    file_dict = original_featurizer.persist("ftr", str(tmp_path))
+
+    # load original_featurizer
+    meta = original_featurizer.component_config.copy()
+    meta.update(file_dict)
+    new_featurizer = CountVectorsFeaturizer.load(
+        meta, str(tmp_path), should_finetune=True
+    )
+
+    additional_train_message = Message(data={"text": additional_train_text})
+    data = TrainingData([train_message, additional_train_message])
+    tokenizer.train(data)
+
+    with pytest.warns(UserWarning) as warning:
+        new_featurizer.train(data)
+    assert "New data contains vocabulary of size" in warning[0].message.args[0]
diff --git a/tests/nlu/featurizers/test_regex_featurizer.py b/tests/nlu/featurizers/test_regex_featurizer.py
index 79df4675d144..e31db2e9f35a 100644
--- a/tests/nlu/featurizers/test_regex_featurizer.py
+++ b/tests/nlu/featurizers/test_regex_featurizer.py
@@ -2,6 +2,9 @@
 
 import numpy as np
 import pytest
+from pathlib import Path
+from _pytest.logging import LogCaptureFixture
+import logging
 
 from rasa.shared.nlu.training_data.training_data import TrainingData
 from rasa.shared.nlu.training_data.message import Message
@@ -15,19 +18,20 @@
 
 @pytest.mark.parametrize(
     "sentence, expected_sequence_features, expected_sentence_features,"
-    "labeled_tokens",
+    "labeled_tokens, additional_vocabulary_size",
     [
         (
             "hey how are you today",
             [
-                [0.0, 1.0, 0.0],
-                [0.0, 0.0, 0.0],
-                [0.0, 0.0, 0.0],
-                [0.0, 0.0, 0.0],
-                [0.0, 0.0, 0.0],
+                [0.0, 1.0, 0.0, 0.0, 0.0],
+                [0.0, 0.0, 0.0, 0.0, 0.0],
+                [0.0, 0.0, 0.0, 0.0, 0.0],
+                [0.0, 0.0, 0.0, 0.0, 0.0],
+                [0.0, 0.0, 0.0, 0.0, 0.0],
             ],
-            [0.0, 1.0, 0.0],
+            [0.0, 1.0, 0.0, 0.0, 0.0],
             [0],
+            2,
         ),
         (
             "hey 456 how are you",
@@ -40,18 +44,31 @@
             ],
             [1.0, 1.0, 0.0],
             [1, 0],
+            0,
         ),
         (
             "blah balh random eh",
-            [[0.0, 0.0, 0.0], [0.0, 0.0, 0.0], [0.0, 0.0, 0.0], [0.0, 0.0, 0.0],],
-            [0.0, 0.0, 0.0],
+            [
+                [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
+                [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
+                [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
+                [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
+            ],
+            [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
             [],
+            None,
         ),
         (
             "a 1 digit number",
-            [[0.0, 0.0, 0.0], [1.0, 0.0, 1.0], [0.0, 0.0, 0.0], [0.0, 0.0, 0.0]],
-            [1.0, 0.0, 1.0],
+            [
+                [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
+                [1.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
+                [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
+                [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
+            ],
+            [1.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
             [1, 1],
+            None,
         ),
     ],
 )
@@ -59,7 +76,8 @@ def test_regex_featurizer(
     sentence: Text,
     expected_sequence_features: List[float],
     expected_sentence_features: List[float],
-    labeled_tokens: List[float],
+    labeled_tokens: List[int],
+    additional_vocabulary_size: int,
     spacy_nlp: Any,
 ):
     from rasa.nlu.featurizers.sparse_featurizer.regex_featurizer import RegexFeaturizer
@@ -69,7 +87,10 @@ def test_regex_featurizer(
         {"pattern": "\\bhey*", "name": "hello", "usage": "intent"},
         {"pattern": "[0-1]+", "name": "binary", "usage": "intent"},
     ]
-    ftr = RegexFeaturizer({}, known_patterns=patterns)
+    ftr = RegexFeaturizer(
+        {"number_additional_patterns": additional_vocabulary_size},
+        known_patterns=patterns,
+    )
 
     # adds tokens to the message
     tokenizer = SpacyTokenizer({})
@@ -150,7 +171,9 @@ def test_lookup_tables_without_use_word_boundaries(
         {"name": "cites", "elements": ["北京", "上海", "广州", "深圳", "杭州"],},
         {"name": "dates", "elements": ["昨天", "今天", "明天", "后天"],},
     ]
-    ftr = RegexFeaturizer({"use_word_boundaries": False})
+    ftr = RegexFeaturizer(
+        {"use_word_boundaries": False, "number_additional_patterns": 0}
+    )
     training_data = TrainingData()
     training_data.lookup_tables = lookups
     ftr.train(training_data)
@@ -215,7 +238,7 @@ def test_lookup_tables(
         },
         {"name": "plates", "elements": "data/test/lookup_tables/plates.txt"},
     ]
-    ftr = RegexFeaturizer()
+    ftr = RegexFeaturizer({"number_additional_patterns": 0})
     training_data = TrainingData()
     training_data.lookup_tables = lookups
     ftr.train(training_data)
@@ -265,7 +288,7 @@ def test_regex_featurizer_no_sequence(
         {"pattern": "\\bhey*", "name": "hello", "usage": "intent"},
         {"pattern": "[0-1]+", "name": "binary", "usage": "intent"},
     ]
-    ftr = RegexFeaturizer({}, known_patterns=patterns)
+    ftr = RegexFeaturizer({"number_additional_patterns": 0}, known_patterns=patterns)
 
     # adds tokens to the message
     tokenizer = SpacyTokenizer()
@@ -290,7 +313,9 @@ def test_regex_featurizer_train():
         {"pattern": "[0-1]+", "name": "binary", "usage": "intent"},
     ]
 
-    featurizer = RegexFeaturizer.create({}, RasaNLUModelConfig())
+    featurizer = RegexFeaturizer.create(
+        {"number_additional_patterns": 0}, RasaNLUModelConfig()
+    )
 
     sentence = "hey how are you today 19.12.2019 ?"
     message = Message(data={TEXT: sentence})
@@ -360,7 +385,10 @@ def test_regex_featurizer_case_sensitive(
         {"pattern": "\\bhey*", "name": "hello", "usage": "intent"},
         {"pattern": "[0-1]+", "name": "binary", "usage": "intent"},
     ]
-    ftr = RegexFeaturizer({"case_sensitive": case_sensitive}, known_patterns=patterns)
+    ftr = RegexFeaturizer(
+        {"case_sensitive": case_sensitive, "number_additional_patterns": 0},
+        known_patterns=patterns,
+    )
 
     # adds tokens to the message
     tokenizer = SpacyTokenizer()
@@ -399,7 +427,9 @@ def test_lookup_with_and_without_boundaries(
     use_word_boundaries: bool,
     spacy_nlp: Any,
 ):
-    ftr = RegexFeaturizer({"use_word_boundaries": use_word_boundaries})
+    ftr = RegexFeaturizer(
+        {"use_word_boundaries": use_word_boundaries, "number_additional_patterns": 0}
+    )
     training_data = TrainingData()
 
     # we use lookups because the "use_word_boundaries" flag is only used when
@@ -441,3 +471,172 @@ def test_lookup_with_and_without_boundaries(
         num_matches = sum(token_matches)
         # labeled_tokens should list the token(s) which match a pattern
         assert num_matches == labeled_tokens.count(i)
+
+
+def test_persist_load_for_finetuning(tmp_path: Path):
+    patterns = [
+        {"pattern": "[0-9]+", "name": "number", "usage": "intent"},
+        {"pattern": "\\bhey*", "name": "hello", "usage": "intent"},
+        {"pattern": "[0-1]+", "name": "binary", "usage": "intent"},
+    ]
+
+    featurizer = RegexFeaturizer.create(
+        {"number_additional_patterns": 5}, RasaNLUModelConfig()
+    )
+
+    sentence = "hey how are you today 19.12.2019 ?"
+    message = Message(data={TEXT: sentence})
+    message.set(RESPONSE, sentence)
+    message.set(INTENT, "intent")
+    WhitespaceTokenizer().train(TrainingData([message]))
+
+    featurizer.train(
+        TrainingData([message], regex_features=patterns), RasaNLUModelConfig()
+    )
+
+    persist_value = featurizer.persist("ftr", str(tmp_path))
+
+    # Test all artifacts stored as part of persist
+    assert persist_value["file"] == "ftr"
+    assert (tmp_path / "ftr.patterns.pkl").exists()
+    assert (tmp_path / "ftr.vocabulary_stats.pkl").exists()
+    assert featurizer.vocabulary_stats == {
+        "max_number_patterns": 8,
+        "pattern_slots_filled": 3,
+    }
+
+    loaded_featurizer = RegexFeaturizer.load(
+        meta={"number_additional_patterns": 5, "file": persist_value["file"],},
+        should_finetune=True,
+        model_dir=str(tmp_path),
+    )
+
+    # Test component loaded in finetune mode and also with
+    # same patterns as before and vocabulary statistics
+    assert loaded_featurizer.known_patterns == featurizer.known_patterns
+    assert loaded_featurizer.finetune_mode
+    assert loaded_featurizer.pattern_vocabulary_stats == featurizer.vocabulary_stats
+
+    new_lookups = [{"name": "plates", "elements": "data/test/lookup_tables/plates.txt"}]
+
+    training_data = TrainingData()
+    training_data.lookup_tables = new_lookups
+    loaded_featurizer.train(training_data)
+
+    # Test merging of a new pattern to an already trained component.
+    assert len(loaded_featurizer.known_patterns) == 4
+    assert loaded_featurizer.vocabulary_stats == {
+        "max_number_patterns": 8,
+        "pattern_slots_filled": 4,
+    }
+
+
+def test_incremental_train_featurization(tmp_path: Path):
+    patterns = [
+        {"pattern": "[0-9]+", "name": "number", "usage": "intent"},
+        {"pattern": "\\bhey*", "name": "hello", "usage": "intent"},
+        {"pattern": "[0-1]+", "name": "binary", "usage": "intent"},
+    ]
+
+    featurizer = RegexFeaturizer.create(
+        {"number_additional_patterns": 5}, RasaNLUModelConfig()
+    )
+
+    sentence = "hey how are you today 19.12.2019 ?"
+    message = Message(data={TEXT: sentence})
+    message.set(RESPONSE, sentence)
+    message.set(INTENT, "intent")
+    WhitespaceTokenizer().train(TrainingData([message]))
+
+    featurizer.train(
+        TrainingData([message], regex_features=patterns), RasaNLUModelConfig()
+    )
+
+    # Test featurization of message
+    expected = np.array([0, 1, 0, 0, 0, 0, 0, 0])
+    expected_cls = np.array([1, 1, 1, 0, 0, 0, 0, 0])
+
+    seq_vecs, sen_vec = message.get_sparse_features(TEXT, [])
+    if seq_vecs:
+        seq_vecs = seq_vecs.features
+    if sen_vec:
+        sen_vec = sen_vec.features
+
+    assert (6, 8) == seq_vecs.shape
+    assert (1, 8) == sen_vec.shape
+    assert np.all(seq_vecs.toarray()[0] == expected)
+    assert np.all(sen_vec.toarray()[-1] == expected_cls)
+
+    persist_value = featurizer.persist("ftr", str(tmp_path))
+    loaded_featurizer = RegexFeaturizer.load(
+        meta={"number_additional_patterns": 5, "file": persist_value["file"],},
+        should_finetune=True,
+        model_dir=str(tmp_path),
+    )
+
+    new_patterns = [
+        {"pattern": "\\btoday*", "name": "day", "usage": "intent"},
+        {"pattern": "\\bhey+", "name": "hello", "usage": "intent"},
+    ]
+
+    message = Message(data={TEXT: sentence})
+    message.set(RESPONSE, sentence)
+    message.set(INTENT, "intent")
+    WhitespaceTokenizer().train(TrainingData([message]))
+
+    loaded_featurizer.train(
+        TrainingData([message], regex_features=patterns + new_patterns),
+        RasaNLUModelConfig(),
+    )
+
+    # Test featurization of message, this time for the extra pattern as well.
+    expected_token_1 = np.array([0, 1, 0, 0, 0, 0, 0, 0])
+    expected_token_2 = np.array([0, 0, 0, 1, 0, 0, 0, 0])
+    expected_cls = np.array([1, 1, 1, 1, 0, 0, 0, 0])
+
+    seq_vecs, sen_vec = message.get_sparse_features(TEXT, [])
+    if seq_vecs:
+        seq_vecs = seq_vecs.features
+    if sen_vec:
+        sen_vec = sen_vec.features
+
+    assert (6, 8) == seq_vecs.shape
+    assert (1, 8) == sen_vec.shape
+    assert np.all(seq_vecs.toarray()[0] == expected_token_1)
+    assert np.all(seq_vecs.toarray()[-2] == expected_token_2)
+    assert np.all(sen_vec.toarray()[-1] == expected_cls)
+
+    # we also modified a pattern, check if that is correctly modified
+    pattern_to_check = [
+        pattern
+        for pattern in loaded_featurizer.known_patterns
+        if pattern["name"] == "hello"
+    ]
+    assert pattern_to_check == [new_patterns[1]]
+
+
+def test_vocabulary_overflow_log():
+    patterns = [
+        {"pattern": "[0-9]+", "name": "number", "usage": "intent"},
+        {"pattern": "\\bhey*", "name": "hello", "usage": "intent"},
+        {"pattern": "[0-1]+", "name": "binary", "usage": "intent"},
+    ]
+
+    featurizer = RegexFeaturizer(
+        {"number_additional_patterns": 1},
+        known_patterns=patterns,
+        finetune_mode=True,
+        pattern_vocabulary_stats={"max_number_patterns": 4, "pattern_slots_filled": 3},
+    )
+
+    additional_patterns = [
+        {"pattern": "\\btoday*", "name": "day", "usage": "intent"},
+        {"pattern": "\\bhello+", "name": "greet", "usage": "intent"},
+    ]
+
+    with pytest.warns(UserWarning) as warning:
+        featurizer.train(TrainingData([], regex_features=additional_patterns))
+    assert (
+        "The originally trained model was configured to handle a maximum number of 4 patterns"
+        in warning[0].message.args[0]
+    )
diff --git a/tests/nlu/selectors/test_selectors.py b/tests/nlu/selectors/test_selectors.py
index d25d28c0367e..560bf9f9b67c 100644
--- a/tests/nlu/selectors/test_selectors.py
+++ b/tests/nlu/selectors/test_selectors.py
@@ -1,6 +1,7 @@
 from pathlib import Path
 
 import pytest
+from typing import List, Dict, Text, Any
 
 from rasa.nlu import train
 from rasa.nlu.components import ComponentBuilder
@@ -238,3 +239,46 @@ async def test_train_model_checkpointing(
     """
     all_files = list(best_model_file.rglob("*.*"))
     assert len(all_files) > 4
+
+
+async def _train_persist_load_with_different_settings(
+    pipeline: List[Dict[Text, Any]],
+    component_builder: ComponentBuilder,
+    tmp_path: Path,
+    should_finetune: bool,
+):
+    _config = RasaNLUModelConfig({"pipeline": pipeline, "language": "en"})
+
+    (trainer, trained, persisted_path) = await train(
+        _config,
+        path=str(tmp_path),
+        data="data/examples/rasa/demo-rasa.md",
+        component_builder=component_builder,
+    )
+
+    assert trainer.pipeline
+    assert trained.pipeline
+
+    loaded = Interpreter.load(
+        persisted_path,
+        component_builder,
+        new_config=_config if should_finetune else None,
+    )
+
+    assert loaded.pipeline
+    assert loaded.parse("Rasa is great!") == trained.parse("Rasa is great!")
+
+
+@pytest.mark.skip_on_windows
+async def test_train_persist_load(component_builder: ComponentBuilder, tmpdir: Path):
+    pipeline = [
+        {"name": "WhitespaceTokenizer"},
+        {"name": "CountVectorsFeaturizer"},
+        {"name": "ResponseSelector", EPOCHS: 1},
+    ]
+    await _train_persist_load_with_different_settings(
+        pipeline, component_builder, tmpdir, False
+    )
+    await _train_persist_load_with_different_settings(
+        pipeline, component_builder, tmpdir, True
+    )
diff --git a/tests/shared/nlu/training_data/test_training_data.py b/tests/shared/nlu/training_data/test_training_data.py
index b828a37c7b84..a0acba1460d1 100644
--- a/tests/shared/nlu/training_data/test_training_data.py
+++ b/tests/shared/nlu/training_data/test_training_data.py
@@ -12,10 +12,13 @@
     ENTITY_ATTRIBUTE_VALUE,
     ENTITY_ATTRIBUTE_TYPE,
     ENTITIES,
+    INTENT,
+    ACTION_NAME,
 )
 from rasa.nlu.convert import convert_training_data
 from rasa.nlu.extractors.mitie_entity_extractor import MitieEntityExtractor
 from rasa.nlu.tokenizers.whitespace_tokenizer import WhitespaceTokenizer
+from rasa.shared.nlu.training_data.message import Message
 from rasa.shared.nlu.training_data.training_data import TrainingData
 from rasa.shared.nlu.training_data.loading import guess_format, UNK, load_data
 from rasa.shared.nlu.training_data.util import (
@@ -633,3 +636,25 @@ def test_fingerprint_is_same_when_loading_data_again():
     td1 = training_data_from_paths(files, language="en")
     td2 = training_data_from_paths(files, language="en")
     assert td1.fingerprint() == td2.fingerprint()
+
+
+@pytest.mark.parametrize(
+    "message",
+    [
+        Message({INTENT: "intent2"}),
+        Message({ENTITIES: [{"entity": "entity2"}]}),
+        Message({ENTITIES: [{"entity": "entity1", "group": "new_group"}]}),
+        Message({ENTITIES: [{"entity": "entity1", "role": "new_role"}]}),
+        Message({ACTION_NAME: "action_name2"}),
+    ],
+)
+def test_label_fingerprints(message: Message):
+    training_data1 = TrainingData(
+        [
+            Message({INTENT: "intent1"}),
+            Message({ENTITIES: [{"entity": "entity1"}]}),
+            Message({ACTION_NAME: "action_name1"}),
+        ]
+    )
+    training_data2 = training_data1.merge(TrainingData([message]))
+    assert training_data1.label_fingerprint() != training_data2.label_fingerprint()
diff --git a/tests/core/test_model.py b/tests/test_model.py
similarity index 68%
rename from tests/core/test_model.py
rename to tests/test_model.py
index 20092001e8a8..75e7240e4009 100644
--- a/tests/core/test_model.py
+++ b/tests/test_model.py
@@ -5,10 +5,14 @@
 import shutil
 from pathlib import Path
 from typing import Text, Optional, Any
+from unittest import mock
 from unittest.mock import Mock
 
 import pytest
+from _pytest.monkeypatch import MonkeyPatch
 
+import rasa
+import rasa.constants
 from rasa.shared.importers.importer import TrainingDataImporter
 from rasa.shared.importers.rasa import RasaFileImporter
 from rasa.shared.constants import (
@@ -19,9 +23,11 @@
 )
 from rasa.shared.core.domain import KEY_RESPONSES
 from rasa.shared.core.domain import Domain
+import rasa.shared.utils.io
 from rasa import model
 from rasa.model import (
     FINGERPRINT_CONFIG_KEY,
+    FINGERPRINT_CONFIG_WITHOUT_EPOCHS_KEY,
     FINGERPRINT_DOMAIN_WITHOUT_NLG_KEY,
     FINGERPRINT_NLG_KEY,
     FINGERPRINT_FILE_PATH,
@@ -33,6 +39,7 @@
     FINGERPRINT_CONFIG_NLU_KEY,
     SECTION_CORE,
     SECTION_NLU,
+    can_finetune,
     create_package_rasa,
     get_latest_model,
     get_model,
@@ -102,18 +109,22 @@ def _fingerprint(
     config: Optional[Any] = None,
     config_nlu: Optional[Any] = None,
     config_core: Optional[Any] = None,
+    config_without_epochs: Optional[Any] = None,
     domain: Optional[Any] = None,
     nlg: Optional[Any] = None,
     stories: Optional[Any] = None,
     nlu: Optional[Any] = None,
     rasa_version: Text = "1.0",
-):
+) -> Fingerprint:
     return {
         FINGERPRINT_CONFIG_KEY: config if config is not None else ["test"],
         FINGERPRINT_CONFIG_CORE_KEY: config_core
         if config_core is not None
         else ["test"],
         FINGERPRINT_CONFIG_NLU_KEY: config_nlu if config_nlu is not None else ["test"],
+        FINGERPRINT_CONFIG_WITHOUT_EPOCHS_KEY: config_without_epochs
+        if config_without_epochs
+        else ["test"],
         FINGERPRINT_DOMAIN_WITHOUT_NLG_KEY: domain if domain is not None else ["test"],
         FINGERPRINT_NLG_KEY: nlg if nlg is not None else ["test"],
         FINGERPRINT_TRAINED_AT_KEY: time.time(),
@@ -148,9 +159,10 @@ def test_persist_and_load_fingerprint():
         (_fingerprint(nlg=["other"]), False),
         (_fingerprint(nlu=["test", "other"]), False),
         (_fingerprint(config_nlu=["other"]), False),
+        (_fingerprint(config_without_epochs=["other"]), False),
     ],
 )
-def test_core_fingerprint_changed(fingerprint2, changed):
+def test_core_fingerprint_changed(fingerprint2: Fingerprint, changed: bool):
     fingerprint1 = _fingerprint()
     assert (
         did_section_fingerprint_change(fingerprint1, fingerprint2, SECTION_CORE)
@@ -168,9 +180,10 @@ def test_core_fingerprint_changed(fingerprint2, changed):
         (_fingerprint(nlg=["other"]), False),
         (_fingerprint(config_core=["other"]), False),
         (_fingerprint(stories=["other"]), False),
+        (_fingerprint(config_without_epochs=["other"]), False),
     ],
 )
-def test_nlu_fingerprint_changed(fingerprint2, changed):
+def test_nlu_fingerprint_changed(fingerprint2: Fingerprint, changed: bool):
     fingerprint1 = _fingerprint()
     assert (
         did_section_fingerprint_change(fingerprint1, fingerprint2, SECTION_NLU)
@@ -233,6 +246,124 @@ async def test_fingerprinting_changed_response_text(project: Text):
     assert old_fingerprint[FINGERPRINT_NLG_KEY] != new_fingerprint[FINGERPRINT_NLG_KEY]
 
 
+async def test_fingerprinting_changing_config_epochs(project: Text, tmp_path):
+    config1 = {
+        "language": "en",
+        "pipeline": [
+            {"name": "WhitespaceTokenizer"},
+            {"name": "RegexFeaturizer"},
+            {"name": "LexicalSyntacticFeaturizer"},
+            {"name": "CountVectorsFeaturizer"},
+            {
+                "name": "CountVectorsFeaturizer",
+                "analyzer": "char_wb",
+                "min_ngram": 1,
+                "max_ngram": 4,
+            },
+            {"name": "DIETClassifier", "epochs": 100},
+            {"name": "EntitySynonymMapper"},
+            {"name": "ResponseSelector", "epochs": 100},
+            {
+                "name": "FallbackClassifier",
+                "threshold": 0.3,
+                "ambiguity_threshold": 0.1,
+            },
+        ],
+        "policies": [
+            {"name": "MemoizationPolicy"},
+            {"name": "TEDPolicy", "max_history": 5, "epochs": 100},
+            {"name": "RulePolicy"},
+        ],
+    }
+
+    config1_path = tmp_path / "config1.yml"
+    rasa.shared.utils.io.write_yaml(config1, config1_path, True)
+    importer = TrainingDataImporter.load_from_config(str(config1_path))
+    old_fingerprint = await model_fingerprint(importer)
+
+    config2 = {
+        "language": "en",
+        "pipeline": [
+            {"name": "WhitespaceTokenizer"},
+            {"name": "RegexFeaturizer"},
+            {"name": "LexicalSyntacticFeaturizer"},
+            {"name": "CountVectorsFeaturizer"},
+            {
+                "name": "CountVectorsFeaturizer",
+                "analyzer": "char_wb",
+                "min_ngram": 1,
+                "max_ngram": 4,
+            },
+            {"name": "DIETClassifier", "epochs": 50},
+            {"name": "EntitySynonymMapper"},
+            {"name": "ResponseSelector", "epochs": 50},
+            {
+                "name": "FallbackClassifier",
+                "threshold": 0.3,
+                "ambiguity_threshold": 0.1,
+            },
+        ],
+        "policies": [
+            {"name": "MemoizationPolicy"},
+            {"name": "TEDPolicy", "max_history": 5, "epochs": 50},
+            {"name": "RulePolicy"},
+        ],
+    }
+
+    config2_path = tmp_path / "config2.yml"
+    rasa.shared.utils.io.write_yaml(config2, config2_path, True)
+    importer = TrainingDataImporter.load_from_config(str(config2_path))
+    new_fingerprint = await model_fingerprint(importer)
+
+    assert (
+        old_fingerprint[FINGERPRINT_CONFIG_WITHOUT_EPOCHS_KEY]
+        == new_fingerprint[FINGERPRINT_CONFIG_WITHOUT_EPOCHS_KEY]
+    )
+    assert (
+        old_fingerprint[FINGERPRINT_CONFIG_CORE_KEY]
+        != new_fingerprint[FINGERPRINT_CONFIG_CORE_KEY]
+    )
+    assert (
+        old_fingerprint[FINGERPRINT_CONFIG_NLU_KEY]
+        != new_fingerprint[FINGERPRINT_CONFIG_NLU_KEY]
+    )
+
+    config3 = {
+        "language": "en",
+        "pipeline": [{"name": "WhitespaceTokenizer"},],
+        "policies": [
+            {"name": "MemoizationPolicy"},
+            {"name": "TEDPolicy", "max_history": 5, "epochs": 50},
+            {"name": "RulePolicy"},
+        ],
+    }
+
+    config3_path = tmp_path / "config3.yml"
+    rasa.shared.utils.io.write_yaml(config3, config3_path, True)
+    importer = TrainingDataImporter.load_from_config(str(config3_path))
+    new_fingerprint = await model_fingerprint(importer)
+
+    assert (
+        old_fingerprint[FINGERPRINT_CONFIG_WITHOUT_EPOCHS_KEY]
+        != new_fingerprint[FINGERPRINT_CONFIG_WITHOUT_EPOCHS_KEY]
+    )
+
+
+@pytest.mark.parametrize("empty_key", ["pipeline", "policies"])
+async def test_fingerprinting_config_epochs_empty_pipeline_or_policies(
+    project: Text, tmp_path: Path, empty_key: Text,
+):
+    config = {
+        "language": "en",
+        "pipeline": [{"name": "WhitespaceTokenizer"},],
+        "policies": [{"name": "MemoizationPolicy"},],
+    }
+
+    config[empty_key] = None
+
+    model._get_fingerprint_of_config_without_epochs(config)
+
+
 async def test_fingerprinting_additional_action(project: Text):
     importer = _project_files(project)
 
@@ -418,3 +549,42 @@ async def get_domain() -> Domain:
     actual = Domain.load(tmpdir / DEFAULT_CORE_SUBDIRECTORY_NAME / DEFAULT_DOMAIN_PATH)
 
     assert actual.is_empty()
+
+
+@pytest.mark.parametrize(
+    "min_compatible_version, old_model_version, can_tune",
+    [("2.1.0", "2.1.0", True), ("2.0.0", "2.1.0", True), ("2.1.0", "2.0.0", False),],
+)
+async def test_can_finetune_min_version(
+    project: Text,
+    monkeypatch: MonkeyPatch,
+    old_model_version: Text,
+    min_compatible_version: Text,
+    can_tune: bool,
+):
+    importer = _project_files(project)
+
+    monkeypatch.setattr(
+        rasa.constants, "MINIMUM_COMPATIBLE_VERSION", min_compatible_version
+    )
+    monkeypatch.setattr(rasa, "__version__", old_model_version)
+    old_fingerprint = await model_fingerprint(importer)
+    new_fingerprint = await model_fingerprint(importer)
+
+    with mock.patch("rasa.model.MINIMUM_COMPATIBLE_VERSION", min_compatible_version):
+        assert can_finetune(old_fingerprint, new_fingerprint) == can_tune
+
+
+@pytest.mark.parametrize("empty_key", ["pipeline", "policies"])
+async def test_fingerprinting_config_epochs_empty_pipeline_or_policies(
+    project: Text, tmp_path: Path, empty_key: Text,
+):
+    config = {
+        "language": "en",
+        "pipeline": [{"name": "WhitespaceTokenizer"},],
+        "policies": [{"name": "MemoizationPolicy"},],
+    }
+
+    config[empty_key] = None
+
+    model._get_fingerprint_of_config_without_epochs(config)
diff --git a/tests/test_train.py b/tests/test_train.py
index e22a5b9e0f72..177c1b63b2dc 100644
--- a/tests/test_train.py
+++ b/tests/test_train.py
@@ -1,24 +1,30 @@
-import asyncio
 import sys
 import tempfile
 import os
 from pathlib import Path
-from typing import Text, Dict
-from unittest.mock import Mock
+from typing import Text, Dict, Any
+from unittest.mock import Mock, create_autospec
 
 import pytest
 from _pytest.capture import CaptureFixture
 from _pytest.monkeypatch import MonkeyPatch
 
+from rasa.core.policies.ted_policy import TEDPolicy
 import rasa.model
 import rasa.core
+import rasa.nlu
+from rasa.nlu.classifiers.diet_classifier import DIETClassifier
 import rasa.shared.importers.autoconfig as autoconfig
+import rasa.shared.utils.io
+from rasa.core.agent import Agent
 from rasa.core.interpreter import RasaNLUInterpreter
+from rasa.nlu.model import Interpreter
 
 from rasa.train import train_core, train_nlu, train, dry_run_result
-from tests.conftest import DEFAULT_CONFIG_PATH, DEFAULT_NLU_DATA
+from rasa.utils.tensorflow.constants import EPOCHS
+from tests.conftest import DEFAULT_CONFIG_PATH, DEFAULT_NLU_DATA, AsyncMock
 from tests.core.conftest import DEFAULT_DOMAIN_PATH_WITH_SLOTS, DEFAULT_STORIES_FILE
-from tests.core.test_model import _fingerprint
+from tests.test_model import _fingerprint
 
 
 @pytest.mark.parametrize(
@@ -148,7 +154,7 @@ def test_train_nlu_temp_files(
 
 def test_train_nlu_wrong_format_error_message(
     capsys: CaptureFixture,
-    tmp_path: Text,
+    tmp_path: Path,
     monkeypatch: MonkeyPatch,
     default_stack_config: Text,
     incorrect_nlu_data: Text,
@@ -202,7 +208,7 @@ def test_train_nlu_with_responses_and_domain_no_warns(tmp_path: Path):
 
 def test_train_nlu_no_nlu_file_error_message(
     capsys: CaptureFixture,
-    tmp_path: Text,
+    tmp_path: Path,
     monkeypatch: MonkeyPatch,
     default_stack_config: Text,
 ):
@@ -217,24 +223,21 @@ def test_train_nlu_no_nlu_file_error_message(
     assert "No NLU data given" in captured.out
 
 
-@pytest.mark.timeout(240)  # these can take a longer time than the default timeout
 def test_trained_interpreter_passed_to_core_training(
-    monkeypatch: MonkeyPatch, tmp_path: Path, unpacked_trained_moodbot_path: Text
+    monkeypatch: MonkeyPatch, tmp_path: Path, unpacked_trained_rasa_model: Text
 ):
     # Skip actual NLU training and return trained interpreter path from fixture
-    _train_nlu_with_validated_data = Mock(return_value=unpacked_trained_moodbot_path)
-
     # Patching is bit more complicated as we have a module `train` and function
     # with the same name 😬
     monkeypatch.setattr(
         sys.modules["rasa.train"],
         "_train_nlu_with_validated_data",
-        asyncio.coroutine(_train_nlu_with_validated_data),
+        AsyncMock(return_value=unpacked_trained_rasa_model),
     )
 
     # Mock the actual Core training
-    _train_core = Mock()
-    monkeypatch.setattr(rasa.core, "train", asyncio.coroutine(_train_core))
+    _train_core = AsyncMock()
+    monkeypatch.setattr(rasa.core, "train", _train_core)
 
     train(
         DEFAULT_DOMAIN_PATH_WITH_SLOTS,
@@ -248,9 +251,8 @@ def test_trained_interpreter_passed_to_core_training(
     assert isinstance(kwargs["interpreter"], RasaNLUInterpreter)
 
 
-@pytest.mark.timeout(240)  # these can take a longer time than the default timeout
 def test_interpreter_of_old_model_passed_to_core_training(
-    monkeypatch: MonkeyPatch, tmp_path: Path, trained_moodbot_path: Text
+    monkeypatch: MonkeyPatch, tmp_path: Path, trained_rasa_model: Text
 ):
     # NLU isn't retrained
     monkeypatch.setattr(
@@ -261,12 +263,12 @@ def test_interpreter_of_old_model_passed_to_core_training(
 
     # An old model with an interpreter exists
     monkeypatch.setattr(
-        rasa.model, rasa.model.get_latest_model.__name__, lambda _: trained_moodbot_path
+        rasa.model, rasa.model.get_latest_model.__name__, lambda _: trained_rasa_model
     )
 
     # Mock the actual Core training
-    _train_core = Mock()
-    monkeypatch.setattr(rasa.core, "train", asyncio.coroutine(_train_core))
+    _train_core = AsyncMock()
+    monkeypatch.setattr(rasa.core, "train", _train_core)
 
     train(
         DEFAULT_DOMAIN_PATH_WITH_SLOTS,
@@ -293,7 +295,7 @@ def test_interpreter_from_previous_model_returns_none_for_none():
 
 
 def test_train_core_autoconfig(
-    tmp_path: Text,
+    tmp_path: Path,
     monkeypatch: MonkeyPatch,
     default_domain_path: Text,
     default_stories_file: Text,
@@ -306,11 +308,8 @@ def test_train_core_autoconfig(
     monkeypatch.setattr(autoconfig, "get_configuration", mocked_get_configuration)
 
     # skip actual core training
-    _train_core_with_validated_data = Mock()
     monkeypatch.setattr(
-        sys.modules["rasa.train"],
-        "_train_core_with_validated_data",
-        asyncio.coroutine(_train_core_with_validated_data),
+        sys.modules["rasa.train"], "_train_core_with_validated_data", AsyncMock()
     )
 
     # do training
@@ -327,7 +326,7 @@ def test_train_core_autoconfig(
 
 
 def test_train_nlu_autoconfig(
-    tmp_path: Text,
+    tmp_path: Path,
     monkeypatch: MonkeyPatch,
     default_stack_config: Text,
     default_nlu_data: Text,
@@ -338,12 +337,8 @@ def test_train_nlu_autoconfig(
     mocked_get_configuration = Mock()
     monkeypatch.setattr(autoconfig, "get_configuration", mocked_get_configuration)
 
-    # skip actual NLU training
-    _train_nlu_with_validated_data = Mock()
     monkeypatch.setattr(
-        sys.modules["rasa.train"],
-        "_train_nlu_with_validated_data",
-        asyncio.coroutine(_train_nlu_with_validated_data),
+        sys.modules["rasa.train"], "_train_nlu_with_validated_data", AsyncMock()
     )
 
     # do training
@@ -358,6 +353,495 @@ def test_train_nlu_autoconfig(
     assert args[1] == autoconfig.TrainingType.NLU
 
 
+@pytest.mark.parametrize("use_latest_model", [True, False])
+def test_model_finetuning(
+    tmp_path: Path,
+    monkeypatch: MonkeyPatch,
+    default_domain_path: Text,
+    default_stories_file: Text,
+    default_stack_config: Text,
+    default_nlu_data: Text,
+    trained_rasa_model: Text,
+    use_latest_model: bool,
+):
+    mocked_nlu_training = AsyncMock(return_value="")
+    monkeypatch.setattr(rasa.nlu, rasa.nlu.train.__name__, mocked_nlu_training)
+
+    mocked_core_training = AsyncMock()
+    monkeypatch.setattr(rasa.core, rasa.core.train.__name__, mocked_core_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    if use_latest_model:
+        trained_rasa_model = str(Path(trained_rasa_model).parent)
+
+    train(
+        default_domain_path,
+        default_stack_config,
+        [default_stories_file, default_nlu_data],
+        output=output,
+        force_training=True,
+        model_to_finetune=trained_rasa_model,
+        finetuning_epoch_fraction=0.1,
+    )
+
+    mocked_core_training.assert_called_once()
+    _, kwargs = mocked_core_training.call_args
+    assert isinstance(kwargs["model_to_finetune"], Agent)
+
+    mocked_nlu_training.assert_called_once()
+    _, kwargs = mocked_nlu_training.call_args
+    assert isinstance(kwargs["model_to_finetune"], Interpreter)
+
+
+@pytest.mark.parametrize("use_latest_model", [True, False])
+def test_model_finetuning_core(
+    tmp_path: Path,
+    monkeypatch: MonkeyPatch,
+    trained_moodbot_path: Text,
+    use_latest_model: bool,
+):
+    mocked_core_training = AsyncMock()
+    monkeypatch.setattr(rasa.core, rasa.core.train.__name__, mocked_core_training)
+
+    mock_agent_load = Mock(wraps=Agent.load)
+    monkeypatch.setattr(Agent, "load", mock_agent_load)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    if use_latest_model:
+        trained_moodbot_path = str(Path(trained_moodbot_path).parent)
+
+    # Typically models will be fine-tuned with a smaller number of epochs than training
+    # from scratch.
+    # Fine-tuning will use the number of epochs in the new config.
+    old_config = rasa.shared.utils.io.read_yaml_file("examples/moodbot/config.yml")
+    old_config["policies"][0]["epochs"] = 10
+    new_config_path = tmp_path / "new_config.yml"
+    rasa.shared.utils.io.write_yaml(old_config, new_config_path)
+
+    old_stories = rasa.shared.utils.io.read_yaml_file(
+        "examples/moodbot/data/stories.yml"
+    )
+    old_stories["stories"].append(
+        {"story": "new story", "steps": [{"intent": "greet"}]}
+    )
+    new_stories_path = tmp_path / "new_stories.yml"
+    rasa.shared.utils.io.write_yaml(old_stories, new_stories_path)
+
+    train_core(
+        "examples/moodbot/domain.yml",
+        str(new_config_path),
+        str(new_stories_path),
+        output=output,
+        model_to_finetune=trained_moodbot_path,
+        finetuning_epoch_fraction=0.2,
+    )
+
+    mocked_core_training.assert_called_once()
+    _, kwargs = mocked_core_training.call_args
+    model_to_finetune = kwargs["model_to_finetune"]
+    assert isinstance(model_to_finetune, Agent)
+
+    ted = model_to_finetune.policy_ensemble.policies[0]
+    assert ted.config[EPOCHS] == 2
+    assert ted.finetune_mode
+
+
+def test_model_finetuning_core_with_default_epochs(
+    tmp_path: Path, monkeypatch: MonkeyPatch, trained_moodbot_path: Text,
+):
+    mocked_core_training = AsyncMock()
+    monkeypatch.setattr(rasa.core, rasa.core.train.__name__, mocked_core_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    # Providing a new config with no epochs will mean the default amount are used
+    # and then scaled by `finetuning_epoch_fraction`.
+    old_config = rasa.shared.utils.io.read_yaml_file("examples/moodbot/config.yml")
+    del old_config["policies"][0]["epochs"]
+    new_config_path = tmp_path / "new_config.yml"
+    rasa.shared.utils.io.write_yaml(old_config, new_config_path)
+
+    train_core(
+        "examples/moodbot/domain.yml",
+        str(new_config_path),
+        "examples/moodbot/data/stories.yml",
+        output=output,
+        model_to_finetune=trained_moodbot_path,
+        finetuning_epoch_fraction=2,
+    )
+
+    mocked_core_training.assert_called_once()
+    _, kwargs = mocked_core_training.call_args
+    model_to_finetune = kwargs["model_to_finetune"]
+
+    ted = model_to_finetune.policy_ensemble.policies[0]
+    assert ted.config[EPOCHS] == TEDPolicy.defaults[EPOCHS] * 2
+
+
+def test_model_finetuning_core_new_domain_label(
+    tmp_path: Path, monkeypatch: MonkeyPatch, trained_moodbot_path: Text,
+):
+    mocked_core_training = AsyncMock()
+    monkeypatch.setattr(rasa.core, rasa.core.train.__name__, mocked_core_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    # Simulate addition to training data
+    old_domain = rasa.shared.utils.io.read_yaml_file("examples/moodbot/domain.yml")
+    old_domain["intents"].append("a_new_one")
+    new_domain_path = tmp_path / "new_domain.yml"
+    rasa.shared.utils.io.write_yaml(old_domain, new_domain_path)
+
+    with pytest.raises(SystemExit):
+        train_core(
+            domain=str(new_domain_path),
+            config="examples/moodbot/config.yml",
+            stories="examples/moodbot/data/stories.yml",
+            output=output,
+            model_to_finetune=trained_moodbot_path,
+        )
+
+    mocked_core_training.assert_not_called()
+
+
+def test_model_finetuning_new_domain_label_stops_all_training(
+    tmp_path: Path, monkeypatch: MonkeyPatch, trained_moodbot_path: Text,
+):
+    mocked_core_training = AsyncMock()
+    mocked_nlu_training = AsyncMock()
+    monkeypatch.setattr(rasa.core, rasa.core.train.__name__, mocked_core_training)
+    monkeypatch.setattr(rasa.nlu, rasa.nlu.train.__name__, mocked_nlu_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    old_domain = rasa.shared.utils.io.read_yaml_file("examples/moodbot/domain.yml")
+    old_domain["intents"].append("a_new_one")
+    new_domain_path = tmp_path / "new_domain.yml"
+    rasa.shared.utils.io.write_yaml(old_domain, new_domain_path)
+
+    with pytest.raises(SystemExit):
+        train(
+            domain=str(new_domain_path),
+            config="examples/moodbot/config.yml",
+            training_files=[
+                "examples/moodbot/data/stories.yml",
+                "examples/moodbot/data/nlu.yml",
+            ],
+            output=output,
+            model_to_finetune=trained_moodbot_path,
+        )
+
+    mocked_core_training.assert_not_called()
+    mocked_nlu_training.assert_not_called()
+
+
+@pytest.mark.parametrize("use_latest_model", [True, False])
+def test_model_finetuning_nlu(
+    tmp_path: Path,
+    monkeypatch: MonkeyPatch,
+    trained_nlu_moodbot_path: Text,
+    use_latest_model: bool,
+):
+    mocked_nlu_training = AsyncMock(return_value="")
+    monkeypatch.setattr(rasa.nlu, rasa.nlu.train.__name__, mocked_nlu_training)
+
+    mock_interpreter_create = Mock(wraps=Interpreter.create)
+    monkeypatch.setattr(Interpreter, "create", mock_interpreter_create)
+
+    mock_DIET_load = Mock(wraps=DIETClassifier.load)
+    monkeypatch.setattr(DIETClassifier, "load", mock_DIET_load)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    if use_latest_model:
+        trained_nlu_moodbot_path = str(Path(trained_nlu_moodbot_path).parent)
+
+    # Typically models will be fine-tuned with a smaller number of epochs than training
+    # from scratch.
+    # Fine-tuning will use the number of epochs in the new config.
+    old_config = rasa.shared.utils.io.read_yaml_file("examples/moodbot/config.yml")
+    old_config["pipeline"][-1][EPOCHS] = 10
+    new_config_path = tmp_path / "new_config.yml"
+    rasa.shared.utils.io.write_yaml(old_config, new_config_path)
+
+    old_nlu = rasa.shared.utils.io.read_yaml_file("examples/moodbot/data/nlu.yml")
+    old_nlu["nlu"][-1]["examples"] = "-something else"
+    new_nlu_path = tmp_path / "new_nlu.yml"
+    rasa.shared.utils.io.write_yaml(old_nlu, new_nlu_path)
+
+    train_nlu(
+        str(new_config_path),
+        str(new_nlu_path),
+        domain="examples/moodbot/domain.yml",
+        output=output,
+        model_to_finetune=trained_nlu_moodbot_path,
+        finetuning_epoch_fraction=0.2,
+    )
+
+    assert mock_interpreter_create.call_args[1]["should_finetune"]
+
+    mocked_nlu_training.assert_called_once()
+    _, nlu_train_kwargs = mocked_nlu_training.call_args
+    model_to_finetune = nlu_train_kwargs["model_to_finetune"]
+    assert isinstance(model_to_finetune, Interpreter)
+
+    _, diet_kwargs = mock_DIET_load.call_args
+    assert diet_kwargs["should_finetune"] is True
+
+    new_diet_metadata = model_to_finetune.model_metadata.metadata["pipeline"][-1]
+    assert new_diet_metadata["name"] == "DIETClassifier"
+    assert new_diet_metadata[EPOCHS] == 2
+
+
+def test_model_finetuning_nlu_new_label(
+    tmp_path: Path, monkeypatch: MonkeyPatch, trained_nlu_moodbot_path: Text,
+):
+    mocked_nlu_training = AsyncMock(return_value="")
+    monkeypatch.setattr(rasa.nlu, rasa.nlu.train.__name__, mocked_nlu_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    old_nlu = rasa.shared.utils.io.read_yaml_file("examples/moodbot/data/nlu.yml")
+    old_nlu["nlu"].append({"intent": "a_new_one", "examples": "-blah"})
+    new_nlu_path = tmp_path / "new_nlu.yml"
+    rasa.shared.utils.io.write_yaml(old_nlu, new_nlu_path)
+
+    with pytest.raises(SystemExit):
+        train_nlu(
+            "examples/moodbot/config.yml",
+            str(new_nlu_path),
+            domain="examples/moodbot/domain.yml",
+            output=output,
+            model_to_finetune=trained_nlu_moodbot_path,
+        )
+
+    mocked_nlu_training.assert_not_called()
+
+
+def test_model_finetuning_nlu_new_entity(
+    tmp_path: Path, monkeypatch: MonkeyPatch, trained_nlu_moodbot_path: Text,
+):
+    mocked_nlu_training = AsyncMock(return_value="")
+    monkeypatch.setattr(rasa.nlu, rasa.nlu.train.__name__, mocked_nlu_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    old_nlu = rasa.shared.utils.io.read_yaml_file("examples/moodbot/data/nlu.yml")
+    old_nlu["nlu"][-1]["examples"] = "-[blah](something)"
+    new_nlu_path = tmp_path / "new_nlu.yml"
+    rasa.shared.utils.io.write_yaml(old_nlu, new_nlu_path)
+
+    with pytest.raises(SystemExit):
+        train_nlu(
+            "examples/moodbot/config.yml",
+            str(new_nlu_path),
+            domain="examples/moodbot/domain.yml",
+            output=output,
+            model_to_finetune=trained_nlu_moodbot_path,
+        )
+
+    mocked_nlu_training.assert_not_called()
+
+
+def test_model_finetuning_nlu_new_label_already_in_domain(
+    tmp_path: Path,
+    monkeypatch: MonkeyPatch,
+    trained_rasa_model: Text,
+    default_nlu_data: Text,
+    default_config_path: Text,
+    default_domain_path: Text,
+):
+    mocked_nlu_training = AsyncMock(return_value="")
+    monkeypatch.setattr(rasa.nlu, rasa.nlu.train.__name__, mocked_nlu_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    old_nlu = rasa.shared.utils.io.read_yaml_file(default_nlu_data)
+    # This intent exists in `default_domain_path` but not yet in the nlu data
+    old_nlu["nlu"].append({"intent": "why", "examples": "whyy??"})
+    new_nlu_path = tmp_path / "new_nlu.yml"
+    rasa.shared.utils.io.write_yaml(old_nlu, new_nlu_path)
+
+    with pytest.raises(SystemExit):
+        train_nlu(
+            default_config_path,
+            str(new_nlu_path),
+            domain=default_domain_path,
+            output=output,
+            model_to_finetune=trained_rasa_model,
+        )
+
+    mocked_nlu_training.assert_not_called()
+
+
+def test_model_finetuning_nlu_new_label_to_domain_only(
+    tmp_path: Path, monkeypatch: MonkeyPatch, trained_nlu_moodbot_path: Text,
+):
+    mocked_nlu_training = AsyncMock(return_value="")
+    monkeypatch.setattr(rasa.nlu, rasa.nlu.train.__name__, mocked_nlu_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    old_domain = rasa.shared.utils.io.read_yaml_file("examples/moodbot/domain.yml")
+    old_domain["intents"].append("a_new_one")
+    new_domain_path = tmp_path / "new_domain.yml"
+    rasa.shared.utils.io.write_yaml(old_domain, new_domain_path)
+
+    train_nlu(
+        "examples/moodbot/config.yml",
+        "examples/moodbot/data/nlu.yml",
+        domain=str(new_domain_path),
+        output=output,
+        model_to_finetune=trained_nlu_moodbot_path,
+    )
+
+    mocked_nlu_training.assert_called()
+
+
+def test_model_finetuning_nlu_with_default_epochs(
+    tmp_path: Path, monkeypatch: MonkeyPatch, trained_nlu_moodbot_path: Text,
+):
+    mocked_nlu_training = AsyncMock(return_value="")
+    monkeypatch.setattr(rasa.nlu, rasa.nlu.train.__name__, mocked_nlu_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    # Providing a new config with no epochs will mean the default amount are used
+    # and then scaled by `finetuning_epoch_fraction`.
+    old_config = rasa.shared.utils.io.read_yaml_file("examples/moodbot/config.yml")
+    del old_config["pipeline"][-1][EPOCHS]
+    new_config_path = tmp_path / "new_config.yml"
+    rasa.shared.utils.io.write_yaml(old_config, new_config_path)
+
+    train_nlu(
+        str(new_config_path),
+        "examples/moodbot/data/nlu.yml",
+        output=output,
+        model_to_finetune=trained_nlu_moodbot_path,
+        finetuning_epoch_fraction=0.1,
+    )
+
+    mocked_nlu_training.assert_called_once()
+    _, nlu_train_kwargs = mocked_nlu_training.call_args
+    model_to_finetune = nlu_train_kwargs["model_to_finetune"]
+    new_diet_metadata = model_to_finetune.model_metadata.metadata["pipeline"][-1]
+    assert new_diet_metadata["name"] == "DIETClassifier"
+    assert new_diet_metadata[EPOCHS] == DIETClassifier.defaults[EPOCHS] * 0.1
+
+
+@pytest.mark.parametrize("model_to_fine_tune", ["invalid-path-to-model", "."])
+def test_model_finetuning_with_invalid_model(
+    tmp_path: Path,
+    monkeypatch: MonkeyPatch,
+    default_domain_path: Text,
+    default_stories_file: Text,
+    default_stack_config: Text,
+    default_nlu_data: Text,
+    model_to_fine_tune: Text,
+    capsys: CaptureFixture,
+):
+    mocked_nlu_training = AsyncMock(return_value="")
+    monkeypatch.setattr(rasa.nlu, rasa.nlu.train.__name__, mocked_nlu_training)
+
+    mocked_core_training = AsyncMock()
+    monkeypatch.setattr(rasa.core, rasa.core.train.__name__, mocked_core_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    with pytest.raises(SystemExit):
+        train(
+            default_domain_path,
+            default_stack_config,
+            [default_stories_file, default_nlu_data],
+            output=output,
+            force_training=True,
+            model_to_finetune=model_to_fine_tune,
+            finetuning_epoch_fraction=1,
+        )
+
+    mocked_core_training.assert_not_called()
+    mocked_nlu_training.assert_not_called()
+
+    output = capsys.readouterr().out
+    assert "No NLU model for finetuning found" in output
+
+
+@pytest.mark.parametrize("model_to_fine_tune", ["invalid-path-to-model", "."])
+def test_model_finetuning_with_invalid_model_core(
+    tmp_path: Path,
+    monkeypatch: MonkeyPatch,
+    default_domain_path: Text,
+    default_stories_file: Text,
+    default_stack_config: Text,
+    model_to_fine_tune: Text,
+    capsys: CaptureFixture,
+):
+    mocked_core_training = AsyncMock()
+    monkeypatch.setattr(rasa.core, rasa.core.train.__name__, mocked_core_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    with pytest.raises(SystemExit):
+        train_core(
+            default_domain_path,
+            default_stack_config,
+            default_stories_file,
+            output=output,
+            model_to_finetune=model_to_fine_tune,
+            finetuning_epoch_fraction=1,
+        )
+
+    mocked_core_training.assert_not_called()
+
+    assert "No Core model for finetuning found" in capsys.readouterr().out
+
+
+@pytest.mark.parametrize("model_to_fine_tune", ["invalid-path-to-model", "."])
+def test_model_finetuning_with_invalid_model_nlu(
+    tmp_path: Path,
+    monkeypatch: MonkeyPatch,
+    default_domain_path: Text,
+    default_stack_config: Text,
+    default_nlu_data: Text,
+    model_to_fine_tune: Text,
+    capsys: CaptureFixture,
+):
+    mocked_nlu_training = AsyncMock(return_value="")
+    monkeypatch.setattr(rasa.nlu, rasa.nlu.train.__name__, mocked_nlu_training)
+
+    (tmp_path / "models").mkdir()
+    output = str(tmp_path / "models")
+
+    with pytest.raises(SystemExit):
+        train_nlu(
+            default_stack_config,
+            default_nlu_data,
+            domain=default_domain_path,
+            output=output,
+            model_to_finetune=model_to_fine_tune,
+            finetuning_epoch_fraction=1,
+        )
+
+    mocked_nlu_training.assert_not_called()
+
+    assert "No NLU model for finetuning found" in capsys.readouterr().out
+
+
 @pytest.mark.parametrize(
     "result, code, texts_count",
     [


Code review comments:
diff_hunk: @@ -143,8 +149,20 @@ def _load_OOV_params(self) -> None:
             if self.OOV_words:
                 self.OOV_words = [w.lower() for w in self.OOV_words]
 
+        # Additional vocabulary size to be kept reserved
+        self.additional_vocabulary_size = self.component_config[
+            "additional_vocabulary_size"
+        ]
+        if isinstance(self.additional_vocabulary_size, int):
path: rasa/nlu/featurizers/sparse_featurizer/count_vectors_featurizer.py
text: @user1:
do we allow both `int` and `dict`?

@user1:
I suggest to always have `dict`

@author:
Ahh good that you pointed it out, I had meant to remove this piece of code. Agreed, we should stick to `dict`.

diff_hunk: @@ -341,8 +360,10 @@ def load(
             evaluate_on_num_examples=0,
             batch_strategy=SEQUENCE,
             silent=True,  # don't confuse users with training output
-            loading=True,  # don't use tensorboard while loading
-            eager=True,  # no need to build tf graph, eager is faster here
+            loading=True,  # don't use tensorboard when doing a dummy fit run
+            eager=False
+            if finetune_mode
+            else True,  # load in eager mode only for prediction phase
path: rasa/utils/tensorflow/models.py
text: @user1:
not sure if `black` will complain, but parenthesis make it easier to spot that this is all related to `eager=`
```suggestion
            eager=(False
                if finetune_mode
                else True
            ),  # load in eager mode only for prediction phase
```