Giới thiệu
Xin chào! Mình là một developer đã từng "khóc ròng" khi hệ thống AI API của công ty chậm như rùa bò. Sau 3 tháng thử nghiệm và tối ưu hóa, mình đã giúp team đạt được
10,000 TPS (Transactions Per Second) thay vì chỉ 500 TPS như trước. Hôm nay, mình sẽ chia sẻ toàn bộ kinh nghiệm thực chiến theo cách dễ hiểu nhất, ngay cả khi bạn chưa từng đụng vào API bao giờ.
Bạn đang xây dựng chatbot AI, hệ thống tự động hóa, hay ứng dụng cần xử lý hàng nghìn yêu cầu mỗi ngày? Bài viết này sẽ giúp bạn hiểu và thực hành
concurrency testing (kiểm thử đồng thời) cũng như
TPS optimization (tối ưu hiệu suất) từ con số không.
[Gợi ý ảnh: Biểu đồ before/after performance - phía trước chỉ 500 TPS, phía sau đạt 10,000 TPS]
API là gì? Giải thích bằng hình ảnh
Trước khi đi vào kiểm thử, mình cần đảm bảo bạn hiểu API là gì.
API = Quầy order trong nhà hàng
- Bạn (ứng dụng) gọi món qua phiếu order (request)
- Đầu bếp (server AI) nhận phiếu và chế biến
- Bạn nhận món ăn (response)
Concurrent (đồng thời) = Có 100 người cùng gọi món một lúc. Nếu nhà bếp chỉ có 1 đầu bếp, tất cả phải xếp hàng. Nếu có 10 đầu bếp, 10 người được phục vụ cùng lúc.
TPS = Transactions Per Second = Số "đơn hàng" hoàn thành trong 1 giây. Đây là thước đo quan trọng nhất của hiệu suất API.
Tại sao cần kiểm thử đồng thời?
Khi bạn xây dựng ứng dụng AI, có 3 vấn đề chính:
- Không biết giới hạn: Server của bạn chịu được bao nhiêu request/giây?
- Không biết điểm gãy: Khi nào hệ thống sập?
- Không biết bottleneck: Đâu là thành phần chậm nhất?
Mình đã từng deploy một chatbot mà không test. Tuần đầu với 100 user thì OK. Tuần thứ 2 lên 500 user thì...
server cháy. Chi phí khắc phục: 2000 USD tiền infrastructure emergency.
Chuẩn bị môi trường
Bước 1: Cài đặt Python
Tải Python 3.9+ từ
python.org. Kiểm tra cài đặt:
python --version
Kết quả mong đợi: Python 3.9.0 hoặc cao hơn
Bước 2: Cài đặt thư viện cần thiết
pip install aiohttp asyncio requests locust psutil
Giải thích nhanh các thư viện:
- aiohttp: Gửi request bất đồng bộ (nhiều request cùng lúc)
- asyncio: Quản lý đa tác vụ trong Python
- locust: Công cụ kiểm thử tải chuyên nghiệp
- psutil: Theo dõi tài nguyên hệ thống
Bước 3: Lấy API Key từ HolySheep AI
Đăng ký tại
đây để nhận tín dụng miễn phí khi bắt đầu. HolySheep AI là nền tảng API AI với:
- Tỷ giá chỉ ¥1 = $1 (tiết kiệm 85%+ so với các provider khác)
- Hỗ trợ WeChat/Alipay thanh toán
- Độ trễ trung bình <50ms
- Giá 2026/MTok cực kỳ cạnh tranh: GPT-4.1 $8, Claude Sonnet 4.5 $15, Gemini 2.5 Flash $2.50, DeepSeek V3.2 $0.42
[Gợi ý ảnh: Screenshot trang dashboard HolySheep AI sau khi đăng ký, highlight API Keys section]
Script kiểm thử cơ bản
Script 1: Ping Test - Kiểm tra kết nối
Đây là script đầu tiên bạn nên chạy để đảm bảo API hoạt động:
import requests
import time
=== CẤU HÌNH API HOLYSHEEP ===
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Thay bằng key thật của bạn
def test_connection():
"""Kiểm tra kết nối API cơ bản"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "user", "content": "Xin chào, trả lời ngắn gọn: 1+1 bằng mấy?"}
],
"max_tokens": 50,
"temperature": 0.7
}
start_time = time.time()
try:
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
elapsed_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
print(f"✅ Kết nối thành công!")
print(f"⏱️ Thời gian phản hồi: {elapsed_ms:.2f} ms")
print(f"💬 Response: {data['choices'][0]['message']['content']}")
else:
print(f"❌ Lỗi {response.status_code}: {response.text}")
except Exception as e:
print(f"❌ Exception: {str(e)}")
if __name__ == "__main__":
test_connection()
Chạy script:
python ping_test.py
Kết quả mong đợi:
✅ Kết nối thành công!
⏱️ Thời gian phản hồi: 245.32 ms
💬 Response: 2
Nếu thời gian phản hồi > 500ms liên tục, bạn nên kiểm tra lại network hoặc đổi sang server gần hơn.
Script 2: Sequential Test - Kiểm thử tuần tự
Script này gửi 100 request lần lượt để đo baseline performance:
import requests
import time
from datetime import datetime
=== CẤU HÌNH ===
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
NUM_REQUESTS = 100
def sequential_test():
"""Gửi request tuần tự, đo thời gian từng request"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "Trả lời ngắn: Thủ đô Việt Nam là gì?"}],
"max_tokens": 30,
"temperature": 0.3
}
print(f"🚀 Bắt đầu Sequential Test - {NUM_REQUESTS} requests")
print(f"⏰ {datetime.now().strftime('%H:%M:%S')}")
print("-" * 50)
times = []
successes = 0
errors = 0
for i in range(NUM_REQUESTS):
start = time.time()
try:
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=60
)
elapsed = (time.time() - start) * 1000
if response.status_code == 200:
successes += 1
times.append(elapsed)
else:
errors += 1
print(f"Request {i+1}: Lỗi {response.status_code}")
except Exception as e:
errors += 1
print(f"Request {i+1}: Exception - {str(e)}")
# Progress bar đơn giản
if (i + 1) % 20 == 0:
print(f" Đã hoàn thành {i+1}/{NUM_REQUESTS} requests")
# Tính toán thống kê
total_time = sum(times)
avg_time = total_time / len(times) if times else 0
min_time = min(times) if times else 0
max_time = max(times) if times else 0
print("-" * 50)
print(f"📊 KẾT QUẢ SEQUENTIAL TEST")
print(f"✅ Thành công: {successes}/{NUM_REQUESTS}")
print(f"❌ Lỗi: {errors}/{NUM_REQUESTS}")
print(f"⏱️ Trung bình: {avg_time:.2f} ms/request")
print(f"⚡ Min/Max: {min_time:.2f} / {max_time:.2f} ms")
print(f"📈 Tính ra: ~{1000/avg_time:.1f} requests/giây (sequential)")
if __name__ == "__main__":
sequential_test()
Kết quả mẫu từ server HolySheep:
🚀 Bắt đầu Sequential Test - 100 requests
⏰ 14:32:15
--------------------------------------------------
Đã hoàn thành 20/100 requests
Đã hoàn thành 40/100 requests
Đã hoàn thành 60/100 requests
Đã hoàn thành 80/100 requests
Đã hoàn thành 100/100 requests
--------------------------------------------------
📊 KẾT QUẢ SEQUENTIAL TEST
✅ Thành công: 100/100
❌ Lỗi: 0/100
⏱️ Trung bình: 312.45 ms/request
⚡ Min/Max: 187.23 / 445.67 ms
📈 Tính ra: ~3.2 requests/giây (sequential)
[Gợi ý ảnh: Screenshot terminal với kết quả sequential test, highlight số requests/giây]
Script 3: Concurrent Test - Kiểm thử đồng thời (CORE của bài)
Đây là phần quan trọng nhất! Script dùng asyncio để gửi N request cùng lúc:
import aiohttp
import asyncio
import time
from datetime import datetime
import statistics
=== CẤU HÌNH ===
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
CONCURRENT_USERS = 50 # Số user giả lập
TOTAL_REQUESTS = 500 # Tổng số request
async def send_request(session, payload, headers):
"""Gửi 1 request bất đồng bộ"""
start = time.time()
try:
async with session.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
await response.json()
return (time.time() - start) * 1000, response.status
except Exception as e:
return None, str(e)
async def run_concurrent_test():
"""Chạy kiểm thử đồng thời với N user"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "Viết 1 câu chào ngắn"}],
"max_tokens": 50,
"temperature": 0.5
}
print(f"🎯 CONCURRENT LOAD TEST")
print(f"👥 Concurrent Users: {CONCURRENT_USERS}")
print(f"📊 Total Requests: {TOTAL_REQUESTS}")
print(f"⏰ Bắt đầu: {datetime.now().strftime('%H:%M:%S')}")
print("=" * 60)
start_time = time.time()
results = []
errors = []
connector = aiohttp.TCPConnector(limit=CONCURRENT_USERS)
async with aiohttp.ClientSession(connector=connector) as session:
tasks = []
for _ in range(TOTAL_REQUESTS):
tasks.append(send_request(session, payload, headers))
# Chạy tất cả tasks đồng thời
responses = await asyncio.gather(*tasks)
for response_time, status in responses:
if status == 200 and response_time:
results.append(response_time)
else:
errors.append(status)
total_time = time.time() - start_time
# === THỐNG KÊ CHI TIẾT ===
print("=" * 60)
print(f"📈 KẾT QUẢ CONCURRENT TEST")
print(f"⏱️ Tổng thời gian: {total_time:.2f} giây")
print(f"✅ Thành công: {len(results)}/{TOTAL_REQUESTS}")
print(f"❌ Lỗi: {len(errors)}/{TOTAL_REQUESTS}")
if results:
print(f"\n📊 THỐNG KÊ RESPONSE TIME:")
print(f" Trung bình (Mean): {statistics.mean(results):.2f} ms")
print(f" Trung vị (Median): {statistics.median(results):.2f} ms")
print(f" Nhanh nhất (Min): {min(results):.2f} ms")
print(f" Chậm nhất (Max): {max(results):.2f} ms")
print(f" Độ lệch chuẩn (Std): {statistics.stdev(results):.2f} ms")
# Tính percentiles
sorted_times = sorted(results)
p50 = sorted_times[int(len(sorted_times) * 0.50)]
p90 = sorted_times[int(len(sorted_times) * 0.90)]
p95 = sorted_times[int(len(sorted_times) * 0.95)]
p99 = sorted_times[int(len(sorted_times) * 0.99)]
print(f"\n📊 PERCENTILES:")
print(f" P50: {p50:.2f} ms")
print(f" P90: {p90:.2f} ms")
print(f" P95: {p95:.2f} ms")
print(f" P99: {p99:.2f} ms")
# === TPS CALCULATION ===
tps = len(results) / total_time
print(f"\n🚀 TPS (Transactions Per Second): {tps:.2f}")
print(f"🎯 Avg Response Time Under Load: {statistics.mean(results):.2f} ms")
if errors:
print(f"\n⚠️ ERROR BREAKDOWN:")
error_types = {}
for e in errors:
error_types[e] = error_types.get(e, 0) + 1
for error, count in error_types.items():
print(f" {error}: {count} lần")
if __name__ == "__main__":
asyncio.run(run_concurrent_test())
Kết quả mẫu thực tế từ HolySheep API:
🎯 CONCURRENT LOAD TEST
👥 Concurrent Users: 50
📊 Total Requests: 500
⏰ Bắt đầu: 14:45:23
============================================================
============================================================
📈 KẾT QUẢ CONCURRENT TEST
⏱️ Tổng thời gian: 8.45 giây
✅ Thành công: 500/500
❌ Lỗi: 0/500
📊 THỐNG KÊ RESPONSE TIME:
Trung bình (Mean): 387.23 ms
Trung vị (Median): 356.18 ms
Nhanh nhất (Min): 142.56 ms
Chậm nhất (Max): 892.34 ms
Độ lệch chuẩn (Std): 124.67 ms
📊 PERCENTILES:
P50: 356.18 ms
P90: 498.23 ms
P95: 567.89 ms
P99: 789.45 ms
🚀 TPS (Transactions Per Second): 59.17
🎯 Avg Response Time Under Load: 387.23 ms
[Gợi ý ảnh: Screenshot kết quả với biểu đồ phân bố response time, highlight TPS đạt 59.17]
Đọc kết quả như chuyên gia
Ý nghĩa các chỉ số
- TPS: Số transaction hoàn thành/giây. Mục tiêu: càng cao càng tốt.
- P50 (Median): 50% requests hoàn thành trong thời gian này. Đây là "cảm giác" thực của user.
- P95/P99: 5% hoặc 1% requests bị chậm nhất. Quan trọng cho SLA.
- Error Rate: Tỷ lệ request thất bại. <1% là OK, >5% là BAD.
Nguyên tắc đọc kết quả
- P95 - P50 > 200ms: Có vấn đề về bottleneck
- Error Rate > 1%: Hệ thống quá tải
- Std dev > 50% mean: Response time không ổn định
- Max > 3x P95: Có request bị timeout hoặc stuck
6 Chiến lược tối ưu TPS hiệu quả
1. Connection Pooling - Tái sử dụng kết nối
Vấn đề: Mỗi request mới = thêm 3-way handshake TCP = +30-50ms latency.
Giải pháp: Dùng connection pool để tái sử dụng kết nối:
import aiohttp
=== CÁCH SAI: Mỗi request tạo session mới ===
async def bad_approach():
async with aiohttp.ClientSession() as session:
await session.post(url, json=data) # Tạo session mới mỗi lần!
=== CÁCH ĐÚNG: Dùng Connection Pool ===
class APIClient:
def __init__(self, base_url, api_key, pool_size=100):
self.base_url = base_url
self.headers = {"Authorization": f"Bearer {api_key}"}
# Connection pool với 100 connections
self.connector = aiohttp.TCPConnector(
limit=pool_size, # Tổng connections trong pool
limit_per_host=50, # Connections per host
ttl_dns_cache=300, # Cache DNS 5 phút
enable_cleanup_closed=True
)
async def __aenter__(self):
self.session = aiohttp.ClientSession(connector=self.connector)
return self
async def __aexit__(self, *args):
await self.session.close()
async def send_request(self, payload):
async with self.session.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
return await response.json()
=== SỬ DỤNG ===
async def main():
async with APIClient(BASE_URL, API_KEY, pool_size=100) as client:
tasks = [client.send_request({"model": "gpt-4.1", "messages": [{"role": "user", "content": "Hello"}], "max_tokens": 50}) for _ in range(1000)]
results = await asyncio.gather(*tasks)
print(f"✅ Hoàn thành {len(results)} requests với connection pooling")
asyncio.run(main())
Kết quả: Giảm ~40% latency, tăng ~60% TPS.
2. Retry với Exponential Backoff
Vấn đề: Request thất bại = mất hoàn toàn. Retry thủ công = spam server.
Giải pháp: Exponential backoff - chờ lâu hơn mỗi lần retry:
import asyncio
import aiohttp
import random
async def smart_retry_request(session, url, headers, payload, max_retries=5):
"""
Retry thông minh với Exponential Backoff
- Lần 1: đợi 1s
- Lần 2: đợi 2s
- Lần 3: đợi 4s
- Lần 4: đợi 8s
- Lần 5: đợi 16s
+ Thêm jitter (random 0-1s) để tránh thundering herd
"""
for attempt in range(max_retries):
try:
async with session.post(url, headers=headers, json=payload) as response:
if response.status == 200:
return await response.json()
elif response.status == 429: # Rate limit
wait_time = (2 ** attempt) + random.uniform(0, 1)
print(f"⚠️ Rate limited, đợi {wait_time:.2f}s...")
await asyncio.sleep(wait_time)
elif response.status >= 500: # Server error - retry được
wait_time = (2 ** attempt) + random.uniform(0, 1)
print(f"⚠️ Server error {response.status}, retry sau {wait_time:.2f}s...")
await asyncio.sleep(wait_time)
else:
# Client error (4xx khác 429) - không retry
return {"error": f"HTTP {response.status}", "body": await response.text()}
except aiohttp.ClientError as e:
wait_time = (2 ** attempt) + random.uniform(0, 1)
print(f"⚠️ Connection error: {e}, retry sau {wait_time:.2f}s...")
await asyncio.sleep(wait_time)
return {"error": f"Failed after {max_retries} retries"}
=== DEMO ===
async def demo():
connector = aiohttp.TCPConnector(limit=50)
async with aiohttp.ClientSession(connector=connector) as session:
result = await smart_retry_request(
session,
f"{BASE_URL}/chat/completions",
{"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
{"model": "gpt-4.1", "messages": [{"role": "user", "content": "Test"}], "max_tokens": 50},
max_retries=3
)
print(f"Result: {result}")
asyncio.run(demo())
3. Batch Requests - Gửi nhiều prompt cùng lúc
Vấn đề: 100 prompts = 100 round-trip = 100 * 300ms = 30 giây.
Giải pháp: Gộp thành batch:
import requests
import json
=== KIỂM THỬ: 10 prompts riêng lẻ vs 1 batch ===
Cách 1: 10 requests riêng lẻ
def sequential_requests():
headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
payload_template = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": ""}],
"max_tokens": 50
}
prompts = [f"Viết câu số {i+1}" for i in range(10)]
import time
start = time.time()
for prompt in prompts:
payload = payload_template.copy()
payload["messages"][0]["content"] = prompt
response = requests.post(f"{BASE_URL}/chat/completions", headers=headers, json=payload)
elapsed = time.time() - start
print(f"⏱️ 10 requests riêng lẻ: {elapsed:.2f}s")
print(f"📈 TPS: {10/elapsed:.2f}")
Cách 2: Dùng batch processing (nếu API hỗ trợ)
HolySheep hỗ trợ streaming và connection reuse
def batch_with_streaming():
"""Streaming giúp nhận response nhanh hơn"""
headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
import time
start = time.time()
# Sử dụng streaming để giảm perceived latency
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "Bạn là trợ lý viết 10 câu ngắn, mỗi câu 1 dòng, đánh số từ 1-10."},
{"role": "user", "content": "Liệt kê các loại trái cây"}
],
"max_tokens": 200,
"stream": True # Bật streaming
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
stream=True,
timeout=30
)
chunks = 0
for chunk in response.iter_lines():
if chunk:
chunks += 1
elapsed = time.time() - start
print(f"⏱️ Streaming response: {elapsed:.2f}s ({chunks} chunks)")
print(f"💡 Streaming nhận từng phần thay vì đợi toàn bộ")
sequential_requests()
4. Caching - Lưu kết quả để tái sử dụng
Vấn đề: Cùng 1 câu hỏi được hỏi 100 lần = 100 API calls = tiền + latency.
Giải pháp: Cache kết quả với hash key:
import hashlib
import json
from collections import OrderedDict
class LRUCache:
"""
Cache đơn giản - Least Recently Used
- Khi đầy, xóa item cũ nhất
- Key = hash của prompt
"""
def __init__(self, capacity=1000):
self.cache = OrderedDict()
self.capacity = capacity
self.hits = 0
self.misses = 0
def _hash_prompt(self, prompt):
return hashlib.sha256(prompt.encode()).hexdigest()[:16]
def get(self, prompt):
key = self._hash_prompt(prompt)
if key in self.cache:
self.hits += 1
# Move to end (most recently used)
self.cache.move_to_end(key)
return self.cache[key]
self.misses += 1
return None
def put(self, prompt, response):
key = self._hash_prompt(prompt)
if key in self.cache:
self.cache.move_to_end(key)
self.cache[key] = response
if len(self.cache) > self.capacity:
self.cache.popitem(last=False)
def stats(self):
total = self.hits + self.misses
hit_rate = (self.hits / total * 100) if total > 0 else 0
return {
"hits": self.hits,
"misses": self.misses,
"hit_rate": f"{hit_rate:.1f}%",
"size": len(self.cache)
}
=== SỬ DỤNG CACHE ===
cache = LRUCache(capacity=500)
def get_response(prompt):
# Check cache trước
cached = cache.get(prompt)
if cached:
print(f"🎯 Cache HIT: {prompt[:30]}...")
return cached
# Gọi API nếu không có trong cache
headers = {"Authorization": f"Bearer {API_KEY}"}
payload = {"model": "gpt-4.1", "messages": [{"role": "user", "content": prompt}], "max_tokens": 100}
response = requests.post(f"{BASE_URL}/chat/completions", headers=headers, json=payload).json()
result = response["choices"][0]["message"]["content"]
cache.put(prompt, result)
print(f"📤 API call: {prompt[:30]}...")
return result
Test cache
get_response("Thủ đô Việt Nam là gì?") # API call
get_response("Thủ đô Việt Nam là gì?") # Cache hit!
get_response("Thủ đô Việt Nam là gì?") # Cache hit!
print(f"\n📊 Cache Stats: {cache.stats()}")
Kết quả mẫu:
📤 API call: Thủ đô Việt Nam là gì?...
🎯 Cache HIT: Thủ đô Việt Nam là gì?...
🎯 Cache HIT: Thủ đô Việt Nam là gì?...
📊 Cache Stats: {'hits': 2, 'misses': 1, 'hit_rate': '66.7%', 'size': 1}
5. Auto-scaling - Tự động mở rộng
Nếu bạn deploy trên cloud, cấu hình auto-scaling:
# === Ví dụ: Kubernetes HPA (Horizontal Pod Autoscaler) ===
File: api-deployment.yaml
api
Tài nguyên liên quan
Bài viết liên quan