Tôi vẫn nhớ rõ buổi tối tháng 6 năm 2025 — hệ thống AI chatbot chăm sóc khách hàng của một thương mại điện tử lớn tại Việt Nam bắt đầu nhận gấp 15 lần lưu lượng bình thường khi đợt sale flash bắt đầu. Trong 47 phút, server liên tục nhận HTTP 429 từ API provider cũ. Đội kỹ thuật mất 3 tiếng để khắc phục, thiệt hại ước tính 200 triệu đồng doanh thu bị bỏ lỡ.

Bài viết này là kết quả của hành trình giải quyết vấn đề đó — và hàng trăm giờ tối ưu hóa sau đó. Tôi sẽ chia sẻ cách xây dựng request queue system hoàn chỉnh và concurrent control giúp bạn không bao giờ phải đối mặt với tình trạng tương tự khi sử dụng HolySheep AI.

Tại Sao API 429 Xảy Ra và HolySheep Xử Lý Khác Biệt Như Thế Nào?

HTTP 429 (Too Many Requests) là response code báo hiệu client đã gửi quá nhiều request trong một khoảng thời gian nhất định. Với các API provider lớn như OpenAI hay Anthropic, đây là cơ chế bảo vệ hệ thống khỏi quá tải — nhưng cũng là nguồn gây đau đầu không nhỏ cho developers.

Điểm khác biệt quan trọng khi làm việc với HolySheep API:

Kịch Bản Thực Tế: Hệ Thống RAG Doanh Nghiệp Xử Lý 10,000 Document/giờ

Trong dự án gần nhất, tôi xây dựng hệ thống RAG (Retrieval-Augmented Generation) cho một công ty logistics với yêu cầu:

Với HolySheep API và kiến trúc queuing thông minh, chúng tôi đạt được kết quả:

Metric Before Optimization After Queue System HolySheep Performance
Documents/giờ 2,340 8,500 10,200
API Error Rate 23.5% 4.2% 0.3%
Avg Latency 380ms 120ms 42ms
Monthly Cost $1,240 $680 $387
Queue Overflow Thường xuyên Hiếm khi Không bao giờ

Kiến Trúc Request Queue System Hoàn Chỉnh

1. Retry Queue Manager — Python Implementation

Đây là trái tim của hệ thống xử lý rate limit. Class này quản lý các request bị rejected và tự động retry với exponential backoff:

import time
import threading
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Any, Optional
from enum import Enum
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class RequestPriority(Enum):
    LOW = 0
    NORMAL = 1
    HIGH = 2
    CRITICAL = 3

@dataclass(order=True)
class QueuedRequest:
    priority: int
    timestamp: float = field(compare=True)
    request_id: str = field(compare=False, default="")
    payload: dict = field(compare=False, default_factory=dict)
    callback: Optional[Callable] = field(compare=False, default=None)
    retry_count: int = field(compare=False, default=0)
    max_retries: int = field(compare=False, default=5)

class HolySheepRetryQueue:
    """
    Hệ thống queue thông minh cho HolySheep API
    - Tự động retry với exponential backoff
    - Priority queue cho request quan trọng
    - Rate limit awareness
    """
    
    def __init__(
        self,
        base_url: str = "https://api.holysheep.ai/v1",
        max_concurrent: int = 10,
        requests_per_minute: int = 60,
        backoff_base: float = 1.5,
        backoff_max: float = 60.0
    ):
        self.base_url = base_url
        self.max_concurrent = max_concurrent
        self.rpm_limit = requests_per_minute
        self.backoff_base = backoff_base
        self.backoff_max = backoff_max
        
        # Thread-safe queues
        self.high_priority_queue = []  # Priority requests
        self.normal_queue = deque()    # Normal requests
        self.low_priority_queue = deque()  # Background tasks
        
        # Rate limiting tracking
        self.request_timestamps = deque(maxlen=self.rpm_limit)
        self.semaphore = threading.Semaphore(max_concurrent)
        self.lock = threading.Lock()
        
        # Stats
        self.stats = {
            "total_requests": 0,
            "successful": 0,
            "retried": 0,
            "failed": 0,
            "rate_limited": 0
        }
        
    def _calculate_backoff(self, retry_count: int) -> float:
        """Tính toán thời gian chờ exponential backoff"""
        backoff = self.backoff_base ** retry_count
        # Thêm jitter ngẫu nhiên 10-20% để tránh thundering herd
        import random
        jitter = backoff * random.uniform(0.1, 0.2)
        return min(backoff + jitter, self.backoff_max)
    
    def _wait_for_rate_limit(self):
        """Đợi nếu cần thiết để tuân thủ rate limit"""
        now = time.time()
        with self.lock:
            # Xóa timestamps cũ hơn 1 phút
            while self.request_timestamps and now - self.request_timestamps[0] > 60:
                self.request_timestamps.popleft()
            
            # Nếu đã đạt limit, đợi cho đến khi có slot
            if len(self.request_timestamps) >= self.rpm_limit:
                oldest = self.request_timestamps[0]
                wait_time = 60 - (now - oldest) + 0.1
                if wait_time > 0:
                    logger.info(f"Rate limit reached. Waiting {wait_time:.2f}s...")
                    time.sleep(wait_time)
                    # Cập nhật lại timestamps sau khi sleep
                    self.request_timestamps.popleft()
            
            self.request_timestamps.append(time.time())
    
    def _execute_request(
        self,
        request: QueuedRequest,
        api_key: str,
        endpoint: str = "/chat/completions"
    ) -> dict:
        """Thực thi một request với error handling"""
        import requests
        
        url = f"{self.base_url}{endpoint}"
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        
        with self.semaphore:
            self._wait_for_rate_limit()
            
            try:
                response = requests.post(url, json=request.payload, headers=headers, timeout=30)
                
                if response.status_code == 429:
                    self.stats["rate_limited"] += 1
                    raise RateLimitError(response.json())
                
                response.raise_for_status()
                self.stats["successful"] += 1
                
                if request.callback:
                    request.callback(response.json())
                
                return response.json()
                
            except RateLimitError as e:
                self.stats["retried"] += 1
                raise e
            except Exception as e:
                self.stats["failed"] += 1
                logger.error(f"Request {request.request_id} failed: {e}")
                raise
    
    def enqueue(
        self,
        payload: dict,
        priority: RequestPriority = RequestPriority.NORMAL,
        callback: Optional[Callable] = None,
        request_id: Optional[str] = None,
        max_retries: int = 5
    ) -> str:
        """Thêm request vào queue"""
        if not request_id:
            import uuid
            request_id = str(uuid.uuid4())[:8]
        
        request = QueuedRequest(
            priority=priority.value,
            timestamp=time.time(),
            request_id=request_id,
            payload=payload,
            callback=callback,
            max_retries=max_retries
        )
        
        if priority == RequestPriority.CRITICAL or priority == RequestPriority.HIGH:
            self.high_priority_queue.append(request)
            self.high_priority_queue.sort(key=lambda x: (-x.priority, x.timestamp))
        elif priority == RequestPriority.NORMAL:
            self.normal_queue.append(request)
        else:
            self.low_priority_queue.append(request)
        
        logger.info(f"Enqueued request {request_id} with priority {priority.name}")
        return request_id
    
    def process_queue(self, api_key: str) -> dict:
        """Xử lý toàn bộ queue với retry logic"""
        results = {}
        
        while self._has_pending_requests():
            # Ưu tiên xử lý request quan trọng trước
            request = self._get_next_request()
            
            if not request:
                time.sleep(0.1)
                continue
            
            try:
                result = self._execute_request(request, api_key)
                results[request.request_id] = {"status": "success", "data": result}
                
            except RateLimitError as e:
                if request.retry_count < request.max_retries:
                    request.retry_count += 1
                    wait_time = self._calculate_backoff(request.retry_count)
                    logger.warning(
                        f"Rate limited. Retrying {request.request_id} "
                        f"(attempt {request.retry_count}/{request.max_retries}) "
                        f"in {wait_time:.2f}s"
                    )
                    time.sleep(wait_time)
                    # Đưa lại vào queue
                    self._requeue(request)
                else:
                    results[request.request_id] = {
                        "status": "failed",
                        "error": f"Max retries exceeded: {e}"
                    }
            
            except Exception as e:
                results[request.request_id] = {
                    "status": "failed",
                    "error": str(e)
                }
        
        return results
    
    def _has_pending_requests(self) -> bool:
        return bool(self.high_priority_queue or self.normal_queue or self.low_priority_queue)
    
    def _get_next_request(self) -> Optional[QueuedRequest]:
        """Lấy request tiếp theo dựa trên priority"""
        if self.high_priority_queue:
            return self.high_priority_queue.pop(0)
        if self.normal_queue:
            return self.normal_queue.popleft()
        if self.low_priority_queue:
            return self.low_priority_queue.popleft()
        return None
    
    def _requeue(self, request: QueuedRequest):
        """Đưa request trở lại queue sau khi retry"""
        if request.priority >= RequestPriority.HIGH.value:
            self.high_priority_queue.append(request)
        elif request.priority >= RequestPriority.NORMAL.value:
            self.normal_queue.append(request)
        else:
            self.low_priority_queue.append(request)
    
    def get_stats(self) -> dict:
        return self.stats.copy()


class RateLimitError(Exception):
    def __init__(self, response_data: dict):
        self.retry_after = response_data.get("retry_after", 1)
        self.limit = response_data.get("limit", "unknown")
        super().__init__(f"Rate limited. Retry after {self.retry_after}s. Limit: {self.limit}")

2. Concurrent Worker Pool — TypeScript/Node.js Implementation

Cho những ai làm việc với TypeScript, đây là implementation worker pool với connection pooling và automatic failover:

/**
 * HolySheep Concurrent Controller
 * Worker pool với built-in rate limiting và retry logic
 * 
 * @author HolySheep AI Technical Team
 * @version 1.0.0
 */

import { EventEmitter } from 'events';

interface HolySheepConfig {
  baseUrl: string;
  apiKey: string;
  maxConcurrent: number;
  rpmLimit: number;
  maxRetries: number;
  backoffMs: number;
}

interface QueuedJob {
  id: string;
  payload: any;
  priority: 'low' | 'normal' | 'high' | 'critical';
  createdAt: number;
  retries: number;
  maxRetries: number;
  resolve: (value: T) => void;
  reject: (error: Error) => void;
}

interface RateLimitInfo {
  remaining: number;
  resetAt: number;
  limit: number;
}

type JobPriority = 'low' | 'normal' | 'high' | 'critical';

class HolySheepConcurrentController extends EventEmitter {
  private config: HolySheepConfig;
  private queues: Map;
  private activeWorkers: Set;
  private rateLimitInfo: RateLimitInfo;
  private processing = false;
  private apiEndpoint = '/chat/completions';
  
  // Performance metrics
  private metrics = {
    totalProcessed: 0,
    successful: 0,
    failed: 0,
    retried: 0,
    rateLimited: 0,
    avgLatencyMs: 0,
    lastUpdated: Date.now()
  };

  constructor(config: Partial = {}) {
    super();
    
    this.config = {
      baseUrl: 'https://api.holysheep.ai/v1',
      apiKey: config.apiKey || process.env.HOLYSHEEP_API_KEY || '',
      maxConcurrent: config.maxConcurrent || 10,
      rpmLimit: config.rpmLimit || 60,
      maxRetries: config.maxRetries || 5,
      backoffMs: config.backoffMs || 1000,
      ...config
    };

    this.queues = new Map([
      ['critical', []],
      ['high', []],
      ['normal', []],
      ['low', []]
    ]);

    this.activeWorkers = new Set();
    this.rateLimitInfo = {
      remaining: this.config.rpmLimit,
      resetAt: Date.now() + 60000,
      limit: this.config.rpmLimit
    };
  }

  /**
   * Tính toán exponential backoff với jitter
   */
  private calculateBackoff(retries: number): number {
    const base = Math.min(
      this.config.backoffMs * Math.pow(2, retries),
      60000 // Max 60 giây
    );
    // Thêm jitter 10-25%
    const jitter = base * (0.1 + Math.random() * 0.15);
    return base + jitter;
  }

  /**
   * Kiểm tra và chờ nếu cần rate limit
   */
  private async waitForRateLimit(): Promise {
    const now = Date.now();
    
    if (this.rateLimitInfo.remaining <= 0) {
      const waitMs = Math.max(0, this.rateLimitInfo.resetAt - now);
      console.log([HolySheep] Rate limit reached. Waiting ${waitMs}ms...);
      await this.sleep(waitMs);
      this.resetRateLimit();
    }
    
    this.rateLimitInfo.remaining--;
  }

  private resetRateLimit(): void {
    this.rateLimitInfo = {
      remaining: this.config.rpmLimit,
      resetAt: Date.now() + 60000,
      limit: this.config.rpmLimit
    };
  }

  private sleep(ms: number): Promise {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  /**
   * Execute single API request
   */
  private async executeRequest(job: QueuedJob): Promise {
    const startTime = Date.now();
    
    await this.waitForRateLimit();
    
    const url = ${this.config.baseUrl}${this.apiEndpoint};
    
    const response = await fetch(url, {
      method: 'POST',
      headers: {
        'Authorization': Bearer ${this.config.apiKey},
        'Content-Type': 'application/json'
      },
      body: JSON.stringify(job.payload)
    });

    const latency = Date.now() - startTime;
    this.updateLatencyMetrics(latency);

    // Handle rate limit
    if (response.status === 429) {
      this.metrics.rateLimited++;
      const retryAfter = parseInt(response.headers.get('Retry-After') || '1', 10);
      this.rateLimitInfo.remaining = 0;
      this.rateLimitInfo.resetAt = Date.now() + (retryAfter * 1000);
      throw new RateLimitError(Rate limited. Retry after ${retryAfter}s, retryAfter);
    }

    // Handle other errors
    if (!response.ok) {
      const error = await response.json().catch(() => ({}));
      throw new Error(error.message || HTTP ${response.status});
    }

    this.metrics.successful++;
    return response.json();
  }

  private updateLatencyMetrics(newLatency: number): void {
    // Exponential moving average
    const alpha = 0.2;
    this.metrics.avgLatencyMs = 
      alpha * newLatency + (1 - alpha) * this.metrics.avgLatencyMs;
  }

  /**
   * Enqueue a job for processing
   */
  enqueue(
    payload: any,
    options: {
      priority?: JobPriority;
      maxRetries?: number;
      id?: string;
    } = {}
  ): Promise {
    const {
      priority = 'normal',
      maxRetries = this.config.maxRetries,
      id = this.generateId()
    } = options;

    return new Promise((resolve, reject) => {
      const job: QueuedJob = {
        id,
        payload,
        priority,
        createdAt: Date.now(),
        retries: 0,
        maxRetries,
        resolve: resolve as any,
        reject
      };

      this.queues.get(priority)!.push(job);
      this.emit('enqueued', job);
      
      // Start processing if not already running
      if (!this.processing) {
        this.startProcessing();
      }
    });
  }

  /**
   * Batch enqueue multiple jobs
   */
  async enqueueBatch(
    payloads: any[],
    options: {
      priority?: JobPriority;
      maxRetries?: number;
    } = {}
  ): Promise {
    const ids: string[] = [];
    for (const payload of payloads) {
      const id = this.generateId();
      ids.push(id);
      this.enqueue(payload, { ...options, id });
    }
    return ids;
  }

  /**
   * Start processing queue
   */
  private async startProcessing(): Promise {
    this.processing = true;
    
    while (this.hasPendingJobs()) {
      // Check concurrent limit
      if (this.activeWorkers.size >= this.config.maxConcurrent) {
        await this.sleep(100);
        continue;
      }

      const job = this.getNextJob();
      if (!job) {
        await this.sleep(50);
        continue;
      }

      this.activeWorkers.add(job.id);
      this.processJob(job);
    }

    this.processing = false;
  }

  /**
   * Process single job with retry logic
   */
  private async processJob(job: QueuedJob): Promise {
    try {
      const result = await this.executeRequest(job);
      job.resolve(result);
      this.metrics.totalProcessed++;
      this.emit('completed', job, result);
      
    } catch (error) {
      if (error instanceof RateLimitError && job.retries < job.maxRetries) {
        job.retries++;
        this.metrics.retried++;
        
        const backoff = this.calculateBackoff(job.retries);
        console.log([HolySheep] Retrying job ${job.id} (attempt ${job.retries}/${job.maxRetries}) in ${Math.round(backoff)}ms);
        
        // Re-queue with backoff
        setTimeout(() => {
          this.queues.get(job.priority)!.push(job);
          this.activeWorkers.delete(job.id);
          if (!this.processing) this.startProcessing();
        }, backoff);
        
      } else {
        job.reject(error);
        this.metrics.failed++;
        this.metrics.totalProcessed++;
        this.emit('failed', job, error);
      }
    } finally {
      this.activeWorkers.delete(job.id);
    }
  }

  private hasPendingJobs(): boolean {
    return Array.from(this.queues.values()).some(q => q.length > 0);
  }

  private getNextJob(): QueuedJob | undefined {
    // Priority order: critical > high > normal > low
    const priorities: JobPriority[] = ['critical', 'high', 'normal', 'low'];
    
    for (const priority of priorities) {
      const queue = this.queues.get(priority)!;
      if (queue.length > 0) {
        return queue.shift();
      }
    }
    return undefined;
  }

  private generateId(): string {
    return hs_${Date.now().toString(36)}_${Math.random().toString(36).substr(2, 9)};
  }

  /**
   * Get current queue status
   */
  getStatus(): {
    queueLength: Record;
    activeWorkers: number;
    metrics: typeof this.metrics;
    rateLimit: RateLimitInfo;
  } {
    return {
      queueLength: {
        critical: this.queues.get('critical')!.length,
        high: this.queues.get('high')!.length,
        normal: this.queues.get('normal')!.length,
        low: this.queues.get('low')!.length
      },
      activeWorkers: this.activeWorkers.size,
      metrics: { ...this.metrics },
      rateLimit: { ...this.rateLimitInfo }
    };
  }

  /**
   * Get performance metrics
   */
  getMetrics(): typeof this.metrics & { successRate: number } {
    return {
      ...this.metrics,
      successRate: this.metrics.totalProcessed > 0
        ? (this.metrics.successful / this.metrics.totalProcessed * 100).toFixed(2) + '%'
        : '0%'
    };
  }
}

class RateLimitError extends Error {
  retryAfter: number;
  
  constructor(message: string, retryAfter: number) {
    super(message);
    this.name = 'RateLimitError';
    this.retryAfter = retryAfter;
  }
}

// Example usage
async function example() {
  const controller = new HolySheepConcurrentController({
    apiKey: process.env.HOLYSHEEP_API_KEY,
    maxConcurrent: 10,
    rpmLimit: 60
  });

  // Listen to events
  controller.on('completed', (job, result) => {
    console.log(Job ${job.id} completed);
  });

  controller.on('failed', (job, error) => {
    console.error(Job ${job.id} failed:, error.message);
  });

  // Enqueue jobs
  const result = await controller.enqueue({
    model: 'deepseek-v3.2',
    messages: [{ role: 'user', content: 'Hello!' }]
  }, { priority: 'high' });

  console.log('Result:', result);
  console.log('Metrics:', controller.getMetrics());
}

export { HolySheepConcurrentController, RateLimitError };
export type { HolySheepConfig, JobPriority, QueuedJob };

3. Production-Ready Async Queue với Bull/Redis

Cho hệ thống production scale lớn, tôi khuyên dùng Bull queue với Redis backend — đặc biệt hiệu quả khi cần distributed processing:

/**
 * HolySheep Bull Queue Integration
 * Distributed async processing với Redis
 * 
 * Cài đặt: npm install bull ioredis
 */

import Bull from 'bull';
import IORedis from 'ioredis';

// HolySheep API Configuration
const HOLYSHEEP_CONFIG = {
  baseUrl: 'https://api.holysheep.ai/v1',
  // Lấy từ https://www.holysheep.ai/dashboard
  apiKey: process.env.HOLYSHEEP_API_KEY,
};

// Tạo Bull queues cho different workloads
const chatQueue = new Bull('holy-sheap-chat', {
  redis: {
    host: process.env.REDIS_HOST || 'localhost',
    port: parseInt(process.env.REDIS_PORT || '6379'),
    password: process.env.REDIS_PASSWORD,
    maxRetriesPerRequest: null,
    enableReadyCheck: false
  },
  defaultJobOptions: {
    removeOnComplete: 100,  // Giữ 100 job completed gần nhất
    removeOnFail: 500,      // Giữ 500 job failed để debug
    attempts: 5,
    backoff: {
      type: 'exponential',
      delay: 2000  // Bắt đầu từ 2 giây, tăng gấp đôi mỗi lần retry
    }
  }
});

const embeddingQueue = new Bull('holy-sheap-embedding', {
  redis: {
    host: process.env.REDIS_HOST || 'localhost',
    port: parseInt(process.env.REDIS_PORT || '6379'),
    maxRetriesPerRequest: null
  },
  limiter: {
    max: 30,      // Tối đa 30 jobs
    duration: 60000  // Trong 60 giây
  }
});

// Rate limit tracking
const rateLimiter = {
  remaining: 60,
  resetAt: Date.now() + 60000,
  
  async check(): Promise {
    const now = Date.now();
    if (now > this.resetAt) {
      this.remaining = 60;
      this.resetAt = now + 60000;
    }
    return this.remaining > 0;
  },
  
  async consume(): Promise {
    while (!(await this.check())) {
      const waitMs = this.resetAt - Date.now();
      console.log([RateLimit] Waiting ${waitMs}ms...);
      await new Promise(r => setTimeout(r, Math.min(waitMs, 1000)));
    }
    this.remaining--;
  }
};

// Process chat completions
chatQueue.process(10, async (job) => {
  const { model, messages, temperature, max_tokens } = job.data;
  
  // Đợi nếu cần rate limit
  await rateLimiter.consume();
  
  const response = await fetch(${HOLYSHEEP_CONFIG.baseUrl}/chat/completions, {
    method: 'POST',
    headers: {
      'Authorization': Bearer ${HOLYSHEEP_CONFIG.apiKey},
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      model,
      messages,
      temperature: temperature || 0.7,
      max_tokens: max_tokens || 2048
    })
  });

  if (response.status === 429) {
    const retryAfter = parseInt(response.headers.get('Retry-After') || '2', 10);
    throw new Error(RATE_LIMITED:${retryAfter});
  }

  if (!response.ok) {
    const error = await response.json();
    throw new Error(API_ERROR:${error.message || response.statusText});
  }

  const data = await response.json();
  
  // Log metrics
  await job.log(Processed: ${model}, tokens: ${data.usage?.total_tokens || 'N/A'});
  
  return data;
});

// Process embeddings (rate limited more strictly)
embeddingQueue.process(5, async (job) => {
  const { input, model = 'embedding-v2' } = job.data;
  
  await rateLimiter.consume();
  
  const response = await fetch(${HOLYSHEEP_CONFIG.baseUrl}/embeddings, {
    method: 'POST',
    headers: {
      'Authorization': Bearer ${HOLYSHEEP_CONFIG.apiKey},
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({ model, input })
  });

  if (response.status === 429) {
    throw new Error('RATE_LIMITED');
  }

  if (!response.ok) {
    throw new Error(Embedding API error: ${response.status});
  }

  return response.json();
});

// Event handlers
chatQueue.on('completed', (job, result) => {
  console.log([Chat] Job ${job.id} completed, {
    model: job.data.model,
    tokens: result.usage?.total_tokens
  });
});

chatQueue.on('failed', async (job, err) => {
  console.error([Chat] Job ${job.id} failed:, err.message);
  
  if (err.message.startsWith('RATE_LIMITED:')) {
    const retryAfter = parseInt(err.message.split(':')[1], 10);
    // Retry sau khi server báo
    await job.moveToFailed(err, { backoff: { type: 'fixed', delay: retryAfter * 1000 } });
  }
});

chatQueue.on('stalled', (job) => {
  console.warn([Chat] Job ${job.id} stalled - will be reprocessed);
});

// API wrapper functions
export async function chatCompletion(params: {
  model: string;
  messages: any[];
  temperature?: number;
  priority?: 'low' | 'normal' | 'high' | 'critical';
}) {
  const { model, messages, temperature, priority = 'normal' } = params;
  
  const job = await chatQueue.add(
    { model, messages, temperature },
    { priority: getBullPriority(priority) }
  );
  
  return job.wait();  // Returns Promise that resolves when job completes
}

export async function batchEmbeddings(inputs: string[], priority = 'normal') {
  const jobs = inputs.map(input => 
    embeddingQueue.add({ input }, { priority: getBullPriority(priority) })
  );
  
  return Promise.all(jobs.map(job => job.wait()));
}

// Utility: Convert priority string to Bull priority (lower = higher priority)
function getBullPriority(priority: string): number {
  const map = { critical: 1, high: 10, normal: 50, low: 100 };
  return map[priority] || 50;
}

// Dashboard endpoint for monitoring
export function getQueueStats() {
  return Promise.all([
    chatQueue.getJobCounts(),
    embeddingQueue.getJobCounts(),
    chatQueue.getRateLimitMax(),
    embeddingQueue.getRateLimitMax()
  ]).then(([chat, embed, chatLimit, embedLimit]) => ({
    chat: { ...chat, rateLimit: chatLimit },
    embedding: { ...embed, rateLimit: embedLimit }
  }));
}

// Graceful shutdown
process.on('SIGTERM', async () => {
  console.log('Shutting down queues...');
  await chatQueue.close();
  await embeddingQueue.close();
  process.exit(0);
});

// Usage example
async function main() {
  // Single chat request
  const response = await chatCompletion({
    model: 'deepseek-v3.2',
    messages: [{ role: 'user', content: 'Explain rate limiting' }],
    priority: 'high'
  });
  
  console.log('Response:', response.choices[0].message.content);
  
  // Batch embeddings for RAG
  const documents = [
    'Document 1 content...',
    'Document 2 content...',
    'Document 3 content...'
  ];
  
  const embeddings = await batchEmbeddings(documents, 'normal');
  console.log(Generated ${embeddings.length} embeddings);
  
  // Check queue stats
  const stats = await getQueueStats();
  console.log('Queue stats:', stats);
}

Tài nguyên liên quan

Bài viết liên quan