Source code for langchain_community.embeddings.openai
from __future__ import annotations
import logging
import os
import warnings
from importlib.metadata import version
from typing import (
    Any,
    Callable,
    Dict,
    List,
    Literal,
    Mapping,
    Optional,
    Sequence,
    Set,
    Tuple,
    Union,
    cast,
)
import numpy as np
from langchain_core.embeddings import Embeddings
from langchain_core.pydantic_v1 import BaseModel, Extra, Field, root_validator
from langchain_core.utils import get_from_dict_or_env, get_pydantic_field_names
from packaging.version import Version, parse
from tenacity import (
    AsyncRetrying,
    before_sleep_log,
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)
logger = logging.getLogger(__name__)
def _create_retry_decorator(embeddings: OpenAIEmbeddings) -> Callable[[Any], Any]:
    import openai
    # Wait 2^x * 1 second between each retry starting with
    # retry_min_seconds seconds, then up to retry_max_seconds seconds,
    # then retry_max_seconds seconds afterwards
    # retry_min_seconds and retry_max_seconds are optional arguments of
    # OpenAIEmbeddings
    return retry(
        reraise=True,
        stop=stop_after_attempt(embeddings.max_retries),
        wait=wait_exponential(
            multiplier=1,
            min=embeddings.retry_min_seconds,
            max=embeddings.retry_max_seconds,
        ),
        retry=(
            retry_if_exception_type(openai.error.Timeout)
            | retry_if_exception_type(openai.error.APIError)
            | retry_if_exception_type(openai.error.APIConnectionError)