Day 124 — マルチモーダル学習入門 — 数値×テキストのIntermediate Fusion

📚 背景知識（読んでから問題へ）

🔀 3つのFusion方式 — どの段階で複数モダリティを混ぜるか

今日実装するのは②Intermediate Fusion。数値ブランチ・テキストブランチがそれぞれ同じ次元数の埋め込みベクトルを出力し、torch.catで結合してから予測ヘッドに渡す。

🗂️ データスキーマ

🎯 問題

架空のネットショップの商品600件について、数値特徴量（価格・割引率・閲覧数）と商品説明文（5単語のBag of Words）から評価スコア（rating, 1.0〜5.0）を予測します。評価スコアは実際には観測できない「商品の本当の品質」に強く依存しており、その品質の手がかりは閲覧数と説明文の単語選びの両方に、それぞれ不完全な形で滲み出ています。

import numpy as np
import pandas as pd

np.random.seed(42)
N = 600

# 観測できない「商品の本当の品質」（-∞〜+∞、平均0）
true_quality = np.random.normal(0, 1, size=N)

# --- 数値特徴量 ---
price_scaled = np.random.normal(0, 1, size=N)
discount = np.random.uniform(0, 0.5, size=N)
views_noise = np.random.normal(0, 10, size=N)
views = (50 + true_quality * 8 + views_noise).clip(min=0)

# --- テキスト特徴量（5単語のBag of Words） ---
VOCAB = ["高品質", "耐久", "高級", "評判", "コスパ", "安価", "シンプル", "お手頃", "限定", "人気", "軽量", "新作"]
PREMIUM = {"高品質", "耐久", "高級", "評判"}
BUDGET  = {"コスパ", "安価", "シンプル", "お手頃"}

def generate_description(quality, rng):
    probs = []
    for w in VOCAB:
        base = 1.0
        if w in PREMIUM:
            base = 1.0 + max(quality, 0) * 2
        elif w in BUDGET:
            base = 1.0 + max(-quality, 0) * 2
        probs.append(base)
    probs = np.array(probs)
    probs /= probs.sum()
    return list(rng.choice(VOCAB, size=5, replace=True, p=probs))

rng = np.random.default_rng(42)
descriptions = [generate_description(q, rng) for q in true_quality]

# --- 目的変数 ---
rating = 3.0 + 0.9 * true_quality - 0.4 * price_scaled + 0.6 * discount + np.random.normal(0, 0.3, size=N)
rating = np.clip(rating, 1.0, 5.0)

df = pd.DataFrame({
    "price_scaled": price_scaled, "discount": discount, "views": views,
    "description": [" ".join(d) for d in descriptions], "rating": rating,
})

TRAIN_LEN = 500   # 先頭500件をtrain、残り100件をvalとして評価

💡 ヒント

import torch
import torch.nn as nn

class NumericBranch(nn.Module):
    def __init__(self, in_dim=3, hidden=16):
        super().__init__()
        self.net = nn.Sequential(nn.Linear(in_dim, hidden), nn.ReLU())
    def forward(self, x):
        return self.net(x)   # (batch, hidden)

class TextBranch(nn.Module):
    def __init__(self, vocab_size, hidden=16):
        super().__init__()
        self.embed = nn.EmbeddingBag(vocab_size, hidden, mode="mean")
    def forward(self, text_idx):
        return self.embed(text_idx)   # (batch, hidden) 5語の平均埋め込み

class MultimodalModel(nn.Module):
    def __init__(self, vocab_size, hidden=16):
        super().__init__()
        self.numeric_branch = NumericBranch(hidden=hidden)
        self.text_branch = TextBranch(vocab_size, hidden=hidden)
        self.head = nn.Linear(hidden * 2, 1)
    def forward(self, numeric, text_idx):
        h_num = self.numeric_branch(numeric)
        h_txt = self.text_branch(text_idx)
        h = torch.cat([h_num, h_txt], dim=1)   # ← Intermediate Fusion
        return self.head(h).squeeze(-1)

✅ 模範解答

import numpy as np
import pandas as pd
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader

torch.manual_seed(0)

# --- サンプルデータ生成（問題文と同じ） ---
np.random.seed(42)
N = 600
true_quality = np.random.normal(0, 1, size=N)
price_scaled = np.random.normal(0, 1, size=N)
discount = np.random.uniform(0, 0.5, size=N)
views_noise = np.random.normal(0, 10, size=N)
views = (50 + true_quality * 8 + views_noise).clip(min=0)

VOCAB = ["高品質", "耐久", "高級", "評判", "コスパ", "安価", "シンプル", "お手頃", "限定", "人気", "軽量", "新作"]
PREMIUM = {"高品質", "耐久", "高級", "評判"}
BUDGET  = {"コスパ", "安価", "シンプル", "お手頃"}

def generate_description(quality, rng):
    probs = []
    for w in VOCAB:
        base = 1.0
        if w in PREMIUM:
            base = 1.0 + max(quality, 0) * 2
        elif w in BUDGET:
            base = 1.0 + max(-quality, 0) * 2
        probs.append(base)
    probs = np.array(probs)
    probs /= probs.sum()
    return list(rng.choice(VOCAB, size=5, replace=True, p=probs))

rng = np.random.default_rng(42)
descriptions = [generate_description(q, rng) for q in true_quality]
rating = 3.0 + 0.9 * true_quality - 0.4 * price_scaled + 0.6 * discount + np.random.normal(0, 0.3, size=N)
rating = np.clip(rating, 1.0, 5.0)

df = pd.DataFrame({
    "price_scaled": price_scaled, "discount": discount, "views": views,
    "description": [" ".join(d) for d in descriptions], "rating": rating,
})
TRAIN_LEN = 500

# --- ステップ1: 単語→IDの辞書とテキストのID化 ---
word2idx = {w: i for i, w in enumerate(VOCAB)}
text_idx_all = np.array([[word2idx[w] for w in desc.split()] for desc in df["description"]])

# --- ステップ2: 数値特徴量をtrainだけの統計量で標準化 ---
numeric_raw = df[["price_scaled", "discount", "views"]].values
num_mean = numeric_raw[:TRAIN_LEN].mean(axis=0)
num_std = numeric_raw[:TRAIN_LEN].std(axis=0)
numeric_scaled = (numeric_raw - num_mean) / num_std

class MultimodalDataset(Dataset):
    def __init__(self, numeric, text_idx, rating):
        self.numeric = torch.tensor(numeric, dtype=torch.float32)
        self.text_idx = torch.tensor(text_idx, dtype=torch.long)
        self.rating = torch.tensor(rating, dtype=torch.float32)
    def __len__(self):
        return len(self.rating)
    def __getitem__(self, idx):
        return self.numeric[idx], self.text_idx[idx], self.rating[idx]

train_ds = MultimodalDataset(numeric_scaled[:TRAIN_LEN], text_idx_all[:TRAIN_LEN], rating[:TRAIN_LEN])
val_ds   = MultimodalDataset(numeric_scaled[TRAIN_LEN:], text_idx_all[TRAIN_LEN:], rating[TRAIN_LEN:])
train_loader = DataLoader(train_ds, batch_size=32, shuffle=True)
val_loader   = DataLoader(val_ds, batch_size=32, shuffle=False)

# --- ステップ3: 3種類のモデル ---
HIDDEN = 16
VOCAB_SIZE = len(VOCAB)

class NumericBranch(nn.Module):
    def __init__(self, in_dim=3, hidden=HIDDEN):
        super().__init__()
        self.net = nn.Sequential(nn.Linear(in_dim, hidden), nn.ReLU())
    def forward(self, x):
        return self.net(x)

class TextBranch(nn.Module):
    def __init__(self, vocab_size=VOCAB_SIZE, hidden=HIDDEN):
        super().__init__()
        self.embed = nn.EmbeddingBag(vocab_size, hidden, mode="mean")
    def forward(self, text_idx):
        return self.embed(text_idx)

class NumericOnlyModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.branch = NumericBranch()
        self.head = nn.Linear(HIDDEN, 1)
    def forward(self, numeric, text_idx):
        return self.head(self.branch(numeric)).squeeze(-1)

class TextOnlyModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.branch = TextBranch()
        self.head = nn.Linear(HIDDEN, 1)
    def forward(self, numeric, text_idx):
        return self.head(self.branch(text_idx)).squeeze(-1)

class MultimodalModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.numeric_branch = NumericBranch()
        self.text_branch = TextBranch()
        self.head = nn.Linear(HIDDEN * 2, 1)
    def forward(self, numeric, text_idx):
        h_num = self.numeric_branch(numeric)
        h_txt = self.text_branch(text_idx)
        h = torch.cat([h_num, h_txt], dim=1)
        return self.head(h).squeeze(-1)

# --- ステップ4: 学習と評価を共通化 ---
def train_and_eval(model, epochs=100, lr=0.01):
    optimizer = torch.optim.Adam(model.parameters(), lr=lr)
    for epoch in range(epochs):
        model.train()
        for numeric, text_idx, y in train_loader:
            optimizer.zero_grad()
            pred = model(numeric, text_idx)
            loss = F.mse_loss(pred, y)
            loss.backward()
            optimizer.step()
    model.eval()
    preds, trues = [], []
    with torch.no_grad():
        for numeric, text_idx, y in val_loader:
            preds.append(model(numeric, text_idx))
            trues.append(y)
    preds = torch.cat(preds).numpy()
    trues = torch.cat(trues).numpy()
    return np.sqrt(np.mean((preds - trues) ** 2))

torch.manual_seed(0); numeric_rmse = train_and_eval(NumericOnlyModel())
torch.manual_seed(0); text_rmse = train_and_eval(TextOnlyModel())
torch.manual_seed(0); multi_rmse = train_and_eval(MultimodalModel())

print(f"{'NumericOnly':12s} RMSE={numeric_rmse:.3f}")
print(f"{'TextOnly':12s} RMSE={text_rmse:.3f}")
print(f"{'Multimodal':12s} RMSE={multi_rmse:.3f}")
# 出力例（同条件のOLS線形回帰で概算した参考値。学習率やepoch数で多少変動する）:
# NumericOnly  RMSE=0.72（例）
# TextOnly     RMSE=0.78（例）
# Multimodal   RMSE=0.58（例）
# → 数値のみ・テキストのみよりも、両方を組み合わせたMultimodalModelの方が
#   RMSEが小さくなりやすい（=それぞれのモダリティが持つ「品質」についての
#   不完全な手がかりを補い合っている）。

# --- ステップ5: テキスト埋め込みの解釈（PREMIUM語・BUDGET語のクラスタ確認） ---
multi_model = MultimodalModel()
train_and_eval(multi_model)  # 再学習（本来は学習済みモデルを使い回す）
emb_matrix = multi_model.text_branch.embed.weight.detach()  # (vocab_size, HIDDEN)

def avg_cos_sim(words_a, words_b, same_group=False):
    idx_a = [word2idx[w] for w in words_a]
    idx_b = [word2idx[w] for w in words_b]
    sims = []
    for i in idx_a:
        for j in idx_b:
            if same_group and i == j:
                continue
            sims.append(F.cosine_similarity(emb_matrix[i], emb_matrix[j], dim=0).item())
    return np.mean(sims)

premium_list, budget_list = list(PREMIUM), list(BUDGET)
print("PREMIUM内の平均類似度:", avg_cos_sim(premium_list, premium_list, same_group=True))
print("BUDGET内の平均類似度 :", avg_cos_sim(budget_list, budget_list, same_group=True))
print("PREMIUM-BUDGET間の類似度:", avg_cos_sim(premium_list, budget_list))
# 出力例（学習がうまくいっていれば）:
# PREMIUM内の平均類似度: 0.4x（例）
# BUDGET内の平均類似度 : 0.4x（例）
# PREMIUM-BUDGET間の類似度: -0.1x（例、負またはPREMIUM内より明確に低い）
# → 「高品質という言葉が使われている商品は評価も高い」という統計的な
#   関係だけから、モデルが「PREMIUM語同士は近く、BUDGET語とは遠い」
#   というベクトル空間の構造を自動的に学習していることが分かる。

📊 評価指標の可視化（val 100件のRMSE、小さいほど良い）

🪜 Step-by-Step 解説

1nn.EmbeddingBagは「Embedding lookup + 平均」を1つのレイヤーにまとめたもの

self.embed = nn.EmbeddingBag(vocab_size, hidden, mode="mean")

2数値ブランチとテキストブランチを同じhidden次元に揃えるのは公平な比較のため

HIDDEN = 16
class NumericBranch(nn.Module): ...  # 出力 (batch, 16)
class TextBranch(nn.Module): ...      # 出力 (batch, 16)

3torch.cat([h_num, h_txt], dim=1)が今日の主役——Intermediate Fusionの実体

h = torch.cat([h_num, h_txt], dim=1)   # (batch, hidden*2)
return self.head(h).squeeze(-1)

4Multimodalが両方の単一モダリティモデルを上回る理由——独立ノイズの打ち消し合い

🧮 数学・統計の補足（文系向け）

🏆 Kaggleでの実践的な使い方

よく使われるコンペカテゴリ: ☐ 表形式データ（Tabular） / ☑ 自然言語処理（NLP） / ☑ 画像認識（CV） / ☐ 時系列（Time Series）

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: 今日のモデルにmodality dropout（学習時にランダムな確率でテキストブランチの出力をゼロベクトルに置き換える）を実装し、テキストが欠損した商品でもMultimodalModelがある程度の精度を保てるかを検証してみましょう。またtorch.catではなく要素ごとの積（h_num * h_txt）や、Day123で学んだAttentionで2つのブランチの重要度を動的に変える「Gated Fusion」を試すのもおすすめです
• 次回予告: Day 125 — ソロGold戦略。Phase 5最後のテーマとして、これまで学んだGBDT（Phase 4）・ディープラーニング各分野（Phase 5）を、チームを組まずに1人でKaggleのGoldメダル圏内を狙うための現実的な立ち回り・コンペ選定・時間配分の戦略を扱います。ここでPhase 5（20テーマ）が完了し、Day126からはいよいよPhase 6: Grandmasterに突入します

Phase 5 の学習マップ（全20テーマ予定）

Day121-123で時系列ブロックが完了し、今日Day124のマルチモーダル学習でテーマ19が完了した。残るはテーマ20「ソロGold戦略」のみで、Phase 5（20テーマ）が完了する。

📝 自己評価（解いた後に記入）