ในโลกของ AI ที่กำลังเปลี่ยนแปลงอย่างรวดเร็ว การค้นหาด้วยภาพ (Visual Search) และการเข้าใจเนื้อหาแบบหลายรูปแบบ (Multi-modal Understanding) ไม่ใช่สิ่งที่เราจะทำได้ในอนาคตอีกต่อไป แต่เป็นความจำเป็นของวันนี้ ในบทความนี้ ผมจะพาทุกคนไปสำรวจเทคนิค Multi-modal Embedding ที่ผมได้ใช้จริงในโปรเจกต์ E-commerce Search Engine ของลูกค้ารายหนึ่ง ซึ่งช่วยเพิ่มอัตราการแปลงสินค้าขึ้นถึง 34%

ทำไมต้อง Multi-modal Embedding?

ในอดีต ระบบค้นหาของเราใช้ได้เฉพาะข้อความเท่านั้น ลูกค้าพิมพ์ "กระเป๋าผ้าสีน้ำตาล" แล้วได้ผลลัพธ์ที่ไม่ตรงกับความต้องการบ่อยครั้ง แต่เมื่อเราเริ่มใช้ Multi-modal Embedding ที่รวม Vector ของทั้งข้อความและรูปภาพเข้าด้วยกัน ทุกอย่างเปลี่ยนไป

ความสามารถหลักของ Multi-modal Embedding มีดังนี้:

การตั้งค่า HolySheep AI API สำหรับ Multi-modal Tasks

สำหรับโปรเจกต์นี้ ผมเลือกใช้ HolySheep AI เพราะราคาประหยัดมากกว่า 85% เมื่อเทียบกับ API อื่น ความหน่วงต่ำกว่า 50ms และรองรับชำระเงินผ่าน WeChat/Alipay ได้สะดวก ราคาของโมเดลที่เราจะใช้มีดังนี้:

การตั้งค่าเริ่มต้นใช้เวลาเพียงไม่กี่นาที:

import requests
import base64
import numpy as np
from PIL import Image
import io

class HolySheepMultiModal:
    """Client สำหรับ Multi-modal Embedding ด้วย HolySheep AI"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def encode_image(self, image_path: str) -> str:
        """แปลงรูปภาพเป็น Base64 string"""
        with open(image_path, "rb") as img_file:
            return base64.b64encode(img_file.read()).decode('utf-8')
    
    def get_text_embedding(self, text: str, model: str = "text-embedding-3-large") -> np.ndarray:
        """สร้าง Text Embedding Vector"""
        payload = {
            "input": text,
            "model": model,
            "encoding_format": "float"
        }
        
        response = requests.post(
            f"{self.base_url}/embeddings",
            headers=self.headers,
            json=payload
        )
        
        if response.status_code != 200:
            raise ValueError(f"API Error: {response.status_code} - {response.text}")
        
        data = response.json()
        return np.array(data['data'][0]['embedding'])
    
    def get_image_embedding(self, image_path: str, model: str = "clip-vit-large-patch14") -> np.ndarray:
        """สร้าง Image Embedding Vector ด้วย CLIP"""
        base64_image = self.encode_image(image_path)
        
        payload = {
            "model": model,
            "image": {
                "type": "base64",
                "data": base64_image
            }
        }
        
        response = requests.post(
            f"{self.base_url}/embeddings",
            headers=self.headers,
            json=payload
        )
        
        if response.status_code != 200:
            raise ValueError(f"API Error: {response.status_code} - {response.text}")
        
        data = response.json()
        return np.array(data['data'][0]['embedding'])


ตัวอย่างการใช้งาน

client = HolySheepMultiModal(api_key="YOUR_HOLYSHEEP_API_KEY")
text_vector = client.get_text_embedding("กระเป๋าผ้าสีน้ำตาลลายหนัง")
image_vector = client.get_image_embedding("product_bag.jpg")
print(f"Text Vector Dimension: {len(text_vector)}")
print(f"Image Vector Dimension: {len(image_vector)}")

การรวม Vector ด้วยเทคนิค Late Fusion

หลังจากได้ Vector จากทั้งสองโมเดลแล้ว ขั้นตอนสำคัญคือการรวม Vector ให้เป็นหนึ่งเดียว ผมใช้เทคนิค Late Fusion ที่ผมปรับแต่งจากงานวิจัยล่าสุด โดยมีการ Weighted Concatenation พร้อม Normalization:

from sklearn.preprocessing import normalize

class MultiModalFusion:
    """รวม Text และ Image Embeddings ด้วยหลายวิธี"""
    
    def __init__(self, text_weight: float = 0.6, image_weight: float = 0.4):
        self.text_weight = text_weight
        self.image_weight = image_weight
    
    def weighted_concatenate(self, text_vec: np.ndarray, image_vec: np.ndarray) -> np.ndarray:
        """วิธีที่ 1: Weighted Concatenation พร้อม Normalization"""
        # Normalize ทั้งสอง Vector ก่อน
        text_norm = normalize(text_vec.reshape(1, -1), norm='l2').flatten()
        image_norm = normalize(image_vec.reshape(1, -1), norm='l2').flatten()
        
        # คูณด้วย Weight แล้ว Concatenate
        weighted_text = text_norm * self.text_weight
        weighted_image = image_norm * self.image_weight
        
        fused = np.concatenate([weighted_text, weighted_image])
        
        # Normalize ผลลัพธ์สุดท้าย
        return normalize(fused.reshape(1, -1), norm='l2').flatten()
    
    def hadamard_product(self, text_vec: np.ndarray, image_vec: np.ndarray) -> np.ndarray:
        """วิธีที่ 2: Element-wise Hadamard Product"""
        text_norm = normalize(text_vec.reshape(1, -1), norm='l2').flatten()
        image_norm = normalize(image_vec.reshape(1, -1), norm='l2').flatten()
        
        # Hadamard Product (Element-wise multiplication)
        fused = text_norm * image_norm
        
        return normalize(fused.reshape(1, -1), norm='l2').flatten()
    
    def learned_fusion(self, text_vec: np.ndarray, image_vec: np.ndarray) -> np.ndarray:
        """วิธีที่ 3: Learned Fusion ด้วย Cross-attention"""
        text_norm = normalize(text_vec.reshape(1, -1), norm='l2').flatten()
        image_norm = normalize(image_vec.reshape(1, -1), norm='l2').flatten()
        
        # คำนวณ Cross-attention weight
        attention = np.dot(text_norm, image_norm)
        
        # ปรับ Weight ตาม Attention Score
        dynamic_weight = (1 + attention) / 2
        
        fused = (text_norm * dynamic_weight) + (image_norm * (1 - dynamic_weight))
        
        return normalize(fused.reshape(1, -1), norm='l2').flatten()


ตัวอย่างการใช้งาน

fusion = MultiModalFusion(text_weight=0.6, image_weight=0.4)


รวม Vector ด้วย 3 วิธี

fused_concat = fusion.weighted_concatenate(text_vector, image_vector)
fused_hadamard = fusion.hadamard_product(text_vector, image_vector)
fused_attention = fusion.learned_fusion(text_vector, image_vector)

print(f"Fused Vector Shape: {fused_concat.shape}")
print(f"Hadamard Vector Shape: {fused_hadamard.shape}")
print(f"Attention-based Shape: {fused_attention.shape}")

การสร้างระบบ Visual Search สำหรับ E-commerce

ต่อไปคือการนำ Multi-modal Embedding ไปใช้จริงในระบบ Visual Search ที่ผมพัฒนาให้ร้านค้าออนไลน์ ซึ่งรองรับการค้นหาด้วยรูปภาพหรือข้อความก็ได้:

import faiss
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class Product:
    product_id: str
    name: str
    description: str
    category: str
    image_path: str

class VisualSearchEngine:
    """ระบบค้นหาสินค้าด้วย Multi-modal Embedding"""
    
    def __init__(self, client: HolySheepMultiModal, dimension: int = 1536):
        self.client = client
        self.dimension = dimension
        
        # ใช้ Inner Product Index สำหรับ Cosine Similarity (เมื่อ Vector เป็น Normalized)
        self.index = faiss.IndexFlatIP(dimension)
        self.products: List[Product] = []
    
    def index_products(self, products: List[Product], batch_size: int = 32):
        """สร้าง Index สำหรับร้านค้าทั้งหมด"""
        all_embeddings = []
        
        for i in range(0, len(products), batch_size):
            batch = products[i:i + batch_size]
            
            for product in batch:
                # รวม Vector จากทั้งรูปภาพและข้อความ
                text_emb = self.client.get_text_embedding(
                    f"{product.name} {product.description} {product.category}"
                )
                image_emb = self.client.get_image_embedding(product.image_path)
                
                # Fusion ทั้งสอง Vector
                fusion = MultiModalFusion(text_weight=0.5, image_weight=0.5)
                combined = fusion.weighted_concatenate(text_emb, image_emb)
                
                all_embeddings.append(combined)
                self.products.append(product)
                
                print(f"Indexed: {product.product_id} - {product.name}")
        
        # เพิ่ม Vector ทั้งหมดใน Index
        embeddings_array = np.array(all_embeddings).astype('float32')
        self.index.add(embeddings_array)
        print(f"Total products indexed: {len(self.products)}")
    
    def search(self, query: Optional[str] = None, query_image: Optional[str] = None, 
               top_k: int = 10) -> List[tuple]:
        """ค้นหาสินค้าที่คล้ายกัน"""
        if query and query_image:
            # ทั้งข้อความและรูป
            text_emb = self.client.get_text_embedding(query)
            image_emb = self.client.get_image_embedding(query_image)
            fusion = MultiModalFusion(text_weight=0.5, image_weight=0.5)
            query_vector = fusion.weighted_concatenate(text_emb, image_emb)
        elif query:
            query_vector = self.client.get_text_embedding(query)
        elif query_image:
            query_vector = self.client.get_image_embedding(query_image)
        else:
            raise ValueError("ต้องระบุ query หรือ query_image อย่างน้อยหนึ่งอย่าง")
        
        # ค้นหา Top-K ที่ใกล้เคียงที่สุด
        query_vector = query_vector.reshape(1, -1).astype('float32')
        scores, indices = self.index.search(query_vector, top_k)
        
        results = []
        for score, idx in zip(scores[0], indices[0]):
            if idx < len(self.products):
                results.append((self.products[idx], score))
        
        return results


ตัวอย่างการใช้งาน

engine = VisualSearchEngine(client)


สมมติมีรายการสินค้า

products = [
    Product("P001", "กระเป๋าผ้าสีน้ำตาล", "กระเป๋าผ้าลินินสำหรับใส่ของใช้", "กระเป๋า", "bags/brown_linen.jpg"),
    Product("P002", "เสื้อยืดสีดำ", "เสื้อยืดคอกลมผ้าฝ้าย", "เสื้อผ้า", "shirts/black_tee.jpg"),
    # ... สินค้าอื่นๆ
]


สร้าง Index

engine.index_products(products)


ค้นหาด้วยรูปภาพ

results = engine.search(query_image="customer_upload.jpg", top_k=5)
for product, score in results:
    print(f"[{score:.4f}] {product.product_id}: {product.name}")

การประยุกต์ใช้กับ RAG ขององค์กร

นอกจาก Visual Search แล้ว Multi-modal Embedding ยังสามารถนำไปใช้ในระบบ RAG (Retrieval Augmented Generation) ขององค์กรได้อีกด้วย สมมติว่าบริษัทของคุณมีเอกสารที่มีทั้งรูปภาพและข้อความ คุณสามารถค้นหาเนื้อหาที่เกี่ยวข้องได้ทั้งจากคำอธิบายและภาพประกอบ:

from typing import Dict, Any
import json

class MultiModalRAG:
    """ระบบ RAG ที่รองรับทั้งข้อความและรูปภาพ"""
    
    def __init__(self, client: HolySheepMultiModal):
        self.client = client
        self.document_index = faiss.IndexFlatIP(3072)  # สำหรับ Combined Embedding
        self.documents: List[Dict[str, Any]] = []
    
    def process_document(self, doc_path: str, images: List[str]) -> np.ndarray:
        """ประมวลผลเอกสารพร้อมรูปภาพประกอบ"""
        # อ่านข้อความจากเอกสาร
        with open(doc_path, 'r', encoding='utf-8') as f:
            text_content = f.read()
        
        # สร้าง Text Embedding
        text_emb = self.client.get_text_embedding(text_content)
        
        # สร้าง Image Embedding จากรูปภาพทั้งหมดในเอกสาร
        image_embeddings = []
        for img_path in images:
            img_emb = self.client.get_image_embedding(img_path)
            image_embeddings.append(img_emb)
        
        # หาค่าเฉลี่ยของ Image Embeddings
        avg_image_emb = np.mean(image_embeddings, axis=0) if image_embeddings else np.zeros(1536)
        
        # รวม Text และ Image Embeddings
        fusion = MultiModalFusion(text_weight=0.7, image_weight=0.3)
        combined = fusion.weighted_concatenate(text_emb, avg_image_emb)
        
        return combined
    
    def add_document(self, doc_id: str, doc_path: str, 
                     images: List[str], metadata: Dict[str, Any]):
        """เพิ่มเอกสารในระบบ"""
        combined_emb = self.process_document(doc_path, images)
        
        self.document_index.add(combined_emb.reshape(1, -1).astype('float32'))
        self.documents.append({
            "doc_id": doc_id,
            "path": doc_path,
            "metadata": metadata
        })
    
    def retrieve_context(self, query: str, top_k: int = 5) -> List[Dict[str, Any]]:
        """ดึงเนื้อหาที่เกี่ยวข้องสำหรับ Query"""
        query_emb = self.client.get_text_embedding(query)
        
        # ค้นหาเอกสารที่ใกล้เคียง
        _, indices = self.document_index.search(
            query_emb.reshape(1, -1).astype('float32'), top_k
        )
        
        results = []
        for idx in indices[0]:
            if idx < len(self.documents):
                doc = self.documents[idx]
                # โหลดเนื้อหาจริง
                with open(doc['path'], 'r', encoding='utf-8') as f:
                    content = f.read()
                results.append({
                    "doc_id": doc['doc_id'],
                    "content": content,
                    "metadata": doc['metadata']
                })
        
        return results
    
    def generate_answer(self, query: str, context_docs: List[Dict]) -> str:
        """สร้างคำตอบจาก Context ที่ดึงมา"""
        context_text = "\n\n".join([doc['content'] for doc in context_docs])
        
        prompt = f"""อ่านเนื้อหาต่อไปนี้แล้วตอบคำถาม:

เนื้อหา:
{context_text}

คำถาม: {query}

คำตอบ:"""

        payload = {
            "model": "gpt-4.1",
            "messages": [
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.3,
            "max_tokens": 1000
        }
        
        response = requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {self.client.api_key}",
                "Content-Type": "application/json"
            },
            json=payload
        )
        
        return response.json()['choices'][0]['message']['content']


ตัวอย่างการใช้งาน

rag = MultiModalRAG(client)


เพิ่มเอกสาร

rag.add_document(
    doc_id="DOC001",
    doc_path="documents/product_manual.txt",
    images=["documents/images/diagram1.jpg", "documents/images/diagram2.jpg"],
    metadata={"category": "คู่มือการใช้งาน", "date": "2025-01-15"}
)


ถามคำถาม

results = rag.retrieve_context("วิธีการติดตั้งอุปกรณ์")
answer = rag.generate_answer("วิธีการติดตั้งอุปกรณ์?", results)
print(answer)

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. ข้อผิดพลาด: "Invalid API Key" หรือ Authentication Error

สาเหตุ: API Key ไม่ถูกต้องหรือหมดอายุ หรือใช้ Base URL ผิด

# ❌ วิธีที่ผิด - ใช้ Base URL ของ OpenAI โดยตรง
response = requests.post(
    "https://api.openai.com/v1/embeddings",  # ผิด!
    headers={"Authorization": f"Bearer {api_key}"},
    json=payload
)


✅ วิธีที่ถูก - ใช้ Base URL ของ HolySheep

response = requests.post(
    "https://api.holysheep.ai/v1/embeddings",  # ถูกต้อง!
    headers={
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    },
    json=payload
)


หรือสร้าง Session สำหรับใช้ซ้ำ

class HolySheepClient:
    def __init__(self, api_key: str):
        if not api_key or len(api_key) < 10:
            raise ValueError("API Key ไม่ถูกต้อง กรุณาตรวจสอบที่ https://www.holysheep.ai/register")
        
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def validate_connection(self) -> bool:
        """ตรวจสอบว่า API Key ทำงานได้หรือไม่"""
        try:
            response = self.session.post(
                f"{self.base_url}/models"
            )
            return response.status_code == 200
        except Exception:
            return False


ใช้งาน

client = HolySheepClient("YOUR_HOLYSHEEP_API_KEY")
if client.validate_connection():
    print("เชื่อมต่อสำเร็จ!")
else:
    print("API Key ไม่ถูกต้อง กรุณาตรวจสอบที่ dashboard")

2. ข้อผิดพลาด: Vector Dimension Mismatch

สาเหตุ: Text Embedding และ Image Embedding มี Dimension ไม่เท่ากัน ทำให้ไม่สามารถรวม Vector ได้

# ปัญหา: Text Vector อาจมี 1536 Dimension แต่ Image Vector มี 512 Dimension
text_emb = client.get_text_embedding("สินค้าลดราคา")  # shape: (1536,)
image_emb = client.get_image_embedding("product.jpg")  # shape: (512,)


❌ ผิด - พยายามรวม Vector โดยตรง

fused = np.concatenate([text_emb, image_emb])  # Error!


✅ วิธีแก้ไข - Project ให้อยู่ใน Space เดียวกัน

class VectorProjector:
    """Project Vector ให้อยู่ใน Dimension เดียวกัน"""
    
    def __init__(self, target_dim: int = 512):
        self.target_dim = target_dim
        self.projection_matrix = None
    
    def fit(self, sample_text_embeddings: List[np.ndarray], 
            sample_image_embeddings: List[np.ndarray]):
        """สร้าง Projection Matrix จากตัวอย่าง"""
        # สร้าง Linear Projection สำหรับ Text
        text_dim = sample_text_embeddings[0].shape[0]
        self.text_projection = np.random.randn(text_dim, self.target_dim) * 0.02
        
        # Normalize Projection Matrix
        self.text_projection = self.text_projection / np.linalg.norm(self.text_projection, axis=0)
    
    def project_text(self, text_vec: np.ndarray) -> np.ndarray:
        """Project Text Vector ให้มี Dimension ตามที่ต้องการ"""
        projected = np.dot(text_vec, self.text_projection)
        return projected / np.linalg.norm(projected)
    
    def project_image(self, image_vec: np.ndarray) -> np.ndarray:
        """Image Vector อาจต้อง Pad หรือ Resize"""
        if image_vec.shape[0] < self.target_dim:
            # Pad with zeros
            padded = np.zeros(self.target_dim)
            padded[:image_vec.shape[0]] = image_vec
            return padded / np.linalg.norm(padded)
        elif image_vec.shape[0] > self.target_dim:
            return image_vec[:self.target_dim] / np.linalg.norm(image_vec[:self.target_dim])
        return image_vec / np.linalg.norm(image_vec)


ใช้งาน

projector = VectorProjector(target_dim=512)

Fit ด้วยตัวอย่างก่อน

projector.fit([text_emb], [image_emb])


Project ก่อนรวม

text_projected = projector.project_text(text_emb)  # (512,)
image_projected = projector.project_image(image_emb)  # (512,)


ตอนนี้รวมได้แล้ว

fused = np.concatenate([text_projected, image_projected])  # (1024,)

3. ข้อผิดพลาด: Rate Limit และ Timeout

สาเหตุ: ส่ง Request มากเกินไปในเวลาสั้น หรือ Image Processing ใช้เวลานานเกินไป

import time
from threading import Lock
from functools import wraps

class RateLimitedClient:
    """Client ที่รองรับ Rate Limiting อัตโนมัติ"""
    
    def __init__(self, api_key: str, max_requests_per_minute: int = 60):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.max_requests = max_requests_per_minute
        self.request_times = []
        self.lock = Lock()
        
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def _wait_if_needed(self):
        """รอถ้าจำนวน Request เกิน Limit"""
        with self.lock:
            now = time.time()
            # ลบ Request ที่เก่ากว่า 1 นาที
            self.request_times = [t for t in self.request_times if now - t < 60]
            
            if len(self.request_times) >= self.max_requests:
                # คำนวณเวลาที่ต้องรอ
                oldest = self.request_times[0]
                wait_time = 60 - (now - oldest) + 1
                print(f"Rate limit reached. Waiting {wait_time:.1f} seconds...")
                time.sleep(wait_time)
                self.request_times = []
            
            self.request_times.append(time.time())
    
    def request_with_retry(self, url: str, payload: dict, 
                          max_retries: int = 3, timeout: int = 30) -> dict:
        """ส่ง Requestพร้อม Retry Logic"""
        for attempt in range(max_retries):
            try:
                self._wait_if_needed()
                
                response = self.session.post(
                    url,
                    json=payload,
                    timeout=timeout
                )
                
                if response.status_code == 429:
                    # Rate Limited - รอแล้วลองใหม่
                    retry_after = int(response.headers.get('Retry-After', 60))
                    print(f"Rate limited. Retrying after {retry_after}s...")
                    time.sleep(retry_after)
                    continue
                
                if response.status_code == 500:
                    # Server Error - ลองใหม่
                    wait_time = 2 ** attempt
                    print(f"Server error. Retrying in {wait_time}s...")
                    time.sleep(wait_time)
                    continue
                
                return response.json()
                
            except requests.Timeout:
                print(f"Timeout on attempt {attempt + 1}. Retrying...")
                if attempt < max_retries - 1:
                    time.sleep(5 * (attempt + 1))
                    continue
                raise ValueError("Request timeout after multiple retries")
            
            except requests.RequestException as e:
                print(f"Request failed: {e}")
                raise
        
        raise ValueError(f"Failed after {max_retries} retries")
    
    def get_embedding(self, text: str = None, image_base64: str = None) -> np.ndarray:
        """ดึง Embedding พร้อม Rate Limiting"""
        if text:
            payload = {"input": text, "model": "text-embedding-3-large"}
            url = f"{self.base_url}/embeddings"
        elif image_base64:
            payload = {
                "model": "clip-vit-large-patch14",
                "image": {"type": "base64", "data": image_base64}
            }
            url = f"{self.base_url}/embeddings"
        else:
            raise ValueError("ต้องระบุ text หรือ image_base64")
        
        result = self.request_with_retry(url, payload)
        return np

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง