Mở Đầu: Câu Chuyện Thực Tế Từ Hệ Thống Thương Mại Điện Tử
Tháng 3 năm 2025, đội ngũ kỹ thuật của một nền tảng thương mại điện tử lớn tại Việt Nam đối mặt với một vấn đề nan giải: chatbot chăm sóc khách hàng AI của họ bắt đầu đưa ra những câu trả lời "ảo" về tình trạng tồn kho. Khách hàng được thông báo sản phẩm "còn hàng" nhưng khi đặt hàng thì nhận được thông báo hết hàng. Chỉ trong 2 tuần, tỷ lệ khiếu nại tăng 340%, và đội ngũ không thể xác định chính xác thời điểm nào model bắt đầu "nói dối."
Đây là lý do tôi bắt đầu nghiên cứu sâu về statistical quality monitoring cho output của AI — một lĩnh vực mà nhiều doanh nghiệp bỏ qua cho đến khi gặp khủng hoảng. Trong bài viết này, tôi sẽ chia sẻ framework thực chiến đã giúp đội ngũ của tôi phát hiện drift trong output của LLM với độ chính xác 94.7% trước khi người dùng cuối nhận thấy vấn đề.
Tại Sao Giám Sát Chất Lượng Output AI Lại Quan Trọng?
Khác với traditional software testing, output của LLM có tính không xác định (non-deterministic). Một prompt giống hệt có thể tạo ra response khác nhau tùy thuộc vào:
- Temperature setting và sampling parameters
- Context window length và conversation history
- Model version và fine-tuning updates
- Server load và API latency variations
- Training data distribution shifts
Với chi phí API call trung bình $0.002-0.015 cho mỗi thousand tokens (sử dụng
HolySheheep AI với giá chỉ từ $0.42/MTok cho DeepSeek V3.2), một hệ thống xử lý 10,000 requests/ngày có thể tiêu tốn $200-1,500/ngày. Nếu 15% trong số đó cho ra output chất lượng kém, doanh nghiệp đang lãng phí $30-225 mỗi ngày cho những tương tác vô giá trị — chưa kể uy tín thương hiệu bị ảnh hưởng.
Framework Giám Sát Chất Lượng 4 Giai Đoạn
Giai Đoạn 1: Baseline Establishment
Trước khi phát hiện anomaly, bạn cần establish baseline từ output "tốt". Đây là phase quan trọng nhất nhưng nhiều team bỏ qua.
Giai Đoạn 2: Real-time Metrics Collection
Thu thập các metrics chính trong thời gian thực, không chỉ response quality mà còn technical indicators.
Giai Đoạn 3: Statistical Anomaly Detection
Áp dụng các statistical tests để xác định khi nào metrics deviated đáng kể khỏi baseline.
Giai Đoạn 4: Automated Intervention
Khi anomaly được detected, trigger automated actions: fallback, alert, hoặc model switching.
Triển Khai Chi Tiết Với HolySheep AI
Dưới đây là implementation hoàn chỉnh sử dụng HolySheep AI API — nền tảng tiết kiệm 85%+ chi phí so với OpenAI với độ trễ trung bình dưới 50ms.
1. Thiết Lập Monitoring Dashboard Cơ Bản
# monitor_quality.py - Production-ready quality monitoring system
import httpx
import asyncio
import statistics
from collections import deque
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
import json
import hashlib
@dataclass
class QualityMetrics:
"""Lưu trữ metrics cho một request"""
timestamp: datetime
response_time_ms: float
token_count: int
cost_usd: float
prompt_length: int
response_length: int
latency_p50: float = 0.0
latency_p95: float = 0.0
latency_p99: float = 0.0
@dataclass
class QualityThresholds:
"""Ngưỡng cảnh báo được configure"""
max_latency_ms: float = 2000.0
max_cost_per_request: float = 0.05
min_tokens_per_second: float = 50.0
max_error_rate: float = 0.05
z_score_threshold: float = 2.5
rolling_window_size: int = 100
class AIServiceMonitor:
"""
Monitor chất lượng output từ HolySheep AI API
với statistical anomaly detection
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
thresholds: Optional[QualityThresholds] = None
):
self.api_key = api_key
self.base_url = base_url
self.thresholds = thresholds or QualityThresholds()
# Rolling windows để track metrics
self.latency_history: deque = deque(maxlen=self.thresholds.rolling_window_size)
self.cost_history: deque = deque(maxlen=self.thresholds.rolling_window_size)
self.token_history: deque = deque(maxlen=self.thresholds.rolling_window_size)
# Statistical tracking
self.metrics_log: List[QualityMetrics] = []
self.anomaly_callbacks: List[Callable] = []
# Latency percentiles tracking
self.recent_latencies: List[float] = []
async def call_with_monitoring(
self,
prompt: str,
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: int = 1000,
system_prompt: Optional[str] = None
) -> Dict:
"""
Gọi HolySheep AI API với full monitoring
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
start_time = datetime.now()
latency_ms = 0.0
error_occurred = False
error_message = None
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
end_time = datetime.now()
latency_ms = (end_time - start_time).total_seconds() * 1000
if response.status_code == 200:
data = response.json()
usage = data.get("usage", {})
completion_tokens = usage.get("completion_tokens", 0)
prompt_tokens = usage.get("prompt_tokens", 0)
total_tokens = usage.get("total_tokens", 0)
# Tính chi phí theo bảng giá HolySheep 2026
cost_map = {
"deepseek-v3.2": 0.00042, # $0.42/MTok
"gpt-4.1": 0.008, # $8/MTok
"claude-sonnet-4.5": 0.015, # $15/MTok
"gemini-2.5-flash": 0.0025 # $2.50/MTok
}
cost_usd = (total_tokens / 1000) * cost_map.get(model, 0.001)
# Lưu metrics
metric = QualityMetrics(
timestamp=end_time,
response_time_ms=latency_ms,
token_count=total_tokens,
cost_usd=cost_usd,
prompt_length=len(prompt),
response_length=len(data["choices"][0]["message"]["content"])
)
self._record_metric(metric)
self._check_anomalies(metric)
return {
"success": True,
"content": data["choices"][0]["message"]["content"],
"metrics": metric,
"latency_ms": latency_ms,
"cost_usd": cost_usd
}
else:
error_occurred = True
error_message = f"HTTP {response.status_code}: {response.text}"
except httpx.TimeoutException:
error_occurred = True
error_message = "Request timeout after 30s"
latency_ms = 30000.0
except Exception as e:
error_occurred = True
error_message = str(e)
latency_ms = 0.0
# Ghi nhận error
metric = QualityMetrics(
timestamp=datetime.now(),
response_time_ms=latency_ms,
token_count=0,
cost_usd=0.0,
prompt_length=len(prompt),
response_length=0
)
self.metrics_log.append(metric)
return {
"success": False,
"error": error_message,
"metrics": metric
}
def _record_metric(self, metric: QualityMetrics):
"""Cập nhật rolling windows và log"""
self.latency_history.append(metric.response_time_ms)
self.cost_history.append(metric.cost_usd)
self.token_history.append(metric.token_count)
self.metrics_log.append(metric)
# Update latency percentiles
self.recent_latencies = list(self.latency_history)
if len(self.recent_latencies) >= 20:
sorted_latencies = sorted(self.recent_latencies)
n = len(sorted_latencies)
metric.latency_p50 = sorted_latencies[int(n * 0.5)]
metric.latency_p95 = sorted_latencies[int(n * 0.95)]
metric.latency_p99 = sorted_latencies[int(n * 0.99)]
def _calculate_z_score(self, value: float, window: deque) -> float:
"""Tính z-score để phát hiện statistical anomaly"""
if len(window) < 10:
return 0.0
mean = statistics.mean(window)
stdev = statistics.stdev(window)
if stdev == 0:
return 0.0
return (value - mean) / stdev
def _check_anomalies(self, metric: QualityMetrics):
"""Kiểm tra và trigger alerts cho anomalies"""
anomalies = []
# Check latency anomaly
z_latency = self._calculate_z_score(
metric.response_time_ms,
self.latency_history[:-1] if len(self.latency_history) > 1 else self.latency_history
)
if abs(z_latency) > self.thresholds.z_score_threshold:
anomalies.append({
"type": "latency_spike",
"z_score": z_latency,
"value_ms": metric.response_time_ms,
"severity": "high" if abs(z_latency) > 3.0 else "medium"
})
# Check cost anomaly (indicates token explosion)
if len(self.cost_history) >= 5:
z_cost = self._calculate_z_score(metric.cost_usd, self.cost_history)
if z_cost > self.thresholds.z_score_threshold:
anomalies.append({
"type": "cost_spike",
"z_score": z_cost,
"value_usd": metric.cost_usd,
"severity": "high"
})
# Check if cost exceeds threshold
if metric.cost_usd > self.thresholds.max_cost_per_request:
anomalies.append({
"type": "cost_threshold_exceeded",
"value_usd": metric.cost_usd,
"threshold": self.thresholds.max_cost_per_request,
"severity": "high"
})
# Trigger callbacks if anomalies detected
if anomalies:
for callback in self.anomaly_callbacks:
try:
callback(anomalies, metric)
except Exception as e:
print(f"Callback error: {e}")
def register_anomaly_callback(self, callback: Callable):
"""Đăng ký function được gọi khi có anomaly"""
self.anomaly_callbacks.append(callback)
def get_statistics_report(self) -> Dict:
"""Generate báo cáo thống kê chi tiết"""
if not self.metrics_log:
return {"error": "No data collected yet"}
latencies = [m.response_time_ms for m in self.metrics_log]
costs = [m.cost_usd for m in self.metrics_log]
tokens = [m.token_count for m in self.metrics_log]
successful_requests = [m for m in self.metrics_log if m.response_length > 0]
failed_requests = len(self.metrics_log) - len(successful_requests)
return {
"summary": {
"total_requests": len(self.metrics_log),
"successful": len(successful_requests),
"failed": failed_requests,
"error_rate": failed_requests / len(self.metrics_log),
"total_cost_usd": sum(costs),
"avg_cost_per_request": statistics.mean(costs) if costs else 0,
"total_tokens": sum(tokens)
},
"latency": {
"mean_ms": statistics.mean(latencies),
"median_ms": statistics.median(latencies),
"stdev_ms": statistics.stdev(latencies) if len(latencies) > 1 else 0,
"min_ms": min(latencies),
"max_ms": max(latencies),
"p95_ms": sorted(latencies)[int(len(latencies) * 0.95)] if len(latencies) >= 20 else max(latencies),
"p99_ms": sorted(latencies)[int(len(latencies) * 0.99)] if len(latencies) >= 100 else max(latencies)
},
"cost_analysis": {
"total_spent_usd": round(sum(costs), 4),
"avg_per_request_usd": round(statistics.mean(costs), 6) if costs else 0,
"projected_daily_cost": round(statistics.mean(costs) * 10000, 2), # 10k requests/day
"potential_savings_percent": self._calculate_potential_savings()
}
}
def _calculate_potential_savings(self) -> float:
"""So sánh chi phí HolySheep vs OpenAI"""
if not self.token_history:
return 0.0
total_tokens = sum(self.token_history)
# HolySheep DeepSeek V3.2
holysheep_cost = (total_tokens / 1000) * 0.42
# OpenAI GPT-4o
openai_cost = (total_tokens / 1000) * 15.0
if openai_cost == 0:
return 0.0
return round((1 - holysheep_cost / openai_cost) * 100, 1)
=== USAGE EXAMPLE ===
async def main():
monitor = AIServiceMonitor(
api_key="YOUR_HOLYSHEEP_API_KEY",
thresholds=QualityThresholds(
max_latency_ms=1500.0,
max_cost_per_request=0.02,
z_score_threshold=2.5
)
)
# Register anomaly callback
def on_anomaly(anomalies, metric):
print(f"🚨 ANOMALY DETECTED at {metric.timestamp}:")
for anomaly in anomalies:
print(f" - {anomaly}")
monitor.register_anomaly_callback(on_anomaly)
# Simulate monitoring for e-commerce product queries
test_prompts = [
"Kiểm tra tồn kho sản phẩm SKU-12345",
"Tư vấn laptop cho lập trình viên",
"So sánh iPhone 15 Pro và Samsung S24",
"Hỏi về chính sách đổi trả 30 ngày",
"Tra cứu đơn hàng #ORD-98765"
]
print("Bắt đầu giám sát chất lượng AI responses...")
for i, prompt in enumerate(test_prompts):
result = await monitor.call_with_monitoring(
prompt=prompt,
model="deepseek-v3.2",
system_prompt="Bạn là trợ lý chăm sóc khách hàng thương mại điện tử. Trả lời ngắn gọn, chính xác."
)
if result["success"]:
print(f"✅ Request {i+1}: {result['latency_ms']:.1f}ms, cost: ${result['cost_usd']:.4f}")
else:
print(f"❌ Request {i+1} failed: {result['error']}")
# Generate report
report = monitor.get_statistics_report()
print("\n" + "="*60)
print("BÁO CÁO CHẤT LƯỢNG AI")
print("="*60)
print(json.dumps(report, indent=2, default=str))
if __name__ == "__main__":
asyncio.run(main())
2. Statistical Process Control cho Output Quality
# statistical_quality_control.py - Advanced SPC implementation
import numpy as np
from scipy import stats
from collections import deque
from dataclasses import dataclass
from typing import List, Tuple, Optional
import math
@dataclass
class ControlLimits:
"""Ngưỡng kiểm soát SPC"""
ucl: float # Upper Control Limit
lcl: float # Lower Control Limit
cl: float # Center Line (mean)
sigma: float
@property
def ucl_2sigma(self) -> float:
return self.cl + 2 * self.sigma
@property
def lcl_2sigma(self) -> float:
return self.cl - 2 * self.sigma
@property
def ucl_3sigma(self) -> float:
return self.cl + 3 * self.sigma
@property
def lcl_3sigma(self) -> float:
return self.cl - 3 * self.sigma
@dataclass
class SPCViolation:
"""Ghi nhận vi phạm kiểm soát chất lượng"""
timestamp: float
value: float
violation_type: str
severity: str
details: dict
class StatisticalQualityControl:
"""
Statistical Process Control cho AI model outputs
Phát hiện drift sử dụng Shewhart rules và CUSUM
"""
def __init__(
self,
window_size: int = 50,
target_value: float = 0.0,
control_multiplier: float = 3.0
):
self.window_size = window_size
self.target_value = target_value
self.control_multiplier = control_multiplier
# Data storage
self.observations: deque = deque(maxlen=window_size * 2)
self.timestamps: deque = deque(maxlen=window_size * 2)
# Control limits
self.control_limits: Optional[ControlLimits] = None
# Violations tracking
self.violations: List[SPCViolation] = []
# CUSUM parameters
self.cusum_pos = 0.0
self.cusum_neg = 0.0
self.cusum_target = target_value
self.cusum_delta = 0.5 # Allowable drift
# Run rules tracking
self.consecutive_above: int = 0
self.consecutive_below: int = 0
self.last_8_values: deque = deque(maxlen=8)
def calculate_control_limits(self) -> ControlLimits:
"""
Tính ngưỡng kiểm soát từ historical data
Sử dụng moving range method (không cần biết trước sigma)
"""
if len(self.observations) < 20:
raise ValueError("Cần ít nhất 20 observations để calculate control limits")
data = list(self.observations)
mean = np.mean(data)
# Moving range method
moving_ranges = []
for i in range(1, len(data)):
mr = abs(data[i] - data[i-1])
moving_ranges.append(mr)
# D4 và D3 constants cho n=2 (subgroup size)
# Với individual measurements: d2 = 1.128
d2 = 1.128
mr_bar = np.mean(moving_ranges)
sigma_est = mr_bar / d2
ucl = mean + self.control_multiplier * sigma_est
lcl = mean - self.control_multiplier * sigma_est
self.control_limits = ControlLimits(
ucl=ucl,
lcl=lcl,
cl=mean,
sigma=sigma_est
)
return self.control_limits
def add_observation(self, value: float, timestamp: Optional[float] = None) -> List[SPCViolation]:
"""
Thêm observation mới và kiểm tra violations
Returns: Danh sách các violations được phát hiện
"""
if timestamp is None:
timestamp = len(self.observations)
self.observations.append(value)
self.timestamps.append(timestamp)
self.last_8_values.append(value)
violations = []
# Calculate/update control limits if needed
if len(self.observations) >= 20 and self.control_limits is None:
self.calculate_control_limits()
if self.control_limits:
# Rule 1: Shewhart 3-sigma rule
rule1 = self._check_shewhart_rule(value)
if rule1:
violations.extend(rule1)
# Rule 2: 2 consecutive points beyond 2-sigma
rule2 = self._check_two_sigma_rule(value)
if rule2:
violations.extend(rule2)
# Rule 3: 4 consecutive points beyond 1-sigma
rule3 = self._check_one_sigma_run_rule()
if rule3:
violations.extend(rule3)
# Rule 4: 8 consecutive points on same side of center
rule4 = self._check_trend_rule()
if rule4:
violations.extend(rule4)
# Update CUSUM
self._update_cusum(value)
# Check CUSUM
cusum_violation = self._check_cusum_violation()
if cusum_violation:
violations.append(cusum_violation)
self.violations.extend(violations)
return violations
def _check_shewhart_rule(self, value: float) -> List[SPCViolation]:
"""Rule 1: Giá trị nằm ngoài 3-sigma limits"""
violations = []
if value > self.control_limits.ucl:
violations.append(SPCViolation(
timestamp=self.timestamps[-1],
value=value,
violation_type="shewhart_upper",
severity="critical",
details={
"limit": self.control_limits.ucl,
"distance_sigma": (value - self.control_limits.cl) / self.control_limits.sigma
}
))
elif value < self.control_limits.lcl:
violations.append(SPCViolation(
timestamp=self.timestamps[-1],
value=value,
violation_type="shewhart_lower",
severity="critical",
details={
"limit": self.control_limits.lcl,
"distance_sigma": (self.control_limits.cl - value) / self.control_limits.sigma
}
))
return violations
def _check_two_sigma_rule(self, value: float) -> List[SPCViolation]:
"""Rule 2: 2 consecutive points ngoài 2-sigma zone"""
violations = []
if len(self.observations) < 2:
return violations
prev_value = self.observations[-2]
two_sigma_violated = (
(value > self.control_limits.ucl_2sigma and prev_value > self.control_limits.ucl_2sigma) or
(value < self.control_limits.lcl_2sigma and prev_value < self.control_limits.lcl_2sigma)
)
if two_sigma_violated:
violations.append(SPCViolation(
timestamp=self.timestamps[-1],
value=value,
violation_type="two_sigma_run",
severity="high",
details={
"previous_value": prev_value,
"limit_2sigma": self.control_limits.ucl_2sigma
}
))
return violations
def _check_one_sigma_run_rule(self) -> List[SPCViolation]:
"""Rule 3: 4 consecutive points beyond 1-sigma on same side"""
violations = []
if len(self.last_8_values) < 4:
return violations
recent_4 = list(self.last_8_values)[-4:]
cl = self.control_limits.cl
one_sigma = self.control_limits.sigma
all_above_1sigma = all(v > cl + one_sigma for v in recent_4)
all_below_1sigma = all(v < cl - one_sigma for v in recent_4)
if all_above_1sigma or all_below_1sigma:
violations.append(SPCViolation(
timestamp=self.timestamps[-1],
value=recent_4[-1],
violation_type="one_sigma_run",
severity="medium",
details={
"run_length": 4,
"mean_of_run": np.mean(recent_4)
}
))
return violations
def _check_trend_rule(self) -> List[SPCViolation]:
"""Rule 4: 8 consecutive points on same side of center"""
violations = []
if len(self.last_8_values) < 8:
return violations
recent_8 = list(self.last_8_values)
cl = self.control_limits.cl
all_above = all(v > cl for v in recent_8)
all_below = all(v < cl for v in recent_8)
if all_above or all_below:
violations.append(SPCViolation(
timestamp=self.timestamps[-1],
value=recent_8[-1],
violation_type="centerline_drift",
severity="warning",
details={
"consecutive_count": 8,
"side": "above" if all_above else "below",
"mean_of_run": np.mean(recent_8)
}
))
return violations
def _update_cusum(self, value: float):
"""Cập nhật CUSUM statistics"""
deviation = value - self.cusum_target
self.cusum_pos = max(0, self.cusum_pos + deviation - self.cusum_delta)
self.cusum_neg = max(0, self.cusum_neg - deviation - self.cusum_delta)
def _check_cusum_violation(self) -> Optional[SPCViolation]:
"""Kiểm tra CUSUM vượt ngưỡng"""
# CUSUM threshold thường là 5 * sigma
if self.control_limits is None:
return None
cusum_threshold = 5 * self.control_limits.sigma
if self.cusum_pos > cusum_threshold or self.cusum_neg > cusum_threshold:
return SPCViolation(
timestamp=self.timestamps[-1],
value=self.cusum_pos if self.cusum_pos > self.cusum_neg else -self.cusum_neg,
violation_type="cusum_drift",
severity="high",
details={
"cusum_pos": self.cusum_pos,
"cusum_neg": self.cusum_neg,
"threshold": cusum_threshold
}
)
return None
def detect_drift_type(self) -> str:
"""
Phân loại loại drift dựa trên pattern
"""
if len(self.observations) < 20:
return "insufficient_data"
recent = list(self.observations)[-20:]
early = list(self.observations)[:20]
recent_mean = np.mean(recent)
early_mean = np.mean(early)
# T-test để kiểm tra significance
t_stat, p_value = stats.ttest_ind(early, recent)
if p_value < 0.01:
if recent_mean > early_mean:
return "positive_drift"
else:
return "negative_drift"
# Check variance change
recent_var = np.var(recent)
early_var = np.var(early)
if max(recent_var, early_var) / min(recent_var, early_var) > 2.0:
return "variance_increase"
return "stable"
def get_control_chart_data(self) -> dict:
"""Export data cho visualization"""
return {
"observations": list(self.observations),
"timestamps": list(self.timestamps),
"control_limits": {
"ucl": self.control_limits.ucl if self.control_limits else None,
"lcl": self.control_limits.lcl if self.control_limits else None,
"cl": self.control_limits.cl if self.control_limits else None,
} if self.control_limits else None,
"violations": [
{
"timestamp": v.timestamp,
"value": v.value,
"type": v.violation_type,
"severity": v.severity
}
for v in self.violations
],
"cusum": {
"positive": self.cusum_pos,
"negative": self.cusum_neg
},
"drift_analysis": self.detect_drift_type()
}
=== ADVANCED: Semantic Quality Scoring ===
class SemanticQualityScorer:
"""
Đánh giá chất lượng semantic của AI responses
sử dụng HolySheep AI embeddings
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
async def calculate_embedding(self, text: str) -> List[float]:
"""Lấy embedding vector cho text"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "embedding-v3",
"input": text
}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/embeddings",
headers=headers,
json=payload
)
if response.status_code == 200:
data = response.json()
return data["data"][0]["embedding"]
else:
raise Exception(f"Embedding API error: {response.text}")
@staticmethod
def cosine_similarity(a: List[float], b: List[float]) -> float:
"""Tính cosine similarity giữa 2 vectors"""
dot_product = sum(x * y for x, y in zip(a, b))
norm_a = math.sqrt(sum(x * x for x in a))
norm_b = math.sqrt(sum(y * y for y in b))
if norm_a == 0 or norm_b == 0:
return 0.0
return dot_product / (norm_a * norm_b)
async def score_response_quality(
self,
prompt: str,
response: str,
expected_keywords: List[str] = None,
reference_response: str = None
) -> dict:
"""
Đánh giá chất lượng response với nhiều metrics
"""
scores = {}
# 1. Prompt-Response Relevance
prompt_emb = await self.calculate_embedding(prompt)
response_emb = await self.calculate_embedding(response)
scores["prompt_relevance"] = self.cosine_similarity(prompt_emb, response_emb)
# 2. Response Coherence (self-similarity)
sentences = response.split(".")
if len(sentences) >= 2:
half = len(sentences) // 2
first_half = ".".join(sentences[:half])
second_half = ".".join(sentences[half:])
first_emb = await self.calculate_embedding(first_half)
second_emb = await self.calculate_embedding(second_half)
scores["coherence"] = self.cosine_similarity(first_emb, second_emb)
else:
scores["coherence"] = 1.0
# 3. Keyword Coverage
if expected_keywords:
response_lower = response.lower()
coverage = sum(1 for kw in expected_keywords if kw.lower() in response_lower)
scores["keyword_coverage"] = coverage / len(expected_keywords)
else:
scores["keyword_coverage"] = None
# 4. Reference Similarity (nếu có)
if reference_response:
ref_emb = await self.calculate_embedding(reference_response)
scores["reference_similarity"] = self.cosine_similarity(response_emb, ref_emb)
else:
scores["reference_similarity"] = None
# Overall quality score
relevant_scores = [v for v in scores.values() if v is not None]
scores["overall_quality"] = sum(relevant_scores) / len(relevant_scores) if relevant_scores else 0
return scores
=== INTEGRATION EXAMPLE ===
async def demo_quality_monitoring():
"""Demonstrate full quality monitoring pipeline"""
import httpx
spc = StatisticalQualityControl(
window_size=50,
target_value=500.0, # target latency 500ms
control_multiplier=3.0
)
monitor = AIServiceMonitor(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
Tài nguyên liên quan
Bài viết liên quan