Đêm thứ 6 tuần trước, tôi nhận được cuộc gọi khẩn cấp từ đội backend. Hệ thống xử lý hàng loạt của công ty — vốn dựa trên DeepSeek V4 API chính thức — đã hoàn toàn chết máy. 12.000 yêu cầu xử lý tài liệu bị kẹt trong hàng đợi, khách hàng phàn nàn ào ạt, và đội dev phải làm việc xuyên đêm để khắc phục.
Bài học đắt giá đó là lý do tôi viết bài viết này. Trong 18 tháng triển khai batch processing với DeepSeek, tôi đã thử nghiệm gần như mọi chiến lược tối ưu hóa: từ simple throttling thủ công đến sophisticated token bucket algorithm. Và kết quả? Việc chuyển sang HolySheep AI không chỉ giải quyết vấn đề rate limit mà còn giúp team tiết kiệm 85%+ chi phí API.
Vì Sao Batch Request Thất Bại? Phân Tích Root Cause
Trước khi đi vào giải pháp, cần hiểu rõ tại sao batch request với DeepSeek V4 thường gặp vấn đề nghiêm trọng.
3 Lý Do Chính Gây Ra Rate Limit Crisis
1. Token Budget Exhaustion Không Lường Trước
DeepSeek V4 có cơ chế rate limit phức tạp: không chỉ giới hạn requests/minute mà còn giới hạn tokens/minute và concurrent connections. Khi batch size tăng đột ngột (ví dụ: xử lý data spike từ campaign marketing), hệ thống không kịp thích ứng.
2. Retry Storm — Vòng lặp tử thần
Khi request bị reject, hầu hết developers immediately retry với exponential backoff không đúng cách. Kết quả? 100 failed requests tạo ra 500+ retry attempts trong 30 giây, gây ra cascade failure.
3. Connection Pool Exhaustion
Mỗi HTTP connection có timeout nhất định. Khi pool bị chiếm dụng bởi các request đang chờ, new requests không có slot để khởi tạo, dẫn đến timeout cascade.
# Ví dụ: Retry storm thực tế mà team tôi gặp phải
Khi gặp 429 error, code cũ của team:
import time
import requests
def call_deepseek_batch(prompts):
results = []
for prompt in prompts:
max_retries = 5
for attempt in range(max_retries):
try:
response = requests.post(
"https://api.deepseek.com/v1/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json={"model": "deepseek-chat", "messages": [{"role": "user", "content": prompt}]}
)
if response.status_code == 200:
results.append(response.json())
break
elif response.status_code == 429:
# Vấn đề: sleep ngắn + retry ngay = disaster
time.sleep(0.1) # Quá ngắn!
continue
except Exception as e:
time.sleep(0.1)
continue
return results
Kết quả: 12,000 prompts tạo ra ~50,000 HTTP requests trong 2 phút
Server DeepSeek chính thức block IP team suốt 1 giờ
Kiến Trúc Tối Ưu: HolySheep AI + Smart Rate Limiter
Sau khi benchmark nhiều giải pháp relay API, tôi chọn HolySheep AI với 3 lý do chính:
- Tỷ giá ¥1 = $1 — Tiết kiệm 85%+ so với API chính thức
- Độ trễ trung bình <50ms — Thấp hơn đáng kể so với direct API
- Hỗ trợ WeChat/Alipay — Thuận tiện cho developers Trung Quốc
- Tín dụng miễn phí khi đăng ký — Có thể test production-ready ngay
So sánh chi phí thực tế (tính theo 1 triệu tokens đầu vào):
- GPT-4.1: $8/1M tokens — Chi phí cao nhất
- Claude Sonnet 4.5: $15/1M tokens
- Gemini 2.5 Flash: $2.50/1M tokens
- DeepSeek V3.2: $0.42/1M tokens — Rẻ nhất, và HolySheep còn giảm thêm
Triển Khai Chi Tiết: Từ Zero Đến Production
Bước 1: Cài Đặt Client Với Rate Limiting Tích Hợp
# Cài đặt thư viện cần thiết
pip install aiohttp asyncio-limiter holy Sheep-sdk
File: holy_sheep_batch_client.py
=============================================
import asyncio
import aiohttp
from typing import List, Dict, Any
from holy_sheep import HolySheepClient
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class BatchRequestOptimizer:
"""
Lớp xử lý batch request với rate limiting thông minh
được tối ưu cho HolySheep AI API
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
max_concurrent: int = 20,
requests_per_minute: int = 300,
tokens_per_minute: int = 100000
):
self.api_key = api_key
self.base_url = base_url
self.client = HolySheepClient(api_key=api_key, base_url=base_url)
# Token bucket algorithm cho rate limiting
self.request_bucket = AsyncTokenBucket(
capacity=requests_per_minute,
refill_rate=requests_per_minute / 60.0
)
self.token_bucket = AsyncTokenBucket(
capacity=tokens_per_minute,
refill_rate=tokens_per_minute / 60.0
)
# Semaphore để kiểm soát concurrent connections
self.semaphore = asyncio.Semaphore(max_concurrent)
# Metrics tracking
self.stats = {
"total_requests": 0,
"successful": 0,
"failed": 0,
"retries": 0,
"total_tokens": 0
}
async def process_batch(
self,
prompts: List[str],
model: str = "deepseek-chat",
temperature: float = 0.7,
max_tokens: int = 2048
) -> List[Dict[str, Any]]:
"""
Xử lý batch prompts với rate limiting thông minh
"""
tasks = []
for idx, prompt in enumerate(prompts):
task = self._process_single_with_retry(
prompt=prompt,
model=model,
temperature=temperature,
max_tokens=max_tokens,
request_id=idx
)
tasks.append(task)
# Execute với gather, semaphore tự động kiểm soát concurrency
results = await asyncio.gather(*tasks, return_exceptions=True)
# Log summary
logger.info(f"Batch completed: {self.stats}")
return results
async def _process_single_with_retry(
self,
prompt: str,
model: str,
temperature: float,
max_tokens: int,
request_id: int,
max_retries: int = 3
) -> Dict[str, Any]:
"""
Xử lý single request với exponential backoff
"""
async with self.semaphore:
for attempt in range(max_retries):
try:
# Chờ đến khi có slot trong rate limiter
await self.request_bucket.acquire()
# Estimate tokens và chờ nếu cần
estimated_tokens = len(prompt.split()) * 1.3 # Rough estimate
await self.token_bucket.acquire_amount(int(estimated_tokens))
# Gọi HolySheep API
start_time = time.time()
response = await self.client.chat_completions_create(
model=model,
messages=[{"role": "user", "content": prompt}],
temperature=temperature,
max_tokens=max_tokens
)
latency_ms = (time.time() - start_time) * 1000
self.stats["total_requests"] += 1
self.stats["successful"] += 1
self.stats["total_tokens"] += response.usage.total_tokens
logger.info(f"Request {request_id} success: {latency_ms:.2f}ms")
return {
"success": True,
"data": response,
"latency_ms": latency_ms,
"request_id": request_id
}
except RateLimitError as e:
self.stats["retries"] += 1
wait_time = min(2 ** attempt * 1.5, 30) # Exponential backoff
logger.warning(f"Rate limited, waiting {wait_time}s before retry...")
await asyncio.sleep(wait_time)
except Exception as e:
if attempt == max_retries - 1:
self.stats["failed"] += 1
logger.error(f"Request {request_id} failed permanently: {str(e)}")
return {
"success": False,
"error": str(e),
"request_id": request_id
}
await asyncio.sleep(2 ** attempt)
return {"success": False, "error": "Max retries exceeded", "request_id": request_id}
class AsyncTokenBucket:
"""
Token bucket implementation cho async/await
"""
def __init__(self, capacity: int, refill_rate: float):
self.capacity = capacity
self.tokens = capacity
self.refill_rate = refill_rate
self.last_refill = time.time()
self.lock = asyncio.Lock()
async def acquire(self, amount: int = 1):
async with self.lock:
while self.tokens < amount:
self._refill()
if self.tokens < amount:
await asyncio.sleep(0.1)
self.tokens -= amount
async def acquire_amount(self, amount: int):
await self.acquire(amount)
def _refill(self):
now = time.time()
elapsed = now - self.last_refill
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
Bước 2: Cấu Hình Worker Pool Với Connection Management
# File: batch_worker_pool.py
=============================================
import asyncio
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import Optional
import aiohttp
import certifi
import ssl
@dataclass
class WorkerConfig:
"""Cấu hình worker pool"""
num_workers: int = 10
max_connections_per_host: int = 20
connect_timeout: float = 30.0
read_timeout: float = 60.0
total_timeout: float = 120.0
class ConnectionPoolManager:
"""
Quản lý connection pool cho batch processing
"""
def __init__(self, config: Optional[WorkerConfig] = None):
self.config = config or WorkerConfig()
self._session: Optional[aiohttp.ClientSession] = None
self._connector: Optional[aiohttp.TCPConnector] = None
async def get_session(self) -> aiohttp.ClientSession:
"""Lazy initialization của session"""
if self._session is None or self._session.closed:
self._connector = aiohttp.TCPConnector(
limit=self.config.num_workers * self.config.max_connections_per_host,
limit_per_host=self.config.max_connections_per_host,
ttl_dns_cache=300,
ssl=ssl.create_default_context(cafile=certifi.where())
)
timeout = aiohttp.ClientTimeout(
total=self.config.total_timeout,
connect=self.config.connect_timeout,
sock_read=self.config.read_timeout
)
self._session = aiohttp.ClientSession(
connector=self._connector,
timeout=timeout,
headers={
"Content-Type": "application/json",
"X-Batch-Optimizer": "HolySheep-v2"
}
)
return self._session
async def close(self):
"""Cleanup resources"""
if self._session and not self._session.closed:
await self._session.close()
if self._connector and not self._connector.closed:
await self._connector.close()
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.close()
File: main_batch_processor.py
=============================================
import asyncio
import json
from batch_worker_pool import ConnectionPoolManager, WorkerConfig
from holy_sheep_batch_client import BatchRequestOptimizer
async def main():
"""
Ví dụ sử dụng batch processor với HolySheep AI
"""
# Cấu hình
config = WorkerConfig(
num_workers=15,
max_connections_per_host=25,
connect_timeout=30.0,
read_timeout=60.0,
total_timeout=120.0
)
optimizer = BatchRequestOptimizer(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
max_concurrent=15,
requests_per_minute=500,
tokens_per_minute=200000
)
# Đọc prompts từ file
prompts = []
with open("prompts_batch.json", "r") as f:
data = json.load(f)
prompts = [item["prompt"] for item in data["items"]]
print(f"Processing {len(prompts)} prompts...")
async with ConnectionPoolManager(config) as pool:
results = await optimizer.process_batch(
prompts=prompts,
model="deepseek-chat",
temperature=0.7,
max_tokens=2048
)
# Thống kê kết quả
successful = sum(1 for r in results if isinstance(r, dict) and r.get("success"))
failed = len(results) - successful
print(f"\n=== Batch Processing Summary ===")
print(f"Total prompts: {len(prompts)}")
print(f"Successful: {successful}")
print(f"Failed: {failed}")
print(f"Success rate: {successful/len(prompts)*100:.2f}%")
print(f"Total cost (estimated): ${optimizer.stats['total_tokens'] * 0.42 / 1_000_000:.4f}")
if __name__ == "__main__":
asyncio.run(main())
Bước 3: Monitoring Dashboard Cho Production
# File: batch_monitor.py
=============================================
import time
from dataclasses import dataclass, field
from typing import Dict, List
from collections import deque
from datetime import datetime
import threading
@dataclass
class MetricsSnapshot:
timestamp: datetime
requests_sent: int
requests_success: int
requests_failed: int
requests_retried: int
avg_latency_ms: float
tokens_used: int
queue_size: int
active_connections: int
class BatchMetricsCollector:
"""
Real-time metrics collector cho batch processing
"""
def __init__(self, window_size_seconds: int = 60):
self.window_size = window_size_seconds
self.snapshots: deque = deque(maxlen=100)
# Counters
self._lock = threading.Lock()
self._requests_sent = 0
self._requests_success = 0
self._requests_failed = 0
self._requests_retried = 0
self._tokens_used = 0
self._latencies: List[float] = []
self._latencies_lock = threading.Lock()
# Rate tracking
self._window_requests = deque(maxlen=window_size_seconds)
self._window_start = time.time()
def record_request(
self,
success: bool,
latency_ms: float,
tokens: int = 0,
retried: bool = False
):
with self._lock:
self._requests_sent += 1
if success:
self._requests_success += 1
else:
self._requests_failed += 1
if retried:
self._requests_retried += 1
self._tokens_used += tokens
with self._latencies_lock:
self._latencies.append(latency_ms)
def get_current_snapshot(self) -> MetricsSnapshot:
with self._lock:
with self._latencies_lock:
avg_latency = sum(self._latencies) / len(self._latencies) if self._latencies else 0
return MetricsSnapshot(
timestamp=datetime.now(),
requests_sent=self._requests_sent,
requests_success=self._requests_success,
requests_failed=self._requests_failed,
requests_retried=self._requests_retried,
avg_latency_ms=avg_latency,
tokens_used=self._tokens_used,
queue_size=0, # Từ batch queue
active_connections=0 # Từ connection pool
)
def get_rate_stats(self) -> Dict:
"""Tính requests per minute"""
now = time.time()
elapsed = now - self._window_start
return {
"requests_per_minute": self._requests_sent / max(elapsed / 60, 1),
"success_rate": self._requests_success / max(self._requests_sent, 1),
"retry_rate": self._requests_retried / max(self._requests_sent, 1),
"cost_per_1k_tokens": self._tokens_used * 0.42 / 1_000_000 * 1000
}
def export_prometheus_format(self) -> str:
"""Export metrics cho Prometheus scraping"""
snapshot = self.get_current_snapshot()
rate_stats = self.get_rate_stats()
return f"""
HELP batch_requests_total Total number of batch requests
TYPE batch_requests_total counter
batch_requests_total{{status="success"}} {snapshot.requests_success}
batch_requests_total{{status="failed"}} {snapshot.requests_failed}
HELP batch_retries_total Total number of retries
TYPE batch_retries_total counter
batch_retries_total {snapshot.requests_retried}
HELP batch_avg_latency_ms Average request latency in milliseconds
TYPE batch_avg_latency_ms gauge
batch_avg_latency_ms {snapshot.avg_latency_ms}
HELP batch_tokens_used Total tokens used
TYPE batch_tokens_used counter
batch_tokens_used {snapshot.tokens_used}
HELP batch_rpm Current requests per minute
TYPE batch_rpm gauge
batch_rpm {rate_stats['requests_per_minute']:.2f}
"""
File: alert_manager.py
=============================================
import logging
from enum import Enum
from typing import Callable, Dict
logger = logging.getLogger(__name__)
class AlertLevel(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
class Alert:
def __init__(self, level: AlertLevel, message: str, metric: str = None):
self.level = level
self.message = message
self.metric = metric
self.timestamp = datetime.now()
class BatchAlertManager:
"""
Alert manager cho batch processing
"""
def __init__(self):
self.alert_handlers: Dict[AlertLevel, Callable] = {
AlertLevel.INFO: lambda a: logger.info(f"[ALERT] {a.message}"),
AlertLevel.WARNING: lambda a: logger.warning(f"[ALERT] {a.message}"),
AlertLevel.CRITICAL: lambda a: self._send_critical_alert(a)
}
# Thresholds
self.failure_rate_threshold = 0.05 # 5%
self.retry_rate_threshold = 0.15 # 15%
self.latency_p99_threshold_ms = 5000 # 5 seconds
def check_and_alert(self, collector: BatchMetricsCollector):
"""Kiểm tra metrics và trigger alerts nếu cần"""
snapshot = collector.get_current_snapshot()
rate_stats = collector.get_rate_stats()
# Check failure rate
failure_rate = 1 - rate_stats["success_rate"]
if failure_rate > self.failure_rate_threshold:
self._trigger_alert(
AlertLevel.WARNING,
f"High failure rate: {failure_rate*100:.2f}%",
"failure_rate"
)
# Check retry rate
if rate_stats["retry_rate"] > self.retry_rate_threshold:
self._trigger_alert(
AlertLevel.WARNING,
f"High retry rate: {rate_stats['retry_rate']*100:.2f}%",
"retry_rate"
)
# Check latency
if snapshot.avg_latency_ms > self.latency_p99_threshold_ms:
self._trigger_alert(
AlertLevel.CRITICAL,
f"High latency detected: {snapshot.avg_latency_ms:.2f}ms",
"avg_latency"
)
def _trigger_alert(self, level: AlertLevel, message: str, metric: str = None):
alert = Alert(level, message, metric)
self.alert_handlers[level](alert)
def _send_critical_alert(self, alert: Alert):
"""Gửi critical alert qua nhiều channels"""
logger.critical(f"[CRITICAL ALERT] {alert.message}")
# Slack webhook, email, SMS integration có thể thêm vào đây
Lỗi Thường Gặp Và Cách Khắc Phục
Trong quá trình triển khai batch processing với DeepSeek V4 qua HolySheep AI, tôi đã gặp và xử lý hàng chục lỗi khác nhau. Dưới đây là 5 lỗi phổ biến nhất kèm giải pháp đã được verify.
Lỗi 1: HTTP 429 Too Many Requests Liên Tục
Mô tả: Request bị reject với HTTP 429 dù đã có rate limiting.
Nguyên nhân gốc: Rate limit của HolySheep được tính theo tokens/minute, không chỉ requests/minute. Nếu prompts rất dài, một request có thể tiêu tốn hết budget.
Giải pháp:
# Giải pháp: Dynamic rate limiting dựa trên response headers
import aiohttp
import asyncio
async def smart_request_with_header_parsing(
session: aiohttp.ClientSession,
url: str,
headers: dict,
payload: dict,
optimizer: BatchRequestOptimizer
):
"""
Request với việc parse rate limit headers từ response
"""
max_tries = 3
for attempt in range(max_tries):
async with session.post(url, headers=headers, json=payload) as response:
if response.status == 200:
return await response.json()
elif response.status == 429:
# Parse rate limit headers
remaining = response.headers.get('X-RateLimit-Remaining', '0')
reset_time = response.headers.get('X-RateLimit-Reset')
# Cập nhật rate limiter với giá trị thực từ server
if reset_time:
reset_timestamp = float(reset_time)
current_timestamp = time.time()
wait_seconds = max(0, reset_timestamp - current_timestamp) + 1
# Dynamic adjustment
optimizer.request_bucket.capacity = min(
optimizer.request_bucket.capacity,
int(remaining) + 10
)
await asyncio.sleep(wait_seconds)
else:
# Fallback exponential backoff
await asyncio.sleep(2 ** attempt * 2)
elif response.status >= 500:
# Server error - retry
await asyncio.sleep(2 ** attempt)
continue
else:
# Client error - không retry
error_text = await response.text()
raise Exception(f"Request failed: {response.status} - {error_text}")
raise Exception("Max retries exceeded")
Lỗi 2: Connection Pool Exhaustion
Mô tả: "Cannot create connection to host" error sau vài trăm requests.
Nguyên nhân gốc: aiohttp giữ connection alive quá lâu, không release kịp khi gặp error.
Giải pháp:
# Giải pháp: Strict connection lifecycle management
class StrictConnectionPool:
"""
Connection pool với strict cleanup
"""
def __init__(self, max_total: int = 100, max_per_host: int = 20):
self.connector = aiohttp.TCPConnector(
limit=max_total,
limit_per_host=max_per_host,
force_close=True, # Quan trọng: close connection ngay
enable_cleanup_closed=True
)
self._session = None
self._active_requests = 0
self._lock = asyncio.Lock()
async def request(self, method: str, url: str, **kwargs):
async with self._lock:
self._active_requests += 1
try:
if not self._session or self._session.closed:
self._session = aiohttp.ClientSession(
connector=self.connector,
timeout=aiohttp.ClientTimeout(total=60)
)
async with self._session.request(method, url, **kwargs) as response:
# Ensure response body được đọc để release connection
await response.read()
return response
finally:
async with self._lock:
self._active_requests -= 1
# Force cleanup nếu có quá nhiều requests đang chờ
if self._active_requests > max_total * 0.8:
await self._force_cleanup()
async def _force_cleanup(self):
"""Force cleanup stale connections"""
if self._session and not self._session.closed:
# Close connector để release connections
await self.connector.close()
# Recreate connector
self.connector = aiohttp.TCPConnector(
limit=100,
limit_per_host=20,
force_close=True,
enable_cleanup_closed=True
)
self._session = None
Lỗi 3: Memory Leak Khi Xử Lý Batch Lớn
Mô tả: Memory tăng liên tục, eventually OOM khi batch size > 10,000.
Nguyên nhân gốc: Response objects và parsed JSON được giữ trong memory.
Giải pháp:
# Giải pháp: Streaming processor với batched writes
import gc
class StreamingBatchProcessor:
"""
Xử lý batch với streaming và periodic cleanup
"""
def __init__(self, batch_size: int = 100, gc_interval: int = 500):
self.batch_size = batch_size
self.gc_interval = gc_interval
self.results_buffer = []
self.processed_count = 0
async def process_with_streaming(
self,
prompts: List[str],
output_path: str
):
"""
Xử lý với streaming - không giữ tất cả trong memory
"""
with open(output_path, 'w') as f:
for i in range(0, len(prompts), self.batch_size):
batch = prompts[i:i + self.batch_size]
# Process batch
batch_results = await self._process_single_batch(batch)
# Write immediately
for result in batch_results:
f.write(json.dumps(result) + '\n')
f.flush() # Force write to disk
self.processed_count += len(batch)
# Periodic garbage collection
if self.processed_count % self.gc_interval == 0:
del batch_results
gc.collect()
# Log memory usage
import psutil
process = psutil.Process()
memory_mb = process.memory_info().rss / 1024 / 1024
logger.info(f"Processed {self.processed_count}, Memory: {memory_mb:.2f}MB")
async def _process_single_batch(self, batch: List[str]) -> List[Dict]:
"""Process một batch nhỏ"""
tasks = [process_single_prompt(p) for p in batch]
return await asyncio.gather(*tasks)
Lỗi 4: Token Estimate Không Chính Xác
Mô tả: Token bucket tính sai, dẫn đến hoặc thừa waiting time hoặc rate limit.
Nguyên nhân gốc: Dùng simple word count × 1.3 không accurate với tokenization thực.
Giải pháp:
# Giải pháp: Sử dụng tiktoken hoặc HolySheep built-in tokenizer
from holy_sheep import TokenCounter
Initialize token counter (sử dụng tokenizer tương thích)
token_counter = TokenCounter(model="deepseek-chat")
async def accurate_token_estimate(prompt: str) -> int:
"""
Estimate tokens chính xác hơn
"""
# Nếu có tiktoken, sử dụng nó
try:
import tiktoken
enc = tiktoken.get_encoding("cl100k_base")
return len(enc.encode(prompt))
except ImportError:
# Fallback: rough estimate nhưng tốt hơn
# DeepSeek sử dụng similar tokenization với GPT
chars = len(prompt)
return max(1, chars // 4) # Rough estimate: ~4 chars per token
Sử dụng trong rate limiter
async def process_with_accurate_limiting(
prompt: str,
optimizer: BatchRequestOptimizer
):
accurate_tokens = await accurate_token_estimate(prompt)
await optimizer.token_bucket.acquire_amount(accurate_tokens)
# Plus estimated output tokens
await optimizer.token_bucket.acquire_amount(512) # max_tokens default
return await optimizer.client.chat_completions_create(...)
Lỗi 5: Race Condition Trong Concurrent Requests
Mô tả: Kết quả không đúng thứ tự, hoặc missing responses.
Nguyên nhân gốc: Dùng shared mutable state mà không có proper locking.
Giải pháp:
# Giải pháp: Immutable result wrapper với asyncio.Lock
from dataclasses import dataclass
from typing import Optional
@dataclass(frozen=True)
class ImmutableResult:
index: int
success: bool
data: Optional[dict] = None
error: Optional[str] = None
latency_ms: float = 0.0
class ThreadSafeBatchProcessor:
"""
Batch processor với thread-safety guarantee
"""
def __init__(self, max_concurrent: int = 20):
self.semaphore = asyncio.Semaphore(max_concurrent)
self._lock = asyncio.Lock()
async def process_batch_safe(
self,
prompts: List[str],
processor: Callable
) -> List[ImmutableResult]:
"""
Process batch với đảm bảo thứ tự và thread-safety
"""
async def process_indexed(index: int, prompt: str) -> ImmutableResult:
async with self.semaphore:
start = time.time()
try:
result = await processor(prompt)
latency = (time.time() - start) * 1000
return ImmutableResult(
index=index,
success=True,
data=result,
latency_ms=latency
)
except Exception as e:
latency = (time.time() - start) * 1000
return ImmutableResult(
index=index,
success=False,
error=str(e),
latency_ms=latency
)
# Tạo tasks với index preservation
tasks = [
process_indexed(idx, prompt)
for idx, prompt in enumerate(prompts)
]
# Gather all results
results = await asyncio.gather(*tasks