Metric Learning & CLIP

import torch
from PIL import Image
from transformers import CLIPProcessor, CLIPModel
import torch.nn.functional as F

def clip_multimodal_search(image_paths, search_query):
    """
    CLIP 모델을 사용하여, 주어진 이미지들이 텍스트 쿼리와 
    얼마나 연관성이 있는지(유사도)를 계산합니다.
    """
    
    # 1. 사전 학습된 범용 CLIP 모델 및 프로세서(전처리 장치) 로드
    # "openai/clip-vit-base-patch32"는 가장 대중적인 기본 모델입니다.
    model_id = "openai/clip-vit-base-patch32"
    model = CLIPModel.from_pretrained(model_id)
    processor = CLIPProcessor.from_pretrained(model_id)
    
    # 평가 모드 & 그래디언트 비활성화 (Step 1에서 배운 필수 최적화)
    model.eval()
    
    # 2. 이미지 파일들을 PIL 객체로 로드
    images = [Image.open(path).convert("RGB") for path in image_paths]
    
    # 3. 데이터 전처리 (텍스트 토큰화 및 이미지 리사이징/정규화 동시 수행)
    # return_tensors="pt"는 PyTorch 텐서 형식으로 반환하라는 의미입니다.
    inputs = processor(
        text=[search_query], 
        images=images, 
        return_tensors="pt", 
        padding=True
    )
    
    with torch.no_grad():
        # 4. 모델 추론 (텍스트 임베딩과 이미지 임베딩 추출)
        outputs = model(**inputs)
        
        # 5. 유사도 점수 계산
        # logits_per_image: (이미지 개수 x 텍스트 개수) 형태의 유사도 행렬
        # 코사인 유사도에 기반하여 스케일링된 값을 반환합니다.
        logits_per_image = outputs.logits_per_image 
        
        # 보기 쉽게 Softmax를 취해 확률(0~1)로 변환
        probs = logits_per_image.softmax(dim=0).squeeze()

    # 결과 출력
    print(f"텍스트 쿼리: '{search_query}'")
    for i, path in enumerate(image_paths):
        # 배열 형태에 따라 인덱싱 처리
        prob_score = probs[i].item() if probs.dim() > 0 else probs.item()
        print(f" - [이미지 {i+1}] {path} 일치 확률: {prob_score * 100:.2f}%")
        
    return probs

# ==========================================
# 실행 시뮬레이션
# ==========================================
# 가상의 이미지 3장: 빨간_원피스.jpg, 파란_청바지.jpg, 검은_구두.jpg
# 실제 환경에서는 이 함수를 통해 사용자의 '검색어(Text)'와 DB의 '상품 이미지(Image)'를 직접 매칭합니다.
# 
# query = "화사한 봄에 어울리는 빨간색 원피스"
# results = clip_multimodal_search(
#     ["red_dress.jpg", "blue_jeans.jpg", "black_shoes.jpg"], 
#     query
# )
# 
# 예상 출력:
# 텍스트 쿼리: '화사한 봄에 어울리는 빨간색 원피스'
#  - [이미지 1] red_dress.jpg 일치 확률: 98.50%
#  - [이미지 2] blue_jeans.jpg 일치 확률: 1.20%
#  - [이미지 3] black_shoes.jpg 일치 확률: 0.30%

import torch
from transformers import CLIPProcessor, CLIPModel
from PIL import Image
import torch.nn.functional as F

def clip_zero_shot_classification(image_path, candidate_texts):
    """
    미리 학습된 클래스가 없어도, 우리가 임의로 던져준 후보 텍스트 중
    이미지가 무엇에 가장 가까운지 수학적 내적을 통해 확률을 계산합니다.
    """
    model_id = "openai/clip-vit-base-patch32"
    model = CLIPModel.from_pretrained(model_id)
    processor = CLIPProcessor.from_pretrained(model_id)
    model.eval()

    image = Image.open(image_path).convert("RGB")
    
    # 1. 텐서 변환 및 전처리
    inputs = processor(text=candidate_texts, images=image, return_tensors="pt", padding=True)
    
    with torch.no_grad():
        # 2. 모델 내부의 인코더를 각각 호출하여 '독립적인 벡터(임베딩)'를 추출합니다.
        # Image Feature: [1, 512] 차원 (이미지 1장)
        image_features = model.get_image_features(pixel_values=inputs['pixel_values'])
        # Text Feature: [3, 512] 차원 (텍스트 3개)
        text_features = model.get_text_features(input_ids=inputs['input_ids'], attention_mask=inputs['attention_mask'])
        
        # 3. 벡터 정규화 (L2 Normalization)
        # 길이를 1로 맞추어, 내적(Dot Product)이 곧 코사인 유사도(Cosine Similarity)가 되게 만듭니다.
        image_features = image_features / image_features.norm(dim=-1, keepdim=True)
        text_features = text_features / text_features.norm(dim=-1, keepdim=True)
        
        # 4. 내적 연산 (행렬 곱: Matrix Multiplication)
        # [1, 512] @ [512, 3] = [1, 3] 차원의 유사도 점수(Logits)가 나옵니다.
        # model.logit_scale.exp()는 점수의 크기를 증폭시켜주는 CLIP 고유의 온도(Temperature) 파라미터입니다.
        logit_scale = model.logit_scale.exp()
        logits_per_image = logit_scale * image_features @ text_features.t()
        
        # 5. 확률 변환 (Softmax)
        # 유사도 점수를 합이 1(100%)이 되는 확률로 변환합니다.
        probs = logits_per_image.softmax(dim=1).squeeze()

    # 결과 분석
    print(f"대상 이미지: {image_path}")
    print("-" * 30)
    for i, text in enumerate(candidate_texts):
        print(f"후보 텍스트: '{text}' -> 내적 기반 일치 확률: {probs[i].item() * 100:.2f}%")

# ==========================================
# 실행 시뮬레이션
# ==========================================
# 기존의 ResNet이라면 "이건 홈쇼핑 가전카테고리 3번이야" 같이 정해진 클래스만 내뱉지만,
# CLIP은 우리가 현장에서 지어낸 어떤 텍스트 쿼리와도 실시간으로 유사도를 비교해 냅니다.
# 
# candidates = [
#     "A modern wooden dining table", 
#     "A red leather sofa", 
#     "An abstract oil painting"
# ]
# clip_zero_shot_classification("sample_furniture.jpg", candidates)

import torch
from transformers import CLIPProcessor, CLIPTextModelWithProjection
from typing import List

class HomeShoppingSearchEncoder:
    """
    홈쇼핑 검색 백엔드에 탑재되는 실시간 텍스트 임베딩 전용 클래스.
    실시간 검색 시 무거운 이미지 인코더는 메모리에 올리지 않고 텍스트 인코더만 가동하여 응답 속도를 최적화합니다.
    """
    
    def __init__(self, model_name: str = "openai/clip-vit-base-patch32"):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        
        # 메모리 최적화: Image Encoder를 제외하고 Text Encoder와 Projection Head만 로드합니다.
        self.text_model = CLIPTextModelWithProjection.from_pretrained(model_name).to(self.device)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        
        self.text_model.eval()

    @torch.no_grad()
    def get_search_query_vector(self, search_query: str) -> List[float]:
        """
        고객이 입력한 검색어를 받아 Elasticsearch k-NN 쿼리에 넣을 수 있는
        [512] 차원의 실수 배열(List)로 변환합니다.
        """
        # 1. 고객의 검색어 전처리 (토큰화)
        inputs = self.processor(text=[search_query], return_tensors="pt", padding=True).to(self.device)
        
        # 2. 텍스트 임베딩 벡터 추출
        text_outputs = self.text_model(**inputs)
        text_embeds = text_outputs.text_embeds
        
        # 3. L2 정규화 (Elasticsearch 코사인 유사도 검색을 위해 필수)
        normalized_embeds = text_embeds / text_embeds.norm(dim=-1, keepdim=True)
        
        # 4. JSON 직렬화를 위해 PyTorch Tensor를 순수 Python List로 변환
        return normalized_embeds.squeeze().cpu().tolist()

# ==========================================
# 백엔드 API 컨트롤러 활용 예시
# ==========================================
# encoder = HomeShoppingSearchEncoder()
#
# @app.get("/search")
# def search_products(query: str):
#     # 1. 텍스트 쿼리를 512차원 벡터로 변환 (약 10~20ms 소모)
#     query_vector = encoder.get_search_query_vector(query)
#     
#     # 2. Elasticsearch에 직접 벡터 검색 쿼리(k-NN) 요청
#     es_query = {
#         "knn": {
#             "field": "image_vector", # 오프라인에서 미리 색인해둔 이미지 벡터 필드
#             "query_vector": query_vector,
#             "k": 20,
#             "num_candidates": 200
#         }
#     }
#     # results = es_client.search(index="home_shopping_products", body=es_query)
#     # return parse_results(results)