Rate limiting og retry-logik
Eksponentiel backoff, håndtering af 429 og hensynsfuld brug af API'er.
Rate limiting og retry-logik er en gratis AI-agenter-lektion på CoddyKit. Dette er lektion 4 af 4. Du kan læse hele lektionen gratis nedenfor — og derefter øve dig praktisk i browseren med en indbygget kodeeditor og en AI-vejleder, der er tilgængelig døgnet rundt. Den er en del af læringsforløbet i AI-agenter, og dine fremskridt synkroniseres på tværs af nettet og CoddyKit-appen. AI-agenter-kurset indeholder 4 lektioner i alt.
Hvad er hastighedsbegrænsning?
Hastighedsbegrænsning er den måde, API'er beskytter sig selv mod overbelastning på. Når din agent sender for mange anmodninger for hurtigt, returnerer API'en 429 Too Many Requests. Almindelige grænser er et antal anmodninger pr. sekund, minut eller dag.
Hvis du ignorerer hastighedsbegrænsninger, kan det føre til blokerede agenter, tilbagekaldte API-nøgler og ekstra omkostninger.
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}')Headeren Retry-After
Når en API returnerer 429, indeholder svaret ofte en Retry-After-header, der fortæller præcis, hvor mange sekunder du skal vente, før du forsøger igen. Respektér altid denne header — hvis du ignorerer den og forsøger igen med det samme, får du blot endnu et 429-svar.
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()Eksponentiel ventetid
Eksponentiel ventetid er standardstrategien for nye forsøg: Vent længere efter hvert mislykket forsøg. Hvis forsøg 1 venter 2 sekunder, venter forsøg 2 4, forsøg 3 8 osv. Det reducerer gradvist belastningen på serveren og giver den tid til at komme sig.
Formel: 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')Tilføjelse af variation til ventetiden
Hvis mange agenter forsøger igen på samme tid (et almindeligt scenarie efter et kortvarigt nedbrud), vågner de alle samtidig — og skaber en overvældende flok, der straks udløser hastighedsbegrænsning igen. Hvis du tilføjer tilfældig variation (en tilfældig forsinkelse), spredes de nye forsøg, og serverbelastningen reduceres.
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')tenacity-biblioteket
tenacity er det mest populære Python-bibliotek til logik for nye forsøg. Det håndterer eksponentiel ventetid, tilfældig variation, maksimalt antal forsøg og brugerdefinerede stopbetingelser med en enkel dekoratorsyntaks. Det er langt mere pålideligt end egenbyggede løkker til nye forsøg.
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 med brugerdefineret betingelse for nye forsøg
Du kan lære tenacity kun at forsøge igen ved bestemte statuskoder (f.eks. 429 og 5xx) og straks stoppe ved klientfejl (4xx), som ikke bliver bedre af et nyt forsøg. Brug retry_if_result eller en brugerdefineret funktion til at undersøge responsen.
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()Proaktiv håndtering af hastighedsbegrænsning
Den bedste strategi er at undgå overhovedet at ramme hastighedsbegrænsninger. Kontrollér headerne for hastighedsbegrænsning i hvert svar, og sæt farten ned, når du nærmer dig grænsen. Mange API'er indeholder headerne X-RateLimit-Remaining og 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()Maksimalt antal nye forsøg og opgivelse
Logik for nye forsøg skal altid have en grænse. Hvis du forsøger igen for evigt, kan det skabe kaskadefejl, hvor alle dine agenter sidder fast i løkker med nye forsøg. Efter max_retries skal du udløse en endelig undtagelse med kontekst om, hvad der mislykkedes, så agenten kan logge det og gå videre til andre opgaver.
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)Mønstret med kredsløbsafbryder
Mønstret med kredsløbsafbryder forhindrer din agent i at overbelaste en tjeneste, der fejler. Efter en bestemt tærskel af fejl "åbnes" kredsløbet, og alle anmodninger mislykkes straks uden at ramme netværket. Efter en afkølingsperiode forsøger det med én anmodning — hvis den lykkes, lukkes kredsløbet, og normal drift genoptages.
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}')
Kølægning af anmodninger for at holde sig inden for grænserne
Brug en token-bøtte eller en simpel ventebaseret hastighedsregulering til agenter, der foretager mange kald i en batch, så de holder sig inden for grænserne. Beregn det sikre interval mellem kald baseret på API'ens hastighedsgrænse (f.eks. 60 kald/minut = 1 kald i sekundet).
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 resultsKombination af logik for nye forsøg med ventetidsheadere
Det mest robuste mønster kombinerer serverangivne ventetider (Retry-After) med eksponentiel ventetid som reserve. Følg altid serverens vejledning, når den er tilgængelig — serveren ved præcis, hvornår du kan forsøge igen.
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')Hurtigt tjek: Eksponentiel ventetid
Afprøv din forståelse af strategier for nye forsøg.
Opsummering af hastighedsbegrænsning og nye forsøg
Dine agenter kan nu håndtere hastighedsbegrænsninger på en hensigtsmæssig måde:
- 429 For mange anmodninger — respekter
Retry-After-headeren, og vent, før du forsøger igen - Eksponentiel ventetid —
wait = 2^attemptfordobler ventetiden ved hvert nyt forsøg - Tilfældig variation — spreder nye forsøg tilfældigt på tværs af flere agentinstanser
- tenacity — håndterer al logik for nye forsøg med dekoratorer og enkel konfiguration
- Kredsløbsafbryder — forhindrer overbelastning af en tjeneste, der fejler, efter en bestemt tærskel
- Proaktiv hastighedsregulering — kontrollér
X-RateLimit-Remaining, og sæt farten ned, før du rammer grænsen
Lær AI-agenter med en AI-underviser — gratis
Skriv og kør rigtig kode i din browser, få øjeblikkelig hjælp fra en AI-underviser døgnet rundt, og fortsæt, hvor du slap, på web eller i appen.
- Kurser
- 60
- Lektioner
- 239
Ofte stillede spørgsmål
Er lektionen “Rate limiting og retry-logik” gratis?
Ja — hele teksten til “Rate limiting og retry-logik” kan læses gratis her på nettet. Hvis du vil øve dig interaktivt med en indbygget kodeeditor og en AI-vejleder døgnet rundt og få adgang til resten af AI-agenter-kurset, skal du opgradere til CoddyKit PRO. AI-agenter-kurset indeholder 4 lektioner i alt.
Hvad lærer jeg i “Rate limiting og retry-logik”?
Eksponentiel backoff, håndtering af 429 og hensynsfuld brug af API'er. Du øver dig i AI-agenter med praktisk kode, som du kører direkte i browseren, og en AI-vejleder døgnet rundt besvarer dine spørgsmål, mens du arbejder dig gennem lektionen.
Skal jeg have erfaring for at begynde på AI-agenter?
Der kræves ingen tidligere erfaring. AI-agenter på CoddyKit er tilrettelagt for både begyndere og øvede, så du kan starte her eller fra begyndelsen og lære i dit eget tempo. Dette er lektion 4 af 4.
Hvor lang tid tager lektionen “Rate limiting og retry-logik”?
De fleste CoddyKit-lektioner tager cirka 5–10 minutter. Hver lektion er kort og interaktiv, så du gør løbende fremskridt og kan fortsætte, hvor du slap – på både web og app.
Kan jeg skrive og køre kode i denne AI-agenter-lektion?
Ja. Alle AI-agenter-lektioner har en indbygget kodeeditor, så du kan skrive og køre rigtig kode direkte i din browser og få øjeblikkelig feedback fra AI – uden lokal opsætning.
Alle lektioner i dette kursus
- Grundlæggende REST API for agentudviklere
- Godkendelse: API-nøgler og OAuth
- Håndtering af API-responses og fejl
- Rate limiting og retry-logik