ในบทความนี้เราจะมาเจาะลึกเทคนิคการใช้ Metadata Filtering ใน Vector Database สำหรับระบบ RAG (Retrieval-Augmented Generation) ที่ช่วยให้การค้นหาข้อมูลแม่นยำมากขึ้น ลด hallucination และประหยัดค่าใช้จ่ายได้อย่างมีประสิทธิภาพ โดยเนื้อหาครอบคลุมตั้งแต่พื้นฐานจนถึงการนำไปใช้จริงใน Production
ทำไมต้องใช้ Metadata Filtering ใน RAG
ในระบบ RAG แบบดั้งเดิม ทุกครั้งที่ผู้ใช้ถามคำถาม ระบบจะทำ Semantic Search จาก Vector Database ทั้งหมด ทำให้ผลลัพธ์อาจมีเอกสารที่ไม่เกี่ยวข้องปนเข้ามา เช่น ถามเรื่องนโยบายปี 2024 แต่ได้เอกสารปี 2023 หรือถามเรื่องผลิตภัณฑ์ A แต่ได้เอกสารผลิตภัณฑ์ B
Metadata Filtering ช่วยให้เรากรองผลลัพธ์ตามข้อมูลเชิงบริบท เช่น วันที่ หมวดหมู่ ผู้เขียน หรือแผนก เพื่อให้ได้เอกสารที่ตรงใจมากที่สุด ลด Token Usage และเพิ่มความแม่นยำของคำตอบ
การเปรียบเทียบโครงสร้างข้อมูลแบบต่างๆ
ก่อนจะเริ่มต้นใช้งาน เรามาดูโครงสร้างข้อมูลที่เราจะใช้ในตัวอย่างนี้กัน โดยเราจะจัดเก็บเอกสารคู่มือการใช้งานผลิตภัณฑ์ 3 ประเภท ได้แก่ Hardware, Software และ Cloud Service
# โครงสร้างข้อมูลเอกสารที่จัดเก็บใน Vector Database
documents = [
{
"id": "doc_001",
"content": "คู่มือการตั้งค่า Server แบบครบวงจร...",
"metadata": {
"category": "Hardware",
"product": "Enterprise Server X500",
"created_date": "2024-01-15",
"department": "IT-Infrastructure",
"version": "3.2"
}
},
{
"id": "doc_002",
"content": "การใช้งาน Cloud Dashboard สำหรับผู้ดูแลระบบ...",
"metadata": {
"category": "Cloud Service",
"product": "CloudDash Pro",
"created_date": "2024-03-20",
"department": "Cloud-Operations",
"version": "2.1"
}
},
{
"id": "doc_003",
"content": "การติดตั้งและใช้งาน IDE สำหรับนักพัฒนา...",
"metadata": {
"category": "Software",
"product": "DevStudio Ultimate",
"created_date": "2024-02-10",
"department": "Development",
"version": "5.0"
}
}
]
กรองเฉพาะเอกสารที่มี category = 'Software'
filtered = [doc for doc in documents if doc["metadata"]["category"] == "Software"]
print(f"พบเอกสาร {len(filtered)} รายการ")การตั้งค่า HolySheep AI API สำหรับ Embedding และ RAG
สำหรับการใช้งาน Vector Database ร่วมกับ LLM ในระบบ RAG เราจะใช้ HolySheep AI เป็น API Gateway เพราะให้บริการทั้ง Embedding และ Chat Completion ในราคาที่ประหยัดกว่ามาก คือเริ่มต้นเพียง $0.42/MTok สำหรับ DeepSeek V3.2 รวมถึงรองรับชำระเงินผ่าน WeChat และ Alipay อีกด้วย
import requests
import json
from datetime import datetime
การตั้งค่า HolySheep API
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class HolySheepRAG:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def create_embedding(self, text: str, model: str = "text-embedding-3-small"):
"""สร้าง Embedding vector สำหรับข้อความ"""
response = requests.post(
f"{self.base_url}/embeddings",
headers=self.headers,
json={"input": text, "model": model}
)
if response.status_code == 200:
return response.json()["data"][0]["embedding"]
else:
raise Exception(f"Embedding Error: {response.text}")
def chat_completion(self, messages: list, model: str = "gpt-4o-mini",
temperature: float = 0.3, max_tokens: int = 1000):
"""ส่งคำถามไปยัง LLM พร้อมบริบทจาก RAG"""
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
)
if response.status_code == 200:
return response.json()["choices"][0]["message"]["content"]
else:
raise Exception(f"Chat Error: {response.text}")
ทดสอบการเชื่อมต่อ
rag = HolySheepRAG(HOLYSHEEP_API_KEY)
print("✅ เชื่อมต่อ HolySheep API สำเร็จ")การสร้างระบบ Metadata Filtering แบบละเอียด
ต่อไปเราจะสร้าง Class สำหรับจัดการ Vector Database พร้อมระบบกรองข้อมูลตาม Metadata ที่ซับซ้อน รองรับการกรองหลายเงื่อนไขพร้อมกัน
import numpy as np
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum
class FilterOperator(Enum):
EQ = "eq" # เท่ากับ
NE = "ne" # ไม่เท่ากับ
GT = "gt" # มากกว่า
GTE = "gte" # มากกว่าหรือเท่ากับ
LT = "lt" # น้อยกว่า
LTE = "lte" # น้อยกว่าหรือเท่ากับ
IN = "in" # อยู่ใน list
NOT_IN = "not_in" # ไม่อยู่ใน list
CONTAINS = "contains" # มี substring
@dataclass
class FilterCondition:
field: str
operator: FilterOperator
value: Any
class VectorStoreWithFilter:
def __init__(self, rag_client: HolySheepRAG):
self.rag = rag_client
self.vectors = [] # เก็บ vector embeddings
self.documents = [] # เก็บเอกสารต้นฉบับพร้อม metadata
def add_document(self, content: str, metadata: Dict[str, Any]):
"""เพิ่มเอกสารพร้อมสร้าง embedding"""
embedding = self.rag.create_embedding(content)
doc_id = f"doc_{len(self.documents):04d}"
self.vectors.append(embedding)
self.documents.append({
"id": doc_id,
"content": content,
"metadata": metadata
})
print(f"✅ เพิ่มเอกสาร {doc_id} แล้ว")
def apply_filter(self, documents: List[Dict],
conditions: List[FilterCondition]) -> List[Dict]:
"""กรองเอกสารตามเงื่อนไข metadata"""
filtered = []
for doc in documents:
match = True
for cond in conditions:
value = doc["metadata"].get(cond.field)
if value is None:
match = False
break
if cond.operator == FilterOperator.EQ:
if value != cond.value:
match = False
break
elif cond.operator == FilterOperator.GT:
if not (value > cond.value):
match = False
break
elif cond.operator == FilterOperator.IN:
if value not in cond.value:
match = False
break
elif cond.operator == FilterOperator.CONTAINS:
if cond.value not in str(value):
match = False
break
if match:
filtered.append(doc)
return filtered
def search(self, query: str, top_k: int = 5,
filters: Optional[List[FilterCondition]] = None) -> List[Dict]:
"""ค้นหาด้วย semantic search พร้อมกรอง metadata"""
# สร้าง query embedding
query_embedding = self.rag.create_embedding(query)
# กรองเอกสารตาม metadata ก่อน (ถ้ามีเงื่อนไข)
candidates = self.documents
if filters:
candidates = self.apply_filter(self.documents, filters)
print(f"🔍 กรองเหลือ {len(candidates)} เอกสารจาก {len(self.documents)} รายการ")
if not candidates:
return []
# คำนวณ cosine similarity
scores = []
for i, doc in enumerate(candidates):
vec = self.vectors[self.documents.index(doc)]
similarity = np.dot(query_embedding, vec) / (
np.linalg.norm(query_embedding) * np.linalg.norm(vec)
)
scores.append((similarity, doc))
# เรียงลำดับและเลือก top_k
scores.sort(reverse=True)
results = []
for score, doc in scores[:top_k]:
doc_copy = doc.copy()
doc_copy["similarity_score"] = round(score, 4)
results.append(doc_copy)
return results
ตัวอย่างการใช้งาน
store = VectorStoreWithFilter(rag)
เพิ่มเอกสารตัวอย่าง
store.add_document(
"คู่มือการตั้งค่า Server แบบครบวงจรสำหรับองค์กร...",
{"category": "Hardware", "product": "Server X500", "price": 150000}
)
store.add_document(
"การใช้งาน Cloud Dashboard สำหรับผู้ดูแลระบบ...",
{"category": "Cloud", "product": "CloudDash", "price": 5000}
)
store.add_document(
"การติดตั้ง IDE สำหรับนักพัฒนา Python...",
{"category": "Software", "product": "DevStudio", "price": 2990}
)
print(f"📦 จำนวนเอกสารทั้งหมด: {len(store.documents)}")การค้นหาแบบละเอียดพร้อมตัวอย่างการใช้งานจริง
ต่อไปเราจะมาดูตัวอย่างการค้นหาที่ซับซ้อนขึ้น เช่น การค้นหาสินค้าที่มีราคาในช่วงที่กำหนด หรือการกรองตามหลายหมวดหมู่พร้อมกัน
# ตัวอย่างการค้นหาหลายรูปแบบ
1. ค้นหาสินค้าเฉพาะหมวด Software เท่านั้น
results = store.search(
query="การติดตั้งและใช้งานโปรแกรม",
filters=[FilterCondition("category", FilterOperator.EQ, "Software")]
)
print("📌 ผลการค้นหาหมวด Software:")
for r in results:
print(f" - {r['id']}: {r['metadata']['product']} (score: {r['similarity_score']})")
2. ค้นหาสินค้าที่มีราคาต่ำกว่า 10,000 บาท
results = store.search(
query="เครื่องมือสำหรับทำงาน",
filters=[FilterCondition("price", FilterOperator.LT, 10000)]
)
print("\n📌 ผลการค้นหาราคาต่ำกว่า 10,000:")
for r in results:
print(f" - {r['metadata']['product']}: {r['metadata']['price']} บาท")
3. ค้นหาจากหลายหมวดหมู่
results = store.search(
query="คู่มือการใช้งาน",
filters=[FilterCondition("category", FilterOperator.IN, ["Software", "Cloud"])]
)
print("\n📌 ผลการค้นหาหมวด Software และ Cloud:")
for r in results:
print(f" - {r['metadata']['product']} ({r['metadata']['category']})")การสร้าง RAG Pipeline สมบูรณ์
ตอนนี้เราจะนำทุกอย่างมารวมกันเป็น RAG Pipeline ที่สมบูรณ์ ซึ่งจะทำงานดังนี้ รับคำถามจากผู้ใช้ กรองข้อมูลตามเงื่อนไขที่กำหนด ค้นหาเอกสารที่เกี่ยวข้อง และส่งไปยัง LLM เพื่อสร้างคำตอบ
class RAGPipeline:
def __init__(self, vector_store: VectorStoreWithFilter,
rag_client: HolySheepRAG):
self.store = vector_store
self.rag = rag_client
def query(self, user_question: str,
category_filter: Optional[str] = None,
max_price: Optional[float] = None,
top_k: int = 3) -> str:
"""ประมวลผลคำถามผ่าน RAG Pipeline"""
# สร้างเงื่อนไขการกรอง
filters = []
if category_filter:
filters.append(
FilterCondition("category", FilterOperator.EQ, category_filter)
)
if max_price:
filters.append(
FilterCondition("price", FilterOperator.LT, max_price)
)
# ค้นหาเอกสารที่เกี่ยวข้อง
results = self.store.search(user_question, top_k=top_k, filters=filters)
if not results:
return "ไม่พบเอกสารที่เกี่ยวข้องกับคำถามของคุณ"
# รวบรวมบริบทจากเอกสารที่พบ
context_parts = []
for i, doc in enumerate(results, 1):
context_parts.append(
f"[เอกสาร {i}] {doc['content']}\n"
f"หมวดหมู่: {doc['metadata'].get('category', 'N/A')}\n"
f"ผลิตภัณฑ์: {doc['metadata'].get('product', 'N/A')}"
)
context = "\n\n".join(context_parts)
# สร้าง prompt สำหรับ LLM
system_prompt = """คุณเป็นผู้ช่วยตอบคำถามจากคู่มือการใช้งานผลิตภัณฑ์
ให้ตอบคำถามโดยอิงจากบริบทที่ให้มาเท่านั้น
ถ้าไม่แน่ใจให้ตอบว่าไม่มีข้อมูลในเอกสาร"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"บริบท:\n{context}\n\nคำถาม: {user_question}"}
]
# ส่งไปยัง LLM
answer = self.rag.chat_completion(messages, model="gpt-4o-mini")
return answer
ทดสอบ RAG Pipeline
pipeline = RAGPipeline(store, rag)
ทดสอบการค้นหาแบบกรองหมวดหมู่
answer = pipeline.query(
"วิธีการติดตั้งอย่างไร",
category_filter="Software"
)
print(f"💬 คำตอบ: {answer}")การวัดผลและปรับปรุงประสิทธิภาพ
เพื่อให้ระบบ RAG ทำงานได้อย่างมีประสิทธิภาพ เราควรวัดผลอย่างสม่ำเสมอ โดยมีเมตริกสำคัญดังนี้
- Recall@k - ว่ามีเอกสารที่เกี่ยวข้องติดอยู่ในผลการค้นหา top-k กี่เปอร์เซ็นต์
- Precision@k - ว่าในผลการค้นหา top-k มีเอกสารที่เกี่ยวข้องจริงกี่เปอร์เซ็นต์
- MRR (Mean Reciprocal Rank) - ตำแหน่งเฉลี่ยของเอกสารที่เกี่ยวข้องที่ดีที่สุด
- NDCG - คะแนนคุณภาพการจัดอันดับแบบมีการให้น้ำหนัก
- Latency - เวลาตอบสนองของระบบ ควรน้อยกว่า 50ms
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. ปัญหา: ได้ผลลัพธ์ว่างเปล่าหลังการกรอง Metadata
สาเหตุ: เงื่อนไขการกรองเข้มเกินไปจนไม่มีเอกสารตรงกับเงื่อนไข หรือชื่อฟิลด์ใน FilterCondition ไม่ตรงกับ metadata จริง
# ❌ วิธีที่ผิด: ชื่อฟิลด์ไม่ตรง
filters = [FilterCondition("Category", FilterOperator.EQ, "Software")]
✅ วิธีที่ถูก: ตรวจสอบชื่อฟิลด์ให้ตรงกับ metadata
print("ชื่อฟิลด์ที่มี:", list(store.documents[0]["metadata"].keys()))
ผลลัพธ์: ['category', 'product', 'price']
filters = [FilterCondition("category", FilterOperator.EQ, "Software")]
✅ เพิ่ม fallback: ถ้ากรองแล้วว่าง ให้ค้นหาทั้งหมดแล้วแจ้งผู้ใช้
results = store.search(query, filters=filters)
if not results:
results = store.search(query, filters=None)
print("⚠️ ไม่พบเอกสารตรงเงื่อนไข ขยายการค้นหาเป็นทั้งหมด")2. ปัญหา: Latency สูงเกินไป (>200ms)
สาเหตุ: การสร้าง Embedding ทีละข้อความทำให้เกิด Round Trip หลายครั้ง หรือเลือก model ที่ใหญ่เกินไป
# ❌ วิธีที่ผิด: สร้าง embedding ทีละข้อความในลูป
for doc in many_documents:
embedding = self.rag.create_embedding(doc["content"]) # เพิ่ม latency
✅ วิธีที่ถูก: Batch embedding - ส่งหลายข้อความพร้อมกัน
def batch_create_embeddings(self, texts: List[str],
batch_size: int = 100) -> List[List[float]]:
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
response = requests.post(
f"{self.base_url}/embeddings",
headers=self.headers,
json={"input": batch, "model": "text-embedding-3-small"}
)
embeddings = [item["embedding"] for item in response.json()["data"]]
all_embeddings.extend(embeddings)
return all_embeddings
✅ เปลี่ยน model เป็น embedding model ที่เล็กลงแต่เร็วขึ้น
embedding = self.rag.create_embedding(text, model="text-embedding-3-small")
แทน model="text-embedding-3-large"
3. ปัญหา: RAG ให้คำตอบที่ไม่ตรงกับบริบท (Hallucination)
สาเหตุ: ไม่ได้กรองเอกสารที่ไม่เกี่ยวข้องออกก่อนส่งให้ LLM หรือใช้ค่า Temperature สูงเกินไป
# ❌ วิธีที่ผิด: ใช้ temperature สูงทำให้ LLM แต่งเติมข้อมูล
answer = self.rag.chat_completion(messages, temperature=0.9)
✅ วิธีที่ถูก: กรองเอกสารด้วย similarity threshold
def search_with_threshold(self, query: str,
similarity_threshold: float = 0.7) -> List[Dict]:
results = self.search(query, top_k=10)
# กรองเฉพาะเอกสารที่มีความคล้ายคลึงสูงพอ
filtered = [r for r in results if r["similarity_score"] >= similarity_threshold]
if not filtered:
# ถ้าไม่มีผลลัพธ์ที่ดีพอ แจ้งผู้ใช้
raise ValueError("ไม่พบเอกสารที่เกี่ยวข้องเพียงพอ กรุณาปรับคำถาม")
return filtered
✅ ลด temperature เพื่อความแม่นยำ
answer = self.rag.chat_completion(messages, model="gpt-4o-mini", temperature=0.2)
✅ เพิ่ม citation ในคำตอบ
def generate_with_citation(self, user_question: str, results: List[Dict]) -> str:
# ... ค้นหาเอกสาร ...
# สร้างคำตอบพร้อมอ้างอิงแหล่งที่มา
answer = self.rag.chat_completion(messages, temperature=0.1)
# เพิ่ม footnote แหล่งที่มา
sources = "\n\n📚 แหล่งอ้างอิง:\n"
for i, doc in enumerate(results, 1):
sources += f"[{i}] {doc['id']} - {doc['metadata']['product']}\n"
return answer + sourcesสรุปและข้อแนะนำ
การใช้ Metadata Filtering ในระบบ RAG ช่วยให้เราสามารถค้นหาข้อมูลได้แม่นยำมากขึ้น ลดการใช้ Token โดยไม่จำเป็น และปรับปรุงคุณภาพคำตอบจาก LLM ได้อย่างมีนัยสำคัญ โดยปัจจัยสำคัญที่ควรพิจารณาคือ การออกแบบโครงสร้าง Metadata ให้ครอบคลุม การเลือกใช้ Operator การ