Underground-Digital/Workflow-Engine

1import logging2import time3 4import click5from celery import shared_task6 7from core.rag.datasource.vdb.vector_factory import Vector8from models.dataset import Dataset9from services.dataset_service import DatasetCollectionBindingService10 11 12@shared_task(queue="dataset")13def delete_annotation_index_task(annotation_id: str, app_id: str, tenant_id: str, collection_binding_id: str):14    """15    Async delete annotation index task16    """17    logging.info(click.style("Start delete app annotation index: {}".format(app_id), fg="green"))18    start_at = time.perf_counter()19    try:20        dataset_collection_binding = DatasetCollectionBindingService.get_dataset_collection_binding_by_id_and_type(21            collection_binding_id, "annotation"22        )23 24        dataset = Dataset(25            id=app_id,26            tenant_id=tenant_id,27            indexing_technique="high_quality",28            collection_binding_id=dataset_collection_binding.id,29        )30 31        try:32            vector = Vector(dataset, attributes=["doc_id", "annotation_id", "app_id"])33            vector.delete_by_metadata_field("annotation_id", annotation_id)34        except Exception:35            logging.exception("Delete annotation index failed when annotation deleted.")36        end_at = time.perf_counter()37        logging.info(38            click.style("App annotations index deleted : {} latency: {}".format(app_id, end_at - start_at), fg="green")39        )40    except Exception as e:41        logging.exception("Annotation deleted index failed:{}".format(str(e)))42