Ograniczanie liczby żądań i logika ponawiania
Wykładniczy backoff, obsługa kodu 429 i odpowiedzialne korzystanie z API.
Ograniczanie liczby żądań i logika ponawiania to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 4 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.
Czym jest ograniczanie liczby żądań
Ograniczanie liczby żądań to mechanizm ochrony interfejsów API przed przeciążeniem. Gdy agent wysyła zbyt wiele żądań w zbyt krótkim czasie, interfejs API zwraca odpowiedź 429 Too Many Requests. Typowe limity określają liczbę żądań na sekundę, minutę lub dzień.
Ignorowanie limitów prowadzi do blokowania agentów, unieważniania kluczy API i dodatkowych opłat.
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}')Nagłówek Retry-After
Gdy interfejs API zwraca odpowiedź 429, często zawiera ona nagłówek Retry-After, który informuje dokładnie, ile sekund należy odczekać przed ponowieniem żądania. Zawsze należy respektować ten nagłówek — zignorowanie go i natychmiastowe ponowienie żądania spowoduje tylko kolejną odpowiedź 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()Wykładniczy backoff
Wykładniczy backoff to standardowa strategia ponawiania: po każdej nieudanej próbie należy odczekać dłużej. Jeśli po pierwszej próbie odczekuje się 2 sekundy, po drugiej 4, a po trzeciej 8 itd., stopniowo zmniejsza to obciążenie serwera i daje mu czas na odzyskanie sprawności.
Wzór: 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')Dodawanie jittera do backoffu
Jeśli wiele agentów ponawia żądania w tym samym czasie (co często zdarza się po krótkiej awarii), wszystkie budzą się jednocześnie — tworząc efekt thundering herd, który natychmiast ponownie doprowadza do ograniczenia liczby żądań. Dodanie jittera (losowego opóźnienia) rozprasza ponowienia w czasie i zmniejsza obciążenie serwera.
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')Biblioteka tenacity
tenacity to najpopularniejsza biblioteka Pythona do obsługi logiki ponawiania. Za pomocą przejrzystej składni dekoratorów obsługuje wykładniczy backoff, jitter, maksymalną liczbę ponowień i niestandardowe warunki zatrzymania. Jest znacznie bardziej niezawodna niż ręcznie tworzone pętle ponowień.
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()tenacity z niestandardowym warunkiem ponawiania
Bibliotekę tenacity można skonfigurować tak, aby ponawiała żądania tylko dla określonych kodów statusu (takich jak 429 i 5xx) i natychmiast zatrzymywała się w przypadku błędów klienta (4xx), dla których ponawianie nie przyniesie korzyści. Należy użyć retry_if_result albo własnego obiektu wywoływalnego do sprawdzania odpowiedzi.
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()Proaktywne zarządzanie limitami żądań
Najlepszą strategią jest przede wszystkim nie dopuszczać do osiągania limitów. Należy sprawdzać nagłówki limitów przy każdej odpowiedzi i zwalniać, gdy agent zbliża się do limitu. Wiele interfejsów API zawiera nagłówki X-RateLimit-Remaining i 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()Maksymalna liczba ponowień i rezygnacja
Logika ponawiania musi zawsze mieć ograniczenie. Bezterminowe ponawianie może powodować awarie kaskadowe, w których wszystkie agenty utkną w pętlach ponowień. Po osiągnięciu wartości max_retries należy zgłosić końcowy wyjątek zawierający kontekst tego, co się nie powiodło, aby agent mógł zarejestrować problem i przejść do innych zadań.
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)Wzorzec circuit breaker
Wzorzec circuit breaker zapobiega przeciążaniu przez agenta niesprawnej usługi. Po przekroczeniu progu błędów obwód zostaje „otwarty” i wszystkie żądania kończą się natychmiastową porażką bez nawiązywania połączenia z siecią. Po okresie schładzania wzorzec próbuje wykonać jedno żądanie — jeśli się powiedzie, obwód zostaje zamknięty i zostaje wznowiona normalna praca.
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}')
Kolejkowanie żądań w celu przestrzegania limitów
W przypadku agentów wykonujących wiele wywołań w ramach jednej serii należy użyć mechanizmu token bucket albo prostego ogranicznika opartego na usypianiu, aby mieścić się w limitach. Należy obliczyć bezpieczny odstęp między wywołaniami na podstawie limitu interfejsu API (np. 60 wywołań na minutę = 1 wywołanie na sekundę).
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Łączenie logiki ponawiania z nagłówkami backoffu
Najbardziej niezawodny wzorzec łączy określone przez serwer czasy oczekiwania (Retry-After) z wykładniczym backoffem stosowanym jako rozwiązanie awaryjne. Gdy wskazówki serwera są dostępne, zawsze należy dawać im pierwszeństwo — serwer dokładnie wie, kiedy można ponowić żądanie.
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')Szybkie sprawdzenie: wykładniczy backoff
Sprawdzenie zrozumienia strategii ponawiania.
Podsumowanie ograniczania liczby żądań i ponawiania
Agenci potrafią teraz poprawnie obsługiwać ograniczenia liczby żądań:
- 429 Too Many Requests — należy respektować nagłówek
Retry-Afteri odczekać przed ponowieniem żądania - Wykładniczy backoff —
wait = 2^attemptpodwaja czas oczekiwania przy każdym ponowieniu - Jitter — dodaje losowość, aby rozłożyć ponowienia między wiele instancji agentów
- tenacity — obsługuje całą logikę ponawiania za pomocą dekoratorów i przejrzystej konfiguracji
- Circuit breaker — po przekroczeniu progu przestaje przeciążać niesprawną usługę
- Proaktywne ograniczanie tempa — należy sprawdzać
X-RateLimit-Remainingi zwalniać przed osiągnięciem limitu
Często zadawane pytania
Czy lekcja „Ograniczanie liczby żądań i logika ponawiania” jest bezpłatna?
Tak — pełny tekst „Ograniczanie liczby żądań i logika ponawiania” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.
Co nauczysz się w „Ograniczanie liczby żądań i logika ponawiania”?
Wykładniczy backoff, obsługa kodu 429 i odpowiedzialne korzystanie z API. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.
Czy potrzebuję doświadczenia, aby zacząć AI Agents?
Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 4 z 4.
Ile czasu zajmuje lekcja „Ograniczanie liczby żądań i logika ponawiania”?
Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.
Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?
Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.
Wszystkie lekcje w tym kursie
- Podstawy REST API dla twórców agentów
- Uwierzytelnianie: klucze API i OAuth
- Obsługa odpowiedzi i błędów API
- Ograniczanie liczby żądań i logika ponawiania