Module 2: OpenAI API - Introduction

Error Handling and Retry Strategies

Next →

Capsule overview

APIs fail. OpenAI is no exception. In production you'll see:

  • Rate limit errors (429): Too many requests
  • Timeouts: A slow API
  • Server errors (500, 503): OpenAI is having issues
  • Network errors: Your internet fails

This capsule teaches you to handle all these cases with robust retry strategies.

Time: 25 minutes
Difficulty: Medium-High

🎯 Objectives

  • ✅ Identify common error types
  • ✅ Implement exponential backoff
  • ✅ Build robust retry logic
  • ✅ Log errors

❌ Common Error Types

1. RateLimitError (429)

When: You exceed requests/min or tokens/min

openai.RateLimitError: Rate limit reached for requests

Common cause:

  • Free tier: >60 requests/min
  • A loop without delays
  • A traffic spike

Solution: Retry with backoff

2. APIError (500, 503)

When: OpenAI's servers are having problems

openai.APIError: The server had an error processing your request

Cause:

  • OpenAI outage (rare, but it happens)
  • A deploy in progress

Solution: Automatic retry

3. Timeout

When: A request takes >60s (default timeout)

openai.APITimeoutError: Request timed out

Cause:

  • A slow OpenAI API
  • Network congestion
  • A very long prompt

Solution: Increase the timeout or retry

4. AuthenticationError (401)

When: The API key is invalid

openai.AuthenticationError: Incorrect API key provided

Cause:

  • Wrong key
  • Revoked key
  • A typo in .env

Solution: Check the key (do NOT retry)

5. NetworkError

When: No internet or DNS issues

requests.exceptions.ConnectionError: Failed to establish connection

Cause:

  • No internet
  • DNS failure
  • A firewall blocking it

Solution: Retry with a short backoff

🔄 Retry Strategy: Exponential Backoff

Concept:

Wait a growing amount of time between retries:

  • Retry 1: Wait 1s
  • Retry 2: Wait 2s
  • Retry 3: Wait 4s
  • Retry 4: Wait 8s

Why it works:

  • If OpenAI is saturated, give it time to recover
  • Avoids a "stampede" (everyone retrying at the same time)

Basic implementation:

import time
from openai import OpenAI, RateLimitError, APIError

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def chat_with_retry(prompt: str, max_retries: int = 3):
    """
    Send a prompt with automatic retry.
    
    Args:
        prompt: The user's message
        max_retries: Maximum attempts (default: 3)
        
    Returns:
        GPT's response, or None if it fails
    """
    
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}]
            )
            return response.choices[0].message.content
            
        except RateLimitError as e:
            if attempt == max_retries - 1:  # Last attempt
                print(f"❌ Rate limit after {max_retries} attempts")
                return None
            
            # Exponential backoff
            wait_time = 2 ** attempt  # 1s, 2s, 4s
            print(f"⚠️ Rate limit, waiting {wait_time}s...")
            time.sleep(wait_time)
            
        except APIError as e:
            if attempt == max_retries - 1:
                print(f"❌ API error after {max_retries} attempts: {e}")
                return None
            
            wait_time = 2 ** attempt
            print(f"⚠️ API error, retrying in {wait_time}s...")
            time.sleep(wait_time)
            
        except Exception as e:
            # Unexpected errors (no retry)
            print(f"❌ Unexpected error: {e}")
            return None
    
    return None

# Test
answer = chat_with_retry("What is Python?")
if answer:
    print(answer)
else:
    print("Couldn't get a response")

Improvement: Jitter (randomization)

Add a small random value so retries don't all fire at the same time:

import random

wait_time = (2 ** attempt) + random.uniform(0, 1)

Benefit: Spreads the load across OpenAI's servers

🛠️ Advanced Implementation with Tenacity

The tenacity library (recommended for production):

pip install tenacity

Code:

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
from openai import OpenAI, RateLimitError, APIError, APITimeoutError

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

@retry(
    retry=retry_if_exception_type((RateLimitError, APIError, APITimeoutError)),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    stop=stop_after_attempt(5),
    reraise=True
)
def chat_with_tenacity(prompt: str) -> str:
    """
    Send a prompt with automatic retry (tenacity).
    
    Retry only on transient errors:
    - RateLimitError
    - APIError
    - APITimeoutError
    
    Do NOT retry on:
    - AuthenticationError (invalid key)
    - InvalidRequestError (invalid prompt)
    """
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        timeout=30.0  # 30s timeout
    )
    return response.choices[0].message.content

# Test
try:
    answer = chat_with_tenacity("What is Python?")
    print(answer)
except Exception as e:
    print(f"❌ Error after retries: {e}")

Advantages:

  • Cleaner code (decorator)
  • Built-in exponential backoff
  • Configurable (min/max wait, max attempts)
  • Retries only on transient errors

📊 Error Logging

Implementation with standard logging:

import logging
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('openai_errors.log'),
        logging.StreamHandler()  # Also print to the console
    ]
)

def chat_with_logging(prompt: str):
    """Chat with error logging."""
    
    try:
        response = client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}]
        )
        
        # Log success
        logging.info(f"Successful request | Tokens: {response.usage.total_tokens}")
        
        return response.choices[0].message.content
        
    except RateLimitError as e:
        logging.warning(f"Rate limit error | Prompt: {prompt[:50]}")
        raise
        
    except APIError as e:
        logging.error(f"API error | Status: {e.status_code} | Message: {e.message}")
        raise
        
    except APITimeoutError as e:
        logging.error(f"Timeout error | Prompt length: {len(prompt)}")
        raise
        
    except Exception as e:
        logging.critical(f"Unexpected error | Type: {type(e)} | Message: {e}")
        raise

# Test
try:
    answer = chat_with_logging("What is Python?")
    print(answer)
except Exception as e:
    print(f"Error: {e}")

Output in openai_errors.log:

2024-02-15 10:30:15 - INFO - Successful request | Tokens: 45
2024-02-15 10:31:22 - WARNING - Rate limit error | Prompt: What is Python?
2024-02-15 10:32:45 - ERROR - API error | Status: 503 | Message: Service unavailable

🎯 Complete Error Handling Strategy

Production-ready code:

import time
import logging
from typing import Optional
from openai import OpenAI, RateLimitError, APIError, APITimeoutError, AuthenticationError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def robust_chat(
    prompt: str,
    max_retries: int = 3,
    initial_wait: float = 1.0
) -> Optional[str]:
    """
    Chat with OpenAI with robust error handling.
    
    Handles:
    - Rate limits (retry with backoff)
    - API errors (retry with backoff)
    - Timeouts (retry with a larger timeout)
    - Auth errors (no retry, critical log)
    - Other errors (no retry, log)
    
    Returns:
        GPT's response, or None if it fails after retries
    """
    
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}],
                timeout=30.0 + (attempt * 10)  # Increase the timeout on each retry
            )
            
            logger.info(f"✅ Successful request | Attempt: {attempt + 1} | Tokens: {response.usage.total_tokens}")
            return response.choices[0].message.content
            
        except RateLimitError as e:
            logger.warning(f"⚠️ Rate limit | Attempt: {attempt + 1}/{max_retries}")
            
            if attempt == max_retries - 1:
                logger.error("❌ Persistent rate limit after retries")
                return None
            
            wait_time = initial_wait * (2 ** attempt)
            logger.info(f"Waiting {wait_time}s before retry...")
            time.sleep(wait_time)
            
        except APIError as e:
            logger.warning(f"⚠️ API error {e.status_code} | Attempt: {attempt + 1}/{max_retries}")
            
            if attempt == max_retries - 1:
                logger.error(f"❌ Persistent API error: {e.message}")
                return None
            
            wait_time = initial_wait * (2 ** attempt)
            time.sleep(wait_time)
            
        except APITimeoutError as e:
            logger.warning(f"⚠️ Timeout | Attempt: {attempt + 1}/{max_retries}")
            
            if attempt == max_retries - 1:
                logger.error("❌ Persistent timeout")
                return None
            
            # Retry with a larger timeout (already configured above)
            wait_time = initial_wait
            time.sleep(wait_time)
            
        except AuthenticationError as e:
            logger.critical(f"❌ CRITICAL: Auth error | Check the API key")
            return None  # NO retry (invalid key)
            
        except Exception as e:
            logger.error(f"❌ Unexpected error: {type(e).__name__} | {e}")
            return None  # NO retry (unknown error)
    
    return None

# Test
answer = robust_chat("What is Python?")
if answer:
    print(answer)
else:
    print("Couldn't get a response after retries")

🧪 Testing Error Handling

Simulate a rate limit:

# Force a rate limit (send many requests)
for i in range(100):
    answer = robust_chat(f"Request {i}")
    # Eventually you'll see rate limit errors and retries

Simulate a timeout:

# A very long prompt (more likely to time out)
long_prompt = "Explain in detail " * 1000  # 5000+ tokens
answer = robust_chat(long_prompt)

📊 Summary

Key concepts:

  1. Error types:

    • Rate limit (429) → Retry
    • API error (500, 503) → Retry
    • Timeout → Retry with a larger timeout
    • Auth (401) → NO retry, fix the key
    • Others → NO retry, log

  2. Exponential backoff:

    wait_time = 2^attempt  # 1s, 2s, 4s, 8s

  3. Tenacity (recommended):

    • @retry decorator
    • Configurable
    • Built-in exponential backoff

  4. Logging:

    • INFO: Successful request
    • WARNING: Error with retry
    • ERROR: Persistent error
    • CRITICAL: Auth or critical bug

Checklist:

  • You implemented retry with exponential backoff
  • You handle 3+ error types (rate limit, API error, timeout)
  • Logging configured (file + console)
  • You tested with a simulated rate limit

🔗 Additional resources

  1. OpenAI Error Codes - Official docs
  2. Tenacity - Retry library
  3. Python Logging - Standard docs

➡️ Next step

Next capsule: 08-project-support-chatbot.md

The module's final project!

You'll integrate everything you learned:

  • Setup and API keys
  • Conversations with context
  • Optimized parameters
  • Cost tracking
  • Robust error handling

You'll build a production-ready technical support chatbot.

Time: 60-90 minutes

Estimated time: 25 minutes
Next: 08-project-support-chatbot.md

← PreviousPricing, Rate Limits, and Cost OptimizationNext →Mini-Project: Technical Support Chatbot