fix: add retriever upsert semantics

CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 - `TextFileLoader` and `MarkdownLoader` now convert Unicode decoding and unknown encoding errors to `LoaderError`, with configurable text decoding error handling (closes #24).
 - Validate chunker size and overlap parameters in `FixedSizeChunker` and `SentenceChunker` (closes #25).
+- Ensure retrievers consistently apply last-occurrence-wins semantics for duplicate chunk IDs and return the latest content after upserts.
 
 ## [0.3.0] - 2026-09-25
 

ragframework/base.py

@@ -119,13 +119,18 @@ class Retriever(ABC):
 
     @abstractmethod
     def add(self, chunks: list[Chunk]) -> None:
-        """Add *chunks* to the index.
+        """Add or replace *chunks* in the index.
+
+        Adding a chunk whose :attr:`~Chunk.id` already exists replaces the
+        previously indexed chunk with the same ID. Retrievers therefore expose
+        upsert semantics: repeated ingestion of the same chunk ID does not
+        increase the number of indexed chunks.
 
         Chunks must have their :attr:`~Chunk.embedding` set before calling
         this method.
 
         Args:
-            chunks: Chunks to index.
+            chunks: Chunks to add or replace in the index.
         """
 
     @abstractmethod

ragframework/retriever/chroma.py

@@ -54,35 +54,42 @@ def __init__(
             raise RetrieverError(f"Could not initialize ChromaDB: {exc}") from exc
 
     def add(self, chunks: list[Chunk]) -> None:
-        """Add embedded chunks to the ChromaDB collection."""
+        """Upsert chunks; the last occurrence of each ID wins."""
         if not chunks:
             return
 
-        embeddings: list[Sequence[float]] = []
+        # Validate all chunks and deduplicate IDs.
+        # Reassigning an existing key keeps its original position,
+        # while the latest chunk becomes its value.
+        unique_chunks: dict[str, Chunk] = {}
 
         for chunk in chunks:
             if chunk.embedding is None:
                 raise RetrieverError(
                     f"Chunk '{chunk.id}' has no embedding. "
                     "Embed chunks before adding them to the retriever."
                 )
+            unique_chunks[chunk.id] = chunk
 
-            embeddings.append(chunk.embedding)
+        final_chunks = list(unique_chunks.values())
+        final_embeddings: list[Sequence[float]] = [
+            chunk.embedding for chunk in final_chunks if chunk.embedding is not None
+        ]
 
         try:
             self._collection.upsert(
-                ids=[chunk.id for chunk in chunks],
-                embeddings=embeddings,
-                documents=[chunk.content for chunk in chunks],
-                metadatas=[self._metadata(chunk) for chunk in chunks],
+                ids=[chunk.id for chunk in final_chunks],
+                embeddings=final_embeddings,
+                documents=[chunk.content for chunk in final_chunks],
+                metadatas=[self._metadata(chunk) for chunk in final_chunks],
             )
         except Exception as exc:
             raise RetrieverError(f"Failed to add chunks to ChromaDB: {exc}") from exc
 
         logger.debug(
             "Added chunks count=%d dimension=%d",
-            len(chunks),
-            len(embeddings[0]),
+            len(final_chunks),
+            len(final_embeddings[0]),
         )
 
     def retrieve(

ragframework/retriever/faiss.py

@@ -19,16 +19,18 @@
 from ragframework.utils.vectors import validate_vector
 
 logger = logging.getLogger(__name__)
+
 _INSTALL_HINT = (
-    "FAISS support requires 'ragframework[faiss]'. Install it with: pip install ragframework[faiss]"
+    "FAISS support requires 'ragframework[faiss]'. "
+    "Install it with: pip install ragframework[faiss]"
 )
 
 
 class FAISSRetriever(Retriever):
     """Retrieve chunks with an HNSW index and cosine similarity.
 
     Chunk and query vectors are L2-normalized before being passed to an
-    ``IndexHNSWFlat`` configured for inner product.  Inner product between
+    ``IndexHNSWFlat`` configured for inner product. Inner product between
     normalized vectors is cosine similarity.
 
     Args:
@@ -41,9 +43,15 @@ class FAISSRetriever(Retriever):
         RetrieverError: If an HNSW parameter is not positive.
     """
 
-    def __init__(self, m: int = 32, ef_construction: int = 40, ef_search: int = 16) -> None:
+    def __init__(
+        self,
+        m: int = 32,
+        ef_construction: int = 40,
+        ef_search: int = 16,
+    ) -> None:
         if not _FAISS_AVAILABLE:
             raise ImportError(_INSTALL_HINT)
+
         if any(
             not isinstance(value, int) or isinstance(value, bool) or value <= 0
             for value in (m, ef_construction, ef_search)
@@ -53,87 +61,205 @@ def __init__(self, m: int = 32, ef_construction: int = 40, ef_search: int = 16)
         self._m = m
         self._ef_construction = ef_construction
         self._ef_search = ef_search
+
         self._chunks: list[Chunk] = []
         self._dimension: int | None = None
         self._index: Any | None = None
 
+        # Maps logical chunk IDs to their current physical FAISS labels.
+        # Old labels remain in the FAISS index as tombstones after replacement.
+        self._id_to_index: dict[str, int] = {}
+
+        # Logical chunk ID -> position in self._chunks.
+        self._id_to_position: dict[str, int] = {}
+
+    def _build_index(
+        self,
+        vectors: np.ndarray[Any, Any],
+        dimension: int,
+    ) -> Any:
+        """Build an HNSW index from normalized vectors."""
+        index = faiss.IndexHNSWFlat(
+            dimension,
+            self._m,
+            faiss.METRIC_INNER_PRODUCT,
+        )
+
+        index.hnsw.efConstruction = self._ef_construction
+        index.hnsw.efSearch = self._ef_search
+
+        index.add(vectors)
+
+        return index
+
     def add(self, chunks: list[Chunk]) -> None:
-        """Validate, normalize, and add embedded chunks to the HNSW index.
+        """Validate and upsert chunks; the last duplicate ID in a batch wins.
 
-        An empty batch is a no-op.  Every vector in a batch is validated before
-        either the FAISS index or the chunk mapping is changed.
+        Existing IDs retain their logical position. New IDs are appended.
+        Replaced physical FAISS vectors become tombstones.
         """
         if not chunks:
             return
 
         expected_dimension = self._dimension
-        vectors: list[np.ndarray[Any, np.dtype[np.float32]]] = []
+        validated: dict[str, tuple[Chunk, np.ndarray[Any, np.dtype[np.float32]]]] = {}
+
+        # Validate the entire batch before mutating retriever state.
+        # Dict reassignment keeps the first-seen order but the last value.
         for chunk in chunks:
             if chunk.embedding is None:
                 raise RetrieverError(
                     f"Chunk '{chunk.id}' has no embedding. "
                     "Embed chunks before adding them to the retriever."
                 )
-            vector = validate_vector(chunk.embedding, f"Chunk '{chunk.id}' embedding")
+
+            vector = validate_vector(
+                chunk.embedding,
+                f"Chunk '{chunk.id}' embedding",
+            )
+
             if expected_dimension is None:
                 expected_dimension = int(vector.shape[0])
             elif vector.shape[0] != expected_dimension:
                 raise RetrieverError(
-                    f"Chunk '{chunk.id}' embedding has dimension {vector.shape[0]}; "
-                    f"expected {expected_dimension}."
+                    f"Chunk '{chunk.id}' embedding has dimension "
+                    f"{vector.shape[0]}; expected {expected_dimension}."
                 )
-            vectors.append(vector)
+
+            validated[chunk.id] = (chunk, vector)
 
         assert expected_dimension is not None
-        matrix = np.ascontiguousarray(np.stack(vectors), dtype=np.float32)
-        new_index = self._index
-        try:
-            if new_index is None:
-                new_index = faiss.IndexHNSWFlat(
-                    expected_dimension, self._m, faiss.METRIC_INNER_PRODUCT
-                )
-                new_index.hnsw.efConstruction = self._ef_construction
-                new_index.hnsw.efSearch = self._ef_search
-            new_index.add(matrix)
-        except Exception as exc:
-            raise RetrieverError("FAISS failed to add chunk embeddings.") from exc
 
-        self._index = new_index
-        self._dimension = expected_dimension
-        self._chunks.extend(chunks)
+        unique_chunks = [item[0] for item in validated.values()]
+        vectors = [item[1] for item in validated.values()]
+        matrix = np.ascontiguousarray(
+            np.stack(vectors),
+            dtype=np.float32,
+        )
+
+        # Build a fresh index for the first insertion.
+        if self._index is None:
+            try:
+                new_index = self._build_index(matrix, expected_dimension)
+            except Exception as exc:
+                raise RetrieverError("FAISS failed to add chunk embeddings.") from exc
+
+            self._index = new_index
+            self._dimension = expected_dimension
+
+            for physical_label, chunk in enumerate(unique_chunks):
+                logical_position = len(self._chunks)
+                self._chunks.append(chunk)
+                self._id_to_index[chunk.id] = physical_label
+                self._id_to_position[chunk.id] = logical_position
+
+        else:
+            # Add only the final version of each ID to FAISS.
+            start_label = int(self._index.ntotal)
+            try:
+                self._index.add(matrix)
+            except Exception as exc:
+                raise RetrieverError("FAISS failed to add chunk embeddings.") from exc
+
+            # Update the logical mapping only after FAISS succeeds.
+            for offset, chunk in enumerate(unique_chunks):
+                physical_label = start_label + offset
+                existing_position = self._id_to_position.get(chunk.id)
+
+                if existing_position is None:
+                    logical_position = len(self._chunks)
+                    self._chunks.append(chunk)
+                    self._id_to_position[chunk.id] = logical_position
+                else:
+                    logical_position = existing_position
+                    self._chunks[logical_position] = chunk
+
+                self._id_to_index[chunk.id] = physical_label
+
+            self._dimension = expected_dimension
+
         logger.debug(
             "Added chunks count=%d dimension=%d",
-            len(chunks),
+            len(unique_chunks),
             matrix.shape[1],
         )
 
-    def retrieve(self, query_embedding: list[float], top_k: int = 5) -> list[Chunk]:
-        """Return up to ``top_k`` chunks ordered by approximate cosine similarity.
-
-        Non-positive ``top_k`` values return an empty list.  An empty index also
-        returns an empty list.  Invalid FAISS labels (for example ``-1`` when
-        fewer than ``top_k`` neighbours exist) are ignored.
-        """
+    def retrieve(
+        self,
+        query_embedding: list[float],
+        top_k: int = 5,
+    ) -> list[Chunk]:
+        """Retrieve active chunks using progressively expanded FAISS searches."""
         if not isinstance(top_k, int) or isinstance(top_k, bool):
             raise RetrieverError("top_k must be an integer.")
-        if top_k <= 0 or self._index is None or self._dimension is None:
+
+        if top_k <= 0 or self._index is None or self._dimension is None or not self._chunks:
             return []
 
         query = validate_vector(query_embedding, "Query embedding")
+
         if query.shape[0] != self._dimension:
             raise RetrieverError(
-                f"Query embedding has dimension {query.shape[0]}; expected {self._dimension}."
+                f"Query embedding has dimension {query.shape[0]}; " f"expected {self._dimension}."
             )
 
-        count = min(top_k, len(self._chunks))
+        total = int(self._index.ntotal)
+
+        # Current physical label -> current logical position.
+        active_labels = {
+            physical_label: self._id_to_position[chunk_id]
+            for chunk_id, physical_label in self._id_to_index.items()
+            if chunk_id in self._id_to_position
+        }
+
+        # Start with a bounded over-fetch, not every physical vector.
+        search_count = min(total, max(top_k * 2, top_k + 8))
+        candidates: dict[int, tuple[float, int, Chunk]] = {}
+
         try:
-            _scores, labels = self._index.search(query.reshape(1, -1), count)
+            while True:
+                scores, labels = self._index.search(
+                    query.reshape(1, -1),
+                    search_count,
+                )
+
+                for score, label in zip(
+                    scores[0],
+                    labels[0],
+                    strict=True,
+                ):
+                    physical_label = int(label)
+                    if physical_label < 0:
+                        continue
+
+                    logical_position = active_labels.get(physical_label)
+                    if logical_position is None:
+                        continue
+
+                    chunk = self._chunks[logical_position]
+                    candidates[physical_label] = (
+                        float(score),
+                        logical_position,
+                        chunk,
+                    )
+
+                if len(candidates) >= min(top_k, len(self._chunks)):
+                    break
+
+                if search_count >= total:
+                    break
+
+                search_count = min(total, search_count * 2)
+
         except Exception as exc:
             raise RetrieverError("FAISS failed to search the index.") from exc
 
-        return [
-            self._chunks[int(label)] for label in labels[0] if 0 <= int(label) < len(self._chunks)
-        ]
+        ordered = sorted(
+            candidates.values(),
+            key=lambda item: (-item[0], item[1]),
+        )
+
+        return [chunk for _score, _position, chunk in ordered[:top_k]]
 
     def __len__(self) -> int:
         """Return the number of indexed chunks."""