qdrant / qdrant/qdrant-client

The problem I encountered when using celery task to execute qdrant in fastApi

Open
#672 1 comment 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Python
Stars
1.4k
Forks
301
Avg merge
2d 9h
Merged PRs (30d)
37

Description

I`m using an asynchronous method to perform qdrant operations in the celery task, and set prefer_grpc to True, and the qdrant asynchronous client will be blocked. It should be caused by a problem with the event loop.
My code snippet:

@db_required
async def test(file_id: str):
    # some other code...
    try:
        file_obj: UploadFile = await OSSServices.read_file(file_res, file.file_name)
        collection_name = file.kb_id

        rag = RAGInterface(collection_name)
        # insert with metadata
        metadata = {
            "kb_id": file.kb_id,
            "file_name": file.file_name,
            "file_id": file.file_id,
            "created_at": file.created_at.strftime("%Y-%m-%d %H:%M:%S"),
        }
        await rag.insert(files=file_obj, kwargs=metadata)
        # vector parser success
    except Exception as e:
        logger.exception(f"[vector_create_task]:{file.file_name}\n{str(e)}")


@shared_task
def vector_create_task(file_id: str):
    """向量解析-提交"""

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    try:
        async_result = loop.run_until_complete(test(file_id=file_id))
        return async_result
    finally:
        loop.close()

In my RAGInterface, I made some business encapsulation based on qdrant api. After my debugging, I found that every time I called AsyncQdrantClient, I would get stuck. For example, when I execute: async_client.collection_exists(collection_name), I will be stuck in the qdrant source code:

  async def collection_exists(self, collection_name: str, **kwargs: Any) -> bool:
      if self._prefer_grpc:
          return (
              await self.grpc_collections.CollectionExists(
                  grpc.CollectionExistsRequest(collection_name=collection_name),
                  timeout=self._timeout,
              )
          ).result.exists
      result: Optional[models.CollectionExistence] = (
          await self.http.collections_api.collection_exists(collection_name=collection_name)
      ).result
      assert result is not None, "Collection exists returned None"
      return result.exists

code from : qdrant_client.async_qdrant_remote.AsyncQdrantRemote

I`m very confused in this problem, anybody can help me? thanks very much!!!

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start by reproducing the Celery event-loop setup shown in the issue, then inspect qdrant_client.async_qdrant_remote.AsyncQdrantRemote.collection_exists with prefer_grpc enabled. Compare the gRPC and HTTP paths and capture where execution stops. Done means identifying the event-loop or client interaction that causes the block and validating the behavior with a focused reproduction.

Written by the indexing model from the issue text.

Assessment

Tech stack
fastapi, grpc, python
Domain
backend, database
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.