リプレイベース継続学習（リプレイベースケイゾクガクシュウ）

class ExperienceReplay:
    def __init__(self, buffer_size=5000):
        self.buffer = []  # (input, label, task_id)
        self.buffer_size = buffer_size
    
    def update_buffer(self, task_data, strategy="reservoir"):
        """バッファ更新戦略"""
        if strategy == "reservoir":
            # Reservoir Sampling: 各サンプルが等確率でバッファに入る
            for item in task_data:
                if len(self.buffer) < self.buffer_size:
                    self.buffer.append(item)
                else:
                    # ランダムな既存アイテムと置き換え
                    j = random.randint(0, len(self.buffer) - 1)
                    self.buffer[j] = item
        
        elif strategy == "ring_buffer":
            # リングバッファ: 古いサンプルを順次上書き
            self.buffer = (self.buffer + list(task_data))[-self.buffer_size:]
        
        elif strategy == "class_balanced":
            # クラス均等: 各クラス/ドメインから均等にサンプル保持
            per_class = self.buffer_size // len(set(t for _, _, t in task_data))
            for task_id in set(t for _, _, t in task_data):
                task_samples = [(x, y, t) for x, y, t in task_data if t == task_id]
                self.buffer.extend(random.sample(task_samples,
                    min(per_class, len(task_samples))))
    
    def sample_replay(self, batch_size):
        """学習バッチにリプレイデータを追加"""
        if len(self.buffer) >= batch_size:
            return random.sample(self.buffer, batch_size)
        return self.buffer.copy()
    
    def train_with_replay(self, model, new_task_loader, replay_ratio=0.3):
        """新タスクデータとリプレイデータを混合して学習"""
        optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)
        
        for new_batch in new_task_loader:
            # リプレイバッチサイズを計算
            replay_batch_size = int(len(new_batch) * replay_ratio)
            replay_batch = self.sample_replay(replay_batch_size)
            
            # 混合バッチを作成
            mixed_batch = combine_batches(new_batch, replay_batch)
            
            optimizer.zero_grad()
            loss = model.compute_loss(mixed_batch)
            loss.backward()
            optimizer.step()

class GenerativeReplay:
    def __init__(self, generator_model):
        # 過去タスクのデータを生成するモデル
        # LLMではモデル自身が生成器になれる
        self.generator = generator_model
    
    def generate_pseudo_data(self, task_id, num_samples):
        """タスクIDに対応する擬似データを生成"""
        # LLMの場合: プロンプトで過去ドメインのデータを生成
        prompt = f"Generate {num_samples} example texts in the style of {task_id}:\n"
        pseudo_samples = self.generator.generate(
            prompt,
            max_new_tokens=200,
            num_return_sequences=num_samples,
        )
        return pseudo_samples
    
    def train_with_generative_replay(self, model, new_data, past_tasks):
        for batch in new_data:
            # 全過去タスクから擬似データを生成
            pseudo_batches = [
                self.generate_pseudo_data(task, len(batch)//len(past_tasks))
                for task in past_tasks
            ]
            
            # 新データ + 擬似データで学習
            combined = [batch] + pseudo_batches
            loss = sum(model.compute_loss(b) for b in combined)
            loss.backward()

class DarkExperienceReplay:
    """
    ラベルではなく教師モデルのロジット出力をバッファに保存
    よりリッチな「暗黙知識」を保持する
    """
    def __init__(self, buffer_size=5000):
        self.buffer = []  # (input, logits)  # ← ラベルでなくロジット
    
    def store_with_logits(self, inputs, model):
        with torch.no_grad():
            logits = model(inputs)  # 現在のモデルのロジットを保存
        for inp, log in zip(inputs, logits):
            self.buffer.append((inp, log.clone()))
    
    def der_loss(self, model, replay_batch):
        inputs, stored_logits = zip(*replay_batch)
        current_logits = model(torch.stack(inputs))
        # ロジット整合性損失（MSE）
        return F.mse_loss(current_logits, torch.stack(stored_logits))

# LLaMA-3のCode LlamaレシピにインスパイアされたCPTリプレイ
MIX_RATIO = {
    "new_domain": 0.80,      # 新しいドメイン（医療/法律/コード等）
    "pretrain_sample": 0.15, # 元の事前学習データのサンプル（リプレイ効果）
    "instruction": 0.05,     # 指示追従データ（能力維持）
}

def create_mixed_dataset(new_domain_data, pretrain_sample, instruct_data):
    return interleave_datasets(
        [new_domain_data, pretrain_sample, instruct_data],
        probabilities=list(MIX_RATIO.values()),
    )

# 効果:
# - 元の言語能力の保持
# - 一般知識の維持
# - 指示追従能力の保護

# SFT後のCPT時のアライメント保護
ALIGNMENT_REPLAY_RATIO = 0.1

# アライメント（RLHF/DPO）済みモデルへのドメイン適応
def domain_adapt_aligned_model(model, domain_data, alignment_dataset):
    mixed = interleave_datasets(
        [domain_data, alignment_dataset],
        probabilities=[1 - ALIGNMENT_REPLAY_RATIO, ALIGNMENT_REPLAY_RATIO]
    )
    # アライメント能力を保護しながらドメイン適応
    train(model, mixed)

手法              タスク1精度（5タスク後）  全タスク平均精度
Naive SGD（なし）       19.8%              72.3%
EWC                    58.0%              83.4%
ER (バッファ200)        91.5%              93.8%
DER++ (バッファ200)     92.3%              94.5%
Joint Training（上界）  99.1%              99.0%