Merge pull request #18 from Learn2RAG/feature/align_with_importer

hannred · web-flow · commit cd24f502fc5a · 2026-03-25T15:50:49.000+01:00
clean up json loader, use content_hash instead of chunk_hash
diff --git a/learn2rag/pipeline/data/load_kcenter_dump.py b/learn2rag/pipeline/data/load_kcenter_dump.py
@@ -2,6 +2,7 @@
 import json
 from pathlib import Path
 from bs4 import BeautifulSoup
+import hashlib
 
 root = Path("/home/large-file-storage/download-uknowit")
 paths = list(root.rglob("raw-document.json"))
@@ -18,16 +19,19 @@
     html_content = raw.get("content", "")
     soup = BeautifulSoup(html_content, "html.parser")
     content = soup.get_text(separator=" ", strip=True)
+    content_hash = hashlib.sha256(content.encode('utf-8')).hexdigest()
     
     new = {
         "metadata": {
             "source": raw.get("id"),
-            "content_hash": "",
+            "content_hash": content_hash,
             "source_path": raw['categories'][0]['branchText'],
             "file_extension": "",
             "process_date": "",
             "process_time": "",
             "loader_type": "KCenterFormatter",
+            "loader_id": "my_kcenter_dump",
+            "document_id": document.get("docId"),
             "title": raw.get("title"),
             "summary": raw.get("summary"),
             "keywords": document.get("keywords"),
diff --git a/learn2rag/pipeline/ingestion.py b/learn2rag/pipeline/ingestion.py
@@ -13,7 +13,7 @@
 from qdrant_client.models import PointStruct, Filter, FieldCondition, MatchValue, SparseVector, VectorParams, MultiVectorConfig, MultiVectorComparator, Distance
 
 
-from . import loaders
+from . import json_loader
 from .embeddings import create_embeddings
 
 
@@ -29,11 +29,12 @@ def get_chunks_metadata(chunks: list[Document], item: str) -> Iterator[str]:
         logging.warning('%d out of %d chunks are missing "%s" in metadata; using empty string', missing, len(chunks), item)
 
 
-def point_exists(qdrant: Qdrant, collection_name: str, path: str, chunk_hash:str) -> bool:
+def point_exists(qdrant: Qdrant, collection_name: str, loader_id: str, path: str, content_hash:str) -> bool:
     filter = Filter(
         must=[
+            FieldCondition(key="loader_id", match=MatchValue(value=loader_id)),
             FieldCondition(key="path", match=MatchValue(value=path)),
-            FieldCondition(key="content_hash", match=MatchValue(value=chunk_hash)),
+            FieldCondition(key="content_hash", match=MatchValue(value=content_hash)),
         ]
     )
     result, _ = qdrant.client.scroll(
@@ -96,27 +97,23 @@ def payload(sample: dict[str, Any]) -> dict[str, str]:
     return {
         "content": sample["page_content"],
         "path": sample["metadata"]["source"],
-        "content_hash": sample["chunk_hash"],
+        "content_hash": sample["metadata"]["content_hash"],
         "title": sample["metadata"].get("title",""),
         "uri": sample["metadata"].get("uri",""),
         "loader_id": sample["metadata"]["loader_id"],
         "document_id": sample["metadata"].get("document_id", "")
     }
 
 def index(user_config: dict[str, Any], opt_config: dict[str, Any]) -> None:
-    # TODO: enable list of file paths in loader and adapt user_config
-    # Load the documents from pdf
-    # all_documents = loaders.sync_pdf_loader(user_config["file_path"])
-    # TODO: use ifdt loader to load pdf in json, then:
     logging.info('Loading documents')
-    all_documents = loaders.json_loader(user_config['imported_documents_file_path'])
+    all_documents = json_loader.json_loader(user_config['imported_documents_file_path'])
 
     # Split documents into chunks
     logging.info('Splitting documents into chunks')
     text_splitter = RecursiveCharacterTextSplitter(
         chunk_size=opt_config["chunk_size"], chunk_overlap=opt_config["chunk_overlap"]
     )
-    chunks = text_splitter.split_documents(all_documents)
+    chunks = text_splitter.split_documents(all_documents)[:2]
 
     collection_name = user_config["collection_name"]
 
@@ -126,7 +123,6 @@ def index(user_config: dict[str, Any], opt_config: dict[str, Any]) -> None:
         opt_config=opt_config
     )
 
-
     chunks_content = [chunk.page_content for chunk in chunks]
     if len(opt_config["multi_search"]) > 0 and opt_config["query_mode"] == "multi":
         chunks_metadata =  {}
@@ -140,8 +136,6 @@ def index(user_config: dict[str, Any], opt_config: dict[str, Any]) -> None:
             else:
                 raise TypeError(f"dense_vecs must be np.ndarray, got {type(dense_vecs)}")
                 
-    # TODO: hash if you want to monitore changes in metadata
-    chunk_hash = [hashlib.md5(chunk.page_content.encode()).hexdigest() for chunk in chunks]
     # Todo: handle different vector lengths for batch encoding when using sparse vectors
 
     logging.info('Creating embeddings...')
@@ -158,40 +152,38 @@ def index(user_config: dict[str, Any], opt_config: dict[str, Any]) -> None:
     if isinstance(embeddings, dict) and "dense_vecs" in embeddings:
         if opt_config["search_mode"] == "dense":
             chunks_with_embeddings = [
-                dict(chunk) | {"dense_vec": dense, "chunk_hash": c_hash}
-                for chunk, dense, c_hash in zip(chunks, embeddings["dense_vecs"], chunk_hash)
+                dict(chunk) | {"dense_vec": dense}
+                for chunk, dense in zip(chunks, embeddings["dense_vecs"])
             ]
         if opt_config["search_mode"] == "dense_sparse":
             chunks_with_embeddings = [
                 dict(chunk)
-                | {"dense_vec": dense, "lexical_weights": sparse, "chunk_hash": c_hash}
-                for chunk, dense, sparse, c_hash in zip(
+                | {"dense_vec": dense, "lexical_weights": sparse}
+                for chunk, dense, sparse in zip(
                     chunks,
                     list(embeddings["dense_vecs"]),
                     list(embeddings["lexical_weights"]),
-                    chunk_hash,
                 )
             ]
         if opt_config["search_mode"] == "dense_sparse_colbert":
             chunks_with_embeddings = [
                 dict(chunk)
-                | {"dense_vec": dense, "lexical_weights": sparse, "colbert_vecs": colbert, "chunk_hash": c_hash}
-                for chunk, dense, sparse, colbert, c_hash in zip(
+                | {"dense_vec": dense, "lexical_weights": sparse, "colbert_vecs": colbert}
+                for chunk, dense, sparse, colbert in zip(
                     chunks,
                     list(embeddings["dense_vecs"]),
                     list(embeddings["lexical_weights"]),
                     list(embeddings['colbert_vecs']),
-                    chunk_hash,
                 )
             ]
     else:
         chunks_with_embeddings = [
-            dict(chunk) | {"dense_vec": dense, "chunk_hash": c_hash}
-            for chunk, dense, c_hash in zip(chunks, embeddings, chunk_hash)
+            dict(chunk) | {"dense_vec": dense}
+            for chunk, dense in zip(chunks, embeddings)
         ]
 
     for sample in chunks_with_embeddings:
-        if not point_exists(qdrant, collection_name, sample['metadata']['source'], sample['chunk_hash']):
+        if not point_exists(qdrant, collection_name, sample['metadata']['loader_id'], sample['metadata']['source'], sample['metadata']['content_hash']):
             if opt_config["search_mode"] == "dense_sparse":
                 insert_dense_sparse(qdrant, collection_name, sample)
             elif opt_config["search_mode"] == "dense_sparse_colbert":
diff --git a/learn2rag/pipeline/json_loader.py b/learn2rag/pipeline/json_loader.py
@@ -0,0 +1,12 @@
+from langchain_community.document_loaders import JSONLoader
+from langchain_core.documents import Document
+
+
+def json_loader(file_path: str) -> list[Document]:
+    loader = JSONLoader(
+        file_path,
+        jq_schema=".[]",
+        content_key="content",
+        metadata_func=lambda record, meta: record.get("metadata", {}),
+    )
+    return loader.load()
diff --git a/learn2rag/pipeline/loaders.py b/learn2rag/pipeline/loaders.py
diff --git a/learn2rag/pipeline/main.py b/learn2rag/pipeline/main.py
@@ -17,7 +17,7 @@
 
     from .config import user_config, opt_config
 
-    #ingestion.index(user_config, opt_config)
+    ingestion.index(user_config, opt_config)
 
     if opt_config["query_mode"] == "multi":
         # in query_mode 'multi' different querys for each vector in the multi-vector are allowed
