7장: 멀티모달 임베딩과 크로스모달 검색

import torch
from PIL import Image
from transformers import CLIPProcessor, CLIPModel
 
model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14")
 
# 텍스트 임베딩
texts = ["해변의 석양", "도시의 야경", "숲 속의 오솔길"]
text_inputs = processor(text=texts, return_tensors="pt", padding=True)
 
with torch.no_grad():
    text_features = model.get_text_features(**text_inputs)
    text_features = text_features / text_features.norm(dim=-1, keepdim=True)
 
# 이미지 임베딩
image = Image.open("sunset_beach.jpg")
image_inputs = processor(images=image, return_tensors="pt")
 
with torch.no_grad():
    image_features = model.get_image_features(**image_inputs)
    image_features = image_features / image_features.norm(dim=-1, keepdim=True)
 
# 유사도 계산
similarities = (text_features @ image_features.T).squeeze()
for text, sim in zip(texts, similarities):
    print(f"{text}: {sim.item():.4f}")

import cohere
import base64
 
co = cohere.Client(api_key="...")
 
# 텍스트와 이미지를 같은 공간에 임베딩
response = co.embed(
    model="embed-v4.0",
    input_type="search_document",
    texts=["해변의 석양 사진"],
    images=[base64.standard_b64encode(open("sunset.jpg", "rb").read()).decode()],
)
 
text_embedding = response.embeddings.float_[0]
image_embedding = response.embeddings.float_[1]

from dataclasses import dataclass
from typing import Literal
import numpy as np
 
@dataclass
class MultimodalDocument:
    id: str
    modality: Literal["text", "image", "audio", "video"]
    content_ref: str  # 원본 데이터 참조 (파일 경로 또는 텍스트)
    embedding: np.ndarray
    metadata: dict
 
class MultimodalVectorStore:
    def __init__(self, dimension: int = 768):
        self.documents: list[MultimodalDocument] = []
        self.dimension = dimension
 
    def add(self, doc: MultimodalDocument):
        assert doc.embedding.shape[0] == self.dimension
        self.documents.append(doc)
 
    def search(
        self,
        query_embedding: np.ndarray,
        top_k: int = 10,
        modality_filter: str | None = None,
    ) -> list[tuple[MultimodalDocument, float]]:
        """크로스모달 검색"""
        results = []
 
        for doc in self.documents:
            if modality_filter and doc.modality != modality_filter:
                continue
 
            similarity = np.dot(query_embedding, doc.embedding) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(doc.embedding)
            )
            results.append((doc, float(similarity)))
 
        results.sort(key=lambda x: x[1], reverse=True)
        return results[:top_k]
 
    def text_to_image_search(
        self, text_embedding: np.ndarray, top_k: int = 10
    ) -> list[tuple[MultimodalDocument, float]]:
        """텍스트로 이미지 검색"""
        return self.search(text_embedding, top_k, modality_filter="image")
 
    def image_to_text_search(
        self, image_embedding: np.ndarray, top_k: int = 10
    ) -> list[tuple[MultimodalDocument, float]]:
        """이미지로 텍스트 검색"""
        return self.search(image_embedding, top_k, modality_filter="text")

from pinecone import Pinecone
 
pc = Pinecone(api_key="...")
 
# 멀티모달 인덱스 생성
index = pc.Index("multimodal-search")
 
# 이미지 인덱싱
index.upsert(
    vectors=[
        {
            "id": f"img_{i}",
            "values": image_embedding.tolist(),
            "metadata": {
                "modality": "image",
                "file_path": f"/images/{filename}",
                "description": auto_caption,
                "tags": detected_tags,
            },
        }
        for i, (filename, image_embedding, auto_caption, detected_tags)
        in enumerate(processed_images)
    ]
)
 
# 텍스트로 이미지 검색
text_query = "팀 회의 중인 사람들"
query_embedding = encode_text(text_query)
 
results = index.query(
    vector=query_embedding.tolist(),
    top_k=10,
    filter={"modality": {"$eq": "image"}},
    include_metadata=True,
)

async def index_image_with_caption(
    image_path: str,
    vlm_client,
    clip_model,
    vector_store,
):
    """이미지를 캡셔닝하고 멀티모달 인덱싱"""
    # 1. VLM으로 상세 캡션 생성
    caption = await generate_caption(vlm_client, image_path)
 
    # 2. CLIP 이미지 임베딩
    image_embedding = clip_model.encode_image(image_path)
 
    # 3. CLIP 캡션 텍스트 임베딩
    caption_embedding = clip_model.encode_text(caption)
 
    # 4. 두 임베딩을 결합하여 저장
    # (이미지 임베딩으로 시각적 유사도, 캡션 임베딩으로 의미적 유사도)
    vector_store.add(MultimodalDocument(
        id=f"img_{hash(image_path)}",
        modality="image",
        content_ref=image_path,
        embedding=image_embedding,
        metadata={
            "caption": caption,
            "caption_embedding": caption_embedding.tolist(),
        },
    ))

def hybrid_search(
    query: str,
    clip_model,
    vector_store,
    visual_weight: float = 0.5,
    semantic_weight: float = 0.5,
    top_k: int = 10,
) -> list[dict]:
    """시각적 + 의미적 하이브리드 검색"""
    query_embedding = clip_model.encode_text(query)
 
    # 1. 시각적 유사도 (CLIP 이미지 임베딩 vs 쿼리)
    visual_results = vector_store.search(query_embedding, top_k=top_k * 2)
 
    # 2. 의미적 유사도 (캡션 임베딩 vs 쿼리)
    candidates = {}
    for doc, visual_score in visual_results:
        caption_emb = np.array(doc.metadata.get("caption_embedding", []))
        if caption_emb.size > 0:
            semantic_score = float(np.dot(query_embedding, caption_emb) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(caption_emb)
            ))
        else:
            semantic_score = 0.0
 
        combined_score = (
            visual_weight * visual_score +
            semantic_weight * semantic_score
        )
        candidates[doc.id] = {
            "document": doc,
            "visual_score": visual_score,
            "semantic_score": semantic_score,
            "combined_score": combined_score,
        }
 
    # 3. 결합 점수로 정렬
    sorted_results = sorted(
        candidates.values(),
        key=lambda x: x["combined_score"],
        reverse=True,
    )
    return sorted_results[:top_k]

import hashlib
import json
 
class EmbeddingCache:
    def __init__(self, cache_dir: str = ".embedding_cache"):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
 
    def _key(self, content: str, model: str) -> str:
        return hashlib.sha256(f"{model}:{content}".encode()).hexdigest()
 
    def get(self, content: str, model: str) -> np.ndarray | None:
        path = self.cache_dir / f"{self._key(content, model)}.npy"
        if path.exists():
            return np.load(path)
        return None
 
    def set(self, content: str, model: str, embedding: np.ndarray):
        path = self.cache_dir / f"{self._key(content, model)}.npy"
        np.save(path, embedding)

# 일부 최신 모델은 Matryoshka Representation Learning을 지원
# 임베딩의 앞부분만 잘라도 유의미한 유사도 측정 가능
 
full_embedding = model.encode(text)  # 1024 차원
reduced_embedding = full_embedding[:256]  # 256 차원으로 축소
 
# 정규화
reduced_embedding = reduced_embedding / np.linalg.norm(reduced_embedding)