การลองใหม่ด้วยการหน่วงแบบทวีคูณ

เหตุใดจึงต้องลองใหม่

สิ่งที่ไม่ควรทำ

# BAD
for _ in range(10):
    try:
        return call_api()
    except Exception:
        pass

การหน่วงเวลาแบบทวีคูณ

import time

def call_with_backoff(fn, attempts=5):
    for i in range(attempts):
        try:
            return fn()
        except TransientError:
            if i == attempts - 1:
                raise
            time.sleep(2 ** i)   # 1, 2, 4, 8, 16 seconds

เพิ่มความแปรผันแบบสุ่ม

import random
sleep = (2 ** i) + random.uniform(0, 1)
time.sleep(sleep)

ใช้ Tenacity

from tenacity import retry, wait_exponential_jitter, stop_after_attempt, retry_if_exception_type
import requests

@retry(
    wait=wait_exponential_jitter(initial=1, max=30),
    stop=stop_after_attempt(5),
    retry=retry_if_exception_type((requests.Timeout, requests.ConnectionError))
)
def search(query):
    return requests.get('https://api.tavily.com/search', timeout=10).json()

ลองใหม่เฉพาะข้อผิดพลาดชั่วคราว

การลองใหม่โดยคำนึงถึงขีดจำกัดอัตราการเรียกใช้

if response.status_code == 429:
    wait = int(response.headers.get('Retry-After', 5))
    time.sleep(wait)
    continue

จำกัดเวลารวมของการลองใหม่

from tenacity import stop_after_delay

@retry(
    wait=wait_exponential_jitter(initial=1, max=10),
    stop=stop_after_delay(60)
)
def call_api():
    ...

ส่วนหัวไอดอมโพเทนซี

headers = {'Idempotency-Key': f'agent-call-{run_id}-{tool_id}'}

ลองใหม่กับการเรียกใช้ LLM ด้วย

client = OpenAI(max_retries=5, timeout=30.0)

การลองใหม่แบบอะซิงโครนัส

from tenacity import AsyncRetrying

async for attempt in AsyncRetrying(wait=wait_exponential_jitter(), stop=stop_after_attempt(5)):
    with attempt:
        result = await call_api()

บันทึกการลองใหม่

การลองใหม่กับการกำหนดหมดเวลา

เหตุใดจึงต้องใช้ความแปรผันแบบสุ่ม

สรุป