landing_page/qdrant-landing/content/documentation/tutorials/async-api.md

title, weight

title	weight
Asynchronous API	18

Using Qdrant asynchronously

Databases are often launched as separate services and are accessed via a network. All the interactions with them are IO-bound and can be performed asynchronously so as not to waste time actively waiting for a server response. If you use Python, that is achieved by using async/await syntax. That lets the interpreter switch to another task while waiting for a response from the server.

Qdrant exposes two interfaces: HTTP and gRPC. The official SDKs for different languages are based on the autogenerated clients, and in Python qdrant-client, you can call all the methods asynchronously. This tutorial presents how to do it.

When the usage of async API is justified?

If the application you are writing will never support multiple users at once, for example, it is a script run once daily, then there is no need to use async API. But if you are writing a web service that multiple users will use simultaneously, you shouldn't be blocking the threads of the web server as it limits the number of concurrent requests it can handle. In this case, you should use the async API.

Modern web frameworks like FastAPI and Quart support async API out of the box. Asynchronous code cannot be incorporated into an existing synchronous codebase, so it's better to choose the proper tool from the start.

How to use async API?

As usual, calling any method of Qdrant requires establishing a connection to the server. We need to create an instance of QdrantClient that will act as a gateway. If you want to do it locally, please make sure Qdrant server is running. If it's not, then you can launch it in a Docker container:

docker run -p "6333:6333" -p "6334:6334" qdrant/qdrant:v1.4.0

Alternatively, you can also use a Qdrant Cloud cluster. With a Cloud cluster you will pass the API key to the client constructor.

With that, we are ready to create QdrantClient instance:

from qdrant_client import QdrantClient

client = QdrantClient("localhost")
# If you use Qdrant Cloud, then pass the API key to the constructor:
# client = QdrantClient("https://your-cluster-url.cloud.com", api_key="your-api-key")

The default client only exposes the synchronous methods for interacting with the server. To access the async client and it's underlying methods, we need to use autogenerated async clients. It is possible to use asynchronous HTTP API by using qdrant_client.http.api_client.AsyncApiClient. However, the gRPC API is much more efficient and is the recommended way of using Qdrant asynchronously.

gRPC API

Every instance of QdrantClient has two properties we are going to use: async_grpc_collections and async_grpc_points. They expose the autogenerated clients for collections and points respectively.

The gRPC client uses type definitions for all the interactions with the server. They are autogenerated from the source code of Qdrant server, and we have to import them if we want to call any of the available methods.

from qdrant_client import grpc

If your IDE does not support autocompletion for the autogenerated clients, you can refer to the documentation and see the available fields for each of them.

Creating a collection

Let's create a collection and add some points to it.

response = await client.async_grpc_collections.Create(
    grpc.CreateCollection(
        collection_name="my_collection",
        vectors_config=grpc.VectorsConfig(
            params=grpc.VectorParams(
                size=100,
                distance=grpc.Distance.Cosine,
            )
        ),
        quantization_config=grpc.QuantizationConfig(
            scalar=grpc.ScalarQuantization(
                type=grpc.QuantizationType.Int8,
                always_ram=False,
            )
        ),
        timeout=10,
    )
)

Our collection was set up for 100-dimensional vectors, cosine distance and 8-bit quantization. The timeout is set to 10 seconds, so if the server does not respond within 10 seconds, the request will be aborted.

Checking collection info

We can check the collection info to see if it was created successfully, and configured as we expected:

response = await client.async_grpc_collections.Get(
    grpc.GetCollectionInfoRequest(
        collection_name="my_collection"
    )
)

The response object is going to be an instance of grpc.GetCollectionInfoResponse and we can access the fields we need or simply display its content.

print(response)

result {
  status: Green
  optimizer_status {
    ok: true
  }
  segments_count: 8
  config {
    params {
      shard_number: 1
      on_disk_payload: true
      vectors_config {
        params {
          size: 100
          distance: Cosine
        }
      }
      replication_factor: 1
      write_consistency_factor: 1
    }
    hnsw_config {
      m: 16
      ef_construct: 100
      full_scan_threshold: 10000
      max_indexing_threads: 0
      on_disk: false
    }
    optimizer_config {
      deleted_threshold: 0.2
      vacuum_min_vector_number: 1000
      default_segment_number: 0
      indexing_threshold: 20000
      flush_interval_sec: 5
      max_optimization_threads: 1
    }
    wal_config {
      wal_capacity_mb: 32
      wal_segments_ahead: 0
    }
    quantization_config {
      scalar {
        type: Int8
        always_ram: false
      }
    }
  }
  indexed_vectors_count: 0
}
time: 1.3022e-05

As you can see, the collection was created properly and is ready to be used.

Handling exceptions

If the collection does not exist, the server will return an error. We can handle it by catching the exception. We may expect some specific exceptions, or simply catch all the errors that subclass RpcError from grpc package:

from grpc import RpcError

try:
    response = await client.async_grpc_collections.Get(
        grpc.GetCollectionInfoRequest(
            collection_name="my_collection_2"
        )
    )
except RpcError as e:
    print(e)

Our exception contain the details about the error, including the status code and the message:

<AioRpcError of RPC that terminated with:
	status = StatusCode.NOT_FOUND
	details = "Not found: Collection `my_collection_2` doesn't exist!"
	debug_error_string = "UNKNOWN:Error received from peer  {created_time:"2023-08-29T12:14:05.885634788+02:00", grpc_status:5, grpc_message:"Not found: Collection `my_collection_2` doesn\'t exist!"}"
>

We can use those details to properly handle some specific cases. For example, if we want to create a collection only if it does not exist, we can check the status code and create it if it is StatusCode.NOT_FOUND:

Updating collection configuration

It's quite likely that we will want to update the collection configuration. For example, we may want to change the HNSW parameters to improve the search precision. Let's change the ef_construct parameter to 200:

await client.async_grpc_collections.Update(
    grpc.UpdateCollection(
        collection_name="my_collection",
        hnsw_config=grpc.HnswConfigDiff(
            ef_construct=200,
        ),
    )
)

Deleting a collection

There is plenty of other methods available for collections, like creating aliases or checking the status of the cluster. We are not going to call all of them in this tutorial, but you can always check the list of available methods and their parameters in the documentation.

Right now, we are just going to delete the collection we created:

await client.async_grpc_collections.Delete(
    grpc.DeleteCollection(
        collection_name="my_collection"
    )
)

That's it when it comes to collection management. Let's move on to points (individual vectors) allowing us to load data into our collection

Adding points to the collection

Assuming your collection created successfuly, we can now add some points to it. Let's add our first two points into the collection:

import numpy as np

points = [
    grpc.PointStruct(
        id=grpc.PointId(num=1),
        payload={
            "int_param": grpc.Value(integer_value=32),
            "bool_param": grpc.Value(bool_value=True),
        },
        vectors=grpc.Vectors(
            vector=grpc.Vector(data=np.random.random(100).tolist()),
        ),
    ),
    grpc.PointStruct(
        id=grpc.PointId(num=2),
        payload={
            "int_param": grpc.Value(integer_value=64),
            "bool_param": grpc.Value(bool_value=False),
        },
        vectors=grpc.Vectors(
            vector=grpc.Vector(data=np.random.random(100).tolist()),
        ),
    ),
]

response = await client.async_grpc_points.Upsert(
    grpc.UpsertPoints(
        collection_name="my_collection",
        points=points,
    )
)

Those points are automatically indexed and available for search. In the future, we can add more points, or update the existing ones. All the point related operations are documented in the Qdrant server repository.

The main operation we need from the vector database is semantic search. Let's see how to perform it.

Asynchronous semantic search

Search operation requires a query vector and a collection name. We can also specify the number of results we want to retrieve and some other parameters, but let's take one step at a time. The minimal example may look like this:

response = await client.async_grpc_points.Search(
    grpc.SearchPoints(
        collection_name="my_collection",
        vector=np.random.rand(100).tolist(),
        limit=1,
    )
)

print(response)

The response is going to contain just the point id and similarity score for the vector most similar to the random vector generated in the code. For example, that might be the output of the code above:

result {
  id {
    num: 1
  }
  score: 0.786471784
  version: 1
}
time: 0.000456134

Returning vectors and payloads

If you need the vectors and payloads of the points, you can specify it in the request:

response = await client.async_grpc_points.Search(
    grpc.SearchPoints(
        collection_name="my_collection",
        vector=np.random.rand(100).tolist(),
        limit=1,
        with_payload=grpc.WithPayloadSelector(enable=True),
        with_vectors=grpc.WithVectorsSelector(enable=True),
    )
)

That's all the code you need to bring back the vector and payload.

Applying filters

If you ever used SQL, you are probably familiar with the WHERE clause. It allows you to filter the results of the query. Qdrant has similar functionality. Let's say we want to find all the points similar to given vector, but with int_param equal to 32. We can do it like this:

response = await client.async_grpc_points.Search(
    grpc.SearchPoints(
        collection_name="my_collection",
        vector=np.random.rand(100).tolist(),
        limit=1,
        with_payload=grpc.WithPayloadSelector(enable=True),
        filter=grpc.Filter(
            must=[
                grpc.Condition(
                    field=grpc.FieldCondition(
                        key="int_param",
                        match=grpc.Match(integer=32)
                    )
                )
            ]
        )
    )
)

We can build even more sophisticated filters with the must, should and must_not clauses. You can find more details in the filtering
documentation.

Analogous to search, we can also scroll our collection to retrieve all the points or use the recommendation API to find points similar to positive and dissimilar to negative examples.

Support of Qdrant async API in Python libraries

There is plenty of Python libraries that Qdrant integrates with. Up till now, only Langchain provides async API support. Qdrant is the only vector database with full coverage of async API in Langchain. The documentation describes how to use it.