Multimodal Embedding Pipeline — Text, Vision & Audio dalam Satu Sistem

from PIL import Image
import torch
import clip
from sentence_transformers import SentenceTransformer
import numpy as np

class MultimodalEmbedder:
    def __init__(self):
        # Text: multilingual-e5 untuk support Bahasa Indonesia
        self.text_model = SentenceTransformer("intfloat/multilingual-e5-large")
        # Image: SigLIP lebih baik dari CLIP untuk akurasi
        self.clip_model, self.clip_preprocess = clip.load("ViT-L/14")
        self.clip_model.eval()
        # Projection layer untuk align dimensi (text 1024, CLIP 768)
        self.text_dim = 1024
        self.image_dim = 768
        self.unified_dim = 768  # target: project ke dimensi yang sama

    def embed_text(self, text: str) -> np.ndarray:
        # E5 perlu prefix untuk performance terbaik
        vec = self.text_model.encode(f"query: {text}", normalize_embeddings=True)
        return self._project_text(vec)

    def embed_image(self, image_path: str) -> np.ndarray:
        image = Image.open(image_path)
        image_input = self.clip_preprocess(image).unsqueeze(0)
        with torch.no_grad():
            features = self.clip_model.encode_image(image_input)
        return features.numpy()[0]  # 768-dim

    def embed_audio(self, audio_path: str) -> np.ndarray:
        import whisper
        model = whisper.load_model("base")
        result = model.transcribe(audio_path, language="id")  # Bahasa Indonesia
        return self.embed_text(result["text"])  # audio → text → text embed

    def _project_text(self, vec: np.ndarray) -> np.ndarray:
        # Jika dimensi berbeda, project ke unified dim
        # Untuk production: gunakan learned linear projection layer
        if vec.shape[0] != self.unified_dim:
            return vec[: self.unified_dim]  # simple truncation — production pakai proper projection
        return vec

# Indexing produk e-commerce
embedder = MultimodalEmbedder()
products = [
    {"id": "p1", "name": "Kaos Polos Merah", "image": "p1.jpg"},
    {"id": "p2", "name": "Kemeja Batik Pria", "image": "p2.jpg"},
]
for p in products:
    text_vec = embedder.embed_text(p["name"])
    image_vec = embedder.embed_image(p["image"])
    # Combined embedding: average text + image
    combined = (text_vec + image_vec) / 2
    vector_db.store(p["id"], combined)