AI Engineering Academy · Урок

Балансировка нагрузки и стратегии с несколькими ключами

Урок 2 из 413 шагов

«Балансировка нагрузки и стратегии с несколькими ключами» — бесплатный урок AI Engineering Academy на CoddyKit. Это урок 2 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Engineering Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Engineering Academy содержит 4 уроков всего.

Почему одного ключа API недостаточно

Один ключ API OpenAI имеет фиксированные ограничения частоты, измеряемые в запросах в минуту (RPM) и токенах в минуту (TPM). На уровне 1 GPT-4o допускает 500 RPM и 30 000 TPM. Для рабочего приложения с сотнями одновременно работающих пользователей один ключ будет постоянно достигать этих ограничений. Несколько ключей API пропорционально увеличивают доступный запас производительности.

Создание нескольких ключей API

Вы можете создать несколько ключей API в одной организации OpenAI или создать несколько аккаунтов OpenAI (с отдельной оплатой каждого). Храните каждый ключ в конфигурации окружения и рассматривайте их как пул. Храните ключи в менеджере секретов, например AWS Secrets Manager или HashiCorp Vault, а не в исходном коде или файлах .env, добавленных в систему контроля версий.

import os

API_KEYS = [
    os.environ['OPENAI_KEY_1'],
    os.environ['OPENAI_KEY_2'],
    os.environ['OPENAI_KEY_3'],
    os.environ['OPENAI_KEY_4'],
]

# Total effective RPM = 500 * 4 = 2000 RPM
# Total effective TPM = 30000 * 4 = 120000 TPM

Балансировка нагрузки по циклическому перебору

Циклический перебор равномерно распределяет запросы между всеми ключами, последовательно переходя от одного к другому. Этот подход просто реализовать; со временем каждый ключ обрабатывает примерно одинаковую нагрузку. Используйте потокобезопасный счётчик или атомарное целое число, чтобы два параллельных запроса одновременно не выбрали один и тот же ключ. Циклический перебор хорошо работает, когда у всех ключей одинаковые ограничения частоты.

import itertools
import threading
from openai import OpenAI

class RoundRobinPool:
    def __init__(self, keys: list):
        self._clients = [OpenAI(api_key=k) for k in keys]
        self._cycle = itertools.cycle(range(len(self._clients)))
        self._lock = threading.Lock()

    def get_client(self) -> OpenAI:
        with self._lock:
            idx = next(self._cycle)
        return self._clients[idx]

pool = RoundRobinPool(API_KEYS)
client = pool.get_client()

Взвешенная балансировка нагрузки

Взвешенная балансировка нагрузки назначает ключам более высокого уровня (с более высокими ограничениями частоты) большую долю трафика пропорционально их пропускной способности. Если ключ A относится к уровню 3 (10 000 RPM), а ключ B — к уровню 1 (500 RPM), ключ A должен получать около 95% запросов. Взвешенная балансировка не позволяет ключам низкого уровня становиться узким местом при совместном использовании с ключами более высокого уровня.

import random

class WeightedPool:
    def __init__(self, key_configs: list):
        # key_configs = [{'key': '...', 'weight': 10}, ...]
        self._clients = [OpenAI(api_key=c['key']) for c in key_configs]
        self._weights = [c['weight'] for c in key_configs]

    def get_client(self) -> OpenAI:
        return random.choices(self._clients, weights=self._weights, k=1)[0]

pool = WeightedPool([
    {'key': os.environ['OPENAI_KEY_TIER3'], 'weight': 20},
    {'key': os.environ['OPENAI_KEY_TIER1'], 'weight': 1},
])

Отслеживание состояния ограничений для каждого ключа

API OpenAI возвращает в каждом ответе заголовки с ограничениями частоты: x-ratelimit-remaining-requests и x-ratelimit-remaining-tokens. Отслеживайте эти заголовки для каждого ключа, чтобы знать, какие ключи близки к исчерпанию. Когда ключ сообщает, что в текущей минуте осталось менее 10 запросов, временно направляйте трафик от него, чтобы предотвратить ошибки 429 до их возникновения.

class SmartPool:
    def __init__(self, keys: list):
        self._clients = [OpenAI(api_key=k) for k in keys]
        self._remaining = {i: 500 for i in range(len(keys))}  # initial RPM

    def get_best_client(self):
        # Pick key with most remaining capacity
        best_idx = max(self._remaining, key=lambda i: self._remaining[i])
        return self._clients[best_idx], best_idx

    def update_remaining(self, idx: int, response_headers: dict):
        remaining = int(response_headers.get('x-ratelimit-remaining-requests', 0))
        self._remaining[idx] = remaining

Обработка ошибок ограничения частоты 429

Когда ключ возвращает ошибку 429, немедленно исключите его из пула на срок, указанный в заголовке Retry-After (обычно 60 секунд). Пометьте его как охлаждающийся и направьте весь трафик на оставшиеся ключи. После завершения периода охлаждения верните ключ в пул. Это предотвращает каскадные сбои, при которых повторные попытки с тем же ключом усугубляют ситуацию.

import time
from openai import RateLimitError

class CooldownPool:
    def __init__(self, keys: list):
        self._clients = [(OpenAI(api_key=k), None) for k in keys]  # (client, cooldown_until)

    def get_available_clients(self):
        now = time.time()
        return [
            (i, c) for i, (c, until) in enumerate(self._clients)
            if until is None or until <= now
        ]

    def mark_cooling(self, idx: int, retry_after: int = 60):
        client, _ = self._clients[idx]
        self._clients[idx] = (client, time.time() + retry_after)
        print(f'Key {idx} cooling down for {retry_after}s')

Использование OpenRouter в качестве мультиплексора

OpenRouter — это прокси-сервис, предоставляющий сотни моделей через единую конечную точку API, совместимую с OpenAI. При маршрутизации через OpenRouter Вы автоматически получаете балансировку нагрузки между несколькими аккаунтами базовых поставщиков, переключение на альтернативных поставщиков и доступ к моделям с открытым исходным кодом в качестве резервных. Небольшая надбавка к стоимости компенсируется простотой эксплуатации.

from openai import OpenAI

# OpenRouter uses the same OpenAI SDK interface
client = OpenAI(
    api_key=os.environ['OPENROUTER_API_KEY'],
    base_url='https://openrouter.ai/api/v1'
)

response = client.chat.completions.create(
    model='openai/gpt-4o',  # OpenRouter model name format
    messages=[{'role': 'user', 'content': prompt}]
)
# Automatic failover if OpenAI is down

Мониторинг состояния ключей с помощью метрик

Отслеживайте для каждого ключа количество отправленных запросов, полученных ошибок 429 и время охлаждения за последний час. Ключ с высокой долей ошибок 429 требует либо снижения трафика, либо повышения уровня. Предоставьте эти метрики в конечной точке /metrics в формате Prometheus, чтобы система мониторинга могла предупредить Вас, когда какой-либо ключ постоянно достигает ограничений.

from dataclasses import dataclass, field
from collections import defaultdict

@dataclass
class KeyMetrics:
    requests_sent: int = 0
    rate_limit_errors: int = 0
    total_tokens_used: int = 0
    cooldown_count: int = 0

class MetricPool:
    def __init__(self, keys: list):
        self._clients = [OpenAI(api_key=k) for k in keys]
        self._metrics = [KeyMetrics() for _ in keys]

    def report(self):
        for i, m in enumerate(self._metrics):
            error_rate = m.rate_limit_errors / max(m.requests_sent, 1)
            print(f'Key {i}: {m.requests_sent} req, {error_rate:.1%} 429 rate')

Распределение ключей по регионам

Если пользователи находятся по всему миру, рассмотрите возможность хранения отдельных ключей API для каждого географического региона и маршрутизации запросов к ключу, ближайшему к пользователю. Уменьшение времени сетевого обмена туда и обратно улучшает TTFT. Разместите лёгкий балансировщик нагрузки в каждом регионе (AWS Lambda@Edge или Cloudflare Worker), который выбирает подходящий ключ и проксирует запрос, скрывая ключи от клиентов.

REGIONAL_KEYS = {
    'us-east': os.environ['OPENAI_KEY_US_EAST'],
    'eu-west': os.environ['OPENAI_KEY_EU_WEST'],
    'ap-southeast': os.environ['OPENAI_KEY_AP'],
}

def get_key_for_region(user_region: str) -> str:
    # Default to us-east if region unknown
    return REGIONAL_KEYS.get(user_region, REGIONAL_KEYS['us-east'])

Тестирование балансировщика нагрузки

Напишите нагрузочный тест, который отправляет через пул балансировки 100 параллельных запросов и измеряет распределение, частоту ошибок и процентили задержки. Убедитесь, что ни один ключ не обрабатывает больше своей пропорциональной доли, а частота ошибок 429 ниже 0,1%. Используйте asyncio.gather или инструмент вроде Locust, чтобы смоделировать параллельную нагрузку, с которой в действительности столкнётся рабочая система.

import asyncio
import time

async def load_test(pool, concurrency=100, total=1000):
    sem = asyncio.Semaphore(concurrency)
    results = []

    async def one_request():
        async with sem:
            client = pool.get_client()
            start = time.perf_counter()
            try:
                await client.chat.completions.create(
                    model='gpt-4o-mini',
                    messages=[{'role': 'user', 'content': 'Ping'}],
                    max_tokens=5
                )
                results.append(('ok', time.perf_counter() - start))
            except Exception as e:
                results.append(('error', str(e)))

    await asyncio.gather(*[one_request() for _ in range(total)])
    ok = [r for r in results if r[0] == 'ok']
    print(f'Success rate: {len(ok)/total:.1%}')
    return results

Выбор подходящей стратегии балансировки

Выбирайте стратегию балансировки в соответствии со структурой ограничений частоты. Используйте циклический перебор, когда у всех ключей одинаковые ограничения уровня и трафик распределён равномерно. Используйте взвешенную балансировку, когда ограничения уровня у ключей различаются. Используйте маршрутизацию с учётом состояния (пропуская ключи, близкие к исчерпанию), когда нужно минимизировать ошибки 429 при всплесках трафика. Для большинства рабочих систем наилучший баланс простоты и устойчивости даёт маршрутизация с учётом состояния в сочетании с экспоненциальной задержкой между повторными попытками.

# Strategy selection guide:
# Scenario A: 4 keys all Tier 2 (same limits)
#   -> Round-robin: simple, even distribution
#
# Scenario B: 1 Tier 3 key + 3 Tier 1 keys
#   -> Weighted: Tier 3 gets 10x weight
#
# Scenario C: Variable traffic with burst periods
#   -> Health-aware: track remaining headers, skip near-limit keys
#
# Scenario D: Multi-region, latency-sensitive
#   -> Geographic: regional keys, route by user location

Быстрая проверка

Проверьте своё понимание стратегий балансировки нагрузки для API LLM.

Ответь в интерактивном уроке

Итоги урока

В этом уроке Вы узнали: циклическая и взвешенная балансировка распределяют трафик между несколькими ключами API, увеличивая доступный запас в рамках ограничений частоты; отслеживание периода охлаждения предотвращает каскадные ошибки 429, временно удаляя ключи, для которых применено ограничение; а OpenRouter предоставляет управляемый вариант мультиплексирования с автоматическим переключением. Далее мы реализуем резервных поставщиков и автоматические размыкатели цепи.

Можно начать бесплатно

Изучай Python с ИИ-репетитором — бесплатно

Пиши и запускай код прямо в браузере, получай мгновенную помощь от ИИ-репетитора 24/7 и продолжи учиться на сайте или в приложении.

Курсы: 30
Уроки: 120

Начать учиться бесплатно Посмотреть полный курс

Часто задаваемые вопросы

Урок «Балансировка нагрузки и стратегии с несколькими ключами» бесплатный?

Да — полный текст урока «Балансировка нагрузки и стратегии с несколькими ключами» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Engineering Academy, подпишись на CoddyKit PRO. Курс AI Engineering Academy содержит 4 уроков всего.

Чему я научусь в уроке «Балансировка нагрузки и стратегии с несколькими ключами»?

Реализуйте балансировку нагрузки по кругу и взвешенную балансировку между несколькими ключами API и учётными записями, чтобы увеличить запас по ограничению частоты запросов и уменьшить скачки задержк… Ты практикуешь AI Engineering Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать AI Engineering Academy?

Предыдущий опыт не требуется. AI Engineering Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 2 из 4.

Сколько времени занимает урок «Балансировка нагрузки и стратегии с несколькими ключами»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке AI Engineering Academy?

Да. Каждый урок AI Engineering Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

Измерение задержки LLM: TTFT и TPOT
Балансировка нагрузки и стратегии с несколькими ключами
Резервные поставщики и автоматические размыкатели
Бюджеты времени ожидания и плавное ухудшение качества

← Назад к AI Engineering Academy