langchain/vectorstores/milvus.py/Milvus/__init__
class Milvus:    def __init__(
        self,
        embedding_function: Embeddings,
        collection_name: str = "LangChainCollection",
        connection_args: Optional[dict[str, Any]] = None,
        consistency_level: str = "Session",
        index_params: Optional[dict] = None,
        search_params: Optional[dict] = None,
        drop_old: Optional[bool] = False,
    ):
        """Initialize wrapper around the milvus vector database.

        In order to use this you need to have `pymilvus` installed and a
        running Milvus/Zilliz Cloud instance.

        See the following documentation for how to run a Milvus instance:
        https://milvus.io/docs/install_standalone-docker.md

        If looking for a hosted Milvus, take a looka this documentation:
        https://zilliz.com/cloud

        IF USING L2/IP metric IT IS HIGHLY SUGGESTED TO NORMALIZE YOUR DATA.

        The connection args used for this class comes in the form of a dict,
        here are a few of the options:
            address (str): The actual address of Milvus
                instance. Example address: "localhost:19530"
            uri (str): The uri of Milvus instance. Example uri:
                "http://randomwebsite:19530",
                "tcp:foobarsite:19530",
                "https://ok.s3.south.com:19530".
            host (str): The host of Milvus instance. Default at "localhost",
                PyMilvus will fill in the default host if only port is provided.
            port (str/int): The port of Milvus instance. Default at 19530, PyMilvus
                will fill in the default port if only host is provided.
            user (str): Use which user to connect to Milvus instance. If user and
                password are provided, we will add related header in every RPC call.
            password (str): Required when user is provided. The password
                corresponding to the user.
            secure (bool): Default is false. If set to true, tls will be enabled.
            client_key_path (str): If use tls two-way authentication, need to
                write the client.key path.
            client_pem_path (str): If use tls two-way authentication, need to
                write the client.pem path.
            ca_pem_path (str): If use tls two-way authentication, need to write
                the ca.pem path.
            server_pem_path (str): If use tls one-way authentication, need to
                write the server.pem path.
            server_name (str): If use tls, need to write the common name.

        Args:
            embedding_function (Embeddings): Function used to embed the text.
            collection_name (str): Which Milvus collection to use. Defaults to
                "LangChainCollection".
            connection_args (Optional[dict[str, any]]): The arguments for connection to
                Milvus/Zilliz instance. Defaults to DEFAULT_MILVUS_CONNECTION.
            consistency_level (str): The consistency level to use for a collection.
                Defaults to "Session".
            index_params (Optional[dict]): Which index params to use. Defaults to
                HNSW/AUTOINDEX depending on service.
            search_params (Optional[dict]): Which search params to use. Defaults to
                default of index.
            drop_old (Optional[bool]): Whether to drop the current collection. Defaults
                to False.
        """
        try:
            from pymilvus import Collection, utility
        except ImportError:
            raise ValueError(
                "Could not import pymilvus python package. "
                "Please install it with `pip install pymilvus`."
            )

        # Default search params when one is not provided.
        self.default_search_params = {
            "IVF_FLAT": {"metric_type": "L2", "params": {"nprobe": 10}},
            "IVF_SQ8": {"metric_type": "L2", "params": {"nprobe": 10}},
            "IVF_PQ": {"metric_type": "L2", "params": {"nprobe": 10}},
            "HNSW": {"metric_type": "L2", "params": {"ef": 10}},
            "RHNSW_FLAT": {"metric_type": "L2", "params": {"ef": 10}},
            "RHNSW_SQ": {"metric_type": "L2", "params": {"ef": 10}},
            "RHNSW_PQ": {"metric_type": "L2", "params": {"ef": 10}},
            "IVF_HNSW": {"metric_type": "L2", "params": {"nprobe": 10, "ef": 10}},
            "ANNOY": {"metric_type": "L2", "params": {"search_k": 10}},
            "AUTOINDEX": {"metric_type": "L2", "params": {}},
        }

        self.embedding_func = embedding_function
        self.collection_name = collection_name
        self.index_params = index_params
        self.search_params = search_params
        self.consistency_level = consistency_level

        # In order for a collection to be compatible, pk needs to be auto'id and int
        self._primary_field = "pk"
        # In order for compatiblility, the text field will need to be called "text"
        self._text_field = "text"
        # In order for compatbility, the vector field needs to be called "vector"
        self._vector_field = "vector"
        self.fields: list[str] = []
        # Create the connection to the server
        if connection_args is None:
            connection_args = DEFAULT_MILVUS_CONNECTION
        self.alias = self._create_connection_alias(connection_args)
        self.col: Optional[Collection] = None

        # Grab the existing colection if it exists
        if utility.has_collection(self.collection_name, using=self.alias):
            self.col = Collection(
                self.collection_name,
                using=self.alias,
            )
        # If need to drop old, drop it
        if drop_old and isinstance(self.col, Collection):
            self.col.drop()
            self.col = None

        # Initialize the vector store
        self._init()