Tôi vẫn nhớ rõ cái đêm tháng 11 năm ngoái — ngày siêu sale 11.11 của khách hàng thương mại điện tử. Lúc 11 giờ tối, hệ thống chatbot AI tư vấn sản phẩm bắt đầu trả về timeout liên tục. 15 phút sau, hàng nghìn khách hàng xếp hàng chờ đợi. Thiệt hại ước tính khoảng 45.000 USD doanh thu bị mất trong đợt peak traffic đó. Sau sự cố đó, tôi đã dành 3 tuần để xây dựng hệ thống load testing hoàn chỉnh — và bài viết này sẽ chia sẻ toàn bộ kiến thức tôi đã tích lũy được.

Tại Sao Load Testing Quan Trọng Với API AI?

Khác với API truyền thống chỉ trả về JSON đơn giản, các API AI như chat completion hay embedding generation có những đặc điểm riêng biệt đòi hỏi chiến lược test khác:

So Sánh Locust và k6: Tool Nào Phù Hợp?

Tiêu chí Locust (Python) k6 (JavaScript/Golang)
Ngôn ngữ lập trình Python — dễ học, ecosystem AI phong phú JavaScript/TypeScript — familiar với developer web
Phân tích kết quả Giao diện web dashboard tích hợp Built-in reporter + tích hợp Grafana/InfluxDB
Protocol hỗ trợ HTTP, gRPC, MQTT HTTP, WebSocket, gRPC, Socket.io
Performance Single-threaded (cần gevent), ~1000 RPS/cơ bản Golang-based, có thể đạt 10.000+ RPS
Distributed testing Hỗ trợ master-worker sẵn có Chạy multiple instances thủ công
Learning curve Thấp — viết task bằng Python Trung bình — cần làm quen với VU concept
AI API testing ⭐⭐⭐⭐⭐ — Dễ dùng async/await với aiohttp ⭐⭐⭐ — Cần wrapper cho streaming
Chi phí Miễn phí, open-source Miễn phí (open-source), có bản Cloud trả phí

Kết luận thực chiến của tôi: Với dự án AI, Locust là lựa chọn tối ưu vì Python là ngôn ngữ chính của hệ sinh thái AI. Bạn có thể tái sử dụng các function xử lý response trực tiếp. Tuy nhiên, nếu team đã quen với JavaScript hoặc cần performance cực cao, k6 là giải pháp thay thế xứng đáng.

Setup Môi Trường

# Cài đặt Locust
pip install locust aiohttp asyncio-limiter python-dotenv


Cài đặt k6 (macOS)

brew install k6


Cài đặt k6 (Ubuntu/Debian)

sudo gpg -k
sudo gpg --no-default-keyring --keyring /usr/share/keyrings/k6-archive-keyring.gpg --keyserver hkp://keyserver.ubuntu.com:80 --recv-keys C5AD17C747E3415A3642D57D77C6C491D6AC1D69
echo "deb [signed-by=/usr/share/keyrings/k6-archive-keyring.gpg] https://dl.k6.io/deb stable main" | sudo tee /etc/apt/sources.list.d/k6.list
sudo apt-get update
sudo apt-get install k6

Locust: Load Testing Với AI API

Cấu Hình Base Project

# config.py
import os
from dotenv import load_dotenv

load_dotenv()


HolySheep AI Configuration

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")


Test Configuration

MODEL_CONFIG = {
    "chat": {
        "model": "gpt-4.1",
        "max_tokens": 500,
        "temperature": 0.7
    },
    "embedding": {
        "model": "text-embedding-3-small",
        "dimensions": 1536
    },
    "vision": {
        "model": "gpt-4o-mini",
        "max_tokens": 300
    }
}


Load Test Limits

RATE_LIMIT_RPM = 500  # Requests per minute
RATE_LIMIT_TPM = 150000  # Tokens per minute
CONCURRENT_CONNECTIONS = 50

Locust Task File Chính

# locustfile.py
import os
import json
import time
import random
from locust import task, between, events
from locust.runners import MasterRunner
import aiohttp
import asyncio
from typing import Dict, List, Optional


Configuration

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")


Test Data - Common e-commerce queries

E_COMMERCE_QUERIES = [
    "Tôi muốn tìm áo thun nam size L, màu đen, giá dưới 300k",
    "So sánh iPhone 15 Pro Max và Samsung S24 Ultra",
    "Cách chọn kem chống nắng cho da dầu?",
    "Gợi ý quà tặng sinh nhật cho mẹ 50 tuổi",
    "Tại sao laptop gaming bị lag khi chơi game?",
    "Đánh giá máy lọc nước Karofi có tốt không?",
    "Hướng dẫn setup workspace cho lập trình viên",
    "So sánh các loại sữa rửa mặt cho da nhạy cảm"
]

RAG_CONTEXT_EXAMPLES = [
    "Sản phẩm: Áo sơ mi nam VNXK, chất liệu 100% cotton, giá 450.000đ. Đánh giá 4.5/5 sao từ 2.340 khách hàng. Bảo hành 12 tháng.",
    "Sản phẩm: Giày thể thao Nike Air Max 90, hàng chính hãng, giá 2.850.000đ. Đang có khuyến mãi giảm 15%. Bảo hành 6 tháng.",
    "Chính sách đổi trả: Áp dụng trong 30 ngày với sản phẩm chưa qua sử dụng, còn nguyên tem mác. Hoàn tiền trong 5-7 ngày làm việc."
]

class AIAbstractUser:
    """Base class với shared utilities cho tất cả AI test scenarios"""
    
    def __init__(self):
        self.request_count = 0
        self.token_count = 0
        self.error_count = 0
        self.cost_estimate = 0.0
    
    def build_headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        }
    
    def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """Tính chi phí theo pricing HolySheep 2026"""
        pricing = {
            "gpt-4.1": {"input": 8.0, "output": 8.0},  # $8/MTok
            "gpt-4o-mini": {"input": 0.75, "output": 3.0},
            "claude-sonnet-4.5": {"input": 15.0, "output": 15.0},  # $15/MTok
            "gemini-2.5-flash": {"input": 2.50, "output": 2.50},  # $2.50/MTok
            "deepseek-v3.2": {"input": 0.42, "output": 0.42},  # $0.42/MTok
            "text-embedding-3-small": {"input": 0.02, "output": 0.0}
        }
        rates = pricing.get(model, {"input": 3.0, "output": 3.0})
        cost = (input_tokens / 1_000_000) * rates["input"] + \
               (output_tokens / 1_000_000) * rates["output"]
        return cost
    
    def log_request(self, endpoint: str, latency: float, status: int, tokens: int = 0, cost: float = 0.0):
        """Log metrics cho analytics"""
        self.request_count += 1
        self.token_count += tokens
        self.cost_estimate += cost
        print(f"[{endpoint}] Status: {status} | Latency: {latency:.2f}ms | Tokens: {tokens} | Cost: ${cost:.6f}")


class ChatCompletionUser(AIAbstractUser):
    """Test chatbot AI tư vấn sản phẩm thương mại điện tử"""
    
    def __init__(self):
        super().__init__()
        self.session = None
        self.conversation_history = []
    
    def on_start(self):
        """Khởi tạo HTTP session khi user bắt đầu"""
        import httpx
        self.client = httpx.Client(
            base_url=BASE_URL,
            headers=self.build_headers(),
            timeout=60.0
        )
        self.conversation_history = [
            {"role": "system", "content": "Bạn là trợ lý tư vấn bán hàng chuyên nghiệp cho cửa hàng thương mại điện tử. Hãy trả lời ngắn gọn, hữu ích và thân thiện."}
        ]
    
    def on_stop(self):
        """Cleanup session"""
        if hasattr(self, 'client'):
            self.client.close()
    
    @task(5)  # Weight cao hơn - ưu tiên scenario này
    def test_chat_completion_standard(self):
        """Test standard chat completion với single query"""
        query = random.choice(E_COMMERCE_QUERIES)
        self.conversation_history.append({"role": "user", "content": query})
        
        payload = {
            "model": "gpt-4.1",
            "messages": self.conversation_history[-5:],  # Keep last 5 messages
            "max_tokens": 500,
            "temperature": 0.7,
            "stream": False
        }
        
        start_time = time.time()
        try:
            response = self.client.post("/chat/completions", json=payload)
            latency = (time.time() - start_time) * 1000
            
            if response.status_code == 200:
                data = response.json()
                assistant_message = data["choices"][0]["message"]["content"]
                usage = data.get("usage", {})
                
                input_tokens = usage.get("prompt_tokens", 0)
                output_tokens = usage.get("completion_tokens", 0)
                cost = self.calculate_cost("gpt-4.1", input_tokens, output_tokens)
                
                self.conversation_history.append({"role": "assistant", "content": assistant_message})
                self.log_request("chat/completions", latency, 200, input_tokens + output_tokens, cost)
            else:
                self.error_count += 1
                self.log_request("chat/completions", latency, response.status_code)
                
        except Exception as e:
            self.error_count += 1
            print(f"Error: {str(e)}")
    
    @task(2)
    def test_rag_chat_completion(self):
        """Test RAG-powered chat với context injection"""
        context = random.choice(RAG_CONTEXT_EXAMPLES)
        query = random.choice([
            "Sản phẩm này có bảo hành không?",
            "Giá này đang có khuyến mãi không?",
            "Chính sách đổi trả như thế nào?"
        ])
        
        payload = {
            "model": "gpt-4.1",
            "messages": [
                {"role": "system", "content": f"Sử dụng thông tin sau để trả lời: {context}"},
                {"role": "user", "content": query}
            ],
            "max_tokens": 300,
            "temperature": 0.3
        }
        
        start_time = time.time()
        try:
            response = self.client.post("/chat/completions", json=payload)
            latency = (time.time() - start_time) * 1000
            
            if response.status_code == 200:
                data = response.json()
                output_tokens = data.get("usage", {}).get("completion_tokens", 0)
                cost = self.calculate_cost("gpt-4.1", 100, output_tokens)  # Estimate
                self.log_request("rag-chat", latency, 200, 100 + output_tokens, cost)
            else:
                self.error_count += 1
                
        except Exception as e:
            self.error_count += 1
    
    @task(1)
    def test_vision_analysis(self):
        """Test multi-modal: phân tích hình ảnh sản phẩm"""
        payload = {
            "model": "gpt-4o-mini",
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": "Mô tả sản phẩm trong ảnh này"
                        },
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": "https://example.com/sample-product.jpg"
                            }
                        }
                    ]
                }
            ],
            "max_tokens": 200
        }
        
        start_time = time.time()
        try:
            response = self.client.post("/chat/completions", json=payload)
            latency = (time.time() - start_time) * 1000
            
            if response.status_code == 200:
                self.log_request("vision", latency, 200, 500, 0.001)
            else:
                self.error_count += 1
                
        except Exception as e:
            self.error_count += 1


class EmbeddingUser(AIAbstractUser):
    """Test embedding API cho RAG và semantic search"""
    
    def __init__(self):
        super().__init__()
    
    def on_start(self):
        import httpx
        self.client = httpx.Client(
            base_url=BASE_URL,
            headers=self.build_headers(),
            timeout=30.0
        )
    
    def on_stop(self):
        if hasattr(self, 'client'):
            self.client.close()
    
    @task(3)
    def test_embedding_generation(self):
        """Test embedding generation cho semantic search"""
        texts = random.sample(E_COMMERCE_QUERIES, k=3)
        
        payload = {
            "model": "text-embedding-3-small",
            "input": texts,
            "dimensions": 1536
        }
        
        start_time = time.time()
        try:
            response = self.client.post("/embeddings", json=payload)
            latency = (time.time() - start_time) * 1000
            
            if response.status_code == 200:
                data = response.json()
                total_tokens = sum(item.get("tokens", 0) for item in data.get("usage", {}).values())
                cost = self.calculate_cost("text-embedding-3-small", total_tokens, 0)
                self.log_request("embeddings", latency, 200, total_tokens, cost)
            else:
                self.error_count += 1
                
        except Exception as e:
            self.error_count += 1
    
    @task(1)
    def test_batch_embedding(self):
        """Test batch embedding với 100+ texts"""
        texts = [f"Mô tả sản phẩm #{i}: chất lượng tốt, giá rẻ" for i in range(100)]
        
        payload = {
            "model": "text-embedding-3-small",
            "input": texts
        }
        
        start_time = time.time()
        try:
            response = self.client.post("/embeddings", json=payload)
            latency = (time.time() - start_time) * 1000
            
            if response.status_code == 200:
                self.log_request("batch-embeddings", latency, 200, 5000, 0.0001)
            else:
                self.error_count += 1
                
        except Exception as e:
            self.error_count += 1



Event handlers cho reporting

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    print("=" * 60)
    print("🚀 BẮT ĐẦU LOAD TEST - HolySheep AI API")
    print(f"📍 Base URL: {BASE_URL}")
    print(f"⏰ Thời gian: {time.strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 60)

@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    print("=" * 60)
    print("📊 KẾT QUẢ LOAD TEST")
    print("=" * 60)

Chạy Locust Load Test

# Chạy Locust với UI dashboard
locust -f locustfile.py \
    --host=https://api.holysheep.ai \
    --users=100 \
    --spawn-rate=10 \
    --run-time=5m \
    --headless \
    --csv=results/load_test


Chạy với Master-Worker cho distributed testing

locust -f locustfile.py --master
locust -f locustfile.py --worker --master-host=192.168.1.100


Web UI mode (truy cập http://localhost:8089)

locust -f locustfile.py --host=https://api.holysheep.ai


Phân tích kết quả

python analyze_results.py --csv=results/load_test_stats.csv

k6: Load Testing Với JavaScript

Script k6 Cho AI API Testing

// k6_ai_load_test.js
// Load test configuration for HolySheep AI API

import http from 'k6/http';
import { check, sleep, group } from 'k6';
import { Rate, Trend, Counter } from 'k6/metrics';
import { SharedArray } from 'k6/data';

// Custom metrics
const chatLatency = new Trend('chat_completion_latency');
const embeddingLatency = new Trend('embedding_latency');
const errorRate = new Rate('error_rate');
const tokenCounter = new Counter('total_tokens');
const costTracker = new Trend('estimated_cost');

// Test configuration
const BASE_URL = 'https://api.holysheep.ai/v1';
const API_KEY = __ENV.HOLYSHEEP_API_KEY || 'YOUR_HOLYSHEEP_API_KEY';

// Pricing configuration (HolySheep 2026)
const PRICING = {
    'gpt-4.1': { input: 8.0, output: 8.0 },           // $8/MTok
    'gpt-4o-mini': { input: 0.75, output: 3.0 },       // $0.75 input, $3 output
    'claude-sonnet-4.5': { input: 15.0, output: 15.0 }, // $15/MTok
    'gemini-2.5-flash': { input: 2.50, output: 2.50 }, // $2.50/MTok
    'deepseek-v3.2': { input: 0.42, output: 0.42 },   // $0.42/MTok
    'text-embedding-3-small': { input: 0.02, output: 0.0 } // $0.02/MTok
};

// Test data
const ECOMMERCE_QUERIES = [
    'Tìm áo phông nam chất cotton, giá dưới 300k',
    'So sánh laptop Dell XPS và MacBook Pro M3',
    'Đánh giá kem chống nắng La Roche-Posay',
    'Cách chọn giày chạy bộ cho người mới bắt đầu',
    'Gợi ý quà tặng Noel cho bạn gái',
];

// Shared data for VUs
const queries = new SharedArray('queries', function() {
    return ECOMMERCE_QUERIES;
});

const headers = {
    'Authorization': Bearer ${API_KEY},
    'Content-Type': 'application/json',
};

// Helper function to calculate cost
function calculateCost(model, inputTokens, outputTokens) {
    const rates = PRICING[model] || { input: 3.0, output: 3.0 };
    const inputCost = (inputTokens / 1000000) * rates.input;
    const outputCost = (outputTokens / 1000000) * rates.output;
    return inputCost + outputCost;
}

// Test scenarios
export const options = {
    scenarios: {
        // Warmup phase
        warmup: {
            executor: 'constant-vus',
            vus: 5,
            duration: '30s',
            tags: { test_phase: 'warmup' },
        },
        // Ramp up phase
        ramp_up: {
            executor: 'ramp-vus',
            startVUs: 5,
            stages: [
                { duration: '1m', target: 20 },
                { duration: '2m', target: 50 },
                { duration: '1m', target: 100 },
            ],
            tags: { test_phase: 'ramp_up' },
        },
        // Peak load test
        peak_load: {
            executor: 'constant-vus',
            vus: 100,
            duration: '5m',
            tags: { test_phase: 'peak' },
        },
        // Stress test
        stress_test: {
            executor: 'ramp-vus',
            startVUs: 100,
            stages: [
                { duration: '2m', target: 200 },
                { duration: '3m', target: 200 },
                { duration: '1m', target: 0 },
            ],
            tags: { test_phase: 'stress' },
        },
    },
    thresholds: {
        // P95 latency should be under 3000ms
        'chat_completion_latency': ['p(95)<3000'],
        'embedding_latency': ['p(95)<500'],
        // Error rate should be less than 1%
        'error_rate': ['rate<0.01'],
    },
};

export default function() {
    // Test 1: Chat Completion (70% of traffic)
    group('Chat Completion', () => {
        const query = queries[Math.floor(Math.random() * queries.length)];
        
        const payload = JSON.stringify({
            model: 'gpt-4.1',
            messages: [
                { role: 'system', content: 'Bạn là trợ lý bán hàng chuyên nghiệp.' },
                { role: 'user', content: query }
            ],
            max_tokens: 500,
            temperature: 0.7,
        });

        const startTime = Date.now();
        const response = http.post(${BASE_URL}/chat/completions, payload, {
            headers: headers,
            timeout: '60s',
        });
        const latency = Date.now() - startTime;

        chatLatency.add(latency);

        const checkResult = check(response, {
            'status is 200': (r) => r.status === 200,
            'has content': (r) => r.json('choices[0].message.content') !== undefined,
            'has usage': (r) => r.json('usage') !== undefined,
        });

        if (!checkResult) {
            errorRate.add(1);
            console.error(Chat Error: ${response.status} - ${response.body});
        } else {
            errorRate.add(0);
            const usage = response.json('usage');
            const inputTokens = usage.prompt_tokens || 0;
            const outputTokens = usage.completion_tokens || 0;
            const cost = calculateCost('gpt-4.1', inputTokens, outputTokens);
            
            tokenCounter.add(inputTokens + outputTokens);
            costTracker.add(cost);
        }
    });

    sleep(1);

    // Test 2: Embedding Generation (30% of traffic)
    group('Embedding Generation', () => {
        const batchTexts = queries.slice(0, 5);
        
        const payload = JSON.stringify({
            model: 'text-embedding-3-small',
            input: batchTexts,
        });

        const startTime = Date.now();
        const response = http.post(${BASE_URL}/embeddings, payload, {
            headers: headers,
            timeout: '30s',
        });
        const latency = Date.now() - startTime;

        embeddingLatency.add(latency);

        const checkResult = check(response, {
            'status is 200': (r) => r.status === 200,
            'has embeddings': (r) => r.json('data') !== undefined,
        });

        if (!checkResult) {
            errorRate.add(1);
        } else {
            errorRate.add(0);
            tokenCounter.add(1000); // Estimate
        }
    });

    sleep(2);
}

// Summary handler
export function handleSummary(data) {
    return {
        'stdout': textSummary(data, { indent: ' ', enableColors: true }),
        'summary.json': JSON.stringify(data, null, 2),
    };
}

function textSummary(data, options) {
    let summary = '\n' + '='.repeat(60) + '\n';
    summary += '📊 KẾT QUẢ LOAD TEST - HolySheep AI\n';
    summary += '='.repeat(60) + '\n\n';
    
    summary += ⏱️  Thời gian test: ${data.test_params.duration}s\n;
    summary += 👥 Peak VUs: ${data.metrics.vus_max?.value || 0}\n;
    summary += 📝 Total Requests: ${data.metrics.http_reqs?.values?.passes + data.metrics.http_reqs?.values?.fails || 0}\n;
    summary += ❌ Error Rate: ${((data.metrics.error_rate?.values?.rate || 0) * 100).toFixed(2)}%\n\n;
    
    summary += '📈 Latency Statistics:\n';
    summary +=    • Chat Completion P95: ${(data.metrics.chat_completion_latency?.values?.['p(95)'] || 0).toFixed(2)}ms\n;
    summary +=    • Embedding P95: ${(data.metrics.embedding_latency?.values?.['p(95)'] || 0).toFixed(2)}ms\n;
    summary +=    • Average Chat: ${(data.metrics.chat_completion_latency?.values?.avg || 0).toFixed(2)}ms\n\n;
    
    summary += '💰 Cost Estimation:\n';
    summary +=    • Estimated Total: $${(data.metrics.estimated_cost?.values?.avg || 0).toFixed(6)}\n;
    summary +=    • Total Tokens: ${data.metrics.total_tokens?.values?.count || 0}\n\n;
    
    return summary;
}

Chạy k6 Load Test

# Basic run với k6
k6 run k6_ai_load_test.js \
    --env HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY \
    --summary-export=results/summary.json


Chạy với Cloud execution (cần k6 Cloud account)

k6 cloud k6_ai_load_test.js --env HOLYSHEEP_API_KEY=YOUR_API_KEY


Output to Grafana/InfluxDB

k6 run k6_ai_load_test.js \
    --out influxdb=http://localhost:8086/k6db \
    --env HOLYSHEEP_API_KEY=YOUR_API_KEY


Run specific scenario only

k6 run k6_ai_load_test.js \
    --env HOLYSHEEP_API_KEY=YOUR_API_KEY \
    --tag test_phase=peak


Quick smoke test (1 user, short duration)

k6 run k6_ai_load_test.js \
    --env HOLYSHEEP_API_KEY=YOUR_API_KEY \
    --vus 1 --duration 30s

So Sánh Chi Phí: HolySheep vs OpenAI

Model OpenAI (Input) OpenAI (Output) HolySheep (Input) HolySheep (Output) Tiết kiệm
GPT-4.1 $15/MTok $60/MTok $8/MTok $8/MTok ~85%
Claude Sonnet 4.5 $15/MTok $75/MTok $15/MTok $15/MTok ~80%
Gemini 2.5 Flash $2.50/MTok $10/MTok $2.50/MTok $2.50/MTok ~75%
DeepSeek V3.2 $0.55/MTok $2.75/MTok $0.42/MTok $0.42/MTok ~85%
Embedding 3 Small $0.02/MTok - $0.02/MTok - ~0%

Phù hợp / Không phù hợp với ai

✅ Nên dùng HolySheep AI khi: