บทนำ
ในระบบ RAG (Retrieval-Augmented Generation) คุณภาพของการค้นหาเอกสารเป็นรากฐานสำคัญที่ส่งผลต่อความแม่นยำของคำตอบสุดท้าย หากระบบค้นหาดึงเอกสารที่ไม่เกี่ยวข้องมาให้ LLM ตอบ ผลลัพธ์ย่อมไม่น่าเชื่อถือ ในบทความนี้เราจะมาเรียนรู้วิธีการคำนวณเมตริกสำคัญ 3 ตัว ได้แก่ Recall, MRR และ NDCG พร้อมโค้ด Python ที่พร้อมใช้งานจริง
---
กรณีศึกษา: ทีมสตาร์ทอัพ AI ในกรุงเทพฯ
**บริบทธุรกิจ:** ทีมพัฒนาแชทบอทสำหรับลูกค้าบริการประกันภัยรายใหญ่แห่งหนึ่ง มีฐานความรู้กฎเกณฑ์ประกันภัยมากกว่า 50,000 เอกสาร ต้องการระบบ Q&A ที่แม่นยำสำหรับเจ้าหน้าที่ Call Center
**จุดเจ็บปวด:** ระบบ RAG เดิมที่ใช้ผู้ให้บริการ API รายเดิมมี Recall เพียง 65% ทำให้เจ้าหน้าที่ต้องตรวจสอบคำตอบด้วยตนเองบ่อยครั้ง ใช้เวลาตอบลูกค้านานขึ้น และมีความผิดพลาดในการอ้างอิงข้อมูลผิดเอกสาร
**เหตุผลที่เลือก HolySheep AI:** หลังจากทดสอบพบว่า embedding model ของ HolySheep ให้ความหน่วงเฉลี่ยต่ำกว่า 50ms ราคาถูกกว่า 85% เมื่อเทียบกับผู้ให้บริการรายเดิม และรองรับ Chinese/English/Thai multilingual embeddings ได้ดีเยี่ยม ทีมจึงตัดสินใจย้ายมาใช้
HolySheep AI
**ขั้นตอนการย้าย:** ทีมเริ่มจากการเปลี่ยน base_url จากผู้ให้บริการเดิมมาเป็น https://api.holysheep.ai/v1 พร้อมกับหมุนคีย์ API ใหม่ จากนั้นทำ canary deploy 10% → 30% → 100% โดยเฝ้าระวัง Recall และ MRR อย่างใกล้ชิด
**ตัวชี้วัด 30 วันหลังการย้าย:**
- ความหน่วงเฉลี่ย: 420ms → 180ms (ลดลง 57%)
- ค่าบริการรายเดือน: $4,200 → $680 (ประหยัด 84%)
- Recall: 65% → 89%
- MRR: 0.52 → 0.91
---
ทำความเข้าใจเมตริก RAG Retrieval
1. Recall@k — ความครบถ้วนของเอกสารที่เกี่ยวข้อง
Recall@k วัดว่าในเอกสารที่เกี่ยวข้องทั้งหมดที่มีอยู่ในฐานข้อมูล ระบบสามารถดึงมาได้กี่เปอร์เซ็นต์เมื่อดู k ผลลัพธ์แรก
**สูตร:** Recall@k = |Relevant ∩ Retrieved| / |Relevant|
def calculate_recall_at_k(retrieved_ids: list[str], relevant_ids: set[str], k: int) -> float:
"""
คำนวณ Recall@k
retrieved_ids: รายการ ID ของเอกสารที่ระบบดึงมา (เรียงตามลำดับความเกี่ยวข้อง)
relevant_ids: เซ็ตของ ID เอกสารที่ถูกต้องตาม ground truth
k: จำนวนผลลัพธ์ที่ดู
"""
retrieved_k = set(retrieved_ids[:k])
relevant_retrieved = retrieved_k.intersection(relevant_ids)
if len(relevant_ids) == 0:
return 0.0
recall = len(relevant_retrieved) / len(relevant_ids)
return round(recall, 4)
ตัวอย่างการใช้งาน
retrieved = ["doc_A", "doc_B", "doc_C", "doc_D", "doc_E"]
relevant = {"doc_A", "doc_C", "doc_F"}
print(f"Recall@1: {calculate_recall_at_k(retrieved, relevant, 1)}") # 0.0 (doc_A ไม่อยู่ใน relevant)
print(f"Recall@3: {calculate_recall_at_k(retrieved, relevant, 3)}") # 0.6667 (doc_A, doc_C อยู่)
print(f"Recall@5: {calculate_recall_at_k(retrieved, relevant, 5)}") # 0.6667 (doc_F ไม่อยู่ใน retrieved)
2. MRR (Mean Reciprocal Rank) — ความเร็วในการเจอคำตอบแรกที่ถูกต้อง
MRR วัดตำแหน่งเฉลี่ยของเอกสารที่เกี่ยวข้องลำดับแรกที่ระบบดึงมาได้ ค่ายิ่งสูงยิ่งดี (1.0 คือดีที่สุด)
**สูตร:** MRR = (1/Q) × Σ(1/rank_i) โดย rank_i คือตำแหน่งของ relevant document แรก
def calculate_mrr(queries_results: list[dict]) -> float:
"""
คำนวณ MRR สำหรับหลาย queries
queries_results: list of {"retrieved_ids": [...], "relevant_ids": {...}}
"""
reciprocal_ranks = []
for result in queries_results:
retrieved = result["retrieved_ids"]
relevant = result["relevant_ids"]
# หาตำแหน่งของ relevant document แรกที่เจอ
rank = None
for idx, doc_id in enumerate(retrieved, start=1):
if doc_id in relevant:
rank = idx
break
# ใช้ 0 ถ้าไม่เจอ relevant document เลย
rr = 1.0 / rank if rank is not None else 0.0
reciprocal_ranks.append(rr)
mrr = sum(reciprocal_ranks) / len(reciprocal_ranks)
return round(mrr, 4)
ตัวอย่างการใช้งาน
test_queries = [
{
"retrieved_ids": ["doc_X", "doc_Y", "doc_Z"],
"relevant_ids": {"doc_Z"}
},
{
"retrieved_ids": ["doc_A", "doc_B", "doc_C"],
"relevant_ids": {"doc_B", "doc_D"}
},
{
"retrieved_ids": ["doc_1", "doc_2", "doc_3"],
"relevant_ids": {"doc_9"}
}
]
Query 1: doc_Z อยู่ตำแหน่ง 3 -> 1/3
Query 2: doc_B อยู่ตำแหน่ง 2 -> 1/2
Query 3: ไม่เจอ -> 0
MRR = (1/3 + 1/2 + 0) / 3 = 0.8333 / 3 = 0.2778
print(f"MRR: {calculate_mrr(test_queries)}") # 0.2778
3. NDCG@k (Normalized Discounted Cumulative Gain) — คุณภาพการจัดอันดับแบบมีระดับ
NDCG พิจารณาทั้งการมีเอกสารที่เกี่ยวข้อง และลำดับการจัดอันดับ โดยเอกสารที่เกี่ยวข้องมากที่สุดควรอยู่ลำดับบน
def calculate_dcg(scores: list[float], k: int) -> float:
"""คำนวณ DCG (Discounted Cumulative Gain)"""
dcg = 0.0
for i, score in enumerate(scores[:k], start=1):
dcg += score / (math.log2(i + 1)) # i+1 เพราะ log2(1) = 0
return dcg
def calculate_ndcg(retrieved_scores: list[float], ideal_scores: list[float], k: int) -> float:
"""
คำนวณ NDCG@k
retrieved_scores: คะแนนความเกี่ยวข้องของเอกสารที่ดึงมา (เรียงตามลำดับ)
ideal_scores: คะแนนความเกี่ยวข้องที่ดีที่สุด (เรียงลำดับจากมากไปน้อย)
"""
dcg = calculate_dcg(retrieved_scores, k)
idcg = calculate_dcg(ideal_scores, k)
if idcg == 0:
return 0.0
return round(dcg / idcg, 4)
import math
ตัวอย่าง: ระบบดึงเอกสาร 5 ฉบับ โดยมี relevance grades (0-3)
retrieved_scores: คะแนนจริงที่ระบบให้
ideal_scores: คะแนนที่เรียงจากมากไปน้อย
retrieved_scores = [3, 1, 0, 2, 0] # doc1=3, doc2=1, doc3=0, doc4=2, doc5=0
ideal_scores = sorted(retrieved_scores, reverse=True) # [3, 2, 1, 0, 0]
print(f"NDCG@3: {calculate_ndcg(retrieved_scores, ideal_scores, 3)}")
print(f"NDCG@5: {calculate_ndcg(retrieved_scores, ideal_scores, 5)}")
อธิบาย: ถ้าเอกสารคะแนน 3 อยู่ลำดับ 1 (ถูกต้อง) ได้ NDCG สูง
ถ้าเอกสารคะแนน 2 อยู่ลำดับ 4 (ผิด) ได้ NDCG ต่ำ
---
การวัดผล RAG Pipeline แบบครบวงจร
เมื่อนำเมตริกทั้ง 3 มาประกอบกัน จะเห็นภาพครบถ้วนของประสิทธิภาพระบบ
import json
from typing import Optional
class RAGEvaluator:
"""คลาสสำหรับประเมินผล RAG Retrieval แบบครบวงจร"""
def __init__(self):
self.results = []
def add_query_result(
self,
query_id: str,
retrieved_ids: list[str],
relevant_ids: set[str],
retrieved_scores: Optional[list[float]] = None
):
"""เพิ่มผลลัพธ์ของ query หนึ่ง"""
self.results.append({
"query_id": query_id,
"retrieved_ids": retrieved_ids,
"relevant_ids": relevant_ids,
"retrieved_scores": retrieved_scores or [1.0] * len(retrieved_ids)
})
def evaluate_all(self, k_values: list[int] = [1, 3, 5, 10]) -> dict:
"""ประเมินผลทุกเมตริกพร้อมกัน"""
metrics = {"k_values": k_values}
for k in k_values:
recalls = []
mrrs = []
ndcgs = []
for r in self.results:
retrieved = r["retrieved_ids"]
relevant = r["relevant_ids"]
scores = r["retrieved_scores"]
# Recall@k
recall = calculate_recall_at_k(retrieved, relevant, k)
recalls.append(recall)
# MRR (คำนวณครั้งเดียว ไม่ขึ้นกับ k)
if k == k_values[0]:
mrr = self._calc_single_mrr(retrieved, relevant)
mrrs.append(mrr)
# NDCG@k
if len(scores) >= k:
ideal = sorted(scores, reverse=True)
ndcg = calculate_ndcg(scores, ideal, k)
ndcgs.append(ndcg)
metrics[f"recall@{k}"] = round(sum(recalls) / len(recalls), 4)
if k == k_values[0]:
metrics["mrr"] = round(sum(mrrs) / len(mrrs), 4)
metrics[f"ndcg@{k}"] = round(sum(ndcgs) / len(ndcgs), 4)
return metrics
def _calc_single_mrr(self, retrieved: list, relevant: set) -> float:
for idx, doc_id in enumerate(retrieved, 1):
if doc_id in relevant:
return 1.0 / idx
return 0.0
การใช้งาน
evaluator = RAGEvaluator()
เพิ่มผลลัพธ์จากการทดสอบ
evaluator.add_query_result(
query_id="q1",
retrieved_ids=["doc_1", "doc_2", "doc_3", "doc_4", "doc_5"],
relevant_ids={"doc_2", "doc_5"},
retrieved_scores=[3, 2, 1, 0, 2]
)
evaluator.add_query_result(
query_id="q2",
retrieved_ids=["doc_a", "doc_b", "doc_c"],
relevant_ids={"doc_a", "doc_d"},
retrieved_scores=[3, 1, 2]
)
วัดผลทั้งหมด
results = evaluator.evaluate_all(k_values=[1, 3, 5])
print(json.dumps(results, indent=2))
---
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
กรณีที่ 1: ใช้ API endpoint ผิด ทำให้เรียกโมเดลผิดตัว
**ปัญหา:** หลายคนยังลืมเปลี่ยน base_url จากผู้ให้บริการเดิม ทำให้เรียก OpenAI หรือ Anthropic แทน HolySheep
# ❌ ผิด -
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง