fix: improve indexing robustness and type safety

- Disable persistent storage temporarily due to multi-threading issues
- Improve document deletion using collection-level delete
- Add better type hints and class attributes
- Update build backend and gitignore

Author: ErikBjare

.gitignore

@@ -19,6 +19,7 @@ wheels/
 *.egg-info/
 .installed.cfg
 *.egg
+.*cache/
 
 # Virtual Environments
 .env

gptme_rag/indexing/indexer.py

@@ -3,7 +3,8 @@
 from pathlib import Path
 
 import chromadb
-from chromadb.api import Collection
+from chromadb import Collection
+from chromadb.api import ClientAPI
 from chromadb.config import Settings
 
 from .document import Document
@@ -15,25 +16,34 @@
 class Indexer:
     """Handles document indexing and embedding storage."""
 
+    client: ClientAPI | None = None
+    collection: Collection
+    processor: DocumentProcessor
+    is_persistent: bool = False
+
     def __init__(
         self,
         persist_directory: Path | None,
         collection_name: str = "default",
         chunk_size: int = 1000,
         chunk_overlap: int = 200,
     ):
-        if persist_directory:
+        enable_persist = False
+        if persist_directory and enable_persist:
+            self.is_persistent = True
             persist_directory = Path(persist_directory).expanduser().resolve()
             persist_directory.mkdir(parents=True, exist_ok=True)
             logger.debug(f"Using persist directory: {persist_directory}")
 
         settings = Settings(
             allow_reset=True,  # Allow resetting for testing
-            is_persistent=persist_directory is not None,
+            is_persistent=self.is_persistent,
             anonymized_telemetry=False,
         )
 
-        if persist_directory:
+        # FIXME: persistent storage doesn't work in multi-threaded environments.
+        #        ("table segments already exist", "database is locked", among other issues)
+        if persist_directory and enable_persist:
             settings.persist_directory = str(persist_directory)
             logger.debug(f"Using persist directory: {persist_directory}")
             self.client = chromadb.PersistentClient(
@@ -44,6 +54,7 @@ def __init__(
             self.client = chromadb.Client(settings)
 
         def create_collection():
+            assert self.client
             return self.client.get_or_create_collection(
                 name=collection_name, metadata={"hnsw:space": "cosine"}
             )
@@ -67,11 +78,11 @@ def create_collection():
 
     def __del__(self):
         """Cleanup when the indexer is destroyed."""
-        try:
-            self.client.reset()
-        except Exception as e:
-            if "Resetting is not allowed" not in e.args[0]:
-                logger.exception("Error resetting ChromaDB client")
+        if self.client:
+            try:
+                self.client.reset()
+            except Exception as e:
+                logger.warning(f"Error during cleanup: {e}")
 
     def add_document(self, document: Document, timestamp: int | None = None) -> None:
         """Add a single document to the index."""

gptme_rag/indexing/watcher.py

@@ -178,18 +178,14 @@ def _process_updates(self) -> None:
                         # Read current content for verification
                         current_content = path.read_text()
 
-                        # Remove all old versions from the index
-                        old_docs = self.indexer.search(
-                            "", n_results=100, where={"source": canonical_path}
-                        )[0]
-                        for doc in old_docs:
-                            if doc.doc_id is not None:
-                                if self.indexer.delete_document(doc.doc_id):
-                                    logger.info(f"Deleted old version: {doc.doc_id}")
-                                else:
-                                    logger.warning(
-                                        f"Failed to delete old version: {doc.doc_id}"
-                                    )
+                        # Clear all documents with this source path
+                        try:
+                            self.indexer.collection.delete(
+                                where={"source": canonical_path}
+                            )
+                            logger.info(f"Cleared old versions for: {canonical_path}")
+                        except Exception as e:
+                            logger.warning(f"Error clearing old versions: {e}")
 
                         # Index the new version with retries
                         max_attempts = 3
@@ -263,6 +259,14 @@ def start(self) -> None:
             if not path.exists():
                 logger.warning(f"Watch path does not exist: {path}")
                 continue
+            # Clear any existing documents for this path
+            try:
+                for file in path.glob(self.event_handler.pattern):
+                    canonical_path = str(file.resolve())
+                    self.indexer.collection.delete(where={"source": canonical_path})
+            except Exception as e:
+                logger.warning(f"Error clearing existing documents: {e}")
+
             # Index existing files
             self.indexer.index_directory(path, self.event_handler.pattern)
             # Set up watching

pyproject.toml

@@ -54,4 +54,4 @@ check_untyped_defs = true
 
 [build-system]
 requires = ["poetry-core"]
-build-backend = "poetry-core.backend"
+build-backend = "poetry.core.masonry.api"