Add batched vector sync orchestration across repositories

Author: phernandez

src/basic_memory/config.py

@@ -173,6 +173,11 @@ class BasicMemoryConfig(BaseSettings):
         description="Batch size for embedding generation.",
         gt=0,
     )
+    semantic_embedding_sync_batch_size: int = Field(
+        default=64,
+        description="Batch size for vector sync orchestration flushes.",
+        gt=0,
+    )
     semantic_embedding_cache_dir: str | None = Field(
         default=None,
         description="Optional cache directory for FastEmbed model artifacts.",

src/basic_memory/db.py

@@ -128,8 +128,13 @@ async def _run_semantic_embedding_backfill(
             project_id=project_id,
             app_config=app_config,
         )
-        for entity_id in entity_ids:
-            await search_repository.sync_entity_vectors(entity_id)
+        batch_result = await search_repository.sync_entity_vectors_batch(entity_ids)
+        if batch_result.entities_failed > 0:
+            logger.warning(
+                "Automatic semantic embedding backfill encountered entity failures: "
+                f"project={project_name}, failed={batch_result.entities_failed}, "
+                f"failed_entity_ids={batch_result.failed_entity_ids}"
+            )
 
     logger.info(
         "Automatic semantic embedding backfill complete: "

src/basic_memory/repository/postgres_search_repository.py

@@ -58,6 +58,9 @@ def __init__(
         self._semantic_enabled = self._app_config.semantic_search_enabled
         self._semantic_vector_k = self._app_config.semantic_vector_k
         self._semantic_min_similarity = self._app_config.semantic_min_similarity
+        self._semantic_embedding_sync_batch_size = (
+            self._app_config.semantic_embedding_sync_batch_size
+        )
         self._embedding_provider = embedding_provider
         self._vector_dimensions = 384
         self._vector_tables_initialized = False

src/basic_memory/repository/search_repository.py

@@ -7,14 +7,15 @@
 """
 
 from datetime import datetime
-from typing import List, Optional, Protocol
+from typing import Any, Callable, List, Optional, Protocol
 
 from sqlalchemy import Result
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from basic_memory.config import BasicMemoryConfig, ConfigManager, DatabaseBackend
 from basic_memory.repository.postgres_search_repository import PostgresSearchRepository
 from basic_memory.repository.search_index_row import SearchIndexRow
+from basic_memory.repository.search_repository_base import VectorSyncBatchResult
 from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository
 from basic_memory.schemas.search import SearchItemType, SearchRetrievalMode
 
@@ -69,6 +70,14 @@ async def sync_entity_vectors(self, entity_id: int) -> None:
         """Sync semantic vector chunks for an entity."""
         ...
 
+    async def sync_entity_vectors_batch(
+        self,
+        entity_ids: list[int],
+        progress_callback: Optional[Callable[[int, int, int], Any]] = None,
+    ) -> VectorSyncBatchResult:
+        """Sync semantic vector chunks for a batch of entities."""
+        ...
+
     async def execute_query(self, query, params: dict) -> Result:
         """Execute a raw SQL query."""
         ...