บทนำ: เหตุการณ์จริงที่ทำให้ต้องมีระบบมอนิเตอร์

วันที่ 15 มีนาคม 2024 เวลา 03:47 น. ระบบ RAG ของทีมเราเริ่มส่งคืนข้อมูลที่ไม่เกี่ยวข้องอย่างเห็นได้ชัด ผู้ใช้ถามเกี่ยวกับ "การสมัครงาน" แต่ระบบตอบเรื่อง "การลงทะเบียนเข้างาน" แทน ปัญหานี้เกิดจาก embedding model ที่ทำงานผิดพลาดหลังจาก deployment ใหม่ แต่ไม่มีใครสังเกตเห็นจนกระทั่งลูกค้าส่ง email ตอบกลับมาตุญเตือน หลังจากเหตุการณ์นั้น เราจึงพัฒนาระบบ Agentic RAG monitoring ที่สามารถตรวจจับความผิดปกติได้ภายใน 30 วินาที บทความนี้จะอธิบายวิธีการสร้างระบบ monitoring และ alerting สำหรับ Agentic RAG ตั้งแต่พื้นฐานจนถึงการ implement จริง พร้อมโค้ดที่ใช้งานได้ทันที โดยเราจะใช้ HolySheep AI เป็น LLM backend หลักเนื่องจากมี latency ต่ำกว่า 50ms และราคาประหยัดกว่า 85% เมื่อเทียบกับ OpenAI

Agentic RAG คืออะไร และทำไมต้องมีการมอนิเตอร์

Agentic RAG คือระบบ Retrieval-Augmented Generation ที่ทำงานแบบ autonomous agent หมายความว่าระบบสามารถตัดสินใจเองว่าจะค้นหาข้อมูลจากที่ใด ใช้ query แบบใด และประมวลผลอย่างไร ต่างจาก RAG แบบดั้งเดิมที่มีขั้นตอนตายตัว การทำงานของ Agentic RAG ประกอบด้วยหลาย stage ที่อาจเกิดความผิดพลาดได้ ได้แก่:

สถาปัตยกรรมระบบ Monitoring และ Alerting

ระบบ monitoring ที่ดีต้องครอบคลุม 3 ระดับ:

1. Metric Collection Layer

เก็บข้อมูลพื้นฐานจากทุก stage ของ pipeline ได้แก่ latency, token usage, retrieval score, rerank score และ response quality

2. Anomaly Detection Layer

วิเคราะห์ metrics ที่เก็บได้เพื่อตรวจจับความผิดปกติ โดยใช้ทั้ง rule-based และ ML-based detection

3. Alerting Layer

ส่งการแจ้งเตือนผ่านช่องทางต่างๆ เมื่อตรวจพบความผิดปกติ พร้อมระดับความรุนแรงที่แตกต่างกัน 
import requests
import numpy as np
from datetime import datetime
from collections import deque
from dataclasses import dataclass
from typing import List, Dict, Optional

@dataclass
class RAGMetrics:
    timestamp: datetime
    query_id: str
    stage: str
    latency_ms: float
    retrieval_score: float
    rerank_score: float
    token_count: int
    context_relevance: float
    answer_faithfulness: float

class RecallAnomalyDetector:
    """
    ระบบตรวจจับความผิดปกติสำหรับ Agentic RAG
    ใช้ statistical methods และ rule-based detection
    """
    
    def __init__(self, 
                 window_size: int = 100,
                 z_score_threshold: float = 2.5,
                 min_baseline_samples: int = 50):
        self.window_size = window_size
        self.z_score_threshold = z_score_threshold
        self.min_baseline_samples = min_baseline_samples
        
        # Rolling window สำหรับเก็บ metrics
        self.metrics_history: deque = deque(maxlen=window_size)
        self.baseline_calculated = False
        self.baseline_stats = {}
        
    def add_metric(self, metric: RAGMetrics):
        """เพิ่ม metric ใหม่เข้าไปใน history"""
        self.metrics_history.append(metric)
        
        # คำนวณ baseline หลังจากมี sample เพียงพอ
        if len(self.metrics_history) >= self.min_baseline_samples and not self.baseline_calculated:
            self._calculate_baseline()
            
    def _calculate_baseline(self):
        """คำนวณ baseline statistics จาก historical data"""
        retrieval_scores = [m.retrieval_score for m in self.metrics_history]
        rerank_scores = [m.rerank_score for m in self.metrics_history]
        latencies = [m.latency_ms for m in self.metrics_history]
        context_relevances = [m.context_relevance for m in self.metrics_history]
        
        self.baseline_stats = {
            'retrieval_mean': np.mean(retrieval_scores),
            'retrieval_std': np.std(retrieval_scores),
            'rerank_mean': np.mean(rerank_scores),
            'rerank_std': np.std(rerank_scores),
            'latency_mean': np.mean(latencies),
            'latency_std': np.std(latencies),
            'context_relevance_mean': np.mean(context_relevances),
            'context_relevance_std': np.std(context_relevances)
        }
        self.baseline_calculated = True
        
    def detect_anomaly(self, metric: RAGMetrics) -> Dict:
        """ตรวจจับความผิดปกติจาก metric ใหม่"""
        if not self.baseline_calculated:
            return {
                'is_anomaly': False,
                'reason': 'insufficient_baseline_data',
                'severity': 'none'
            }
        
        anomalies = []
        severity = 'none'
        
        # 1. ตรวจจับ retrieval score ต่ำผิดปกติ
        z_retrieval = self._z_score(metric.retrieval_score, 
                                     self.baseline_stats['retrieval_mean'],
                                     self.baseline_stats['retrieval_std'])
        if abs(z_retrieval) > self.z_score_threshold:
            anomalies.append(f'retrieval_score_z={z_retrieval:.2f}')
            severity = self._max_severity(severity, 'high')
            
        # 2. ตรวจจับ rerank score ต่ำผิดปกติ
        z_rerank = self._z_score(metric.rerank_score,
                                 self.baseline_stats['rerank_mean'],
                                 self.baseline_stats['rerank_std'])
        if abs(z_rerank) > self.z_score_threshold:
            anomalies.append(f'rerank_score_z={z_rerank:.2f}')
            severity = self._max_severity(severity, 'medium')
            
        # 3. ตรวจจับ latency สูงผิดปกติ
        z_latency = self._z_score(metric.latency_ms,
                                  self.baseline_stats['latency_mean'],
                                  self.baseline_stats['latency_std'])
        if z_latency > self.z_score_threshold:
            anomalies.append(f'latency_z={z_latency:.2f}')
            severity = self._max_severity(severity, 'medium')
            
        # 4. ตรวจจับ context relevance ต่ำ
        z_context = self._z_score(metric.context_relevance,
                                   self.baseline_stats['context_relevance_mean'],
                                   self.baseline_stats['context_relevance_std'])
        if z_context < -self.z_score_threshold:
            anomalies.append(f'context_relevance_z={z_context:.2f}')
            severity = self._max_severity(severity, 'high')
            
        # 5. Rule-based: retrieval score < 0.3 ถือว่าผิดปกติเสมอ
        if metric.retrieval_score < 0.3:
            anomalies.append('hard_rule:retrieval_below_0.3')
            severity = self._max_severity(severity, 'critical')
            
        return {
            'is_anomaly': len(anomalies) > 0,
            'anomalies': anomalies,
            'severity': severity,
            'z_scores': {
                'retrieval': z_retrieval,
                'rerank': z_rerank,
                'latency': z_latency,
                'context_relevance': z_context
            }
        }
    
    def _z_score(self, value: float, mean: float, std: float) -> float:
        """คำนวณ z-score"""
        if std == 0:
            return 0.0
        return (value - mean) / std
    
    def _max_severity(self, current: str, new: str) -> str:
        """เลือก severity ที่รุนแรงกว่า"""
        order = {'none': 0, 'low': 1, 'medium': 2, 'high': 3, 'critical': 4}
        return new if order.get(new, 0) > order.get(current, 0) else current


ตัวอย่างการใช้งาน

if __name__ == "__main__":
    detector = RecallAnomalyDetector(window_size=200, z_score_threshold=2.5)
    
    # สร้าง baseline data (ปกติ)
    for i in range(100):
        metric = RAGMetrics(
            timestamp=datetime.now(),
            query_id=f"base_{i}",
            stage="retrieval",
            latency_ms=150 + np.random.normal(0, 20),
            retrieval_score=0.75 + np.random.normal(0, 0.1),
            rerank_score=0.82 + np.random.normal(0, 0.08),
            token_count=500 + int(np.random.normal(0, 50)),
            context_relevance=0.88 + np.random.normal(0, 0.05),
            answer_faithfulness=0.90 + np.random.normal(0, 0.03)
        )
        detector.add_metric(metric)
    
    # ทดสอบกับ anomaly
    anomaly_metric = RAGMetrics(
        timestamp=datetime.now(),
        query_id="test_anomaly",
        stage="retrieval",
        latency_ms=800,  # สูงผิดปกติ
        retrieval_score=0.15,  # ต่ำผิดปกติ
        rerank_score=0.45,
        token_count=500,
        context_relevance=0.20,  # ต่ำมาก
        answer_faithfulness=0.55
    )
    
    result = detector.detect_anomaly(anomaly_metric)
    print(f"Anomaly Detection Result: {result}")
    # Output: {'is_anomaly': True, 'anomalies': ['retrieval_score_z=-6.12', ...], 'severity': 'critical'}

การสร้าง Alerting System ที่ครอบคลุม

ระบบ alerting ที่ดีต้องมีหลายระดับความรุนแรงและหลายช่องทางการแจ้งเตือน เพื่อให้ทีมสามารถตอบสนองได้อย่างเหมาะสมกับระดับปัญหา 
import asyncio
import json
from enum import Enum
from typing import Callable, Dict, List
from dataclasses import dataclass, field
from datetime import datetime
import aiohttp
from collections import defaultdict

class AlertSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"

class AlertChannel(Enum):
    SLACK = "slack"
    PAGERDUTY = "pagerduty"
    EMAIL = "email"
    WEBHOOK = "webhook"
    CONSOLE = "console"

@dataclass
class Alert:
    alert_id: str
    timestamp: datetime
    severity: AlertSeverity
    title: str
    message: str
    metrics: Dict = field(default_factory=dict)
    query_id: str = ""
    stage: str = ""
    suggested_action: str = ""

class RAGAlertManager:
    """
    ระบบจัดการ alerts สำหรับ Agentic RAG
    รองรับหลายช่องทางและหลายระดับความรุนแรง
    """
    
    def __init__(self):
        self.handlers: Dict[AlertChannel, List[Callable]] = defaultdict(list)
        self.alert_history: List[Alert] = []
        self.alert_counter = 0
        self.cooldown_periods: Dict[str, datetime] = {}
        
        # Alert rules configuration
        self.rules = {
            'retrieval_score_low': {
                'threshold': 0.3,
                'severity': AlertSeverity.WARNING,
                'cooldown_minutes': 5
            },
            'latency_high': {
                'threshold': 500,  # ms
                'severity': AlertSeverity.ERROR,
                'cooldown_minutes': 2
            },
            'context_relevance_low': {
                'threshold': 0.4,
                'severity': AlertSeverity.WARNING,
                'cooldown_minutes': 10
            },
            'consecutive_failures': {
                'threshold': 5,
                'time_window_minutes': 5,
                'severity': AlertSeverity.CRITICAL,
                'cooldown_minutes': 15
            }
        }
        
        self.failure_tracker: Dict[str, List[datetime]] = defaultdict(list)
        
    def register_handler(self, channel: AlertChannel, handler: Callable):
        """ลงทะเบียน handler สำหรับช่องทางการแจ้งเตือน"""
        self.handlers[channel].append(handler)
        
    async def send_alert(self, alert: Alert):
        """ส่ง alert ไปยังทุก registered channels"""
        # ตรวจสอบ cooldown
        if self._is_in_cooldown(alert):
            return
            
        self.alert_history.append(alert)
        self.alert_counter += 1
        
        for channel, handlers in self.handlers.items():
            for handler in handlers:
                try:
                    await handler(alert)
                except Exception as e:
                    print(f"Error sending alert via {channel}: {e}")
                    
    def _is_in_cooldown(self, alert: Alert) -> bool:
        """ตรวจสอบว่า alert อยู่ในช่วง cooldown หรือไม่"""
        cooldown_key = f"{alert.severity.value}:{alert.title}"
        if cooldown_key in self.cooldown_periods:
            last_alert_time = self.cooldown_periods[cooldown_key]
            from datetime import timedelta
            cooldown = self.rules.get(alert.title, {}).get('cooldown_minutes', 5)
            if datetime.now() - last_alert_time < timedelta(minutes=cooldown):
                return True
        self.cooldown_periods[cooldown_key] = datetime.now()
        return False
    
    def track_failure(self, query_id: str):
        """ติดตามจำนวน failures สำหรับ detection"""
        self.failure_tracker[query_id].append(datetime.now())
        
        # Clean old entries
        cutoff = datetime.now()
        from datetime import timedelta
        self.failure_tracker[query_id] = [
            t for t in self.failure_tracker[query_id]
            if cutoff - t < timedelta(minutes=5)
        ]
        
    async def process_metric_alert(self, metric: RAGMetrics, anomaly_result: Dict):
        """ประมวลผล metric และสร้าง alert หากจำเป็น"""
        if not anomaly_result['is_anomaly']:
            return
            
        severity = AlertSeverity[anomaly_result['severity'].upper()]
        alert = Alert(
            alert_id=f"ALERT-{self.alert_counter + 1:06d}",
            timestamp=datetime.now(),
            severity=severity,
            title="RAG Recall Anomaly Detected",
            message=f"ตรวจพบความผิดปกติใน stage {metric.stage}: {', '.join(anomaly_result['anomalies'])}",
            metrics={
                'retrieval_score': metric.retrieval_score,
                'rerank_score': metric.rerank_score,
                'latency_ms': metric.latency_ms,
                'context_relevance': metric.context_relevance,
                'z_scores': anomaly_result['z_scores']
            },
            query_id=metric.query_id,
            stage=metric.stage,
            suggested_action=self._get_suggested_action(anomaly_result)
        )
        
        await self.send_alert(alert)
        
    def _get_suggested_action(self, anomaly_result: Dict) -> str:
        """แนะนำ action ตามประเภทของ anomaly"""
        actions = []
        if any('retrieval' in a for a in anomaly_result['anomalies']):
            actions.append("ตรวจสอบ embedding model และ vector database")
        if any('latency' in a for a in anomaly_result['anomalies']):
            actions.append("ตรวจสอบ network connectivity และ server load")
        if any('context' in a for a in anomaly_result['anomalies']):
            actions.append("ทำ re-indexing ของ document store")
        return "; ".join(actions)


Handlers สำหรับแต่ละ channel

async def slack_handler(alert: Alert):
    """ส่ง alert ไป Slack"""
    webhook_url = "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"
    
    color_map = {
        AlertSeverity.INFO: "#36a64f",
        AlertSeverity.WARNING: "#ff9800",
        AlertSeverity.ERROR: "#f44336",
        AlertSeverity.CRITICAL: "#9c27b0"
    }
    
    payload = {
        "attachments": [{
            "color": color_map.get(alert.severity, "#808080"),
            "title": f"[{alert.severity.value.upper()}] {alert.title}",
            "text": alert.message,
            "fields": [
                {"title": "Query ID", "value": alert.query_id, "short": True},
                {"title": "Stage", "value": alert.stage, "short": True},
                {"title": "Timestamp", "value": alert.timestamp.isoformat(), "short": True},
                {"title": "Alert ID", "value": alert.alert_id, "short": True}
            ],
            "footer": "Agentic RAG Monitor",
            "ts": alert.timestamp.timestamp()
        }]
    }
    
    if alert.suggested_action:
        payload["attachments"][0]["fields"].append({
            "title": "Suggested Action",
            "value": alert.suggested_action,
            "short": False
        })
    
    async with aiohttp.ClientSession() as session:
        await session.post(webhook_url, json=payload)

async def console_handler(alert: Alert):
    """Handler สำหรับ console output (development)"""
    print(f"\n{'='*60}")
    print(f"🔔 ALERT: {alert.alert_id}")
    print(f"⏰ {alert.timestamp.isoformat()}")
    print(f"⚠️  Severity: {alert.severity.value.upper()}")
    print(f"📌 {alert.title}")
    print(f"💬 {alert.message}")
    if alert.metrics:
        print(f"📊 Metrics: {json.dumps(alert.metrics, indent=2)}")
    if alert.suggested_action:
        print(f"🔧 Suggested Action: {alert.suggested_action}")
    print(f"{'='*60}\n")


ตัวอย่างการใช้งาน

async def main():
    alert_manager = RAGAlertManager()
    
    # ลงทะเบียน handlers
    alert_manager.register_handler(AlertChannel.CONSOLE, console_handler)
    # alert_manager.register_handler(AlertChannel.SLACK, slack_handler)
    
    # สร้าง alert จาก anomaly
    alert = Alert(
        alert_id="ALERT-000001",
        timestamp=datetime.now(),
        severity=AlertSeverity.CRITICAL,
        title="RAG Recall Anomaly Detected",
        message="Retrieval score ต่ำกว่า 0.3 (z-score: -6.12)",
        metrics={
            'retrieval_score': 0.15,
            'z_scores': {'retrieval': -6.12}
        },
        query_id="q-12345",
        stage="retrieval",
        suggested_action="ตรวจสอบ embedding model และ re-index documents"
    )
    
    await alert_manager.send_alert(alert)

if __name__ == "__main__":
    asyncio.run(main())

การ Integrate กับ HolySheep AI API

หลังจากสร้างระบบ monitoring แล้ว ต่อไปจะเป็นการ integrate กับ HolySheep AI สำหรับ LLM backend ซึ่งมี latency ต่ำกว่า 50ms และราคาประหยัดมาก 
import requests
import time
from typing import List, Dict, Optional
from dataclasses import dataclass

@dataclass
class HolySheepRAGIntegration:
    """
    Integration class สำหรับเชื่อมต่อ Agentic RAG กับ HolySheep AI
    base_url: https://api.holysheep.ai/v1
    """
    api_key: str
    base_url: str = "https://api.holysheep.ai/v1"
    model: str = "gpt-4.1"  # $8/MTok - ราคาประหยัด 85%+
    embedding_model: str = "text-embedding-3-small"
    timeout: int = 30
    
    def __post_init__(self):
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
    def generate_with_context(
        self,
        query: str,
        context_documents: List[Dict],
        system_prompt: Optional[str] = None
    ) -> Dict:
        """
        Generate คำตอบด้วย context จาก retrieval results
        พร้อมเก็บ metrics สำหรับ monitoring
        """
        start_time = time.time()
        
        # สร้าง context string
        context_text = "\n\n".join([
            f"[Document {i+1}] {doc.get('content', '')}"
            for i, doc in enumerate(context_documents)
        ])
        
        # เตรียม messages
        if system_prompt is None:
            system_prompt = """คุณเป็น AI assistant ที่ตอบคำถามโดยอ้างอิงจาก context ที่ให้มา
ตอบเฉพาะสิ่งที่มีใน context เท่านั้น หากไม่แน่ใจให้บอกว่าไม่ทราบ"""
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Context:\n{context_text}\n\nQuestion: {query}"}
        ]
        
        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": 0.3,
            "max_tokens": 1000
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=self.headers,
                json=payload,
                timeout=self.timeout
            )
            
            latency_ms = (time.time() - start_time) * 1000
            
            if response.status_code == 200:
                result = response.json()
                return {
                    'success': True,
                    'answer': result['choices'][0]['message']['content'],
                    'usage': result.get('usage', {}),
                    'latency_ms': latency_ms,
                    'model': self.model
                }
            else:
                return {
                    'success': False,
                    'error': f"HTTP {response.status_code}: {response.text}",
                    'latency_ms': latency_ms
                }
                
        except requests.exceptions.Timeout:
            return {
                'success': False,
                'error': 'ConnectionTimeout: Request exceeded timeout',
                'latency_ms': (time.time() - start_time) * 1000
            }
        except requests.exceptions.ConnectionError as e:
            return {
                'success': False,
                'error': f'ConnectionError: {str(e)}',
                'latency_ms': (time.time() - start_time) * 1000
            }
            
    def get_embeddings(self, texts: List[str]) -> Dict:
        """ดึง embeddings สำหรับ retrieval"""
        start_time = time.time()
        
        payload = {
            "model": self.embedding_model,
            "input": texts
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/embeddings",
                headers=self.headers,
                json=payload,
                timeout=self.timeout
            )
            
            latency_ms = (time.time() -

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง