การติดตามและประเมินคุณภาพ Embedding ถือเป็นหัวใจสำคัญของระบบ Vector Database ที่มีประสิทธิภาพ ในบทความนี้เราจะมาดูวิธีการตรวจสอบคุณภาพ Embedding และการตรวจจับความผิดปกติโดยละเอียด พร้อมตัวอย่างโค้ดที่สามารถนำไปใช้งานได้จริงกับ HolySheep AI

ทำไมต้องมีระบบ Monitoring สำหรับ Embedding

Embedding Vector ที่มีคุณภาพต่ำจะส่งผลกระทบโดยตรงต่อความแม่นยำในการค้นหาและการจับคู่ความหมาย โดยปัญหาที่พบบ่อย ได้แก่:

การติดตั้งระบบ Monitoring พื้นฐาน

เริ่มจากการติดตั้งไลบรารีที่จำเป็นและสร้างโค้ดสำหรับเก็บ Metrics ของ Embedding

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from collections import defaultdict
import time
from datetime import datetime

class EmbeddingMonitor:
    def __init__(self, base_url="https://api.holysheep.ai/v1", api_key="YOUR_HOLYSHEEP_API_KEY"):
        self.base_url = base_url
        self.api_key = api_key
        self.metrics_history = []
        self.drift_threshold = 0.15
        self.outlier_threshold = 3.0
    
    def get_embedding(self, text, model="text-embedding-3-small"):
        """เรียกใช้ HolySheep API เพื่อสร้าง Embedding"""
        import requests
        
        start_time = time.time()
        try:
            response = requests.post(
                f"{self.base_url}/embeddings",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "input": text,
                    "model": model
                }
            )
            latency = (time.time() - start_time) * 1000
            
            if response.status_code == 200:
                data = response.json()
                return {
                    "embedding": np.array(data["data"][0]["embedding"]),
                    "latency_ms": latency,
                    "success": True,
                    "model": model,
                    "timestamp": datetime.now().isoformat()
                }
            else:
                return {
                    "embedding": None,
                    "latency_ms": latency,
                    "success": False,
                    "error": response.text,
                    "timestamp": datetime.now().isoformat()
                }
        except Exception as e:
            return {
                "embedding": None,
                "latency_ms": (time.time() - start_time) * 1000,
                "success": False,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }
    
    def calculate_consistency_score(self, embeddings_list):
        """คำนวณ Consistency Score จาก Embedding หลายตัวของข้อความเดียวกัน"""
        if len(embeddings_list) < 2:
            return None
        
        similarities = []
        base_embedding = embeddings_list[0]
        
        for emb in embeddings_list[1:]:
            sim = cosine_similarity([base_embedding], [emb])[0][0]
            similarities.append(sim)
        
        return np.mean(similarities)
    
    def detect_outliers(self, embeddings):
        """ตรวจจับ Outlier โดยใช้ Z-Score"""
        norms = np.linalg.norm(embeddings, axis=1)
        mean_norm = np.mean(norms)
        std_norm = np.std(norms)
        
        if std_norm == 0:
            return []
        
        z_scores = np.abs((norms - mean_norm) / std_norm)
        outlier_indices = np.where(z_scores > self.outlier_threshold)[0]
        
        return outlier_indices.tolist()
    
    def calculate_drift(self, old_embeddings, new_embeddings):
        """คำนวณ Distribution Drift ระหว่างชุดข้อมูลเก่าและใหม่"""
        if len(old_embeddings) == 0 or len(new_embeddings) == 0:
            return None
        
        old_centroid = np.mean(old_embeddings, axis=0)
        new_centroid = np.mean(new_embeddings, axis=0)
        
        centroid_shift = np.linalg.norm(new_centroid - old_centroid)
        avg_norm = np.linalg.norm(old_centroid)
        
        drift_ratio = centroid_shift / avg_norm if avg_norm > 0 else 0
        
        return {
            "drift_ratio": drift_ratio,
            "is_anomaly": drift_ratio > self.drift_threshold,
            "centroid_shift": centroid_shift
        }

monitor = EmbeddingMonitor()
print("ระบบ Monitoring พร้อมใช้งานแล้ว")

การประเมินคุณภาพ Embedding แบบครอบคลุม

นอกจากการตรวจจับ Outlier แล้ว เรายังต้องประเมินคุณภาพ Embedding ในหลายมิติเพื่อให้เข้าใจภาพรวมของระบบ

class EmbeddingQualityEvaluator:
    def __init__(self):
        self.evaluation_results = []
    
    def evaluate_retrieval_accuracy(self, query_embedding, relevant_docs, irrelevant_docs, top_k=5):
        """ประเมิน Retrieval Accuracy โดยเช็คว่าเอกสารที่เกี่ยวข้องถูกดึงมากี่อันดับแรก"""
        relevant_scores = []
        irrelevant_scores = []
        
        for doc_emb in relevant_docs:
            relevant_scores.append(cosine_similarity([query_embedding], [doc_emb])[0][0])
        
        for doc_emb in irrelevant_docs:
            irrelevant_scores.append(cosine_similarity([query_embedding], [doc_emb])[0][0])
        
        avg_relevant = np.mean(relevant_scores)
        avg_irrelevant = np.mean(irrelevant_scores)
        
        mrr = 0
        sorted_docs = relevant_docs + irrelevant_docs
        for i, doc in enumerate(sorted_docs):
            if doc in relevant_docs:
                mrr = 1 / (i + 1)
                break
        
        return {
            "avg_relevant_score": avg_relevant,
            "avg_irrelevant_score": avg_irrelevant,
            "separation_score": avg_relevant - avg_irrelevant,
            "mrr": mrr
        }
    
    def evaluate_semantic_clustering(self, embeddings, labels):
        """ประเมินคุณภาพของ Semantic Clustering"""
        from sklearn.metrics import silhouette_score
        
        if len(np.unique(labels)) < 2:
            return None
        
        try:
            score = silhouette_score(embeddings, labels)
            return {
                "silhouette_score": score,
                "quality": "ดีเยี่ยม" if score > 0.7 else "ดี" if score > 0.5 else "พอใช้" if score > 0.3 else "ต้องปรับปรุง"
            }
        except Exception as e:
            return {"error": str(e)}
    
    def generate_quality_report(self, monitor_results):
        """สร้างรายงานคุณภาพโดยรวม"""
        total_requests = len(monitor_results)
        successful_requests = sum(1 for r in monitor_results if r["success"])
        success_rate = successful_requests / total_requests if total_requests > 0 else 0
        
        latencies = [r["latency_ms"] for r in monitor_results if r["success"]]
        avg_latency = np.mean(latencies) if latencies else 0
        p95_latency = np.percentile(latencies, 95) if latencies else 0
        
        return {
            "total_requests": total_requests,
            "success_rate": success_rate,
            "avg_latency_ms": avg_latency,
            "p95_latency_ms": p95_latency,
            "overall_score": success_rate * 0.4 + (1 - avg_latency/1000) * 0.3 + (1 - p95_latency/1000) * 0.3
        }

evaluator = EmbeddingQualityEvaluator()
print("ตัวประเมินคุณภาพพร้อมใช้งาน")

ระบบ Alerting สำหรับ Anomaly Detection

เมื่อพบความผิดปกติ ระบบควรแจ้งเตือนทันทีเพื่อให้ทีมสามารถตอบสนองได้อย่างรวดเร็ว

import json

class AnomalyAlerter:
    def __init__(self, alert_config=None):
        self.alert_config = alert_config or {
            "latency_threshold_ms": 500,
            "success_rate_threshold": 0.95,
            "drift_threshold": 0.15,
            "outlier_ratio_threshold": 0.05
        }
        self.alert_history = []
    
    def check_anomalies(self, quality_report, drift_result, outlier_count, total_count):
        """ตรวจสอบทุกเงื่อนไขและส่ง Alert หากพบความผิดปกติ"""
        alerts = []
        
        if quality_report["avg_latency_ms"] > self.alert_config["latency_threshold_ms"]:
            alerts.append({
                "type": "HIGH_LATENCY",
                "severity": "WARNING",
                "message": f"Latency สูงเกิน {quality_report['avg_latency_ms']:.2f}ms",
                "current_value": quality_report["avg_latency_ms"]
            })
        
        if quality_report["success_rate"] < self.alert_config["success_rate_threshold"]:
            alerts.append({
                "type": "LOW_SUCCESS_RATE",
                "severity": "CRITICAL",
                "message": f"Success Rate ต่ำกว่าเกณฑ์ {quality_report['success_rate']:.2%}",
                "current_value": quality_report["success_rate"]
            })
        
        if drift_result and drift_result["is_anomaly"]:
            alerts.append({
                "type": "DISTRIBUTION_DRIFT",
                "severity": "CRITICAL",
                "message": f"พบ Distribution Drift ที่ drift_ratio={drift_result['drift_ratio']:.4f}",
                "current_value": drift_result["drift_ratio"]
            })
        
        outlier_ratio = outlier_count / total_count if total_count > 0 else 0
        if outlier_ratio > self.alert_config["outlier_ratio_threshold"]:
            alerts.append({
                "type": "HIGH_OUTLIER_RATIO",
                "severity": "WARNING",
                "message": f"Outlier Ratio สูง {outlier_ratio:.2%}",
                "current_value": outlier_ratio
            })
        
        self.alert_history.extend(alerts)
        return alerts
    
    def get_alert_summary(self):
        """สรุป Alert ทั้งหมดในรูปแบบ JSON"""
        return json.dumps({
            "total_alerts": len(self.alert_history),
            "critical_count": sum(1 for a in self.alert_history if a["severity"] == "CRITICAL"),
            "warning_count": sum(1 for a in self.alert_history if a["severity"] == "WARNING"),
            "recent_alerts": self.alert_history[-10:]
        }, indent=2)

alerter = AnomalyAlerter()
print("ระบบ Alerting พร้อมใช้งาน")

การทดสอบระบบแบบ End-to-End

# ตัวอย่างการใช้งานระบบ Monitoring ทั้งหมด
def run_monitoring_demo():
    monitor = EmbeddingMonitor()
    evaluator = EmbeddingQualityEvaluator()
    alerter = AnomalyAlerter()
    
    # 1. ทดสอบสร้าง Embedding
    test_texts = [
        "แมวกินปลา",
        "หมาวิ่งเล่น",
        "นกบินในฟ้า",
        "ปลาว่ายน้ำในทะเล",
        "สุนัขเล่นลูกบอล"
    ]
    
    embeddings = []
    monitor_results = []
    
    for text in test_texts:
        result = monitor.get_embedding(text)
        monitor_results.append(result)
        if result["success"]:
            embeddings.append(result["embedding"])
            print(f"✓ {text} - Latency: {result['latency_ms']:.2f}ms")
        else:
            print(f"✗ {text} - Error: {result.get('error', 'Unknown')}")
    
    # 2. ตรวจจับ Outliers
    outliers = monitor.detect_outliers(np.array(embeddings))
    print(f"\nพบ Outlier {len(outliers)} ตัว: {outliers}")
    
    # 3. คำนวณ Drift (สมมติมีข้อมูลเก่า)
    old_embeddings = embeddings[:2]
    new_embeddings = embeddings[2:]
    drift_result = monitor.calculate_drift(old_embeddings, new_embeddings)
    print(f"Drift Analysis: {drift_result}")
    
    # 4. สร้าง Quality Report
    quality_report = evaluator.generate_quality_report(monitor_results)
    print(f"\n=== Quality Report ===")
    print(f"Success Rate: {quality_report['success_rate']:.2%}")
    print(f"Avg Latency: {quality_report['avg_latency_ms']:.2f}ms")
    print(f"P95 Latency: {quality_report['p95_latency_ms']:.2f}ms")
    print(f"Overall Score: {quality_report['overall_score']:.2f}")
    
    # 5. ตรวจสอบ Anomalies
    alerts = alerter.check_anomalies(
        quality_report, 
        drift_result, 
        len(outliers), 
        len(embeddings)
    )
    
    if alerts:
        print(f"\n⚠️ พบ {len(alerts)} Alerts:")
        for alert in alerts:
            print(f"  [{alert['severity']}] {alert['message']}")
    else:
        print("\n✓ ไม่พบความผิดปกติ")
    
    return {
        "monitor_results": monitor_results,
        "embeddings": embeddings,
        "quality_report": quality_report,
        "alerts": alerts
    }


รัน Demo

result = run_monitoring_demo()
print("\n=== Alert Summary ===")
print(alerter.get_alert_summary())

ข้อมูลราคาและความสามารถของโมเดล Embedding

เมื่อใช้งานกับ HolySheep AI คุณจะได้รับประโยชน์จาก: