Ограничение частоты запросов и логика повторных попыток
Экспоненциальная задержка перед повтором, обработка кода 429 и корректное использование API.
«Ограничение частоты запросов и логика повторных попыток» — бесплатный урок AI Agents на CoddyKit. Это урок 4 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Agents, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Agents содержит 4 уроков всего.
Что такое ограничение частоты запросов
Ограничение частоты запросов — это способ, с помощью которого API защищают себя от перегрузки. Когда агент отправляет слишком много запросов за короткое время, API возвращает 429 Too Many Requests. Обычно ограничения задаются для числа запросов в секунду, минуту или день.
Игнорирование ограничений частоты приводит к блокировке агентов, отзыву ключей API и дополнительным расходам.
import requests
response = requests.get(
'https://api.example.com/data',
headers={'Authorization': 'Bearer YOUR_KEY'}
)
if response.status_code == 429:
print('Rate limit exceeded!')
# Check headers for limit details
limit = response.headers.get('X-RateLimit-Limit')
remaining = response.headers.get('X-RateLimit-Remaining')
reset = response.headers.get('X-RateLimit-Reset')
print(f'Limit: {limit}, Remaining: {remaining}, Reset: {reset}')Заголовок Retry-After
Возвращая ошибку 429, API часто включает заголовок Retry-After, который сообщает точное число секунд до следующей попытки. Всегда соблюдайте значение этого заголовка: если проигнорировать его и повторить запрос немедленно, Вы снова получите ошибку 429.
import requests
import time
def request_with_retry_after(url, headers):
response = requests.get(url, headers=headers)
if response.status_code == 429:
retry_after = int(response.headers.get('Retry-After', 60))
print(f'Rate limited. Waiting {retry_after} seconds...')
time.sleep(retry_after)
# Retry once after waiting
response = requests.get(url, headers=headers)
response.raise_for_status()
return response.json()Экспоненциальная задержка перед повтором
Экспоненциальная задержка перед повтором — стандартная стратегия повторных попыток: после каждой неудачной попытки нужно ждать дольше. Если после первой попытки ожидание составляет 2 секунды, после второй — 4, после третьей — 8 и так далее. Это постепенно снижает нагрузку на сервер и даёт ему время восстановиться.
Формула: wait = 2 ** attempt
import requests
import time
def get_with_exponential_backoff(url, headers, max_retries=5):
for attempt in range(max_retries):
response = requests.get(url, headers=headers, timeout=(5, 30))
if response.status_code == 200:
return response.json()
if response.status_code in (429, 500, 502, 503):
wait = 2 ** attempt # 1, 2, 4, 8, 16 seconds
print(f'Attempt {attempt+1} failed ({response.status_code}). '
f'Waiting {wait}s before retry...')
time.sleep(wait)
else:
response.raise_for_status() # non-retryable error
raise Exception(f'Failed after {max_retries} retries')Добавление случайной задержки к интервалу повторов
Если несколько агентов повторяют запрос одновременно (что часто происходит после кратковременного сбоя), они просыпаются в один момент и создают лавину одновременных запросов, из-за которой ограничение частоты срабатывает снова. Добавление случайной задержки распределяет повторные попытки во времени и снижает нагрузку на сервер.
import requests
import time
import random
def get_with_jittered_backoff(url, headers, max_retries=5):
for attempt in range(max_retries):
response = requests.get(url, headers=headers, timeout=(5, 30))
if response.status_code == 200:
return response.json()
if response.status_code in (429, 500, 502, 503):
base_wait = 2 ** attempt
# Add random jitter: actual wait is 50%-100% of base
jitter = random.uniform(0.5, 1.0)
wait = base_wait * jitter
print(f'Waiting {wait:.1f}s (attempt {attempt+1})')
time.sleep(wait)
else:
response.raise_for_status()
raise Exception(f'Failed after {max_retries} retries')Библиотека повторных попыток
Библиотека повторных попыток для Python — самый популярный инструмент для логики повторных запросов. Она обрабатывает экспоненциальные задержки, случайные задержки, максимальное число повторов и пользовательские условия остановки с помощью удобного синтаксиса декораторов. Это намного надёжнее самописных циклов повторных попыток.
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_exception_type, before_sleep_log
)
import requests
import logging
logger = logging.getLogger(__name__)
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=2, max=60),
retry=retry_if_exception_type(requests.exceptions.HTTPError),
before_sleep=before_sleep_log(logger, logging.WARNING)
)
def fetch_data(url, headers):
response = requests.get(url, headers=headers, timeout=(5, 30))
if response.status_code == 429:
response.raise_for_status() # triggers retry
response.raise_for_status()
return response.json()Пользовательское условие повторных попыток
Можно настроить повторные попытки только для определённых кодов состояния (например, 429 и 5xx) и немедленно прекращать их при ошибках клиента (4xx), для которых повторы не помогут. Используйте retry_if_result или пользовательский вызываемый объект для проверки ответа.
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_result
)
import requests
def is_retryable_response(response):
return response.status_code in (429, 500, 502, 503, 504)
@retry(
stop=stop_after_attempt(4),
wait=wait_exponential(multiplier=2, min=2, max=30),
retry=retry_if_result(is_retryable_response)
)
def resilient_get(url, headers):
response = requests.get(url, headers=headers, timeout=(5, 30))
return response # retry logic inspects the response object
# Usage
response = resilient_get(
'https://api.example.com/data',
{'Authorization': 'Bearer YOUR_KEY'}
)
data = response.json()Упреждающее управление ограничением частоты
Лучшая стратегия — изначально не доводить дело до ограничения частоты. Проверяйте заголовки ограничения частоты в каждом ответе и снижайте скорость, когда приближаетесь к пределу. Многие API включают заголовки X-RateLimit-Remaining и X-RateLimit-Reset.
import requests
import time
class RateLimitAwareClient:
def __init__(self, base_url, api_key):
self.base_url = base_url
self.headers = {'Authorization': f'Bearer {api_key}'}
self.remaining = 1000 # assume generous limit
def get(self, path):
# Proactively slow down if nearly exhausted
if self.remaining < 10:
print('Rate limit nearly exhausted, sleeping 5s...')
time.sleep(5)
response = requests.get(
f'{self.base_url}{path}', headers=self.headers
)
# Update remaining from response headers
remaining_str = response.headers.get('X-RateLimit-Remaining')
if remaining_str:
self.remaining = int(remaining_str)
response.raise_for_status()
return response.json()Максимальное число повторов и отказ от дальнейших попыток
Логика повторных попыток всегда должна иметь ограничение. Бесконечные повторы могут вызвать каскадные сбои, при которых все агенты застрянут в циклах повторных попыток. После max_retries создайте итоговое исключение с описанием сбоя, чтобы агент мог записать его в журнал и перейти к другой работе.
import requests
import time
class MaxRetriesExceeded(Exception):
def __init__(self, url, attempts, last_status):
self.url = url
self.attempts = attempts
self.last_status = last_status
super().__init__(
f'Failed {url} after {attempts} attempts '
f'(last status: {last_status})'
)
def fetch_with_limit(url, headers, max_retries=3):
last_response = None
for attempt in range(max_retries):
last_response = requests.get(url, headers=headers)
if last_response.status_code == 200:
return last_response.json()
time.sleep(2 ** attempt)
raise MaxRetriesExceeded(url, max_retries, last_response.status_code)Паттерн размыкателя цепи
Паттерн размыкателя цепи не позволяет агенту непрерывно нагружать неисправный сервис. После достижения порога ошибок цепь «размыкается», и все запросы немедленно завершаются с ошибкой, не обращаясь к сети. По истечении периода охлаждения выполняется один запрос: если он успешен, цепь замыкается и обычная работа возобновляется.
import time
class CircuitBreaker:
CLOSED, OPEN, HALF_OPEN = 'closed', 'open', 'half_open'
def __init__(self, failure_threshold=5, recovery_timeout=60):
self.state = self.CLOSED
self.failures = 0
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.opened_at = None
def call(self, func, *args, **kwargs):
if self.state == self.OPEN:
if time.time() - self.opened_at > self.recovery_timeout:
self.state = self.HALF_OPEN
else:
raise Exception('Circuit OPEN — service unavailable')
try:
result = func(*args, **kwargs)
self.failures = 0
self.state = self.CLOSED
return result
except Exception as e:
self.failures += 1
if self.failures >= self.failure_threshold:
self.state = self.OPEN
self.opened_at = time.time()
print(f'Circuit OPENED after {self.failures} failures')
raise
# --- demo ---
def flaky():
raise ValueError('upstream 500')
def works():
return 'ok'
cb = CircuitBreaker(failure_threshold=3, recovery_timeout=60)
for i in range(3):
try:
cb.call(flaky)
except Exception as e:
print(f'call {i+1} failed: {e}')
print(f'Breaker state after 3 failures: {cb.state}')
try:
cb.call(flaky)
except Exception as e:
print(f'Rejected without calling flaky(): {e}')
Постановка запросов в очередь для соблюдения ограничений
Для агентов, выполняющих много вызовов пакетами, используйте ведро токенов или простое ограничение скорости с помощью пауз, чтобы не выходить за пределы. Рассчитайте безопасный интервал между вызовами на основе ограничения API (например, 60 вызовов в минуту = 1 вызов в секунду).
import requests
import time
def batch_requests(urls, headers, calls_per_minute=60):
interval = 60.0 / calls_per_minute # seconds between calls
results = []
for i, url in enumerate(urls):
start = time.time()
response = requests.get(url, headers=headers, timeout=(5, 30))
response.raise_for_status()
results.append(response.json())
print(f'Processed {i+1}/{len(urls)}')
# Sleep for remaining time in the interval
elapsed = time.time() - start
sleep_time = interval - elapsed
if sleep_time > 0:
time.sleep(sleep_time)
return resultsОбъединение логики повторных попыток с заголовками задержки
Самый надёжный подход сочетает заданное сервером время ожидания (Retry-After) с экспоненциальной задержкой в качестве запасного варианта. Если сервер предоставил указания, всегда отдавайте им предпочтение: сервер точно знает, когда можно повторить запрос.
import requests
import time
import random
def smart_retry(url, headers, max_retries=5):
for attempt in range(max_retries):
response = requests.get(url, headers=headers, timeout=(5, 30))
if response.status_code == 200:
return response.json()
if response.status_code == 429:
# Use Retry-After if provided, else exponential backoff
retry_after = response.headers.get('Retry-After')
if retry_after:
wait = int(retry_after)
else:
wait = (2 ** attempt) + random.uniform(0, 1)
print(f'429 rate limit. Waiting {wait:.1f}s...')
time.sleep(wait)
elif response.status_code >= 500:
wait = (2 ** attempt) + random.uniform(0, 1)
print(f'Server error {response.status_code}. Waiting {wait:.1f}s...')
time.sleep(wait)
else:
response.raise_for_status() # non-retryable
raise Exception(f'Gave up after {max_retries} attempts')Быстрая проверка: экспоненциальная задержка перед повтором
Проверьте, насколько хорошо Вы понимаете стратегии повторных попыток.
Итоги: ограничение частоты и повторные попытки
Теперь Ваши агенты умеют корректно обрабатывать ограничения частоты:
- 429 Too Many Requests — соблюдайте заголовок
Retry-Afterи подождите перед повторной попыткой - Экспоненциальная задержка перед повтором —
wait = 2^attemptудваивает время ожидания после каждой повторной попытки - Случайная задержка — распределяет повторные попытки между несколькими экземплярами агента
- Библиотека повторных попыток — обрабатывает всю логику повторов с помощью декораторов и понятной настройки
- Размыкатель цепи — прекращает непрерывно нагружать неисправный сервис после достижения порога ошибок
- Упреждающее ограничение скорости — проверяйте
X-RateLimit-Remainingи снижайте скорость до достижения предела
Часто задаваемые вопросы
Урок «Ограничение частоты запросов и логика повторных попыток» бесплатный?
Да — полный текст урока «Ограничение частоты запросов и логика повторных попыток» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Agents, подпишись на CoddyKit PRO. Курс AI Agents содержит 4 уроков всего.
Чему я научусь в уроке «Ограничение частоты запросов и логика повторных попыток»?
Экспоненциальная задержка перед повтором, обработка кода 429 и корректное использование API. Ты практикуешь AI Agents с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.
Нужен ли мне опыт, чтобы начать AI Agents?
Предыдущий опыт не требуется. AI Agents на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 4 из 4.
Сколько времени занимает урок «Ограничение частоты запросов и логика повторных попыток»?
Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.
Можно ли писать и запускать код в этом уроке AI Agents?
Да. Каждый урок AI Agents включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.
Все уроки этого курса
- Основы REST API для разработчиков агентов
- Аутентификация: ключи API и OAuth
- Обработка ответов API и ошибок
- Ограничение частоты запросов и логика повторных попыток