Trong thế giới AI đang thay đổi từng ngày, việc chọn đúng mô hình ngôn ngữ cho ứng dụng của bạn không chỉ là vấn đề kỹ thuật — mà là quyết định kinh doanh quan trọng. Với mức giá chênh lệch 36 lần giữa DeepSeek V3.2 ($0.42/MTok) và Claude Sonnet 4.5 ($15/MTok), một chiến lược A/B testing tốt có thể tiết kiệm hàng ngàn đô la mỗi tháng.

Bài viết này tôi sẽ chia sẻ kinh nghiệm thực chiến xây dựng hệ thống A/B test đa mô hình tại HolySheep AI — nền tảng tích hợp nhiều provider AI hàng đầu với chi phí tối ưu.

Bảng So Sánh Chi Phí Các Mô Hình 2026

Dữ liệu giá thực tế tính đến 2026 đã được xác minh:

Mô hình Output ($/MTok) 10M token/tháng 100M token/tháng
Claude Sonnet 4.5 $15.00 $150 $1,500
GPT-4.1 $8.00 $80 $800
Gemini 2.5 Flash $2.50 $25 $250
DeepSeek V3.2 $0.42 $4.20 $42
HolySheep (tất cả) ¥1=$1 (85%+ tiết kiệm) Tiết kiệm 85%+ Tiết kiệm 85%+

A/B Testing Đa Mô Hình Là Gì?

A/B test đa mô hình là quá trình so sánh hiệu suất của nhiều LLM API trên cùng một tập dữ liệu để đưa ra quyết định dựa trên dữ liệu thay vì linh cảm. Điều quan trọng là bạn cần test đồng thời:

Kiến Trúc Hệ Thống A/B Test

Tôi đã xây dựng kiến trúc này cho một startup AI, xử lý khoảng 50 triệu token mỗi tháng. Kết quả? Tiết kiệm $3,200/tháng chỉ bằng việc chọn đúng mô hình cho từng use case.

# Hệ thống A/B Test đa mô hình - Core Engine
import asyncio
import aiohttp
import time
import hashlib
from dataclasses import dataclass
from typing import List, Dict, Optional
from collections import defaultdict
import json

@dataclass
class ModelConfig:
    name: str
    provider: str
    base_url: str  # Sử dụng HolySheep unified endpoint
    model_id: str
    api_key: str
    cost_per_mtok: float  # Chi phí $/MTok

@dataclass
class TestResult:
    model: str
    prompt_tokens: int
    completion_tokens: int
    latency_ms: float
    success: bool
    error: Optional[str]
    timestamp: float

class MultiModelABTester:
    """
    Hệ thống A/B test đa mô hình - X supports multiple providers
    thông qua unified API endpoint của HolySheep
    """
    
    # Cấu hình các mô hình - tất cả qua HolySheep unified endpoint
    MODELS = {
        'gpt4.1': ModelConfig(
            name='GPT-4.1',
            provider='openai',
            base_url='https://api.holysheep.ai/v1',
            model_id='gpt-4.1',
            api_key='YOUR_HOLYSHEEP_API_KEY',
            cost_per_mtok=8.00
        ),
        'claude_sonnet_4.5': ModelConfig(
            name='Claude Sonnet 4.5',
            provider='anthropic',
            base_url='https://api.holysheep.ai/v1',
            model_id='claude-sonnet-4-20250514',
            api_key='YOUR_HOLYSHEEP_API_KEY',
            cost_per_mtok=15.00
        ),
        'gemini_flash': ModelConfig(
            name='Gemini 2.5 Flash',
            provider='google',
            base_url='https://api.holysheep.ai/v1',
            model_id='gemini-2.5-flash',
            api_key='YOUR_HOLYSHEEP_API_KEY',
            cost_per_mtok=2.50
        ),
        'deepseek_v3': ModelConfig(
            name='DeepSeek V3.2',
            provider='deepseek',
            base_url='https://api.holysheep.ai/v1',
            model_id='deepseek-v3.2',
            api_key='YOUR_HOLYSHEEP_API_KEY',
            cost_per_mtok=0.42
        )
    }
    
    def __init__(self):
        self.results: Dict[str, List[TestResult]] = defaultdict(list)
        self.session: Optional[aiohttp.ClientSession] = None
    
    async def initialize(self):
        """Khởi tạo aiohttp session với connection pooling"""
        timeout = aiohttp.ClientTimeout(total=60)
        self.session = aiohttp.ClientSession(timeout=timeout)
    
    async def close(self):
        """Đóng session và cleanup"""
        if self.session:
            await self.session.close()
    
    def _create_hash(self, prompt: str, model: str) -> str:
        """Tạo hash để đảm bảo test的一致性"""
        data = f"{prompt}:{model}"
        return hashlib.sha256(data.encode()).hexdigest()[:8]
    
    async def call_model(self, config: ModelConfig, prompt: str) -> TestResult:
        """Gọi model qua unified HolySheep endpoint"""
        start_time = time.time()
        
        try:
            headers = {
                'Authorization': f'Bearer {config.api_key}',
                'Content-Type': 'application/json'
            }
            
            # Unified request format cho tất cả providers
            payload = {
                'model': config.model_id,
                'messages': [
                    {'role': 'user', 'content': prompt}
                ],
                'temperature': 0.7,
                'max_tokens': 2048
            }
            
            async with self.session.post(
                f'{config.base_url}/chat/completions',
                headers=headers,
                json=payload
            ) as response:
                elapsed_ms = (time.time() - start_time) * 1000
                
                if response.status == 200:
                    data = await response.json()
                    usage = data.get('usage', {})
                    
                    return TestResult(
                        model=config.name,
                        prompt_tokens=usage.get('prompt_tokens', 0),
                        completion_tokens=usage.get('completion_tokens', 0),
                        latency_ms=elapsed_ms,
                        success=True,
                        error=None,
                        timestamp=time.time()
                    )
                else:
                    error_text = await response.text()
                    return TestResult(
                        model=config.name,
                        prompt_tokens=0,
                        completion_tokens=0,
                        latency_ms=elapsed_ms,
                        success=False,
                        error=f'HTTP {response.status}: {error_text}',
                        timestamp=time.time()
                    )
        
        except asyncio.TimeoutError:
            return TestResult(
                model=config.name,
                prompt_tokens=0,
                completion_tokens=0,
                latency_ms=(time.time() - start_time) * 1000,
                success=False,
                error='Timeout after 60s',
                timestamp=time.time()
            )
        except Exception as e:
            return TestResult(
                model=config.name,
                prompt_tokens=0,
                completion_tokens=0,
                latency_ms=(time.time() - start_time) * 1000,
                success=False,
                error=str(e),
                timestamp=time.time()
            )
    
    async def run_ab_test(
        self,
        prompt: str,
        models: List[str],
        runs: int = 5
    ) -> Dict[str, List[TestResult]]:
        """Chạy A/B test cho nhiều models"""
        tasks = []
        
        for model_key in models:
            if model_key not in self.MODELS:
                print(f"⚠️ Model '{model_key}' không tồn tại, bỏ qua")
                continue
            
            config = self.MODELS[model_key]
            
            # Chạy nhiều lần để lấy trung bình
            for _ in range(runs):
                tasks.append(self.call_model(config, prompt))
        
        # Execute all requests concurrently
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Group results by model
        grouped = defaultdict(list)
        for i, result in enumerate(results):
            if isinstance(result, TestResult):
                model_key = models[i % len(models)]
                grouped[model_key].append(result)
        
        return dict(grouped)
    
    def generate_report(self, results: Dict[str, List[TestResult]]) -> str:
        """Tạo báo cáo chi tiết"""
        report_lines = ["=" * 60]
        report_lines.append("📊 A/B TEST REPORT - Multi-Model Comparison")
        report_lines.append("=" * 60)
        
        summary = []
        
        for model_key, model_results in results.items():
            successful = [r for r in model_results if r.success]
            
            if not successful:
                report_lines.append(f"\n❌ {model_key}: Tất cả requests thất bại")
                continue
            
            avg_latency = sum(r.latency_ms for r in successful) / len(successful)
            total_tokens = sum(
                r.prompt_tokens + r.completion_tokens 
                for r in successful
            )
            
            model_config = self.MODELS.get(model_key)
            cost_per_run = (total_tokens / 1_000_000) * model_config.cost_per_mtok if model_config else 0
            
            summary.append({
                'model': model_key,
                'success_rate': len(successful) / len(model_results) * 100,
                'avg_latency': avg_latency,
                'total_tokens': total_tokens,
                'estimated_cost': cost_per_run
            })
            
            report_lines.append(f"\n✅ {model_key}")
            report_lines.append(f"   Success Rate: {len(successful)/len(model_results)*100:.1f}%")
            report_lines.append(f"   Avg Latency: {avg_latency:.1f}ms")
            report_lines.append(f"   Total Tokens: {total_tokens:,}")
            report_lines.append(f"   Est. Cost: ${cost_per_run:.4f}")
        
        # Sort by latency for recommendation
        summary.sort(key=lambda x: x['avg_latency'])
        
        report_lines.append("\n" + "=" * 60)
        report_lines.append("🏆 RECOMMENDATION")
        report_lines.append("=" * 60)
        
        if summary:
            best = summary[0]
            report_lines.append(f"\n🥇 Best Latency: {best['model']} ({best['avg_latency']:.1f}ms)")
            
            # Best cost
            summary.sort(key=lambda x: x['estimated_cost'])
            cheapest = summary[0]
            report_lines.append(f"💰 Best Cost: {cheapest['model']} (${cheapest['estimated_cost']:.4f})")
        
        return "\n".join(report_lines)



Cách sử dụng

async def main():
    tester = MultiModelABTester()
    await tester.initialize()
    
    try:
        # Test prompt
        test_prompt = "Giải thích quantum computing trong 3 câu"
        
        # Run test trên tất cả models
        results = await tester.run_ab_test(
            prompt=test_prompt,
            models=['deepseek_v3', 'gemini_flash', 'gpt4.1', 'claude_sonnet_4.5'],
            runs=5
        )
        
        # Generate report
        report = tester.generate_report(results)
        print(report)
        
    finally:
        await tester.close()

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

Chiến Lược Phân Bổ Traffic A/B

Khi tôi triển khai hệ thống này cho production, key insight là không phải lúc nào model đắt nhất cũng tốt nhất. DeepSeek V3.2 rẻ nhưng với nhiều task đơn giản, nó cho kết quả tương đương GPT-4.1.

# Router thông minh - Dynamic Model Selection
from enum import Enum
from typing import Callable
import random

class TaskType(Enum):
    SIMPLE_SUMMARIZATION = "simple_summarize"
    COMPLEX_REASONING = "complex_reasoning"
    CODE_GENERATION = "code_gen"
    CREATIVE_WRITING = "creative"
    GENERAL = "general"

class SmartRouter:
    """
    Router thông minh phân bổ traffic dựa trên:
    1. Loại task
    2. Kết quả A/B test trước đó
    3. Budget constraints
    """
    
    # Chiến lược phân bổ dựa trên A/B test results
    # Format: {task_type: {model: weight}}
    ROUTING_STRATEGY = {
        TaskType.SIMPLE_SUMMARIZATION: {
            'deepseek_v3': 0.6,      # 60% - rẻ + đủ tốt
            'gemini_flash': 0.3,     # 30%
            'gpt4.1': 0.1,           # 10%
        },
        TaskType.COMPLEX_REASONING: {
            'gpt4.1': 0.4,
            'claude_sonnet_4.5': 0.35,
            'deepseek_v3': 0.25,
        },
        TaskType.CODE_GENERATION: {
            'gpt4.1': 0.45,
            'claude_sonnet_4.5': 0.45,
            'deepseek_v3': 0.10,
        },
        TaskType.CREATIVE_WRITING: {
            'claude_sonnet_4.5': 0.5,
            'gpt4.1': 0.4,
            'gemini_flash': 0.1,
        },
        TaskType.GENERAL: {
            'deepseek_v3': 0.35,
            'gemini_flash': 0.35,
            'gpt4.1': 0.20,
            'claude_sonnet_4.5': 0.10,
        }
    }
    
    # Budget cap per day (USD)
    DAILY_BUDGET = 50.0
    current_spend = 0.0
    
    @classmethod
    def classify_task(cls, prompt: str) -> TaskType:
        """Phân loại task dựa trên keywords"""
        prompt_lower = prompt.lower()
        
        if any(kw in prompt_lower for kw in ['tóm tắt', 'summarize', 'tổng kết']):
            return TaskType.SIMPLE_SUMMARIZATION
        elif any(kw in prompt_lower for kw in ['giải thích', 'phân tích', 'reasoning']):
            return TaskType.COMPLEX_REASONING
        elif any(kw in prompt_lower for kw in ['viết code', 'function', 'python', 'javascript']):
            return TaskType.CODE_GENERATION
        elif any(kw in prompt_lower for kw in ['sáng tạo', 'creative', 'viết', 'write']):
            return TaskType.CREATIVE_WRITING
        else:
            return TaskType.GENERAL
    
    @classmethod
    def select_model(cls, task_type: TaskType, prompt: str = "") -> str:
        """Chọn model dựa trên weighted routing + budget check"""
        
        # Check budget
        if cls.current_spend >= cls.DAILY_BUDGET:
            # Over budget, force to cheapest
            return 'deepseek_v3'
        
        # Get routing weights for this task type
        weights = cls.ROUTING_STRATEGY.get(task_type, cls.ROUTING_STRATEGY[TaskType.GENERAL])
        
        # Weighted random selection
        models = list(weights.keys())
        probabilities = list(weights.values())
        
        selected = random.choices(models, weights=probabilities, k=1)[0]
        
        # Budget safeguard - never exceed daily limit
        estimated_cost = {
            'deepseek_v3': 0.00042,      # ~$0.42 per 1M tokens
            'gemini_flash': 0.0025,
            'gpt4.1': 0.008,
            'claude_sonnet_4.5': 0.015,
        }.get(selected, 0.001)
        
        if cls.current_spend + estimated_cost > cls.DAILY_BUDGET:
            return 'deepseek_v3'
        
        return selected
    
    @classmethod
    def record_spend(cls, model: str, tokens: int):
        """Ghi nhận chi phí để track budget"""
        cost_per_token = {
            'deepseek_v3': 0.42 / 1_000_000,
            'gemini_flash': 2.50 / 1_000_000,
            'gpt4.1': 8.00 / 1_000_000,
            'claude_sonnet_4.5': 15.00 / 1_000_000,
        }.get(model, 0)
        
        cls.current_spend += tokens * cost_per_token
    
    @classmethod
    def reset_daily_budget(cls):
        """Reset budget cho ngày mới"""
        cls.current_spend = 0.0



Integration với A/B Tester

class ABRouter:
    """
    Kết hợp A/B testing với Smart Routing
    - Phase 1: Pure A/B (uniform distribution)
    - Phase 2: Smart routing (dựa trên kết quả)
    """
    
    def __init__(self, ab_tester: MultiModelABTester):
        self.ab_tester = ab_tester
        self.phase = "ab_testing"  # or "production"
        self.min_ab_runs = 100  # Minimum runs before switching to smart routing
    
    async def process_request(self, prompt: str) -> TestResult:
        """Process request dựa trên phase hiện tại"""
        
        if self.phase == "ab_testing":
            # Phase 1: Chạy A/B test với uniform distribution
            models = list(self.ab_tester.MODELS.keys())
            selected_model = random.choice(models)
            config = self.ab_tester.MODELS[selected_model]
            
            result = await self.ab_tester.call_model(config, prompt)
            
            # Check if enough data collected
            total_runs = sum(
                len(runs) for runs in self.ab_tester.results.values()
            )
            
            if total_runs >= self.min_ab_runs:
                self._optimize_routing()
                self.phase = "production"
                print(f"✅ Đã chuyển sang production mode với {total_runs} A/B runs")
            
            return result
        
        else:
            # Phase 2: Smart routing
            task_type = SmartRouter.classify_task(prompt)
            selected_model = SmartRouter.select_model(task_type, prompt)
            config = self.ab_tester.MODELS[selected_model]
            
            result = await self.ab_tester.call_model(config, prompt)
            
            if result.success:
                SmartRouter.record_spend(
                    selected_model,
                    result.prompt_tokens + result.completion_tokens
                )
            
            return result
    
    def _optimize_routing(self):
        """Tối ưu hóa routing dựa trên A/B test results"""
        # Analyze results and update SmartRouter.ROUTING_STRATEGY
        # Implementation details...
        print("🔧 Đang tối ưu hóa routing strategy...")



Ví dụ sử dụng đầy đủ

async def production_example():
    # Khởi tạo
    tester = MultiModelABTester()
    await tester.initialize()
    
    router = ABRouter(tester)
    
    # Batch test prompts
    test_prompts = [
        "Tóm tắt bài viết sau: AI đang thay đổi thế giới...",
        "Giải thích thuật toán Dijkstra",
        "Viết function Python sort một array",
        "Sáng tác một bài thơ về mùa xuân",
    ]
    
    try:
        for i, prompt in enumerate(test_prompts):
            result = await router.process_request(prompt)
            task_type = SmartRouter.classify_task(prompt)
            
            print(f"\n[{i+1}] Task: {task_type.value}")
            print(f"    Selected: {result.model}")
            print(f"    Latency: {result.latency_ms:.1f}ms")
            print(f"    Success: {result.success}")
            print(f"    Current Spend: ${SmartRouter.current_spend:.4f}")
    
    finally:
        await tester.close()

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

Đo Lường và Phân Tích Kết Quả

Trong quá trình triển khai, tôi phát hiện ra 3 metrics quan trọng nhất cần theo dõi:

# Metrics Dashboard - Real-time monitoring
from dataclasses import dataclass
from typing import Dict, List
from datetime import datetime, timedelta
import statistics

@dataclass
class ModelMetrics:
    model: str
    total_requests: int = 0
    successful_requests: int = 0
    failed_requests: int = 0
    total_latencies: List[float] = None
    total_cost: float = 0.0
    quality_scores: List[float] = None
    
    def __post_init__(self):
        self.total_latencies = []
        self.quality_scores = []
    
    @property
    def success_rate(self) -> float:
        if self.total_requests == 0:
            return 0.0
        return self.successful_requests / self.total_requests * 100
    
    @property
    def avg_latency(self) -> float:
        if not self.total_latencies:
            return 0.0
        return statistics.mean(self.total_latencies)
    
    @property
    def p99_latency(self) -> float:
        if not self.total_latencies:
            return 0.0
        sorted_latencies = sorted(self.total_latencies)
        idx = int(len(sorted_latencies) * 0.99)
        return sorted_latencies[min(idx, len(sorted_latencies) - 1)]
    
    @property
    def avg_quality(self) -> float:
        if not self.quality_scores:
            return 0.0
        return statistics.mean(self.quality_scores)
    
    @property
    def cost_efficiency(self) -> float:
        """Quality per dollar spent"""
        if self.total_cost == 0 or self.avg_quality == 0:
            return 0.0
        return self.avg_quality / self.total_cost

class MetricsDashboard:
    """Dashboard theo dõi metrics theo thời gian thực"""
    
    def __init__(self):
        self.metrics: Dict[str, ModelMetrics] = {}
        self.start_time = datetime.now()
    
    def record_request(
        self,
        model: str,
        latency_ms: float,
        success: bool,
        tokens: int,
        quality_score: float = None,
        cost_per_mtok: float = 0.0
    ):
        """Ghi nhận một request"""
        if model not in self.metrics:
            self.metrics[model] = ModelMetrics(model=model)
        
        m = self.metrics[model]
        m.total_requests += 1
        
        if success:
            m.successful_requests += 1
            m.total_latencies.append(latency_ms)
            m.total_cost += (tokens / 1_000_000) * cost_per_mtok
            
            if quality_score is not None:
                m.quality_scores.append(quality_score)
        else:
            m.failed_requests += 1
    
    def generate_summary(self) -> str:
        """Tạo bảng tóm tắt metrics"""
        lines = []
        lines.append("\n" + "=" * 80)
        lines.append("📊 METRICS DASHBOARD")
        lines.append(f"⏱️  Runtime: {datetime.now() - self.start_time}")
        lines.append("=" * 80)
        
        # Sort by cost efficiency
        sorted_models = sorted(
            self.metrics.values(),
            key=lambda x: x.cost_efficiency,
            reverse=True
        )
        
        header = f"{'Model':<20} {'Success':<10} {'Avg Lat':<12} {'P99 Lat':<12} {'Avg Quality':<15} {'Cost':<12} {'Efficiency':<15}"
        lines.append("\n" + header)
        lines.append("-" * 96)
        
        for m in sorted_models:
            lines.append(
                f"{m.model:<20} "
                f"{m.success_rate:>6.1f}%   "
                f"{m.avg_latency:>8.1f}ms  "
                f"{m.p99_latency:>8.1f}ms  "
                f"{m.avg_quality:>12.2f}   "
                f"${m.total_cost:>8.4f}  "
                f"{m.cost_efficiency:>12.2f}"
            )
        
        # Budget analysis
        total_cost = sum(m.total_cost for m in self.metrics.values())
        total_requests = sum(m.total_requests for m in self.metrics.values())
        
        lines.append("\n" + "=" * 80)
        lines.append("💰 BUDGET ANALYSIS")
        lines.append("=" * 80)
        lines.append(f"Total Requests: {total_requests:,}")
        lines.append(f"Total Cost: ${total_cost:.4f}")
        
        # Potential savings if using best model for all
        if sorted_models:
            best_model = sorted_models[0]
            optimal_cost = (total_requests * (best_model.total_cost / max(best_model.total_requests, 1)))
            potential_savings = total_cost - optimal_cost
            
            lines.append(f"\nOptimal Cost (using {best_model.model} only): ${optimal_cost:.4f}")
            lines.append(f"💵 Potential Savings: ${potential_savings:.4f}")
            lines.append(f"📈 Savings Percentage: {potential_savings/total_cost*100:.1f}%")
        
        return "\n".join(lines)



Quality evaluation using LLM-as-Judge

class QualityEvaluator:
    """
    Đánh giá chất lượng output sử dụng LLM-as-Judge
    """
    
    SYSTEM_PROMPT = """Bạn là một chuyên gia đánh giá chất lượng câu trả lời.
Đánh giá câu trả lời dựa trên:
1. Accuracy (độ chính xác) - 0-10
2. Relevance (tính liên quan) - 0-10
3. Coherence (tính mạch lạc) - 0-10

Trả về JSON format: {"accuracy": X, "relevance": X, "coherence": X}"""

    @classmethod
    async def evaluate(
        cls,
        ab_tester: MultiModelABTester,
        prompt: str,
        response: str
    ) -> float:
        """Đánh giá chất lượng response"""
        
        evaluation_prompt = f"""Prompt: {prompt}

Response: {response}

Hãy đánh giá chất lượng response này và trả về JSON format."""

        # Use best model for evaluation (typically Claude or GPT-4)
        config = ab_tester.MODELS['claude_sonnet_4.5']
        result = await ab_tester.call_model(config, evaluation_prompt)
        
        if result.success:
            # Parse JSON response
            try:
                import json
                # Extract JSON from response (handle potential formatting)
                response_text = ""  # Would need to get actual response
                scores = json.loads(response_text)
                return (scores['accuracy'] + scores['relevance'] + scores['coherence']) / 3
            except:
                return 5.0  # Default score if parsing fails
        
        return 0.0



Full integration example

async def complete_ab_test_pipeline():
    """
    Pipeline hoàn chỉnh: A/B Test → Evaluate → Optimize
    """
    print("🚀 Starting A/B Test Pipeline\n")
    
    # Initialize
    tester = MultiModelABTester()
    await tester.initialize()
    dashboard = MetricsDashboard()
    
    # Test cases
    test_cases = [
        {
            'prompt': 'Tóm tắt: AI là công nghệ...',
            'expected_type': TaskType.SIMPLE_SUMMARIZATION
        },
        {
            'prompt': 'Giải thích: Machine Learning hoạt động như thế nào?',
            'expected_type': TaskType.COMPLEX_REASONING
        },
        {
            'prompt': 'Viết code Python để đọc file CSV',
            'expected_type': TaskType.CODE_GENERATION
        },
    ]
    
    try:
        # Phase 1: A/B Testing
        print("📊 Phase 1: Running A/B Tests...")
        
        for test in test_cases:
            results = await tester.run_ab_test(
                prompt=test['prompt'],
                models=['deepseek_v3', 'gemini_flash', 'gpt4.1', 'claude_sonnet_4.5'],
                runs=10
            )
            
            # Record metrics
            for model_key, model_results in results.items():
                for result in model_results:
                    config = tester.MODELS[model_key]
                    tokens = result.prompt_tokens + result.completion_tokens
                    
                    dashboard.record_request(
                        model=model_key,
                        latency_ms=result.latency_ms,
                        success=result.success,
                        tokens=tokens,
                        cost_per_mtok=config.cost_per_mtok
                    )
        
        # Print results
        print(dashboard.generate_summary())
        
        # Phase 2: Export for further analysis
        print("\n📁 Exporting results to JSON...")
        
        export_data = {
            'test_date': datetime.now().isoformat(),
            'models': {}
        }
        
        for model_key, metrics in dashboard.metrics.items():
            export_data['models'][model_key] = {
                'total_requests': metrics.total_requests,
                'success_rate': metrics.success_rate,
                'avg_latency': metrics.avg_latency,
                'p99_latency': metrics.p99_latency,
                'total_cost': metrics.total_cost,
                'cost_efficiency': metrics.cost_efficiency
            }
        
        # Save to file
        with open('ab_test_results.json', 'w') as f:
            json.dump(export_data, f, indent=2)
        
        print("✅ Results saved to ab_test_results.json")
        
    finally:
        await tester.close()

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

Phù hợp / không

Tài nguyên liên quan

Bài viết liên quan