0Pricing
AI Agents · Pelajaran

Pembatasan Laju dan Logika Percobaan Ulang

Kemunduran eksponensial, penanganan 429, dan penggunaan API secara bijaksana.

Pembatasan Laju dan Logika Percobaan Ulang adalah pelajaran AI Agents gratis di CoddyKit. Ini adalah pelajaran 4 dari 4. Kamu bisa membaca pelajaran lengkapnya di bawah secara gratis — lalu praktikkan langsung di browser dengan editor kode bawaan dan tutor AI 24/7. Ini adalah bagian dari jalur belajar AI Agents, dan progresmu tersinkronisasi di web dan aplikasi CoddyKit. Kursus AI Agents mencakup 4 pelajaran total.

Apa Itu Pembatasan Laju?

Pembatasan laju adalah cara API melindungi diri agar tidak kewalahan. Ketika agen Anda mengirim terlalu banyak permintaan dalam waktu singkat, API mengembalikan 429 Too Many Requests. Batas yang umum mencakup jumlah permintaan per detik, per menit, atau per hari.

Mengabaikan batas laju dapat menyebabkan agen diblokir, kunci API dicabut, dan biaya tambahan.

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}')

Header Retry-After

Ketika API mengembalikan 429, API sering menyertakan header Retry-After yang memberi tahu secara tepat berapa detik Anda harus menunggu sebelum mencoba lagi. Selalu patuhi header ini—mengabaikannya dan langsung mencoba lagi hanya akan membuat Anda menerima 429 lagi.

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()

Penundaan Eksponensial

Penundaan eksponensial adalah strategi percobaan ulang standar: tunggu lebih lama setelah setiap upaya yang gagal. Jika upaya pertama menunggu 2 detik, upaya kedua menunggu 4 detik, upaya ketiga 8 detik, dan seterusnya. Strategi ini secara bertahap mengurangi beban pada server dan memberinya waktu untuk pulih.

Rumus: 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')

Menambahkan Keacakan pada Penundaan

Jika banyak agen mencoba lagi pada waktu yang sama (situasi yang umum setelah gangguan singkat), semuanya aktif kembali secara bersamaan—menciptakan gerombolan besar yang segera terkena pembatasan laju lagi. Menambahkan keacakan (jeda acak) menyebarkan percobaan ulang, sehingga mengurangi beban server.

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')

Pustaka Ketekunan

Ketekunan adalah pustaka Python yang paling populer untuk logika percobaan ulang. Pustaka ini menangani penundaan eksponensial, keacakan, jumlah percobaan ulang maksimum, dan kondisi penghentian khusus dengan sintaks dekorator yang rapi. Pustaka ini jauh lebih andal daripada perulangan percobaan ulang yang dibuat secara manual.

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()

Ketekunan dengan Kondisi Percobaan Ulang Khusus

Anda dapat mengajarkan ketekunan untuk mencoba lagi hanya pada kode status tertentu (seperti 429 dan 5xx), lalu segera berhenti pada kesalahan klien (4xx) yang tidak akan terbantu dengan percobaan ulang. Gunakan retry_if_result atau callable khusus untuk memeriksa respons.

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()

Pengelolaan Batas Laju secara Proaktif

Strategi terbaik adalah menghindari batas laju sejak awal. Periksa header batas laju pada setiap respons dan perlambat proses ketika Anda mendekati batas. Banyak API menyertakan header X-RateLimit-Remaining dan 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()

Percobaan Ulang Maksimum dan Berhenti

Logika percobaan ulang harus selalu memiliki batas. Mencoba lagi tanpa batas dapat menyebabkan kegagalan berantai ketika semua agen terjebak dalam perulangan percobaan ulang. Setelah max_retries, munculkan pengecualian terakhir yang menyertakan konteks tentang kegagalan tersebut, sehingga agen dapat mencatatnya dan melanjutkan pekerjaan lain.

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)

Pola Pemutus Sirkuit

Pola pemutus sirkuit mencegah agen Anda terus-menerus membebani layanan yang gagal. Setelah jumlah kegagalan melewati ambang batas, sirkuit "terbuka" dan semua permintaan langsung gagal tanpa mengakses jaringan. Setelah periode pendinginan, pola ini mencoba satu permintaan—jika berhasil, sirkuit tertutup dan operasi normal dilanjutkan.

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}')

Mengantrekan Permintaan agar Tetap dalam Batas

Untuk agen yang melakukan banyak panggilan dalam satu kelompok, gunakan keranjang token atau pembatas sederhana berbasis jeda untuk tetap berada dalam batas. Hitung interval aman antar-panggilan berdasarkan batas laju API (misalnya, 60 panggilan/menit = 1 panggilan per detik).

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

Menggabungkan Logika Percobaan Ulang dengan Header Penundaan

Pola yang paling tangguh menggabungkan waktu tunggu yang ditentukan server (Retry-After) dengan penundaan eksponensial sebagai cadangan. Selalu dahulukan panduan server jika tersedia—server mengetahui secara tepat kapan Anda dapat mencoba lagi.

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')

Uji Singkat: Penundaan Eksponensial

Uji pemahaman Anda tentang strategi percobaan ulang.

Ringkasan Pembatasan Laju dan Percobaan Ulang

Agen Anda kini dapat menangani batas laju dengan baik:

  • 429 Too Many Requests—patuhi header Retry-After; tunggu sebelum mencoba lagi
  • Penundaan eksponensial—wait = 2^attempt menggandakan waktu tunggu pada setiap percobaan ulang
  • Keacakan—menambahkan unsur acak untuk menyebarkan percobaan ulang di antara beberapa instans agen
  • Ketekunan—menangani seluruh logika percobaan ulang dengan dekorator dan konfigurasi yang rapi
  • Pemutus sirkuit—berhenti terus-menerus membebani layanan yang gagal setelah melewati ambang batas
  • Pembatasan proaktif—periksa X-RateLimit-Remaining dan perlambat proses sebelum mencapai batas

Pertanyaan yang Sering Diajukan

Apakah pelajaran “Pembatasan Laju dan Logika Percobaan Ulang” gratis?

Ya — teks lengkap “Pembatasan Laju dan Logika Percobaan Ulang” gratis dibaca di sini di web. Untuk praktiknya secara interaktif (editor kode bawaan dan tutor AI 24/7) dan buka sisa kursus AI Agents, upgrade ke CoddyKit PRO. Kursus AI Agents mencakup 4 pelajaran total.

Apa yang akan aku pelajari di “Pembatasan Laju dan Logika Percobaan Ulang”?

Kemunduran eksponensial, penanganan 429, dan penggunaan API secara bijaksana. Kamu berlatih AI Agents dengan kode praktik yang langsung kamu jalankan di browser, dan tutor AI 24/7 menjawab pertanyaanmu saat kamu mengerjakan pelajaran ini.

Apakah aku perlu pengalaman untuk memulai AI Agents?

Tidak diperlukan pengalaman sebelumnya. AI Agents di CoddyKit dirancang untuk pemula hingga pelajar tingkat lanjut, jadi kamu bisa memulai di sini atau dari awal dan belajar sesuai kecepatan kamu sendiri. Ini adalah pelajaran 4 dari 4.

Berapa lama pelajaran “Pembatasan Laju dan Logika Percobaan Ulang” memakan waktu?

Sebagian besar pelajaran CoddyKit memakan waktu sekitar 5–10 menit. Setiap pelajaran ringkas dan interaktif, jadi kamu membuat kemajuan stabil dan melanjutkan dari tempat kamu tinggalkan di web dan aplikasi.

Bisakah aku menulis dan menjalankan kode dalam pelajaran AI Agents ini?

Ya. Setiap pelajaran AI Agents menyertakan editor kode bawaan, jadi kamu menulis dan menjalankan kode nyata langsung di browser dan mendapatkan umpan balik AI instan — tidak diperlukan penyiapan lokal.

Semua pelajaran dalam kursus ini

  1. Dasar-Dasar REST API untuk Pengembang Agen
  2. Autentikasi: Kunci API dan OAuth
  3. Menangani Respons dan Error API
  4. Pembatasan Laju dan Logika Percobaan Ulang
← Kembali ke AI Agents