Lifelong Learning

Agents that continuously learn and adapt over their lifetime without catastrophic forgetting.

Lifelong Learning Fundamentals

Lifelong Learning
├── Continual Learning
│   ├── Regularization-based
│   ├── Replay-based
│   ├── Architecture-based
│   └── Parameter Isolation
├── Forward Transfer
│   ├── Knowledge Transfer
│   ├── Skill Transfer
│   └── Representation Transfer
├── Backward Transfer
│   ├── Retroactive Interference
│   ├── Catastrophic Forgetting
│   └── Consolidation
├── Meta-Learning
│   ├── MAML
│   ├── Reptile
│   ├── Meta-SGD
│   └── ANIL
├── Curriculum Learning
│   ├── Difficulty Scheduling
│   ├── Competence-Based
│   └── Self-Paced
└── Open-Ended Learning
    ├── Intrinsic Motivation
    ├── Goal Generation
    ├── Environment Design
    └── Autotelic Agents

Continual Learning Strategies

1. Regularization-Based (EWC, SI, LwF)

class EWC:
    """Elastic Weight Consolidation"""
    def __init__(self, model: nn.Module, lambda_: float = 1000):
        self.model = model
        self.lambda_ = lambda_
        self.fisher = {}
        self.params = {}

    def consolidate(self, dataloader: DataLoader):
        """Compute Fisher Information after task completion"""
        self.model.eval()
        fisher = {n: torch.zeros_like(p) for n, p in self.model.named_parameters()}

        for batch in dataloader:
            self.model.zero_grad()
            loss = self.model(batch)
            loss.backward()

            for n, p in self.model.named_parameters():
                if p.grad is not None:
                    fisher[n] += p.grad.data ** 2

        self.fisher = {n: f / len(dataloader) for n, f in fisher.items()}
        self.params = {n: p.clone().detach() for n, p in self.model.named_parameters()}

    def penalty(self) -> torch.Tensor:
        """EWC penalty: sum_i (lambda/2) * F_i * (theta_i - theta*_i)^2"""
        penalty = 0
        for n, p in self.model.named_parameters():
            if n in self.fisher:
                penalty += (self.fisher[n] * (p - self.params[n]) ** 2).sum()
        return self.lambda_ * penalty

class SynapticIntelligence:
    """Synaptic Intelligence (SI) - Online importance estimation"""
    def __init__(self, model: nn.Module, xi: float = 0.1):
        self.model = model
        self.xi = xi
        self.omega = {n: torch.zeros_like(p) for n, p in model.named_parameters()}
        self.prev_params = {n: p.clone().detach() for n, p in model.named_parameters()}
        self.grad_accum = {n: torch.zeros_like(p) for n, p in model.named_parameters()}

    def update_importance(self):
        """Update importance weights after each parameter update"""
        for n, p in self.model.named_parameters():
            if p.grad is not None:
                # Accumulate gradient * parameter change
                delta = p.data - self.prev_params[n]
                self.grad_accum[n] += p.grad * delta
                self.prev_params[n] = p.clone().detach()

    def update_omega(self):
        """Consolidate importance weights after task"""
        for n, p in self.model.named_parameters():
            delta = p.data - self.prev_params[n]
            # Avoid division by zero
            omega_add = self.grad_accum[n] / (delta ** 2 + 1e-8)
            self.omega[n] += omega_add
            self.grad_accum[n].zero_()
            self.prev_params[n] = p.clone().detach()

    def penalty(self) -> torch.Tensor:
        penalty = 0
        for n, p in self.model.named_parameters():
            if n in self.omega:
                penalty += (self.omega[n] * (p - self.prev_params[n]) ** 2).sum()
        return self.xi * penalty

Learning without Forgetting (LwF)

class LearningWithoutForgetting:
    def __init__(self, model: nn.Module, teacher_model: nn.Module, 
                 lambda_: float = 1.0, temperature: float = 2.0):
        self.model = model
        self.teacher = teacher_model
        self.lambda_ = lambda_
        self.temperature = temperature

    def loss(self, inputs: torch.Tensor, targets: torch.Tensor) -> torch.Tensor:
        # Task loss
        outputs = self.model(inputs)
        task_loss = F.cross_entropy(outputs, targets)

        # Distillation loss (preserve old knowledge)
        with torch.no_grad():
            teacher_outputs = self.teacher(inputs)

        student_logits = self.model(inputs) / self.temperature
        teacher_logits = teacher_outputs / self.temperature

        distill_loss = F.kl_div(
            F.log_softmax(student_logits, dim=1),
            F.softmax(teacher_logits, dim=1),
            reduction='batchmean'
        ) * (self.temperature ** 2)

        return task_loss + self.lambda_ * distill_loss

Gradient Episodic Memory (GEM)

class GradientEpisodicMemory:
    def __init__(self, model: nn.Module, memory_size: int = 1000):
        self.model = model
        self.memory = EpisodicMemory(memory_size)
        self.optimizer = Adam(model.parameters())

    def train_step(self, task_id: int, batch: Batch):
        # 1. Compute gradient on current task
        loss = self.model.loss(batch)
        grad = torch.autograd.grad(loss, self.model.parameters())

        # 2. Project gradient to not interfere with previous tasks
        if task_id > 0:
            mem_grads = self.memory.get_gradients(task_id)
            grad = self.project_gradient(grad, mem_grads)

        # 3. Update
        self.apply_gradients(grad)

        # 3. Store gradient in memory
        self.memory.store_gradient(task_id, loss, self.model.parameters())

    def project_gradient(self, grad, mem_grads):
        """Project gradient to not increase loss on previous tasks"""
        for g, mg in zip(grad, mem_grads):
            if (g * mg).sum() < 0:  # Negative dot product = interference
                g = g - (g @ mg) / (mg @ mg) * mg
        return grad

Replay-Based Methods

Experience Replay

class ExperienceReplay:
    def __init__(self, capacity: int = 10000):
        self.buffer = ReservoirBuffer(capacity)

    def add(self, experience: Experience):
        self.buffer.add(experience)

    def sample(self, batch_size: int) -> list[Experience]:
        return self.buffer.sample(batch_size)

class ReservoirBuffer:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.buffer = []
        self.n_seen = 0

    def add(self, item):
        self.n_seen += 1
        if len(self.buffer) < self.capacity:
            self.buffer.append(item)
        else:
            # Reservoir sampling
            idx = random.randint(0, self.n_seen - 1)
            if idx < self.capacity:
                self.buffer[idx] = item

class ExperienceReplayAgent:
    def __init__(self, model, replay_buffer: ExperienceReplay):
        self.model = model
        self.replay = replay_buffer

    async def train_step(self, current_batch: Batch):
        # 1. Train on current batch
        loss = self.model.loss(current_batch)
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

        # 2. Replay from memory
        if len(self.replay.buffer) > self.batch_size:
            mem_batch = self.replay.sample(self.batch_size)
            mem_loss = self.model.loss(mem_batch)
            self.optimizer.zero_grad()
            mem_loss.backward()
            self.optimizer.step()

Generative Replay (Deep Generative Replay)

class GenerativeReplay:
    def __init__(self, generator: Generator, solver: Solver):
        self.generator = generator
        self.solver = solver

    async def train_task(self, task_id: int, data: Dataset):
        # 1. Train generator on current task
        await self.train_generator(task_id, data)

        # 2. Generate pseudo-data for previous tasks
        for prev_task in range(task_id):
            pseudo_data = self.generator.generate(prev_task, num_samples=1000)
            await self.train_on_replay(prev_task, pseudo_data)

        # 3. Train solver on current task
        await self.train_solver(task_id, data)

    def sample_replay(self, task_id: int, n_samples: int) -> Batch:
        """Generate pseudo-data for previous task"""
        return self.generator.generate(task_id, n_samples)

Architecture-Based Methods

Progressive Neural Networks

class ProgressiveNet(nn.Module):
    def __init__(self, input_dim: int, hidden_dims: list[int], output_dim: int):
        super().__init__()
        self.columns = nn.ModuleList()

        # First column
        self.columns.append(self.make_column(input_dim, hidden_dims, output_dim))

        # Lateral connections
        self.lateral_connections = nn.ModuleList()

    def add_column(self, input_dim: int, hidden_dims: list[int], output_dim: int):
        """Add new column for new task"""
        new_col = self.make_column(input_dim, hidden_dims, output_dim)
        self.columns.append(new_col)

        # Add lateral connections from all previous columns
        lateral = nn.ModuleList()
        for col in self.columns[:-1]:
            lateral.append(nn.Linear(col.output_dim, self.columns[-1].input_dim))
        self.lateral_connections.append(lateral)

    def forward(self, x: torch.Tensor, task_id: int) -> torch.Tensor:
        # Forward through all columns up to task_id
        outputs = []
        for i, col in enumerate(self.columns[:task_id+1]):
            if i == 0:
                out = col(x)
            else:
                # Add lateral connections from all previous columns
                lateral_input = sum(
                    lat(out) for lat, out in zip(self.lateral_connections[i-1], outputs)
                )
                out = col(x + lateral_input)
            outputs.append(out)

        return outputs[task_id]

PackNet (Parameter Packing)

class PackNet:
    def __init__(self, model: nn.Module, prune_ratio: float = 0.5):
        self.model = model
        self.prune_ratio = prune_ratio
        self.masks = {}
        self.task_masks = {}

    def train_task(self, task_id: int, train_loader: DataLoader):
        # 1. Train on new task (with previous masks frozen)
        self.freeze_previous_masks()
        self.train(model, train_loader)

        # 2. Prune unimportant weights
        mask = self.prune(model, self.prune_ratio)
        self.masks[task_id] = mask

        # 3. Retrain on remaining weights
        self.freeze_mask(mask)
        self.retrain(model, train_loader)

        # 4. Store task mask
        self.task_masks[task_id] = mask.copy()

    def freeze_previous_masks(self):
        """Freeze weights allocated to previous tasks"""
        for name, param in self.model.named_parameters():
            for task_id, mask in self.task_masks.items():
                param.grad[mask] = 0

    def forward(self, x: torch.Tensor, task_id: int) -> torch.Tensor:
        # Apply task-specific mask
        with torch.no_grad():
            for name, param in self.model.named_parameters():
                if task_id in self.task_masks:
                    param.data = param.data * self.task_masks[task_id][name]

        return self.model(x)

Hard Attention to the Task (HAT)

class HAT:
    def __init__(self, model: nn.Module, embedding_dim: int = 100):
        self.model = model
        self.task_embeddings = nn.Embedding(num_tasks, embedding_dim)
        self.gate = nn.Sigmoid()

    def forward(self, x: torch.Tensor, task_id: int) -> torch.Tensor:
        # Get task embedding
        e = self.task_embeddings(torch.tensor(task_id))

        # Generate masks for each layer
        masks = self.generate_masks(e)

        # Apply masks during forward pass
        return self.model(x, masks)

    def generate_masks(self, embedding: torch.Tensor) -> dict:
        masks = {}
        for name, module in self.model.named_modules():
            if isinstance(module, (nn.Linear, nn.Conv2d)):
                # Generate mask from embedding
                mask_logits = self.mask_generator[name](embedding)
                masks[name] = self.gate(mask_logits)
        return masks

    def regularization(self, current_task: int) -> torch.Tensor:
        """Prevent forgetting by encouraging similar masks for similar tasks"""
        reg = 0
        for i in range(current_task):
            for name in self.task_masks[current_task]:
                reg += F.mse_loss(
                    self.task_masks[current_task][name],
                    self.task_masks[i][name]
                )
        return reg

Meta-Learning

MAML (Model-Agnostic Meta-Learning)

class MAML:
    def __init__(self, model: nn.Module, meta_lr: float = 1e-3, inner_lr: float = 0.01):
        self.model = model
        self.meta_lr = meta_lr
        self.inner_lr = inner_lr
        self.meta_optimizer = Adam(model.parameters(), lr=meta_lr)

    def inner_update(self, model: nn.Module, batch: Batch, lr: float) -> nn.Module:
        """Single gradient step on task"""
        loss = self.compute_loss(model, batch)
        grads = torch.autograd.grad(loss, model.parameters(), create_graph=True)

        # Create updated model
        updated_model = copy.deepcopy(model)
        for param, grad in zip(updated_model.parameters(), grads):
            param.data -= self.inner_lr * grad
        return updated_model

    def meta_update(self, tasks: list[Task]):
        meta_loss = 0

        for task in tasks:
            # Split into support/query
            support, query = task.split()

            # Inner loop: adapt to task
            adapted_model = self.model
            for _ in range(self.inner_steps):
                adapted_model = self.inner_update(adapted_model, task.support)

            # Compute meta-loss on query set
            query_loss = self.compute_loss(adapted_model, task.query)
            meta_loss += query_loss

        # Meta-update
        self.meta_optimizer.zero_grad()
        meta_loss.backward()
        self.meta_optimizer.step()

Reptile (First-Order MAML)

class Reptile:
    def __init__(self, model: nn.Module, inner_lr: float = 0.01, meta_lr: float = 1e-3):
        self.model = model
        self.inner_lr = inner_lr
        self.meta_lr = meta_lr

    def meta_update(self, tasks: list[Task]):
        # Store original parameters
        original_params = {n: p.clone() for n, p in self.model.named_parameters()}

        for task in tasks:
            # Train on task
            for _ in range(self.inner_steps):
                loss = self.compute_loss(task)
                self.optimizer.zero_grad()
                loss.backward()
                self.optimizer.step()

        # Move towards task-specific parameters
        for name, param in self.model.named_parameters():
            param.data = original_params[name] + self.meta_lr * (param.data - original_params[name])

Meta-SGD

class MetaSGD:
    def __init__(self, model: nn.Module):
        self.model = model
        # Learn per-parameter learning rates
        self.lr_params = {n: nn.Parameter(torch.ones_like(p) * 0.01) 
                         for n, p in model.named_parameters()}
        self.optimizer = Adam(list(model.parameters()) + list(self.lr_params.values()))

    def inner_update(self, loss: torch.Tensor):
        grads = torch.autograd.grad(loss, self.model.parameters(), create_graph=True)

        for (name, param), grad in zip(self.model.named_parameters(), grads):
            # Use learned per-parameter learning rate
            lr = self.lr_params[name]
            param.data -= lr * grad

ANIL (Almost No Inner Loop)

class ANIL:
    def __init__(self, model: nn.Module, inner_lr: float = 0.01):
        self.model = model
        # Split into body (shared) and head (task-specific)
        self.body = model.body
        self.head = model.head
        self.inner_lr = inner_lr

    def inner_update(self, task: Task):
        # Only adapt head, freeze body
        for param in self.body.parameters():
            param.requires_grad = False
        for param in self.head.parameters():
            param.requires_grad = True

        # Standard gradient steps on head only
        for _ in range(self.inner_steps):
            loss = self.compute_loss(task)
            self.optimizer.zero_grad()
            loss.backward()
            self.optimizer.step()

Curriculum Learning

Difficulty Scheduling

class CurriculumScheduler:
    def __init__(self, difficulty_estimator: DifficultyEstimator):
        self.estimator = difficulty_estimator
        self.schedule = []

    def generate_curriculum(self, tasks: list[Task], max_difficulty: float = 1.0) -> list[Task]:
        # Sort by difficulty
        tasks_with_difficulty = [(t, self.estimator.estimate(t)) for t in tasks]
        tasks_with_difficulty.sort(key=lambda x: x[1])

        # Schedule: start easy, gradually increase
        curriculum = []
        current_difficulty = 0.1
        for task, diff in tasks_with_difficulty:
            if diff <= current_difficulty:
                curriculum.append(task)
            else:
                # Add intermediate tasks or wait
                curriculum.append(task)
                current_difficulty = min(diff, max_difficulty)

        return curriculum

    def adaptive_schedule(self, performance_history: list[float]) -> float:
        """Adjust difficulty based on performance"""
        recent_perf = statistics.mean(performance_history[-10:])

        if recent_perf > 0.9:
            return min(current_difficulty * 1.2, 1.0)  # Increase difficulty
        elif recent_perf < 0.6:
            return max(current_difficulty * 0.8, 0.1)  # Decrease difficulty
        return current_difficulty

Competence-Based Curriculum

class CompetenceBasedCurriculum:
    def __init__(self, competence_estimator: CompetenceEstimator):
        self.estimator = competence_estimator

    async def select_next_task(self, agent: Agent, task_pool: list[Task]) -> Task:
        """Select task at boundary of competence"""
        competencies = await asyncio.gather(*[
            self.estimator.estimate(agent, task) for task in task_pool
        ])

        # Select task at boundary (competence ~ 0.5-0.7)
        candidates = [
            (task, comp) for task, comp in zip(task_pool, competencies)
            if 0.3 <= comp <= 0.7
        ]

        if candidates:
            return max(candidates, key=lambda x: x[1])[0]
        else:
            # Fallback: easiest unmastered task
            unmastered = [t for t, c in zip(task_pool, competencies) if c < 0.3]
            return min(unmastered, key=lambda t: self.estimator.estimate_difficulty(t))

Open-Ended Learning

Intrinsic Motivation

class IntrinsicMotivation:
    def __init__(self):
        self.curiosity = CuriosityModule()
        self.competence = CompetenceProgress()
        self.empowerment = Empowerment()

    def intrinsic_reward(self, state: State, action: Action, next_state: State) -> float:
        """Compute intrinsic reward"""
        curiosity_reward = self.curiosity.reward(state, next_state)
        competence_reward = self.competence.progress(state, action, next_state)
        empowerment_reward = self.empowerment.reward(state, action)

        return (
            0.5 * curiosity_reward + 
            0.3 * competence_reward + 
            0.2 * empowerment_reward
        )

class CuriosityModule:
    def __init__(self, encoder: Encoder, forward_model: ForwardModel, inverse_model: InverseModel):
        self.encoder = encoder
        self.forward = forward_model
        self.inverse = inverse_model

    def reward(self, state: State, next_state: State) -> float:
        # Prediction error as curiosity
        encoded_s = self.encoder(state)
        encoded_next = self.encoder(next_state)

        pred_next = self.forward(encoded_s, action)
        curiosity = F.mse_loss(pred_next, encoded_next).item()

        return curiosity

class CompetenceProgress:
    def __init__(self):
        self.success_history = defaultdict(list)

    def progress(self, state: State, action: Action, next_state: State) -> float:
        # Competence = success rate in similar situations
        key = self.state_action_key(state, action)
        history = self.success_history[key]

        if len(history) < 10:
            return 0.5  # Uncertain

        recent_success = statistics.mean(history[-10:])
        overall_success = statistics.mean(history)

        # Progress = improvement in competence
        return recent_success - overall_success

Goal Generation

class GoalGenerator:
    def __init__(self, world_model: WorldModel):
        self.world_model = world_model

    async def generate_goals(self, state: State, num_goals: int = 5) -> list[Goal]:
        # Sample diverse goals from world model
        goals = []
        for _ in range(num_goals):
            # Sample from latent space
            z = torch.randn(1, self.latent_dim)
            goal_state = self.world_model.decode(z)

            # Filter: reachable, novel, meaningful
            if self.is_reachable(state, goal_state) and self.is_novel(goal_state):
                goals.append(Goal(state=goal_state))

        return goals

    def is_novel(self, state: State) -> bool:
        # Check against visited states
        embedding = self.encoder(state)
        distances = [cosine_similarity(embedding, v) for v in self.visited_embeddings]
        return max(distances) < self.novelty_threshold

Autotelic Agents

class AutotelicAgent:
    def __init__(self):
        self.goal_generator = GoalGenerator()
        self.planner = Planner()
        self.motivation = IntrinsicMotivation()
        self.skill_library = SkillLibrary()

    async def step(self, state: State) -> Action:
        # 1. Generate goals
        goals = await self.goal_generator.generate_goals(state)

        # 2. Select goal based on motivation
        goal = await self.select_goal(goals, state)

        # 3. Plan to achieve goal
        plan = await self.planner.plan(state, goal)

        # 3. Execute first action
        action = plan.actions[0]

        # 4. Execute and learn
        next_state = await self.execute(action)
        reward = self.motivation.reward(state, action, next_state)

        # Update skills
        self.update_skills(state, action, next_state, reward)

        return action

    def update_skills(self, state, action, next_state, reward):
        # Extract skill from successful trajectory
        if reward > self.skill_threshold:
            skill = self.extract_skill(state, action, next_state)
            self.skill_library.add(skill)

Evaluation of Lifelong Learning

Metrics

Metrics Implementation

class LifelongMetrics:
    def __init__(self):
        self.task_accuracies = defaultdict(list)

    def record(self, task_id: int, accuracy: float):
        self.task_accuracies[task_id].append(accuracy)

    def compute_metrics(self) -> LifelongMetrics:
        tasks = sorted(self.task_accuracies.keys())

        # Average accuracy
        final_acc = [self.task_accuracies[t][-1] for t in tasks]
        avg_acc = statistics.mean(final_acc)

        # Backward Transfer
        bwt = 0
        for i, task_id in enumerate(tasks):
            if i > 0:
                initial = self.task_accuracies[task_id][0]
                final = self.task_accuracies[task_id][-1]
                bwt += final - initial
        bwt /= max(1, len(tasks) - 1)

        # Forward Transfer
        fwt = 0
        for i, task_id in enumerate(tasks):
            if i > 0:
                # Performance on task i before training on it
                first_attempt = self.task_accuracies[task_id][0]
                random_guess = 1.0 / num_classes
                fwt += first_attempt - random_guess
        fwt /= max(1, len(tasks) - 1)

        # Forgetting
        forgetting = 0
        for i, task_id in enumerate(tasks):
            if i < len(tasks) - 1:
                max_acc = max(self.task_accuracies[task_id])
                final_acc = self.task_accuracies[task_id][-1]
                forgetting += max(0, max_acc - final_acc)
        forgetting /= max(1, len(tasks) - 1)

        return LifelongMetrics(
            avg_accuracy=avg_acc,
            bwt=bwt,
            fwt=fwt,
            forgetting=forgetting
        )

Related Notes

• Self Improving Agents — Self-improvement
• Neurosymbolic Integration — Neurosymbolic lifelong learning
• Formal Verification — Verified lifelong learning
• Agent Computer Interface — ACI for lifelong learning
• Reasoning Models — Reasoning in lifelong learning
• Mixture of Experts — MoE for lifelong learning
• Reasoning Models — Reasoning in lifelong learning
• Agent Benchmarks — Benchmarks
• Evaluation Frameworks — Evaluation frameworks
• Research Directions — Research directions
• Engineering Culture — Learning culture
• Research Agents — Research agents
• Coding Agents — Coding agents learning

Sources: "Continual Learning" surveys (Parisi et al.), "Continual Learning" (De Lange et al.), "EWC" (Kirkpatrick et al.), "Synaptic Intelligence" (Zenke et al.), "LwF" (Li & Hoiem), "GEM" (Lopez-Paz & Ranzato), "Progressive Nets" (Rusu et al.), "PackNet" (Mallya & Lazebnik), "HAT" (Serra et al.), "MAML" (Finn et al.), "Reptile" (Nichol et al.), "Meta-SGD" (Li et al.), "ANIL" (Raghu et al.), "Curriculum Learning" (Bengio et al.), "Self-Paced Learning" (Kumar et al.), "Intrinsic Motivation" (Oudeyer & Kaplan), "Curiosity-driven Exploration" (Pathak et al.), "Empowerment" (Klyubin et al.), "Autotelic Agents" (Colas et al.), "Open-Ended Learning" (Stanley et al.), "POET" (Wang et al.), "Meta-Learning" surveys, "Continual Learning" benchmarks (CL Benchmarks, CLEAR, CORe50, Split MNIST, Permuted MNIST, Split CIFAR).