AI API 并发连接数限制突破方案： Giải pháp thực chiến cho doanh nghiệp Việt Nam

Ngày 15 tháng 3 năm 2026, tôi đang deploy hệ thống chatbot AI cho một doanh nghiệp bán lẻ lớn tại TP.HCM. Họ cần xử lý 500+ yêu cầu đồng thời trong giờ cao điểm. Và rồi, thảm họa ập đến:

ERROR: ConnectionError: timeout after 30s
ERROR: 429 Too Many Requests - Rate limit exceeded
ERROR: 401 Unauthorized - Invalid API key
ERROR: Connection pool exhausted, max connections: 100

Đây là bài học đắt giá nhất trong sự nghiệp tôi. Sau 72 giờ không ngủ, tôi đã tìm ra giải pháp toàn diện để vượt qua giới hạn concurrent connection của AI API. Bài viết này sẽ chia sẻ toàn bộ kinh nghiệm thực chiến, kèm code mẫu có thể chạy ngay.

Tại Sao AI API Lại Giới Hạn Concurrent Connection?

Trước khi đi vào giải pháp, hãy hiểu rõ vấn đề nền tảng. Các nhà cung cấp AI API như OpenAI, Anthropic, Google đều áp dụng giới hạn:

• Rate Limit: Số request được phép gửi trong một khoảng thời gian (thường tính bằng RPM - requests per minute)
• Token Limit: Tổng token xử lý trong phút (TPM - tokens per minute)
• Concurrent Connection Limit: Số kết nối đồng thời tối đa được mở
• TPM per Connection: Giới hạn token cho mỗi kết nối

Với gói Free tier của OpenAI, bạn chỉ được 3 concurrent connections. Gói Pay-as-you-go nâng lên 60. Ngay cả gói Enterprise cũng chỉ có 500. Điều này gây ra bottleneck nghiêm trọng cho ứng dụng cần xử lý lớn.

Giải Pháp 1: Intelligent Request Queue với Retry Logic

Đây là phương pháp tôi sử dụng đầu tiên và đã giải quyết 80% vấn đề. Code dưới đây implement một queue thông minh với exponential backoff.

import asyncio
import aiohttp
import time
from collections import deque
from typing import Optional, Callable, Any
import logging

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

class IntelligentRequestQueue:
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        max_concurrent: int = 50,
        max_retries: int = 5,
        base_delay: float = 1.0
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.request_queue = deque()
        self.stats = {"success": 0, "failed": 0, "retried": 0}
        
    async def _make_request_with_retry(
        self,
        session: aiohttp.ClientSession,
        endpoint: str,
        payload: dict,
        retry_count: int = 0
    ) -> dict:
        """Thực hiện request với retry logic và exponential backoff"""
        
        async with self.semaphore:
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }
            
            try:
                async with session.post(
                    f"{self.base_url}{endpoint}",
                    json=payload,
                    headers=headers,
                    timeout=aiohttp.ClientTimeout(total=60)
                ) as response:
                    
                    if response.status == 200:
                        self.stats["success"] += 1
                        return await response.json()
                    
                    elif response.status == 429:
                        # Rate limit - retry với backoff
                        if retry_count < self.max_retries:
                            delay = self.base_delay * (2 ** retry_count)
                            self.stats["retried"] += 1
                            logger.warning(
                                f"Rate limited, retry {retry_count + 1}/{self.max_retries} "
                                f"after {delay}s"
                            )
                            await asyncio.sleep(delay)
                            return await self._make_request_with_retry(
                                session, endpoint, payload, retry_count + 1
                            )
                        else:
                            self.stats["failed"] += 1
                            raise Exception(f"Max retries exceeded after {response.status}")
                    
                    elif response.status == 401:
                        self.stats["failed"] += 1
                        raise Exception("401 Unauthorized - Kiểm tra API key")
                    
                    else:
                        self.stats["failed"] += 1
                        error_text = await response.text()
                        raise Exception(f"HTTP {response.status}: {error_text}")
                        
            except asyncio.TimeoutError:
                self.stats["failed"] += 1
                logger.error("Request timeout sau 60 giây")
                raise
                
            except aiohttp.ClientError as e:
                self.stats["failed"] += 1
                logger.error(f"Connection error: {str(e)}")
                raise

    async def batch_chat_completions(
        self,
        messages_list: list,
        model: str = "gpt-4.1"
    ) -> list:
        """Xử lý batch nhiều request chat cùng lúc"""
        results = []
        
        connector = aiohttp.TCPConnector(
            limit=100,
            limit_per_host=50,
            ttl_dns_cache=300
        )
        
        async with aiohttp.ClientSession(connector=connector) as session:
            tasks = []
            for messages in messages_list:
                task = self._make_request_with_retry(
                    session,
                    "/chat/completions",
                    {"model": model, "messages": messages}
                )
                tasks.append(task)
            
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
        return results
    
    def get_stats(self) -> dict:
        return self.stats

Sử dụng
async def main():
    queue = IntelligentRequestQueue(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        max_concurrent=50,
        max_retries=5
    )
    
    # Tạo 200 requests mẫu
    messages_list = [
        [{"role": "user", "content": f"Tin nhắn {i}"}]
        for i in range(200)
    ]
    
    start = time.time()
    results = await queue.batch_chat_completions(messages_list)
    elapsed = time.time() - start
    
    print(f"Hoàn thành trong {elapsed:.2f}s")
    print(f"Stats: {queue.get_stats()}")
    print(f"Throughput: {200/elapsed:.2f} requests/giây")

if __name__ == "__main__":
    asyncio.run(main())

Giải Pháp 2: Connection Pooling với Smart Load Balancer

Với các hệ thống production cần xử lý hàng nghìn request/giây, bạn cần implement connection pooling thông minh. Phương pháp này giúp tái sử dụng kết nối và phân phối tải đều.

import threading
import queue
import time
from dataclasses import dataclass
from typing import List, Dict, Optional
import requests
from concurrent.futures import ThreadPoolExecutor, Future
import heapq

@dataclass
class ConnectionNode:
    connection_id: str
    is_available: bool = True
    last_used: float = 0.0
    request_count: int = 0
    avg_latency: float = 0.0

class SmartConnectionPool:
    """Connection pool với load balancing thông minh"""
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        pool_size: int = 100,
        health_check_interval: int = 30
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.pool_size = pool_size
        self.connections: List[ConnectionNode] = []
        self.connection_lock = threading.Lock()
        self.request_queue = queue.Queue()
        self.active_requests = 0
        self.max_concurrent = 200
        
        # Khởi tạo pool
        self._init_pool()
        
        # Health check thread
        self.health_check_interval = health_check_interval
        self._start_health_check()
        
    def _init_pool(self):
        """Khởi tạo các connection nodes"""
        for i in range(self.pool_size):
            self.connections.append(
                ConnectionNode(connection_id=f"conn_{i}")
            )
    
    def _get_best_connection(self) -> ConnectionNode:
        """Chọn connection tốt nhất dựa trên latency và availability"""
        available_conns = [
            c for c in self.connections 
            if c.is_available
        ]
        
        if not available_conns:
            # Fallback: chờ đợi connection释放
            time.sleep(0.1)
            return self._get_best_connection()
        
        # Ưu tiên connection có latency thấp nhất
        return min(available_conns, key=lambda x: x.avg_latency)
    
    def _acquire_connection(self) -> ConnectionNode:
        """Acquire một connection từ pool"""
        with self.connection_lock:
            conn = self._get_best_connection()
            conn.is_available = False
            conn.last_used = time.time()
            conn.request_count += 1
            self.active_requests += 1
            return conn
    
    def _release_connection(self, conn: ConnectionNode, latency: float):
        """Release connection về pool"""
        with self.connection_lock:
            conn.is_available = True
            # Cập nhật latency trung bình
            conn.avg_latency = (
                (conn.avg_latency * (conn.request_count - 1) + latency) 
                / conn.request_count
            )
            self.active_requests -= 1
    
    def _start_health_check(self):
        """Background health check"""
        def health_check():
            while True:
                time.sleep(self.health_check_interval)
                self._perform_health_check()
        
        thread = threading.Thread(target=health_check, daemon=True)
        thread.start()
    
    def _perform_health_check(self):
        """Kiểm tra sức khỏe các connections"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        for conn in self.connections:
            start = time.time()
            try:
                response = requests.post(
                    f"{self.base_url}/chat/completions",
                    json={
                        "model": "gpt-4.1",
                        "messages": [{"role": "user", "content": "ping"}],
                        "max_tokens": 1
                    },
                    headers=headers,
                    timeout=5
                )
                
                if response.status_code == 200:
                    conn.avg_latency = (time.time() - start) * 1000
                    
            except Exception as e:
                # Đánh dấu connection có vấn đề
                conn.avg_latency = 99999
    
    def send_request(
        self,
        messages: List[Dict],
        model: str = "gpt-4.1"
    ) -> dict:
        """Gửi request qua pool"""
        
        if self.active_requests >= self.max_concurrent:
            raise Exception(f"Max concurrent ({self.max_concurrent}) exceeded")
        
        conn = self._acquire_connection()
        start_time = time.time()
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                json={"model": model, "messages": messages},
                headers=headers,
                timeout=60
            )
            
            latency = (time.time() - start_time) * 1000
            
            if response.status_code == 200:
                result = response.json()
                result["_connection_latency"] = latency
                result["_connection_id"] = conn.connection_id
                return result
                
            elif response.status_code == 429:
                raise Exception("Rate limit exceeded")
                
            else:
                raise Exception(f"HTTP {response.status_code}: {response.text}")
                
        finally:
            self._release_connection(conn, (time.time() - start_time) * 1000)
    
    def batch_send(self, requests_data: List[dict]) -> List[dict]:
        """Gửi batch requests với thread pool"""
        results = []
        
        with ThreadPoolExecutor(max_workers=50) as executor:
            futures = []
            for data in requests_data:
                future = executor.submit(
                    self.send_request,
                    data["messages"],
                    data.get("model", "gpt-4.1")
                )
                futures.append(future)
            
            for future in futures:
                try:
                    results.append(future.result(timeout=120))
                except Exception as e:
                    results.append({"error": str(e)})
        
        return results
    
    def get_pool_status(self) -> dict:
        """Lấy trạng thái pool"""
        with self.connection_lock:
            available = sum(1 for c in self.connections if c.is_available)
            avg_latency = sum(c.avg_latency for c in self.connections) / len(self.connections)
            
            return {
                "total_connections": len(self.connections),
                "available": available,
                "in_use": len(self.connections) - available,
                "active_requests": self.active_requests,
                "avg_latency_ms": round(avg_latency, 2)
            }

Sử dụng
if __name__ == "__main__":
    pool = SmartConnectionPool(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        pool_size=100
    )
    
    # Single request
    result = pool.send_request(
        messages=[{"role": "user", "content": "Xin chào"}]
    )
    print(f"Response: {result['choices'][0]['message']['content']}")
    print(f"Pool status: {pool.get_pool_status()}")
    
    # Batch requests
    batch_data = [
        {"messages": [{"role": "user", "content": f"Câu {i}"}]}
        for i in range(50)
    ]
    batch_results = pool.batch_send(batch_data)
    print(f"Batch completed: {len(batch_results)} requests")

Giải Pháp 3: Horizontal Scaling với Multiple API Keys

Đây là chiến lược tôi áp dụng khi cần scale lên 10,000+ requests/giây. Kỹ thuật này sử dụng nhiều API keys để nhân đôi capacity.

import asyncio
import aiohttp
import hashlib
from typing import List, Dict
from collections import defaultdict
import time

class MultiKeyLoadBalancer:
    """Load balancer sử dụng nhiều API keys để tăng throughput"""
    
    def __init__(
        self,
        api_keys: List[str],
        base_url: str = "https://api.holysheep.ai/v1",
        keys_per_endpoint: int = 3
    ):
        self.api_keys = api_keys
        self.base_url = base_url
        self.keys_per_endpoint = keys_per_endpoint
        
        # Track usage per key
        self.key_usage = defaultdict(int)
        self.key_locks = {key: asyncio.Lock() for key in api_keys}
        
        # Rate tracking
        self.key_timestamps = defaultdict(list)
        
    def _select_key_for_request(self) -> str:
        """Chọn key có ít request nhất trong thời gian gần đây"""
        current_time = time.time()
        now = time.time()
        
        # Clean old timestamps (giữ chỉ 60 giây gần nhất)
        for key in self.api_keys:
            self.key_timestamps[key] = [
                t for t in self.key_timestamps[key]
                if now - t < 60
            ]
        
        # Chọn key có ít request nhất
        min_usage = float('inf')
        selected_key = self.api_keys[0]
        
        for key in self.api_keys:
            count = len(self.key_timestamps[key])
            if count < min_usage:
                min_usage = count
                selected_key = key
        
        return selected_key
    
    async def _make_request(
        self,
        session: aiohttp.ClientSession,
        endpoint: str,
        payload: dict,
        api_key: str
    ) -> dict:
        """Thực hiện request với key cụ thể"""
        
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        
        async with session.post(
            f"{self.base_url}{endpoint}",
            json=payload,
            headers=headers,
            timeout=aiohttp.ClientTimeout(total=60)
        ) as response:
            # Ghi lại timestamp
            self.key_timestamps[api_key].append(time.time())
            
            if response.status == 200:
                return await response.json()
            elif response.status == 429:
                raise Exception(f"Rate limit on key: {api_key[:8]}...")
            else:
                text = await response.text()
                raise Exception(f"Error {response.status}: {text}")
    
    async def chat_completion(
        self,
        messages: List[Dict],
        model: str = "gpt-4.1"
    ) -> dict:
        """Gửi chat completion request với load balancing"""
        
        api_key = self._select_key_for_request()
        
        connector = aiohttp.TCPConnector(limit=100)
        async with aiohttp.ClientSession(connector=connector) as session:
            return await self._make_request(
                session,
                "/chat/completions",
                {"model": model, "messages": messages},
                api_key
            )
    
    async def batch_chat_completions(
        self,
        requests: List[Dict],
        model: str = "gpt-4.1"
    ) -> List[dict]:
        """Xử lý batch với auto-scaling across keys"""
        
        async def process_single(req_id: int, messages: List[Dict]):
            api_key = self._select_key_for_request()
            
            connector = aiohttp.TCPConnector(limit=50)
            async with aiohttp.ClientSession(connector=connector) as session:
                try:
                    result = await self._make_request(
                        session,
                        "/chat/completions",
                        {"model": model, "messages": messages},
                        api_key
                    )
                    return {"id": req_id, "result": result}
                except Exception as e:
                    return {"id": req_id, "error": str(e)}
        
        # Xử lý đồng thời với concurrency limit
        semaphore = asyncio.Semaphore(len(self.api_keys) * 10)
        
        async def bounded_process(req_id: int, messages: List[Dict]):
            async with semaphore:
                return await process_single(req_id, messages)
        
        tasks = [
            bounded_process(i, req["messages"])
            for i, req in enumerate(requests)
        ]
        
        results = await asyncio.gather(*tasks)
        
        # Sort theo id
        results.sort(key=lambda x: x["id"])
        return results
    
    def get_usage_stats(self) -> dict:
        """Lấy thống kê sử dụng các keys"""
        stats = {}
        now = time.time()
        
        for key in self.api_keys:
            recent = [t for t in self.key_timestamps[key] if now - t < 60]
            stats[key[:12] + "..."] = {
                "requests_last_60s": len(recent),
                "total_tracked": self.key_usage[key]
            }
        
        return stats

Ví dụ sử dụng nhiều keys
async def main():
    # Khởi tạo với 5 API keys
    api_keys = [
        "YOUR_HOLYSHEEP_API_KEY_1",
        "YOUR_HOLYSHEEP_API_KEY_2",
        "YOUR_HOLYSHEEP_API_KEY_3",
        "YOUR_HOLYSHEEP_API_KEY_4",
        "YOUR_HOLYSHEEP_API_KEY_5"
    ]
    
    # Chỉ dùng 1 key trong demo, production dùng nhiều
    balancer = MultiKeyLoadBalancer(
        api_keys=[api_keys[0]],  # Demo với 1 key
        keys_per_endpoint=3
    )
    
    # Single request
    result = await balancer.chat_completion([
        {"role": "user", "content": "Viết code Python để gọi API"}
    ])
    print(f"Response: {result['choices'][0]['message']['content'][:100]}...")
    
    # Batch với 500 requests
    batch_requests = [
        {"messages": [{"role": "user", "content": f"Yêu cầu {i}"}]}
        for i in range(500)
    ]
    
    start = time.time()
    results = await balancer.batch_chat_completions(batch_requests)
    elapsed = time.time() - start
    
    successful = sum(1 for r in results if "result" in r)
    print(f"Hoàn thành: {successful}/500 trong {elapsed:.2f}s")
    print(f"Throughput: {500/elapsed:.1f} requests/s")

if __name__ == "__main__":
    asyncio.run(main())

Giải Pháp 4: Caching Layer với Redis

Một cách hiệu quả để giảm tải API là implement caching thông minh. Với các câu hỏi trùng lặp hoặc tương tự, cache giúp tiết kiệm 30-70% requests.

import redis
import hashlib
import json
import time
from typing import Optional, Dict, Any

class SemanticCache:
    """Semantic caching với Redis cho AI API responses"""
    
    def __init__(
        self,
        redis_host: str = "localhost",
        redis_port: int = 6379,
        ttl: int = 3600,
        similarity_threshold: float = 0.95
    ):
        self.redis_client = redis.Redis(
            host=redis_host,
            port=redis_port,
            decode_responses=True
        )
        self.ttl = ttl
        self.similarity_threshold = similarity_threshold
        self.cache_hits = 0
        self.cache_misses = 0
    
    def _generate_cache_key(self, messages: list, model: str) -> str:
        """Tạo cache key từ messages"""
        # Chỉ hash phần nội dung
        content_str = json.dumps(
            [m.get("content", "") for m in messages],
            sort_keys=True
        )
        hash_obj = hashlib.sha256(content_str.encode())
        return f"ai_cache:{model}:{hash_obj.hexdigest()[:16]}"
    
    def _get_embedding_key(self, content: str) -> str:
        """Tạo key cho embedding để so sánh similarity"""
        return f"embedding:{hashlib.md5(content.encode()).hexdigest()}"
    
    def get(self, messages: list, model: str) -> Optional[dict]:
        """Lấy response từ cache"""
        cache_key = self._generate_cache_key(messages, model)
        
        cached = self.redis_client.get(cache_key)
        if cached:
            self.cache_hits += 1
            return json.loads(cached)
        
        # Kiểm tra similar content
        primary_content = messages[-1].get("content", "")
        similar_key = self._get_embedding_key(primary_content)
        similar = self.redis_client.get(similar_key)
        
        if similar:
            data = json.loads(similar)
            # Trong production, dùng vector similarity ở đây
            self.cache_hits += 1
            return data
        
        self.cache_misses += 1
        return None
    
    def set(
        self,
        messages: list,
        model: str,
        response: dict,
        embedding: Optional[list] = None
    ):
        """Lưu response vào cache"""
        cache_key = self._generate_cache_key(messages, model)
        
        # Lưu response
        self.redis_client.setex(
            cache_key,
            self.ttl,
            json.dumps(response)
        )
        
        # Lưu embedding nếu có (cho similarity search)
        if embedding:
            primary_content = messages[-1].get("content", "")
            embedding_key = self._get_embedding_key(primary_content)
            self.redis_client.setex(
                embedding_key,
                self.ttl,
                json.dumps(response)
            )
    
    def get_stats(self) -> dict:
        """Lấy thống kê cache"""
        total = self.cache_hits + self.cache_misses
        hit_rate = (self.cache_hits / total * 100) if total > 0 else 0
        
        return {
            "hits": self.cache_hits,
            "misses": self.cache_misses,
            "hit_rate": f"{hit_rate:.1f}%",
            "memory_used": self.redis_client.info("memory")["used_memory_human"]
        }

Integration với request queue
class CachedRequestQueue(IntelligentRequestQueue):
    """Request queue với semantic caching"""
    
    def __init__(self, *args, use_cache: bool = True, **kwargs):
        super().__init__(*args, **kwargs)
        self.use_cache = use_cache
        self.cache = SemanticCache() if use_cache else None
    
    async def cached_chat_completion(
        self,
        messages: list,
        model: str = "gpt-4.1"
    ) -> dict:
        """Chat completion với cache"""
        
        if self.use_cache and self.cache:
            cached = self.cache.get(messages, model)
            if cached:
                cached["cached"] = True
                return cached
        
        # Gọi API nếu không có cache
        connector = aiohttp.TCPConnector(limit=100)
        async with aiohttp.ClientSession(connector=connector) as session:
            result = await self._make_request_with_retry(
                session,
                "/chat/completions",
                {"model": model, "messages": messages}
            )
        
        # Lưu vào cache
        if self.use_cache and self.cache:
            self.cache.set(messages, model, result)
        
        return result

Lỗi Thường Gặp và Cách Khắc Phục

# Implement exponential backoff
import time
import asyncio

async def retry_with_backoff(func, max_retries=5):
    for attempt in range(max_retries):
        try:
            return await func()
        except Exception as e:
            if "429" in str(e) and attempt < max_retries - 1:
                delay = 2 ** attempt  # 1s, 2s, 4s, 8s, 16s
                await asyncio.sleep(delay)
            else:
                raise

# Kiểm tra và xoay vòng keys
class KeyManager:
    def __init__(self, keys: list):
        self.keys = keys
        self.current_index = 0
    
    def get_current_key(self) -> str:
        return self.keys[self.current_index]
    
    def rotate_key(self):
        self.current_index = (self.current_index + 1) % len(self.keys)
        return self.get_current_key()
    
    def validate_key(self, key: str) -> bool:
        # Gọi endpoint kiểm tra
        import requests
        try:
            r = requests.get(
                "https://api.holysheep.ai/v1/models",
                headers={"Authorization": f"Bearer {key}"}
            )
            return r.status_code == 200
        except:
            return False

# Tăng pool size và implement connection limiting
import asyncio

class ConnectionLimiter:
    def __init__(self, max_connections: int = 100):
        self.semaphore = asyncio.Semaphore(max_connections)
        self.active = 0
    
    async def __aenter__(self):
        await self.semaphore.acquire()
        self.active += 1
        return self
    
    async def __aexit__(self, *args):
        self.active -= 1
        self.semaphore.release()

Sử dụng
async def limited_request():
    async with ConnectionLimiter(max_connections=50):
        # Thực hiện request ở đây
        await make_api_call()

# Tăng timeout và implement fallback
import requests

def robust_request_with_fallback(messages):
    # Thử với timeout dài hơn
    try:
        response = requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            json={"model": "gpt-4.1", "messages": messages},
            headers={"Authorization": f"Bearer {YOUR_HOLYSHEEP_API_KEY}"},
            timeout=120  # Tăng lên 120s
        )
        return response.json()
    except requests.Timeout:
        # Fallback sang model nhanh hơn
        response = requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            json={"model": "deepseek-v3.2", "messages": messages},
            headers={"Authorization": f"Bearer {YOUR_HOLYSHEEP_API_KEY}"},
            timeout=30
        )
        return response.json()

# Implement automatic truncation
def truncate_messages(messages, max_tokens=3000):
    total_tokens = 0
    truncated = []
    
    # Duyệt từ cuối lên đầu
    for msg in reversed(messages):
        msg_tokens = len(msg["content"].split()) * 1.3  # Ước tính
        if total_tokens + msg_tokens <= max_tokens:
            truncated.insert(0, msg)
            total_tokens += msg_tokens
        else:
            break
    
    return truncated

Sử dụng
messages = truncate_messages(messages, max_tokens=6000)

So Sánh Các Nhà Cung Cấp AI API