Why can't I connect remote Milvus server in Python SDK?

[lingvisa]

A remote Milvus distributed instance is deployed. And I can access it through "curl":

 curl --request GET \
    --url 'https://rag-db.c-97a1353.stage.kyma.ondemand.com/v1/vector/collections' \
    --header "Authorization: Bearer xxx:xxx" \
    --header 'accept: application/json' \
    --header 'content-type: application/json'
{"code":200,"data":["demo"]}%    

But then in my Python client, it reports this error:

from pymilvus import MilvusClient
client = MilvusClient(uri="https://rag-db.c-97a1353.stage.kyma.ondemand.com", token="xxx:xxx", db_name="default")
print(client)

The error message is:

raise _InactiveRpcError(state)  # pytype: disable=not-instantiable
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
   status = StatusCode.UNAUTHENTICATED
   details = "Received http2 header with status: 401"
   debug_error_string = "UNKNOWN:Error received from peer  {grpc_message:"Received http2 header with status: 401", grpc_status:16, created_time:"2024-09-12T17:25:47.431059-07:00"}"

In both cases, the credentials are the same: "xxx:xxx". My version:
pymilvus==2.4.6

Why is that?

[lingvisa]

Hello, this is an urgent issue. Can you please help me with this? Thank you

[yhmo]

This is not a problem of milvus/pymilvus. Search the keyword "Received http2 header with status: 401" to get more info.

[lingvisa]

Authentification turned off for the time being. I loaded some documents, one collection with multiple partitions. But why does it show each individual

's partition's count is 0, and all the counts are under 'default partition'?

[yhmo]

The interface to get row count of partitions is a pretty old interface, the returned number is not real-time.
If you click the "Flush" button in the "Segments" page, you will see the row count updated.

[lingvisa]

So I have 8 partitions and they are placed into two segments.
Another little issue is, the function 'collection.has_index()' only works for a single index. If I have two indexes, this function reports an error, saying that it has multiple indexes, so it's ambiguous. I can use 'collection.indexes' to test whether it has any index as a workaround.

[yhmo]

has_index() has a parameter 'index_name". By default, index name is the vector field name.

import random

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

connections.connect(host='localhost',port='19530')

schema = CollectionSchema(fields=[
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=False),
    FieldSchema(name="vector1", dtype=DataType.FLOAT_VECTOR, dim=128),
    FieldSchema(name="vector2", dtype=DataType.FLOAT_VECTOR, dim=256),
])

collection_name = "BBB"
utility.drop_collection(collection_name)
collection = Collection(collection_name, schema)

index_params_ivf={"index_type": "IVF_FLAT", "params": {"nlist": 128}, "metric_type": "L2"}
collection.create_index("vector1", index_params=index_params_ivf)
collection.create_index('vector2', index_params=index_params_ivf)

for idx in collection.indexes:
    print(idx.to_dict())

print("vector1 has index:", collection.has_index(index_name="vector1"))
print("vector2 has index:", collection.has_index(index_name="vector2"))

collection.drop_index(index_name="vector1")
print("vector1 has index:", collection.has_index(index_name="vector1"))

[lingvisa]

Right. I should use "index_name=xxx".

[lingvisa]

@yhmo Is it possible that langchain can also provide a version of MilvusCollectionHybridSearchRetriever that can return the scores as the non-hybrid version does as below?

retriever = Milvus(
        dense_embedding_func,
        collection_name=collection_name,
        vector_field="dense_vector",
        consistency_level="Strong",
    )

Although I can create multiple partitions in one collection in Milvus in hybrid search, that also requires to create a single keyword index for the whole document set. If I have documents from different sources it is logical to create one collection for each data source, and this will reduce the bm25 index building's scalability requirement.

Does this make sense, or any recommendation on this concern?

[lingvisa]

When do i need to use flush() and load() in creating a new collection?

           collection = Collection(self.collection_name, schema, consistency_level="Strong") 
           collection.create_partition(partition_name=self.partition_name) 
           sparse_index = {"index_type": "SPARSE_INVERTED_INDEX", "metric_type": "IP"}
           collection.create_index("sparse_vector", sparse_index)
           dense_index = {"index_type": "AUTOINDEX", "metric_type": "IP"}
            collection.create_index("dense_vector", dense_index)

Now, do i need the two statements below:

collection.flush()
collection.load()

Or are they only needed when I insert new entities to the collection?

 for i in range(0, len(docs), batch_size):
            batched_entities = [
                docs[i: i + batch_size],
                sparse_embeddings[i: i + batch_size],
                dense_embeddings[i: i + batch_size],
            ]
        collection.insert(batched_entities)
        collection.flush()
        collection.load()

[yhmo]

You don't need to call flush(). In fact, we strongly recommend users don't call flush() interface manually. Milvus internally automatically flushes data.

The following workflow is good practice:
create_collection() ----> insert() ---> ..... ---> insert() ---> create_index() ----> load() ---> search()---> ... ---> search()

[lingvisa]

why does create_index() run after insert()? create_index() should occur when a collection is initialized or created, and then you insert() data, right?

[xiaofan-luan]

why does create_index() run after insert()? create_index() should occur when a collection is initialized or created, and then you insert() data, right?

you can do both.
either create index first or insert data first.
Personally I recommend to create index first then insert data

[lingvisa]

That's good to know! @xiaofan-luan

[lingvisa]

@xiaofan-luan and @yhmo For MilvusCollectionHybridSearchRetriever,

 retriever = MilvusCollectionHybridSearchRetriever(
        collection=collection,
        # rerank=WeightedRanker(0.5, 0.5),
        rerank=RRFRanker(),
        anns_fields=["dense_vector", "sparse_vector"],
        field_embeddings=[dense_embedding_func, sparse_embedding_func],
        field_search_params=[dense_search_params, sparse_search_params],
        top_k=TOP_K,
        text_field='text',
    )

If I change top_k from 5 to 20 without changing anything else, the orders of ranked results are not exactly the same. No matter I use 5 or 20, the top 5 results should be exactly the same, right?

[yhmo]

Definitely different.
The topk value will affect hybrid_search result. Read this discussion for more info: #36329

[lingvisa]

In the example,

connections.connect(
    host="localhost",
    port="19530",
    user="root",
    password="password",
    db_name="default",
)
print(utility.get_server_version())
collection = Collection(collection_name)
client = MilvusClient(
    host="localhost",
    port="19530",
    user="root",
    password="password",
    db_name="default",
)

If I have the collection object, I don't need another 'client' object. The 'collection' works for both hybrid and non-hybrid search, right? is the MilvusClient() the old API, and "collections.connect(...), collection=Collection(collection_name)" are the new API?

[lingvisa]

Once a collection is created and I want to switch to another similarity metric for testing, can I switch similarity metric in web-based browser, or I need to do it in code without the need to re-create the collection?

Also, how to test searches in the browser? Please see the attachment.

[xiaofan-luan]

Use pretrained BM25 index is not recommeneded. If your document is totally different from trained dataset, you will get empty sparse vector.

Milvus 2.5 is out with built in BM25 support, which should be much better for ur case.

[lingvisa]

Is Milvus 2.5 out? I checked this documentation, https://milvus.io/docs/multi-vector-search.md, not updated yet. Any instruction on how to trim BM25 with my own data with Milvus?

[xiaofan-luan]

Branch is already created. Document and image should be ready for today.

[lingvisa]

Does it have instruction on how to train my own BM25? That would be great if that can be included.

[lingvisa]

I did this: CustomBM25SparseEmbedding(corpus, model_path=sparse_model_path), which creates my own IDf.json. By customizing, I persisted the index. If you mean this by 'pretrained BM25 index', I already did this.

[lingvisa]

Hi, @yhmo

async embed_and_insert_documents(self, ...):

    ...

    self.collection.insert(batched_entities, partition_name=partition_name)

Do I need to use a lock for the insert() function when adding entities into milvus in an async context? Or internally it already takes care of concurrency issue so I don' t need to worry about it?

[yhmo]

No need to add lock, milvus insert() interface supports parallel insertion.

[lingvisa]

I have a query "What is CRM-S-SOM?", which generates an empt sparse vector, and because of that, it triggered an internal exception, like this one: #35463

Is there a better way to handle this in Milvus 2.5.0? For example, if sparse vector is empty, can the hybrid_search just use dense_vector to search, instead of throw an exception with a warning message?

search_result = self.collection.hybrid_search(
            requests, self.rerank, limit=self.top_k, output_fields=self.output_fields
        )

[yhmo]

I think it is by design.
Empty sparse vector is illegal for search(), if you call search() with an empty sparse vector, you will get the same exception. Each sub-request of hybrid_search() internally equals a search request. If a sub-request failed, we should tell users the input is illegal. Otherwise, the illegal input is omitted, the user might get an unexpected result but he don't know why.

[lingvisa]

In order to drop a partition, I need to release it first, but i didn't find a button to do that. How to release a partition from the UI?

[yhmo]

Currently, Attu doesn't support separate load/release partitions. In our plan, this feature will be available in a minor version of Attu 2.5.x, maybe v2.5.1 or v2.5.2.

[lingvisa]

Hi, @yhmo I need to benchmark our existing RAG and Milvus hybrid rag prototype latency. Do you have benchmark scripts that I can refer to? Milvus 2.4.11 is deployed in a cluster

[lingvisa]

In collection building, I use consistency_level=Strong. When benchmarking the latency at query time, I set consistency_level=Bounded as follows, but the latency doesn't really change as follows (using langchain retrieval):

retriever = CustomMilvusCollectionHybridSearchRetriever(
            collection=collection,
            rerank=RRFRanker(),
            anns_fields=["dense_vector", "sparse_vector"],
            field_embeddings=[get_dense_embedding_func(), sparse_embedding_func],
            field_search_params=[dense_search_params, sparse_search_params],
            top_k=top_k,
            text_field='content',
            consistency_level="Bounded",
        )

consistency_level="Bounded", latency doesn't change when I switch to "Strong". Is that because I used "Strong" in building the collection?

[yhmo]

Yes, if your data has no change, "Eventually" is better.

Normally, newly inserted data will be invisible for hundreds of milliseconds.

Default consistency level is "Bounded" https://github.com/milvus-io/pymilvus/blob/ca9d40df30faeab23206210467c3561e5685e33a/pymilvus/client/constants.py#L9

[yhmo]

"consistency_level doesn't take effect" ------------------ a possible reason is this parameter is not passed into pymilvus hybrid_search interface correctly. Debug into your code, set a breakpoint in pymilvus hybrid_search interface to observe the parameters.

[lingvisa]

I tracked the code here from venv/lib/python3.10/site-packages/pymilvus/client/ts_utils.py. 'use_default'=True, consistency_level=None, which means the user passed value Bounded is not correctly obtained here, but since use_default_consistency

def construct_guarantee_ts(collection_name: str, kwargs: Dict):
    consistency_level = kwargs.get("consistency_level")
    use_default = consistency_level is None
    if use_default:
        # in case of the default consistency is Customized or Session,
        # we set guarantee_timestamp to the cached mutation ts or 1
        kwargs["guarantee_timestamp"] = get_collection_ts(collection_name) or get_eventually_ts()
        return True
    consistency_level = get_consistency_level(consistency_level)
    kwargs["consistency_level"] = consistency_level
    if consistency_level == ConsistencyLevel.Strong:
        # Milvus will assign a newest ts.
        kwargs["guarantee_timestamp"] = 0
    elif consistency_level == ConsistencyLevel.Session:
        # Using the last write ts of the collection.
        # TODO: get a timestamp from server?
        kwargs["guarantee_timestamp"] = get_collection_ts(collection_name) or get_eventually_ts()
    elif consistency_level == ConsistencyLevel.Bounded:
        # Milvus will assign ts according to the server timestamp and a configured time interval
        kwargs["guarantee_timestamp"] = get_bounded_ts()
    else:
        # Users customize the consistency level, no modification on `guarantee_timestamp`.
        kwargs.setdefault("guarantee_timestamp", get_eventually_ts())
    return use_default

Then, here
use_default_consistency = ts_utils.construct_guarantee_ts(collection_name, kwargs)

use_default_consitency=True.

If the default consistency=Bounded, i suppose i still applied the right consistency_level i want. Then the next question is,

Average Latency: 622.81 ms

This average latency is normal for about 1million chunks? I suppose it should be under 100ms for most queries. Do you have other tests for average query time?

[yhmo]

Pretty easy to test:

import random
import time

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

connections.connect(host='localhost', port=19530)
print(utility.get_server_version())

collection_name = "AAA"
dim = 512

fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=False),
    FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=dim),
]
schema = CollectionSchema(fields)
utility.drop_collection(collection_name)
collection = Collection(collection_name, schema)
print(collection_name, "created")

desc = collection.describe()
# 0 is Strong, 1 is Session, 2 is Bounded, 3 is Eventually
assert desc["consistency_level"] == 2

batch = 10000
for i in range(100):
    data = [
        [i*batch+k for k in range(batch)],
        [[random.random() for _ in range(dim)] for _ in range(batch)],
    ]
    collection.insert(data=data)
    print("inserted", i)

collection.flush()

print("indexing...")
collection.create_index(field_name="vector", index_params={
        "index_type": "IVF_FLAT",
        "metric_type": "L2",
        "params": {"nlist": 256},
    })
collection.load()

collection = Collection(collection_name)
print("row count:", collection.query(expr="", output_fields=["count(*)"]))

def search_test(level:str):
    print("====================================================================")
    print("test search with consistency level:", level)
    repeat = 10
    start = time.time()
    for i in range(repeat):
        t1 = time.time()
        results = collection.search(data=[[random.random() for _ in range(dim)]],
                                    limit=10,
                                    param={'nprobe': 16},
                                    anns_field="vector",
                                    consistency_level=level)
        t2 = time.time()
        print(f"search time cost: {(t2-t1)*1000} ms")
        # for res in results[0]:
        #     print(res)

    end = time.time()
    cost = (end-start)*1000
    print(f"total time cost: {cost} ms")
    print(f"average time cost per request: {cost/repeat} ms")
    print("====================================================================")

search_test("Strong")
search_test("Bounded")
search_test("Eventually")

row count: data: ["{'count(*)': 1000000}"] 
====================================================================
test search with consistency level: Strong
search time cost: 444.4272518157959 ms
search time cost: 199.91588592529297 ms
search time cost: 192.0621395111084 ms
search time cost: 419.44003105163574 ms
search time cost: 188.24410438537598 ms
search time cost: 393.41282844543457 ms
search time cost: 406.45408630371094 ms
search time cost: 200.07729530334473 ms
search time cost: 192.39377975463867 ms
search time cost: 407.3817729949951 ms
total time cost: 3044.2254543304443 ms
average time cost per request: 304.42254543304443 ms
====================================================================
====================================================================
test search with consistency level: Bounded
search time cost: 14.234542846679688 ms
search time cost: 16.038894653320312 ms
search time cost: 11.225700378417969 ms
search time cost: 7.4920654296875 ms
search time cost: 6.23774528503418 ms
search time cost: 5.858898162841797 ms
search time cost: 6.477832794189453 ms
search time cost: 5.587339401245117 ms
search time cost: 6.140708923339844 ms
search time cost: 6.526947021484375 ms
total time cost: 86.1048698425293 ms
average time cost per request: 8.61048698425293 ms
====================================================================
====================================================================
test search with consistency level: Eventually
search time cost: 5.95402717590332 ms
search time cost: 5.578517913818359 ms
search time cost: 6.160497665405273 ms
search time cost: 5.881071090698242 ms
search time cost: 12.094497680664062 ms
search time cost: 6.348848342895508 ms
search time cost: 5.700111389160156 ms
search time cost: 6.645679473876953 ms
search time cost: 6.487369537353516 ms
search time cost: 5.950450897216797 ms
total time cost: 66.95914268493652 ms
average time cost per request: 6.695914268493652 ms
====================================================================

[lingvisa]

Very helpful. Looks like the parameter isn't passed correctly, and even the use_default_consistency=True doesn't use Bounded. I will use Bounded to create a new collection by defining Bounded in the schema, not depending on it in query time, and test to see.

…

On Thu, Dec 19, 2024, 12:38 AM groot ***@***.***> wrote: Pretty easy to test: import random import time from pymilvus import ( connections, FieldSchema, CollectionSchema, DataType, Collection, utility, ) connections.connect(host='localhost', port=19530) print(utility.get_server_version()) collection_name = "AAA" dim = 512 fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=False), FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=dim), ] schema = CollectionSchema(fields) utility.drop_collection(collection_name) collection = Collection(collection_name, schema) print(collection_name, "created") desc = collection.describe() # 0 is Strong, 1 is Session, 2 is Bounded, 3 is Eventually assert desc["consistency_level"] == 2 batch = 10000 for i in range(100): data = [ [i*batch+k for k in range(batch)], [[random.random() for _ in range(dim)] for _ in range(batch)], ] collection.insert(data=data) print("inserted", i) collection.flush() print("indexing...") collection.create_index(field_name="vector", index_params={ "index_type": "IVF_FLAT", "metric_type": "L2", "params": {"nlist": 256}, }) collection.load() collection = Collection(collection_name) print("row count:", collection.query(expr="", output_fields=["count(*)"])) def search_test(level:str): print("====================================================================") print("test search with consistency level:", level) repeat = 10 start = time.time() for i in range(repeat): t1 = time.time() results = collection.search(data=[[random.random() for _ in range(dim)]], limit=10, param={'nprobe': 16}, anns_field="vector", consistency_level=level) t2 = time.time() print(f"search time cost: {(t2-t1)*1000} ms") # for res in results[0]: # print(res) end = time.time() cost = (end-start)*1000 print(f"total time cost: {cost} ms") print(f"average time cost per request: {cost/repeat} ms") print("====================================================================") search_test("Strong") search_test("Bounded") search_test("Eventually") row count: data: ["{'count(*)': 1000000}"] ==================================================================== test search with consistency level: Strong search time cost: 444.4272518157959 ms search time cost: 199.91588592529297 ms search time cost: 192.0621395111084 ms search time cost: 419.44003105163574 ms search time cost: 188.24410438537598 ms search time cost: 393.41282844543457 ms search time cost: 406.45408630371094 ms search time cost: 200.07729530334473 ms search time cost: 192.39377975463867 ms search time cost: 407.3817729949951 ms total time cost: 3044.2254543304443 ms average time cost per request: 304.42254543304443 ms ==================================================================== ==================================================================== test search with consistency level: Bounded search time cost: 14.234542846679688 ms search time cost: 16.038894653320312 ms search time cost: 11.225700378417969 ms search time cost: 7.4920654296875 ms search time cost: 6.23774528503418 ms search time cost: 5.858898162841797 ms search time cost: 6.477832794189453 ms search time cost: 5.587339401245117 ms search time cost: 6.140708923339844 ms search time cost: 6.526947021484375 ms total time cost: 86.1048698425293 ms average time cost per request: 8.61048698425293 ms ==================================================================== ==================================================================== test search with consistency level: Eventually search time cost: 5.95402717590332 ms search time cost: 5.578517913818359 ms search time cost: 6.160497665405273 ms search time cost: 5.881071090698242 ms search time cost: 12.094497680664062 ms search time cost: 6.348848342895508 ms search time cost: 5.700111389160156 ms search time cost: 6.645679473876953 ms search time cost: 6.487369537353516 ms search time cost: 5.950450897216797 ms total time cost: 66.95914268493652 ms average time cost per request: 6.695914268493652 ms ==================================================================== — Reply to this email directly, view it on GitHub <#36226 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AB4J634S4HI4PAWW4QMWYDL2GKAZJAVCNFSM6AAAAABOEKBZKCVHI2DSMVQWIX3LMV43URDJONRXK43TNFXW4Q3PNVWWK3TUHMYTCNRRGQ2TKMQ> . You are receiving this because you were mentioned.Message ID: ***@***.***>

[lingvisa]

In this class, if topK=10, does it get top 5 from bm25 and another top 5 from semantic search and then, merge and rerank?

retriever = CustomMilvusCollectionHybridSearchRetriever(
collection=collection,
rerank=RRFRanker(),
anns_fields=["dense_vector", "sparse_vector"],
field_embeddings=[get_dense_embedding_func(), sparse_embedding_func],
field_search_params=[dense_search_params, sparse_search_params],
top_k=top_k,
text_field='content',
consistency_level="Bounded",
)

[yhmo]

In langchain source code, the hybrid_search() is called and assigned multiple AnnSearchRequest with the same topk values:
https://github.com/langchain-ai/langchain-milvus/blob/b925dac8b1b3b4cff1f1972a9644584d113fe3f6/libs/milvus/langchain_milvus/retrievers/milvus_hybrid_search.py#L154
https://github.com/langchain-ai/langchain-milvus/blob/b925dac8b1b3b4cff1f1972a9644584d113fe3f6/libs/milvus/langchain_milvus/retrievers/milvus_hybrid_search.py#L121C18-L121C30
https://github.com/langchain-ai/langchain-milvus/blob/b925dac8b1b3b4cff1f1972a9644584d113fe3f6/libs/milvus/langchain_milvus/retrievers/milvus_hybrid_search.py#L59

If the top_k = 10, it gets top 10 from "dense_vector" and gets top 10 from "sparse_vector". And rerank the 20 items with RRF to get the final top 10 items.

[lingvisa]

If my topK is set to 10, and I want the results from semantic and keyword searches to be based on a ratio of 0.6/0.4 (e.g., 6 results from semantic search and 4 from keyword search), I would merge these results and re-rank them. This approach allows me to prioritize semantic search while still including a few top results from keyword search, without distributing them evenly. How can I implement this? Thanks.

[yhmo]

Milvus only supports 2 rerankers: Weighted and RRF. Both of them cannot ensure that "6 results from A and 4 from B".
Currently milvus doesn't support customized reranker.
You can:

1. call normal search() to A to get 10 results
2. call normal search() to B to get 10 results
3. rerank the 20 items by yourself at your will