クロスモーダル埋め込み: AIモダリティの橋渡し

import torch
from transformers import CLIPProcessor, CLIPModel
from PIL import Image

# 事前トレーニング済みCLIPモデルの読み込み
model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

# 入力の準備
image = Image.open("example.jpg")
texts = ["a photo of a cat", "a photo of a dog", "a photo of a bird"]

# 入力の処理と埋め込みの取得
inputs = processor(text=texts, images=image, return_tensors="pt", padding=True)

# クロスモーダル類似度スコアの取得
outputs = model(**inputs)
logits_per_image = outputs.logits_per_image
probs = logits_per_image.softmax(dim=1)

print(f"確率: {probs}")

import open_clip

model, _, preprocess = open_clip.create_model_and_transforms(
    'ViT-L-14', 
    pretrained='laion2b_s32b_b82k'
)
tokenizer = open_clip.get_tokenizer('ViT-L-14')

import numpy as np
from typing import List, Tuple
import faiss
from transformers import CLIPModel, CLIPProcessor

class CrossModalSearchEngine:
    def __init__(self, model_name: str = "openai/clip-vit-base-patch32"):
        self.model = CLIPModel.from_pretrained(model_name)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        self.image_index = None
        self.image_metadata = []
        
    def encode_images(self, images: List) -> np.ndarray:
        """画像を埋め込みに変換"""
        inputs = self.processor(images=images, return_tensors="pt", padding=True)
        with torch.no_grad():
            embeddings = self.model.get_image_features(**inputs)
        return embeddings.cpu().numpy()
    
    def encode_text(self, texts: List[str]) -> np.ndarray:
        """テキストクエリを埋め込みに変換"""
        inputs = self.processor(text=texts, return_tensors="pt", padding=True)
        with torch.no_grad():
            embeddings = self.model.get_text_features(**inputs)
        return embeddings.cpu().numpy()
    
    def build_index(self, image_embeddings: np.ndarray):
        """効率的な類似度検索のためにFAISSインデックスを構築"""
        dimension = image_embeddings.shape[1]
        
        # コサイン類似度のために埋め込みを正規化
        faiss.normalize_L2(image_embeddings)
        
        # 大規模な展開のためにHNSWを使用
        self.image_index = faiss.IndexHNSWFlat(dimension, 32)
        self.image_index.hnsw.efConstruction = 40
        self.image_index.add(image_embeddings)
    
    def search(self, query: str, k: int = 10) -> List[Tuple[int, float]]:
        """テキストクエリで画像を検索"""
        query_embedding = self.encode_text([query])
        faiss.normalize_L2(query_embedding)
        
        distances, indices = self.image_index.search(query_embedding, k)
        return list(zip(indices[0], distances[0]))

# 使用例
engine = CrossModalSearchEngine()

# 画像コレクションからインデックスを構築
image_embeddings = engine.encode_images(image_collection)
engine.build_index(image_embeddings)

# テキストで検索
results = engine.search("sunset over mountains", k=5)

from transformers import CLIPModel, CLIPProcessor, AdamW
import torch.nn as nn

class FineTuneCLIP:
    def __init__(self, model_name: str, num_epochs: int = 10):
        self.model = CLIPModel.from_pretrained(model_name)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        self.num_epochs = num_epochs
        
    def contrastive_loss(self, image_embeddings, text_embeddings, temperature=0.07):
        """InfoNCEコントラスト損失を計算"""
        # 埋め込みを正規化
        image_embeddings = nn.functional.normalize(image_embeddings, dim=1)
        text_embeddings = nn.functional.normalize(text_embeddings, dim=1)
        
        # 類似度マトリクスを計算
        logits = torch.matmul(image_embeddings, text_embeddings.T) / temperature
        
        # ラベルは対角線（一致ペア）
        labels = torch.arange(len(logits), device=logits.device)
        
        # 対称損失
        loss_i = nn.functional.cross_entropy(logits, labels)
        loss_t = nn.functional.cross_entropy(logits.T, labels)
        
        return (loss_i + loss_t) / 2
    
    def train(self, dataloader, learning_rate=5e-6):
        """ドメイン固有のデータで微調整"""
        optimizer = AdamW(self.model.parameters(), lr=learning_rate)
        
        self.model.train()
        for epoch in range(self.num_epochs):
            total_loss = 0
            for batch in dataloader:
                images, texts = batch['images'], batch['texts']
                
                # 入力の処理
                inputs = self.processor(
                    text=texts, 
                    images=images, 
                    return_tensors="pt", 
                    padding=True
                )
                
                # フォワードパス
                outputs = self.model(**inputs, return_loss=True)
                loss = outputs.loss
                
                # バックワードパス
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()
                
                total_loss += loss.item()
            
            avg_loss = total_loss / len(dataloader)
            print(f"Epoch {epoch+1}/{self.num_epochs}, Loss: {avg_loss:.4f}")

import torch
from torch.quantization import quantize_dynamic

# CPU推論用の動的量子化
quantized_model = quantize_dynamic(
    model, 
    {torch.nn.Linear}, 
    dtype=torch.qint8
)

import torch.onnx

dummy_input = processor(text=["sample"], return_tensors="pt")
torch.onnx.export(
    model,
    tuple(dummy_input.values()),
    "clip_model.onnx",
    input_names=['input_ids', 'attention_mask'],
    output_names=['output'],
    dynamic_axes={
        'input_ids': {0: 'batch_size'},
        'attention_mask': {0: 'batch_size'}
    }
)

def batch_encode(items: List, batch_size: int = 32):
    """効率性のためにアイテムをバッチで処理"""
    embeddings = []
    for i in range(0, len(items), batch_size):
        batch = items[i:i+batch_size]
        batch_embeddings = encode_batch(batch)
        embeddings.append(batch_embeddings)
    return np.concatenate(embeddings, axis=0)

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# スキーマの定義
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=512),
    FieldSchema(name="metadata", dtype=DataType.JSON)
]
schema = CollectionSchema(fields, description="画像の埋め込み")

# コレクションの作成
collection = Collection("images", schema)

# インデックスの作成
index_params = {
    "metric_type": "IP",  # 内積（コサイン類似度）
    "index_type": "IVF_FLAT",
    "params": {"nlist": 1024}
}
collection.create_index("embedding", index_params)

def zero_shot_classify(image, candidate_labels: List[str], model, processor):
    """任意のカテゴリに画像を分類"""
    # テキストプロンプトの作成
    text_inputs = [f"a photo of a {label}" for label in candidate_labels]
    
    # 埋め込みの取得
    inputs = processor(
        text=text_inputs, 
        images=image, 
        return_tensors="pt", 
        padding=True
    )
    outputs = model(**inputs)
    
    # 確率の計算
    logits = outputs.logits_per_image
    probs = logits.softmax(dim=1)
    
    # ランク付けされた予測の返却
    sorted_indices = probs.argsort(descending=True)[0]
    return [(candidate_labels[idx], probs[0][idx].item()) for idx in sorted_indices]

# 使用例
labels = ["cat", "dog", "bird", "fish", "horse"]
predictions = zero_shot_classify(image, labels, model, processor)
print(f"トップ予測: {predictions[0]}")

class MultimodalRAG:
    def __init__(self, clip_model, llm_model):
        self.clip = clip_model
        self.llm = llm_model
        self.knowledge_base = []
    
    def add_documents(self, images, texts, metadata):
        """マルチモーダルドキュメントを知識ベースに追加"""
        image_embeds = self.clip.encode_images(images)
        text_embeds = self.clip.encode_text(texts)
        
        # 組み合わせた情報を保存
        for i, (img_emb, txt_emb, meta) in enumerate(
            zip(image_embeds, text_embeds, metadata)
        ):
            self.knowledge_base.append({
                'image_embedding': img_emb,
                'text_embedding': txt_emb,
                'metadata': meta,
                'index': i
            })
    
    def retrieve(self, query: str, k: int = 5):
        """関連するマルチモーダルコンテンツを取得"""
        query_embedding = self.clip.encode_text([query])[0]
        
        # 類似度の計算
        similarities = []
        for doc in self.knowledge_base:
            # モダリティごとの類似度の平均
            img_sim = np.dot(query_embedding, doc['image_embedding'])
            txt_sim = np.dot(query_embedding, doc['text_embedding'])
            combined_sim = (img_sim + txt_sim) / 2
            similarities.append((combined_sim, doc))
        
        # トップkの返却
        similarities.sort(reverse=True)
        return [doc for _, doc in similarities[:k]]
    
    def answer_query(self, query: str):
        """取得されたマルチモーダルコンテキストを使ってクエリに回答"""
        retrieved_docs = self.retrieve(query)
        
        # 取得されたドキュメントからコンテキストの構築
        context = "\n".join([doc['metadata']['text'] for doc in retrieved_docs])
        
        # LLMによる回答の生成
        prompt = f"コンテキスト:\n{context}\n\n質問: {query}\n\n回答:"
        answer = self.llm.generate(prompt)
        
        return answer, retrieved_docs

class ContentModerator:
    def __init__(self, model, processor):
        self.model = model
        self.processor = processor
        
        # 不安全なカテゴリの定義
        self.unsafe_categories = [
            "暴力的な内容",
            "成人向けの内容",
            "憎悪のイメージ",
            "グラフィックな暴力",
            "露骨な内容"
        ]
        
        self.safe_categories = [
            "職場でも安全",
            "ファミリーフレンドリー",
            "教育的な内容"
        ]
    
    def moderate_image(self, image, threshold: float = 0.3):
        """画像に不安全な内容が含まれているかを確認"""
        # すべてのカテゴリを組み合わせる
        all_categories = self.unsafe_categories + self.safe_categories
        text_inputs = [f"image containing {cat}" for cat in all_categories]
        
        # 予測の取得
        inputs = self.processor(
            text=text_inputs, 
            images=image, 
            return_tensors="pt"
        )
        outputs = self.model(**inputs)
        probs = outputs.logits_per_image.softmax(dim=1)[0]
        
        # 不安全なカテゴリの確認
        unsafe_scores = probs[:len(self.unsafe_categories)]
        max_unsafe_score = unsafe_scores.max().item()
        
        return {
            'is_safe': max_unsafe_score < threshold,
            'confidence': 1 - max_unsafe_score,
            'flagged_categories': [
                self.unsafe_categories[i] 
                for i, score in enumerate(unsafe_scores) 
                if score > threshold
            ]
        }

from torchvision import transforms

# 標準的なCLIP前処理
clip_transform = transforms.Compose([
    transforms.Resize(224, interpolation=transforms.InterpolationMode.BICUBIC),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(
        mean=[0.48145466, 0.4578275, 0.40821073],
        std=[0.26862954, 0.26130258, 0.27577711]
    )
])

def assess_embedding_quality(embeddings: np.ndarray):
    """埋め込み品質のメトリクスを計算"""
    # 平均ペアワイズ距離の計算
    distances = np.linalg.norm(
        embeddings[:, None] - embeddings[None, :], 
        axis=2
    )
    avg_distance = distances.mean()
    
    # クラスタリングのチェック（低イントラクラスタ距離）
    from sklearn.cluster import KMeans
    kmeans = KMeans(n_clusters=10)
    labels = kmeans.fit_predict(embeddings)
    
    # シルエットスコアの計算
    from sklearn.metrics import silhouette_score
    score = silhouette_score(embeddings, labels)
    
    return {
        'avg_pairwise_distance': avg_distance,
        'silhouette_score': score
    }

FROM nvidia/cuda:11.8.0-cudnn8-runtime-ubuntu22.04

# Pythonと依存関係のインストール
RUN apt-get update && apt-get install -y python3-pip

WORKDIR /app

# 依存関係のインストール
COPY requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt

# アプリケーションコードのコピー
COPY . .

# APIポートの公開
EXPOSE 8000

# APIサーバーの実行
CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

# api.py - クロスモーダル埋め込み用のFastAPIサーバー
from fastapi import FastAPI, File, UploadFile
from pydantic import BaseModel
import torch
from PIL import Image
import io

app = FastAPI()

# 起動時にモデルをロード
@app.on_event("startup")
async def load_model():
    global model, processor
    model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
    processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
    if torch.cuda.is_available():
        model = model.cuda()

class TextQuery(BaseModel):
    text: str

@app.post("/embed/text")
async def embed_text(query: TextQuery):
    inputs = processor(text=[query.text], return_tensors="pt", padding=True)
    if torch.cuda.is_available():
        inputs = {k: v.cuda() for k, v in inputs.items()}
    
    with torch.no_grad():
        embeddings = model.get_text_features(**inputs)
    
    return {"embedding": embeddings.cpu().numpy().tolist()}

@app.post("/embed/image")
async def embed_image(file: UploadFile = File(...)):
    image = Image.open(io.BytesIO(await file.read()))
    inputs = processor(images=image, return_tensors="pt")
    if torch.cuda.is_available():
        inputs = {k: v.cuda() for k, v in inputs.items()}
    
    with torch.no_grad():
        embeddings = model.get_image_features(**inputs)
    
    return {"embedding": embeddings.cpu().numpy().tolist()}

@app.post("/similarity")
async def compute_similarity(
    text: TextQuery, 
    file: UploadFile = File(...)
):
    # 両方の埋め込みの取得
    image = Image.open(io.BytesIO(await file.read()))
    inputs = processor(text=[text.text], images=image, return_tensors="pt", padding=True)
    
    if torch.cuda.is_available():
        inputs = {k: v.cuda() for k, v in inputs.items()}
    
    outputs = model(**inputs)
    similarity = outputs.logits_per_image[0][0].item()
    
    return {"similarity": similarity}