Frekvensbegrensning og logikk for nye forsøk
Eksponentiell backoff, håndtering av 429 og hensynsfull bruk av API-er.
Frekvensbegrensning og logikk for nye forsøk er en gratis leksjon i AI-agenter på CoddyKit. Dette er leksjon 4 av 4. Du kan lese hele leksjonen gratis nedenfor – og deretter øve praktisk i nettleseren med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt. Den er en del av læringsløpet i AI-agenter, og fremdriften din synkroniseres mellom nettet og CoddyKit-appen. Kurset i AI-agenter inneholder totalt 4 leksjoner.
Hva er hastighetsbegrensning?
Hastighetsbegrensning er måten API-er beskytter seg mot å bli overbelastet på. Når agenten sender for mange forespørsler for raskt, returnerer API-et 429 Too Many Requests. Vanlige grenser er antall forespørsler per sekund, minutt eller dag.
Hvis du ignorerer hastighetsgrensene, kan agenter bli blokkert, API-nøkler bli tilbakekalt og kostnadene øke.
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-headeren
Når et API returnerer 429, inneholder svaret ofte en Retry-After-header som forteller nøyaktig hvor mange sekunder du skal vente før du prøver på nytt. Respekter alltid denne headeren – hvis du ignorerer den og prøver på nytt umiddelbart, får du bare en ny 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()Eksponentiell backoff
Eksponentiell backoff er standardstrategien for nye forsøk: vent lenger etter hvert mislykkede forsøk. Hvis forsøk 1 venter i 2 sekunder, venter forsøk 2 i 4, forsøk 3 i 8 og så videre. Dette reduserer belastningen på serveren gradvis og gir den tid til å hente seg inn.
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')Legge til jitter i backoff
Hvis mange agenter prøver på nytt samtidig (et vanlig scenario etter et kortvarig driftsavbrudd), våkner de alle samtidig – og skaper en thundering herd som umiddelbart blir utsatt for en ny hastighetsbegrensning. Ved å legge til jitter (tilfeldig forsinkelse) spres de nye forsøkene utover, slik at serverbelastningen reduseres.
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-biblioteket for logikk for nye forsøk. Det håndterer eksponentiell backoff, jitter, maksimalt antall forsøk og egendefinerte stoppbetingelser med en ryddig dekoratorsyntaks. Det er langt mer pålitelig enn egenlagde løkker for nye forsøk.
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 egendefinert betingelse for nye forsøk
Du kan konfigurere tenacity til å prøve på nytt bare ved bestemte statuskoder (for eksempel 429 og 5xx) og stoppe umiddelbart ved klientfeil (4xx), som det ikke hjelper å prøve på nytt etter. Bruk retry_if_result eller en egendefinert callable til å undersøke 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 av hastighetsgrenser
Den beste strategien er å unngå å nå hastighetsgrensene i utgangspunktet. Kontroller hastighetsgrensehodene i hver respons, og senk tempoet når du nærmer deg grensen. Mange API-er inkluderer hodene 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 antall forsøk og når du gir opp
Logikk for nye forsøk må alltid ha en grense. Hvis du prøver for alltid, kan det føre til kaskadefeil der alle agentene blir sittende fast i løkker med nye forsøk. Etter max_retries skal du utløse et siste unntak med kontekst om hva som mislyktes, slik at agenten kan logge det og gå videre til annet arbeid.
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ønsteret circuit breaker
Mønsteret circuit breaker hindrer agenten i å overbelaste en tjeneste som har feil. Når en terskel for antall feil er nådd, «åpnes» kretsen, og alle forespørsler mislykkes umiddelbart uten å kontakte nettverket. Etter en nedkjølingsperiode prøver den én forespørsel – hvis den lykkes, lukkes kretsen, og normal drift gjenopptas.
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}')
Sette forespørsler i kø for å holde seg innenfor grensene
For agenter som foretar mange kall i en batch, kan du bruke en token bucket eller en enkel ventebasert struping for å holde deg innenfor grensene. Beregn et trygt intervall mellom kall basert på API-ets hastighetsgrense (f.eks. 60 kall/minutt = 1 kall per 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 resultsKombinere logikk for nye forsøk med backoff-headere
Det mest robuste mønsteret kombinerer ventetider angitt av serveren (Retry-After) med eksponentiell backoff som reserve. Foretrekk alltid serverens veiledning når den er tilgjengelig – serveren vet nøyaktig når du kan prøve på nytt.
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')Kort test: eksponentiell backoff
Test forståelsen din av strategier for nye forsøk.
Oppsummering av hastighetsbegrensning og nye forsøk
Agentene dine kan nå håndtere hastighetsbegrensninger på en god måte:
- 429 Too Many Requests – respekter
Retry-After-headeren, og vent før du prøver på nytt - Eksponentiell backoff –
wait = 2^attemptdobler ventetiden ved hvert nye forsøk - Jitter – tilfører tilfeldighet slik at nye forsøk spres mellom flere agentinstanser
- tenacity – håndterer all logikk for nye forsøk med dekoratorer og ryddig konfigurasjon
- Circuit breaker – hindrer gjentatte kall til en tjeneste som har feil, etter en angitt terskel
- Proaktiv struping – kontroller
X-RateLimit-Remaining, og senk tempoet før du når grensen
Lær deg AI-agenter med en AI-veileder – gratis
Skriv og kjør ekte kode i nettleseren, få umiddelbar hjelp fra en AI-veileder som er tilgjengelig døgnet rundt, og fortsett der du slapp – på nettet eller i appen.
- Kurs
- 60
- Leksjoner
- 239
Ofte stilte spørsmål
Er leksjonen «Frekvensbegrensning og logikk for nye forsøk» gratis?
Ja – hele teksten i «Frekvensbegrensning og logikk for nye forsøk» er gratis å lese her på nettet. For å øve interaktivt med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt, og for å låse opp resten av AI-agenter-kurset, kan du oppgradere til CoddyKit PRO. Kurset i AI-agenter inneholder totalt 4 leksjoner.
Hva lærer jeg i «Frekvensbegrensning og logikk for nye forsøk»?
Eksponentiell backoff, håndtering av 429 og hensynsfull bruk av API-er. Du øver på AI-agenter med praktisk kode som du kjører direkte i nettleseren, mens en AI-veileder som er tilgjengelig døgnet rundt, svarer på spørsmålene dine mens du jobber deg gjennom leksjonen.
Trenger jeg erfaring for å begynne med AI-agenter?
Ingen tidligere erfaring er nødvendig. AI-agenter på CoddyKit er lagt opp for både nybegynnere og viderekomne, så De kan begynne her eller helt fra start og lære i Deres eget tempo. Dette er leksjon 4 av 4.
Hvor lang tid tar leksjonen «Frekvensbegrensning og logikk for nye forsøk»?
De fleste CoddyKit-leksjoner tar omtrent 5–10 minutter. Hver leksjon er kort og interaktiv, slik at De gjør jevne fremskritt og kan fortsette akkurat der De slapp – både på nettet og i appen.
Kan jeg skrive og kjøre kode i denne AI-agenter-leksjonen?
Ja. Alle AI-agenter-leksjoner har en innebygd kodeeditor, slik at De kan skrive og kjøre ekte kode direkte i nettleseren og få umiddelbar tilbakemelding fra AI – uten lokal konfigurering.
Alle leksjonene i dette kurset
- Grunnleggende REST API for agentutviklere
- Autentisering: API-nøkler og OAuth
- Håndtere API-svar og feil
- Frekvensbegrensning og logikk for nye forsøk