AI 智能體遇上 429 錯誤便崩潰，還在白白燒錢

Your AI Agent Crashes on 429 Errors and You're Burning Money

Handling rate limits the right way saves you from unexpected bills and keeps your app reliable.

Why 429 Errors Matter

When you call an AI service (OpenAI, Anthropic, Cohere, etc.) you're usually limited by a rate limit – a maximum number of requests per minute or per day. If you exceed it, the API returns HTTP 429 Too Many Requests.

If your code treats a 429 like any other error and just crashes or retries blindly, two bad things happen:

Lost work – the request that triggered the limit is dropped, so you may need to redo expensive generations.
Cost spiral – naïve retry loops can hammer the endpoint, causing more 429s and potentially triggering over‑usage fees or account suspensions.

Handling 429s gracefully means:

Pausing just long enough for the limit to reset.
Using the server's hint (Retry-After header) when available.
Applying exponential back‑off to avoid thundering‑herd problems.
Logging what happened so you can tune your traffic later.

Beginner‑Friendly Retry Logic in Python

Below is a self‑contained script that:

Wraps any API call in a retry function.
Detects 429 responses.
Honors the Retry-After header (seconds) or falls back to exponential back‑off.
Limits total attempts to avoid infinite loops.
Prints helpful logs so you can see what's happening.

#!/usr/bin/env python3
"""
Example: Robust AI API caller with 429 handling.
Replace the dummy `call_ai_api` with your actual SDK call.
"""

import time
import random
import logging
from typing import Any, Callable

# ----------------------------------------------------------------------
# Configuration – tweak these for your workload
# ----------------------------------------------------------------------
MAX_RETRIES = 5               # How many times we'll retry a 429
BASE_BACKOFF = 1.0            # Seconds for the first back‑off step
JITTER = 0.1                  # Random jitter to avoid sync retries
LOG_LEVEL = logging.INFO

logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s %(levelname)s %(message)s",
)

# ----------------------------------------------------------------------
# Dummy AI API function – replace with real call (e.g., openai.ChatCompletion.create)
# ----------------------------------------------------------------------
def call_ai_api(prompt: str) -> dict:
    """
    Simulates an AI endpoint that randomly returns 429.
    In real code, this would be the actual HTTP request to the provider.
    """
    # Simulate network latency
    time.sleep(random.uniform(0.2, 0.5))

    # 20% chance of hitting a rate limit for demo purposes
    if random.random() < 0.2:
        # Pretend the server asks us to wait 2 seconds
        fake_response = type("Response", (), {})()
        fake_response.status_code = 429
        fake_response.headers = {"Retry-After": "2"}
        fake_response.text = "Rate limit exceeded"
        return fake_response

    # Successful response
    fake_response = type("Response", (), {})()
    fake_response.status_code = 200
    fake_response.json = lambda: {"answer": f"You said: {prompt}"}
    return fake_response


# ----------------------------------------------------------------------
# Core retry logic
# ----------------------------------------------------------------------
def robust_ai_call(
    func: Callable[..., Any],
    *args,
    max_retries: int = MAX_RETRIES,
    base_backoff: float = BASE_BACKOFF,
    jitter: float = JITTER,
    **kwargs,
) -> Any:
    """
    Calls `func(*args, **kwargs)` and retries on HTTP 429.
    Returns the successful response or raises the last exception.
    """
    attempt = 0
    while True:
        attempt += 1
        response = func(*args, **kwargs)

        # Assume response objects have a `status_code` attribute (like requests)
        if getattr(response, "status_code", None) == 429:
            if attempt > max_retries:
                logging.error(
                    f"Exceeded max retries ({max_retries}) for 429. Giving up."
                )
                raise RuntimeError("Rate limit exceeded after retries")

            # Use Retry-After header if present, else exponential back‑off
            retry_after = response.headers.get("Retry-After")
            if retry_after:
                try:
                    wait = float(retry_after)
                except ValueError:
                    wait = base_backoff * (2 ** (attempt - 1))
            else:
                wait = base_backoff * (2 ** (attempt - 1))

            # Add jitter to avoid synchronized retries across many instances
            wait += random.uniform(-jitter, jitter)
            wait = max(wait, 0)  # guard against negative

            logging.warning(
                f"Received 429 on attempt {attempt}. "
                f"Waiting {wait:.2f}s before retry..."
            )
            time.sleep(wait)
            continue

        # Any non‑429 status is treated as final (success or other error)
        if getattr(response, "status_code", None) != 200:
            logging.error(
                f"Non‑200 response ({response.status_code}): {getattr(response, 'text', '')}"
            )
            return response  # let caller decide what to do

        # Success!
        logging.info(f"Successful call on attempt {attempt}")
        return response


# ----------------------------------------------------------------------
# Example usage
# ----------------------------------------------------------------------
if __name__ == "__main__":
    prompt = "Explain quantum entanglement in one sentence."
    try:
        result = robust_ai_call(call_ai_api, prompt)
        if result.status_code == 200:
            print("AI response:", result.json()["answer"])
        else:
            print("Failed after retries.")
    except Exception as exc:
        print("Error:", exc)

How the script works

Step	What happens
1️⃣ Call	`robust_ai_call` invokes your real API function (`call_ai_api`).
2️⃣ Check 429	If the response status is `429`, we look for a `Retry-After` header.
3️⃣ Wait	Use the header value; otherwise, wait `base_backoff * 2^(attempt‑1)` seconds, plus a small random jitter.
4️⃣ Retry	Loop up to `MAX_RETRIES` times. If we exceed it, we log an error and raise.
5️⃣ Success	On any non‑429 response (including 200), we return the result to the caller.
6️⃣ Logging	Every attempt, wait, and outcome is logged so you can monitor traffic.

You can drop this function into any project—just replace call_ai_api with the actual request you make to your AI provider (OpenAI, Azure, HuggingFace Inference API, etc.). The retry logic stays the same.

Why This Prevents Costly Surprises

Predictable spend – By respecting the provider's back‑off, you avoid sending thousands of redundant requests that would otherwise be billed.
Better user experience – Instead of crashing, your app gracefully pauses and shows a friendly "please wait" message.
Operational visibility – Logs give you concrete data on how often you hit limits, helping you decide whether to request a higher quota or smooth traffic with a queue.

In short, handling 429s isn't just about avoiding errors; it's a cost‑control mechanism that keeps your AI‑powered service reliable and affordable.

What's your experience?

Have you ever seen a sudden spike in your AI bill because of unhandled rate limits? What strategies (queues, token buckets, etc.) have you tried to smooth traffic? Share your thoughts and any tips in the comments below!

Tags: ai, python, webdev, beginners