การสร้าง RAG (Retrieval-Augmented Generation) Pipeline ที่มีประสิทธิภาพสูงไม่ใช่เรื่องง่าย ผมเคยเจอสถานการณ์ที่ทำให้หัวหน้าโครงการเรียกประชุมด่วนหลังจากระบบค้นหาข้อมูลให้คำตอบผิดเพี้ยนไปจากความเป็นจริง สาเหตุหลักคือการที่ embedding model ปกติไม่สามารถจับ semantic similarity ได้ดีพอ โดยเฉพาะเมื่อ query มีความซับซ้อน การใช้ Cohere Rerank จึงกลายเป็นทางออกที่ดีที่สุด แต่กว่าจะ integrate สำเร็จ ผมต้องเจอกับข้อผิดพลาดหลายต่อหลายครั้ง
RAG Pipeline พื้นฐานทำงานอย่างไร
ก่อนจะไปถึงการใช้ Rerank เรามาทำความเข้าใจ flow ของ RAG Pipeline กันก่อน ขั้นตอนแรกคือ Indexing โดยเราจะแบ่งเอกสารออกเป็น chunks แล้วส่งแต่ละ chunk ไปสร้าง vector embedding จากนั้นเก็บ vector เหล่านั้นไว้ใน vector database อย่าง ChromaDB หรือ Pinecone ต่อมาคือ Retrieval เมื่อ user ถามคำถาม ระบบจะแปลงคำถามนั้นเป็น vector แล้วค้นหา documents ที่ใกล้เคียงที่สุดด้วย cosine similarity สุดท้ายคือ Generation คือการส่ง retrieved documents พร้อมกับ query ไปให้ LLM ตอบ
ทำไมต้องใช้ Rerank
ปัญหาของ semantic search แบบดั้งเดิมคือมันเป็น just "approximate nearest neighbor search" ซึ่งอาจดึง documents ที่เกี่ยวข้องบางส่วนมาแต่ไม่ใช่ผลลัพธ์ที่ดีที่สุด โดยเฉพาะในกรณีที่ query มี nuance เช่น ถามว่า "วิธีแก้ปัญหา production server down" แต่ได้ผลลัพธ์เป็นเอกสารเกี่ยวกับ development server แทน Rerank model จะทำหน้าที่ re-scoring documents ที่ retrieve มาได้ด้วย cross-encoder ที่เข้าใจความสัมพันธ์ระหว่าง query กับ document อย่างแท้จริง
การติดตั้ง Cohere Rerank ผ่าน HolySheep AI
HolySheep AI เป็น API gateway ที่ให้บริการ Cohere models รวมถึง Rerank ด้วย ข้อดีคือราคาประหยัดมาก เพียง ¥1 ต่อ $1 ซึ่งประหยัดกว่า 85% เมื่อเทียบกับ direct API นอกจากนี้ยังรองรับ WeChat/Alipay, latency ต่ำกว่า 50ms และมีเครดิตฟรีเมื่อลงทะเบียน สามารถสมัครที่นี่ได้เลย
Setup และ Installation
# ติดตั้ง dependencies ที่จำเป็น
pip install cohere chromadb openai python-dotenv
สร้าง .env file
echo "HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY" > .env
เริ่มใช้งาน ChromaDB
import chromadb
chroma_client = chromadb.Client()การสร้าง RAG Pipeline พร้อม Rerank
import cohere
import chromadb
from openai import OpenAI
from dotenv import load_dotenv
import os
โหลด API key
load_dotenv()
cohere_api_key = os.getenv("HOLYSHEEP_API_KEY")
สร้าง Cohere client โดยใช้ HolySheep base URL
cohere_client = cohere.Client(
api_key=cohere_api_key,
base_url="https://api.holysheep.ai/v1"
)
OpenAI client สำหรับ embedding และ generation
openai_client = OpenAI(
api_key=cohere_api_key, # ใช้ API key เดียวกัน
base_url="https://api.holysheep.ai/v1"
)
class RAGRerankPipeline:
def __init__(self, collection_name="documents"):
self.chroma_client = chromadb.Client()
self.collection = self.chroma_client.get_or_create_collection(
name=collection_name
)
self.top_k_initial = 50 # ดึงมาก่อน 50 อัน
self.top_k_final = 10 # หลัง rerank เหลือ 10 อัน
def index_documents(self, documents: list[str], ids: list[str]):
"""Index เอกสารเข้า vector database"""
for i, doc in enumerate(documents):
response = openai_client.embeddings.create(
model="text-embedding-3-small",
input=doc
)
embedding = response.data[0].embedding
self.collection.add(
documents=[doc],
ids=[ids[i]],
embeddings=[embedding]
)
print(f"Indexed {len(documents)} documents")
def retrieve_with_rerank(self, query: str):
"""ค้นหาเอกสารด้วย Rerank"""
# Step 1: สร้าง query embedding
query_response = openai_client.embeddings.create(
model="text-embedding-3-small",
input=query
)
query_embedding = query_response.data[0].embedding
# Step 2: Initial retrieval
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=self.top_k_initial
)
retrieved_docs = results['documents'][0]
retrieved_ids = results['ids'][0]
# Step 3: Rerank ด้วย Cohere
rerank_response = cohere_client.rerank(
model="rerank-multilingual-v2.0",
query=query,
documents=retrieved_docs,
top_n=self.top_k_final,
return_documents=True
)
# Step 4: รวบรวมผลลัพธ์ที่ rerank แล้ว
reranked_results = []
for result in rerank_response.results:
reranked_results.append({
'document': result.document.text,
'id': retrieved_ids[result.index],
'relevance_score': result.relevance_score
})
return reranked_results
def generate_answer(self, query: str, context_docs: list[dict]):
"""สร้างคำตอบด้วย LLM"""
context = "\n\n".join([
f"[Document {i+1}] {doc['document']}"
for i, doc in enumerate(context_docs)
])
prompt = f"""Based on the following context, answer the question.
Context:
{context}
Question: {query}
Answer:"""
response = openai_client.chat.completions.create(
model="gpt-4o-mini", # ราคาเพียง $2.50/MTok บน HolySheep
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
ทดสอบการทำงาน
pipeline = RAGRerankPipeline()
Index ตัวอย่างเอกสาร
sample_docs = [
"วิธีติดตั้ง Docker บน Ubuntu 22.04",
"การแก้ไขปัญหา Kubernetes Pod CrashLoopBackOff",
"สอนใช้งาน Git เบื้องต้น",
"วิธีตั้งค่า Nginx reverse proxy",
"การ deploy Next.js บน Vercel"
]
doc_ids = [f"doc_{i}" for i in range(len(sample_docs))]
pipeline.index_documents(sample_docs, doc_ids)
ทดสอบค้นหาด้วย Rerank
query = "วิธีแก้ปัญหา server ล่ม"
results = pipeline.retrieve_with_rerank(query)
print("\nReranked Results:")
for i, r in enumerate(results):
print(f"{i+1}. Score: {r['relevance_score']:.4f} - {r['document']}")Advanced: Streaming RAG พร้อม Caching
import hashlib
from functools import lru_cache
class OptimizedRAGRerankPipeline(RAGRerankPipeline):
def __init__(self, *args, cache_size=1000, **kwargs):
super().__init__(*args, **kwargs)
self.cache_size = cache_size
self._cache = {}
def _get_cache_key(self, query: str, top_k: int) -> str:
"""สร้าง cache key จาก query และ top_k"""
content = f"{query}:{top_k}"
return hashlib.sha256(content.encode()).hexdigest()
@lru_cache(maxsize=100)
def _cached_embedding(self, text: str) -> list[float]:
"""Cache embedding result"""
response = openai_client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
def batch_rerank(self, queries: list[str], all_docs: list[str]):
"""Rerank multiple queries พร้อมกัน (efficient)"""
# Batch embedding สำหรับทุก query
embeddings = [self._cached_embedding(q) for q in queries]
# Batch retrieve
all_results = []
for emb in embeddings:
results = self.collection.query(
query_embeddings=[emb],
n_results=self.top_k_initial
)
all_results.append(results['documents'][0])
# Batch rerank
rerank_responses = cohere_client.rerank(
model="rerank-multilingual-v2.0",
query=queries, # ส่ง list ของ queries
documents=all_results,
top_n=self.top_k_final
)
return rerank_responses
ใช้งาน batch processing
optimized_pipeline = OptimizedRAGRerankPipeline(cache_size=500)
queries = [
"วิธีติดตั้ง Docker",
"การ deploy web application",
"แก้ไข error 404"
]
all_documents = [
"Docker Installation Guide",
"Nginx Configuration Tutorial",
"React Deployment Best Practices",
"HTTP Error Codes Explained",
"CI/CD Pipeline Setup"
]
results = optimized_pipeline.batch_rerank(queries, all_documents)
for i, resp in enumerate(results):
print(f"\nQuery: {queries[i]}")
for r in resp.results[:3]:
print(f" - {r.document.text} ({r.relevance_score:.3f})")ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. ConnectionError: HTTPSConnectionPool Timeout
อาการ: เกิด error ประเภท ConnectionError: HTTPSConnectionPool(host='api.holysheep.ai', port=443): Max retries exceeded หรือ timeout บ่อยๆ โดยเฉพาะเมื่อมี request จำนวนมาก
สาเหตุ: ปกติเกิดจาก rate limiting หรือ network timeout ที่ไม่เพียงพอ และบางครั้งอาจเกิดจากการใช้ wrong base_url
# ❌ วิธีที่ผิด - timeout too short
cohere_client = cohere.Client(
api_key=api_key,
base_url="https://api.holysheep.ai/v1",
timeout=10 # too short!
)
✅ วิธีที่ถูก - เพิ่ม timeout และ retry logic
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
def rerank_with_retry(client, query, docs, top_n):
try:
response = client.rerank(
model="rerank-multilingual-v2.0",
query=query,
documents=docs,
top_n=top_n,
return_documents=True
)
return response
except Exception as e:
print(f"Rerank failed: {e}, retrying...")
raise
ใช้งาน
cohere_client = cohere.Client(
api_key=api_key,
base_url="https://api.holysheep.ai/v1",
timeout=120 # 2 minutes timeout
)
เรียกใช้ฟังก์ชันพร้อม retry
result = rerank_with_retry(cohere_client, query, documents, 10)2. 401 Unauthorized - Invalid API Key
อาการ: ได้รับ cohere.error.UnauthorizedError: 401 Client Error: Unauthorized ทันทีหลังจากเรียก API
สาเหตุ: API key ไม่ถูกต้อง หรือ key หมดอายุ หรือใช้ key จาก provider อื่นโดยไม่ได้ตั้งค่า base_url ถูกต้อง
# ❌ ผิด - ใช้ key จาก OpenAI แต่เรียก Cohere endpoint
openai_key = "sk-xxxxx" # OpenAI key ไม่ทำงานกับ Cohere!
client = cohere.Client(
api_key=openai_key,
base_url