Tôi vẫn nhớ rõ ngày đó — 3 giờ sáng, deadline sản phẩm còn 4 tiếng, và tôi nhận được notification từ hệ thống CI/CD: "Model inference output mismatch between staging và production". Hai phiên bản cùng một prompt, cùng model, nhưng tạo ra hai kết quả hoàn toàn khác nhau. Đó là lần đầu tiên tôi thực sự hiểu tại sao reproducibility verification lại quan trọng đến vậy.
Bài viết này sẽ hướng dẫn bạn cách xây dựng hệ thống xác minh reproducibility cho AI model inference một cách chuyên nghiệp, kèm theo những kinh nghiệm thực chiến mà tôi đã đúc kết qua hơn 2 năm làm việc với các mô hình AI tại HolySheep AI.
Tại Sao Reproducibility Verification Lại Quan Trọng?
Trong production environment, reproducibility không chỉ là "nice to have" mà là yêu cầu bắt buộc. Khi bạn deploy model lên nhiều server, mỗi request phải trả về kết quả nhất quán. Nếu không có verification system, bạn sẽ gặp những vấn đề nghiêm trọng:
- User experience không nhất quán — Cùng một câu hỏi nhưng được trả lời khác nhau
- Debugging nightmare — Không thể tái hiện lỗi để fix
- Compliance violation — Không đáp ứng được yêu cầu audit
- Revenue impact — Ảnh hưởng trực tiếp đến doanh thu
Kiến Trúc Verification System
Tôi đã thiết kế hệ thống verification với 4 layer chính, mỗi layer giải quyết một khía cạnh khác nhau của reproducibility:
1. Hash-based Content Verification
Đây là layer foundation — nơi chúng ta tạo deterministic hash từ output và lưu trữ để so sánh. Với HolySheep AI, tôi sử dụng streaming response và hash generation đồng thời để đảm bảo performance.
import hashlib
import json
import time
from dataclasses import dataclass
from typing import Optional, Dict, Any
import requests
@dataclass
class InferenceResult:
content: str
hash: str
model: str
latency_ms: float
timestamp: float
request_id: str
@dataclass
class VerificationReport:
is_reproducible: bool
expected_hash: str
actual_hash: str
similarity_score: float
error_message: Optional[str]
class HolySheepReproducibilityVerifier:
"""Verification system cho AI model inference reproducibility"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.hash_history: Dict[str, str] = {}
def generate_deterministic_hash(self, content: str, model: str,
temperature: float, seed: int) -> str:
"""
Tạo hash deterministic từ content và các tham số.
Sử dụng SHA-256 với salt từ model parameters.
"""
hash_input = json.dumps({
"content": content,
"model": model,
"temperature": temperature,
"seed": seed
}, sort_keys=True)
return hashlib.sha256(hash_input.encode()).hexdigest()
def verify_inference(self, prompt: str, model: str = "gpt-4.1",
temperature: float = 0.7,
expected_hash: Optional[str] = None,
num_runs: int = 3) -> VerificationReport:
"""
Chạy multiple inference runs và verify reproducibility.
Returns detailed report về consistency giữa các runs.
"""
results = []
latencies = []
print(f"🔄 Starting {num_runs} inference runs for verification...")
for run in range(num_runs):
start_time = time.time()
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": temperature
},
timeout=30
)
latency_ms = (time.time() - start_time) * 1000
latencies.append(latency_ms)
if response.status_code != 200:
return VerificationReport(
is_reproducible=False,
expected_hash=expected_hash or "N/A",
actual_hash="N/A",
similarity_score=0.0,
error_message=f"HTTP {response.status_code}: {response.text}"
)
data = response.json()
content = data["choices"][0]["message"]["content"]
result_hash = self.generate_deterministic_hash(
content, model, temperature, run
)
results.append(InferenceResult(
content=content,
hash=result_hash,
model=model,
latency_ms=latency_ms,
timestamp=time.time(),
request_id=data.get("id", "unknown")
))
print(f" Run {run + 1}: Hash={result_hash[:16]}..., "
f"Latency={latency_ms:.2f}ms")
# Calculate reproducibility metrics
hashes = [r.hash for r in results]
unique_hashes = set(hashes)
is_reproducible = len(unique_hashes) == 1
# Calculate average latency
avg_latency = sum(latencies) / len(latencies)
print(f"\n📊 Verification Results:")
print(f" Unique hashes: {len(unique_hashes)}/{num_runs}")
print(f" Average latency: {avg_latency:.2f}ms")
print(f" Reproducible: {'✅ YES' if is_reproducible else '❌ NO'}")
return VerificationReport(
is_reproducible=is_reproducible,
expected_hash=expected_hash or hashes[0],
actual_hash=hashes[0],
similarity_score=1.0 if is_reproducible else 0.0,
error_message=None
)
=== USAGE EXAMPLE ===
if __name__ == "__main__":
verifier = HolySheepReproducibilityVerifier(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
test_prompt = "Explain quantum entanglement in one sentence."
report = verifier.verify_inference(
prompt=test_prompt,
model="gpt-4.1",
temperature=0.0, # Temperature = 0 cho deterministic output
num_runs=5
)
if not report.is_reproducible:
print(f"\n⚠️ WARNING: Non-reproducible inference detected!")
print(f"Expected: {report.expected_hash}")
print(f"Actual: {report.actual_hash}")2. Temperature và Seed Configuration
Trong thực tế, tôi phát hiện rằng temperature = 0 không phải lúc nào cũng đảm bảo 100% deterministic output. Một số model có internal randomness ngay cả ở temperature 0. Giải pháp là sử dụng seed parameter nếu model hỗ trợ.
import asyncio
import aiohttp
from typing import List, Tuple, Optional
import statistics
class AdvancedReproducibilityTester:
"""
Advanced testing với multiple configurations và statistical analysis.
"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.test_results: List[dict] = []
async def run_inference_async(self, session: aiohttp.ClientSession,
prompt: str, model: str,
temperature: float, seed: Optional[int],
max_tokens: int = 100) -> dict:
"""Run single inference với async support."""
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": temperature,
"max_tokens": max_tokens
}
if seed is not None:
payload["seed"] = seed
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
start = asyncio.get_event_loop().time()
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
data = await response.json()
latency_ms = (asyncio.get_event_loop().time() - start) * 1000
return {
"status": response.status,
"content": data.get("choices", [{}])[0].get("message", {}).get("content", ""),
"latency_ms": latency_ms,
"model": model,
"temperature": temperature,
"seed": seed,
"usage": data.get("usage", {})
}
async def test_temperature_variations(self, prompt: str,
model: str = "gpt-4.1") -> dict:
"""
Test reproducibility ở different temperature settings.
Temperature variations: 0.0, 0.3, 0.7, 1.0
"""
print(f"🧪 Testing temperature variations for model: {model}")
results = {}
async with aiohttp.ClientSession() as session:
# Test each temperature with 3 runs
for temp in [0.0, 0.3, 0.7, 1.0]:
print(f"\n📌 Temperature = {temp}")
tasks = [
self.run_inference_async(session, prompt, model, temp, None)
for _ in range(3)
]
responses = await asyncio.gather(*tasks)
latencies = [r["latency_ms"] for r in responses]
contents = [r["content"] for r in responses]
# Check if all outputs are identical
all_same = len(set(contents)) == 1
results[f"temp_{temp}"] = {
"reproducible": all_same,
"unique_outputs": len(set(contents)),
"avg_latency_ms": statistics.mean(latencies),
"min_latency_ms": min(latencies),
"max_latency_ms": max(latencies),
"stddev_ms": statistics.stdev(latencies) if len(latencies) > 1 else 0,
"sample_output": contents[0][:100] + "..." if contents[0] else "EMPTY"
}
print(f" Reproducible: {'✅' if all_same else '❌'}")
print(f" Unique outputs: {len(set(contents))}")
print(f" Avg latency: {statistics.mean(latencies):.2f}ms")
return results
async def test_seed_based_reproducibility(self, prompt: str,
model: str = "gpt-4.1") -> dict:
"""
Test reproducibility với explicit seed parameter.
Seed-based approach cho deterministic output.
"""
print(f"🧬 Testing seed-based reproducibility for model: {model}")
seed_results = {}
async with aiohttp.ClientSession() as session:
# Test same seed across multiple runs
test_seeds = [42, 123, 999]
for seed in test_seeds:
print(f"\n🎲 Testing with seed = {seed}")
tasks = [
self.run_inference_async(session, prompt, model, 0.7, seed)
for _ in range(5)
]
responses = await asyncio.gather(*tasks)
contents = [r["content"] for r in responses]
all_same = len(set(contents)) == 1
seed_results[f"seed_{seed}"] = {
"reproducible": all_same,
"unique_outputs": len(set(contents)),
"avg_latency_ms": statistics.mean([r["latency_ms"] for r in responses]),
"outputs": contents
}
print(f" Reproducible: {'✅' if all_same else '❌'}")
print(f" Unique outputs: {len(set(contents))}")
return seed_results
=== ADVANCED USAGE ===
async def main():
tester = AdvancedReproducibilityTester(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
test_prompt = "What is the capital of Vietnam? Answer in one word only."
# Temperature variations test
temp_results = await tester.test_temperature_variations(test_prompt)
print("\n" + "="*60)
print("📊 TEMPERATURE TEST SUMMARY")
print("="*60)
for temp_key, result in temp_results.items():
status = "✅" if result["reproducible"] else "❌"
print(f"{status} {temp_key}: {result['unique_outputs']} unique outputs, "
f"avg {result['avg_latency_ms']:.2f}ms")
# Seed-based test
seed_results = await tester.test_seed_based_reproducibility(test_prompt)
print("\n" + "="*60)
print("📊 SEED-BASED TEST SUMMARY")
print("="*60)
for seed_key, result in seed_results.items():
status = "✅" if result["reproducible"] else "❌"
print(f"{status} {seed_key}: {result['unique_outputs']} unique outputs")
if __name__ == "__main__":
asyncio.run(main())3. Real-time Monitoring Dashboard
Để production-ready, tôi đã xây dựng một monitoring system giúp track reproducibility metrics theo thời gian thực. Dashboard này sử dụng HolySheep AI API với streaming để cập nhật metrics liên tục.
import time
import json
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass, asdict
from typing import Dict, List, Optional
import requests
@dataclass
class ReproducibilityMetrics:
"""Metrics container cho reproducibility tracking."""
total_requests: int = 0
reproducible_count: int = 0
non_reproducible_count: int = 0
avg_latency_ms: float = 0.0
p95_latency_ms: float = 0.0
total_cost_usd: float = 0.0
last_updated: str = ""
class ReproducibilityMonitor:
"""
Production monitoring system cho AI inference reproducibility.
Features:
- Real-time metrics tracking
- Alerting when reproducibility drops
- Cost tracking per model
- Historical trend analysis
"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.metrics: Dict[str, ReproducibilityMetrics] = defaultdict(
lambda: ReproducibilityMetrics()
)
self.request_log: List[dict] = []
self.alert_threshold = 0.95 # 95% reproducibility threshold
def record_inference(self, model: str, prompt_hash: str,
output_hash: str, latency_ms: float,
tokens_used: int, cost_usd: float,
temperature: float):
"""Record a single inference result."""
metric = self.metrics[model]
metric.total_requests += 1
metric.total_cost_usd += cost_usd
# Update latency metrics
latencies = [r["latency_ms"] for r in self.request_log
if r["model"] == model][-100:] # Last 100
latencies.append(latency_ms)
metric.avg_latency_ms = sum(latencies) / len(latencies)
latencies_sorted = sorted(latencies)
metric.p95_latency_ms = latencies_sorted[int(len(latencies_sorted) * 0.95)]
# Log request
self.request_log.append({
"timestamp": datetime.now().isoformat(),
"model": model,
"prompt_hash": prompt_hash,
"output_hash": output_hash,
"latency_ms": latency_ms,
"tokens_used": tokens_used,
"cost_usd": cost_usd,
"temperature": temperature
})
metric.last_updated = datetime.now().isoformat()
# Check reproducibility (same prompt_hash vs output_hash patterns)
self._update_reproducibility_rate(model)
def _update_reproducibility_rate(self, model: str):
"""Calculate reproducibility rate for a model."""
metric = self.metrics[model]
recent_logs = [r for r in self.request_log
if r["model"] == model][-50:]
if len(recent_logs) < 2:
return
# Group by prompt_hash
groups = defaultdict(list)
for log in recent_logs:
groups[log["prompt_hash"]].append(log["output_hash"])
# Calculate reproducibility
reproducible = 0
total_groups = len(groups)
for prompt_hash, output_hashes in groups.items():
if len(output_hashes) >= 2:
# Check if all outputs for same prompt are identical
if len(set(output_hashes)) == 1:
reproducible += 1
if total_groups > 0:
rate = reproducible / total_groups
metric.reproducible_count = reproducible
metric.non_reproducible_count = total_groups - reproducible
# Trigger alert if threshold breached
if rate < self.alert_threshold:
self._trigger_alert(model, rate)
def _trigger_alert(self, model: str, current_rate: float):
"""Trigger alert when reproducibility drops below threshold."""
print(f"🚨 ALERT: Model '{model}' reproducibility rate "
f"dropped to {current_rate:.2%} (threshold: {self.alert_threshold:.2%})")
# In production, integrate with PagerDuty, Slack, etc.
# self.send_alert_notification(model, current_rate)
def get_metrics_dashboard(self) -> Dict:
"""Generate metrics dashboard data."""
dashboard = {
"generated_at": datetime.now().isoformat(),
"models": {}
}
for model, metric in self.metrics.items():
dashboard["models"][model] = {
"total_requests": metric.total_requests,
"reproducible_count": metric.reproducible_count,
"non_reproducible_count": metric.non_reproducible_count,
"reproducibility_rate": (
metric.reproducible_count /
(metric.reproducible_count + metric.non_reproducible_count)
if (metric.reproducible_count + metric.non_reproducible_count) > 0
else 1.0
),
"avg_latency_ms": round(metric.avg_latency_ms, 2),
"p95_latency_ms": round(metric.p95_latency_ms, 2),
"total_cost_usd": round(metric.total_cost_usd, 4),
"last_updated": metric.last_updated
}
return dashboard
def run_continuous_verification(self, test_prompts: List[str],
interval_seconds: int = 60):
"""
Run continuous verification với periodic health checks.
"""
print("🔄 Starting continuous reproducibility monitoring...")
print(f"📝 Monitoring {len(test_prompts)} test prompts every {interval_seconds}s")
while True:
for prompt in test_prompts:
for model in ["gpt-4.1", "claude-sonnet-4.5"]:
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.0
},
timeout=30
)
if response.status_code == 200:
data = response.json()
output = data["choices"][0]["message"]["content"]
prompt_hash = hashlib.sha256(prompt.encode()).hexdigest()
output_hash = hashlib.sha256(output.encode()).hexdigest()
tokens = data.get("usage", {}).get("total_tokens", 0)
# Calculate cost based on model pricing
cost_per_1k = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
cost = (tokens / 1000) * cost_per_1k.get(model, 8.0)
self.record_inference(
model=model,
prompt_hash=prompt_hash,
output_hash=output_hash,
latency_ms=response.elapsed.total_seconds() * 1000,
tokens_used=tokens,
cost_usd=cost,
temperature=0.0
)
except Exception as e:
print(f"❌ Error verifying {model}: {str(e)}")
# Print dashboard
dashboard = self.get_metrics_dashboard()
print(f"\n📊 Dashboard Update - {dashboard['generated_at']}")
for model, metrics in dashboard["models"].items():
print(f" {model}: {metrics['reproducibility_rate']:.1%} reproducible, "
f"${metrics['total_cost_usd']:.4f} total cost")
time.sleep(interval_seconds)
=== MONITORING USAGE ===
if __name__ == "__main__":
import hashlib
monitor = ReproducibilityMonitor(api_key="YOUR_HOLYSHEEP_API_KEY")
# Test prompts cho continuous monitoring
test_prompts = [
"What is 2 + 2?",
"List the colors of a rainbow.",
"What is the chemical symbol for gold?"
]
# Run one-time verification
print("📊 Running verification...")
dashboard = monitor.get_metrics_dashboard()
print(json.dumps(dashboard, indent=2))
# For continuous monitoring (uncomment to enable)
# monitor.run_continuous_verification(test_prompts, interval_seconds=60)Bảng So Sánh Model Pricing - HolySheep AI 2026
Khi implement reproducibility verification, việc hiểu rõ pricing giúp bạn optimize cost hiệu quả. HolySheep AI cung cấp mức giá cực kỳ cạnh tranh với tỷ giá ¥1 = $1 USD:
| Model | Giá/MTok (USD) | Tính năng | Khuyến nghị |
|---|---|---|---|
| GPT-4.1 | $8.00 | High accuracy, complex reasoning | Production critical tasks |
| Claude Sonnet 4.5 | $15.00 | Extended context, nuance | Long-form content |
| Gemini 2.5 Flash | $2.50 | Fast, cost-effective | High volume, non-critical |
| DeepSeek V3.2 | $0.42 | Budget-friendly | Testing, development |
Với DeepSeek V3.2 chỉ $0.42/MTok, bạn có thể chạy extensive reproducibility tests với chi phí cực thấp. Tôi thường dùng DeepSeek V3.2 cho CI/CD pipeline và GPT-4.1 cho production verification.
Lỗi Thường Gặp và Cách Khắc Phục
Qua quá trình implement reproducibility verification, tôi đã gặp và xử lý nhiều lỗi khác nhau. Dưới đây là 5 lỗi phổ biến nhất và giải pháp chi tiết:
1. Lỗi "ConnectionError: timeout" - Timeout Khi Run Multiple Verification
# ❌ CÁCH SAI - Gây timeout khi chạy nhiều requests
import requests
def verify_multiple_runs(prompt: str, num_runs: int = 10):
results = []
for i in range(num_runs):
response = requests.post(
f"{self.base_url}/chat/completions",
json={"model": "gpt-4.1", "messages": [...]},
timeout=5 # Too short!
)
results.append(response.json())
return results
✅ CÁCH ĐÚNG - Incremental timeout với retry logic
import time
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
def create_session_with_retry():
"""Create requests session với exponential backoff retry."""
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
return session
def verify_with_proper_timeout(prompt: str, num_runs: int = 10):
"""Verify với adaptive timeout và retry."""
session = create_session_with_retry()
results = []
for i in range(num_runs):
try:
response = session.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 100
},
timeout=(5, 60) # (connect_timeout, read_timeout)
)
results.append(response.json())
except requests.exceptions.Timeout:
print(f"⚠️ Request {i+1} timeout, retrying...")
time.sleep(2 ** i) # Exponential backoff
continue
except requests.exceptions.ConnectionError as e:
print(f"❌ Connection error: {e}")
time.sleep(5)
continue
return results2. Lỗi "401 Unauthorized" - API Key Authentication
# ❌ CÁCH SAI - Hardcode API key trong code
response = requests.post(
url,
headers={"Authorization": "Bearer sk-1234567890abcdef"} # Security risk!
)
❌ CÁCH SAI - Sai format Authorization header
headers = {"Authorization": "Token YOUR_API_KEY"} # Sai prefix
✅ CÁCH ĐÚNG - Environment variable + correct format
import os
from dotenv import load_dotenv
load_dotenv() # Load .env file
class HolySheepAPIClient:
"""Production-ready API client."""
def __init__(self):
self.api_key = os.getenv("HOLYSHEEP_API_KEY")
if not self.api_key:
raise ValueError(
"HOLYSHEEP_API_KEY not found. "
"Set it in .env file or environment variable."
)
if not self.api_key.startswith("hs_"):
raise ValueError(
"Invalid API key format. HolySheep API keys start with 'hs_'"
)
def create_headers(self, additional_headers: dict = None) -> dict:
"""Create proper authentication headers."""
headers = {
"Authorization": f"Bearer {self.api_key}", # MUST use "Bearer"
"Content-Type": "application/json"
}
if additional_headers:
headers.update(additional_headers)
return headers
def verify_connection(self) -> bool:
"""Test API connection với /models endpoint."""
try:
response = requests.get(
"https://api.holysheep.ai/v1/models",
headers=self.create_headers(),
timeout=10
)
if response.status_code == 200:
print("✅ API connection successful")
return True
elif response.status_code == 401:
print("❌ Authentication failed. Check your API key.")
return False
else:
print(f"⚠️ Unexpected status: {response.status_code}")
return False
except requests.exceptions.RequestException as e:
print(f"❌ Connection error: {e}")
return False
=== VERIFICATION SCRIPT ===
if __name__ == "__main__":
client = HolySheepAPIClient()
if client.verify_connection():
print("🚀 Ready to verify reproducibility!")3. Lỗi "Rate Limit Exceeded" - Quá Nhiều Verification Requests
# ❌ CÁCH SAI - Không có rate limiting
def run_verification_batch(prompts: list):
results = []
for prompt in prompts:
response = requests.post(url, json={...}) # Spam API!
results.append(response.json())
return results
✅ CÁCH ĐÚNG - Rate limiting với token bucket algorithm
import time
import threading
from collections import deque
class RateLimiter:
"""Token bucket rate limiter for API calls."""
def __init__(self, requests_per_second: float = 10,
burst_size: int = 20):
self.rate = requests_per_second
self.burst = burst_size
self.tokens = burst_size
self.last_update = time.time()
self.lock = threading.Lock()
def acquire(self, tokens: int = 1):
"""Acquire tokens, blocking if necessary."""
with self.lock:
now = time.time()
elapsed = now - self.last_update
self.tokens = min(
self.burst,
self.tokens + elapsed * self.rate
)
self.last_update = now
if self.tokens < tokens:
wait_time = (tokens - self.tokens) / self.rate
time.sleep(wait_time)
self.tokens = 0
else:
self.tokens -= tokens
def get_wait_time(self) -> float:
"""Get estimated wait time for next request."""
with self.lock:
if self.tokens >= 1:
return 0
return (1 - self.tokens) / self.rate
class ThrottledVerificationClient:
"""API client với built-in rate limiting."""
def __init__(self, api_key: str, rpm: int = 60):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.limiter = RateLimiter(
requests_per_second=rpm / 60,
burst_size=rpm // 10
)
self.request_times = deque(maxlen=100)
def _check_rate_limit_headers(self, response: requests.Response):
"""Parse rate limit headers từ response."""
headers = response.headers
remaining = headers.get("X-RateLimit-Remaining")
reset_time = headers.get("X-RateLimit-Reset")
if remaining is not None:
print(f"📊 Rate limit remaining: {remaining}")
if remaining and int(remaining) < 10:
print("⚠️ Approaching rate limit!")
def throttled_post(self, endpoint: str, payload: dict) -> dict:
"""Post request với rate limiting."""
# Wait for rate limit
wait_time = self.limiter.get_wait_time()
if wait_time > 0:
print(f"⏳ Rate limited, waiting {wait_time:.2f}s")
time.sleep(wait_time)
self.limiter.acquire()
start = time.time()
response = requests.post(
f"{self.base_url}{endpoint}",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json=payload,
timeout=(10, 60)
)
elapsed = time.time() - start
self.request_times.append(elapsed)
self._check_rate_limit_headers(response)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"⛔ Rate limited! Retrying after {retry_after}s")
time.sleep(retry_after)
return self.throttled_post(endpoint, payload)
return response.json()
=== USAGE ===
def run_ratelimited_verification(prompts: list, model: str = "gpt-4.1"):
client = ThrottledVerificationClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
rpm=60 # 60 requests per minute
)
results = []
for i, prompt in enumerate(prompts):
print(f"Processing {i+1}/{len(prompts)}: {prompt[:50]}...")
result = client.throttled_post(
"/chat/completions",
{
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.0
}
)
results.append(result)
return results4. Lỗi "Streaming Response Hash Mismatch" - Hashing Streamed Output
# ❌ CÁCH SAI - Hash từng chunk riêng lẻ
def stream_and_hash_wrong(prompt: str):
response = requests.post(url, stream=True)
full_content = ""
for chunk in response.iter_lines():
if chunk:
full_content += chunk.decode()
# Hash mỗi chunk -> different hash per chunk!