Indexing wiki

.gitignore

@@ -48,4 +48,10 @@ kotaemon
 logs/
 old_logs/
 
-data2/
+data2/
+
+*.db
+
+volumes/
+volumes/*
+!volumes/.gitkeep # Keep the placeholder

.hydra_config/config.yaml

@@ -57,6 +57,8 @@ verbose:
 paths:
   prompts_dir: ${oc.env:PROMPTS_DIR, ../prompts}
   data_dir: ${oc.env:DATA_DIR, ../data}
+  volumes_dir: ${oc.env:DOCKER_VOLUME_DIRECTORY, ../volumes}
+
 
 prompt:
   rag_sys_pmpt: rag_sys_prompt_template.txt # rag_sys_pmpt_tmpl_ifa.txt

docker-compose.yaml

@@ -12,10 +12,11 @@ x-ragondin: &ragondin_template
     - ./model_weights:/app/model_weights
     - ./data:/app/data # PDF data for RAG
     - ./ragondin:/app/ragondin # For dev mode
+    - ${DATA_VOLUME_DIR:-./volumes}:/app/volumes
 
   ports:
     - "${APP_PORT}:${APP_PORT}"
-    - 8265:8265 # for ray
+    - 8267:8265 # for ray
   env_file:
     - .env
   shm_size: 10.24gb

ragondin/components/indexer/chunker.py

@@ -42,7 +42,7 @@ async def split_document(self, doc: Document):
 **Contexte** :
 - Source : {source}
 - Première page :
-{first_page}
+{first_chunk}
 - Fragment précédent :
 {prev_chunk}
 
@@ -81,7 +81,7 @@ def __init__(
 
     async def generate_context(
         self,
-        first_page: str,
+        first_chunk: str,
         prev_chunk: str,
         chunk: str,
         source: str,
@@ -92,7 +92,7 @@ async def generate_context(
             try:
                 return await self.context_generator.ainvoke(
                     {
-                        "first_page": first_page,
+                        "first_chunk": first_chunk,
                         "prev_chunk": prev_chunk,
                         "chunk": chunk,
                         "source": source,
@@ -122,7 +122,7 @@ async def contextualize(
         Raises:
             Exception: If an error occurs during the contextualization process, it logs a warning with the error message.
         """
-        if not self.contextual_retrieval:
+        if not self.contextual_retrieval or len(chunks) == 1:
             return [chunk.page_content for chunk in chunks]
 
         try:
@@ -132,7 +132,7 @@ async def contextualize(
                 curr_chunk = chunks[i]
                 tasks.append(
                     self.generate_context(
-                        first_page=pages[0],
+                        first_chunk=chunks[0],
                         prev_chunk=prev_chunk.page_content,
                         chunk=curr_chunk.page_content,
                         source=source,
@@ -339,5 +339,5 @@ def create_chunker(
                 )
 
         # Include contextual retrieval if specified
-        chunker_params["llm"] = ChatOpenAI(**config.llm)
+        chunker_params["llm"] = ChatOpenAI(**config.vlm)
         return chunker_class(**chunker_params)

ragondin/components/indexer/indexer.py

@@ -11,19 +11,21 @@
 from .loaders.loader import DocSerializer
 from .vectordb import ConnectorFactory
 
+
 if not ray.is_initialized():
     ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
 
+
 if torch.cuda.is_available():
-    gpu, cpu = 1, 0
+    gpu, cpu = 1, 24
 else:
-    gpu, cpu = 0, 1
+    gpu, cpu = 0, 24
 
 
 @ray.remote(
     num_cpus=cpu,
     num_gpus=gpu,
-    concurrency_groups={"compute": 2, "serialize": 2, "chunk": 2},
+    concurrency_groups={"compute": 25},
     max_task_retries=2,
     max_restarts=-1,
 )
@@ -60,7 +62,6 @@ def __init__(self, config, logger, device=None) -> None:
         self.default_partition = "_default"
         self.enable_insertion = self.config.vectordb["enable"]
 
-    # @ray.method(concurrency_group="serialize")
     async def serialize(self, path: str, metadata: Optional[Dict] = {}):
         self.logger.info(f"Starting serialization of documents from {path}...")
         doc: Document = await self.serializer.serialize_document(
@@ -69,7 +70,6 @@ async def serialize(self, path: str, metadata: Optional[Dict] = {}):
         self.logger.info("Serialization completed.")
         return doc
 
-    # @ray.method(concurrency_group="chunk")
     async def chunk(self, doc: Document, file_path: str):
         if doc is not None:
             self.logger.info("Starting chunking")
@@ -109,6 +109,12 @@ async def add_file(
                 torch.cuda.empty_cache()
                 torch.cuda.ipc_collect()
 
+    def delete_partition(self, partition: str):
+        return self.vectordb.delete_partition(partition)
+
+    def check_file_exists_in_partition(self, file_id: str, partition: str):
+        return self.vectordb.file_exists(file_id=file_id, partition=partition)
+
     def delete_file(self, file_id: str, partition: str):
         """
         Deletes files from the vector database based on the provided filters.
@@ -135,7 +141,7 @@ def delete_file(self, file_id: str, partition: str):
                 self.logger.info(f"No points found for file_id: {file_id}")
                 return
             # Delete the points
-            self.vectordb.delete_points(points)
+            self.vectordb.delete_file_points(points, file_id, partition)
             self.logger.info(f"File {file_id} deleted.")
         except Exception as e:
             self.logger.error(f"Error in `delete_files` for file_id {file_id}: {e}")

ragondin/components/indexer/vectordb/__init__.py

@@ -0,0 +1 @@
+from .vectordb import *