Adding Openai v1.0.0+ support into data index (#33440)

* changes in ai-generative index

* changes in resources.

* spelling

* adding tqdm

* remove tqdm from required installation.

* fix "Unknown word ( )" issue

* Fixing Cspell complains

Author: shpengmsft

.vscode/cspell.json

@@ -33,6 +33,7 @@
     "sdk/batch/azure-batch/**",
     "sdk/ai/**/index/**",
     "sdk/ai/azure-ai-generative/tests/**",
+    "sdk/ai/azure-ai-resources/azure/ai/resources/_index/_langchain/vendor/**",
     "sdk/cognitiveservices/azure-cognitiveservices-search-autosuggest/**",
     "sdk/cognitiveservices/azure-cognitiveservices-search-customimagesearch/**",
     "sdk/cognitiveservices/azure-cognitiveservices-search-customsearch/**",
@@ -1412,7 +1413,16 @@
     {
       "filename": "sdk/ai/azure-ai-resources/**",
       "words": [
-        "etest"
+        "azuread",
+        "all-mpnet-base-v2",
+        "dtype",
+        "LLMRAG",
+        "pipelinerunid",
+        "pydantic",
+        "swigfaiss",
+        "tiktokens",
+        "tqdm",
+        "wasb"
       ]
     },
     {

sdk/ai/azure-ai-generative/azure/ai/generative/index/_embeddings/__init__.py

@@ -42,14 +42,17 @@ def _args_to_openai_embedder(arguments: dict):
 
     if langchain_version > "0.0.154":
         embedder = OpenAIEmbeddings(
-            openai_api_base=arguments.get("api_base", openai.api_base),
+            openai_api_base=arguments.get("api_base", openai.api_base if hasattr(openai, "api_base") else openai.base_url),
             openai_api_type=arguments.get("api_type", openai.api_type),
             openai_api_version=arguments.get("api_version", openai.api_version),
             openai_api_key=arguments.get("api_key", openai.api_key),
             max_retries=100,  # TODO: Make this configurable
         )
     else:
-        openai.api_base = arguments.get("api_base", openai.api_base)
+        if hasattr(openai, "api_base"):
+            openai.api_base = arguments.get("api_base", openai.api_base)
+        else:
+            openai.base_url = arguments.get("api_base", openai.base_url)
         openai.api_type = arguments.get("api_type", openai.api_type)
         openai.api_version = arguments.get("api_version", openai.api_version)
         embedder = OpenAIEmbeddings(
@@ -85,9 +88,10 @@ def get_langchain_embeddings(embedding_kind: str, arguments: dict, credential: O
 
         arguments = init_open_ai_from_config(arguments, credential=credential)
 
+        # In openai v1.0.0 and above, openai.api_base is replaced by openai.base_url
         embedder = OpenAIEmbedder(
             model=arguments.get("model"),
-            api_base=arguments.get("api_base", openai.api_base),
+            api_base=arguments.get("api_base", openai.api_base if hasattr(openai, "api_base") else openai.base_url),
             api_type=arguments.get("api_type", openai.api_type),
             api_version=arguments.get("api_version", openai.api_version),
             api_key=arguments.get("api_key", openai.api_key),
@@ -171,9 +175,10 @@ def get_embed_fn(embedding_kind: str, arguments: dict, credential: Optional[Toke
 
         arguments = init_open_ai_from_config(arguments, credential=credential)
 
+        # In openai v1.0.0 and above, openai.api_base is replaced by openai.base_url
         embedder = OpenAIEmbedder(
             model=arguments.get("model"),
-            api_base=arguments.get("api_base", openai.api_base),
+            api_base=arguments.get("api_base", openai.api_base if hasattr(openai, "api_base") else openai.base_url),
             api_type=arguments.get("api_type", openai.api_type),
             api_version=arguments.get("api_version", openai.api_version),
             api_key=arguments.get("api_key", openai.api_key),
@@ -228,7 +233,7 @@ def get_query_embed_fn(embedding_kind: str, arguments: dict, credential: Optiona
 
         embedder = OpenAIEmbedder(
             model=arguments.get("model"),
-            api_base=arguments.get("api_base", openai.api_base),
+            api_base=arguments.get("api_base", openai.api_base if hasattr(openai, "api_base") else openai.base_url),
             api_type=arguments.get("api_type", openai.api_type),
             api_version=arguments.get("api_version", openai.api_version),
             api_key=arguments.get("api_key", openai.api_key),
@@ -332,7 +337,9 @@ def get_embeddings(self) -> str:
     def open_embedding_file(cls, path) -> pa.Table:
         """Open the embedding file and cache it."""
         if cls._last_opened_embeddings is None or cls._last_opened_embeddings[0] != path:
-            logger.debug(f"caching embeddings file: \n{path}\n   previous path cached was: \n{cls._last_opened_embeddings}")
+            logger.debug(
+                f"caching embeddings file: \n{path}\n   previous path cached was: \n{cls._last_opened_embeddings}"
+            )
             table = pq.read_table(path)
             cls._last_opened_embeddings = (path, table)
 
@@ -460,7 +467,7 @@ def get_metadata(self):
         if "open_ai" in self.kind:
             if "api_base" not in arguments:
                 import openai
-                arguments["api_base"] = openai.api_base
+                arguments["api_base"] = openai.api_base if hasattr(openai, "api_base") else openai.base_url
             if "api_key" in arguments:
                 del arguments["api_key"]
             if "key" in arguments:

sdk/ai/azure-ai-generative/azure/ai/generative/index/_embeddings/openai.py

@@ -2,10 +2,12 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
 """OpenAI Embeddings generation and management tools."""
+import os
 import time
 from typing import Any, Dict, List, Optional
 
 from azure.ai.generative.index._utils.logging import get_logger
+from packaging import version
 
 logger = get_logger("embeddings.openai")
 
@@ -17,7 +19,7 @@ def __init__(
         self,
         api_base: str,
         api_type: str,
-        api_version: str = "2023-03-15-preview",
+        api_version: str = None,
         api_key: Optional[str] = None,
         azure_credential: Optional[Any] = None,
         model: str = "text-embedding-ada-002",
@@ -31,8 +33,7 @@ def __init__(
         """Initialize an OpenAI Embedding client."""
         self.api_base = api_base
         self.api_type = api_type
-        self.api_version = api_version
-        self.api_key = api_key
+        self.api_key = api_key or os.getenv("AZURE_OPENAI_KEY") or ""
         # TODO: If azure_credential set, check api_type is azure or azure_ad and setup auth accordingly
         self.azure_credential = azure_credential
 
@@ -44,7 +45,7 @@ def __init__(
         self._dynamic_batch_size = None
 
         if max_retries is None:
-            max_retries = 20
+            max_retries = 10
         self.max_retries = max_retries
 
         if model is None:
@@ -60,14 +61,67 @@ def __init__(
         self.embedding_ctx_length = embedding_ctx_length
 
         self.show_progress_bar = show_progress_bar
+        self.openai_passthrough_args = openai_passthrough_args or {}
 
         try:
             import openai
         except ImportError as e:
             raise ImportError("Please install openai via `pip install openai`") from e
 
-        self.openai_passthrough_args = openai_passthrough_args or {}
-        self.embedding_client = openai.Embedding
+        if version.parse(openai.version.VERSION) >= version.parse("1.0.0"):
+            self.openai_v1plus = True
+            self.api_version = api_version if api_version else "2023-05-15"
+
+            if "azure" in self.api_type:
+                client = openai.AzureOpenAI(
+                    api_key=self.api_key,
+                    api_version=self.api_version,
+                    azure_endpoint=self.api_base,
+                )
+            else:
+                client = openai.OpenAI(
+                    api_key=self.api_key,
+                    base_url=self.api_base,
+                )
+
+            self.embedding_client = client.embeddings
+
+            self._params = {
+                "model": self.model,
+                **self.openai_passthrough_args,
+            }
+            self._retry_exceptions = [
+                openai._exceptions.APIStatusError,
+                openai._exceptions.APITimeoutError,
+                openai._exceptions.APIError,
+                openai._exceptions.APIConnectionError,
+                openai._exceptions.RateLimitError,
+                openai._exceptions.InternalServerError,
+                openai._exceptions.APIResponseValidationError,
+            ]
+            self._RateLimitError = openai._exceptions.RateLimitError
+        else:
+            self.openai_v1plus = False
+            self.api_version = api_version if api_version else "2023-03-15-preview"
+            self.embedding_client = openai.Embeddings
+            self._params = {
+                "model": self.model,
+                "api_base": self.api_base,
+                "api_type": self.api_type,
+                "api_version": self.api_version,
+                "api_key": self.api_key,
+                **self.openai_passthrough_args,
+            }
+            if self.deployment is not None:
+                self._params["engine"] = self.deployment
+            self._retry_exceptions = [
+                openai.error.Timeout,
+                openai.error.APIError,
+                openai.error.APIConnectionError,
+                openai.error.RateLimitError,
+                openai.error.ServiceUnavailableError,
+            ]
+            self._RateLimitError = openai.error.RateLimitError
 
         self._statistics = {
             "num_retries": 0,
@@ -77,28 +131,11 @@ def __init__(
 
     @property
     def _openai_client_params(self) -> dict:
-        params = {
-            "model": self.model,
-            "api_base": self.api_base,
-            "api_type": self.api_type,
-            "api_version": self.api_version,
-            "api_key": self.api_key,
-            **self.openai_passthrough_args,
-        }
-        if self.deployment is not None:
-            params["engine"] = self.deployment
-        return params
+        return self._params
 
     @property
     def _retryable_openai_errors(self) -> List[Exception]:
-        import openai
-        return [
-            openai.error.Timeout,
-            openai.error.APIError,
-            openai.error.APIConnectionError,
-            openai.error.RateLimitError,
-            openai.error.ServiceUnavailableError,
-        ]
+        return self._retry_exceptions
 
     def _dynamic_batch_size_embed_request(self, tokenized_texts: List[List[int]], **kwargs) -> dict:
         try:
@@ -142,17 +179,18 @@ def _dynamic_batch_size_embed_request(self, tokenized_texts: List[List[int]], **
 
     def _embed_request(self, tokenized_texts: List[List[int]], **kwargs) -> dict:
         try:
-            min_seconds = 4
-            max_seconds = 10
             total_delay = 0
             last_exception = None
             for retry in range(self.max_retries):
                 logger.info(f"Attempt {retry} to embed {len(tokenized_texts)} documents.")
                 try:
-                    return self.embedding_client.create(
+                    response = self.embedding_client.create(
                         input=tokenized_texts,
                         **kwargs,
                     )
+                    if self.openai_v1plus:
+                        response = {"object": "list", "data": [{"object": "embedding", "embedding": d.embedding} for d in response.data]}
+                    return response
                 except Exception as e:
                     err_msg = str(e)
                     logger.warning(f"Error embedding: {err_msg}", exc_info=e)
@@ -161,16 +199,18 @@ def _embed_request(self, tokenized_texts: List[List[int]], **kwargs) -> dict:
                     for retryable_error in self._retryable_openai_errors:
                         if isinstance(e, retryable_error):
                             retrying = True
-                            import openai
-
-                            # Retry with retry-after set by openai for RateLimitError
-                            if isinstance(e, openai.error.RateLimitError) and "Retry-After" in e.headers:
-                                delay = int(e.headers["Retry-After"])
-                                logger.warning(f"OpenAI throws RateLimitError with Retry-After set to {delay}")
-                            # Retry with exponential backoff
-                            else:
-                                exp = 2 ** (retry - 1)
-                                delay = max(min(1 * exp, max_seconds), min_seconds)
+
+                            # Retry with retry-after if found in RateLimitError
+                            if isinstance(e, self._RateLimitError):
+                                logger.warning(f"Retrying error type {type(e)}.")
+                                response_headers = e.headers if hasattr(e, "headers") else {}
+                                if "Retry-After" in response_headers:
+                                    delay = int(response_headers["Retry-After"])
+                                    logger.warning(f"OpenAI throws RateLimitError with Retry-After {delay} seconds.")
+                                else:
+                                    # Wait for 1 minute as suggested by openai https://help.openai.com/en/articles/6897202-ratelimiterror
+                                    logger.warning("Retry after 60 seconds.")
+                                    delay = 60
                             total_delay += delay
                             logger.warning(f"Sleeping for {delay} seconds before retrying.")
                             time.sleep(delay)

sdk/ai/azure-ai-generative/azure/ai/generative/index/_langchain/openai.py

@@ -55,6 +55,25 @@ def __call__(self, retry_state) -> bool:
     # Copied from https://github.com/hwchase17/langchain/blob/511c12dd3985ce682226371c12f8fa70d8c9a8e1/langchain/embeddings/openai.py#L34
     def _create_retry_decorator(embeddings):
         import openai
+        from packaging import version
+
+        if version.parse(openai.version.VERSION) >= version.parse("1.0.0"):
+            retry_exceptions = (
+                retry_if_exception_type(openai._exceptions.APITimeoutError)
+                | retry_if_exception_type(openai._exceptions.APIError)
+                | retry_if_exception_type(openai._exceptions.APIConnectionError)
+                | retry_if_exception_type(openai._exceptions.RateLimitError)
+                | retry_if_exception_type(openai._exceptions.InternalServerError)
+                | retry_if_exception_type(openai._exceptions.APIResponseValidationError)
+            )
+        else:
+            retry_exceptions = (
+                retry_if_exception_type(openai.error.Timeout)
+                | retry_if_exception_type(openai.error.APIError)
+                | retry_if_exception_type(openai.error.APIConnectionError)
+                | retry_if_exception_type(openai.error.RateLimitError)
+                | retry_if_exception_type(openai.error.ServiceUnavailableError)
+            )
 
         min_seconds = 4
         max_seconds = 10
@@ -65,13 +84,7 @@ def _create_retry_decorator(embeddings):
             # stop=stop_after_attempt(embeddings.max_retries),
             stop=stop_after_delay_that_works(max_seconds_retrying, activity_logger),
             wait=wait_exponential(multiplier=1, min=min_seconds, max=max_seconds),
-            retry=(
-                retry_if_exception_type(openai.error.Timeout)
-                | retry_if_exception_type(openai.error.APIError)
-                | retry_if_exception_type(openai.error.APIConnectionError)
-                | retry_if_exception_type(openai.error.RateLimitError)
-                | retry_if_exception_type(openai.error.ServiceUnavailableError)
-            ),
+            retry=retry_exceptions,
             before_sleep=_log_it,
         )
 

sdk/ai/azure-ai-generative/azure/ai/generative/index/_mlindex.py

@@ -155,6 +155,8 @@ def as_langchain_vectorstore(self, credential: Optional[TokenCredential] = None)
             activity_logger.activity_info["embeddings_kind"] = self.embeddings_config.get("kind", "none")
             activity_logger.activity_info["embeddings_api_type"] = self.embeddings_config.get("api_type", "none")
 
+            langchain_pkg_version = pkg_version.parse(langchain_version)
+
             if index_kind == "acs":
                 from azure.ai.generative.index._indexes.azure_search import import_azure_search_or_so_help_me
 
@@ -252,6 +254,11 @@ def as_langchain_vectorstore(self, credential: Optional[TokenCredential] = None)
                         self.embeddings_config.copy()
                     ).as_langchain_embeddings(credential=credential)
 
+
+                    # langchain fix https://github.com/langchain-ai/langchain/pull/10823 released in 0.0.318
+                    if langchain_pkg_version >= pkg_version.parse("0.0.318"):
+                        embeddings = embeddings.embed_query
+
                     fs, uri = url_to_fs(self.base_uri)
 
                     with tempfile.TemporaryDirectory() as tmpdir:

sdk/ai/azure-ai-generative/azure/ai/generative/index/_models.py

@@ -120,7 +120,7 @@ def init_open_ai_from_config(config: dict, credential: Optional[TokenCredential]
             logger.warning(f"Failed to get credential for ACS with {e}, falling back to env vars.")
             config["api_key"] = os.environ["OPENAI_API_KEY"]
             config["api_type"] = os.environ.get("OPENAI_API_TYPE", "azure")
-            config["api_base"] = os.environ.get("OPENAI_API_BASE", openai.api_base)
+            config["api_base"] = os.environ.get("OPENAI_API_BASE", openai.api_base if hasattr(openai, "api_base") else openai.base_url)
             config["api_version"] = os.environ.get("OPENAI_API_VERSION", openai.api_version)
         else:
             raise e

sdk/ai/azure-ai-generative/azure/ai/generative/index/_tasks/generate_qa.py

@@ -72,7 +72,11 @@ def get_model_config(llm_config: Dict[str, str], openai_api_type: str, openai_ap
 
     # Only add base and version if using AOAI
     if model_config["api_type"] == "azure":
-        openai.api_base = model_config["api_base"]
+        # openai.api_base is replaced by openai.base_url in openai 1.x
+        if hasattr(openai, "api_base"):
+            openai.api_base = model_config["api_base"]
+        else:
+            openai.base_url = model_config["api_base"]
         openai.api_version = model_config["api_version"]
     return model_config
 

sdk/ai/azure-ai-generative/azure/ai/generative/index/_utils/deployment.py

@@ -22,11 +22,15 @@ def infer_deployment(aoai_connection, model_name):
     openai.api_version = connection_metadata.get(
         "ApiVersion", connection_metadata.get("apiVersion", "2023-03-15-preview")
     )
-    openai.api_base = get_target_from_connection(aoai_connection)
+    api_base = get_target_from_connection(aoai_connection)
+    if hasattr(openai, "api_base"):
+        openai.api_base = api_base
+    else:
+        openai.base_url = api_base
     credential = connection_to_credential(aoai_connection)
     openai.api_key = credential.key if isinstance(credential, AzureKeyCredential) else credential.get_token().token
     deployment_list = convert_to_dict(
-        Deployment.list(api_key=openai.api_key, api_base=openai.api_base, api_type=openai.api_type)
+        Deployment.list(api_key=openai.api_key, api_base=api_base, api_type=openai.api_type)
     )
     for deployment in deployment_list["data"]:
         if deployment["model"] == model_name: