การพัฒนาเครื่องมือค้นหาที่เข้าใจภาษามนุษย์เป็นความท้าทายที่นักพัฒนาหลายคนต้องเผชิญ ในบทความนี้เราจะมาเรียนรู้วิธีสร้างระบบค้นหาภาษาธรรมชาติโดยใช้ Large Language Model หรือ LLM เพื่อให้ผู้ใช้สามารถค้นหาข้อมูลได้อย่างเป็นธรรมชาติเหมือนกับการถามคนจริงๆ
เปรียบเทียบบริการ AI API ยอดนิยม
ก่อนจะเริ่มพัฒนา เรามาดูการเปรียบเทียบค่าใช้จ่ายและความสามารถของบริการต่างๆ กัน โดยเฉพาะ HolySheep AI ที่มีความคุ้มค่าสูงสุดสำหรับนักพัฒนาไทย
| บริการ | ราคา ($/MTok) | ความหน่วง (ms) | วิธีการชำระเงิน | เครดิตฟรี |
|---|---|---|---|---|
| HolySheep AI | $0.42 - $15 | <50 | WeChat, Alipay, บัตร | ✓ มีเมื่อลงทะเบียน |
| API อย่างเป็นทางการ | $2.50 - $60 | 100-300 | บัตรเครดิตเท่านั้น | ✗ ไม่มี |
| บริการรีเลย์อื่นๆ | $1.50 - $30 | 80-200 | หลากหลาย | ขึ้นอยู่กับผู้ให้บริการ |
จะเห็นได้ว่า HolySheep AI มีความได้เปรียบเรื่องราคาที่ประหยัดได้ถึง 85% เมื่อเทียบกับ API อย่างเป็นทางการ พร้อมทั้งความหน่วงที่ต่ำกว่า 50 มิลลิวินาที ทำให้เหมาะสำหรับการพัฒนาเครื่องมือค้นหาที่ต้องการความเร็วสูง
หลักการทำงานของ Semantic Search
การค้นหาภาษาธรรมชาติแตกต่างจากการค้นหาแบบดั้งเดิมที่ใช้ Keyword Matching โดยระบบจะทำความเข้าใจความหมายของคำถาม แทนที่จะแค่จับคู่คำที่ตรงกันเท่านั้น ซึ่งทำให้ผู้ใช้สามารถถามคำถามได้หลากหลายรูปแบบและยังคงได้ผลลัพธ์ที่ถูกต้อง
การตั้งค่าโปรเจกต์และติดตั้ง Dependencies
เริ่มต้นด้วยการสร้างโปรเจกต์ Python และติดตั้งไลบรารีที่จำเป็นสำหรับการพัฒนาระบบค้นหา
pip install openai tiktoken numpy scikit-learn fastapi uvicornการสร้าง Client สำหรับเชื่อมต่อกับ LLM API
ส่วนสำคัญที่สุดคือการตั้งค่า Client สำหรับเชื่อมต่อกับ API ซึ่งเราจะใช้ HolySheep AI เป็นตัวอย่างเนื่องจากมีราคาที่คุ้มค่าและความหน่วงต่ำ
import openai
from typing import List, Dict, Optional
class LLMClient:
"""Client สำหรับเชื่อมต่อกับ LLM API ผ่าน HolySheep"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.client = openai.OpenAI(
api_key=api_key,
base_url=base_url
)
def create_embedding(self, text: str, model: str = "text-embedding-3-small") -> List[float]:
"""สร้าง Embedding จากข้อความ"""
response = self.client.embeddings.create(
model=model,
input=text
)
return response.data[0].embedding
def chat_completion(
self,
messages: List[Dict],
model: str = "gpt-4o",
temperature: float = 0.7
) -> str:
"""ส่งข้อความไปยัง LLM และรับคำตอบ"""
response = self.client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature
)
return response.choices[0].message.content
ตัวอย่างการใช้งาน
api_key = "YOUR_HOLYSHEEP_API_KEY" # แทนที่ด้วย API Key ของคุณ
llm_client = LLMClient(api_key=api_key)การสร้างระบบ Semantic Search Engine
ต่อไปจะเป็นการสร้างคลาสหลักสำหรับระบบค้นหา โดยจะประกอบด้วยฟังก์ชันสำหรับสร้าง Embedding ของเอกสาร การค้นหาด้วย Cosine Similarity และการใช้ LLM ตอบคำถาม
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class SemanticSearchEngine:
"""ระบบค้นหาภาษาธรรมชาติโดยใช้ LLM"""
def __init__(self, llm_client: LLMClient):
self.llm_client = llm_client
self.documents: List[Dict] = []
self.embeddings: Optional[np.ndarray] = None
def index_documents(self, docs: List[Dict[str, str]]):
"""ทำดัชนีเอกสารทั้งหมด"""
self.documents = docs
texts = [doc["content"] for doc in docs]
# สร้าง Embedding สำหรับแต่ละเอกสาร
self.embeddings = np.array([
self.llm_client.create_embedding(text) for text in texts
])
print(f"✓ ทำดัชนีเอกสาร {len(docs)} ฉบับเสร็จสิ้น")
def search(self, query: str, top_k: int = 5) -> List[Dict]:
"""ค้นหาเอกสารที่เกี่ยวข้องที่สุด"""
# สร้าง Embedding ของคำถาม
query_embedding = np.array([
self.llm_client.create_embedding(query)
])
# คำนวณ Cosine Similarity
similarities = cosine_similarity(query_embedding, self.embeddings)[0]
# เรียงลำดับตามความคล้ายคลึง
top_indices = np.argsort(similarities)[::-1][:top_k]
results = []
for idx in top_indices:
results.append({
"document": self.documents[idx],
"score": float(similarities[idx])
})
return results
def answer_question(self, question: str) -> str:
"""ใช้ LLM ตอบคำถามโดยอ้างอิงจากเอกสารที่ค้นหาเจอ"""
# ค้นหาเอกสารที่เกี่ยวข้อง
relevant_docs = self.search(question, top_k=3)
# สร้าง Context จากเอกสารที่พบ
context = "\n\n".join([
f"[เอกสาร {i+1}]: {doc['document']['content']}"
for i, doc in enumerate(relevant_docs)
])
# สร้าง Prompt สำหรับ LLM
messages = [
{
"role": "system",
"content": "คุณเป็นผู้ช่วยตอบคำถามโดยอ้างอิงจากเอกสารที่ให้มา หากไม่แน่ใจให้ตอบว่าไม่ทราบ"
},
{
"role": "user",
"content": f"เอกสารที่เกี่ยวข้อง:\n{context}\n\nคำถาม: {question}"
}
]
return self.llm_client.chat_completion(messages, model="gpt-4o")
ตัวอย่างการใช้งาน
engine = SemanticSearchEngine(llm_client)
เพิ่มเอกสารตัวอย่าง
sample_docs = [
{"title": "การติดตั้ง Python", "content": "Python สามารถติดตั้งได้จาก python.org โดยดาวน์โหลดตัวติดตั้งและรัน"},
{"title": "ตัวแปรใน Python", "content": "ตัวแปรใน Python ไม่ต้องประกาศชนิดข้อมูล เช่น x = 10"},
{"title": "ฟังก์ชันใน Python", "content": "ฟังก์ชันสร้างด้วย def เช่น def hello(): print('สวัสดี')"}
]
engine.index_documents(sample_docs)
ค้นหา
result = engine.answer_question("วิธีสร้างตัวแปรใน Python เป็นอย่างไร?")
print(result)การปรับปรุงประสิทธิภาพด้วย Batching
สำหรับการทำดัชนีเอกสารจำนวนมาก การใช้ Batching จะช่วยลดเวลาและค่าใช้จ่ายได้อย่างมาก โดย HolySheep AI รองรับการสร้าง Embedding หลายตัวพร้อมกันในคำขอเดียว
from concurrent.futures import ThreadPoolExecutor
import time
class OptimizedSearchEngine(SemanticSearchEngine):
"""ระบบค้นหาที่ปรับปรุงประสิทธิภาพแล้ว"""
def index_documents_batched(self, docs: List[Dict[str, str]], batch_size: int = 100):
"""ทำดัชนีเอกสารแบบ Batch เพื่อประหยัดเวลาและค่าใช้จ่าย"""
self.documents = docs
texts = [doc["content"] for doc in docs]
all_embeddings = []
total_batches = (len(texts) + batch_size - 1) // batch_size
start_time = time.time()
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
batch_num = i // batch_size + 1
# สร้าง Embedding หลายตัวในคำขอเดียว (Batching)
response = self.llm_client.client.embeddings.create(
model="text-embedding-3-small",
input=batch
)
batch_embeddings = [item.embedding for item in response.data]
all_embeddings.extend(batch_embeddings)
elapsed = time.time() - start_time
print(f"✓ Batch {batch_num}/{total_batches} เสร็จสิ้น (ใช้เวลา {elapsed:.1f}s)")
self.embeddings = np.array(all_embeddings)
print(f"✓ ทำดัชนี {len(docs)} ฉบับ เสร็จสิ้นใน {time.time() - start_time:.1f}s")
def index_parallel(self, docs: List[Dict[str, str]], max_workers: int = 5):
"""ทำดัชนีเอกสารแบบขนานเพื่อความเร็วสูงสุด"""
self.documents = docs
def get_embedding(text: str) -> np.ndarray:
return np.array(self.llm_client.create_embedding(text))
start_time = time.time()
with ThreadPoolExecutor(max_workers=max_workers) as executor:
self.embeddings = np.array(
list(executor.map(get_embedding, [doc["content"] for doc in docs]))
)
print(f"✓ ทำดัชนีแบบขนาน {len(docs)} ฉบับ เสร็จสิ้นใน {time.time() - start_time:.2f}s")
ทดสอบกับเอกสาร 1000 ฉบับ
large_dataset = [
{"title": f"เอกสาร {i}", "content": f"เนื้อหาตัวอย่าง {i} " * 50}
for i in range(1000)
]
optimized_engine = OptimizedSearchEngine(llm_client)
optimized_engine.index_documents_batched(large_dataset, batch_size=50)การสร้าง API Server สำหรับ Search Engine
เพื่อให้ Search Engine พร้อมใช้งานจริง เราจะสร้าง API Server โดยใช้ FastAPI ซึ่งจะรองรับการค้นหาและการตอบคำถามผ่าน HTTP Endpoint
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn
app = FastAPI(title="AI Semantic Search API")
สร้าง Search Engine Instance (ใช้ Singleton Pattern)
search_engine: Optional[SemanticSearchEngine] = None
class Document(BaseModel):
title: str
content: str
class SearchRequest(BaseModel):
query: str
top_k: int = 5
class QuestionRequest(BaseModel):
question: str
model: str = "gpt-4o"
@app.on_event("startup")
async def startup_event():
global search_engine
api_key = "YOUR_HOLYSHEEP_API_KEY"
llm_client = LLMClient(api_key=api_key)
search_engine = SemanticSearchEngine(llm_client)
# โหลดเอกสารตัวอย่าง
sample_docs = [
{"title": "HolySheep AI", "content": "บริการ AI API ราคาประหยัด ความหน่วงต่ำ"},
{"title": "การใช้งาน API", "content": "สมัครบัญชีแล้วนำ API Key มาใช้งาน"}
]
search_engine.index_documents(sample_docs)
print("✓ Search Engine พร้อมใช้งาน")
@app.post("/search")
async def search(request: SearchRequest):
if search_engine is None:
raise HTTPException(status_code=503, detail="Search Engine ยังไม่พร้อมใช้งาน")
results = search_engine.search(request.query, request.top_k)
return {"results": results}
@app.post("/answer")
async def answer_question(request: QuestionRequest):
if search_engine is None:
raise HTTPException(status_code=503, detail="Search Engine ยังไม่พร้อมใช้งาน")
answer = search_engine.answer_question(
request.question,
model=request.model
)
return {"answer": answer}
@app.post("/index")
async def index_documents(documents: List[Document]):
if search_engine is None:
raise HTTPException(status_code=503, detail="Search Engine ยังไม่พร้อมใช้งาน")
docs = [{"title": d.title, "content": d.content} for d in documents]
search_engine.index_documents(docs)
return {"message": f"ทำดัชนีเอกสาร {len(documents)} ฉบับเสร็จสิ้น"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. ข้อผิดพลาด AuthenticationError: Invalid API Key
สาเหตุ: API Key ไม่ถูกต้องหรือหมดอายุ
# ❌ วิธีที่ผิด - Key ไม่ถูกต้อง
client = openai.OpenAI(
api_key="sk-wrong-key",
base_url="https://api.holysheep.ai/v1"
)
✅ วิธีที่ถูกต้อง - ตรวจสอบ Key และใช้ Environment Variable
import os
from dotenv import load_dotenv
load_dotenv() # โหลด .env file
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError("กรุณาตั้งค่า HOLYSHEEP_API_KEY ในไฟล์ .env")
client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)วิธีแก้ไข: ตรวจสอบว่าได้สมัครสมาชิกและคัดลอก API Key ที่ถูกต้องจาก หน้าผู้ใช้งาน พร้อมทั้งเก็บ Key ไว้ใน Environment Variable แทนการเขียนตรงในโค้ด
2. ข้อผิดพลาด RateLimitError: Too Many Requests
สาเหตุ: ส่งคำขอมากเกินกว่าขีดจำกัดที่กำหนด
import time
from functools import wraps
def rate_limit(max_requests: int = 60, period: int = 60):
"""Decorator สำหรับจำกัดจำนวนคำขอ"""
min_interval = period / max_requests
last_called = [0.0]
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() - last_called[0]
if elapsed < min_interval:
time.sleep(min_interval - elapsed)
last_called[0] = time.time()
return func(*args, **kwargs)
return wrapper
return decorator
class RateLimitedClient(LLMClient):
"""Client ที่มีการจำกัดอัตราคำขออัตโนมัติ"""
@rate_limit(max_requests=50, period=60) # สูงสุด 50 คำขอต่อนาที
def create_embedding(self, text: str):
return super().create_embedding(text)
@rate_limit(max_requests=20, period=60)
def chat_completion(self, messages, model="gpt-4o", temperature=0.7):
return super().chat_completion(messages, model, temperature)
หรือใช้ Exponential Backoff สำหรับ Retry
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
def safe_embedding_call(client, text):
"""เรียก API พร้อม Retry อัตโนมัติเมื่อเกิด Rate Limit"""
return client.create_embedding(text)วิธีแก้ไข: ใช้ Rate Limiting หรือ Exponential Backoff เพื่อหลีกเลี่ยงการถูกจำกัด รวมถึงพิจารณาใช้ Batching เพื่อรวมคำขอหลายคำให้เป็นคำขอเดียว
3. ข้อผิดพลาด BadRequestError: Invalid Input - Text Too Long
สาเหตุ: ข้อความที่ส่งไปยาวเกินขีดจำกัดของ Model
import tiktoken
class TextProcessor:
"""ประมวลผลข้อความให้เหมาะสมก่อนส่งไปยัง API"""
def __init__(self, model: str = "gpt-4o"):
self.encoding = tiktoken.encoding_for_model(model)
self.max_tokens = {
"gpt-4o": 128000,
"gpt-4-turbo": 128000,
"gpt-3.5-turbo": 16385
}
self.model = model
def count_tokens(self, text: str) -> int:
"""นับจำนวน Token ในข้อความ"""
return len(self.encoding.encode(text))
def truncate_text(self, text: str, max_tokens: int) -> str:
"""ตัดข้อความให้เหลือตามจำนวน Token ที่กำหนด"""
tokens = self.encoding.encode(text)
if len(tokens) <= max_tokens:
return text
truncated_tokens = tokens[:max_tokens]
return self.encoding.decode(truncated_tokens)
def prepare_for_embedding(self, text: str, max_chars: int = 8000) -> str:
"""เตรียมข้อความสำหรับสร้าง Embedding"""
# ตัดข้อความถ้ายาวเกินไป (ประมาณ 8000 ตัวอักษร ≈ 10000 tokens)
if len(text) > max_chars:
text = text[:max_chars] + "..."
return text
วิธีใช้งาน
processor = TextProcessor(model="gpt-4o")
ตรวจสอบก่อนส่ง
long_text = "ข้อความยาวมาก..." * 1000
safe_text = processor.prepare_for_embedding(long_text)
embedding = llm_client.create_embedding(safe_text)
print(f"✓ Token ที่ใช้: {processor.count_tokens(safe_text)}")วิธีแก้ไข: ใช้ TextProcessor เพื่อตรวจสอบและตัดข้อความให้เหมาะสมก่อนส่งไปยัง API รวมถึงแบ่งเอกสารยาวเป็นส่วนๆ แล้วค่อยสร้าง Embedding แต่ละส่วน
4. ข้อผิดพลาด TimeoutError: Connection Timeout
สาเหตุ: เครือข่ายช้าหรือ API ไม่ตอบสนอง
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retry(retries: int = 3, backoff_factor: float = 0.5):
"""สร้าง Session ที่มีการ Retry อัตโนมัติ"""
session = requests.Session()
retry_strategy = Retry(
total=retries,
backoff_factor=backoff_factor,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST", "GET"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
class RobustLLMClient(LLMClient):
"""Client ที่มีความทนทานต่อข้อผิดพลาดเครือข่าย"""
def __init__(self, api_key: str, timeout: int = 60):
super().__init__(api_key)
self.timeout = timeout
self.session = create_session_with_retry()
def create_embedding(self, text: str, model: str = "text-embedding-3-small"):
try:
response = self.session.post(
f"{self.client.base_url}embeddings",
headers={"Authorization": f"Bearer {self.client.api_key}"},
json={"model": model, "input": text},
timeout=self.timeout
)
response.raise_for_status()
return response.json()["data"][0]["embedding"]
except requests.exceptions.Timeout:
print(f"⚠️ Connection Timeout - ลองใช้ Embedding Cache แทน")
return self._get_cached_embedding(text)
ใช้งาน
robust_client = RobustLLMClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
timeout=120 # เพิ่ม timeout เป็น 120 วินาที
)วิธีแก้ไข: เพิ่มค่า Timeout และใช้การ Retry อัตโนมัติ รวมถึงพิจารณาใช้ Caching สำหรับข้อความที่ถูกถามบ่อยๆ เพื่อลดการเรียก API ซ้ำ
สรุป
การพัฒนาเครื่องมือค