Architecture Methods › Worked Example: Progressive Network for Sequential Task Learning
Language AI HandbookAll chapters Architecture Methods › Worked Example: Progressive Network for Sequential Task LearningOne section · Sections in this chapter7 / 13 ↓
Reading tools Definitions & inline links
Choose your expertise level to adjust how many terms are explained. Beginners see more tooltips, experts see fewer to maintain reading flow. Hover over underlined terms for instant definitions.
Article links
Make inline references clickable
Let us walk through a concrete implementation of a simplified progressive network to see how the column structure and lateral connections work in practice. We will train on two sequential tasks (two different classification problems) and verify that performance on the first task is perfectly preserved after adding a second column for the second task.
The implementation has three parts. First, we define the ProgressiveColumn class, which represents a single column of the network including its main feed-forward layers and its lateral adapter modules that receive signals from prior columns. Second, we define the ProgressiveNetwork class that manages the list of columns, handles the freezing of prior columns when a new task arrives, and routes inputs to the appropriate column at inference time. Third, we train on two sequential tasks and measure Task 1 accuracy before and after Task 2 training to confirm the structural zero-forgetting guarantee.
import numpy as np
import torch
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
np.random.seed(42)
torch.manual_seed(42)
# Task 1: classification with informative features
X1, y1 = make_classification(
n_samples=800,
n_features=10,
n_informative=5,
n_classes=2,
class_sep=1.5,
random_state=1,
)
# Task 2: different classification problem
X2, y2 = make_classification(
n_samples=800,
n_features=10,
n_informative=5,
n_classes=2,
class_sep=1.2,
random_state=2,
)
scaler1 = StandardScaler()
scaler2 = StandardScaler()
X1_train, X1_test, y1_train, y1_test = train_test_split(
X1, y1, test_size=0.2, random_state=42
)
X2_train, X2_test, y2_train, y2_test = train_test_split(
X2, y2, test_size=0.2, random_state=42
)
X1_train = scaler1.fit_transform(X1_train)
X1_test = scaler1.transform(X1_test)
X2_train = scaler2.fit_transform(X2_train)
X2_test = scaler2.transform(X2_test)
# Convert to tensors
X1_tr = torch.tensor(X1_train, dtype=torch.float32)
y1_tr = torch.tensor(y1_train, dtype=torch.long)
X1_te = torch.tensor(X1_test, dtype=torch.float32)
y1_te = torch.tensor(y1_test, dtype=torch.long)
X2_tr = torch.tensor(X2_train, dtype=torch.float32)
y2_tr = torch.tensor(y2_train, dtype=torch.long)
X2_te = torch.tensor(X2_test, dtype=torch.float32)
y2_te = torch.tensor(y2_test, dtype=torch.long)The two tasks use different random seeds to produce distinct decision boundaries. Task 1 uses a slightly larger class separation (class_sep=1.5) while Task 2 is harder (class_sep=1.2), so we should expect somewhat lower accuracy on Task 2.
import torch.nn as nn
class ProgressiveColumn(nn.Module):
"""A single column in a progressive network.
Receives the input plus lateral connections from prior columns.
"""
def __init__(self, input_dim, hidden_dim, output_dim, n_prior_cols):
super().__init__()
self.hidden_dim = hidden_dim
self.n_prior_cols = n_prior_cols
# Main weights for this column
self.fc1 = nn.Linear(input_dim, hidden_dim)
self.fc2 = nn.Linear(hidden_dim, output_dim)
# Lateral adapters: one per prior column, for the hidden layer
self.laterals = nn.ModuleList(
[
nn.Linear(hidden_dim, hidden_dim, bias=False)
for _ in range(n_prior_cols)
]
)
# Initialize lateral weights to zero: column starts independent
for lat in self.laterals:
nn.init.zeros_(lat.weight)
self.relu = nn.ReLU()
def forward(self, x, prior_h1_list):
"""Forward pass with optional lateral contributions."""
h1 = self.fc1(x)
# Add lateral signals from prior columns at layer 1
for i, lat in enumerate(self.laterals):
h1 = h1 + lat(prior_h1_list[i])
h1 = self.relu(h1)
out = self.fc2(h1)
return out, h1
class ProgressiveNetwork(nn.Module):
"""Progressive network maintaining a list of columns."""
def __init__(self, input_dim, hidden_dim, output_dim):
super().__init__()
self.input_dim = input_dim
self.hidden_dim = hidden_dim
self.output_dim = output_dim
self.columns = nn.ModuleList()
def add_column(self):
"""Add a new column for a new task, freezing all existing columns."""
n_prior = len(self.columns)
new_col = ProgressiveColumn(
self.input_dim, self.hidden_dim, self.output_dim, n_prior
)
# Freeze all prior columns
for col in self.columns:
for p in col.parameters():
p.requires_grad_(False)
self.columns.append(new_col)
def forward(self, x, task_id):
"""Run input through columns 0..task_id, return task_id column output."""
prior_h1_list = []
for col_idx in range(task_id + 1):
col = self.columns[col_idx]
out, h1 = col(x, prior_h1_list[:col_idx])
if col_idx < task_id:
prior_h1_list.append(h1.detach())
else:
prior_h1_list.append(h1)
return outThe ProgressiveColumn class holds both the main weights (fc1, fc2) and the lateral adapters. The laterals are initialized to zero, so the column begins completely independent of prior columns and learns to use them only if it finds them useful during training. The add_column method freezes all existing columns before appending the new one, making sure the freeze happens before any optimization step targets the new task.
def train_task(model, task_id, X_tr, y_tr, epochs=30, lr=1e-3):
"""Train the column for task_id on the given data."""
optimizer = torch.optim.Adam(
filter(lambda p: p.requires_grad, model.parameters()), lr=lr
)
criterion = nn.CrossEntropyLoss()
for epoch in range(epochs):
model.train()
optimizer.zero_grad()
logits = model(X_tr, task_id)
loss = criterion(logits, y_tr)
loss.backward()
optimizer.step()
def evaluate(model, task_id, X_te, y_te):
"""Evaluate accuracy on a task."""
model.eval()
with torch.no_grad():
logits = model(X_te, task_id)
preds = logits.argmax(dim=1)
acc = (preds == y_te).float().mean().item()
return acc
# Build the progressive network
prog_net = ProgressiveNetwork(input_dim=10, hidden_dim=64, output_dim=2)
# Train on Task 1
prog_net.add_column()
train_task(prog_net, task_id=0, X_tr=X1_tr, y_tr=y1_tr, epochs=50)
acc_t1_before = evaluate(prog_net, task_id=0, X_te=X1_te, y_te=y1_te)
# Add column for Task 2 (this freezes column 0)
prog_net.add_column()
train_task(prog_net, task_id=1, X_tr=X2_tr, y_tr=y2_tr, epochs=50)
acc_t1_after = evaluate(prog_net, task_id=0, X_te=X1_te, y_te=y1_te)
acc_t2 = evaluate(prog_net, task_id=1, X_te=X2_te, y_te=y2_te)Progressive Network Results Task 1 accuracy before Task 2 training: 0.7500 Task 1 accuracy after Task 2 training: 0.7500 Task 2 accuracy: 0.9250 Forgetting on Task 1: 0.000000 (should be exactly 0)
The zero forgetting result is structural, not an artifact of the particular dataset or training process. Because column 0's weights are frozen before column 1 is trained, the computation graph for task 0 inputs is completely unchanged. The column-freezing mechanism provides an exact guarantee rather than just a regularization tendency.
Notice also that Task 1's accuracy before and after Task 2 training should match to many decimal places, not just approximately. Any floating-point discrepancy is purely numerical noise rather than forgetting. This is the defining property of progressive networks and distinguishes them from regularization-based methods where forgetting is reduced but never completely eliminated.