Enable Qdrant vectorstore (#1076)

* enable qdrant vectorstore Signed-off-by: yuwenzho <yuwen.zhou@intel.com> Co-authored-by: XuhuiRen

Author: yuwenzho

intel_extension_for_transformers/langchain/vectorstores/__init__.py

@@ -16,3 +16,4 @@
 # limitations under the License.
 
 from .chroma import Chroma
+from .qdrant import Qdrant

intel_extension_for_transformers/langchain/vectorstores/qdrant.py

@@ -0,0 +1,290 @@
+# !/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2023 Intel Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import logging
+from typing import Any, Type, List, Optional, TYPE_CHECKING
+
+from langchain_core.documents import Document
+from langchain_core.embeddings import Embeddings
+from langchain.vectorstores.qdrant import Qdrant as Qdrant_origin
+from intel_extension_for_transformers.transformers.utils.utility import LazyImport
+
+logging.basicConfig(
+ format="%(asctime)s %(name)s:%(levelname)s:%(message)s",
+ datefmt="%d-%M-%Y %H:%M:%S",
+ level=logging.INFO
+)
+
+if TYPE_CHECKING:
+ from qdrant_client.conversions import common_types
+
+_DEFAULT_PERSIST_DIR = './output'
+
+qdrant_client = LazyImport("qdrant_client")
+
+class Qdrant(Qdrant_origin):
+
+ _LANGCHAIN_DEFAULT_COLLECTION_NAME = "langchain"
+ 
+ @classmethod
+ def from_documents(
+ cls,
+ documents: List[Document],
+ embedding: Embeddings,
+ sign: Optional[str] = None,
+ location: Optional[str] = None,
+ url: Optional[str] = None,
+ api_key: Optional[str] = None,
+ host: Optional[str]= None,
+ persist_directory: Optional[str] = None,
+ collection_name: Optional[str] = _LANGCHAIN_DEFAULT_COLLECTION_NAME,
+ force_recreate: Optional[bool] = False,
+ **kwargs: Any,
+ ):
+ """Create a Qdrant vectorstore from a list of documents.
+
+ Args:
+ documents (List[Document]): List of documents to add to the vectorstore.
+ embedding (Optional[Embeddings]): A subclass of `Embeddings`, responsible for text vectorization.
+ sign (Optional[str], optional): sign for retrieval_type of 'child_parent'. Defaults to None.
+ location (Optional[str], optional): 
+ If `:memory:` - use in-memory Qdrant instance.
+ If `str` - use it as a `url` parameter.
+ If `None` - fallback to relying on `host` and `port` parameters. 
+ Defaults to None.
+ url (Optional[str], optional): either host or str of "Optional[scheme], host, Optional[port],
+ Optional[prefix]". Defaults to None.
+ api_key (Optional[str], optional): API key for authentication in Qdrant Cloud. Defaults to None.
+ host (Optional[str], optional): Host name of Qdrant service. If url and host are None, set to
+ 'localhost'. Defaults to None.
+ persist_directory (Optional[str], optional): Path in which the vectors will be stored while using
+ local mode. Defaults to None.
+ collection_name (Optional[str], optional): Name of the Qdrant collection to be used. 
+ Defaults to _LANGCHAIN_DEFAULT_COLLECTION_NAME.
+ force_recreate (bool, optional): _description_. Defaults to False.
+ """
+ if sum([param is not None for param in (location, url, host, persist_directory)]) == 0:
+ # One of 'location', 'url', 'host' or 'persist_directory' should be specified.
+ persist_directory = _DEFAULT_PERSIST_DIR
+ if sign == "child":
+ persist_directory = persist_directory + "_child"
+ texts = [d.page_content for d in documents]
+ metadatas = [d.metadata for d in documents]
+ return cls.from_texts(
+ texts, 
+ embedding, 
+ metadatas=metadatas, 
+ location=location,
+ url=url,
+ api_key=api_key,
+ host=host,
+ path=persist_directory, 
+ collection_name=collection_name,
+ force_recreate=force_recreate,
+ **kwargs)
+ 
+ @classmethod
+ def build(
+ cls,
+ documents: List[Document],
+ embedding: Optional[Embeddings],
+ sign: Optional[str] = None,
+ location: Optional[str] = None,
+ url: Optional[str] = None,
+ api_key: Optional[str] = None,
+ host: Optional[str]= None,
+ persist_directory: Optional[str] = None,
+ collection_name: Optional[str] = _LANGCHAIN_DEFAULT_COLLECTION_NAME,
+ force_recreate: Optional[bool] = False,
+ **kwargs: Any,
+ ):
+ """Build a Qdrant vectorstore.
+
+ Args:
+ documents (List[Document]): List of documents to add to the vectorstore.
+ embedding (Optional[Embeddings]): A subclass of `Embeddings`, responsible for text vectorization.
+ sign (Optional[str], optional): sign for retrieval_type of 'child_parent'. Defaults to None.
+ location (Optional[str], optional): 
+ If `:memory:` - use in-memory Qdrant instance.
+ If `str` - use it as a `url` parameter.
+ If `None` - fallback to relying on `host` and `port` parameters. 
+ Defaults to None.
+ url (Optional[str], optional): either host or str of "Optional[scheme], host, Optional[port],
+ Optional[prefix]". Defaults to None.
+ api_key (Optional[str], optional): API key for authentication in Qdrant Cloud. Defaults to None.
+ host (Optional[str], optional): Host name of Qdrant service. If url and host are None, set to
+ 'localhost'. Defaults to None.
+ persist_directory (Optional[str], optional): Path in which the vectors will be stored while using
+ local mode. Defaults to None.
+ collection_name (Optional[str], optional): Name of the Qdrant collection to be used. 
+ Defaults to _LANGCHAIN_DEFAULT_COLLECTION_NAME.
+ force_recreate (bool, optional): _description_. Defaults to False.
+ kwargs: 
+ Current used:
+ port (Optional[int], optional): Port of the REST API interface. Defaults to 6333.
+ grpc_port (int, optional): Port of the gRPC interface. Defaults to 6334.
+ prefer_grpc (bool, optional): If true - use gPRC interface whenever possible in custom methods. 
+ Defaults to False.
+ https (Optional[bool], optional): If true - use HTTPS(SSL) protocol.
+ prefix (Optional[str], optional): 
+ If not None - add prefix to the REST URL path.
+ Example: service/v1 will result in
+ http://localhost:6333/service/v1/{qdrant-endpoint} for REST API.
+ timeout (Optional[float], optional): 
+ Timeout for REST and gRPC API requests.
+ 
+ distance_func (str, optional): Distance function. One of: "Cosine" / "Euclid" / "Dot".
+ Defaults to "Cosine".
+ content_payload_key (str, optional): A payload key used to store the content of the document. 
+ Defaults to CONTENT_KEY.
+ metadata_payload_key (str, optional): A payload key used to store the metadata of the document.
+ Defaults to METADATA_KEY.
+ vector_name (Optional[str], optional): Name of the vector to be used internally in Qdrant.
+ Defaults to VECTOR_NAME.
+ shard_number (Optional[int], optional): Number of shards in collection.
+ replication_factor (Optional[int], optional): 
+ Replication factor for collection.
+ Defines how many copies of each shard will be created.
+ Have effect only in distributed mode.
+ write_consistency_factor (Optional[int], optional): 
+ Write consistency factor for collection.
+ Defines how many replicas should apply the operation for us to consider
+ it successful. Increasing this number will make the collection more
+ resilient to inconsistencies, but will also make it fail if not enough
+ replicas are available.
+ Does not have any performance impact.
+ Have effect only in distributed mode.
+ on_disk_payload (Optional[bool], optional):
+ If true - point`s payload will not be stored in memory.
+ It will be read from the disk every time it is requested.
+ This setting saves RAM by (slightly) increasing the response time.
+ Note: those payload values that are involved in filtering and are
+ indexed - remain in RAM.
+ hnsw_config (Optional[common_types.HnswConfigDiff], optional): Params for HNSW index.
+ optimizers_config (Optional[common_types.OptimizersConfigDiff], optional): Params for optimizer.
+ wal_config (Optional[common_types.WalConfigDiff], optional): Params for Write-Ahead-Log.
+ quantization_config (Optional[common_types.QuantizationConfig], optional): 
+ Params for quantization, if None - quantization will be disable.
+ init_from (Optional[common_types.InitFrom], optional): 
+ Use data stored in another collection to initialize this collection.
+ on_disk (Optional[bool], optional): if True, vectors will be stored on disk.
+ If None, default value will be used.
+ """
+ if sum([param is not None for param in (location, url, host, persist_directory)]) == 0:
+ # One of 'location', 'url', 'host' or 'persist_directory' should be specified.
+ persist_directory = _DEFAULT_PERSIST_DIR
+ if sign == "child":
+ persist_directory = persist_directory + "_child"
+ if persist_directory and os.path.exists(persist_directory):
+ if bool(os.listdir(persist_directory)):
+ logging.info("Load the existing database!")
+ texts = [d.page_content for d in documents]
+ qdrant_collection = cls.construct_instance(
+ texts=texts,
+ embedding=embedding,
+ location=location,
+ url=url,
+ api_key=api_key,
+ host=host,
+ path=persist_directory,
+ collection_name=collection_name,
+ force_recreate=force_recreate,
+ **kwargs
+ )
+ return qdrant_collection
+ else:
+ logging.info("Create a new knowledge base...")
+ qdrant_collection = cls.from_documents(
+ documents=documents,
+ embedding=embedding,
+ location=location,
+ url=url,
+ api_key=api_key,
+ host=host,
+ persist_directory=persist_directory,
+ collection_name=collection_name,
+ force_recreate=force_recreate,
+ **kwargs,
+ )
+ return qdrant_collection
+ 
+ 
+ @classmethod
+ def reload(
+ cls,
+ embedding: Optional[Embeddings],
+ location: Optional[str] = None,
+ url: Optional[str] = None,
+ api_key: Optional[str] = None,
+ host: Optional[str]= None,
+ persist_directory: Optional[str] = None,
+ collection_name: Optional[str] = _LANGCHAIN_DEFAULT_COLLECTION_NAME,
+ force_recreate: bool = False,
+ **kwargs: Any,
+ ):
+ """Reload a Qdrant vectorstore.
+
+ Args:
+ embedding (Optional[Embeddings]): A subclass of `Embeddings`, responsible for text vectorization.
+ location (Optional[str], optional): 
+ If `:memory:` - use in-memory Qdrant instance.
+ If `str` - use it as a `url` parameter.
+ If `None` - fallback to relying on `host` and `port` parameters. 
+ Defaults to None.
+ url (Optional[str], optional): either host or str of "Optional[scheme], host, Optional[port],
+ Optional[prefix]". Defaults to None.
+ api_key (Optional[str], optional): API key for authentication in Qdrant Cloud. Defaults to None.
+ host (Optional[str], optional): Host name of Qdrant service. If url and host are None, set to
+ 'localhost'. Defaults to None.
+ persist_directory (Optional[str], optional): Path in which the vectors will be stored while using
+ local mode. Defaults to None.
+ collection_name (Optional[str], optional): Name of the Qdrant collection to be used. 
+ Defaults to _LANGCHAIN_DEFAULT_COLLECTION_NAME.
+ force_recreate (bool, optional): _description_. Defaults to False.
+ """
+ if sum([param is not None for param in (location, url, host, persist_directory)]) == 0:
+ # One of 'location', 'url', 'host' or 'persist_directory' should be specified.
+ persist_directory = _DEFAULT_PERSIST_DIR
+
+ # for a single quick embedding to get vector size
+ tmp_texts = ["foo"]
+ 
+ qdrant_collection = cls.construct_instance(
+ texts=tmp_texts,
+ embedding=embedding,
+ location=location,
+ url=url,
+ api_key=api_key,
+ host=host,
+ path=persist_directory,
+ collection_name=collection_name,
+ force_recreate=force_recreate,
+ **kwargs
+ )
+ return qdrant_collection
+ 
+ 
+ def is_local(
+ self,
+ ):
+ """Determine whether a client is local."""
+ if hasattr(self.client, "_client") and \
+ isinstance(self.client._client, qdrant_client.local.qdrant_local.QdrantLocal):
+ return True
+ else:
+ return False

intel_extension_for_transformers/neural_chat/pipeline/plugins/retrieval/retrieval_agent.py

@@ -26,7 +26,7 @@
  HuggingFaceInstructEmbeddings, HuggingFaceBgeEmbeddings
 from langchain.embeddings import GooglePalmEmbeddings
 from langchain.text_splitter import RecursiveCharacterTextSplitter
-from intel_extension_for_transformers.langchain.vectorstores import Chroma
+from intel_extension_for_transformers.langchain.vectorstores import Chroma, Qdrant
 import uuid
 from langchain_core.documents import Document
 import logging
@@ -133,10 +133,12 @@ def __init__(self,
  logging.info("The format of parsed documents is transferred.")
 
  if self.vector_database == "Chroma":
- self.database = Chroma()
+ self.database = Chroma
+ elif self.vector_database == "Qdrant":
+ self.database = Qdrant
  # elif self.vector_database == "PGVector":
  # self.database = PGVector()
- 
+
  if self.retrieval_type == 'default': # Using vector store retriever
  if append:
  knowledge_base = self.database.from_documents(documents=langchain_documents, embedding=self.embeddings,
@@ -145,6 +147,9 @@ def __init__(self,
  knowledge_base = self.database.build(documents=langchain_documents, embedding=self.embeddings, **kwargs)
  self.retriever = RetrieverAdapter(retrieval_type=self.retrieval_type, document_store=knowledge_base, \
  **kwargs).retriever
+ if self.vector_database == "Qdrant" and knowledge_base.is_local():
+ # one local storage folder cannot be accessed by multiple instances of Qdrant client simultaneously.
+ knowledge_base.client.close()
  elif self.retrieval_type == "child_parent": # Using child-parent store retriever
  child_documents = self.splitter.split_documents(langchain_documents)
  if append:
@@ -158,6 +163,12 @@ def __init__(self,
  sign='child', **kwargs)
  self.retriever = RetrieverAdapter(retrieval_type=self.retrieval_type, document_store=knowledge_base, \
  child_document_store=child_knowledge_base, **kwargs).retriever
+ if self.vector_database == "Qdrant" :
+ # one local storage folder cannot be accessed by multiple instances of Qdrant client simultaneously.
+ if knowledge_base.is_local():
+ knowledge_base.client.close()
+ if child_knowledge_base.is_local():
+ child_knowledge_base.client.close()
  logging.info("The retriever is successfully built.")
 
  def reload_localdb(self, local_persist_dir, **kwargs):

intel_extension_for_transformers/neural_chat/requirements.txt

@@ -71,3 +71,4 @@ urllib3
 langid
 diffusers==0.12.1
 transformers_stream_generator
+qdrant-client

intel_extension_for_transformers/neural_chat/requirements_cpu.txt

@@ -48,4 +48,5 @@ einops
 cchardet
 zhconv
 urllib3
-langid
+langid
+qdrant-client

intel_extension_for_transformers/neural_chat/requirements_hpu.txt

@@ -43,3 +43,4 @@ einops
 zhconv
 urllib3
 langid
+qdrant-client

intel_extension_for_transformers/neural_chat/requirements_pc.txt

@@ -46,3 +46,4 @@ langid
 pymysql
 deepface
 exifread
+qdrant-client

intel_extension_for_transformers/neural_chat/requirements_xpu.txt

@@ -39,3 +39,4 @@ exifread
 zhconv
 urllib3
 langid
+qdrant-client

intel_extension_for_transformers/neural_chat/tests/ci/api/test_chatbot_build_api.py

@@ -150,6 +150,7 @@ def test_build_chatbot_with_retrieval_plugin_bge_int8(self):
  def test_build_chatbot_with_retrieval_plugin_using_local_file(self):
 
  def _run_retrieval(local_dir):
+ plugins.tts.enable = False
  plugins.retrieval.enable = True
  plugins.retrieval.args["input_path"] = "../../../README.md"
  plugins.retrieval.args["embedding_model"] = local_dir