qdrant component

chainfury.components.qdrant.disable_indexing(collection_name: str)[source]

Disable indexing for a collection, use this in conjunction with enable_indexing. Read more here.

Parameters:: collection_name (str) – collection name
Returns:: success
Return type:: bool

chainfury.components.qdrant.enable_indexing(collection_name: str, indexing_threshold: int = 20000) → bool[source]

Enable indexing for a collection, use this in conjunction with disable_indexing. Read more here.

Example

>>> from chainfury.components.qdrant import enable_indexing, disable_indexing, qdrant_write
>>> disable_indexing("my_collection")
>>> qdrant_write([[1, 2, 3] for _ in range(100)], "my_collection")
>>> enable_indexing("my_collection")

Parameters:

collection_name (str) – collection name
indexing_threshold (int, optional) – indexing threshold. Defaults to 20000.

Returns:

success

Return type:

bool

chainfury.components.qdrant.qdrant_read(embeddings: List[List[float]], collection_name: str, cutoff_score: float = 0.0, top: int = 5, limit: int = 0, offset: int = 0, filters: Dict[str, Dict[str, str]] = {}, qdrant_url: Secret = '', qdrant_api_key: Secret = '', qdrant_search_hnsw_ef: int = 0, qdrant_search_exact: bool = False, batch_search: bool = False) → Tuple[Dict[str, List[Dict[str, float | int]]], Exception | None][source]

Read from the Qdrant DB using the Qdrant client. In order to use this access via the memory_registry:

Example

>>> from chainfury import memory_registry
>>> mem = memory_registry.get_read("qdrant")
>>> sentence = "Who was the Cafavy?"
>>> out, err = mem(
        {
            "items": [sentence],
            "collection_name": "my_test_collection",
            "embedding_model": "openai-embedding"
        }
    )
>>> if err:
        print("TRACE:", out)
    else:
        print(out)

Note

batch_search is not implemented yet. There’s some issues from the qdrant_client library.

Parameters:

embeddings (List[List[float]]) – list of embeddings
collection_name (str) – collection name
cutoff_score (float, optional) – cutoff score. Defaults to 0.0.
limit (int, optional) – limit. Defaults to 3.
offset (int, optional) – offset. Defaults to 0.
qdrant_url (Secret, optional) – qdrant url or set env var QDRANT_API_URL.
qdrant_api_key (Secret, optional) – qdrant api key or set env var QDRANT_API_KEY.
qdrant_search_hnsw_ef (int, optional) – qdrant search beam size, the larger the beam size the more accurate the search, if not set uses default value.
qdrant_search_exact (bool, optional) – qdrant search exact. Defaults to False.
batch_search (bool, optional) – batch search. Defaults to False.

Returns:

list of results and error

Return type:

Tuple[List[Dict[str, Union[float, int]]], Optional[Exception]]

chainfury.components.qdrant.qdrant_write(embeddings: List[List[float]], collection_name: str, qdrant_url: Secret = '', qdrant_api_key: Secret = '', extra_payload: List[Dict[str, str]] = [], wait: bool = True, create_if_not_present: bool = True, distance: str = 'cosine') → Tuple[str, Exception | None][source]

Write to the Qdrant DB using the Qdrant client. In order to use this, access via the memory_registry:

Example

>>> from chainfury import memory_registry
>>> mem = memory_registry.get_write("qdrant")
>>> sentence = "C.P. Cavafy is widely considered the most distinguished Greek poet of the 20th century."
>>> out, err = mem(
        {
            "items": [sentence],
            "extra_payload": [
                {"data": sentence},
            ],
            "collection_name": "my_test_collection",
            "embedding_model": "openai-embedding",
            "create_if_not_present": True,
        }
    )
>>> if err:
        print("TRACE:", out)
    else:
        print(out)

Parameters:

embeddings (List[List[float]]) – list of embeddings
collection_name (str) – collection name
qdrant_url (Secret, optional) – qdrant url or set env var QDRANT_API_URL.
qdrant_api_key (Secret, optional) – qdrant api key or set env var QDRANT_API_KEY.
extra_payload (List[Dict[str, str]], optional) – extra payload. Defaults to [].
wait (bool, optional) – wait for the response. Defaults to True.
create_if_not_present (bool, optional) – create collection if not present. Defaults to True.
distance (str, optional) – distance metric. Defaults to “cosine”.

Returns:

status and error

Return type:

Tuple[str, Optional[Exception]]

chainfury.components.qdrant.recreate_collection(collection_name: str, embedding_dim: int) → bool[source]

Deletes and recreates a collection

Note

This will delete all the data in the collection, use with caution

Parameters:

collection_name (str) – collection name
embedding_dim (int) – embedding dimension

Returns:

success

Return type:

bool