6Ig@ddlZddlmZddlZddlZddlZddlm Z m Z m Z ddl m Z ddlmZddlmZmZddlmZddlZddlZddlmZdd lmZddlZeGd d Zy) N)embedding_functions)ListDictOptional) CrossEncoder)RecursiveCharacterTextSplitter)ThreadPoolExecutor as_completed) PyPDFLoader)datetime) load_dotenvceZdZd$dedeededededededefd Zd Z d eeeffd Z d Z ded e efdZ dZd%dededededededed efdZd&dede ede ed efdZdZded dfdZd'd eded!ed"ed eef d#Zy)( HybridSearchcollection_nameuser folder_name duckdb_pathuuid_storage_path chunk_size chunk_overlapalphac tjddzt|jzdzt|z} ||_||_||_| dz|z|_tdd|_ |j|_ | dz} tjj| stj| tj | |_t%j&tjdd } |j"j)|| |_t-j.| dz|z|_|j3t5||t6d |_y)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_efs 3/var/www/html/answerous/dependencies/chromautils.py__init__zHybridSearch.__init__sF))I&s*3t~~+>>sB3{CSS.$*!(->!>*+Q^ab002 $L0ww~~/0 KK) *%66>$00 1d,,c2 'dyy ' '  ' s A!!A+ct|jd5}tj|j|ddddy#1swYyxYw)z6 Save the UUID storage to a JSON file w)indentN)rMrrNdumpr*rPs r<_save_uuid_storagezHybridSearch._save_uuid_storageQs?$((# . 6$ IIdnnd1 5 6 6 6s #AA pdf_namec`|jjD]\}}|d|k(s|cSy)z8 Retrieve the UUID for a given PDF name rXN)r*items)r9rXuuiddetailss r< get_pdf_uuidzHybridSearch.get_pdf_uuidXs;"^^113 MD'z"h.  r>cjjdjjdjjdjjdjjd jjdfdd y #tj$rYy wxYw) 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_scorec.j|||||S)N) _bm25_score)tfdfdoc_len avg_doc_len total_docsr9s r<z,HybridSearch._setup_duckdb..s$$RWk:Nr>DOUBLE) return_typeN)r5executecreate_functionr3CatalogException)r9s`r<r6zHybridSearch._setup_duckdbas          Z[ NO  II % %O$ &  &&   s !B,,CCrbrcrdrerfk1bctj||z dz|dzz dz}||dzz||d|z ||z|z zzzz } t|| zS)z-Calculate BM25 score for a term in a document?)nplogfloat) r9rbrcrdrerfrmrnidf tf_adjusteds r<razHybridSearch._bm25_scoreseffj2o+S9A=>R!V}bAEAK+Npdf_pathpdf_uuidmetadatac |ttj}|i}tjj |j dd}|||d|j|<|jt|}|j}|jjd|g|jjd|g|jjd|id }|r|jj| g}t!5} g} t#|D]4\} } | j%| j'|j(| | |||6t+| D]} | j- dddt/j0|}|jjd |jjd |j2g|S#1swYbxYw) zG Process and upsert a PDF document into both databases N.r)rXrwry(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*rWr rOr5rjr2getdeleter enumerateappendsubmit process_pager resultpd DataFramer)r9rwrxryrXloaderpages existing_ids all_chunksexecutorfuturespage_numpagefuture chunks_dfs r< upsert_pdfzHybridSearch.upsert_pdfs  4::<(H  H77##H-33C8;!  $ x !X&   6 J   e J --11X8N1OPUV   " " ) )l ) ;  ! XG#,E"2 s$xt/@/@$RZ\dfpqr s'w/    LL,  IJ  "" # %3 s A%GG(c |jj|j}t|D]\}}|d|d|} t j } |||| j d|} |jj| g|g| g|j| ||||| | d|jj} tj| j} g}| jD]\}}|j|| |d|stj |}|j"j%dy) 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_idrxr@rrrry)termrrbz'INSERT INTO terms SELECT * FROM term_dfN)r8 split_text page_contentrr now isoformatr2addrlowerrrSeries value_countsrZrr5rj)r9rrrxryrchunks chunk_numrFrrchunk_metadataterms term_freq term_datarfreqterm_dfs r<rzHybridSearch.process_pagesh ##..t/@/@A )& 1- M Iu"2hZr)=H!J%' )(224   N  " " & &J ')* '    $$ ' )(* KKM'')E %(557II'oo/  d   (" ,,y1 !!"KL[- Mr>c^|jjd|g|jjd|g |jjd|id}|r|jj |t d |d y#t $r}t d|Yd}~,d}~wwxYw) 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.)r5rjr2rr Exceptionprint)r9rxres r<delete_collection_by_uuidz&HybridSearch.delete_collection_by_uuids  6 J   e J  >1155!8,6L&&--,-? 28* 8< = = >s?B B,B''B,rDtop_kfinal_nc |jj|gd|i|}|jj}dj dj |Dcgc]}dc}} |j j| ||jg|z|||gzj} |dd} | rHt| } t| } t|dd| Dcic]\}}|d || z | | z z z }}}ni}| jDcic]\}}|d |d }}}|rtt|j}t|j}||kDr-|jDcic]\}}|||z ||z z }}}n|Dcic]}|d }}t!|j#t!|j#z}g}|D]}|j%|d}|j%|d}||zd |z |zz}|j jd |gjj&d}|j)||dt+|dt+|d|d|||d|r|j-||}t/|ddd|Scc}wcc}}wcc}}wcc}}wcc}w)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 ? ,? distancesrrrqrr_rpz*SELECT * FROM documents WHERE chunk_id = ?r@rrry)rr@rrrycombined_score vector_scorer_c |dS)NrA)xs r<rgz%HybridSearch.search..s a8H6Ir>T)keyreverseN)r2rDrrformatjoinr5rjrfetchdfminmaxrCiterrowsvaluesrZsetkeysrilocrintrIsorted)r9rDrxrrrvector_results query_terms_ bm25_query bm25_resultsvector_distances min_distance max_distancerdistance vector_scoresrow bm25_scoresmin_bm25_scoremax_bm25_scorerHrcombined_resultsrr_r chunk_detailss r<searchzHybridSearch.search3sA //55x(6 kkm))+ 4 h F388+6QS67 8i lyy((  t++ ,{ :hRW=X X  ')  *+6q9 /0L/0L+.nU.CA.FHX*Y&Hh!x,6<,;VWWWM M '//1 3  OS. .    !3!3!56N !3!3!56N.,7+<+<+>'%u~5.>:YZZ  >IIx}I I++-.[5E5E5G1HH " H(,,Xq9L$15J#\1QY*4LLN!II--< giQ M  # #$(3"=#?@ #M.$A B)*5"0 ,( %  0 #}}U4DE &,ISWXYaZabbG7  Js J$ "J)J/6J5 J;)z search.dbzpdf_uuids.jsonirp)g?g?)NN)rp)__name__ __module__ __qualname__r&rrrrtr=rIr)rWrr]r6rarrrrrr>r<rrsd* * 4:* C* ^a* EH* il* DG* V[* XDdO6SXc]3j(c(s(S(u(Z](ch(sx(EJ( B3B(3-BRZ[_R`BloBH4MlQ#Q$Q:EcCEc3EcsEc#EccghlcmEcr>r)r-chromadb.utilsrr3pandasrnumpyrrtypingrrrsentence_transformersrlangchain.text_splitterrconcurrent.futuresr r langchain.document_loadersr r[r$r dotenvr rNrrr>r<rsH. ''.B?2  dcdcr>