Tôi vẫn nhớ rõ ngày hôm đó - deadline sản phẩm còn 3 ngày, hệ thống báo chi phí API tăng 340% so với tháng trước. Cuộc gọi điện với kế toán kéo dài 2 tiếng để giải thích tại sao "máy tính lại tốn tiền như vậy". Kể từ đó, tôi xây dựng một hệ thống theo dõi token chi tiết đến từng mili-giây — và hôm nay, tôi sẽ chia sẻ toàn bộ phương pháp này với bạn.
Vấn đề thực tế: Khi hóa đơn API trở thành "hộp đen"
Khi sử dụng các AI programming assistant như Claude, GPT-4 hay Gemini thông qua API, hầu hết developers gặp phải một vấn đề nan giải: không biết token đã tiêu thụ ở đâu và tại sao.
- Không phân biệt được input token vs output token
- Không biết request nào tiêu tốn nhiều token nhất
- Không thể dự đoán chi phí khi scale hệ thống
- Thiếu dữ liệu để tối ưu prompt
Kiến trúc theo dõi Token toàn diện
Dưới đây là kiến trúc tôi đã triển khai thực tế, cho phép theo dõi chi tiết từng byte token trong hệ thống.
1. Wrapper Class cho API Calls
import time
import json
from datetime import datetime
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
import httpx
@dataclass
class TokenUsage:
"""Lớp lưu trữ thông tin sử dụng token"""
timestamp: str
model: str
input_tokens: int
output_tokens: int
total_tokens: int
latency_ms: float
cost_usd: float
request_id: Optional[str] = None
prompt_preview: str = ""
status: str = "success"
error_message: Optional[str] = None
class HolySheepTokenTracker:
"""Theo dõi token consumption cho HolySheep API"""
BASE_URL = "https://api.holysheep.ai/v1"
# Bảng giá tham khảo (USD per 1M tokens) - cập nhật 2026
PRICING = {
"gpt-4.1": {"input": 2.00, "output": 8.00}, # $8/M output
"claude-sonnet-4.5": {"input": 3.00, "output": 15.00}, # $15/M output
"gemini-2.5-flash": {"input": 0.10, "output": 2.50}, # $2.50/M
"deepseek-v3.2": {"input": 0.10, "output": 0.42}, # $0.42/M
}
def __init__(self, api_key: str):
self.api_key = api_key
self.usage_history: List[TokenUsage] = []
self.session = httpx.Client(timeout=60.0)
def calculate_cost(self, model: str, input_tokens: int,
output_tokens: int) -> float:
"""Tính chi phí USD cho một request"""
if model not in self.PRICING:
raise ValueError(f"Model '{model}' không có trong bảng giá")
rates = self.PRICING[model]
input_cost = (input_tokens / 1_000_000) * rates["input"]
output_cost = (output_tokens / 1_000_000) * rates["output"]
return round(input_cost + output_cost, 6) # Chính xác 6 chữ số thập phân
def chat_completion(self, model: str, messages: List[Dict],
**kwargs) -> Tuple[str, TokenUsage]:
"""Gọi API với tracking chi tiết"""
start_time = time.perf_counter()
usage_record = TokenUsage(
timestamp=datetime.utcnow().isoformat(),
model=model,
input_tokens=0,
output_tokens=0,
total_tokens=0,
latency_ms=0,
cost_usd=0,
prompt_preview=messages[-1]["content"][:100] if messages else ""
)
try:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
**kwargs
}
response = self.session.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload
)
elapsed_ms = (time.perf_counter() - start_time) * 1000
if response.status_code == 200:
data = response.json()
usage = data.get("usage", {})
usage_record.input_tokens = usage.get("prompt_tokens", 0)
usage_record.output_tokens = usage.get("completion_tokens", 0)
usage_record.total_tokens = usage.get("total_tokens", 0)
usage_record.latency_ms = round(elapsed_ms, 2)
usage_record.cost_usd = self.calculate_cost(
model,
usage_record.input_tokens,
usage_record.output_tokens
)
usage_record.request_id = data.get("id")
usage_record.status = "success"
self.usage_history.append(usage_record)
return data["choices"][0]["message"]["content"], usage_record
else:
usage_record.status = "error"
usage_record.error_message = f"HTTP {response.status_code}: {response.text}"
self.usage_history.append(usage_record)
raise Exception(usage_record.error_message)
except httpx.TimeoutException:
usage_record.status = "timeout"
usage_record.error_message = "Connection timeout (>60s)"
usage_record.latency_ms = 60000
self.usage_history.append(usage_record)
raise
except Exception as e:
usage_record.status = "error"
usage_record.error_message = str(e)
self.usage_history.append(usage_record)
raise
def get_cost_summary(self, days: int = 30) -> Dict:
"""Tổng hợp chi phí theo khoảng thời gian"""
cutoff = datetime.utcnow().timestamp() - (days * 86400)
recent = [u for u in self.usage_history
if datetime.fromisoformat(u.timestamp).timestamp() > cutoff]
if not recent:
return {"total_cost": 0, "total_requests": 0, "avg_latency_ms": 0}
by_model = {}
for record in recent:
if record.model not