การค้นหาข้อมูลแบบดั้งเดิมของ Elasticsearch อาศัยการจับคู่คำตรงตัว (Keyword Matching) ซึ่งมีข้อจำกัดเมื่อผู้ใช้ค้นหาด้วยคำที่มีความหมายใกล้เคียงกันแต่ใช้คำ不同 บทความนี้จะสอนวิธีตั้งค่า AI Semantic Matching เพื่อให้ Elasticsearch เข้าใจความหมายของคำค้นหา ไม่ใช่แค่ตัวอักษร

ตารางเปรียบเทียบบริการ AI API สำหรับ Semantic Search

เกณฑ์ HolySheep AI OpenAI API Anthropic API
อัตราแลกเปลี่ยน ¥1 = $1 (ประหยัด 85%+ เมื่อเทียบกับ OpenAI) $1 ≈ ฿35 $1 ≈ ฿35
ราคา GPT-4.1 $8/MTok $60/MTok -
ราคา Claude Sonnet 4.5 $15/MTok - $18/MTok
ราคา Gemini 2.5 Flash $2.50/MTok - -
ราคา DeepSeek V3.2 $0.42/MTok - -
ความเร็วในการตอบสนอง ต่ำกว่า 50ms 200-500ms 300-800ms
ช่องทางชำระเงิน WeChat, Alipay, บัตรเครดิต บัตรเครดิตระหว่างประเทศเท่านั้น บัตรเครดิตระหว่างประเทศเท่านั้น
เครดิตฟรี มีเมื่อลงทะเบียน $5 เมื่อสมัครใหม่ ไม่มี

AI Semantic Matching คืออะไร

AI Semantic Matching คือเทคนิคการค้นหาที่ใช้โมเดล AI สร้าง Vector Embedding ของทั้งเอกสารและคำค้นหา แล้วคำนวณระยะทางความคล้ายคลึงกัน (Cosine Similarity) แทนการใช้ TF-IDF หรือ BM25 แบบดั้งเดิม

ข้อดีของ Semantic Matching

การติดตั้ง Elasticsearch และ Plugin

สำหรับการตั้งค่า Semantic Search คุณต้องมี Elasticsearch เวอร์ชัน 8.x ขึ้นไป พร้อมกับ Plugin สำหรับ Vector Search

# ติดตั้ง Elasticsearch 8.x บน Ubuntu/Debian
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -
echo "deb https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.11.0-amd64.deb" \
    | sudo tee /etc/apt/sources.list.d/elastic-8.x.list
sudo apt-get update && sudo apt-get install elasticsearch


เริ่มบริการ

sudo systemctl start elasticsearch
sudo systemctl enable elasticsearch


ตรวจสอบสถานะ

curl -k -u elastic:YOUR_PASSWORD https://localhost:9200

การสร้าง Index พร้อม Dense Vector Field

ขั้นตอนแรกคือการสร้าง Index ที่รองรับ Dense Vector สำหรับเก็บ Embedding จาก AI Model

# สร้าง Index พร้อม mapping สำหรับ Semantic Search
PUT /semantic_products
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0,
    "index": {
      "knn": true
    }
  },
  "mappings": {
    "properties": {
      "product_id": { "type": "keyword" },
      "product_name": { "type": "text", "analyzer": "thai" },
      "description": { "type": "text", "analyzer": "thai" },
      "category": { "type": "keyword" },
      "price": { "type": "float" },
      "name_embedding": {
        "type": "dense_vector",
        "dims": 1536,
        "index": true,
        "similarity": "cosine"
      },
      "description_embedding": {
        "type": "dense_vector",
        "dims": 1536,
        "index": true,
        "similarity": "cosine"
      },
      "combined_embedding": {
        "type": "dense_vector",
        "dims": 1536,
        "index": true,
        "similarity": "cosine"
      }
    }
  }
}


ตรวจสอบ Index ที่สร้าง

GET /semantic_products/_mapping

สคริปต์ Python สร้าง Embedding และ Indexing ด้วย HolySheep AI

ตัวอย่างนี้ใช้ HolySheep AI สำหรับสร้าง Embedding ด้วยโมเดล text-embedding-3-small ซึ่งมีค่าใช้จ่ายเพียง $0.42/MTok (DeepSeek V3.2) หรือ $8/MTok (GPT-4.1)

# ติดตั้งไลบรารีที่จำเป็น
pip install elasticsearch openai python-dotenv


config.py

import os
from dotenv import load_dotenv

load_dotenv()


กำหนดค่าการเชื่อมต่อ HolySheep AI

HOLYSHEEP_CONFIG = {
    "base_url": "https://api.holysheep.ai/v1",
    "api_key": "YOUR_HOLYSHEEP_API_KEY",  # ใส่ API Key จาก HolySheep
    "embedding_model": "text-embedding-3-small",
    "embedding_dimensions": 1536
}


กำหนดค่าการเชื่อมต่อ Elasticsearch

ES_CONFIG = {
    "hosts": ["https://localhost:9200"],
    "basic_auth": ("elastic", "YOUR_ES_PASSWORD"),
    "verify_certs": False,
    "ssl_show_warn": False
}


ตัวอย่างการสร้าง Embedding ด้วย HolySheep

import openai

client = openai.OpenAI(
    base_url=HOLYSHEEP_CONFIG["base_url"],
    api_key=HOLYSHEEP_CONFIG["api_key"]
)

def create_embedding(text: str) -> list[float]:
    """สร้าง Embedding vector จากข้อความ"""
    response = client.embeddings.create(
        model=HOLYSHEEP_CONFIG["embedding_model"],
        input=text
    )
    return response.data[0].embedding


ทดสอบการสร้าง Embedding

test_embedding = create_embedding("สมาร์ทโฟนไทม์แลดดี้ไฟต์โปร")
print(f"Embedding dimensions: {len(test_embedding)}")
print(f"First 5 values: {test_embedding[:5]}")

สคริปต์ Indexing ข้อมูลสินค้าพร้อม Embedding

# indexer.py
import json
from elasticsearch import Elasticsearch
from config import client, HOLYSHEEP_CONFIG, ES_CONFIG

class ProductIndexer:
    def __init__(self):
        self.es = Elasticsearch(**ES_CONFIG)
        self.embedding_client = client
        self.model = HOLYSHEEP_CONFIG["embedding_model"]
    
    def create_embeddings_batch(self, texts: list[str]) -> list[list[float]]:
        """สร้าง Embedding หลายข้อความพร้อมกัน"""
        response = self.embedding_client.embeddings.create(
            model=self.model,
            input=texts
        )
        return [item.embedding for item in response.data]
    
    def prepare_document(self, product: dict) -> dict:
        """เตรียมเอกสารพร้อม Embedding สำหรับ Indexing"""
        # สร้าง Embedding จากชื่อและรายละเอียดสินค้า
        name_embedding = self.create_embeddings_batch([product["name"]])[0]
        desc_embedding = self.create_embeddings_batch([product["description"]])[0]
        
        # รวม Embedding จากชื่อและรายละเอียด
        combined_text = f"{product['name']} {product['description']}"
        combined_embedding = self.create_embeddings_batch([combined_text])[0]
        
        return {
            "product_id": product["id"],
            "product_name": product["name"],
            "description": product["description"],
            "category": product["category"],
            "price": product["price"],
            "name_embedding": name_embedding,
            "description_embedding": desc_embedding,
            "combined_embedding": combined_embedding
        }
    
    def index_products(self, products: list[dict], batch_size: int = 50):
        """Index สินค้าพร้อม Embedding"""
        for i in range(0, len(products), batch_size):
            batch = products[i:i+batch_size]
            operations = []
            
            for product in batch:
                doc = self.prepare_document(product)
                operations.append({"index": {"_index": "semantic_products", "_id": doc["product_id"]}})
                operations.append(doc)
            
            # Bulk indexing
            response = self.es.bulk(operations=operations, refresh=True)
            
            if response["errors"]:
                for item in response["items"]:
                    if "error" in item["index"]:
                        print(f"Error indexing {item['index']['_id']}: {item['index']['error']}")
            else:
                print(f"Successfully indexed batch {i//batch_size + 1} ({len(batch)} products)")
        
        # ตรวจสอบจำนวนเอกสารที่ index
        count = self.es.count(index="semantic_products")
        print(f"Total documents indexed: {count['count']}")


ตัวอย่างข้อมูลสินค้า

sample_products = [
    {"id": "P001", "name": "iPhone 15 Pro Max", "description": "สมาร์ทโฟนระดับพรีเมียมจาก Apple หน้าจอ 6.7 นิ้ว กล้อง 48MP", "category": "สมาร์ทโฟน", "price": 49900.00},
    {"id": "P002", "name": "Samsung Galaxy S24 Ultra", "description": "มือถือ Android ระดับไฮเอนด์ หน้าจอ AMOLED 6.8 นิ้ว", "category": "สมาร์ทโฟน", "price": 45900.00},
    {"id": "P003", "name": "MacBook Pro M3 Max", "description": "แล็ปท็อประดับมืออาชีพ ชิพ M3 Max หน้าจอ 16.2 นิ้ว", "category": "แล็ปท็อป", "price": 119900.00},
    {"id": "P004", "name": "Sony WH-1000XM5", "description": "หูฟังไร้สายตัดเสียงระดับท็อป แบตเตอรี่ 30 ชั่วโมง", "category": "หูฟัง", "price": 12900.00}
]


รันการ Indexing

indexer = ProductIndexer()
indexer.index_products(sample_products)

การค้นหาด้วย Semantic Query

เมื่อ Index ข้อมูลเรียบร้อยแล้ว ต่อไปจะเป็นการค้นหาด้วย Semantic Query โดยใช้ KNN (K-Nearest Neighbors)

# search_engine.py
import openai
from elasticsearch import Elasticsearch
from config import ES_CONFIG, HOLYSHEEP_CONFIG

client = openai.OpenAI(
    base_url=HOLYSHEEP_CONFIG["base_url"],
    api_key=HOLYSHEEP_CONFIG["api_key"]
)
es = Elasticsearch(**ES_CONFIG)

def semantic_search(query: str, top_k: int = 10, field: str = "combined_embedding"):
    """
    ค้นหาด้วย Semantic Search
    - query: ข้อความค้นหา
    - top_k: จำนวนผลลัพธ์ที่ต้องการ
    - field: field ที่ใช้ค้นหา (name_embedding, description_embedding, combined_embedding)
    """
    # สร้าง Query Embedding
    query_embedding = client.embeddings.create(
        model=HOLYSHEEP_CONFIG["embedding_model"],
        input=query
    ).data[0].embedding
    
    # ค้นหาด้วย KNN
    response = es.search(
        index="semantic_products",
        knn={
            "field": field,
            "query_vector": query_embedding,
            "k": top_k,
            "num_candidates": 100
        },
        source=["product_id", "product_name", "description", "category", "price"]
    )
    
    return response["hits"]["hits"]

def hybrid_search(query: str, top_k: int = 10):
    """
    ค้นหาแบบผสม (Hybrid Search)
    - รวม Semantic Search + Keyword Search
    - ใช้ RRF (Reciprocal Rank Fusion) ในการรวมผลลัพธ์
    """
    # สร้าง Query Embedding
    query_embedding = client.embeddings.create(
        model=HOLYSHEEP_CONFIG["embedding_model"],
        input=query
    ).data[0].embedding
    
    # Semantic Search
    semantic_response = es.search(
        index="semantic_products",
        knn={
            "field": "combined_embedding",
            "query_vector": query_embedding,
            "k": top_k * 2,
            "num_candidates": 100
        },
        source=["product_name", "description", "price"]
    )
    
    # Keyword Search
    keyword_response = es.search(
        index="semantic_products",
        query={
            "multi_match": {
                "query": query,
                "fields": ["product_name^3", "description"],
                "type": "best_fields"
            }
        },
        size=top_k * 2,
        source=["product_name", "description", "price"]
    )
    
    # รวมผลลัพธ์ด้วย RRF
    from collections import defaultdict
    
    rank_scores = defaultdict(lambda: {"semantic": None, "keyword": None})
    
    for rank, hit in enumerate(semantic_response["hits"]["hits"]):
        rank_scores[hit["_id"]]["semantic"] = rank + 1
    
    for rank, hit in enumerate(keyword_response["hits"]["hits"]):
        rank_scores[hit["_id"]]["keyword"] = rank + 1
    
    # คำนวณ RRF Score
    results = []
    for doc_id, ranks in rank_scores.items():
        rrf_score = 0
        if ranks["semantic"]:
            rrf_score += 1 / (60 + ranks["semantic"])
        if ranks["keyword"]:
            rrf_score += 1 / (60 + ranks["keyword"])
        results.append((doc_id, rrf_score))
    
    # เรียงลำดับตาม RRF Score
    results.sort(key=lambda x: x[1], reverse=True)
    
    return results[:top_k]


ทดสอบการค้นหา

if __name__ == "__main__":
    # ทดสอบ Semantic Search
    print("=== Semantic Search ===")
    results = semantic_search("โทรศัพท์ระดับพรีเมียมกล้องดี")
    for result in results:
        print(f"- {result['_source']['product_name']}: {result['_score']:.4f}")
    
    print("\n=== Hybrid Search ===")
    hybrid_results = hybrid_search("สมาร์ทโฟนแอปเปิล")
    for doc_id, score in hybrid_results:
        doc = es.get(index="semantic_products", id=doc_id)["_source"]
        print(f"- {doc['product_name']}: RRF={score:.4f}")

การปรับแต่งประสิทธิภาพและ Ranking

เพื่อให้ผลการค้นหามีความแม่นยำมากขึ้น ควรปรับแต่ง Query และใช้เทคนิคต่างๆ ดังนี้

# advanced_search.py
from elasticsearch import Elasticsearch
from config import ES_CONFIG, HOLYSHEEP_CONFIG, client

es = Elasticsearch(**ES_CONFIG)

def advanced_semantic_search(
    query: str,
    category_filter: str = None,
    min_price: float = None,
    max_price: float = None,
    boost_name: float = 2.0,
    top_k: int = 20
):
    """
    ค้นหาขั้นสูงพร้อมตัวกรองและการปรับน้ำหนัก
    """
    # สร้าง Query Embedding
    query_embedding = client.embeddings.create(
        model=HOLYSHEEP_CONFIG["embedding_model"],
        input=query
    ).data[0].embedding
    
    # สร้าง filter clauses
    filter_clauses = []
    if category_filter:
        filter_clauses.append({"term": {"category": category_filter}})
    if min_price is not None or max_price is not None:
        price_range = {}
        if min_price is not None:
            price_range["gte"] = min_price
        if max_price is not None:
            price_range["lte"] = max_price
        filter_clauses.append({"range": {"price": price_range}})
    
    # Query ที่ซับซ้อน
    search_body = {
        "query": {
            "bool": {
                "must": {
                    "script_score": {
                        "query": {"match_all": {}},
                        "script": {
                            "source": "cosineSimilarity(params.query_vector, 'combined_embedding') + 1.0",
                            "params": {"query_vector": query_embedding}
                        }
                    }
                },
                "filter": filter_clauses
            }
        },
        "knn": {
            "field": "combined_embedding",
            "query_vector": query_embedding,
            "k": top_k,
            "num_candidates": 100,
            "filter": filter_clauses
        },
        "rank": {
            "rrf": {}
        },
        "_source": ["product_id", "product_name", "description", "category", "price"]
    }
    
    response = es.search(index="semantic_products", body=search_body)
    return response["hits"]["hits"]

def multi_vector_search(query: str, top_k: int = 10):
    """
    ค้นหาด้วยหลาย Vector Fieldsพร้อมกัน
    """
    query_embedding = client.embeddings.create(
        model=HOLYSHEEP_CONFIG["embedding_model"],
        input=query
    ).data[0].embedding
    
    # ค้นหาจากทุก embedding fields แล้วรวมผลลัพธ์
    fields = ["name_embedding", "description_embedding", "combined_embedding"]
    all_scores = {}
    
    for field in fields:
        response = es.search(
            index="semantic_products",
            knn={
                "field": field,
                "query_vector": query_embedding,
                "k": top_k * 2,
                "num_candidates": 100
            },
            _source=["product_name", "price"]
        )
        
        # กำหนดน้ำหนักตาม field
        weights = {
            "name_embedding": 0.4,
            "description_embedding": 0.3,
            "combined_embedding": 0.3
        }
        
        for hit in response["hits"]["hits"]:
            doc_id = hit["_id"]
            score = hit["_score"] * weights[field]
            if doc_id not in all_scores:
                all_scores[doc_id] = {"score": 0, "source": hit["_source"]}
            all_scores[doc_id]["score"] += score
    
    # เรียงลำดับตามคะแนนรวม
    sorted_results = sorted(all_scores.items(), key=lambda x: x[1]["score"], reverse=True)
    return sorted_results[:top_k]


ทดสอบการค้นหาขั้นสูง

results = advanced_semantic_search(
    query="หูฟังเสียงดีราคาไม่แพง",
    category_filter="หูฟัง",
    max_price=15000,
    top_k=5
)
print("ผลการค้นหาหูฟังราคาไม่เกิน 15000 บาท:")
for r in results:
    print(f"  - {r['_source']['product_name']}: ฿{r['_source']['price']:,.2f}")

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

ข้อผิดพลาดที่ 1: Elasticsearch Security Exception หรือ 403 Forbidden

สาเหตุ: การตั้งค่า Security ของ Elasticsearch 8.x ที่เปิดใช้งานโดยค่าเริ่มต้น ทำให้การเชื่อมต่อถูกปฏิเสธ

# วิธีแก้ไขที่ 1: เพิ่ม SSL Verification Option
from elasticsearch import Elasticsearch

es = Elasticsearch(
    ["https://localhost:9200"],
    basic_auth=("elastic", "YOUR_PASSWORD"),
    verify_certs=True,          # หรือ False สำหรับ Development
    ssl_show_warn=False,
    ca_certs="/path/to/http_ca.crt"  # Certificate จาก Elasticsearch
)


วิธีแก้ไขที่ 2: ปิด Security (ไม่แนะนำสำหรับ Production)


แก้ไขไฟล์ /etc/elasticsearch/elasticsearch.yml


xpack.security.enabled: false


network.host: 0.0.0.0



วิธีแก้ไขที่ 3: สร้าง API Key แทนการใช้ Password


curl -X POST "https://localhost:9200/_security/api_key" \


  -u elastic:YOUR_PASSWORD \


  -H "Content-Type: application/json" \


  -d '{"name": "semantic-search-key", "role_descriptors": {"semantic_role": {"cluster": [], "index": [{"names": ["semantic_*"], "privileges": ["read", "write"]}]}}}'


es_with_api_key = Elasticsearch(
    ["https://localhost:9200"],
    api_key=("YOUR_API_KEY_ID", "YOUR_API_KEY_SECRET"),
    verify_certs=True,
    ca_certs="/path/to/http_ca.crt"
)

ข้อผิดพลาดที่ 2: Dimension Mismatch Error ใน Vector Field

สาเหตุ: ขนาดของ Embedding Vector ที่สร้างจาก AI Model ไม่ตรงกับขนาดที่กำหนดไว้ใน Index Mapping

# วิธีแก้ไขที่ 1: ตรวจสอบและสร้าง Index ใหม่พร้อม Dimensions ที่ถูกต้อง

ตรวจสอบ Dimensions ของ Model ที่ใช้

EMBEDDING_MODEL = "text-embedding-3-small"  # 1536 dimensions

EMBEDDING_MODEL = "text-embedding-3-large"  # 3072 dimensions



สร้าง Index ใหม่พร้อม Dimensions ที่ถูกต้อง

PUT /semantic_products_v2
{
  "settings": {
    "index": {
      "knn": true
    }
  },
  "mappings": {
    "properties": {
      "name_embedding": {
        "type": "dense_vector",
        "dims": 1536,  # ต้องตรงกับ Model ที่ใช้
        "index": true,
        "similarity": "cosine"
      }
    }
  }
}


วิธีแก้ไขที่ 2: Reindex ข้อมูลเก่าไปยัง Index ใหม่

POST /_reindex
{
  "source": {
    "index": "semantic_products_old"
  },
  "dest": {
    "index": "semantic_products_v2"
  }
}


วิธีแก้ไขที่ 3: ใช้ Dimension Reduction หากต้องการใช้ Model ที่ใหญ่กว่า

def reduce_embedding_dimension(embedding: list[float], target_dim: int) -> list[float]:
    """ลดขนาด Embedding โดยใช้ Truncation (ไม่แนะนำ ควรใช้ Matryoshka Representation)"""
    return embedding[:target_dim]


วิธีที่ดีกว่า: ใช้ Matryoshka Representation หรือเปลี่ยน Model


text-embedding-3-small รองรับการตั้งค่า dimensions ตอนสร้าง Embedding

def create_embedding_with_custom_dim(text: str, dimensions: int = 1536) -> list[float]:
    response = client.embeddings.create(
        model="text-embedding-3-large",  # ใช้ Model ที่รองรับ 3072+
        input=text,
        dimensions=dimensions  # ตั้งค่า dimensions ที่ต้องการ
    )
    return response.data[0].embedding

ข้อผิดพลาดที่ 3: Rate Limit หรือ 429 Too Many Requests จ

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

บทความที่เกี่ยวข้อง