Computing Task Aware/ Task Agnostic Matrices + terminology clarification

[francesco-p]

I have troubles understanding the evaluation protocol. Given splitcifar100, I want to train a model on experience 0 and test on all the other experiences, then I want to train on experience 1 (updating the corriuspettive head) and compute the accuracy for each other experiences...and so on... up to the last experience. The thing that I want is to update the correct head at training time, but at evaluation time I need two evaluation methods:

1. test by selecting the right head. This is the task aware evaluation i.e. I have the task label at test time
2. test without selecting the right head [forward to all the heads and take the class with maximum probability among all heads]. This is the task agnostic evaluation i.e. I don't have the task label at test time.

I think I'm not understanding the difference between stream task experience...so far I got this code.

The end goal is to create a task aware matrix and a task agnostic accuracy matrix, where in the diagonal I have the performance of the model on the current experience, in the lower triangle the performance on old tasks and in upper triangle in (never seen) future tasks. Something like this:

1   0   0   0
0.8 1   0   0
0.2 0.5  1  0
0.6 0.4 0.8 1 

So far I have this code, but I struggle to understand the evaluation plugin:

import torch
import timm 
from torch.nn import CrossEntropyLoss
from torch.optim import SGD
import avalanche
from avalanche.models import MultiHeadClassifier
from avalanche.training.strategies import Naive
from avalanche.benchmarks.classic import SplitCIFAR100
from avalanche.logging import InteractiveLogger
from avalanche.training.plugins import EvaluationPlugin
from avalanche.logging import InteractiveLogger, TextLogger, TensorboardLogger
from avalanche.training.plugins import EvaluationPlugin
from avalanche.evaluation import metrics
import torchvision


class MTResnet18(avalanche.models.MultiTaskModule):
    def __init__(self, pretrained=False):
        super().__init__()
        self.resnet = timm.create_model('resnet18', pretrained=pretrained, num_classes=0)
        self.classifier = MultiHeadClassifier(512)

    def forward(self, x, task_labels):
        out = self.resnet(x)
        out = out.view(out.size(0), -1)
        return self.classifier(out, task_labels)


SEED = 0
N_EXPERIENCES = 10
PRETRAINED = False
EPOCHS = 2
MINI_BATCH = 128
DEVICE = 0 

device = torch.device(f"cuda:{DEVICE}"
                          if torch.cuda.is_available() and
                          DEVICE >= 0 else "cpu")

# Scenario
scenario = SplitCIFAR100(n_experiences=N_EXPERIENCES, seed=SEED, return_task_id=True)

# Model
model = MTResnet18(pretrained=PRETRAINED)

# Metrics and Logging
eval_plugin = EvaluationPlugin(
            metrics.accuracy_metrics(epoch=False, experience=True, stream=False),
            loggers=[InteractiveLogger()], benchmark=scenario)

# Strategy
optimizer = SGD(model.parameters(), lr=0.01)
criterion = CrossEntropyLoss()
strategy = Naive(
    model=model, optimizer=optimizer, criterion=criterion,
    train_mb_size=MINI_BATCH, train_epochs=EPOCHS, eval_mb_size=MINI_BATCH, device=device,
    evaluator=eval_plugin)


results = []
for exp in scenario.train_stream:
    print(f"Start of experience: {exp.current_experience}\nCurrent Classes: {exp.classes_in_this_experience}")
    
    # Adds Head
    model.adaptation(exp.dataset)
    
    # Train
    strategy.train(exp)
    
    # Test
    results.append(strategy.eval(scenario.test_stream))
    print(results)

[AndreaCossu]

Hi, thanks for reaching out! As for the concepts related to streams, experience and task, does this short introduction help you? Otherwise, let us know if you have specific doubts about the terminology.

In your code, you are testing the evaluation setup 1. (task aware). So, the performance will be reported for a multi-headed model. The evaluation plugin simply computes the accuracy on each test experience and report the result back to you through InteractiveLogger. Each experience will have a growing task label associated to it, apart from the experience ID (in this case, they will be the same).

If you want to test evaluation setup 2 (task-agnostic), you can set return_task_id=False and use a model without MultiHeadClassifier but with a simple linear classifier instead. You don't have to change the evaluation plugin, which will still report the accuracy for each test experience.

Note also that you don't have to manually call adaptation on your model, the strategy will take care of that for you.

[francesco-p]

Thanks for the reply, the thing is that:
if I don't use MultiHeadClassifier, then it means I know a-priori the total number of classes of the scenario. This is unrealistic, instead, as I face experiences, I want to instantiate a dedicated head and let the gradient to flow only there. Otherwise I know I can use a dynamical head, but I want to preserve the ability to switch between task agnostic / aware, so I need a classifier with different heads. I was trying to implement this framework so I wrote that code. But if I'm right I just need to run twice the code with the return_task_id=False I am trying to do it all at once, with the gradient happening in the correct head.

Yeah I think I'm doing a bit of confusion in the logs about experience and task which I saw from the resource that they are synonyms. Thanks anyway!

[AntonioCarta]

First of all, let's clarify the definitions:

• stream: a list of experiences
• experience: all the information you have available at a certain point in time. In supervised CL this is the current batch of data. Notice that Avalanche tags experience with an ID, but you should not use it during training, otherwise it's like having a task label. It's there to distinguish experiences during the evaluation.
• task: in Avalanche we don't have the notion of task. We have task labels, and they can be different for each sample.

Unfortunately, the MultiHeadClassifier does not support your use case but you can easily modify it to do it. Consider that in Avalanche you may have different task that reuse the same class labels (Task 1 with classes [0, 1], Task 2 with classes [0, 1], ...) so be careful to have different class labels if you want to combine different heads.

Now, an example of what your multi-head may look like:

class MyMHead(MultiHeadClassifier):
    def __init__(...):
          self.task_agnostic_eval = True  # change this flag to switch eval head behavior

    def adaptation(...):
        super.adaptation(...)
        if not self.training:
             # when the model is in eval mode, update your single head by combining all the multi-heads
             self.single_head = nn.Linear(...)
             self.single_head.weight = ....

    def forward(...):
         # two branches here, depending on `self.task_agnostic_eval`.

for exp in scenario.train_stream:
print(f"Start of experience: {exp.current_experience}\nCurrent Classes: {exp.classes_in_this_experience}")

# Adds Head
model.adaptation(exp.dataset)

# Train
strategy.train(exp)
```

you don't need to call model.adaptation because it's already called inside the strategy.