Prompt Reliability Scoring & SLAs — Mục tiêu: Define SLA for prompt success, monitor QoS

Chào anh em, hôm nay mình muốn chia sẻ về một chủ đề hơi “meta” một chút — Prompt Reliability Scoring & SLAs. Nghe có vẻ abstract nhưng thực ra nó rất practical, đặc biệt nếu bạn đang build hệ thống AI tương tác với người dùng.

1. Vì sao cần Prompt Reliability Scoring?

Khi làm việc với LLM, bạn sẽ nhận ra một điều đau lòng: kết quả trả về không phải lúc nào cũng giống nhau dù cùng một prompt. Điều này khác hoàn toàn với các API truyền thống — gọi 10 lần, nhận 10 kết quả giống hệt nhau.

Vấn đề: Bạn không thể đảm bảo 100% prompt sẽ trả về đúng format, đúng thông tin cần thiết.

Đó là lý do chúng ta cần Reliability Scoring — đo lường mức độ tin cậy của kết quả prompt trước khi đưa nó vào hệ thống.

2. Các thành phần của Prompt Reliability Scoring

2.1. Metrics cần đo lường

2.2. Công thức tổng hợp Reliability Score

\huge Reliability\_Score = \frac{FC \times 0.4 + SA \times 0.4 + (1 - ER) \times 0.2}{1}

Giải thích:
- Format Compliance và Semantic Accuracy mỗi cái chiếm 40% vì đây là 2 yếu tố quan trọng nhất
- Error Rate chiếm 20% vì một prompt trả về đúng format nhưng bị lỗi timeout vẫn chưa thể dùng được

3. Define SLA cho Prompt Success

3.1. Các tiêu chí SLA

3.2. SLA Dashboard (Ví dụ thực tế)

┌─────────────────────────────────────────────────────────────┐
│               PROMPT RELIABILITY DASHBOARD                 │
├─────────────────────────────────────────────────────────────┤
│ Availability:   99.73%   ┌──────────────────────────────┐ │
│ Response Time:  1.8s     │    Latency Distribution     │ │
│ Success Rate:   87.2%    │    P50: 1.2s                │ │
│ Error Rate:     1.3%     │    P95: 1.8s                │ │
│                    ┌─────────────────────────────────────┤ │
│                    │    Reliability Score Trend         │ │
│                    │    Last 24h: 0.872                 │ │
│                    │    Last 7d: 0.865                  │ │
│                    └─────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

4. Monitor QoS (Quality of Service)

4.1. Cơ chế giám sát

import time
from typing import Dict, Any, Tuple
from dataclasses import dataclass
from enum import Enum

class PromptStatus(Enum):
    SUCCESS = "success"
    FORMAT_ERROR = "format_error"
    SEMANTIC_ERROR = "semantic_error"
    TIMEOUT = "timeout"
    UNKNOWN = "unknown"

@dataclass
class PromptResult:
    status: PromptStatus
    response: str
    latency: float
    timestamp: float
    metadata: Dict[str, Any] = None

class PromptMonitor:
    def __init__(self, max_latency: float = 5.0):
        self.max_latency = max_latency
        self.results: list[PromptResult] = []
        self.window_size = 1000  # Số request trong window

    def record(self, result: PromptResult):
        """Ghi nhận kết quả prompt"""
        self.results.append(result)
        # Giữ chỉ window_size kết quả gần nhất
        if len(self.results) > self.window_size:
            self.results = self.results[-self.window_size:]

    def calculate_metrics(self) -> Dict[str, float]:
        """Tính toán các metrics"""
        if not self.results:
            return {
                "format_compliance": 0.0,
                "semantic_accuracy": 0.0,
                "avg_latency": 0.0,
                "error_rate": 1.0
            }

        total = len(self.results)
        format_ok = sum(1 for r in self.results if r.status in [PromptStatus.SUCCESS])
        semantic_ok = sum(1 for r in self.results if r.status == PromptStatus.SUCCESS)
        errors = sum(1 for r in self.results if r.status != PromptStatus.SUCCESS)
        total_latency = sum(r.latency for r in self.results)

        return {
            "format_compliance": format_ok / total,
            "semantic_accuracy": semantic_ok / total,
            "avg_latency": total_latency / total,
            "error_rate": errors / total
        }

    def calculate_reliability_score(self) -> float:
        """Tính toán Reliability Score"""
        metrics = self.calculate_metrics()
        fc = metrics["format_compliance"]
        sa = metrics["semantic_accuracy"]
        er = metrics["error_rate"]

        # Công thức Reliability Score
        score = (fc * 0.4 + sa * 0.4 + (1 - er) * 0.2)
        return round(score, 3)

4.2. Alerting Mechanism

class AlertManager:
    def __init__(self, thresholds: Dict[str, float]):
        self.thresholds = thresholds
        self.alerts = []

    def check_alerts(self, metrics: Dict[str, float]) -> list[str]:
        """Kiểm tra các alert"""
        new_alerts = []

        if metrics["reliability_score"] < self.thresholds["min_reliability"]:
            new_alerts.append(f"Reliability score too low: {metrics['reliability_score']} < {self.thresholds['min_reliability']}")

        if metrics["error_rate"] > self.thresholds["max_error_rate"]:
            new_alerts.append(f"Error rate too high: {metrics['error_rate']} > {self.thresholds['max_error_rate']}")

        if metrics["avg_latency"] > self.thresholds["max_latency"]:
            new_alerts.append(f"Average latency too high: {metrics['avg_latency']} > {self.thresholds['max_latency']}")

        self.alerts.extend(new_alerts)
        return new_alerts

5. Use Case kỹ thuật: Hệ thống Chatbot Support

Giả sử bạn đang build hệ thống chatbot cho customer support, xử lý 10,000 request/giờ.

5.1. Kiến trúc hệ thống

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   User Query    │    │ Prompt Engine   │    │   Response      │
│   (Text)        │───▶│   (LLM)         │───▶│   Validator     │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                            │                       │
                            ▼                       ▼
                   ┌─────────────────┐    ┌─────────────────┐
                   │   Monitor       │    │   Fallback      │
                   │   (QoS)         │    │   (Rule-based)  │
                   └─────────────────┘    └─────────────────┘

5.2. Flow xử lý

1. User gửi query → Prompt Engine tạo prompt
2. Gọi LLM API → Thu thập metrics (latency, status)
3. Validate response → Check format + semantic
4. Monitor ghi nhận → Cập nhật dashboard
5. Nếu reliability score < 0.7 → Fallback sang rule-based system

5.3. Code xử lý thực tế

import openai
from typing import Optional
import time
from pydantic import BaseModel, validator

class ChatResponse(BaseModel):
    answer: str
    confidence: float
    metadata: dict

    @validator("answer")
    def check_format(cls, v):
        """Validate format"""
        if not v or len(v.strip()) == 0:
            raise ValueError("Answer cannot be empty")
        return v

class PromptEngine:
    def __init__(self, openai_client, monitor: PromptMonitor):
        self.client = openai_client
        self.monitor = monitor

    def generate_response(self, user_query: str) -> Optional[ChatResponse]:
        """Tạo response với monitoring"""
        start_time = time.time()

        try:
            # Gọi LLM API
            response = self.client.chat.completions.create(
                model="gpt-4",
                messages=[{"role": "user", "content": user_query}],
                max_tokens=1000,
                temperature=0.7
            )

            latency = time.time() - start_time
            content = response.choices[0].message.content

            # Validate format
            try:
                parsed = ChatResponse(answer=content, confidence=0.8, metadata={})
                status = PromptStatus.SUCCESS
            except Exception as e:
                status = PromptStatus.FORMAT_ERROR
                parsed = None

            # Ghi nhận kết quả
            result = PromptResult(
                status=status,
                response=content,
                latency=latency,
                timestamp=time.time()
            )
            self.monitor.record(result)

            # Tính toán reliability score
            score = self.monitor.calculate_reliability_score()

            # Nếu score quá thấp, dùng fallback
            if score < 0.7:
                return self.fallback_response(user_query)

            return parsed

        except Exception as e:
            latency = time.time() - start_time
            result = PromptResult(
                status=PromptStatus.UNKNOWN,
                response="",
                latency=latency,
                timestamp=time.time()
            )
            self.monitor.record(result)
            return None

    def fallback_response(self, query: str) -> ChatResponse:
        """Fallback response (rule-based)"""
        return ChatResponse(
            answer=f"Rất tiếc, hiện tại chúng tôi đang gặp sự cố. Vui lòng thử lại sau.",
            confidence=0.0,
            metadata={"fallback": True}
        )

6. So sánh các giải pháp giám sát

7. Best Practices

⚡ Always set timeout — Không bao giờ để prompt treo vô thời hạn.

🛡️ Validate everything — Đừng tin vào kết quả từ LLM, luôn validate format và semantic.

📊 Monitor in production — Không có monitoring, bạn đang bay mù.

🔁 Implement fallback — Luôn có kế hoạch B khi LLM không hoạt động.

8. Tổng kết (Key Takeaways)

1. Reliability Scoring là cần thiết — Không thể đảm bảo 100% prompt thành công.
2. SLA cần cụ thể hóa — Dùng metrics đo lường được như latency, error rate, reliability score.
3. Monitor + Alerting — Giám sát QoS giúp phát hiện vấn đề trước khi user phàn nàn.

9. Thảo luận

Anh em đã từng gặp trường hợp prompt trả về kết quả không nhất quán chưa? Giải quyết thế nào? Chia sẻ kinh nghiệm ở phần comment nhé!

Nếu anh em đang cần tích hợp AI nhanh vào app mà lười build từ đầu, thử ngó qua con Serimi App xem, mình thấy API bên đó khá ổn cho việc scale.