Mở đầu: Câu chuyện thực tế từ một startup AI ở Hà Nội
Một startup AI ở Hà Nội chuyên cung cấp dịch vụ chatbot cho ngành tài chính đã gặp phải vấn đề nghiêm trọng: khách hàng phản hồi rằng câu trả lời của bot "có vẻ đúng nhưng thiếu chính xác về số liệu". Đội kỹ thuật mất 3 tuần để phát hiện rằng tỷ lệ hallucinations (ảo giác AI) đã tăng từ 2% lên 18% sau khi nâng cấp model lên phiên bản mới. Hóa đơn hàng tháng từ nhà cung cấp cũ là $4,200, độ trễ trung bình 420ms, và không có công cụ giám sát chất lượng đầu ra.
Sau khi chuyển sang HolySheep AI với base_url https://api.holysheep.ai/v1, đội đã triển khai hệ thống giám sát chất lượng thống kê trong 30 ngày. Kết quả: độ trễ giảm xuống còn 180ms, chi phí hóa đơn hàng tháng chỉ còn $680, và tỷ lệ phát hiện anomalies giảm 73%. Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống tương tự từ A đến Z.
Tại sao cần giám sát chất lượng đầu ra AI?
Khi triển khai AI vào production, đầu ra của model không chỉ cần "nhìn có vẻ đúng" — nó cần đo lường được, có thể kiểm soát được, và phải phát hiện được regression một cách tự động. Một hệ thống giám sát chất lượng thống kê giúp bạn:
**Phát hiện sớm các vấn đề tiềm ẩn**: Trước khi khách hàng phàn nàn, hệ thống đã ghi nhận sự thay đổi bất thường trong phân phối đầu ra. **Giảm chi phí vận hành**: Với HolySheep AI có giá chỉ từ $0.42/MTok cho DeepSeek V3.2, việc tối ưu hóa prompt dựa trên dữ liệu giám sát giúp giảm token consumption đáng kể. **Đảm bảo compliance**: Trong ngành tài chính và y tế, đầu ra AI cần đáp ứng các tiêu chuẩn nhất định về độ chính xác.
Kiến trúc hệ thống giám sát chất lượng AI
1. Thiết lập kết nối HolySheep AI
Đầu tiên, bạn cần kết nối đến HolySheep AI API. HolySheep hỗ trợ thanh toán qua WeChat và Alipay, đăng ký tại đây để nhận tín dụng miễn phí khi bắt đầu.
import requests
import json
from datetime import datetime
import hashlib
class HolySheepAIMonitor:
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.request_history = []
self.quality_metrics = {
"response_lengths": [],
"token_counts": [],
"latencies": [],
"error_rates": [],
"hallucination_scores": []
}
def call_chat_completion(self, messages: list, model: str = "gpt-4.1") -> dict:
"""
Gọi API với tracking đầy đủ
Model pricing 2026 (USD/MTok):
- gpt-4.1: $8
- claude-sonnet-4.5: $15
- gemini-2.5-flash: $2.50
- deepseek-v3.2: $0.42
"""
start_time = datetime.now()
payload = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2048
}
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
latency = (datetime.now() - start_time).total_seconds() * 1000
result = response.json()
result["_metadata"] = {
"latency_ms": latency,
"timestamp": datetime.now().isoformat(),
"model": model,
"status_code": response.status_code
}
self._record_metrics(result)
return result
except Exception as e:
self.quality_metrics["error_rates"].append({
"timestamp": datetime.now().isoformat(),
"error": str(e)
})
raise
def _record_metrics(self, result: dict):
"""Ghi nhận metrics cho phân tích chất lượng"""
if "usage" in result:
self.quality_metrics["token_counts"].append({
"prompt_tokens": result["usage"].get("prompt_tokens", 0),
"completion_tokens": result["usage"].get("completion_tokens", 0),
"total_tokens": result["usage"].get("total_tokens", 0),
"timestamp": result["_metadata"]["timestamp"]
})
self.quality_metrics["latencies"].append({
"value": result["_metadata"]["latency_ms"],
"timestamp": result["_metadata"]["timestamp"]
})
if "choices" in result and len(result["choices"]) > 0:
content = result["choices"][0]["message"]["content"]
self.quality_metrics["response_lengths"].append({
"value": len(content),
"timestamp": result["_metadata"]["timestamp"]
})
Khởi tạo monitor
api_key = "YOUR_HOLYSHEEP_API_KEY" # Thay bằng API key thực tế
monitor = HolySheepAIMonitor(api_key)
2. Triển khai Statistical Process Control (SPC) cho đầu ra AI
Statistical Process Control là phương pháp tiêu chuẩn trong manufacturing để giám sát chất lượng. Áp dụng vào AI monitoring, chúng ta sử dụng Control Limits để phát hiện khi nào đầu ra deviated khỏi baseline.
import numpy as np
from scipy import stats
from dataclasses import dataclass
from typing import List, Tuple, Optional
@dataclass
class ControlLimits:
ucl: float # Upper Control Limit
lcl: float # Lower Control Limit
center_line: float
sigma: float
@dataclass
class SPCViolation:
timestamp: str
metric_name: str
value: float
limit_exceeded: str # "UCL" or "LCL"
severity: str # "warning" or "critical"
class AIQualityMonitor:
def __init__(self, confidence_level: float = 0.997):
"""
confidence_level: 3-sigma = 0.997 (99.7%)
"""
self.confidence_level = confidence_level
self.z_score = stats.norm.ppf((1 + confidence_level) / 2)
self.baseline_data = {
"latency": [],
"response_length": [],
"token_efficiency": [],
"error_rate": []
}
self.control_limits = {}
self.violations = []
self.is_baseline_stable = False
def calculate_control_limits(self, metric_name: str, data: List[float]) -> ControlLimits:
"""Tính toán Control Limits từ baseline data"""
mean = np.mean(data)
std = np.std(data, ddof=1)
return ControlLimits(
ucl=mean + (self.z_score * std),
lcl=max(0, mean - (self.z_score * std)), # Latency không âm
center_line=mean,
sigma=std
)
def establish_baseline(self, historical_data: dict, window_days: int = 7):
"""
Thiết lập baseline từ dữ liệu lịch sử
Trong thực tế, đội ở Hà Nội đã dùng 30 ngày data để establish baseline
"""
for metric_name, values in historical_data.items():
if len(values) >= 100: # Minimum samples cho statistical validity
limits = self.calculate_control_limits(metric_name, values)
self.control_limits[metric_name] = limits
self.baseline_data[metric_name] = values
self.is_baseline_stable = self._check_stability()
return self.is_baseline_stable
def _check_stability(self) -> bool:
"""Kiểm tra baseline có stable theo Western Electric Rules"""
for metric_name, data in self.baseline_data.items():
if len(data) < 20:
return False
# Rule 1: Không có điểm nào ngoài 3-sigma
limits = self.control_limits.get(metric_name)
if limits:
violations = [x for x in data if x > limits.ucl or x < limits.lcl]
if len(violations) > 0:
return False
# Rule 2: 2 trong 3 điểm liên tiếp ngoài 2-sigma
two_sigma = 2 * limits.sigma
for i in range(len(data) - 2):
window = data[i:i+3]
above_ucl2 = sum(1 for x in window if x > limits.center_line + two_sigma)
below_lcl2 = sum(1 for x in window if x < limits.center_line - two_sigma)
if above_ucl2 >= 2 or below_lcl2 >= 2:
return False
return True
def check_metric(self, metric_name: str, value: float, timestamp: str) -> Optional[SPCViolation]:
"""Kiểm tra một metric mới có vi phạm control limits không"""
if metric_name not in self.control_limits:
return None
limits = self.control_limits[metric_name]
if value > limits.ucl:
severity = "critical" if value > limits.ucl + limits.sigma else "warning"
violation = SPCViolation(
timestamp=timestamp,
metric_name=metric_name,
value=value,
limit_exceeded="UCL",
severity=severity
)
self.violations.append(violation)
return violation
if value < limits.lcl:
severity = "critical" if value < limits.lcl - limits.sigma else "warning"
violation = SPCViolation(
timestamp=timestamp,
metric_name=metric_name,
value=value,
limit_exceeded="LCL",
severity=severity
)
self.violations.append(violation)
return violation
return None
def detect_anomalies(self, recent_data: List[float], window_size: int = 10) -> List[int]:
"""
Phát hiện anomalies sử dụng IQR method
Trả về indices của các điểm bất thường
"""
if len(recent_data) < window_size:
return []
window = recent_data[-window_size:]
q1 = np.percentile(window, 25)
q3 = np.percentile(window, 75)
iqr = q3 - q1
lower_bound = q1 - 1.5 * iqr
upper_bound = q3 + 1.5 * iqr
anomalies = []
for i, val in enumerate(recent_data):
if val < lower_bound or val > upper_bound:
anomalies.append(i)
return anomalies
Khởi tạo quality monitor
quality_monitor = AIQualityMonitor(confidence_level=0.997)
3. Dashboard theo dõi trực quan với alerting
import time
from collections import deque
class QualityDashboard:
def __init__(self, monitor: AIQualityMonitor, alert_webhook: str = None):
self.monitor = monitor
self.alert_webhook = alert_webhook
self.realtime_buffer = {
"latency": deque(maxlen=1000),
"response_length": deque(maxlen=1000),
"error_rate": deque(maxlen=1000)
}
self.alert_history = []
def update_realtime(self, api_response: dict):
"""Cập nhật dashboard với response mới"""
metadata = api_response["_metadata"]
timestamp = metadata["timestamp"]
# Latency monitoring
latency = metadata["latency_ms"]
self.realtime_buffer["latency"].append({
"value": latency,
"timestamp": timestamp
})
# Check latency violation
latency_violation = self.monitor.check_metric("latency", latency, timestamp)
if latency_violation:
self._send_alert(latency_violation, "HIGH_LATENCY")
# Response length monitoring
if "choices" in api_response:
content = api_response["choices"][0]["message"]["content"]
length = len(content)
self.realtime_buffer["response_length"].append({
"value": length,
"timestamp": timestamp
})
length_violation = self.monitor.check_metric("response_length", length, timestamp)
if length_violation:
self._send_alert(length_violation, "RESPONSE_LENGTH_ANOMALY")
# Error rate monitoring
if api_response.get("error"):
self.realtime_buffer["error_rate"].append({
"value": 1,
"timestamp": timestamp
})
else:
self.realtime_buffer["error_rate"].append({
"value": 0,
"timestamp": timestamp
})
# Calculate rolling error rate
recent_errors = [x["value"] for x in list(self.realtime_buffer["error_rate"])[-100:]]
error_rate = sum(recent_errors) / len(recent_errors) if recent_errors else 0
if error_rate > 0.05: # Alert khi error rate > 5%
self._send_alert({
"timestamp": timestamp,
"metric_name": "error_rate",
"value": error_rate,
"severity": "critical"
}, "HIGH_ERROR_RATE")
def _send_alert(self, violation: dict, alert_type: str):
"""Gửi alert qua webhook hoặc log"""
alert = {
"type": alert_type,
"violation": violation,
"timestamp": datetime.now().isoformat()
}
self.alert_history.append(alert)
if self.alert_webhook:
try:
requests.post(self.alert_webhook, json=alert)
except Exception as e:
print(f"Failed to send alert: {e}")
# In ra console cho demo
print(f"🚨 ALERT [{alert_type}]: {violation}")
def get_summary_report(self) -> dict:
"""Tạo báo cáo tổng hợp 30 ngày"""
latency_values = [x["value"] for x in self.realtime_buffer["latency"]]
error_values = [x["value"] for x in self.realtime_buffer["error_rate"]]
return {
"period": "30_ngay",
"total_requests": len(latency_values),
"latency": {
"p50": np.percentile(latency_values, 50),
"p95": np.percentile(latency_values, 95),
"p99": np.percentile(latency_values, 99),
"avg": np.mean(latency_values)
},
"error_rate": {
"total": sum(error_values),
"percentage": (sum(error_values) / len(error_values) * 100) if error_values else 0
},
"alerts_triggered": len(self.alert_history),
"holy_sheep_cost_estimate": self._estimate_cost()
}
def _estimate_cost(self) -> dict:
"""
Ước tính chi phí HolySheep AI
Giá 2026 (USD/MTok):
- GPT-4.1: $8
- Claude Sonnet 4.5: $15
- Gemini 2.5 Flash: $2.50
- DeepSeek V3.2: $0.42
"""
token_data = list(monitor.quality_metrics["token_counts"])
total_tokens = sum(t["total_tokens"] for t in token_data)
total_tokens_millions = total_tokens / 1_000_000
return {
"if_gpt_41": round(total_tokens_millions * 8, 2),
"if_claude_sonnet": round(total_tokens_millions * 15, 2),
"if_deepseek": round(total_tokens_millions * 0.42, 2),
"savings_with_deepseek_vs_gpt": round(
(total_tokens_millions * 8) - (total_tokens_millions * 0.42), 2
)
}
Khởi tạo dashboard
dashboard = QualityDashboard(
monitor=quality_monitor,
alert_webhook="https://your-webhook-endpoint.com/alerts"
)
Pipeline hoàn chỉnh: Từ request đến monitoring
Dưới đây là pipeline hoàn chỉnh kết hợp tất cả components, được đội ở Hà Nội sử dụng để giám sát chatbot tài chính của họ:
import threading
from typing import Generator
class AIMonitoringPipeline:
"""
Pipeline hoàn chỉnh cho monitoring AI outputs
Baseline data: 30 ngày production (startup Hà Nội)
- Average latency: 420ms → sau tối ưu: 180ms
- Error rate: 2.3% → 0.4%
- Monthly cost: $4,200 → $680 (sử dụng DeepSeek V3.2)
"""
def __init__(self, api_key: str):
self.monitor = HolySheepAIMonitor(api_key)
self.quality_monitor = AIQualityMonitor(confidence_level=0.997)
self.dashboard = QualityDashboard(self.monitor)
self._baseline_established = False
def initialize_baseline(self, historical_logs_path: str):
"""Khởi tạo baseline từ log files"""
print("📊 Đang thiết lập baseline từ 30 ngày data...")
# Trong thực tế, đọc từ database/logs
# Ví dụ: load historical data từ BigQuery/S3
historical_latencies = []
historical_lengths = []
historical_errors = []
# Simulate baseline data (thay bằng data thật)
for _ in range(500):
historical_latencies.append(np.random.normal(420, 50))
historical_lengths.append(np.random.normal(500, 100))
historical_errors.append(np.random.choice([0, 1], p=[0.977, 0.023]))
baseline_data = {
"latency": historical_latencies,
"response_length": historical_lengths,
"error_rate": historical_errors
}
self._baseline_established = self.quality_monitor.establish_baseline(baseline_data)
print(f"✅ Baseline established: {self._baseline_established}")
print(f"📈 Control Limits: {self.quality_monitor.control_limits}")
def process_request(self, user_query: str, context: dict = None) -> dict:
"""Xử lý request với full monitoring"""
# Build messages
system_prompt = """Bạn là chatbot tư vấn tài chính.
Trả lời CHÍNH XÁC về các con số. Nếu không chắc, nói rõ 'Tôi không biết'."""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_query}
]
# Gọi API với monitoring
response = self.monitor.call_chat_completion(
messages=messages,
model="deepseek-v3.2" # Model rẻ nhất, đủ dùng
)
# Cập nhật quality monitoring
self.quality_monitor._record_metric_from_response(response)
# Cập nhật dashboard
self.dashboard.update_realtime(response)
return response
def run_monitoring_loop(self, duration_seconds: int = 86400):
"""
Chạy monitoring loop trong N giây
86400 giây = 24 giờ
"""
print(f"🔄 Bắt đầu monitoring trong {duration_seconds} giây...")
start_time = time.time()
test_queries = [
"Tỷ giá USD/VND hôm nay là bao nhiêu?",
"Lãi suất tiết kiệm ngân hàng Techcombank?",
"Chỉ số VN-Index hiện tại?",
]
while time.time() - start_time < duration_seconds:
query = test_queries[int(time.time()) % len(test_queries)]
try:
response = self.process_request(query)
print(f"✅ Response ({response['_metadata']['latency_ms']:.0f}ms): "
f"{response['choices'][0]['message']['content'][:100]}...")
except Exception as e:
print(f"❌ Error: {e}")
time.sleep(5) # Request mỗi 5 giây
def generate_monthly_report(self) -> dict:
"""Tạo báo cáo 30 ngày"""
if not self._baseline_established:
print("⚠️ Warning: Baseline chưa established, report có thể không chính xác")
return self.dashboard.get_summary_report()
==================== MAIN EXECUTION ====================
if __name__ == "__main__":
# Khởi tạo pipeline
pipeline = AIMonitoringPipeline(api_key="YOUR_HOLYSHEEP_API_KEY")
# Thiết lập baseline
pipeline.initialize_baseline("path/to/historical/logs")
# Chạy monitoring trong 1 phút (demo)
# Thực tế: run_monitoring_loop(86400) cho 24 giờ
pipeline.run_monitoring_loop(duration_seconds=60)
# Xuất báo cáo
report = pipeline.generate_monthly_report()
print("\n" + "="*60)
print("📋 BÁO CÁO 30 NGÀY")
print("="*60)
print(f"Tổng requests: {report['total_requests']}")
print(f"Latency P50: {report['latency']['p50']:.0f}ms")
print(f"Latency P95: {report['latency']['p95']:.0f}ms")
print(f"Error rate: {report['error_rate']['percentage']:.2f}%")
print(f"\n💰 Chi phí ước tính:")
print(f" - Nếu dùng GPT-4.1: ${report['holy_sheep_cost_estimate']['if_gpt_41']}")
print(f" - Nếu dùng DeepSeek V3.2: ${report['holy_sheep_cost_estimate']['if_deepseek']}")
print(f" - Tiết kiệm: ${report['holy_sheep_cost_estimate']['savings_with_deepseek_vs_gpt']}")
Lỗi thường gặp và cách khắc phục
1. Lỗi 401 Unauthorized - Invalid API Key
**Mô tả lỗi**: Khi gọi API nhận được response
{"error": {"message": "Invalid authentication", "type": "invalid_request_error"}}. Nguyên nhân thường là key bị sai format hoặc chưa copy đầy đủ.
**Cách khắc phục**:
❌ SAI - Key bị cắt hoặc có khoảng trắng thừa
api_key = " sk-holysheep-xxxxx "
✅ ĐÚNG - Strip whitespace và validate format
api_key = os.environ.get("HOLYSHEEP_API_KEY", "").strip()
if not api_key.startswith("sk-"):
raise ValueError("API key phải bắt đầu bằng 'sk-'")
if len(api_key) < 32:
raise ValueError("API key quá ngắn, có thể bị cắt")
Verify key bằng cách gọi API health check
def verify_api_key(base_url: str, api_key: str) -> bool:
headers = {"Authorization": f"Bearer {api_key}"}
response = requests.get(f"{base_url}/models", headers=headers, timeout=10)
return response.status_code == 200
if not verify_api_key("https://api.holysheep.ai/v1", api_key):
raise ValueError("API key không hợp lệ hoặc đã hết hạn")
2. Lỗi Rate Limit khi monitoring số lượng lớn requests
**Mô tả lỗi**: Nhận được
429 Too Many Requests khi hệ thống monitoring gửi quá nhiều requests. Startup ở Hà Nội đã gặp lỗi này khi thử monitoring 1000 requests/giây.
**Cách khắc phục**:
import time
from threading import Lock
from ratelimit import limits, sleep_and_retry
class RateLimitedMonitor:
def __init__(self, api_key: str, requests_per_second: int = 50):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.rate_limit = requests_per_second
self.last_request_time = 0
self.lock = Lock()
self.request_count = 0
self.reset_time = time.time()
@sleep_and_retry
@limits(calls=50, period=1.0)
def _throttled_request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
"""Gọi API với rate limiting"""
with self.lock:
current_time = time.time()
# Reset counter mỗi 60 giây
if current_time - self.reset_time > 60:
self.request_count = 0
self.reset_time = current_time
# Đảm bảo không vượt rate limit
if self.request_count >= self.rate_limit:
sleep_time = 60 - (current_time - self.reset_time)
if sleep_time > 0:
time.sleep(sleep_time)
self.request_count = 0
self.reset_time = time.time()
self.request_count += 1
headers = kwargs.pop("headers", {})
headers["Authorization"] = f"Bearer {self.api_key}"
url = f"{self.base_url}{endpoint}"
response = requests.request(method, url, headers=headers, **kwargs)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"⏳ Rate limited, retry sau {retry_after} giây...")
time.sleep(retry_after)
return self._throttled_request(method, endpoint, **kwargs)
return response
def call_with_retry(self, payload: dict, max_retries: int = 3) -> dict:
"""Gọi API với exponential backoff retry"""
for attempt in range(max_retries):
try:
response = self._throttled_request(
"POST",
"/chat/completions",
json=payload
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
wait_time = 2 ** attempt # Exponential backoff: 1s, 2s, 4s
print(f"⚠️ Attempt {attempt + 1} failed: {e}. Retrying in {wait_time}s...")
time.sleep(wait_time)
return None
Sử dụng
monitor = RateLimitedMonitor(api_key="YOUR_HOLYSHEEP_API_KEY", requests_per_second=50)
3. Lỗi Baseline Drift - Control Limits không còn chính xác
**Mô tả lỗi**: Sau vài tuần, các violations báo động liên tục dù AI vẫn hoạt động bình thường. Đây là hiện tượng "baseline drift" - khi usage patterns thay đổi theo thời gian, baseline cũ trở nên outdated.
**Cách khắc phục**:
class AdaptiveBaselineMonitor:
"""
Monitor với adaptive baseline - tự động cập nhật baseline
theo thời gian để tránh drift
"""
def __init__(self, baseline_window_days: int = 30,
adaptation_threshold: float = 0.1):
self.baseline_window_days = baseline_window_days
self.adaptation_threshold = adaptation_threshold
self.current_baseline = None
self.baseline_version = 0
self.historical_baselines = []
def calculate_baseline_drift(self, old_baseline: dict,
new_data: List[float]) -> float:
"""Tính độ drift giữa baseline cũ và data mới"""
old_mean = old_baseline["mean"]
new_mean = np.mean(new_data)
old_std = old_baseline["std"]
new_std = np.std(new_data)
mean_drift = abs(new_mean - old_mean) / old_mean if old_mean != 0 else 0
std_drift = abs(new_std - old_std) / old_std if old_std != 0 else 0
return (mean_drift + std_drift) / 2
def check_and_adapt_baseline(self, recent_data: List[float]) -> bool:
"""
Kiểm tra và tự động adapt baseline nếu cần
Trả về True nếu baseline đã được cập nhật
"""
if self.current_baseline is None:
self._set_baseline(recent_data)
return True
drift = self.calculate_baseline_drift(self.current_baseline, recent_data)
if drift > self.adaptation_threshold:
print(f"📈 Baseline drift detected: {drift:.2%}. Adapting...")
# Lưu baseline cũ vào history
self.historical_baselines.append({
**self.current_baseline,
"version": self.baseline_version,
"valid_until": datetime.now().isoformat()
})
# Cập nhật baseline mới
self._set_baseline(recent_data)
self.baseline_version += 1
return True
return False
def _set_baseline(self, data: List[float]):
"""Thiết lập baseline mới"""
self.current_baseline = {
"mean": np.mean(data),
"std": np.std(data),
"median": np.median(data),
"p95": np.percentile(data, 95),
"p99": np.percentile(data, 99),
"sample_size": len(data),
"created_at": datetime.now().isoformat()
}
# Tính lại control limits
self.control_limits = self._calculate_adaptive_limits(data)
def _calculate_adaptive_limits(self, data: List[float]) -> dict:
"""Tính control limits với buffer cho seasonal patterns"""
mean = np.mean(data)
std = np.std(data)
z_score = 3
# Thêm buffer 10% để tránh false positives
buffer = 1 + self.adaptation_threshold
return {
"ucl": (mean + z_score * std) * buffer,
"lcl": max(0, (mean - z_score * std) / buffer),
"center_line": mean
}
def get_baseline_comparison_report(self) -> dict:
"""So sánh baseline cũ và mới để xem trend"""
if len(self.historical_baselines) < 2:
return {"message": "Chưa đủ data để so sánh"}
latest = self.current_baseline
previous = self.historical_baselines[-1]
return {
"current": {
"mean": latest["mean"],
"version": self.baseline_version
},
"previous": {
"mean": previous["mean"],
"version": previous["version"]
},
"change_percentage": (
(latest["mean"] - previous["mean"]) / previous["mean"] * 100
if previous["mean"] != 0 else 0
),
"total_adaptations": self.baseline_version
Tài nguyên liên quan
Bài viết liên quan