Migrating raw OpenAI SDK error handling to LangChain

Migrating openai sdk error handling langchain migration is a common task when moving from direct API calls to a framework that manages retries, fallbacks, and observability for you. The raw OpenAI SDK exposes a granular exception hierarchy that maps directly to HTTP status codes, while LangChain wraps these into its own abstraction layer with configurable retry policies. This guide walks through the migration end to end, showing how to preserve your existing error-handling semantics while gaining LangChain’s built-in resilience features.

Step 1: Understand the raw OpenAI SDK error hierarchy

Before rewriting anything, catalog the exceptions your current code catches. The OpenAI Python SDK (v1.x) raises exceptions from openai that all inherit from openai.APIError:

from openai import (
    APIError,
    APIConnectionError,
    APITimeoutError,
    RateLimitError,
    AuthenticationError,
    PermissionDeniedError,
    NotFoundError,
    UnprocessableEntityError,
    InternalServerError,
    BadRequestError,
)

Each maps to a specific HTTP status:

• RateLimitError → 429
• AuthenticationError → 401
• PermissionDeniedError → 403
• NotFoundError → 404
• UnprocessableEntityError → 422
• BadRequestError → 400
• InternalServerError → 500
• APITimeoutError → timeout (no HTTP response)
• APIConnectionError → network-level failure

Typical raw SDK error handling looks like this:

from openai import OpenAI, RateLimitError, APITimeoutError, APIConnectionError

client = OpenAI()

def chat_with_retry(messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(
                model="gpt-4o-mini",
                messages=messages,
                timeout=30.0,
            )
        except RateLimitError as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt * 1.5  # exponential backoff
            time.sleep(wait)
        except (APITimeoutError, APIConnectionError) as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(1.5 ** attempt)
        except APIError as e:
            # Non-retryable: bad request, auth, etc.
            raise

Verify: Run this against a test endpoint and confirm it retries on 429/5xx/timeout and fails fast on 400/401/403.

Step 2: Map OpenAI exceptions to LangChain’s error types

LangChain’s OpenAI integration (langchain-openai) wraps the same SDK but surfaces errors through langchain_core.exceptions. The mapping is not one-to-one — LangChain collapses several SDK exceptions into broader categories:

Import the LangChain versions:

from langchain_core.exceptions import (
    RateLimitError as LCRateLimitError,
    APITimeoutError as LCAPITimeoutError,
    APIConnectionError as LCAPIConnectionError,
    AuthenticationError as LCAuthenticationError,
    NotFoundError as LCNotFoundError,
    BadRequestError as LCBadRequestError,
    InternalServerError as LCInternalServerError,
)

Verify: Check langchain_core.exceptions.__all__ to confirm the current exports match your version.

Step 3: Replace manual retry loops with LangChain’s configurable retry policy

LangChain’s ChatOpenAI (and OpenAI for completions) accepts a max_retries parameter and a request_timeout that apply to the underlying SDK calls. The default retry policy retries on:

• Rate limits (429)
• Timeouts
• Connection errors
• 5xx server errors

It does not retry on 4xx client errors (except 429).

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    max_retries=3,           # matches your manual loop
    request_timeout=30.0,    # seconds, maps to SDK timeout
    # Optional: customize which status codes trigger retries
    # retry_on_status_codes=[429, 500, 502, 503, 504],
)

messages = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content="Explain quantum computing in one sentence."),
]

response = llm.invoke(messages)
print(response.content)

If you need custom backoff logic, pass a retry config using tenacity (LangChain’s underlying retry library):

from tenacity import wait_exponential_jitter, stop_after_attempt, retry_if_exception_type
from langchain_core.exceptions import RateLimitError, APITimeoutError, APIConnectionError

llm = ChatOpenAI(
    model="gpt-4o-mini",
    max_retries=3,
    request_timeout=30.0,
    retry=dict(
        wait=wait_exponential_jitter(initial=1, max=10),
        stop=stop_after_attempt(3),
        retry=retry_if_exception_type((RateLimitError, APITimeoutError, APIConnectionError)),
    ),
)

Verify: Set max_retries=0, hit a rate-limited endpoint, and confirm it raises RateLimitError immediately. Then restore retries and confirm it succeeds after backoff.

Step 4: Handle streaming errors and partial responses

Streaming introduces a new failure mode: the stream starts successfully, then fails mid-way. The raw SDK raises exceptions from the iterator. LangChain’s stream() and astream() methods surface the same exceptions but wrap partial output in AIMessageChunk objects.

Raw SDK streaming with error handling:

def stream_with_retry(messages, max_retries=2):
    for attempt in range(max_retries):
        try:
            stream = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=messages,
                stream=True,
                timeout=30.0,
            )
            for chunk in stream:
                if chunk.choices[0].delta.content:
                    yield chunk.choices[0].delta.content
            return  # success
        except (RateLimitError, APITimeoutError, APIConnectionError) as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)
        except APIError:
            raise

LangChain equivalent — note that stream() yields AIMessageChunk and you must handle exceptions around the iteration:

from langchain_core.exceptions import RateLimitError, APITimeoutError, APIConnectionError

def stream_with_langchain(messages, max_retries=2):
    llm = ChatOpenAI(model="gpt-4o-mini", max_retries=max_retries, request_timeout=30.0)
    
    for attempt in range(max_retries):
        try:
            for chunk in llm.stream(messages):
                if chunk.content:
                    yield chunk.content
            return
        except (RateLimitError, APITimeoutError, APIConnectionError):
            if attempt == max_retries - 1:
                raise
            # LangChain already retries internally via max_retries;
            # this outer loop is only needed if you want custom logic between attempts.

For async streaming, use astream() with the same pattern. The key difference: LangChain’s internal retry (via max_retries) applies to the entire request, including stream establishment. If the stream breaks mid-way, LangChain does not automatically resume — you must implement resumption logic yourself if needed.

Verify: Simulate a mid-stream failure (e.g., kill the network after first chunk) and confirm your code catches the exception and decides whether to retry from scratch.

Step 5: Add structured logging and observability

LangChain integrates with standard Python logging. Configure it to capture retry attempts, latency, and error context — this replaces ad-hoc print statements in raw SDK code.

import logging
import json
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult

class StructuredLogger(BaseCallbackHandler):
    def __init__(self, logger_name="llm_calls"):
        self.logger = logging.getLogger(logger_name)
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter('%(message)s'))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
        self.logger.propagate = False

    def on_llm_start(self, serialized, prompts, **kwargs):
        self.logger.info(json.dumps({
            "event": "llm_start",
            "model": serialized.get("kwargs", {}).get("model"),
            "prompt_tokens": sum(len(p.split()) for p in prompts),  # rough estimate
        }))

    def on_llm_end(self, response: LLMResult, **kwargs):
        usage = response.llm_output.get("token_usage", {}) if response.llm_output else {}
        self.logger.info(json.dumps({
            "event": "llm_end",
            "prompt_tokens": usage.get("prompt_tokens"),
            "completion_tokens": usage.get("completion_tokens"),
            "total_tokens": usage.get("total_tokens"),
        }))

    def on_llm_error(self, error, **kwargs):
        self.logger.error(json.dumps({
            "event": "llm_error",
            "error_type": type(error).__name__,
            "error_message": str(error),
        }))

# Attach to the LLM
llm = ChatOpenAI(
    model="gpt-4o-mini",
    max_retries=3,
    request_timeout=30.0,
    callbacks=[StructuredLogger()],
)

For production, forward these logs to your observability stack (Datadog, Splunk, etc.). The callback also fires on retries — each attempt generates llm_start/llm_end or llm_error.

Verify: Run a request that triggers a retry (e.g., against a flaky test double) and confirm the log shows multiple llm_start events for a single logical call.

Step 6: Implement fallback routing for provider failures

If your application uses multiple providers (OpenAI, Anthropic, etc.), LangChain’s Runnable interface makes fallbacks explicit. This is where the migration pays off — raw SDK code requires manual try/except chains; LangChain expresses it declaratively.

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables import RunnableSequence
from langchain_core.exceptions import RateLimitError, APIConnectionError, InternalServerError

primary = ChatOpenAI(model="gpt-4o-mini", max_retries=2, request_timeout=20.0)
fallback = ChatAnthropic(model="claude-3-haiku-20240307", max_retries=2, request_timeout=20.0)

# Chain with fallback on specific exceptions
chain = primary.with_fallbacks(
    [fallback],
    exception_types=(RateLimitError, APIConnectionError, InternalServerError),
)

# Usage
response = chain.invoke(messages)
print(response.content)

The with_fallbacks method retries the primary up to its max_retries, then fails over to the fallback (which also respects its own max_retries). You can stack multiple fallbacks:

tertiary = ChatOpenAI(model="gpt-3.5-turbo", max_retries=1, request_timeout=15.0)
chain = primary.with_fallbacks([fallback, tertiary], exception_types=(RateLimitError, APIConnectionError, InternalServerError))

Verify: Mock the primary to always raise RateLimitError and confirm the fallback responds. Then mock the fallback to also fail and confirm the tertiary (or final exception) surfaces.

Step 7: Verify the migration end to end

Create a test matrix that exercises every error path. Use a local mock server (e.g., pytest-httpserver or respx) to return specific status codes.

# test_error_handling.py
import pytest
import respx
import httpx
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.exceptions import (
    RateLimitError, AuthenticationError, BadRequestError, InternalServerError
)

@respx.mock
def test_rate_limit_triggers_retry_then_succeeds():
    # First two calls return 429, third succeeds
    route = respx.post("https://api.openai.com/v1/chat/completions").mock(
        side_effect=[
            httpx.Response(429, json={"error": {"message": "Rate limit", "type": "rate_limit_error"}}),
            httpx.Response(429, json={"error": {"message": "Rate limit", "type": "rate_limit_error"}}),
            httpx.Response(200, json={
                "id": "test", "object": "chat.completion", "created": 123,
                "model": "gpt-4o-mini", "choices": [{
                    "index": 0, "message": {"role": "assistant", "content": "OK"}, "finish_reason": "stop"
                }],
                "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}
            }),
        ]
    )
    
    llm = ChatOpenAI(model="gpt-4o-mini", max_retries=3, request_timeout=5.0)
    response = llm.invoke([HumanMessage(content="test")])
    
    assert response.content == "OK"
    assert route.call_count == 3

@respx.mock
def test_auth_error_fails_fast_no_retry():
    respx.post("https://api.openai.com/v1/chat/completions").mock(
        return_value=httpx.Response(401, json={"error": {"message": "Invalid key", "type": "authentication_error"}})
    )
    
    llm = ChatOpenAI(model="gpt-4o-mini", max_retries=3, request_timeout=5.0)
    
    with pytest.raises(AuthenticationError):
        llm.invoke([HumanMessage(content="test")])

@respx.mock
def test_bad_request_fails_fast():
    respx.post("https://api.openai.com/v1/chat/completions").mock(
        return_value=httpx.Response(400, json={"error": {"message": "Bad request", "type": "invalid_request_error"}})
    )
    
    llm = ChatOpenAI(model="gpt-4o-mini", max_retries=3, request_timeout=5.0)
    
    with pytest.raises(BadRequestError):
        llm.invoke([HumanMessage(content="test")])

@respx.mock
def test_server_error_retries_then_fails():
    route = respx.post("https://api.openai.com/v1/chat/completions").mock(
        return_value=httpx.Response(500, json={"error": {"message": "Internal error", "type": "server_error"}})
    )
    
    llm = ChatOpenAI(model="gpt-4o-mini", max_retries=2, request_timeout=5.0)
    
    with pytest.raises(InternalServerError):
        llm.invoke([HumanMessage(content="test")])
    
    # Initial + 2 retries = 3 calls
    assert route.call_count == 3

Run with pytest -v test_error_handling.py. Every test should pass.

Step 8: Clean up dead code and update documentation

After verification, remove the old manual retry functions, custom exception imports, and any wrapper classes that only existed to normalize errors. Update your internal docs to reference the new patterns:

• Retry policy: controlled by max_retries and request_timeout on ChatOpenAI/OpenAI
• Custom backoff: pass retry= dict with tenacity wait/stop/retry predicates
• Streaming errors: wrap stream()/astream() iteration in try/except; mid-stream failures require application-level resumption
• Fallbacks: use with_fallbacks([fallback_llm, ...], exception_types=(...))
• Observability: implement BaseCallbackHandler for structured logs

Verify: Grep your codebase for from openai import and openai. exception references — only the LangChain imports should remain in application code (tests may still import raw SDK for mocking).

The migration is complete when your test suite passes, logs show structured retry/fallback events, and no raw SDK error-handling code remains in production paths. LangChain’s abstraction means you write less boilerplate, but you still own the policy decisions — which exceptions trigger retries, how many attempts, and what happens when all providers fail.