การสร้างระบบค้นหาภายในองค์กรที่เข้าใจความหมายของผู้ใช้งานมากกว่าแค่คำที่พิมพ์ คือกุญแจสำคัญสู่การเข้าถึงข้อมูลอย่างรวดเร็วและแม่นยำ ในบทความนี้เราจะพาคุณสร้าง Knowledge Base Search ที่ใช้งานจริงในระดับ Production โดยใช้ HolySheep AI Embedding API ซึ่งมีค่าบริการที่ประหยัดกว่า 85% เมื่อเทียบกับบริการอื่นในตลาด พร้อม Latency ต่ำกว่า 50ms
สถาปัตยกรรมระบบโดยรวม
ระบบของเราประกอบด้วย 4 ส่วนหลักที่ทำงานร่วมกัน:
- Document Ingestion Pipeline — รับเอกสารจากหลายแหล่ง ทำความสะอาด และแบ่งเป็น chunks
- Embedding Generation Service — แปลงข้อความเป็น Vector 128 มิติ หรือ 1536 มิติ
- Vector Database — จัดเก็บและทำ Index สำหรับการค้นหาแบบ Approximate Nearest Neighbor
- Query Engine — รับคำถาม ค้นหา และจัดลำดับผลลัพธ์
ทำความเข้าใจ Embedding และ Semantic Search
Embedding คือการแปลงข้อความให้เป็นตัวเลขใน Vector Space โดยที่ข้อความที่มีความหมายคล้ายกันจะอยู่ใกล้กันในปริภูมิหลายมิติ เมื่อผู้ใช้ค้นหาด้วยคำถาม ระบบจะแปลงคำถามนั้นเป็น Vector เดียวกัน แล้วค้นหาเอกสารที่ใกล้เคียงที่สุด
// ตัวอย่าง Concept: ความสัมพันธ์ของ Vectors ใน Semantic Space
//
// คำว่า "รายงานการเงิน" และ "งบการเงิน" มี Embedding ใกล้กัน
// คำว่า "รายงานการเงิน" และ "อาหารการกิน" มี Embedding ไกลกัน
//
// Vector Space (2D representation สำหรับ visualization):
//
// Financial Terms Food Terms
// [0.8, 0.9] [0.1, 0.2]
// * รายงานการเงิน * อาหารเช้า
// * งบการเงิน * อาหารกลางวัน
//
// origin [0, 0]การตั้งค่า Environment และ Dependencies
# requirements.txt
openai==1.12.0
faiss-cpu==1.7.4
pypdf2==3.0.1
python-dotenv==1.0.0
tiktoken==0.5.2
numpy==1.26.3
pydantic==2.5.3
สร้างไฟล์ .env
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
EMBEDDING_MODEL=text-embedding-3-small
CHUNK_SIZE=500
CHUNK_OVERLAP=50
ติดตั้ง dependencies
pip install -r requirements.txtCore Implementation: Document Processing Pipeline
import os
import hashlib
from typing import List, Optional
from dotenv import load_dotenv
from openai import OpenAI
import tiktoken
load_dotenv()
class HolySheepEmbedding:
"""Client สำหรับ HolySheep AI Embedding API"""
def __init__(self, api_key: str = None):
self.client = OpenAI(
api_key=api_key or os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1" # URL ที่ถูกต้องสำหรับ HolySheep
)
self.encoder = tiktoken.get_encoding("cl100k_base")
def get_embedding(self, text: str, model: str = "text-embedding-3-small") -> List[float]:
"""สร้าง Embedding vector สำหรับข้อความเดียว"""
# ตัดข้อความให้เหมาะสม (max 8192 tokens สำหรับ text-embedding-3-small)
tokens = self.encoder.encode(text)
if len(tokens) > 8000:
text = self.encoder.decode(tokens[:8000])
response = self.client.embeddings.create(
model=model,
input=text
)
return response.data[0].embedding
def get_embeddings_batch(self, texts: List[str], model: str = "text-embedding-3-small") -> List[List[float]]:
"""สร้าง Embedding vectors หลายรายการพร้อมกัน (Batch Processing)"""
# จำกัด batch size เพื่อประสิทธิภาพ
batch_size = 100
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
# ตัดข้อความที่ยาวเกิน
cleaned_batch = []
for text in batch:
tokens = self.encoder.encode(text)
if len(tokens) > 8000:
text = self.encoder.decode(tokens[:8000])
cleaned_batch.append(text)
response = self.client.embeddings.create(
model=model,
input=cleaned_batch
)
all_embeddings.extend([item.embedding for item in response.data])
return all_embeddings
class DocumentChunker:
"""แบ่งเอกสารออกเป็น chunks สำหรับการทำ Embedding"""
def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.encoder = tiktoken.get_encoding("cl100k_base")
def chunk_text(self, text: str, metadata: dict = None) -> List[dict]:
"""แบ่งข้อความออกเป็น chunks พร้อม metadata"""
tokens = self.encoder.encode(text)
chunks = []
for i in range(0, len(tokens), self.chunk_size - self.chunk_overlap):
chunk_tokens = tokens[i:i + self.chunk_size]
chunk_text = self.encoder.decode(chunk_tokens)
chunk_id = hashlib.md5(f"{text[:50]}_{i}".encode()).hexdigest()
chunks.append({
"id": chunk_id,
"text": chunk_text,
"metadata": {
**(metadata or {}),
"char_start": len(self.encoder.decode(tokens[:i])),
"char_end": len(self.encoder.decode(tokens[:i + self.chunk_size])),
"token_count": len(chunk_tokens)
}
})
return chunks
def chunk_document(self, content: str, source: str, doc_metadata: dict = None) -> List[dict]:
"""ประมวลผลเอกสารเต็ม: แบ่ง chunks และเพิ่ม metadata"""
base_metadata = {
"source": source,
**(doc_metadata or {})
}
# ทำความสะอาดข้อความ
cleaned_content = self._clean_text(content)
# แบ่งเป็น chunks
chunks = self.chunk_text(cleaned_content, base_metadata)
return chunks
def _clean_text(self, text: str) -> str:
"""ทำความสะอาดข้อความ"""
import re
# ลบ whitespaces ที่ซ้ำซ้อน
text = re.sub(r'\s+', ' ', text)
# ลบ special characters ที่ไม่จำเป็น
text = re.sub(r'[\x00-\x08\x0b-\x0c\x0e-\x1f\x7f-\x9f]', '', text)
return text.strip()Vector Database Implementation ด้วย FAISS
import faiss
import numpy as np
from typing import List, Tuple, Optional
from dataclasses import dataclass
import json
import os
@dataclass
class SearchResult:
"""ผลลัพธ์การค้นหา"""
id: str
text: str
metadata: dict
score: float
class KnowledgeBaseVectorStore:
"""จัดการ Vector Database สำหรับ Knowledge Base"""
def __init__(self, dimension: int = 1536, index_type: str = "Flat"):
self.dimension = dimension
self.index_type = index_type
self.index = None
self.documents = {} # id -> {text, metadata}
self._initialize_index()
def _initialize_index(self):
"""สร้าง FAISS Index ตามประเภทที่กำหนด"""
if self.index_type == "Flat":
# Exact search - ช้าแต่แม่นยำ 100%
self.index = faiss.IndexFlatIP(self.dimension)
elif self.index_type == "IVF":
# Approximate search - เร็วกว่าสำหรับ dataset ใหญ่
quantizer = faiss.IndexFlatIP(self.dimension)
self.index = faiss.IndexIVFFlat(quantizer, self.dimension, 100)
elif self.index_type == "HNSW":
# Hierarchical NSW - สมดุลระหว่างความเร็วและความแม่นยำ
self.index = faiss.IndexHNSWFlat(self.dimension, 32)
else:
raise ValueError(f"Unknown index type: {self.index_type}")
def add_documents(self, embeddings: List[List[float]], chunks: List[dict]):
"""เพิ่มเอกสารและ embeddings ลงใน index"""
if len(embeddings) != len(chunks):
raise ValueError("จำนวน embeddings และ chunks ไม่ตรงกัน")
# Normalize vectors สำหรับ cosine similarity
vectors = np.array(embeddings).astype('float32')
faiss.normalize_L2(vectors)
# Train index (จำเป็นสำหรับ IVF)
if self.index_type == "IVF" and not self.index.is_trained:
self.index.train(vectors)
# เพิ่มลงใน index
self.index.add(vectors)
# เก็บ documents metadata
for chunk in chunks:
self.documents[chunk["id"]] = {
"text": chunk["text"],
"metadata": chunk["metadata"]
}
def search(self, query_embedding: List[float], top_k: int = 5) -> List[SearchResult]:
"""ค้นหาเอกสารที่เกี่ยวข้องมากที่สุด"""
query_vector = np.array([query_embedding]).astype('float32')
faiss.normalize_L2(query_vector)
# ค้นหา
scores, indices = self.index.search(query_vector, top_k)
results = []
for score, idx in zip(scores[0], indices[0]):
if