o CibI@sddlZddlZddlZddlmZddlZddlZddlZddl m Z dddZ dddZ ddd Z d d ZGd d d ZGdddZdejejdddddddddddfddZdS)N) ThreadPool) get_invlistcCs~|j\}}|j|fks Jt|tjr|d9}||jksJ|dur/|j|fks*Jt|}||t||t|dS)zP Add elements to an IVF index, where the assignment is already computed N)shape isinstancefaissIndexBinaryIVFdswig_ptradd_core) index_ivfxaidsnr ro/var/www/html/fyndo/python/python_agents/rag_suite/venv/lib/python3.10/site-packages/faiss/contrib/ivf_tools.pyadd_preassigneds   rcCs|j\}}t|tjr|d9}d}nd}||jksJ|j||jfks%J|dur4tj||jf|d}n |j||jfks>J|||||S)z Perform a search in the IVF index, with predefined lists to search into. Supports indexes with pretransforms (as opposed to the IndexIVF.search_preassigned, that cannot be applied with pretransform). rint32float32Ndtype) rrrrr nprobenpzerossearch_preassigned)r xqklist_nos coarse_disrr dis_typerrrr#s  rcCs|j\}}t|tjr|d9}d}nd}|dur#tj||jf|d}n |j||jfks-J||jks4J|j||jfks>Jt|}tj } | || ||| || ||t |j |d } t| d} t |j|  } t |j|  } | | | fS)z[ Perform a range search in the IVF index, with predefined lists to search into rrrNr)rrrrremptyrr RangeSearchResultr range_search_preassigned rev_swig_ptrlimscopyint distanceslabels)r r radiusrrrr r resspr' num_resultsdistindicesrrrr%=s,       r%cCs|jdkr|j}||||n|j|jksJ|j}d|_t|j}|j |||_t |dr?|j ||S|g|_ |S)zQ replace the IVF quantizer with a flat quantizer and return the old quantizerrFreferenced_objects)ntotal quantizer reconstruct_ntrainaddnlist own_fieldsrdowncast_indexthisownhasattrr2append)r new_quantizer centroidsold_own old_quantizerrrrreplace_ivf_quantizercs        rCc@s~eZdZdZ  dddZddZdd Zd d Zd d ZddZ ddZ ddZ ddZ ddZ ddZddZddZdS) BigBatchSearcherz Object that manages all the data related to the computation except the actual within-bucket matching and the organization of the computation (parallel or not) rFcCs`||_g|_||_||_||_t|j}tjt |||d|_ dgd|_ t |_ |_dS)N)keep_maxr)verbosetictocrindex use_float16ris_similarity_metric metric_type ResultHeaplenrht_accutime t_displayt0)selfrIrrrGrJrErrr__init__s  zBigBatchSearcher.__init__cCst|_dSN)rQ t_accu_t0rTrrr start_t_accuszBigBatchSearcher.start_t_accucCs |j|t|j7<dSrV)rPrQrW)rTrrrr stop_t_accus zBigBatchSearcher.stop_t_accucCs.|tf|_|jdkrt|ddddSdS)Nr Tendflush)rQrHrGprint)rTnamerrrtics zBigBatchSearcher.ticcCs:|j\}}t|}|jdkrt|d|dd|S)Nrz: .3fz s)rHrQrGr_)rTr`rSdtrrrtocs   zBigBatchSearcher.toccCs|jdks|dkrt|jdkrdStdt|jdd|d|jjd|jd d d |jdd d |jd d d|jdd d|jdd dddt|_dS)Nr!ig?[z.1fz s] list /z times prep q rrbz prep b z comp z res z wait r[Tr\)rGrQrRr_rSrIr8rP)rTlrrrreports$ $     zBigBatchSearcher.reportcCs|dd}t|j}tj||jjfdd}td||D] }t|||}|jj |j|||jj\}}||||<q| ||_ dS)Nzcoarse quantizationirrr) rarNrrr#rIrrangeminr4searchrdq_assign)rTbsnqroi0i1q_dis_i q_assign_irrrcoarse_quantizations   z$BigBatchSearcher.coarse_quantizationcCsz|d|j}|d7}tj|j|jjddd|_|j|_|j dkr-t d|jd|jdd|_|`| dS)Nz bucket sortr!)nbucketntrz number of -1s:) rarormatrix_bucket_sort_inplacerIr8 bucket_limsravel query_idsrGr_rd)rTrorrrreorder_assigns    zBigBatchSearcher.reorder_assignc Cst}|j}|j||j|d}}|j||}|j|}|jr+||j|}t}t|j |\} } |j durA| } n| | } |j rS| d} | d}t} |jd||7<|jd| |7<||| | fS)z4 prepare the queries and database items for bucket lr!Nfloat16r)rQrIr{r}r by_residualr4 reconstructrinvlists decode_funcr|rJastyperP) rTrjrSrIrrrsq_subsetxq_lt1list_idsxb_lt2rrrprepare_buckets&       zBigBatchSearcher.prepare_bucketcCsX|durdSt}|dur|}n||}|j||||jdt|7<dS)z,add the bucket results to the heap structureNrh)rQrOadd_result_subsetrP)rTrDrIrSrrradd_results_to_heapsz$BigBatchSearcher.add_results_to_heapcCs|jj|jj|jjfSrV)rrrIrr8rXrrrsizes_in_checkpointsz$BigBatchSearcher.sizes_in_checkpointcCsD|d}t|||jj|jjfdt|ddt||dS)Nz.tmp)sizes cur_list_norOwbr") pickledumprrOrropenosreplace)rTfnamertmpnamerrrwrite_checkpoints z!BigBatchSearcher.write_checkpointcCs\tt|d}|d|ksJ|dd|jjdd<|dd|jjdd<|dS)NrbrrOrr!r)rloadrrrOrr)rTrckprrrread_checkpoints z BigBatchSearcher.read_checkpointN)rF)__name__ __module__ __qualname____doc__rUrYrZrardrkrvr~rrrrrrrrrrD{s"      rDc@s,eZdZdZdejejfddZddZdS) BlockComputerz computation within one bucket knn_functioncCs||_|jtjkrt|j|j}dd}d}nL|jtjkr;t|j|j j |j j |j}|j |_ |j j }d|_ |j}n)|jtjkr[t|j|jj|j}|j|_|jj }d|_ |j}n td|jd||_|dkrmdn||_||_||_||_||_dS)NcSs |dS)Nr)view)r rrrs z(BlockComputer.__init__..FTz index type z not supportedrI)rI __class__r IndexIVFFlat IndexFlatr rL IndexIVFPQIndexPQpqMnbitsdecode is_trainedrIndexIVFScalarQuantizerIndexScalarQuantizersqqtype RuntimeError index_helprmethodpairwise_distancesknn)rTrIrrrrrrrrrrUs8    zBlockComputer.__init__c Ks|jj}|jdks|jdkrd}}||fS|jdkr6t||jjt||j_ |j ||\}}||fS|jdkrI|j |||d}d}||fS|jdkr]|j |||fd|i|\}}||fS)NrrIr)metricrr) rIrLsizerrcopy_array_to_vectorrcodesrNr3rnrr) rTrrrr extra_argsrLrrrrr block_search3s     zBlockComputer.block_searchN) rrrrrrrrUrrrrrr s #rrFr@r"c* sj}|dvs J|j}t||tdj}t|tdjtdj}|||}|dkrKtd|d|d|d |d |d d d t|||dt|||dj _ j _ | durm n| _ |dur{j}| dur||fdjfksJtj| rtd| | }td|ntd|dkrt||D]3}||\}}}}t}|||\}}jdt|7<||||qnP|dkrGfdd}|}d} td}!t||D]=}||!|| |df}"|\}}}}|||\}}d||||f} |"}dqj| |!n||dksRJdkrZt ntj | dkrjt| }#fdd}$t!" i fdd}%|$g|}&g} td}!d}'t||D]z}||&}(|!|$| |f}'g} | dkr|(D]\}}}}|||\}}| #||||fqnt$|# |%|(} d||krdd|'}&d| dur|| dkrtd|%| |q| D]})j|)q|!dkr*| dkr3|#&dj'()j'j*j'j+fS)a Search queries xq in the IVF index, with a search function that collects batches of query vectors per inverted list. This can be faster than the regular search indexes. Supports IVFFlat, IVFPQ and IVFScalarQuantizer. Supports three computation methods: method = "index": build a flat index and populate it separately for each index method = "pairwise_distances": decompress codes and compute all pairwise distances for the queries and index and add result to heap method = "knn_function": decompress codes and compute knn results for the queries threaded=0: sequential execution threaded=1: prefetch next bucket while computing the current one threaded>1: prefetch this many buckets at a time. compute_threads>1: the knn function will get an additional thread_no that tells which worker should handle this. In threaded mode, the computation is tiled with the bucket perparation and the writeback of results (useful to maximize GPU utilization). use_float16: convert all matrices to float16 (faster for GPU gemm) q_assign: override coarse assignment, should be a matrix of size nq * nprobe checkpointing (only for threaded > 1): checkpoint: file where the checkpoints are stored checkpoint_freq: when to perform checkpoinging. Should be a multiple of threaded start_list, end_list: process only a subset of invlists )rIrrrint64rrzmemory: queries z assign z result z total z = i@rbz GiB)rGrJ)rrrNzrecovering checkpointz start at listz$no checkpoint: starting from scratchrgr!cs*|dur j||jkr|SdS)z` perform the addition for the previous bucket and prefetch the next (if applicable) N)rr8r)to_addrj)bbsrIrradd_results_and_prefetchs    z2big_batch_search..add_results_and_prefetchricsffdd}dkr||n||f}t|t|j}tj|}dkr1||S)Ncs |D] }|dur j|qdSrV)r)rta)rrr add_resultss  zMbig_batch_search..add_results_and_prefetch_batch..add_resultsr) apply_asyncrlrmr8listrget)rrjradd_a next_listsr-)rrI list_step prefetch_map prefetch_poolprefetch_threadsrradd_results_and_prefetch_batchs  z8big_batch_search..add_results_and_prefetch_batchc s|\}}}}zt}Wn%ty3t}|t<Wdn1s,wYYnwj||||d\}}||||fS)N) thread_id) threading get_identKeyErrorrNr)bucketrrrrtidrr)comprthread_id_to_seqthread_id_to_seq_lockrrdo_comps   z!big_batch_search..do_compzwriting checkpoint %sz finalize heap),rnbytesrNrritemsizer_rDrrrrvror~r8rpathexistsrrlrkrrQrrPrrrrYrZrclosemaprLockr>rrrarOfinalizerdrr)*rIrrrrrrGthreadedrJrcomputation_threadsro checkpointcheckpoint_freq start_listend_listcrash_atr mem_queries mem_assignmem_resmem_totrjrrrrt0irrrprefetched_bucketrpoolprefetched_bucket_a comp_poolrrprefetched_bucketsprefetched_buckets_abucketsrr) rrrIrrrrrrrrbig_batch_searchEs4                                     rrV)rQrrmultiprocessing.poolrrnumpyrrfaiss.contrib.inspect_toolsrrrr%rCrDrrrrrrrrs<     &: