Milvus Insert

Inputs

Outputs

Source

python

# Input preparation
inp = __ivy_ctx__["nodes"][__ivy_node_id__]["input"]
milvus_client = inp["milvus_client"]
collection_name = inp["collection_name"]
text_lines = inp["text_lines"]
text_vectorizer = inp["text_vectorizer"]

# Compute
from typing import List
from pymilvus import MilvusClient

class MilvusInsert:
    """
    Handles inserting text embeddings into a Milvus vector database.

    This class is responsible for:
    - Managing a Milvus collection.
    - Generating text embeddings using a provided text vectorizer.
    - Inserting the embeddings into the Milvus database.
    """

    def __init__(self, milvus_client: MilvusClient, collection_name: str, text_vectorizer: TextVectorizer):
        """
        Initializes the MilvusInsert instance with a Milvus client, collection name, and a text vectorizer.

        :param milvus_client: An instance of MilvusClient for database operations.
        :param collection_name: The name of the Milvus collection to store embeddings.
        :param text_vectorizer: A text vectorizer instance responsible for generating embeddings.
        """
        self.milvus_client = milvus_client
        self.collection_name = collection_name
        self.text_vectorizer = text_vectorizer
        # Determine embedding dimension once during initialization
        self.embedding_dim = len(self.text_vectorizer.vectorizer_text("This is a test"))

    def insert(self, text_lines: List[str]) -> None:
        """
        Inserts text embeddings into the Milvus collection.

        - If the collection already exists, it will be dropped and recreated.
        - Converts text into embeddings using the provided text vectorizer.
        - Stores the embeddings in the Milvus database.

        :param text_lines: A list of text strings to be converted into embeddings and stored.
        :raises Exception: If an unexpected error occurs during processing.
        """
        try:
            # Check if the collection exists and drop it if it does
            if self.milvus_client.has_collection(self.collection_name):
                try:
                    self.milvus_client.drop_collection(self.collection_name)
                except Exception as e:
                    print(f"Error dropping collection {e}: {e}")

            # Create a new collection with the pre-determined embedding dimension
            self.milvus_client.create_collection(
                collection_name=self.collection_name,
                dimension=self.embedding_dim,
                metric_type="IP",
                consistency_level="Strong",
            )

            # Generate embeddings and insert into Milvus
            data = self.text_vectorizer.vectorizer_texts(text_lines)
            self.milvus_client.insert(collection_name=self.collection_name, data=data)
            print(f"Successfully inserted {len(data)} embeddings into Milvus.")

        except Exception as e:
            print(f"Unexpected error: {e}")

milvus_insert = MilvusInsert(milvus_client=milvus_client, collection_name=collection_name, text_vectorizer=text_vectorizer)

milvus_insert.insert(text_lines=text_lines)

# Output collection (runner reads __ivy_ctx__)
out = __ivy_ctx__["nodes"][__ivy_node_id__]["output"]

Tests

Requires: python:3.11

• basic-insert

  Insert embeddings with valid inputs (should succeed).

  insertbasic
• error-empty-text-lines

  Empty text_lines should handle gracefully.

  error-handlingempty-input