Siamese Networksの概要とアルゴリズム及び実装例

import torch
import torch.nn as nn
import torch.nn.functional as F
from torchvision import datasets, transforms
from torch.utils.data import Dataset, DataLoader
import random
import numpy as np
from PIL import Image

# 1. データ前処理とペア生成
class SiameseMNIST(Dataset):
    def __init__(self, mnist_dataset):
        self.mnist_dataset = mnist_dataset
        self.targets = mnist_dataset.targets
        self.data = mnist_dataset.data

    def __getitem__(self, index):
        img1, label1 = self.data[index], int(self.targets[index])
        # 同じラベルのペア or 異なるラベルのペアをランダムに選択
        should_get_same_class = random.randint(0, 1)
        while True:
            index2 = random.randint(0, len(self.data) - 1)
            label2 = int(self.targets[index2])
            if should_get_same_class == (label1 == label2):
                break
        img2 = self.data[index2]

        transform = transforms.Compose([transforms.ToPILImage(), transforms.ToTensor()])
        return (transform(img1), transform(img2), torch.tensor([int(label1 == label2)], dtype=torch.float32))

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

# 2. Siamese ネットワーク定義
class SiameseNetwork(nn.Module):
    def __init__(self):
        super(SiameseNetwork, self).__init__()
        self.cnn = nn.Sequential(
            nn.Conv2d(1, 16, 3), nn.ReLU(), nn.MaxPool2d(2),
            nn.Conv2d(16, 32, 3), nn.ReLU(), nn.MaxPool2d(2)
        )
        self.fc = nn.Sequential(
            nn.Linear(32 * 5 * 5, 128),
            nn.ReLU(),
            nn.Linear(128, 64)
        )

    def forward_once(self, x):
        x = self.cnn(x)
        x = x.view(x.size()[0], -1)
        return self.fc(x)

    def forward(self, input1, input2):
        output1 = self.forward_once(input1)
        output2 = self.forward_once(input2)
        return output1, output2

# 3. Contrastive Loss 定義
class ContrastiveLoss(nn.Module):
    def __init__(self, margin=1.0):
        super(ContrastiveLoss, self).__init__()
        self.margin = margin

    def forward(self, output1, output2, label):
        euclidean_distance = F.pairwise_distance(output1, output2)
        loss = torch.mean((1 - label) * torch.pow(euclidean_distance, 2) +
                          (label) * torch.pow(torch.clamp(self.margin - euclidean_distance, min=0.0), 2))
        return loss

# 4. 学習
def train(model, dataloader, loss_fn, optimizer, epochs=5):
    for epoch in range(epochs):
        total_loss = 0
        for img1, img2, label in dataloader:
            img1, img2, label = img1.cuda(), img2.cuda(), label.cuda()
            output1, output2 = model(img1, img2)
            loss = loss_fn(output1, output2, label)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            total_loss += loss.item()
        print(f"Epoch {epoch+1}, Loss: {total_loss:.4f}")

# 5. 実行
if __name__ == "__main__":
    transform = transforms.ToTensor()
    mnist_train = datasets.MNIST(root='./data', train=True, download=True, transform=transform)
    siamese_dataset = SiameseMNIST(mnist_train)
    dataloader = DataLoader(siamese_dataset, shuffle=True, batch_size=64)

    model = SiameseNetwork().cuda()
    loss_fn = ContrastiveLoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

    train(model, dataloader, loss_fn, optimizer, epochs=5)

from sentence_transformers import SentenceTransformer, util

# 1. モデルの読み込み（日本語や多言語にも対応する場合は別モデル可）
model = SentenceTransformer('all-MiniLM-L6-v2')  # 軽量で高速

# 2. 比較対象の文を用意
sentences = [
    "This is a book about deep learning.",
    "This book explains neural networks in detail.",
    "I like to play soccer on weekends.",
]

# 3. 文章をベクトルに変換
embeddings = model.encode(sentences, convert_to_tensor=True)

# 4. 類似度を計算（コサイン類似度）
cosine_scores = util.pytorch_cos_sim(embeddings, embeddings)

# 5. 出力
print("文間の類似度行列:")
for i in range(len(sentences)):
    for j in range(len(sentences)):
        print(f"({i}, {j}): {cosine_scores[i][j]:.4f}")

# ライブラリ
pip install torch torchvision scikit-learn matplotlib

# データ定義
from torch.utils.data import Dataset
import os
from PIL import Image
import random
import torch
from torchvision import transforms

class ISICSiameseDataset(Dataset):
    def __init__(self, image_dir, labels_dict, transform=None):
        self.image_dir = image_dir
        self.labels_dict = labels_dict  # filename -> label
        self.transform = transform or transforms.ToTensor()
        self.image_filenames = list(labels_dict.keys())

    def __getitem__(self, idx):
        img1_name = self.image_filenames[idx]
        label1 = self.labels_dict[img1_name]

        # 同じ/異なるラベル画像をランダムに選択
        should_match = random.randint(0, 1)
        while True:
            img2_name = random.choice(self.image_filenames)
            label2 = self.labels_dict[img2_name]
            if (label1 == label2) == should_match:
                break

        img1 = Image.open(os.path.join(self.image_dir, img1_name)).convert('RGB')
        img2 = Image.open(os.path.join(self.image_dir, img2_name)).convert('RGB')

        return self.transform(img1), self.transform(img2), torch.tensor([int(label1 == label2)], dtype=torch.float32)

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

# モデル定義
import torch.nn as nn
import torchvision.models as models

class SiameseResNet(nn.Module):
    def __init__(self):
        super().__init__()
        base_model = models.resnet18(pretrained=True)
        base_model.fc = nn.Identity()  # Remove classification layer
        self.encoder = base_model
        self.fc = nn.Sequential(
            nn.Linear(512, 256),
            nn.ReLU(),
            nn.Linear(256, 128)
        )

    def forward_once(self, x):
        x = self.encoder(x)
        x = self.fc(x)
        return x

    def forward(self, x1, x2):
        out1 = self.forward_once(x1)
        out2 = self.forward_once(x2)
        return out1, out2

# Contrastive Loss
import torch.nn.functional as F

class ContrastiveLoss(nn.Module):
    def __init__(self, margin=2.0):
        super().__init__()
        self.margin = margin

    def forward(self, out1, out2, label):
        distance = F.pairwise_distance(out1, out2)
        loss = torch.mean((1 - label) * distance.pow(2) +
                          (label) * F.relu(self.margin - distance).pow(2))
        return loss

# 学習ループ
def train(model, dataloader, optimizer, criterion, epochs=5):
    model.train()
    for epoch in range(epochs):
        total_loss = 0
        for x1, x2, label in dataloader:
            x1, x2, label = x1.cuda(), x2.cuda(), label.cuda()
            out1, out2 = model(x1, x2)
            loss = criterion(out1, out2, label)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            total_loss += loss.item()
        print(f"Epoch {epoch+1}: Loss {total_loss:.4f}")

# 類似画像検索の推論
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def find_similar(query_img, dataset, model, top_k=5):
    model.eval()
    with torch.no_grad():
        q_emb = model.forward_once(query_img.unsqueeze(0).cuda()).cpu().numpy()
        embeddings = []
        paths = []

        for img_path in dataset.image_filenames:
            img = Image.open(os.path.join(dataset.image_dir, img_path)).convert('RGB')
            img_tensor = dataset.transform(img).unsqueeze(0).cuda()
            emb = model.forward_once(img_tensor).cpu().numpy()
            embeddings.append(emb)
            paths.append(img_path)

        similarities = cosine_similarity(q_emb, np.vstack(embeddings))[0]
        top_indices = similarities.argsort()[::-1][:top_k]
        return [paths[i] for i in top_indices], [similarities[i] for i in top_indices]

# 結果の可視化（matplotlib）
import matplotlib.pyplot as plt

def show_similar(query_img, similar_imgs, sim_scores, dataset):
    plt.figure(figsize=(15, 3))
    plt.subplot(1, len(similar_imgs)+1, 1)
    plt.imshow(query_img.permute(1, 2, 0))
    plt.title("Query")
    plt.axis('off')
    for i, (img_path, score) in enumerate(zip(similar_imgs, sim_scores)):
        img = Image.open(os.path.join(dataset.image_dir, img_path)).convert('RGB')
        plt.subplot(1, len(similar_imgs)+1, i+2)
        plt.imshow(img)
        plt.title(f"Score: {score:.2f}")
        plt.axis('off')
    plt.show()

# ライブラリインストール
pip install torch transformers scikit-learn

# data
product_data = [
{"title": "Wireless Bluetooth Earphones with Noise Cancellation", "label": "audio"},
{"title": "Over-Ear Noise Cancelling Headphones", "label": "audio"},
{"title": "Smart Fitness Watch with Heart Rate Monitor", "label": "wearable"},
{"title": "Leather Smart Watch Band for Apple Watch", "label": "wearable"},
{"title": "Portable Bluetooth Speaker for Outdoors", "label": "audio"},
{"title": "Cotton Yoga Pants for Women", "label": "clothing"},
]

# siameseコード
import random

def make_pairs(data):
    pairs = []
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            label = 1 if data[i]["label"] == data[j]["label"] else 0
            pairs.append((data[i]["title"], data[j]["title"], label))
    return pairs

from transformers import AutoTokenizer, AutoModel
import torch.nn as nn
import torch

class SiameseBERT(nn.Module):
    def __init__(self, model_name="sentence-transformers/all-MiniLM-L6-v2"):
        super().__init__()
        self.bert = AutoModel.from_pretrained(model_name)
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)

    def encode(self, sentences):
        encoded = self.tokenizer(sentences, padding=True, truncation=True, return_tensors="pt")
        with torch.no_grad():
            model_output = self.bert(**encoded)
        embeddings = model_output.last_hidden_state[:, 0, :]  # CLS token
        return embeddings

from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def recommend(query_text, product_data, model, top_k=3):
    all_titles = [item["title"] for item in product_data]
    embeddings = model.encode(all_titles)
    query_emb = model.encode([query_text])
    sims = cosine_similarity(query_emb, embeddings)[0]
    top_idx = sims.argsort()[::-1][1:top_k+1]
    return [(all_titles[i], sims[i]) for i in top_idx]

model = SiameseBERT()

query = "Compact Bluetooth Noise Cancelling Earbuds"
recommendations = recommend(query, product_data, model)

for title, score in recommendations:
    print(f"{title} (Score: {score:.4f})")