大規模言語モデルのファインチューニングとRLHF(Reinforcement Learning from Human Feedback)

import tensorflow as tf
from tensorflow.keras.applications import VGG16
from tensorflow.keras.preprocessing.image import ImageDataGenerator
from tensorflow.keras.layers import Dense, Dropout, Flatten
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam

# ファインチューニングするクラス数
num_classes = 10

# 画像のサイズ
img_rows, img_cols = 224, 224

# 学習率
learning_rate = 0.0001

# 事前学習済みのVGG16モデルの読み込み
base_model = VGG16(weights='imagenet', include_top=False, input_shape=(img_rows, img_cols, 3))

# VGG16モデルの最後の層を取得
x = base_model.output
x = Flatten()(x)

# 新しい全結合層を追加
x = Dense(512, activation='relu')(x)
x = Dropout(0.5)(x)

# 新しい全結合層を追加（出力の数はクラス数に対応）
predictions = Dense(num_classes, activation='softmax')(x)

# 新しいモデルを作成（既存のVGG16モデルに新しい層を追加）
model = Model(inputs=base_model.input, outputs=predictions)

# 既存のVGG16の層は再学習しないようにする
for layer in base_model.layers:
    layer.trainable = False

# モデルのコンパイル
model.compile(optimizer=Adam(lr=learning_rate), loss='categorical_crossentropy', metrics=['accuracy'])

# データ拡張の設定
train_datagen = ImageDataGenerator(
    rescale=1./255,
    shear_range=0.2,
    zoom_range=0.2,
    horizontal_flip=True
)

test_datagen = ImageDataGenerator(rescale=1./255)

# 訓練データとテストデータのディレクトリ
train_dir = 'path_to_training_data_directory'
test_dir = 'path_to_test_data_directory'

# 訓練データの生成
train_generator = train_datagen.flow_from_directory(
    train_dir,
    target_size=(img_rows, img_cols),
    batch_size=32,
    class_mode='categorical'
)

# テストデータの生成
test_generator = test_datagen.flow_from_directory(
    test_dir,
    target_size=(img_rows, img_cols),
    batch_size=32,
    class_mode='categorical'
)

# モデルの学習
history = model.fit(
    train_generator,
    steps_per_epoch=len(train_generator),
    epochs=10,
    validation_data=test_generator,
    validation_steps=len(test_generator)
)

import torch
from transformers import BertTokenizer, BertForSequenceClassification
from torch.utils.data import DataLoader
from transformers import AdamW
from transformers import get_linear_schedule_with_warmup
from tqdm import tqdm

# デバイスの設定（GPUが利用可能な場合はGPUを使用）
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# パラメータ
MAX_LEN = 128
BATCH_SIZE = 16
EPOCHS = 3
LEARNING_RATE = 2e-5

# データセットのパス
train_data_path = "path_to_train_dataset"
val_data_path = "path_to_validation_dataset"

# 事前学習済みのBERTモデルとトークナイザーを読み込む
model = BertForSequenceClassification.from_pretrained('bert-base-uncased', num_labels=2)
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')

# モデルをデバイスに移動
model.to(device)

# データセットの読み込みとトークン化
class CustomDataset(torch.utils.data.Dataset):
    def __init__(self, tokenizer, data_path):
        self.tokenizer = tokenizer
        self.data = []
        with open(data_path, 'r') as f:
            for line in f:
                text, label = line.strip().split('\t')
                self.data.append((text, int(label)))

    def __len__(self):
        return len(self.data)

    def __getitem__(self, idx):
        text, label = self.data[idx]
        inputs = self.tokenizer.encode_plus(
            text,
            None,
            add_special_tokens=True,
            max_length=MAX_LEN,
            pad_to_max_length=True,
            return_token_type_ids=True,
            truncation=True
        )
        return {
            'input_ids': torch.tensor(inputs['input_ids'], dtype=torch.long),
            'attention_mask': torch.tensor(inputs['attention_mask'], dtype=torch.long),
            'token_type_ids': torch.tensor(inputs["token_type_ids"], dtype=torch.long),
            'labels': torch.tensor(label, dtype=torch.long)
        }

# データセットを作成
train_dataset = CustomDataset(tokenizer, train_data_path)
val_dataset = CustomDataset(tokenizer, val_data_path)

# DataLoaderを作成
train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=BATCH_SIZE, shuffle=False)

# オプティマイザーとスケジューラーの設定
optimizer = AdamW(model.parameters(), lr=LEARNING_RATE, eps=1e-8)
total_steps = len(train_loader) * EPOCHS
scheduler = get_linear_schedule_with_warmup(optimizer, num_warmup_steps=0, num_training_steps=total_steps)

# ファインチューニングのループ
for epoch in range(EPOCHS):
    model.train()
    train_loss = 0.0
    for batch in tqdm(train_loader, desc=f"Epoch {epoch + 1}/{EPOCHS}"):
        input_ids = batch['input_ids'].to(device)
        attention_mask = batch['attention_mask'].to(device)
        token_type_ids = batch['token_type_ids'].to(device)
        labels = batch['labels'].to(device)

        model.zero_grad()

        outputs = model(input_ids=input_ids,
                        attention_mask=attention_mask,
                        token_type_ids=token_type_ids,
                        labels=labels)

        loss = outputs.loss
        train_loss += loss.item()

        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)

        optimizer.step()
        scheduler.step()

    # バリデーションの評価
    model.eval()
    val_loss = 0.0
    val_corrects = 0
    val_total = 0
    with torch.no_grad():
        for batch in tqdm(val_loader, desc=f"Validating Epoch {epoch + 1}/{EPOCHS}"):
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            token_type_ids = batch['token_type_ids'].to(device)
            labels = batch['labels'].to(device)

            outputs = model(input_ids=input_ids,
                            attention_mask=attention_mask,
                            token_type_ids=token_type_ids,
                            labels=labels)

            loss = outputs.loss
            val_loss += loss.item()

            _, preds = torch.max(outputs.logits, 1)
            val_corrects += torch.sum(preds == labels).item()
            val_total += labels.size(0)

    train_loss /= len(train_loader)
    val_loss /= len(val_loader)
    val_acc = val_corrects / val_total

    print(f"Epoch {epoch + 1}/{EPOCHS}, Train Loss: {train_loss}, Val Loss: {val_loss}, Val Acc: {val_acc}")

import numpy as np

# エージェントの状態数と行動数
num_states = 5
num_actions = 2

# Qテーブルの初期化
Q_table = np.zeros((num_states, num_actions))

# フィードバックに基づいてQ値を更新する関数
def update_Q(state, action, feedback, alpha):
    if feedback > 0:
        Q_table[state, action] += alpha * feedback

# エージェントの学習と行動選択
def agent_action(state):
    # エージェントの行動選択（例：ε-greedy法）
    epsilon = 0.2
    if np.random.rand() < epsilon:
        action = np.random.randint(num_actions)
    else:
        action = np.argmax(Q_table[state])
    return action

# エピソードごとの学習とフィードバックを受け取りながらの更新
def train_agent(num_episodes):
    alpha = 0.1  # 学習率

    for episode in range(num_episodes):
        # エージェントの初期状態設定
        state = np.random.randint(num_states)
        
        # エージェントの行動選択
        action = agent_action(state)

        # 人間からのフィードバックをシミュレート（例：1は正解、-1は不正解）
        feedback = np.random.choice([1, -1])

        # Q値の更新
        update_Q(state, action, feedback, alpha)

        # 学習の進捗を出力
        print(f"Episode {episode + 1}/{num_episodes}, State: {state}, Action: {action}, Feedback: {feedback}")

# エージェントの学習を実行
train_agent(num_episodes=10)