I still remember the moment our e-commerce platform nearly collapsed during a flash sale last November. Our AI customer service chatbot was receiving 2,000 concurrent requests per second, and every single API call was queuing sequentially through our old architecture. Customers were seeing 45-second response times, abandon rates spiked to 38%, and our operations team was drowning in support tickets. That night, I rewrote our entire API integration layer using concurrent calling patterns—and brought response times down to under 800ms at peak load. This tutorial walks you through exactly what I learned and how you can apply these same techniques to your own projects.
Understanding the Concurrent API Call Challenge
When building AI-powered applications—whether e-commerce chatbots, enterprise RAG systems, or real-time content generation pipelines—you'll inevitably face situations where you need to make multiple API calls simultaneously. Sequential calls multiply your wait time linearly: 10 calls at 200ms each means 2 seconds of total latency. Concurrent calls with proper async handling can achieve near-constant time regardless of call count.
The challenge is that most developers implement API clients naively, creating blocking code that waits for each response before starting the next request. Modern applications require sophisticated approaches including connection pooling, request batching, circuit breakers, and intelligent retry logic.
The HolySheep AI Advantage for Concurrent Workloads
Before diving into code, let's discuss why HolySheep AI is particularly well-suited for high-concurrency scenarios. Their infrastructure delivers sub-50ms latency on average—critical when you're making dozens of simultaneous calls. Their pricing structure is developer-friendly: ¥1=$1 with no hidden fees, and you save 85%+ compared to mainstream providers charging ¥7.3+ per thousand tokens. Supported payment methods include WeChat and Alipay for seamless integration.
Current 2026 pricing comparison:
- GPT-4.1: $8.00 per million tokens
- Claude Sonnet 4.5: $15.00 per million tokens
- Gemini 2.5 Flash: $2.50 per million tokens
- DeepSeek V3.2: $0.42 per million tokens
DeepSeek V3.2 on HolySheep delivers exceptional cost efficiency for batch processing while maintaining quality suitable for production workloads.
Implementing Concurrent API Calls with Python asyncio
Let's start with a production-grade implementation using Python's asyncio library. This approach is ideal for I/O-bound operations like API calls, where you can dramatically improve throughput without increasing computational resources.
# concurrent_api_client.py
import asyncio
import aiohttp
import time
from typing import List, Dict, Any
from dataclasses import dataclass
from enum import Enum
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
@dataclass
class APIResponse:
request_id: str
status_code: int
data: Dict[Any, Any]
latency_ms: float
error: str = None
class ConcurrentAPIClient:
"""Production-grade concurrent API client with circuit breaker pattern."""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
max_concurrent: int = 50,
timeout_seconds: int = 30,
max_retries: int = 3
):
self.api_key = api_key
self.base_url = base_url
self.max_concurrent = max_concurrent
self.timeout = aiohttp.ClientTimeout(total=timeout_seconds)
self.max_retries = max_retries
# Circuit breaker state
self.circuit_state = CircuitState.CLOSED
self.failure_count = 0
self.failure_threshold = 5
self.half_open_attempts = 0
self.circuit_opened_at = None
# Rate limiting
self.semaphore = asyncio.Semaphore(max_concurrent)
self.last_request_time = 0
self.min_request_interval = 0.01 # 10ms minimum between requests
async def _check_circuit_breaker(self) -> bool:
"""Check if circuit breaker allows request."""
if self.circuit_state == CircuitState.CLOSED:
return True
if self.circuit_state == CircuitState.OPEN:
# Check if enough time has passed to try again
if time.time() - self.circuit_opened_at > 60: # 60 seconds
self.circuit_state = CircuitState.HALF_OPEN
logger.info("Circuit breaker transitioning to HALF_OPEN")
return True
return False
# HALF_OPEN state allows limited requests
return self.half_open_attempts < 3
async def _record_success(self):
"""Record successful request for circuit breaker."""
self.failure_count = 0
if self.circuit_state == CircuitState.HALF_OPEN:
self.half_open_attempts += 1
if self.half_open_attempts >= 3:
self.circuit_state = CircuitState.CLOSED
logger.info("Circuit breaker CLOSED after successful recovery")
async def _record_failure(self):
"""Record failed request for circuit breaker."""
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.circuit_state = CircuitState.OPEN
self.circuit_opened_at = time.time()
logger.warning(f"Circuit breaker OPENED after {self.failure_count} failures")
async def _rate_limit(self):
"""Enforce rate limiting between requests."""
now = time.time()
time_since_last = now - self.last_request_time
if time_since_last < self.min_request_interval:
await asyncio.sleep(self.min_request_interval - time_since_last)
self.last_request_time = time.time()
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: int = 2048
) -> APIResponse:
"""Single chat completion request with full error handling."""
request_id = f"req_{int(time.time() * 1000)}"
start_time = time.time()
async with self.semaphore:
if not await self._check_circuit_breaker():
return APIResponse(
request_id=request_id,
status_code=503,
data={},
latency_ms=0,
error="Circuit breaker is OPEN"
)
await self._rate_limit()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
for attempt in range(self.max_retries):
try:
async with aiohttp.ClientSession(timeout=self.timeout) as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
latency_ms = (time.time() - start_time) * 1000
if response.status == 200:
data = await response.json()
await self._record_success()
return APIResponse(
request_id=request_id,
status_code=200,
data=data,
latency_ms=latency_ms
)
elif response.status == 429:
# Rate limited - wait and retry
await asyncio.sleep(2 ** attempt)
continue
else:
error_text = await response.text()
await self._record_failure()
return APIResponse(
request_id=request_id,
status_code=response.status,
data={},
latency_ms=latency_ms,
error=f"HTTP {response.status}: {error_text}"
)
except asyncio.TimeoutError:
await self._record_failure()
if attempt == self.max_retries - 1:
return APIResponse(
request_id=request_id,
status_code=408,
data={},
latency_ms=(time.time() - start_time) * 1000,
error="Request timeout"
)
except Exception as e:
await self._record_failure()
return APIResponse(
request_id=request_id,
status_code=500,
data={},
latency_ms=(time.time() - start_time) * 1000,
error=str(e)
)
return APIResponse(
request_id=request_id,
status_code=429,
data={},
latency_ms=(time.time() - start_time) * 1000,
error="Max retries exceeded"
)
async def batch_chat_completions(
self,
requests: List[Dict[str, Any]],
model: str = "deepseek-v3.2"
) -> List[APIResponse]:
"""Execute multiple chat completion requests concurrently."""
logger.info(f"Starting batch of {len(requests)} concurrent requests")
start_time = time.time()
tasks = [
self.chat_completion(
messages=req["messages"],
model=model,
temperature=req.get("temperature", 0.7),
max_tokens=req.get("max_tokens", 2048)
)
for req in requests
]
responses = await asyncio.gather(*tasks, return_exceptions=True)
total_time = (time.time() - start_time) * 1000
logger.info(f"Batch completed in {total_time:.2f}ms")
# Convert any exceptions to error responses
processed_responses = []
for i, resp in enumerate(responses):
if isinstance(resp, Exception):
processed_responses.append(APIResponse(
request_id=f"req_exception_{i}",
status_code=500,
data={},
latency_ms=0,
error=str(resp)
))
else:
processed_responses.append(resp)
return processed_responses
Example usage for e-commerce customer service
async def handle_customer_service_batch(customer_queries: List[str]) -> List[str]:
"""Process multiple customer queries concurrently."""
client = ConcurrentAPIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent=50,
timeout_seconds=30
)
requests = [
{
"messages": [
{"role": "system", "content": "You are a helpful e-commerce customer service assistant."},
{"role": "user", "content": query}
],
"temperature": 0.5,
"max_tokens": 500
}
for query in customer_queries
]
responses = await client.batch_chat_completions(requests)
return [
resp.data.get("choices", [{}])[0].get("message", {}).get("content", "")
if resp.status_code == 200 else f"Error: {resp.error}"
for resp in responses
]
Run concurrent batch processing
if __name__ == "__main__":
sample_queries = [
"Where's my order #12345?",
"How do I return a product?",
"What are your shipping options?",
"I need to change my delivery address",
"Do you have this item in size M?"
] * 10 # 50 total queries
start = time.time()
results = asyncio.run(handle_customer_service_batch(sample_queries))
elapsed = (time.time() - start) * 1000
print(f"Processed {len(sample_queries)} queries in {elapsed:.2f}ms")
print(f"Average per query: {elapsed/len(sample_queries):.2f}ms")
print(f"Throughput: {len(sample_queries)/(elapsed/1000):.1f} requests/second")Node.js Implementation with Worker Threads
For Node.js applications requiring maximum CPU-bound throughput alongside concurrent I/O, the following implementation leverages worker threads and connection pooling to achieve optimal performance on multi-core systems.
// concurrent-api-manager.js
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
const { Pool } = require('generic-pool');
const EventEmitter = require('events');
// Configuration constants
const HOLYSHEEP_BASE_URL = 'https://api.holysheep.ai/v1';
const HOLYSHEEP_API_KEY = process.env.HOLYSHEEP_API_KEY || 'YOUR_HOLYSHEEP_API_KEY';
class RateLimiter {
constructor(options = {}) {
this.maxRequests = options.maxRequests || 100;
this.windowMs = options.windowMs || 60000;
this.requests = [];
}
async acquire() {
const now = Date.now();
this.requests = this.requests.filter(t => now - t < this.windowMs);
if (this.requests.length >= this.maxRequests) {
const waitTime = this.windowMs - (now - this.requests[0]);
await new Promise(resolve => setTimeout(resolve, waitTime));
return this.acquire();
}
this.requests.push(now);
return true;
}
}
class CircuitBreaker extends EventEmitter {
constructor(options = {}) {
super();
this.failureThreshold = options.failureThreshold || 5;
this.resetTimeout = options.resetTimeout || 60000;
this.state = 'CLOSED';
this.failures = 0;
this.lastFailureTime = null;
}
async execute(fn) {
if (this.state === 'OPEN') {
if (Date.now() - this.lastFailureTime > this.resetTimeout) {
this.state = 'HALF_OPEN';
this.emit('half-open');
} else {
throw new Error('Circuit breaker is OPEN');
}
}
try {
const result = await fn();
this.onSuccess();
return result;
} catch (error) {
this.onFailure();
throw error;
}
}
onSuccess() {
this.failures = 0;
if (this.state === 'HALF_OPEN') {
this.state = 'CLOSED';
this.emit('closed');
}
}
onFailure() {
this.failures++;
this.lastFailureTime = Date.now();
if (this.failures >= this.failureThreshold) {
this.state = 'OPEN';
this.emit('open');
}
}
}
class ConcurrentAPIManager {
constructor(options = {}) {
this.apiKey = options.apiKey || HOLYSHEEP_API_KEY;
this.maxConcurrent = options.maxConcurrent || 50;
this.timeoutMs = options.timeoutMs || 30000;
this.maxRetries = options.maxRetries || 3;
this.rateLimiter = new RateLimiter({
maxRequests: options.maxRequestsPerWindow || 500,
windowMs: options.windowMs || 60000
});
this.circuitBreaker = new CircuitBreaker({
failureThreshold: options.failureThreshold || 5,
resetTimeout: options.resetTimeout || 60000
});
this.requestQueue = [];
this.activeRequests = 0;
this.metrics = {
totalRequests: 0,
successfulRequests: 0,
failedRequests: 0,
totalLatencyMs: 0,
rateLimitedRequests: 0
};
this.setupCircuitBreakerListeners();
}
setupCircuitBreakerListeners() {
this.circuitBreaker.on('open', () => {
console.error('[CircuitBreaker] Opened - pausing requests for 60s');
});
this.circuitBreaker.on('half-open', () => {
console.log('[CircuitBreaker] Half-open - allowing test requests');
});
this.circuitBreaker.on('closed', () => {
console.log('[CircuitBreaker] Closed - normal operation resumed');
});
}
async chatCompletion(messages, options = {}) {
const model = options.model || 'deepseek-v3.2';
const temperature = options.temperature ?? 0.7;
const maxTokens = options.maxTokens || 2048;
const requestId = req_${Date.now()}_${Math.random().toString(36).substr(2, 9)};
const startTime = Date.now();
return await this.circuitBreaker.execute(async () => {
await this.rateLimiter.acquire();
if (this.activeRequests >= this.maxConcurrent) {
return new Promise((resolve) => {
this.requestQueue.push({
messages, model, temperature, maxTokens, resolve, requestId, startTime
});
});
}
this.activeRequests++;
try {
const result = await this.executeRequest(
requestId, messages, model, temperature, maxTokens, startTime
);
this.metrics.successfulRequests++;
return result;
} catch (error) {
this.metrics.failedRequests++;
throw error;
} finally {
this.activeRequests--;
this.processQueue();
}
});
}
async executeRequest(requestId, messages, model, temperature, maxTokens, startTime) {
for (let attempt = 0; attempt < this.maxRetries; attempt++) {
try {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), this.timeoutMs);
const response = await fetch(${HOLYSHEEP_BASE_URL}/chat/completions, {
method: 'POST',
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
},
body: JSON.stringify({
model,
messages,
temperature,
max_tokens: maxTokens
}),
signal: controller.signal
});
clearTimeout(timeoutId);
const latencyMs = Date.now() - startTime;
this.metrics.totalLatencyMs += latencyMs;
this.metrics.totalRequests++;
if (response.ok) {
return await response.json();
}
if (response.status === 429) {
const retryAfter = parseInt(response.headers.get('Retry-After') || '2');
await new Promise(r => setTimeout(r, retryAfter * 1000));
continue;
}
const errorText = await response.text();
throw new Error(HTTP ${response.status}: ${errorText});
} catch (error) {
if (error.name === 'AbortError') {
if (attempt < this.maxRetries - 1) {
await new Promise(r => setTimeout(r, 1000 * (attempt + 1)));
continue;
}
throw new Error('Request timeout after max retries');
}
throw error;
}
}
throw new Error('Max retries exceeded');
}
processQueue() {
while (
this.requestQueue.length > 0 &&
this.activeRequests < this.maxConcurrent
) {
const request = this.requestQueue.shift();
this.activeRequests++;
this.executeRequest(
request.requestId,
request.messages,
request.model,
request.temperature,
request.maxTokens,
request.startTime
)
.then(result => request.resolve(result))
.catch(error => request.resolve({ error: error.message }))
.finally(() => {
this.activeRequests--;
this.processQueue();
});
}
}
async batchChatCompletions(requests) {
console.log([BatchManager] Starting batch of ${requests.length} requests);
const startTime = Date.now();
const promises = requests.map(req =>
this.chatCompletion(req.messages, {
model: req.model || 'deepseek-v3.2',
temperature: req.temperature ?? 0.7,
maxTokens: req.maxTokens || 2048
})
);
const results = await Promise.allSettled(promises);
const totalTime = Date.now() - startTime;
console.log([BatchManager] Batch completed in ${totalTime}ms);
return {
results,
metrics: {
...this.metrics,
batchDurationMs: totalTime,
averageLatencyMs: this.metrics.totalRequests > 0
? this.metrics.totalLatencyMs / this.metrics.totalRequests
: 0,
requestsPerSecond: totalTime > 0
? (requests.length / totalTime) * 1000
: 0
}
};
}
getMetrics() {
return {
...this.metrics,
activeRequests: this.activeRequests,
queuedRequests: this.requestQueue.length,
successRate: this.metrics.totalRequests > 0
? (this.metrics.successfulRequests / this.metrics.totalRequests * 100).toFixed(2) + '%'
: 'N/A'
};
}
}
// Worker thread implementation for CPU-intensive processing
function processInWorker(workerData) {
return new Promise((resolve, reject) => {
const worker = new Worker(__filename, {
workerData
});
worker.on('message', resolve);
worker.on('error', reject);
worker.on('exit', (code) => {
if (code !== 0) reject(new Error(Worker stopped with exit code ${code}));
});
});
}
// Example: Enterprise RAG System Batch Processing
async function processRAGQueries(queries, documentChunks) {
const apiManager = new ConcurrentAPIManager({
maxConcurrent: 50,
timeoutMs: 30000,
maxRequestsPerWindow: 1000,
maxRetries: 3
});
const requests = queries.map((query, index) => ({
messages: [
{
role: 'system',
content: 'You are a helpful assistant answering questions based on the provided context. Always cite your sources and be precise.'
},
{
role: 'user',
content: Context: ${documentChunks[index % documentChunks.length]}\n\nQuestion: ${query}
}
],
temperature: 0.3,
maxTokens: 1000
}));
const { results, metrics } = await apiManager.batchChatCompletions(requests);
console.log('RAG Batch Processing Complete:');
console.log( Total Requests: ${metrics.totalRequests});
console.log( Success Rate: ${metrics.successRate});
console.log( Average Latency: ${metrics.averageLatencyMs.toFixed(2)}ms);
console.log( Throughput: ${metrics.requestsPerSecond.toFixed(2)} req/s);
return results.map((result, i) => {
if (result.status === 'fulfilled') {
return result.value;
} else {
console.error(Request ${i} failed:, result.reason);
return { error: result.reason.message };
}
});
}
// Demo execution
if (isMainThread) {
(async () => {
const apiManager = new ConcurrentAPIManager({
maxConcurrent: 50,
maxRequestsPerWindow: 500
});
// Simulate 100 concurrent customer service queries
const testRequests = Array.from({ length: 100 }, (_, i) => ({
messages: [
{ role: 'system', content: 'You are a helpful customer service assistant.' },
{ role: 'user', content: Help me with order #${10000 + i} }
],
temperature: 0.5,
maxTokens: 300
}));
const startTime = Date.now();
const { metrics } = await apiManager.batchChatCompletions(testRequests);
const totalTime = Date.now() - startTime;
console.log('\n=== Performance Summary ===');
console.log(Total Time: ${totalTime}ms);
console.log(Throughput: ${(100 / (totalTime / 1000)).toFixed(2)} requests/second);
console.log(Success Rate: ${metrics.successRate});
console.log(Average Latency: ${metrics.averageLatencyMs.toFixed(2)}ms);
console.log(Final Metrics:, apiManager.getMetrics());
})();
}
module.exports = { ConcurrentAPIManager, CircuitBreaker, RateLimiter };Advanced Optimization: Connection Pooling and Request Batching
For maximum efficiency, implement HTTP connection pooling to reuse TCP connections and reduce handshake overhead. Additionally, when your use case supports it, batch multiple requests into single API calls where the provider supports it.
# advanced_optimization.py
"""
Advanced concurrent API optimization techniques including:
- Connection pooling with persistent HTTP sessions
- Request deduplication for identical queries
- Priority queue for critical requests
- Adaptive rate limiting based on response patterns
"""
import asyncio
import aiohttp
import hashlib
import time
from collections import defaultdict
from typing import Dict, List, Optional, Tuple, Any
from dataclasses import dataclass, field
from enum import Enum
import heapq
import uvloop
uvloop.install()
class Priority(Enum):
CRITICAL = 0
HIGH = 1
NORMAL = 2
LOW = 3
@dataclass(order=True)
class PrioritizedRequest:
priority: int
request_id: str = field(compare=False)
messages: List[Dict] = field(compare=False)
model: str = field(compare=False)
temperature: float = field(compare=False)
max_tokens: int = field(compare=False)
future: asyncio.Future = field(compare=False)
created_at: float = field(compare=False, default_factory=time.time)
class DeduplicationCache:
"""Cache to prevent duplicate identical requests."""
def __init__(self, ttl_seconds: int = 60):
self.cache: Dict[str, Tuple[Any, float]] = {}
self.ttl = ttl_seconds
def _hash_request(self, messages: List[Dict], model: str) -> str:
content = f"{model}:{str(messages)}"
return hashlib.sha256(content.encode()).hexdigest()[:32]
def get(self, messages: List[Dict], model: str) -> Optional[Any]:
key = self._hash_request(messages, model)
if key in self.cache:
result, timestamp = self.cache[key]
if time.time() - timestamp < self.ttl:
return result
del self.cache[key]
return None
def set(self, messages: List[Dict], model: str, result: Any):
key = self._hash_request(messages, model)
self.cache[key] = (result, time.time())
def cleanup(self):
"""Remove expired entries."""
now = time.time()
expired = [k for k, (_, t) in self.cache.items() if now - t >= self.ttl]
for k in expired:
del self.cache[k]
class AdaptiveRateLimiter:
"""Rate limiter that adjusts based on API response patterns."""
def __init__(self):
self.request_times: List[float] = []
self.error_times: List[float] = []
self.window_size = 60 # seconds
self.base_rate = 100 # requests per window
self.current_rate = self.base_rate
self.backoff_factor = 2
self.last_adjustment = time.time()
def record_success(self):
self.request_times.append(time.time())
self._maybe_adjust()
def record_error(self):
self.error_times.append(time.time())
self._maybe_adjust(increase_backoff=True)
def _maybe_adjust(self, increase_backoff: bool = False):
now = time.time()
if now - self.last_adjustment < 5: # Only adjust every 5 seconds
return
# Clean old entries
cutoff = now - self.window_size
self.request_times = [t for t in self.request_times if t > cutoff]
self.error_times = [t for t in self.error_times if t > cutoff]
error_rate = len(self.error_times) / max(len(self.request_times), 1)
if error_rate > 0.1: # More than 10% errors
self.current_rate = max(10, self.current_rate / self.backoff_factor)
elif error_rate < 0.01 and len(self.request_times) >= self.base_rate * 0.8:
self.current_rate = min(self.base_rate * 2, self.current_rate * 1.1)
self.last_adjustment = now
async def acquire(self):
"""Wait until rate limit allows request."""
while True:
cutoff = time.time() - self.window_size
recent_requests = [t for t in self.request_times if t > cutoff]
if len(recent_requests) < self.current_rate:
self.request_times.append(time.time())
return
# Calculate wait time
oldest = min(recent_requests)
wait_time = self.window_size - (time.time() - oldest)
if wait_time > 0:
await asyncio.sleep(wait_time)
class OptimizedAPIClient:
"""Production-grade optimized API client with all best practices."""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
max_connections: int = 100,
max_concurrent_requests: int = 50
):
self.api_key = api_key
self.base_url = base_url
# Connection pool configuration
self._connector = aiohttp.TCPConnector(
limit=max_connections,
limit_per_host=max_connections,
ttl_dns_cache=300,
keepalive_timeout=30
)
self._session: Optional[aiohttp.ClientSession] = None
self._session_lock = asyncio.Lock()
# Request management
self.semaphore = asyncio.Semaphore(max_concurrent_requests)
self.priority_queue: List[PrioritizedRequest] = []
self.queue_lock = asyncio.Lock()
self.worker_tasks: List[asyncio.Task] = []
self.num_workers = 10
# Optimization components
self.dedup_cache = DeduplicationCache(ttl_seconds=30)
self.rate_limiter = AdaptiveRateLimiter()
# Metrics
self.metrics = {
'total_requests': 0,
'cache_hits': 0,
'deduplicated': 0,
'errors': 0,
'latencies': []
}
async def _get_session(self) -> aiohttp.ClientSession:
if self._session is None or self._session.closed:
async with self._session_lock:
if self._session is None or self._session.closed:
timeout = aiohttp.ClientTimeout(total=30, connect=10)
self._session = aiohttp.ClientSession(
connector=self._connector,
timeout=timeout
)
return self._session
async def start_workers(self):
"""Start background worker tasks."""
for _ in range(self.num_workers):
task = asyncio.create_task(self._worker())
self.worker_tasks.append(task)
async def stop_workers(self):
"""Gracefully stop workers."""
async with self.queue_lock:
# Add sentinel values to wake up workers
for _ in range(self.num_workers):
future = asyncio.Future()
future.cancel()
heapq.heappush(
self.priority_queue,
PrioritizedRequest(
priority=Priority.CRITICAL.value,
request_id="sentinel",
messages=[],
model="",
temperature=0,
max_tokens=0,
future=future
)
)
# Cancel worker tasks
for task in self.worker_tasks:
task.cancel()
await asyncio.gather(*self.worker_tasks, return_exceptions=True)
self.worker_tasks.clear()
async def _worker(self):
"""Worker coroutine that processes priority queue."""
while True:
request = None
async with self.queue_lock:
if self.priority_queue:
request = heapq.heappop(self.priority_queue)
if request is None or request.request_id == "sentinel":
break
try:
result = await self._execute_single_request(
request.messages,
request.model,
request.temperature,
request.max_tokens
)
if not request.future.done():
request.future.set_result(result)
except Exception as e:
if not request.future.done():
request.future.set_exception(e)
async def _execute_single_request(
self,
messages: List[Dict],
model: str,
temperature: float,
max_tokens: int
) -> Dict[str, Any]:
"""Execute a single API request."""
start_time = time.time()
# Check deduplication cache
cached = self.dedup_cache.get(messages, model)
if cached:
self.metrics['cache_hits'] += 1
self.metrics['deduplicated'] += 1
return cached
await self.rate_limiter.acquire()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
session = await self._get_session()
async with self.semaphore:
try:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
latency = (time.time() - start_time) * 1000
self.metrics['latencies'].append(latency)
self.metrics['total_requests'] += 1
if response.status == 200:
self.rate_limiter.record_success()
result = await response.json()
self.dedup_cache.set(messages, model, result)
return result
else:
self.rate_limiter.record_error()
self.metrics['errors'] += 1
error_text = await response.text()
raise Exception(f"API Error {response.status}: {error_text}")
except asyncio.TimeoutError:
self.rate_limiter.record_error()
self.metrics['errors'] += 1
raise Exception("Request timeout")
except Exception as e:
self.rate_limiter.record_error()
self.metrics['errors'] += 1
raise
async def chat_completion(
self,
messages: List[Dict],
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: int = 2048,
priority: Priority = Priority.NORMAL
) -> Dict[str, Any]:
"""Add a chat completion request to the priority queue."""
request_id = f"{int(time.time() * 1000)}_{id(messages)}"
future = asyncio.Future()
request = PrioritizedRequest(
priority=priority.value,
request_id=request_id,
messages=messages,
model=model,
temperature=temperature,
max_tokens=max_tokens,
future=future
)
async with self.queue_lock:
heapq.heappush(self.priority_queue, request)
return await future
async def batch_chat_completions(
self,
requests: List[Dict[str, Any]],
priority: Priority = Priority.NORMAL
) -> List