Khi xây dựng hệ thống RAG (Retrieval-Augmented Generation), việc thay đổi embedding model hoặc chiến lược chunking có thể mang lại cải thiện đáng kể — hoặc phá vỡ hoàn toàn trải nghiệm người dùng. Là một developer từng "đổ bể" production vì thay đổi embedding không được kiểm chứng, tôi hiểu rằng gray-scale deployment (triển khai dần) là cách an toàn nhất để đánh giá các thay đổi này trước khi roll-out toàn bộ.

RAG Evaluation Là Gì? Tại Sao Bạn Cần Nó?

Trong bài viết này, tôi sẽ hướng dẫn bạn xây dựng một pipeline đánh giá RAG grayscale hoàn chỉnh sử dụng HolySheep AI. Bạn sẽ học cách:

HolySheep AI Khác Gì So Với OpenAI/Claude?

Tiêu chí HolySheep AI OpenAI Anthropic
Giá GPT-4.1/Claude 4.5 $8 / $15 $8 / $15 $8 / $15
DeepSeek V3.2 $0.42 Không có Không có
Tỷ giá ¥1 = $1 Thanh toán USD Thanh toán USD
Tốc độ trung bình <50ms 200-500ms 300-600ms
Thanh toán WeChat/Alipay Visa/Mastercard Visa/Mastercard
Tín dụng miễn phí Có, khi đăng ký $5 trial $5 trial

📌 Lưu ý quan trọng: Với chi phí DeepSeek V3.2 chỉ $0.42/MTok (rẻ hơn 85%+ so với GPT-4), bạn có thể chạy hàng nghìn evaluation queries mà không lo về chi phí.

Kiến Trúc Tổng Quan Của Pipeline RAG Evaluation

Trước khi vào code, hãy hiểu pipeline hoạt động như thế nào:

┌─────────────────────────────────────────────────────────────────┐
│                    RAG GRAYSCALE EVALUATION PIPELINE            │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌──────────┐    ┌──────────────┐    ┌────────────────────┐    │
│  │ Test Set │───▶│ Query Engine │───▶│ Old vs New Config  │    │
│  │ (100-1K) │    │   v1/v2      │    │    Comparison      │    │
│  └──────────┘    └──────────────┘    └─────────┬──────────┘    │
│                                               │                │
│                          ┌────────────────────┴─────┐          │
│                          ▼                          ▼          │
│               ┌──────────────────┐       ┌──────────────────┐  │
│               │   Retrieval      │       │   Generation     │  │
│               │   Metrics        │       │   Metrics        │  │
│               │ • Hit Rate       │       │ • ROUGE/LESER    │  │
│               │ • MRR            │       │ • Faithfulness   │  │
│               │ • NDCG           │       │ • Answer Relevancy│  │
│               └──────────────────┘       └──────────────────┘  │
│                          │                          │          │
│                          └──────────┬─────────────────┘          │
│                                     ▼                           │
│                          ┌──────────────────┐                    │
│                          │   Final Report   │                    │
│                          │   + Decision     │                    │
│                          └──────────────────┘                    │
└─────────────────────────────────────────────────────────────────┘

Chuẩn Bị Môi Trường Và Cài Đặt

Bước 1: Đăng ký tài khoản HolySheep

Trước tiên, bạn cần tạo tài khoản tại đăng ký tại đây để nhận API key và tín dụng miễn phí. Giao diện đăng ký hỗ trợ WeChat, Alipay và email — rất thuận tiện cho developer Việt Nam.

Bước 2: Cài đặt thư viện cần thiết

# Cài đặt các thư viện cần thiết
pip install requests pandas numpy scikit-learn ragas openai tiktoken


Kiểm tra phiên bản

python -c "import ragas; print(f'RAGAS version: {ragas.__version__}')"

Output: RAGAS version: 0.1.x

Bước 3: Cấu hình API Client

import os
import requests
import json
from typing import List, Dict, Any
from dataclasses import dataclass
from datetime import datetime


============================================


HOLYSHEEP AI - CẤU HÌNH API CHÍNH THỨC


============================================


⚠️ QUAN TRỌNG: Không dùng api.openai.com


URL: https://api.holysheep.ai/v1


============================================


HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Thay bằng API key của bạn
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

@dataclass
class HolySheepConfig:
    """Cấu hình cho HolySheep AI API"""
    api_key: str
    base_url: str = "https://api.holysheep.ai/v1"
    embedding_model: str = "text-embedding-3-large"
    chat_model: str = "deepseek-v3.2"
    
    def headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
    
    def embedding_url(self) -> str:
        return f"{self.base_url}/embeddings"
    
    def chat_url(self) -> str:
        return f"{self.base_url}/chat/completions"


Khởi tạo client

config = HolySheepConfig(api_key=HOLYSHEEP_API_KEY)


Test kết nối

def test_connection():
    response = requests.post(
        config.chat_url(),
        headers=config.headers(),
        json={
            "model": config.chat_model,
            "messages": [{"role": "user", "content": "Hello"}],
            "max_tokens": 10
        }
    )
    if response.status_code == 200:
        print("✅ Kết nối HolySheep AI thành công!")
        return True
    else:
        print(f"❌ Lỗi kết nối: {response.status_code} - {response.text}")
        return False

test_connection()

Xây Dựng Test Dataset Cho RAG Evaluation

Một bộ test dataset chất lượng là nền tảng của evaluation thành công. Tôi khuyên bạn nên chuẩn bị ít nhất 100-500 sample questions.

from dataclasses import dataclass, field
from typing import List, Optional
import json

@dataclass
class RAGEvaluationSample:
    """Một sample cho việc đánh giá RAG"""
    question: str                    # Câu hỏi của user
    ground_truth_answer: str         # Câu trả lời đúng (golden answer)
    ground_truth_contexts: List[str] # Context cần thiết để trả lời
    metadata: Dict[str, Any] = field(default_factory=dict)

class TestDatasetBuilder:
    """Builder để tạo test dataset"""
    
    def __init__(self):
        self.samples: List[RAGEvaluationSample] = []
    
    def add_sample(
        self,
        question: str,
        ground_truth: str,
        contexts: List[str],
        category: str = "general",
        difficulty: str = "medium"
    ) -> "TestDatasetBuilder":
        """Thêm một sample vào dataset"""
        sample = RAGEvaluationSample(
            question=question,
            ground_truth_answer=ground_truth,
            ground_truth_contexts=contexts,
            metadata={"category": category, "difficulty": difficulty}
        )
        self.samples.append(sample)
        return self
    
    def build(self) -> List[Dict]:
        """Xuất dataset dạng list dictionary"""
        return [
            {
                "user_input": s.question,
                "ground_truth": s.ground_truth_answer,
                "retrieved_contexts": s.ground_truth_contexts,
                **s.metadata
            }
            for s in self.samples
        ]
    
    def save_to_json(self, filepath: str):
        """Lưu dataset ra file JSON"""
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(self.build(), f, ensure_ascii=False, indent=2)
        print(f"💾 Đã lưu {len(self.samples)} samples vào {filepath}")


============================================


VÍ DỤ: Tạo test dataset mẫu cho tài liệu kỹ thuật


============================================


builder = TestDatasetBuilder()


Thêm các sample mẫu (bạn nên thay bằng dữ liệu thực tế)

builder.add_sample(
    question="Cách cấu hình rate limiting trong API Gateway?",
    ground_truth="Để cấu hình rate limiting, bạn cần thiết lập các tham số: requests_per_minute, burst_size, và whitelist IPs trong config.yaml",
    contexts=[
        "Rate limiting được cấu hình trong section 'rate_limit' của config.yaml",
        "Tham số requests_per_minute giới hạn số request mỗi phút",
        "Burst_size cho phép burst traffic tạm thời"
    ],
    category="configuration",
    difficulty="easy"
)

builder.add_sample(
    question="Sự khác biệt giữa authentication và authorization là gì?",
    ground_truth="Authentication (xác thực) là quá trình xác minh identity của user, còn Authorization (ủy quyền) là quá trình kiểm tra quyền truy cập tài nguyên sau khi đã xác thực",
    contexts=[
        "Authentication: Xác thực người dùng - 'Bạn là ai?'",
        "Authorization: Ủy quyền truy cập - 'Bạn được làm gì?'",
        "OAuth 2.0 là protocol phổ biến cho authorization"
    ],
    category="security",
    difficulty="medium"
)


Xuất dataset

test_dataset = builder.build()
print(f"📊 Dataset có {len(test_dataset)} samples")
print(json.dumps(test_dataset[0], indent=2, ensure_ascii=False))

Triển Khai Embedding Comparison Engine

Đây là phần core của pipeline — so sánh embedding models và chiến lược chunking khác nhau.

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from concurrent.futures import ThreadPoolExecutor
import time

class EmbeddingComparisonEngine:
    """
    Engine so sánh embedding models cho RAG evaluation
    Hỗ trợ: text-embedding-3-large, text-embedding-3-small, multilingual
    """
    
    def __init__(self, config: HolySheepConfig):
        self.config = config
    
    def get_embedding(self, text: str, model: str = "text-embedding-3-large") -> np.ndarray:
        """
        Lấy embedding vector từ HolySheep API
        """
        start_time = time.time()
        
        response = requests.post(
            self.config.embedding_url(),
            headers=self.config.headers(),
            json={
                "model": model,
                "input": text
            }
        )
        
        latency = (time.time() - start_time) * 1000  # ms
        
        if response.status_code != 200:
            raise Exception(f"Embedding API Error: {response.text}")
        
        data = response.json()
        embedding = np.array(data['data'][0]['embedding'])
        
        return embedding, latency
    
    def batch_get_embeddings(
        self, 
        texts: List[str], 
        model: str = "text-embedding-3-large",
        max_workers: int = 5
    ) -> tuple:
        """
        Batch embedding với parallel processing
        """
        embeddings = []
        total_latency = 0
        
        def single_embedding(text):
            try:
                emb, lat = self.get_embedding(text, model)
                return emb, lat
            except Exception as e:
                print(f"⚠️ Lỗi embedding: {e}")
                return None, 0
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            results = list(executor.map(single_embedding, texts))
        
        for emb, lat in results:
            if emb is not None:
                embeddings.append(emb)
                total_latency += lat
        
        return np.array(embeddings), total_latency / len(embeddings) if embeddings else 0
    
    def calculate_retrieval_metrics(
        self,
        query: str,
        retrieved_chunks: List[str],
        relevant_chunks: List[str],
        model: str = "text-embedding-3-large"
    ) -> Dict[str, float]:
        """
        Tính toán retrieval metrics cho một query
        """
        # Lấy embedding của query
        query_emb, query_lat = self.get_embedding(query, model)
        
        # Lấy embedding của retrieved chunks
        retrieved_embs, avg_retrieve_lat = self.batch_get_embeddings(retrieved_chunks, model)
        
        # Lấy embedding của relevant chunks  
        relevant_embs, _ = self.batch_get_embeddings(relevant_chunks, model)
        
        # Tính Hit Rate (recall)
        hits = 0
        for rel_emb in relevant_embs:
            similarities = cosine_similarity([query_emb], [rel_emb])[0][0]
            if similarities > 0.8:  # threshold
                hits += 1
        hit_rate = hits / len(relevant_chunks) if relevant_chunks else 0
        
        # Tính MRR (Mean Reciprocal Rank)
        mrr = 0
        similarities = cosine_similarity([query_emb], retrieved_embs)[0]
        ranked_indices = np.argsort(similarities)[::-1]
        
        for rank, idx in enumerate(ranked_indices, 1):
            if idx < len(relevant_chunks):
                mrr = 1 / rank
                break
        
        return {
            "hit_rate": hit_rate,
            "mrr": mrr,
            "avg_retrieval_latency_ms": avg_retrieve_lat,
            "query_latency_ms": query_lat
        }


============================================


SỬ DỤNG ENGINE


============================================


engine = EmbeddingComparisonEngine(config)


Ví dụ đánh giá một query

sample_query = "Cách cấu hình rate limiting?"
retrieved = [
    "Rate limiting config trong gateway.yaml",
    "Authentication methods overview",
    "API endpoint documentation"
]
relevant = [
    "Rate limiting config trong gateway.yaml",
    "Rate limiting policy settings"
]

metrics = engine.calculate_retrieval_metrics(
    query=sample_query,
    retrieved_chunks=retrieved,
    relevant_chunks=relevant,
    model="text-embedding-3-large"
)

print("📊 Retrieval Metrics:")
for k, v in metrics.items():
    print(f"   {k}: {v:.4f}")

Chiến Lược Chunking: So Sánh Fixed-Size vs Semantic vs Recursive

Chiến lược chunking ảnh hưởng lớn đến chất lượng retrieval. Tôi sẽ so sánh 3 chiến lược phổ biến nhất.

class ChunkingStrategy:
    """Base class cho các chiến lược chunking"""
    
    def chunk(self, document: str) -> List[str]:
        raise NotImplementedError

class FixedSizeChunking(ChunkingStrategy):
    """
    Chunking theo kích thước cố định
    Đơn giản nhưng có thể cắt giữa câu
    """
    
    def __init__(self, chunk_size: int = 500, overlap: int = 50):
        self.chunk_size = chunk_size
        self.overlap = overlap
    
    def chunk(self, document: str) -> List[str]:
        words = document.split()
        chunks = []
        
        for i in range(0, len(words), self.chunk_size - self.overlap):
            chunk = ' '.join(words[i:i + self.chunk_size])
            if chunk:
                chunks.append(chunk)
        
        return chunks

class RecursiveChunking(ChunkingStrategy):
    """
    Recursive chunking: Cắt theo hierarchy (paragraph -> sentence -> word)
    Giữ được ngữ cảnh tốt hơn
    """
    
    def __init__(self, separators: List[str] = None, chunk_size: int = 500):
        self.separators = separators or ['\n\n', '\n', '. ', ' ']
        self.chunk_size = chunk_size
    
    def chunk(self, document: str) -> List[str]:
        chunks = []
        remaining = document
        
        while remaining:
            split_done = False
            
            for sep in self.separators:
                if sep in remaining:
                    parts = remaining.split(sep)
                    current_chunk = ""
                    
                    for part in parts:
                        test_chunk = current_chunk + sep + part if current_chunk else part
                        
                        if len(test_chunk.split()) <= self.chunk_size:
                            current_chunk = test_chunk
                        else:
                            if current_chunk:
                                chunks.append(current_chunk.strip())
                            current_chunk = part
                    
                    remaining = current_chunk
                    split_done = True
                    break
            
            if not split_done:
                chunks.append(remaining[:self.chunk_size * 5])
                break
        
        return [c for c in chunks if c.strip()]

class SemanticChunking(ChunkingStrategy):
    """
    Semantic chunking: Nhóm các đoạn có ngữ cảnh liên quan
    Cần embedding để xác định semantic boundaries
    """
    
    def __init__(self, similarity_threshold: float = 0.7, min_chunk_size: int = 100):
        self.similarity_threshold = similarity_threshold
        self.min_chunk_size = min_chunk_size
        self.embedding_engine = None
    
    def set_embedding_engine(self, engine: EmbeddingComparisonEngine):
        self.embedding_engine = engine
    
    def chunk(self, document: str) -> List[str]:
        # Tách document thành các đoạn
        paragraphs = [p.strip() for p in document.split('\n') if p.strip()]
        
        if not paragraphs:
            return [document]
        
        if self.embedding_engine is None:
            # Fallback: chunk đơn giản
            return paragraphs
        
        # Lấy embedding cho mỗi paragraph
        embeddings, _ = self.embedding_engine.batch_get_embeddings(paragraphs)
        
        # Nhóm các paragraph có similarity cao
        chunks = []
        current_chunk = [paragraphs[0]]
        current_emb = [embeddings[0]]
        
        for i in range(1, len(paragraphs)):
            sim = cosine_similarity([embeddings[i]], [embeddings[i-1]])[0][0]
            
            if sim >= self.similarity_threshold:
                current_chunk.append(paragraphs[i])
                current_emb.append(embeddings[i])
            else:
                # Lưu chunk hiện tại nếu đủ lớn
                if len(' '.join(current_chunk).split()) >= self.min_chunk_size:
                    chunks.append(' '.join(current_chunk))
                current_chunk = [paragraphs[i]]
                current_emb = [embeddings[i]]
        
        # Lưu chunk cuối
        if current_chunk:
            chunks.append(' '.join(current_chunk))
        
        return chunks


============================================


SO SÁNH 3 CHIẾN LƯỢC


============================================


test_document = """
Hệ thống RAG (Retrieval-Augmented Generation) là một kiến trúc AI kết hợp khả năng 
truy xuất thông tin từ cơ sở dữ liệu với khả năng sinh text của LLM.

Ưu điểm của RAG bao gồm:
1. Cập nhật knowledge base dễ dàng mà không cần retrain
2. Giảm hiện tượng hallucination vì câu trả lời dựa trên dữ liệu thực
3. Tăng tính traceable - có thể kiểm tra nguồn gốc câu trả lời

Embedding model đóng vai trò quan trọng trong việc chuyển đổi text thành vector.
Các model phổ biến bao gồm OpenAI text-embedding-3, Cohere, và sentence-transformers.

Chiến lược chunking ảnh hưởng lớn đến chất lượng retrieval.
Fixed-size chunking đơn giản nhưng có thể cắt giữa câu.
Recursive chunking giữ được ngữ cảnh tốt hơn.
Semantic chunking nhóm các đoạn có ngữ cảnh liên quan.
"""

strategies = {
    "Fixed-Size (500 words)": FixedSizeChunking(chunk_size=500, overlap=50),
    "Recursive": RecursiveChunking(chunk_size=500),
    "Semantic (threshold=0.7)": SemanticChunking(similarity_threshold=0.7)
}

print("📊 So sánh Chiến lược Chunking:\n")
print("-" * 60)

for name, strategy in strategies.items():
    # Semantic cần embedding engine
    if isinstance(strategy, SemanticChunking):
        strategy.set_embedding_engine(engine)
    
    chunks = strategy.chunk(test_document)
    
    total_words = sum(len(c.split()) for c in chunks)
    avg_chunk_size = total_words / len(chunks) if chunks else 0
    
    print(f"\n🔹 {name}")
    print(f"   Số chunks: {len(chunks)}")
    print(f"   Kích thước TB: {avg_chunk_size:.1f} words")
    print(f"   Chunk đầu tiên: {chunks[0][:80] if chunks else 'N/A'}...")

print("\n" + "=" * 60)

Đánh Giá Chất Lượng Câu Trả Lời Với RAGAS

RAGAS (RAG Assessment) là framework chuẩn để đánh giá RAG pipelines. Kết hợp với HolySheep AI, bạn có thể đo lường 4 metrics chính.

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)
from datasets import Dataset

class RAGQualityEvaluator:
    """
    Evaluator đánh giá chất lượng câu trả lời RAG
    Sử dụng RAGAS metrics
    """
    
    def __init__(self, config: HolySheepConfig):
        self.config = config
    
    def generate_answer(
        self,
        question: str,
        context: List[str]
    ) -> str:
        """
        Sinh câu trả lời sử dụng HolySheep AI (DeepSeek V3.2)
        """
        context_text = "\n\n".join(context)
        
        prompt = f"""Dựa trên ngữ cảnh sau để trả lời câu hỏi:

Ngữ cảnh:
{context_text}

Câu hỏi: {question}

Trả lời:"""

        response = requests.post(
            self.config.chat_url(),
            headers=self.config.headers(),
            json={
                "model": "deepseek-v3.2",  # Model rẻ nhất, chất lượng tốt
                "messages": [
                    {"role": "system", "content": "Bạn là assistant trả lời dựa trên ngữ cảnh được cung cấp."},
                    {"role": "user", "content": prompt}
                ],
                "temperature": 0.3,
                "max_tokens": 500
            }
        )
        
        if response.status_code != 200:
            raise Exception(f"Generation error: {response.text}")
        
        return response.json()['choices'][0]['message']['content']
    
    def evaluate_sample(
        self,
        question: str,
        ground_truth: str,
        context: List[str]
    ) -> Dict[str, float]:
        """
        Đánh giá một sample
        """
        # Sinh câu trả lời
        generated_answer = self.generate_answer(question, context)
        
        # Tính các metrics (sử dụng RAGAS)
        # Lưu ý: Trong production, bạn nên dùng async để tăng tốc
        
        # Faithfulness: Mức độ answer trung thành với context
        # Answer Relevancy: Mức độ answer liên quan đến question
        # Context Precision: Độ chính xác của context được retrieve
        # Context Recall: Độ phủ của context với ground truth
        
        return {
            "question": question,
            "ground_truth": ground_truth,
            "generated_answer": generated_answer,
            "faithfulness": 0.85,  # Placeholder - dùng RAGAS thực tế
            "answer_relevancy": 0.82,
            "context_precision": 0.78,
            "context_recall": 0.88
        }
    
    def evaluate_dataset(
        self,
        dataset: List[Dict],
        sample_size: int = 10  # Giới hạn để tiết kiệm chi phí
    ) -> Dict[str, Any]:
        """
        Đánh giá toàn bộ dataset
        """
        results = []
        total_cost = 0
        
        print(f"🚀 Bắt đầu đánh giá {min(sample_size, len(dataset))} samples...")
        
        for i, sample in enumerate(dataset[:sample_size]):
            try:
                result = self.evaluate_sample(
                    question=sample["user_input"],
                    ground_truth=sample["ground_truth"],
                    context=sample["retrieved_contexts"]
                )
                results.append(result)
                
                # Ước tính chi phí (DeepSeek V3.2 = $0.42/MTok input, $0.42/MTok output)
                # Giả sử mỗi sample ~500 tokens input + 200 tokens output
                cost = (500 + 200) / 1_000_000 * 0.42
                total_cost += cost
                
                print(f"✅ Sample {i+1}/{sample_size} hoàn thành (~$ {cost:.4f})")
                
            except Exception as e:
                print(f"❌ Lỗi sample {i+1}: {e}")
        
        # Tính trung bình metrics
        avg_metrics = {
            "faithfulness": np.mean([r["faithfulness"] for r in results]),
            "answer_relevancy": np.mean([r["answer_relevancy"] for r in results]),
            "context_precision": np.mean([r["context_precision"] for r in results]),
            "context_recall": np.mean([r["context_recall"] for r in results])
        }
        
        return {
            "sample_count": len(results),
            "total_cost_usd": total_cost,
            "avg_metrics": avg_metrics,
            "detailed_results": results
        }


============================================


CHẠY EVALUATION


============================================


evaluator = RAGQualityEvaluator(config)


Đánh giá dataset mẫu

evaluation_results = evaluator.evaluate_dataset(test_dataset, sample_size=2)

print("\n" + "=" * 60)
print("📊 KẾT QUẢ ĐÁNH GIÁ RAG")
print("=" * 60)
print(f"\nSố samples đánh giá: {evaluation_results['sample_count']}")
print(f"Tổng chi phí: ${evaluation_results['total_cost_usd']:.4f}")
print("\n📈 Metrics Trung Bình:")
for metric, value in evaluation_results['avg_metrics'].items():
    print(f"   {metric}: {value:.2%}")

A/B Testing Với Grayscale Deployment

Đây là phần quan trọng nhất — triển khai an toàn để so sánh old vs new config.

import random
from collections import defaultdict
from datetime import datetime, timedelta

@dataclass
class GrayscaleConfig:
    """Cấu hình cho grayscale deployment"""
    name: str
    embedding_model: str
    chunking_strategy: str
    chunk_size: int
    weight: float = 0.5  # % traffic

class GrayscaleRouter:
    """
    Router cho grayscale deployment
    Phân phối traffic theo trọng số cấu hình
    """
    
    def __init__(self, configs: List[GrayscaleConfig]):
        self.configs = configs
        self.results = defaultdict(list)
        
        # Validate weights
        total_weight = sum(c.weight for c in configs)
        if abs(total_weight - 1.0) > 0.01:
            # Normalize
            for c in configs:
                c.weight /= total_weight
    
    def get_config_for_request(self, request_id: str = None) -> GrayscaleConfig:
        """
        Chọn config dựa trên weighted random
        """
        if request_id is None:
            request_id = str(random.random())
        
        # Deterministic selection dựa trên request_id
        random.seed(hash(request_id) % (2**32))
        
        r = random.random()
        cumulative = 0
        
        for config in self.configs:
            cumulative += config.weight
            if r <= cumulative:
                return config
        
        return self.configs[-1]
    
    def record_result(self, config_name: str, query: str, answer: str, latency_ms: float):
        """Ghi lại kết quả cho một config"""
        self.results[config_name].append({
            "query": query,
            "answer": answer,
            "latency_ms": latency_ms,
            "timestamp": datetime.now().isoformat()
        })
    
    def get_comparison_report(self) -> Dict[str, Any]:
        """
        Tạo báo cáo so sánh giữa các configs
        """
        report = {}
        
        for config_name, results in self.results.items():
            if not results:
                continue
            
            latencies = [r["latency_ms"] for r in results]
            
            report[config_name] = {
                "request_count": len(results),
                "avg_latency_ms": np.mean(latencies),
                "p50_latency_ms": np.percentile(latencies, 50),
                "p95_latency_ms": np.percentile(latencies, 95),
                "p99_latency_ms": np.percentile(latencies, 99),
                "sample_answers": [r["answer"][:100] for r in results[:3]]
            }
        
        return report


============================================


THIẾT LẬP GRAYSCALE TEST


============================================



Config cũ (baseline)

old_config

Tài nguyên liên quan
Bài viết liên quan