Использование асинхронного API Vector DB для эффективного создания приложений на Python#

Асинхронное программирование широко применяется в экосистеме Python. Инструменты вроде FastAPI приняли эту новую парадигму, но она также становится стандартом для моделей машинного обучения, предоставляемых по модели SaaS. Например, SDK от компании Cohere предоставляет асинхронный клиент наряду с синхронным аналогом.

Базы данных часто запускаются как отдельные сервисы и доступны через сеть. Все взаимодействия с ними связаны с вводом-выводом и могут выполняться асинхронно, чтобы избежать потери времени при активном ожидании ответа сервера. В Python это достигается путем использования синтаксиса async/await. Это позволяет интерпретатору переключаться на другую задачу во время ожидания ответа от сервера.

Когда следует использовать асинхронный API#

Нет необходимости использовать асинхронный API, если приложение никогда не будет поддерживать одновременную работу нескольких пользователей (например, оно выполняется один раз в день). Однако, если разрабатывается веб-сервис, который одновременно используют несколько пользователей, не стоит блокировать потоки веб-сервера, так как это ограничивает количество параллельных запросов, которые он может обрабатывать. В этом случае необходимо использовать асинхронный API.

Современные веб-фреймворки, такие как FastAPI и Platform V Vector DB (далее - Vector DB) поддерживают асинхронные API из коробки. Смешивание асинхронного кода с существующим синхронным кодом может представлять собой вызов. Синтаксис async/await нельзя использовать в синхронных функциях. С другой стороны, выполнение операции ввода-вывода синхронно в асинхронном коде считается антипаттерном. Поэтому, если строится асинхронный веб-сервис, доступ к которому осуществляется через сервер ASGI, необходимо использовать асинхронный API для всех взаимодействий с Vector DB.

Весь асинхронный код должен запускаться в асинхронном контексте. Обычно это означает, что нужно использовать asyncio.run или asyncio.create_task для их выполнения. Пожалуйста, обратитесь к документации asyncio за подробностями.

Асинхронная работа с Vector DB#

Самый простой способ запуска асинхронного кода — определить функцию async и использовать ее следующим образом:

from qdrant_client import models

import qdrant_client
import asyncio


async def main():
    client = qdrant_client.AsyncQdrantClient("localhost")

    # Create a collection
    await client.create_collection(
        collection_name="my_collection",
        vectors_config=models.VectorParams(size=4, distance=models.Distance.COSINE),
    )

    # Insert a vector
    await client.upsert(
        collection_name="my_collection",
        points=[
            models.PointStruct(
                id="5c56c793-69f3-4fbf-87e6-c4bf54c28c26",
                payload={
                    "color": "red",
                },
                vector=[0.9, 0.1, 0.1, 0.5],
            ),
        ],
    )

    # Search for nearest neighbors
    points = await client.query_points(
        collection_name="my_collection",
        query=[0.9, 0.1, 0.1, 0.5],
        limit=2,
    ).points

    # Your async code using AsyncQdrantClient might be put here
    # ...


asyncio.run(main())

AsyncQdrantClient предоставляет те же методы, что и его синхронный аналог QdrantClient. Если уже есть синхронная база кода, переход на асинхронный API сводится просто к замене QdrantClient на AsyncQdrantClient и добавлению await перед каждым вызовом метода.