o xq@g@@sddlZddlmZddlZddlZddlZddlm Z m Z m Z ddl m Z ddlmZddlmZmZddlmZddlZddlZddlmZdd lmZddlZeGd d d ZdS) N)embedding_functions)ListDictOptional) CrossEncoder)RecursiveCharacterTextSplitter)ThreadPoolExecutor as_completed) PyPDFLoader)datetime) load_dotenvc@seZdZd8dedeeded ed ed ed ed efddZddZ deeeffddZ ddZ dede efddZ ddZd9deded ed!ed"ed#ed$edefd%d&Zd:d(ed)e ed*e edefd+d,Zd-d.Zd)edd'fd/d0Zd;d3ed)ed4ed5edeef d6d7Zd'S)< HybridSearch search.dbpdf_uuids.json?collection_nameuser folder_name duckdb_pathuuid_storage_path chunk_size chunk_overlapalphac Cstddt|jdt|} ||_||_||_| d||_tddd|_ | |_ | d} tj | s=t| tj| d|_tjtdd d } |jj|| d |_t| d||_|t||td d |_dS)z5 Initialize the hybrid search system DB_PATH/z$cross-encoder/ms-marco-MiniLM-L-6-v2i) max_lengthz /chroma_db)pathOPENAI_API_KEYztext-embedding-3-small)api_key model_name)nameembedding_functionF)rrlength_functionis_separator_regexN)osgetenvstr user_coderrrrrreranker_model_load_uuid_storage pdf_uuidsrexistsmakedirschromadbPersistentClient chroma_clientrOpenAIEmbeddingFunctionget_or_create_collectionchroma_collectionduckdbconnectconn _setup_duckdbrlen text_splitter) selfrrrrrrrrrpersist_directory openai_efr>9/var/www/html/answerit/server/dependencies/chromautils.py__init__s6$    zHybridSearch.__init__cs:|jfdd|D}t||D]\}}||d<q|S)Ncsg|]}|dfqS)contentr>).0chunkqueryr>r? Csz)HybridSearch.reranker..reranker_score)r*predictzip)r;rEresultsscoresscorerCr>rDr?rerankerAs zHybridSearch.rerankerreturncCsHtj|jr"t|jd }t|WdS1swYiS)z8 Load the UUID storage from a JSON file rN)r&rr-ropenjsonloadr;filer>r>r?r+Hs  zHybridSearch._load_uuid_storagecCsDt|jd}tj|j|ddWddS1swYdS)z6 Save the UUID storage to a JSON file w)indentN)rPrrQdumpr,rSr>r>r?_save_uuid_storageQs"zHybridSearch._save_uuid_storagepdf_namecCs,|jD]\}}|d|kr|SqdS)z8 Retrieve the UUID for a given PDF name rZN)r,items)r;rZuuiddetailsr>r>r? get_pdf_uuidXs  zHybridSearch.get_pdf_uuidcstjdjdjdjdjdzjjdfddd d Wd Stjy9Yd Sw) z2 Set up DuckDB tables and indexes aG CREATE TABLE IF NOT EXISTS documents ( chunk_id VARCHAR PRIMARY KEY, pdf_uuid VARCHAR, content TEXT, page_number INTEGER, chunk_number INTEGER, created_at TIMESTAMP, metadata JSON ) z CREATE TABLE IF NOT EXISTS terms ( term VARCHAR, chunk_id VARCHAR, tf INTEGER, UNIQUE(term, chunk_id) ) z CREATE TABLE IF NOT EXISTS doc_stats ( collection_name VARCHAR PRIMARY KEY, total_chunks INTEGER, avg_chunk_length DOUBLE ) z>CREATE INDEX IF NOT EXISTS idx_pdf_uuid ON documents(pdf_uuid)z2CREATE INDEX IF NOT EXISTS idx_term ON terms(term) bm25_scorecs|||||S)N) _bm25_score)tfdfdoc_len avg_doc_len total_docsr;r>r?sz,HybridSearch._setup_duckdb..DOUBLE) return_typeN)r7executecreate_functionr5CatalogExceptionrfr>rfr?r8as    zHybridSearch._setup_duckdb??rarbrcrdrek1bc CsRt||d|dd}||d||d||||} t|| S)z-Calculate BM25 score for a term in a documentr)nplogfloat) r;rarbrcrdrerorpidf tf_adjustedr>r>r?r`s( zHybridSearch._bm25_scoreNpdf_pathpdf_uuidmetadatac CsF|dur tt}|duri}tj|dd}|||d|j|<|t |}| }|j d|g|j d|g|j jd|idd }|rT|j j|d g}t,} g} t|D]\} } | | |j| | |||q`t| D]} | qwWdn1swYt|}|j d |j d |jg|S) zG Process and upsert a PDF document into both databases N.r)rZrwry(DELETE FROM documents WHERE pdf_uuid = ?WDELETE FROM terms WHERE chunk_id IN (SELECT chunk_id FROM documents WHERE pdf_uuid = ?)rxwhereidsrz-INSERT INTO documents SELECT * FROM chunks_dfz INSERT OR REPLACE INTO doc_stats SELECT ? as collection_name, COUNT(*) as total_chunks, AVG(LENGTH(content)) as avg_chunk_length FROM documents )r(r\uuid4r&rbasenamesplitr,rYr rRr7rjr4getdeleter enumerateappendsubmit process_pager resultpd DataFramer)r;rwrxryrZloaderpages existing_ids all_chunksexecutorfuturespage_numpagefuture chunks_dfr>r>r? upsert_pdfsL      zHybridSearch.upsert_pdfc Cs|j|j}t|D]d\}}|d|d|} t} |||| d|} |jj| g|g| gd| | ||||| | d| } t | } g}| D] \}}| || |dqT|rot |}|jdq dS) z| Process a single page and add chunks to databases. This method will run in parallel for each page. z-pz-c)rx page_number chunk_number created_at)r documents metadatas)chunk_idrxrArrrry)termrraz'INSERT INTO terms SELECT * FROM term_dfN)r: split_text page_contentrr now isoformatr4addrlowerrrSeries value_countsr[rr7rj)r;rrrxryrchunks chunk_numrCrrchunk_metadataterms term_freq term_datarfreqterm_dfr>r>r?rsN     zHybridSearch.process_pagec Cs|jd|g|jd|gz|jjd|idd}|r%|jj|dWnty?}z td|WYd}~nd}~wwtd |d dS) z Delete all data associated with a given PDF UUID from DuckDB and ChromaDB. Args: pdf_uuid (str): The unique identifier of the PDF to delete. r{r|rxr}rrz$Error while deleting from ChromaDB: Nz"All data associated with PDF UUID z has been deleted.)r7rjr4rr Exceptionprint)r;rxrer>r>r?delete_collection_by_uuids,z&HybridSearch.delete_collection_by_uuidrEtop_kfinal_nc s|jj|gd|i|d}|}dddd|D}|j|||jg||||g } |dd} | rTt | t | fd d t |d d| D} ni} d d | D} | rt | t | kr~fd d | D} ndd | D} t| t| B} g}| D]?}| |d}| |d}||d||}|jd|g jd}|||dt|dt|d|d|||dq|r|||}t|ddddd|S)z= Perform hybrid search within a specific PDF rx) query_textsr~ n_resultsa# WITH pdf_stats AS ( -- Calculate statistics specific to the PDF SELECT COUNT(*) AS total_chunks, AVG(LENGTH(content)) AS avg_chunk_length FROM documents WHERE pdf_uuid = ? -- Filter by the specific PDF ), doc_lengths AS ( -- Get the length of each chunk within the specific PDF SELECT d.chunk_id, LENGTH(d.content) AS doc_len FROM documents d WHERE d.pdf_uuid = ? -- Filter by the specific PDF ), term_stats AS ( -- Calculate term statistics specific to the PDF SELECT t.term, t.chunk_id, t.tf, COUNT(*) OVER (PARTITION BY t.term) AS df -- Document frequency within this PDF FROM terms t JOIN documents d ON t.chunk_id = d.chunk_id WHERE t.term IN ({}) AND d.pdf_uuid = ? -- Filter terms by the specific PDF ), scores AS ( -- Calculate BM25 score for each chunk in the specific PDF SELECT t.chunk_id, SUM(bm25_score( t.tf, t.df, dl.doc_len, ps.avg_chunk_length, -- Use PDF-level average chunk length ps.total_chunks -- Use PDF-level total chunks )) AS bm25_score FROM term_stats t JOIN doc_lengths dl ON t.chunk_id = dl.chunk_id CROSS JOIN pdf_stats ps -- Join with PDF-specific statistics GROUP BY t.chunk_id ) SELECT d.chunk_id, d.content, d.page_number, d.chunk_number, d.metadata, COALESCE(s.bm25_score, 0) AS bm25_score -- Default to 0 if no BM25 score FROM documents d LEFT JOIN scores s ON d.chunk_id = s.chunk_id WHERE d.pdf_uuid = ? -- Filter by the specific PDF ORDER BY bm25_score DESC -- Sort by BM25 score LIMIT ? ,cSsg|]}dqS)?r>)rB_r>r>r?rFusz'HybridSearch.search.. distancesrcs&i|]\}}|d|qS)rqr>)rBrdistance) max_distance min_distancer>r? sz'HybridSearch.search..rcSsi|] \}}|d|dqS)rr_r>)rBrrowr>r>r?rs cs"i|] \}}||qSr>r>)rBrrL)max_bm25_scoremin_bm25_scorer>r?rscSsi|]}|dqS)rr>)rBrr>r>r?rsrqz*SELECT * FROM documents WHERE chunk_id = ?rArrry)rrArrrycombined_score vector_scorer_cSs|dS)NrGr>)xr>r>r?rgsz%HybridSearch.search..T)keyreverseN)r4rErrformatjoinr7rjrfetchdfminmaxrIiterrowsvaluesr[setkeysrilocrintrMsorted)r;rErxrrrvector_results query_terms bm25_query bm25_resultsvector_distances vector_scores bm25_scoresrcombined_resultsrrr_r chunk_detailsr>)rrrrr?search3sv 46           zHybridSearch.search)rrrrr)rmrn)NN)rrr)__name__ __module__ __qualname__r(rrrrtr@rMr+rYrr^r8r`rrrrr>r>r>r?r s0,  ,5$D6(r )r/chromadb.utilsrr5pandasrnumpyrrtypingrrrsentence_transformersrlangchain.text_splitterrconcurrent.futuresrr langchain.document_loadersr r\r&r dotenvr rQr r>r>r>r?s"