บทนำ: ทำไม BGE-M3 ถึงเป็นมาตรฐานใหม่ของวงการ Embedding

BGE-M3 (BAAI General Embedding-M3) คือโมเดล embedding หลายภาษาจาก BAAI ที่รองรับมากกว่า 100 ภาษา รวมถึงภาษาไทย ด้วยความสามารถในการสร้าง vector ขนาด 1,024 มิติที่มีความแม่นยำสูงในงาน semantic search, document retrieval และ RAG (Retrieval-Augmented Generation) บทความนี้จะพาคุณเข้าใจวิธีการต่อ BGE-M3 ผ่าน API แบบ step-by-step พร้อม case study จริงจากทีมที่ประสบความสำเร็จในการย้ายระบบ

กรณีศึกษา: ผู้ให้บริการอีคอมเมิร์ซในเชียงใหม่

บริบทธุรกิจ

ทีมสตาร์ทอัพ AI ในเชียงใหม่ที่ให้บริการแพลตฟอร์มค้นหาสินค้าอัจฉริยะสำหรับร้านค้าออนไลน์ SME ได้พัฒนาระบบ product search ด้วย embedding model เพื่อให้ลูกค้าสามารถค้นหาสินค้าด้วยคำบรรยายหรือรูปภาพแทนการพิมพ์ keyword แบบเดิม

จุดเจ็บปวดของผู้ให้บริการเดิม

ทีมเดิมใช้ OpenAI API สำหรับ text-embedding-3-large แต่เจอปัญหาหลายประการ:

การย้ายสู่ HolySheep AI

หลังจากทดสอบหลายเจ้าผู้ให้บริการ ทีมเลือก HolySheep AI เพราะ:

ขั้นตอนการย้ายระบบ

การย้ายระบบใช้เวลาทั้งหมด 3 วัน แบ่งเป็น:

ผลลัพธ์ 30 วันหลังการย้าย

ตัวชี้วัดก่อนย้ายหลังย้ายการเปลี่ยนแปลง
Latency เฉลี่ย420ms180msลดลง 57%
ค่าใช้จ่ายรายเดือน$4,200$680ประหยัด 84%
ความแม่นยำ (NDCG@10)0.720.89เพิ่มขึ้น 24%
ความพึงพอใจลูกค้า3.2/54.6/5เพิ่มขึ้น 44%

BGE-M3 คืออะไร และทำไมต้องใช้ผ่าน API

BGE-M3 พัฒนาโดย Beijing Academy of Artificial Intelligence (BAAI) มีความโดดเด่นในหลายด้าน:

การใช้ BGE-M3 ผ่าน API ช่วยให้คุณไม่ต้องดูแล infrastructure เอง ประหยัด GPU resource และได้รับการอัปเดตโมเดลอัตโนมัติ

การติดตั้งและตั้งค่าเบื้องต้น

ติดตั้ง Client Library

pip install requests huggingface-hub sentence-transformers

สำหรับการใช้งานผ่าน API เราจะใช้ requests library เพื่อควบคุมทุกอย่างได้อย่างละเอียด

สร้าง API Client พื้นฐาน

import requests
import numpy as np
from typing import List, Union

class BGE_M3_Client:
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.embeddings_endpoint = f"{self.base_url}/embeddings"
    
    def get_embedding(
        self, 
        text: str, 
        model: str = "bge-m3",
        dimensions: int = 1024
    ) -> List[float]:
        """สร้าง embedding จากข้อความเดียว"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "input": text,
            "dimensions": dimensions
        }
        
        response = requests.post(
            self.embeddings_endpoint,
            headers=headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code != 200:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
        
        result = response.json()
        return result["data"][0]["embedding"]
    
    def get_embeddings_batch(
        self, 
        texts: List[str], 
        model: str = "bge-m3"
    ) -> List[List[float]]:
        """สร้าง embeddings จากหลายข้อความพร้อมกัน"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "input": texts
        }
        
        response = requests.post(
            self.embeddings_endpoint,
            headers=headers,
            json=payload,
            timeout=60
        )
        
        if response.status_code != 200:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
        
        result = response.json()
        # เรียงลำดับตาม input ต้นฉบับ
        embeddings_map = {item["index"]: item["embedding"] for item in result["data"]}
        return [embeddings_map[i] for i in range(len(texts))]


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

client = BGE_M3_Client(api_key="YOUR_HOLYSHEEP_API_KEY")


Embedding ข้อความเดียว

embedding = client.get_embedding("ร้านกาแฟอร์จานาวาโด้ สาขาเชียงใหม่")
print(f"Embedding dimensions: {len(embedding)}")


Embedding หลายข้อความพร้อมกัน

texts = [
    "เสื้อยืดผ้าฝ้าย สีดำ ขนาด M",
    "Cotton t-shirt black size M",
    "Chemise en coton noire taille M"
]
embeddings = client.get_embeddings_batch(texts)
print(f"Batch embeddings: {len(embeddings)} items")

การคำนวณความคล้ายคลึงและ Semantic Search

หลังจากได้ embedding แล้ว ขั้นตอนถัดไปคือการนำไปใช้ในงาน semantic search โดยใช้ cosine similarity

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class SemanticSearch:
    def __init__(self, client: BGE_M3_Client):
        self.client = client
        self.document_embeddings = []
        self.documents = []
    
    def index_documents(self, documents: List[str]) -> None:
        """สร้าง index สำหรับเอกสารทั้งหมด"""
        print(f"Indexing {len(documents)} documents...")
        
        # เรียก API แบบ batch เพื่อประหยัด cost และเวลา
        batch_size = 32
        all_embeddings = []
        
        for i in range(0, len(documents), batch_size):
            batch = documents[i:i + batch_size]
            batch_embeddings = self.client.get_embeddings_batch(batch)
            all_embeddings.extend(batch_embeddings)
            print(f"  Processed {min(i + batch_size, len(documents))}/{len(documents)}")
        
        self.document_embeddings = np.array(all_embeddings)
        self.documents = documents
        print(f"Indexing complete. Shape: {self.document_embeddings.shape}")
    
    def search(
        self, 
        query: str, 
        top_k: int = 5
    ) -> List[tuple]:
        """ค้นหาเอกสารที่เกี่ยวข้องมากที่สุด"""
        # สร้าง embedding ของ query
        query_embedding = np.array(self.client.get_embedding(query))
        
        # คำนวณ cosine similarity
        similarities = cosine_similarity(
            [query_embedding], 
            self.document_embeddings
        )[0]
        
        # เรียงลำดับและเลือก top_k
        top_indices = np.argsort(similarities)[::-1][:top_k]
        
        results = [
            (self.documents[i], similarities[i], i) 
            for i in top_indices
        ]
        
        return results


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

search_engine = SemanticSearch(client)


สร้าง product catalog

products = [
    "กาแฟอร์จานาวาโด้ ปิดท้าย ร้านกาแฟคลาสสิกในเชียงใหม่",
    "Espresso เข้มข้นจากเมล็ดกาแฟอาราบิก้าคุณภาพสูง",
    "Cold brew กาแฟเย็นสกัดเย็น 24 ชั่วโมง",
    "Latte กาแฟนมร้อน สูตรอิตาเลียนแท้",
    "Cappuccino กาแฟนมฟูเนื้อละเอียด",
    "Americano กาแฟดำเข้มข้นรสชาติจัดจ้าน",
    "Matcha latte ชาเขียวมัทฉะญี่ปุ่นผสมนมสด",
    "Thai iced tea ชาชงเย็นรสหวานมัน"
]

search_engine.index_documents(products)


ทดสอบค้นหา

query = "กาแฟเย็นร้อนอะไรก็ได้ที่ทำจากเมล็ดคุณภาพ"
results = search_engine.search(query, top_k=3)

print(f"\nQuery: '{query}'")
print("-" * 60)
for doc, score, idx in results:
    print(f"[{score:.4f}] {doc}")

การใช้งาน BGE-M3 สำหรับ RAG Pipeline

BGE-M3 เหมาะมากสำหรับ RAG (Retrieval-Augmented Generation) เพราะรองรับ multilingual query และสามารถค้นหาข้ามภาษาได้

import json
from datetime import datetime

class RAG_Pipeline:
    def __init__(self, client: BGE_M3_Client, embedding_model: str = "bge-m3"):
        self.client = client
        self.embedding_model = embedding_model
        self.chunks = []
        self.chunk_embeddings = None
        self.metadata = []
    
    def add_documents(
        self, 
        documents: List[dict],
        chunk_size: int = 512,
        overlap: int = 64
    ) -> None:
        """เพิ่มเอกสารและสร้าง chunks พร้อม embeddings"""
        for doc in documents:
            text = doc["content"]
            doc_id = doc.get("id", hash(text) % 1000000)
            source = doc.get("source", "unknown")
            
            # แบ่งเอกสารเป็น chunks
            chunks = self._chunk_text(text, chunk_size, overlap)
            
            for i, chunk in enumerate(chunks):
                self.chunks.append(chunk)
                self.metadata.append({
                    "doc_id": doc_id,
                    "source": source,
                    "chunk_index": i,
                    "timestamp": datetime.now().isoformat

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