Getting Error while performing Hybrid search using Milvus

[Preet1234-ux]

I am trying to perform Hybrid search in milvus using the below code :

`from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_milvus.retrievers import MilvusCollectionHybridSearchRetriever
from langchain_milvus.utils.sparse import BM25SparseEmbedding
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
import torch
from dotenv import load_dotenv
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from database_operation import initialise_milvus, data_ingest_milvus_db, data_load_from_milvus
import time
from groq import Groq
import os, logging
from langchain_community.document_loaders import UnstructuredPDFLoader
from bs4 import BeautifulSoup
from pathlib import Path
from pymilvus import (
Collection,
CollectionSchema,
DataType,
FieldSchema,
WeightedRanker,
connections,
)

CONNECTION_URI = "http://localhost:19530"
#export OPENAI_API_KEY=<your_api_key>

folder_path = "/home/rndadmin/dev/test/"

def preprocess_document(docs):
table_fragments = {}
data_string = ''

for doc in docs:
    if doc.page_content:
        # Extract and process HTML table fragments
        if doc.metadata.get("text_as_html", ''):
            table_data = doc.metadata.get("text_as_html", '')
            # Parse the HTML
            soup = BeautifulSoup(table_data, 'html.parser')
            
            # Remove style attributes
            for tag in soup.find_all(style=True):
                del tag['style']
            # Remove extra white spaces
            for tag in soup.find_all(["th", "td"]):
                if tag.string:
                    tag.string = tag.string.strip()
            
            # Get table ID or another unique identifier
            table_id = soup.find('table')['id'] if soup.find('table') else 'unknown'
            cleaned_table = str(soup.table)
            
            if table_id not in table_fragments:
                table_fragments[table_id] = []
            table_fragments[table_id].append(cleaned_table)
            
        data_string += doc.page_content + "\n\n"

# Combine table fragments
combined_tables = []
for table_id, fragments in table_fragments.items():
    combined_table = ''.join(fragments)
    combined_tables.append(combined_table)

# Create final document with combined tables
final_doc = docs[0]
final_doc.page_content = data_string + "\n\n".join(combined_tables)
source = docs[0].metadata.get("source", '')

if source:
    file_path = Path(source)
    file_name = file_path.name  # storing only filename
    final_doc.metadata = {"source": file_name} if file_name else {"source": source}
else:
    final_doc.metadata = {"source": source}

final_doc = [final_doc]
print(final_doc)
return final_doc

def get_documents(folder_path):
try:
documents = []

    files = [ext for ext in os.listdir(folder_path)]
    print("length of the file: {}".format(len(files)))

    for file in files:    
        if ('.pdf' in file ) and not file.startswith("~$"):
            full_path = os.path.join(folder_path, file)
            pdf_loader = UnstructuredPDFLoader(full_path, mode="elements", strategy="fast",)
            print("Processing this file-> {}".format(full_path))
            pdf_document = pdf_loader.load()
            pdf_document= preprocess_document(pdf_document)
            documents += pdf_document
 
    return documents
except Exception as ex:
    print("issue while processing this file :{}:: {}".format(full_path, ex))

def load_and_process_documents():
documents = get_documents(folder_path)
print('Files Loaded')

text_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=100)
texts = text_splitter.split_documents(documents)
return texts

texts= load_and_process_documents()

print(type(texts))

dense_embedding_func = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
text_contents = [doc.page_content for doc in texts] # Extract text content from each Document
dense_dim = len(dense_embedding_func.embed_query(text_contents[1]))

corpus=str(texts)
print(type(corpus))
sparse_embedding_func = BM25SparseEmbedding(corpus=corpus)

connections.connect(uri=CONNECTION_URI)

pk_field = "doc_id"
dense_field = "dense_vector"
sparse_field = "sparse_vector"
text_field = "text"
fields = [
FieldSchema(
name=pk_field,
dtype=DataType.VARCHAR,
is_primary=True,
auto_id=True,
max_length=100,
),
FieldSchema(name=dense_field, dtype=DataType.FLOAT_VECTOR, dim=dense_dim),
FieldSchema(name=sparse_field, dtype=DataType.SPARSE_FLOAT_VECTOR),
FieldSchema(name=text_field, dtype=DataType.VARCHAR, max_length=65_535),
]

schema = CollectionSchema(fields=fields, enable_dynamic_field=False)
collection = Collection(
name="Galitt_test_hybrid", schema=schema, consistency_level="Strong"
)

dense_index = {"index_type": "FLAT", "metric_type": "IP"}
collection.create_index("dense_vector", dense_index)
sparse_index = {"index_type": "SPARSE_INVERTED_INDEX", "metric_type": "IP"}
collection.create_index("sparse_vector", sparse_index)
collection.flush()

entities = []
for text in text_contents:
dense_vector = dense_embedding_func.embed_documents([text])[0]
sparse_vector = sparse_embedding_func.embed_documents([text])[0]

print(f"Dense Vector Type: {type(dense_vector)}")
print(f"Sparse Vector Type: {type(sparse_vector)}")


# Check the format of sparse_vector here
print(f"Sparse Vector: {sparse_vector}")

entity = {
    dense_field: dense_vector,
    sparse_field: sparse_vector,
    text_field: text,
}
entities.append(entity)

print(entities)

Filter out entities with empty sparse vectors

valid_entities = [entity for entity in entities if entity[sparse_field]]
print("Valid Entities after excluding empty sparse vectors")
print(valid_entities)

if valid_entities:
collection.insert(valid_entities)
print("Entities insertes in vector DB")
else:
print("No valid entities to insert.")
collection.load()

sparse_search_params = {"metric_type": "IP"}
dense_search_params = {"metric_type": "IP", "params": {}}
retriever = MilvusCollectionHybridSearchRetriever(
collection=collection,
rerank=WeightedRanker(0.5, 0.5),
anns_fields=[dense_field, sparse_field],
field_embeddings=[dense_embedding_func, sparse_embedding_func],
field_search_params=[dense_search_params, sparse_search_params],
top_k=3,
text_field=text_field,
)

result=retriever.invoke("What does SDA mean?")

print(result)

`

My retriever is created sucesfully but when I am doing any query to the retriever , I am getting below error :

**RPC error: [hybrid_search], <MilvusException: (code=65535, message=fail to search on QueryNode 3: worker(3) query failed: Assert "static_cast(field_meta.get_data_type()) == static_cast(info.type())" at /go/src/github.com/milvus-io/milvus/internal/core/src/query/Plan.cpp:48
=> vector type must be the same, field sparse_vector - type VECTOR_SPARSE_FLOAT, search info type VECTOR_FLOAT)>, <Time:{'RPC start': '2024-08-14 04:47:17.316357', 'RPC error': '2024-08-14 04:47:18.708334'}>
Traceback (most recent call last):
File "/home/rndadmin/galitt_dev/./test.py", line 194, in
result=retriever.invoke("What does SDA mean?")
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/langchain_core/retrievers.py", line 221, in invoke
raise e
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/langchain_core/retrievers.py", line 214, in invoke
result = self._get_relevant_documents(
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/langchain_milvus/retrievers/milvus_hybrid_search.py", line 157, in _get_relevant_documents
search_result = self.collection.hybrid_search(
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/pymilvus/orm/collection.py", line 943, in hybrid_search
resp = conn.hybrid_search(
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/pymilvus/decorators.py", line 147, in handler
raise e from e
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/pymilvus/decorators.py", line 143, in handler
return func(*args, **kwargs)
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/pymilvus/decorators.py", line 182, in handler
return func(self, *args, **kwargs)
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/pymilvus/decorators.py", line 122, in handler
raise e from e
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/pymilvus/decorators.py", line 87, in handler
return func(*args, kwargs)
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/pymilvus/client/grpc_handler.py", line 850, in hybrid_search
return self._execute_hybrid_search(
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/pymilvus/client/grpc_handler.py", line 761, in _execute_hybrid_search
raise e from e
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/pymilvus/client/grpc_handler.py", line 754, in _execute_hybrid_search
check_status(response.status)
File "/home/rndadmin/galitt_dev/galitt_env/lib/python3.10/site-packages/pymilvus/client/utils.py", line 63, in check_status
raise MilvusException(status.code, status.reason, status.error_code)
pymilvus.exceptions.MilvusException: <MilvusException: (code=65535, message=fail to search on QueryNode 3: worker(3) query failed: Assert "static_cast(field_meta.get_data_type()) == static_cast(info.type())" at /go/src/github.com/milvus-io/milvus/internal/core/src/query/Plan.cpp:48
=> vector type must be the same, field sparse_vector - type VECTOR_SPARSE_FLOAT, search info type VECTOR_FLOAT)>

can you tell me a work around for this ?

[yhmo]

I just simplified the script, and test with milvus v2.4.6, I didn't encounter the error.
My simplified script:

import random
from typing import List, Dict

from langchain_milvus.retrievers import MilvusCollectionHybridSearchRetriever
from langchain_milvus.utils.sparse import BaseSparseEmbedding
from langchain_core.embeddings import Embeddings

from pymilvus import (
    Collection,
    CollectionSchema,
    DataType,
    FieldSchema,
    WeightedRanker,
    connections,
)

CONNECTION_URI = "http://localhost:19530"
#export OPENAI_API_KEY=<your_api_key>

dense_dim = 128

connections.connect(uri=CONNECTION_URI)

pk_field = "doc_id"
dense_field = "dense_vector"
sparse_field = "sparse_vector"
text_field = "text"
fields = [
    FieldSchema(
        name=pk_field,
        dtype=DataType.VARCHAR,
        is_primary=True,
        auto_id=True,
        max_length=100,
    ),
    FieldSchema(name=dense_field, dtype=DataType.FLOAT_VECTOR, dim=dense_dim),
    FieldSchema(name=sparse_field, dtype=DataType.SPARSE_FLOAT_VECTOR),
    FieldSchema(name=text_field, dtype=DataType.VARCHAR, max_length=65535),
]

schema = CollectionSchema(fields=fields, enable_dynamic_field=False)
collection = Collection(name="test", schema=schema, consistency_level="Strong")

dense_index = {"index_type": "FLAT", "metric_type": "IP"}
collection.create_index("dense_vector", dense_index)
sparse_index = {"index_type": "SPARSE_INVERTED_INDEX", "metric_type": "IP"}
collection.create_index("sparse_vector", sparse_index)
collection.flush()

class DenseFunc(Embeddings):
    def embed_query(self, txt: str):
        return [random.random() for _ in range(dense_dim)]

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        return [[random.random() for _ in range(dense_dim)] for i in range(len(texts))]

dense_embedding_func = DenseFunc()

class SparseFunc(BaseSparseEmbedding):
    def generate_sparse_vector(self, dimension: int, non_zero_count: int) -> dict:
        indices = random.sample(range(dimension), non_zero_count)
        values = [random.random() for _ in range(non_zero_count)]
        sparse_vector = {index: value for index, value in zip(indices, values)}
        return sparse_vector

    def embed_query(self, txt: str):
        return self.generate_sparse_vector(20, 10)

    def embed_documents(self, texts: List[str])-> List[Dict[int, float]]:
        d = random.randint(5, 10)
        return [self.generate_sparse_vector(20, d) for _ in range(len(texts))]

sparse_embedding_func = SparseFunc()

text_contents = [
    "this is a chunk",
    "not a chunk",
    "what is chunk",
    "delete chunk",
]
entities = []
for text in text_contents:
    dense_vector = dense_embedding_func.embed_documents([text])[0]
    sparse_vector = sparse_embedding_func.embed_documents([text])[0]

    print(f"Dense Vector Type: {type(dense_vector)}")
    print(f"Sparse Vector Type: {type(sparse_vector)}")


    # Check the format of sparse_vector here
    print(f"Sparse Vector: {sparse_vector}")

    entity = {
        dense_field: dense_vector,
        sparse_field: sparse_vector,
        text_field: text,
    }
    entities.append(entity)

collection.insert(entities)
print("Entities insertes in vector DB")
collection.load()


sparse_search_params = {"metric_type": "IP"}
dense_search_params = {"metric_type": "IP", "params": {}}
retriever = MilvusCollectionHybridSearchRetriever(
    collection=collection,
    rerank=WeightedRanker(0.5, 0.5),
    anns_fields=[dense_field, sparse_field],
    field_embeddings=[dense_embedding_func, sparse_embedding_func],
    field_search_params=[dense_search_params, sparse_search_params],
    top_k=3,
    text_field=text_field,
)

print("\nDo hybrid search...")
results=retriever.invoke("What does SDA mean?")
for result in results:
    print(result)

[xiaofan-luan]

print(f"Dense Vector Type: {type(dense_vector)}")
print(f"Sparse Vector Type: {type(sparse_vector)}")

Check the format of sparse_vector here

print(f"Sparse Vector: {sparse_vector}")

entity = {
dense_field: dense_vector,
sparse_field: sparse_vector,
text_field: text,
}
entities.append(entity)

what is the result looks like of your dense embeddings and your sparse embedding?

print(f"Dense Vector Type: {type(dense_vector)}")
print(f"Sparse Vector Type: {type(sparse_vector)}")

Check the format of sparse_vector here

print(f"Sparse Vector: {sparse_vector}")

entity = {
dense_field: dense_vector,
sparse_field: sparse_vector,
text_field: text,
}
entities.append(entity)

from the error log, milvus believe you pass in a floatvector on a sparse vector field

[yhmo]

Although you checked the entities and only inserted valid_entities, the sentence "What does SDA mean?" is converted to an empty sparse embedding by the sparse_embedding_func so that the Milvus returns this error.

A workaround is, extend the BM25SparseEmbedding and check it output, something like:

class MyBM25SparseEmbedding(BM25SparseEmbedding):
    def embed_query(self, txt: str):
        sparse = self.embed_query(txt)
        if not sparse:
            raise Exception("An empty sparse embedding is generated by query text!")
        return sparse

sparse_embedding_func = MyBM25SparseEmbedding(corpus=corpus)

[nlp-greyfoss]

This piece of code will cause a recursion error.

Relpace self.embed_query with super().embed_query.

class MyBM25SparseEmbedding(BM25SparseEmbedding):
    def embed_query(self, txt: str):
        sparse = super().embed_query(txt)
        if not sparse:
            raise Exception("An empty sparse embedding is generated by query text!")
        return sparse

sparse_embedding_func = MyBM25SparseEmbedding(corpus=corpus)

[zc277584121]

There might be some problem when the query output of BM25SparseEmbedding is empty dict {}, I will try to fix it as a work around to solve this problem. you can also subclass the BM25SparseEmbedding just like what yhmo did. Milvus will also fix this issue in a short future.

[Preet1234-ux]

Now I am able to do Hybrid search .
Thanks all .

[kagaho]

Hi all,

Thanks for the clear explanations! I ran with similar issue and the workaround didn't work.

RPC error: [hybrid_search], <MilvusException: (code=65535, message=fail to search on QueryNode 2: worker(2) query failed: Assert "static_cast(field_meta.get_data_type()) == static_cast(info.type())" => vector type must be the same, field sparse_vector - type VECTOR_SPARSE_FLOAT, search info type VECTOR_FLOAT at /workspace/source/internal/core/src/query/Plan.cpp:48
)>, <Time:{'RPC start': '2024-10-29 18:16:07.940001', 'RPC error': '2024-10-29 18:16:08.748929'}>

The possible cause is that I used BGE M3 model to embed the text as dense and sparse vectors (ref: https://github.com/milvus-io/pymilvus/blob/master/examples/hello_hybrid_sparse_dense.py)
However, for the LangChain retrieval (on a different module) I used BGEM3 for dense and BM25 for sparse (could not find method in Langchain).
To fix the issue, I will redo the Milvus vectorstore with BGE M3 (dense) and BM25 (sparse) and see the results.

[xiaofan-luan]

@zc277584121 could you help on this one?

[Swati02041991]

Hi all, we observed the same issue with hybrid search. I am using Milvus 2.5.0
MilvusException: (code=65535, message=fail to search on QueryNode 98: worker(98) query failed: Operator::GetOutput failed for [Operator:, plan node id: 11536] : map::at)>

I am using BM25 function available inbuilt from pymilvus.
This error is gone if I drop the collection and recreate it, but the error occurs abruptly without any change to the collection for days.

[xiaofan-luan]

Hi all, we observed the same issue with hybrid search. I am using Milvus 2.5.0 MilvusException: (code=65535, message=fail to search on QueryNode 98: worker(98) query failed: Operator::GetOutput failed for [Operator:, plan node id: 11536] : map::at)>

I am using BM25 function available inbuilt from pymilvus. This error is gone if I drop the collection and recreate it, but the error occurs abruptly without any change to the collection for days.

can you upgrade to 2.5.6 and see if it's working?
If not, can u offer your search request and full milvus log?

[Swati02041991]

Hi @xiaofan-luan , please find attached search request and logs:
community.zip
As you can see I am not doing anything complex, just basic insert and query. We are trying to implement this for enterprise grade use cases. I tired releasing the collection and then trying loading and query. But error still exists.
This error disappears if i drop and recreate the collection. But this is a temporary fix, cannot be taken to production.

[xiaofan-luan]

double checked the client side code, it seems to be all good.

[xiaofan-luan]

@Swati02041991
what milvus version are u using?
did reboot or upgrade to 2.5.6 fixed the issue?

[Swati02041991]

@xiaofan-luan, I am using Milvus 2.5.0 version.
I doubt if we can upgrade as we have the pipelines set for this specific version only and upgrade goes through a long checklist.
Few observations:

Hybrid search with full text search (BM25 Sparse embedding generation inbuilt) is giving ' QueryNode failed to search' error only for 1 specific type of collection. While works for other collections with same schema. The collection with issue has semi-structured data (tabular data, special charecters like $, plus plain text ) being vectorised while the collections with no issue have unstructured data (pure text). Is BM25 inbuilt function tested for semi-structured data input?

[xiaofan-luan]

@xiaofan-luan, I am using Milvus 2.5.0 version. I doubt if we can upgrade as we have the pipelines set for this specific version only and upgrade goes through a long checklist. Few observations:

Hybrid search with full text search (BM25 Sparse embedding generation inbuilt) is giving ' QueryNode failed to search' error only for 1 specific type of collection. While works for other collections with same schema. The collection with issue has semi-structured data (tabular data, special charecters like $, plus plain text ) being vectorised while the collections with no issue have unstructured data (pure text). Is BM25 inbuilt function tested for semi-structured data input?

It is highly suggest to use 2.5.7.
many of the issue has been fixed.

[xiaofan-luan]

There are some known issues that has been fixed on multi chunk scalar datas.

Right now there seems to be no significant clue about how does this happened. (The error don't have stack trace, I opened another issue for that).

But I'm most likely sure you problem could be fixed via upgrade

[xiaofan-luan]

Sorry about this, but if upgrade is not help, we might need the segment dump of your errored data to reproduce this issue.