Resilient API Integration: Streaming & Retries

Moving from prototype to production means your AI integrations need to handle the real world: rate limits, server errors, network timeouts, partial responses, and the need to stream output to users as it's generated. This lesson covers the patterns that keep AI applications running when the API doesn't cooperate.

Streaming Responses

Streaming is critical for user experience. Without it, users stare at a blank screen for 5–10 seconds, then see the full response appear at once. With streaming, they see tokens appear in real time — much less jarring for long outputs.

Basic Streaming

By default an API call blocks until the entire completion is ready. Passing stream=True instead returns an iterator that yields the response in small chunks as the model generates them — the difference between a spinner and text that appears live, like ChatGPT's typewriter effect. You loop over the chunks and pull the incremental text out of each delta.

from openai import OpenAI
                      
                      client = OpenAI()
                      
                      
                      def stream_to_console(prompt: str) -> str:
                          """Stream a response, printing each chunk and returning the full text."""
                          full_text = ""
                      
                          with client.chat.completions.stream(
                              model="gpt-4o",
                              max_tokens=1024,
                              messages=[{"role": "user", "content": prompt}],
                          ) as stream:
                              for text_chunk in stream.text_stream:
                                  print(text_chunk, end="", flush=True)
                                  full_text += text_chunk
                      
                          print()  # newline at end
                          return full_text
                      
                      
                      # Usage
                      response = stream_to_console("Explain transformer attention in detail.")

Streaming with Metadata

Streaming the text is only half the job — you usually also need the finish reason and token usage once the stream completes, for logging and cost tracking. This version accumulates the text as it arrives and captures the final metadata from the last chunk, so you get live output and a complete record of the call.

from openai import OpenAI
                      import time
                      
                      def stream_with_metadata(prompt: str) -> dict:
                          """Stream a response and collect usage metadata."""
                          start = time.monotonic()
                          chunks = []
                          final_message = None
                      
                          with client.chat.completions.stream(
                              model="gpt-4o",
                              max_tokens=1024,
                              messages=[{"role": "user", "content": prompt}],
                          ) as stream:
                              for text in stream.text_stream:
                                  chunks.append(text)
                                  yield text  # if used as a generator
                      
                              # Final message available after stream completes
                              final_message = stream.get_final_message()
                      
                          latency = (time.monotonic() - start) * 1000
                          full_text = "".join(chunks)
                      
                          return {
                              "text": full_text,
                              "input_tokens": final_message.usage.prompt_tokens,
                              "output_tokens": final_message.usage.completion_tokens,
                              "latency_ms": latency,
                          }
                      
                      
                      # Async streaming
                      async def async_stream(prompt: str):
                          """Async streaming for concurrent applications."""
                          async with AsyncOpenAI() as async_client:
                              async with async_client.chat.completions.stream(
                                  model="gpt-4o",
                                  max_tokens=1024,
                                  messages=[{"role": "user", "content": prompt}],
                              ) as stream:
                                  async for text in stream.text_stream:
                                      yield text  # async generator

Production Retry Strategies

Idempotency: Safe to Retry

An operation is idempotent if calling it multiple times has the same effect as calling it once. LLM read requests are naturally idempotent — asking the same question twice doesn't change the world. Write operations (persisting results, billing) must be idempotent by design.

import hashlib
                      import json
                      from typing import Optional
                      
                      # Idempotency key pattern for write operations
                      def generate_idempotency_key(request_data: dict) -> str:
                          """Generate a deterministic key from request content."""
                          canonical = json.dumps(request_data, sort_keys=True)
                          return hashlib.sha256(canonical.encode()).hexdigest()[:16]
                      
                      
                      # Cache results to avoid duplicate charges
                      _request_cache: dict[str, str] = {}
                      
                      def idempotent_complete(
                          prompt: str,
                          model: str = "gpt-4o",
                          use_cache: bool = True,
                      ) -> str:
                          """Cache LLM responses by content hash — prevents duplicate API charges on retry."""
                          cache_key = generate_idempotency_key({"prompt": prompt, "model": model})
                      
                          if use_cache and cache_key in _request_cache:
                              return _request_cache[cache_key]
                      
                          response = client.chat.completions.create(
                              model=model,
                              max_tokens=1024,
                              messages=[{"role": "user", "content": prompt}],
                          )
                          result = response.choices[0].message.content
                      
                          if use_cache:
                              _request_cache[cache_key] = result
                      
                          return result

Robust Retry with tenacity

Transient failures — rate limits, timeouts, brief 5xx errors — are normal at scale, and the fix is to retry with exponential backoff rather than fail. The tenacity library turns that into a declarative decorator: you specify how many attempts, how long to wait between them, and which exception types are worth retrying (retrying a 400 Bad Request would just waste calls). It's the Polly of the Python world.

from openai import OpenAI
                      import logging
                      from tenacity import (
                          retry, stop_after_attempt, wait_exponential,
                          retry_if_exception_type, before_sleep_log, after_log
                      )
                      
                      logger = logging.getLogger(__name__)
                      
                      
                      def is_retryable(exception: Exception) -> bool:
                          """Custom predicate — retry on transient errors only."""
                          if isinstance(exception, openai.RateLimitError):
                              return True
                          if isinstance(exception, openai.InternalServerError):
                              return exception.status_code >= 500
                          if isinstance(exception, openai.APIConnectionError):
                              return True
                          return False
                      
                      
                      @retry(
                          retry=retry_if_exception_type((
                              openai.RateLimitError,
                              openai.InternalServerError,
                              openai.APIConnectionError,
                          )),
                          wait=wait_exponential(multiplier=2, min=2, max=60),
                          stop=stop_after_attempt(6),
                          before_sleep=before_sleep_log(logger, logging.WARNING),
                          reraise=True,
                      )
                      def resilient_create(client: OpenAI, **kwargs) -> openai.types.chat.ChatCompletion:
                          """Drop-in wrapper for client.chat.completions.create with automatic retry."""
                          return client.chat.completions.create(**kwargs)

Timeout Management

LLM requests can hang for 30+ seconds on overloaded servers. Setting timeouts prevents your application from waiting indefinitely:

from openai import OpenAI
                      import httpx
                      
                      # Configure timeout at client level
                      client = OpenAI(
                          timeout=httpx.Timeout(
                              connect=5.0,    # TCP connection timeout
                              read=60.0,      # Time to wait for server response (long for streaming)
                              write=10.0,     # Time to send request body
                              pool=5.0,       # Time to acquire connection from pool
                          )
                      )
                      
                      # Or override per-request
                      def fast_complete(prompt: str, timeout_seconds: float = 10.0) -> str | None:
                          """Complete with aggressive timeout — for latency-critical paths."""
                          try:
                              response = client.chat.completions.create(
                                  model="gpt-4o-mini-20240307",  # faster model for short timeout
                                  max_tokens=256,
                                  messages=[{"role": "user", "content": prompt}],
                                  timeout=timeout_seconds,
                              )
                              return response.choices[0].message.content
                          except openai.APITimeoutError:
                              logger.warning(f"Request timed out after {timeout_seconds}s")
                              return None

Circuit Breaker Pattern

A circuit breaker stops sending requests to a service that's consistently failing, giving it time to recover. This prevents your app from hammering a failing API:

import time
                      from enum import Enum
                      from dataclasses import dataclass, field
                      
                      class CircuitState(Enum):
                          CLOSED = "closed"       # normal operation
                          OPEN = "open"           # blocking requests
                          HALF_OPEN = "half_open" # testing recovery
                      
                      
                      @dataclass
                      class CircuitBreaker:
                          """Simple circuit breaker for AI API calls."""
                          failure_threshold: int = 5       # failures before opening
                          recovery_timeout: float = 60.0   # seconds before trying again
                      
                          _state: CircuitState = field(default=CircuitState.CLOSED, init=False)
                          _failure_count: int = field(default=0, init=False)
                          _last_failure_time: float = field(default=0.0, init=False)
                      
                          def call(self, func, *args, **kwargs):
                              if self._state == CircuitState.OPEN:
                                  if time.monotonic() - self._last_failure_time > self.recovery_timeout:
                                      self._state = CircuitState.HALF_OPEN
                                      logger.info("Circuit breaker: half-open, testing API")
                                  else:
                                      raise RuntimeError("Circuit breaker OPEN — API calls blocked")
                      
                              try:
                                  result = func(*args, **kwargs)
                                  self._on_success()
                                  return result
                              except (openai.RateLimitError, openai.InternalServerError) as e:
                                  self._on_failure()
                                  raise
                      
                          def _on_success(self):
                              self._failure_count = 0
                              self._state = CircuitState.CLOSED
                      
                          def _on_failure(self):
                              self._failure_count += 1
                              self._last_failure_time = time.monotonic()
                              if self._failure_count >= self.failure_threshold:
                                  self._state = CircuitState.OPEN
                                  logger.error(f"Circuit breaker OPENED after {self._failure_count} failures")
                      
                      
                      # Usage
                      breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30.0)
                      
                      def protected_call(prompt: str) -> str:
                          return breaker.call(
                              client.chat.completions.create,
                              model="gpt-4o",
                              max_tokens=512,
                              messages=[{"role": "user", "content": prompt}],
                          ).choices[0].message.content

Rate Limit Awareness

Retrying blindly still hits the wall if you're simply sending requests too fast. Being rate-limit-aware means reading the limit headers the API returns and proactively pacing yourself — backing off before you get a 429 rather than after. The pattern below inspects remaining-quota headers and sleeps when you're close to the limit.

import time
                      import threading
                      from collections import deque
                      
                      class RateLimiter:
                          """Token bucket rate limiter — respect API rate limits proactively."""
                      
                          def __init__(self, requests_per_minute: int = 50):
                              self._rpm = requests_per_minute
                              self._window = 60.0
                              self._timestamps: deque = deque()
                              self._lock = threading.Lock()
                      
                          def acquire(self) -> None:
                              """Block until a request slot is available."""
                              with self._lock:
                                  now = time.monotonic()
                      
                                  # Remove timestamps older than our window
                                  while self._timestamps and now - self._timestamps[0] > self._window:
                                      self._timestamps.popleft()
                      
                                  # If we've hit the limit, wait until the oldest request is outside the window
                                  if len(self._timestamps) >= self._rpm:
                                      wait_time = self._window - (now - self._timestamps[0])
                                      if wait_time > 0:
                                          time.sleep(wait_time)
                      
                                  self._timestamps.append(time.monotonic())
                      
                      
                      limiter = RateLimiter(requests_per_minute=40)  # conservative buffer below actual limit
                      
                      def rate_limited_call(prompt: str) -> str:
                          limiter.acquire()  # blocks if needed
                          return client.chat.completions.create(
                              model="gpt-4o",
                              max_tokens=512,
                              messages=[{"role": "user", "content": prompt}],
                          ).choices[0].message.content