def _check_response(response: dict, skip_empty: bool = False) -> dict:
    if any(len(d["embedding"]) == 1 for d in response["data"]) and not skip_empty:
        import openai
        raise openai.error.APIError("OpenAI API returned an empty embedding")
    return response
[docs]def embed_with_retry(embeddings: OpenAIEmbeddings, **kwargs: Any) -> Any:
    """Use tenacity to retry the embedding call."""
    if _is_openai_v1():
        return embeddings.client.create(**kwargs)
    retry_decorator = _create_retry_decorator(embeddings)
    @retry_decorator
    def _embed_with_retry(**kwargs: Any) -> Any:
        response = embeddings.client.create(**kwargs)
        return _check_response(response, skip_empty=embeddings.skip_empty)
    return _embed_with_retry(**kwargs)
[docs]async def async_embed_with_retry(embeddings: OpenAIEmbeddings, **kwargs: Any) -> Any:
    """Use tenacity to retry the embedding call."""
    if _is_openai_v1():
        return await embeddings.async_client.create(**kwargs)
    @_async_retry_decorator(embeddings)
    async def _async_embed_with_retry(**kwargs: Any) -> Any:
        response = await embeddings.client.acreate(**kwargs)
        return _check_response(response, skip_empty=embeddings.skip_empty)
    return await _async_embed_with_retry(**kwargs)
def _is_openai_v1() -> bool:
    _version = parse(version("openai"))
    return _version >= Version("1.0.0")
[docs]class OpenAIEmbeddings(BaseModel, Embeddings):
    """OpenAI embedding models.
    To use, you should have the ``openai`` python package installed, and the