wip: tune fastembed defaults

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/repository/embedding_provider_factory.py

@@ -1,5 +1,6 @@
 """Factory for creating configured semantic embedding providers."""
 
+import os
 from threading import Lock
 
 from basic_memory.config import BasicMemoryConfig
@@ -20,17 +21,54 @@
 _EMBEDDING_PROVIDER_CACHE_LOCK = Lock()
 
 
+def _available_cpu_count() -> int | None:
+    """Return the CPU budget available to this process when the runtime exposes it."""
+    process_cpu_count = getattr(os, "process_cpu_count", None)
+    if callable(process_cpu_count):
+        cpu_count = process_cpu_count()
+        if cpu_count is not None and cpu_count > 0:
+            return cpu_count
+
+    cpu_count = os.cpu_count()
+    return cpu_count if cpu_count is not None and cpu_count > 0 else None
+
+
+def _resolve_fastembed_runtime_knobs(app_config: BasicMemoryConfig) -> tuple[int | None, int | None]:
+    """Resolve FastEmbed threads/parallel from explicit config or CPU-aware defaults."""
+    configured_threads = app_config.semantic_embedding_threads
+    configured_parallel = app_config.semantic_embedding_parallel
+    if configured_threads is not None or configured_parallel is not None:
+        return configured_threads, configured_parallel
+
+    available_cpus = _available_cpu_count()
+    if available_cpus is None:
+        return None, None
+
+    # Trigger: local laptops and cloud workers expose different CPU budgets.
+    # Why: FastEmbed throughput wants enough ONNX threads to use the machine,
+    # but the multiprocessing-style ``parallel`` fan-out can add a lot of
+    # overhead for this workload and make full rebuilds slower instead of faster.
+    # Outcome: when config leaves the knobs unset, each process uses a bounded
+    # thread count and keeps FastEmbed on the simpler single-process path.
+    if available_cpus <= 2:
+        return available_cpus, 1
+
+    threads = min(8, available_cpus)
+    return threads, 1
+
+
 def _provider_cache_key(app_config: BasicMemoryConfig) -> ProviderCacheKey:
     """Build a stable cache key from provider-relevant semantic embedding config."""
+    resolved_threads, resolved_parallel = _resolve_fastembed_runtime_knobs(app_config)
     return (
         app_config.semantic_embedding_provider.strip().lower(),
         app_config.semantic_embedding_model,
         app_config.semantic_embedding_dimensions,
         app_config.semantic_embedding_batch_size,
         app_config.semantic_embedding_request_concurrency,
         app_config.semantic_embedding_cache_dir,
-        app_config.semantic_embedding_threads,
-        app_config.semantic_embedding_parallel,
+        resolved_threads,
+        resolved_parallel,
     )
 
 
@@ -61,12 +99,13 @@ def create_embedding_provider(app_config: BasicMemoryConfig) -> EmbeddingProvide
         # Deferred import: fastembed (and its onnxruntime dep) may not be installed
         from basic_memory.repository.fastembed_provider import FastEmbedEmbeddingProvider
 
+        resolved_threads, resolved_parallel = _resolve_fastembed_runtime_knobs(app_config)
         if app_config.semantic_embedding_cache_dir is not None:
             extra_kwargs["cache_dir"] = app_config.semantic_embedding_cache_dir
-        if app_config.semantic_embedding_threads is not None:
-            extra_kwargs["threads"] = app_config.semantic_embedding_threads
-        if app_config.semantic_embedding_parallel is not None:
-            extra_kwargs["parallel"] = app_config.semantic_embedding_parallel
+        if resolved_threads is not None:
+            extra_kwargs["threads"] = resolved_threads
+        if resolved_parallel is not None:
+            extra_kwargs["parallel"] = resolved_parallel
 
         provider = FastEmbedEmbeddingProvider(
             model_name=app_config.semantic_embedding_model,

src/basic_memory/repository/search_repository_base.py

@@ -800,6 +800,7 @@ async def _sync_entity_vectors_internal(
         batch_start = time.perf_counter()
         backend_name = type(self).__name__.removesuffix("SearchRepository").lower()
 
+        self._log_vector_sync_runtime_settings(backend_name=backend_name, entities_total=total_entities)
         logger.info(
             "Vector batch sync start: project_id={project_id} entities_total={entities_total} "
             "sync_batch_size={sync_batch_size} prepare_window_size={prepare_window_size}",
@@ -1595,6 +1596,51 @@ def _finalize_completed_entity_syncs(
 
         return queue_wait_seconds_total
 
+    def _log_vector_sync_runtime_settings(self, *, backend_name: str, entities_total: int) -> None:
+        """Log the resolved embedding runtime knobs before the first prepare window.
+
+        Trigger: a vector sync batch is about to start real work.
+        Why: operators need one place to confirm the provider/runtime settings that
+        this run will actually use, especially when threads/parallel are auto-tuned.
+        Outcome: the log shows the resolved values once per batch without changing
+        the hot-path control flow or adding more telemetry structure.
+        """
+        assert self._embedding_provider is not None
+
+        from basic_memory.repository.fastembed_provider import FastEmbedEmbeddingProvider
+
+        provider = self._embedding_provider
+        if isinstance(provider, FastEmbedEmbeddingProvider):
+            logger.info(
+                "Vector batch runtime settings: project_id={project_id} backend={backend} "
+                "entities_total={entities_total} provider={provider} model_name={model_name} "
+                "dimensions={dimensions} provider_batch_size={provider_batch_size} "
+                "sync_batch_size={sync_batch_size} threads={threads} "
+                "configured_parallel={configured_parallel} effective_parallel={effective_parallel}",
+                project_id=self.project_id,
+                backend=backend_name,
+                entities_total=entities_total,
+                provider=type(provider).__name__,
+                model_name=provider.model_name,
+                dimensions=provider.dimensions,
+                provider_batch_size=provider.batch_size,
+                sync_batch_size=self._semantic_embedding_sync_batch_size,
+                threads=provider.threads,
+                configured_parallel=provider.parallel,
+                effective_parallel=provider._effective_parallel(),
+            )
+            return
+
+        logger.info(
+            "Vector batch runtime settings: project_id={project_id} backend={backend} "
+            "entities_total={entities_total} provider={provider} sync_batch_size={sync_batch_size}",
+            project_id=self.project_id,
+            backend=backend_name,
+            entities_total=entities_total,
+            provider=type(provider).__name__,
+            sync_batch_size=self._semantic_embedding_sync_batch_size,
+        )
+
     def _log_vector_sync_complete(
         self,
         *,

tests/repository/test_openai_provider.py

@@ -8,6 +8,7 @@
 import pytest
 
 from basic_memory.config import BasicMemoryConfig
+import basic_memory.repository.embedding_provider_factory as embedding_provider_factory_module
 from basic_memory.repository.embedding_provider_factory import (
     create_embedding_provider,
     reset_embedding_provider_cache,
@@ -264,6 +265,52 @@ def test_embedding_provider_factory_forwards_fastembed_runtime_knobs():
     assert provider.parallel == 2
 
 
+def test_embedding_provider_factory_auto_tunes_fastembed_runtime_knobs_from_cpu_budget(monkeypatch):
+    """Unset FastEmbed runtime knobs should resolve from available CPU budget."""
+    monkeypatch.setattr(embedding_provider_factory_module.os, "process_cpu_count", lambda: 8)
+    monkeypatch.setattr(embedding_provider_factory_module.os, "cpu_count", lambda: 8)
+
+    config = BasicMemoryConfig(
+        env="test",
+        projects={"test-project": "/tmp/basic-memory-test"},
+        default_project="test-project",
+        semantic_search_enabled=True,
+        semantic_embedding_provider="fastembed",
+        semantic_embedding_threads=None,
+        semantic_embedding_parallel=None,
+    )
+
+    provider = create_embedding_provider(config)
+
+    assert isinstance(provider, FastEmbedEmbeddingProvider)
+    assert provider.threads == 8
+    assert provider.parallel == 1
+
+
+def test_embedding_provider_factory_auto_tuning_stays_conservative_on_small_cpu_budget(
+    monkeypatch,
+):
+    """Small workers should not get an oversized FastEmbed runtime footprint."""
+    monkeypatch.setattr(embedding_provider_factory_module.os, "process_cpu_count", lambda: 2)
+    monkeypatch.setattr(embedding_provider_factory_module.os, "cpu_count", lambda: 2)
+
+    config = BasicMemoryConfig(
+        env="test",
+        projects={"test-project": "/tmp/basic-memory-test"},
+        default_project="test-project",
+        semantic_search_enabled=True,
+        semantic_embedding_provider="fastembed",
+        semantic_embedding_threads=None,
+        semantic_embedding_parallel=None,
+    )
+
+    provider = create_embedding_provider(config)
+
+    assert isinstance(provider, FastEmbedEmbeddingProvider)
+    assert provider.threads == 2
+    assert provider.parallel == 1
+
+
 def test_embedding_provider_factory_reuses_provider_for_same_cache_key():
     """Factory should reuse the same provider instance for identical config values."""
     config_a = BasicMemoryConfig(
@@ -289,6 +336,36 @@ def test_embedding_provider_factory_reuses_provider_for_same_cache_key():
     assert provider_a is provider_b
 
 
+def test_embedding_provider_factory_reuses_auto_tuned_provider_for_same_cpu_budget(monkeypatch):
+    """Auto-tuned FastEmbed providers should still reuse the process cache."""
+    monkeypatch.setattr(embedding_provider_factory_module.os, "process_cpu_count", lambda: 8)
+    monkeypatch.setattr(embedding_provider_factory_module.os, "cpu_count", lambda: 8)
+
+    config_a = BasicMemoryConfig(
+        env="test",
+        projects={"test-project": "/tmp/basic-memory-test"},
+        default_project="test-project",
+        semantic_search_enabled=True,
+        semantic_embedding_provider="fastembed",
+        semantic_embedding_threads=None,
+        semantic_embedding_parallel=None,
+    )
+    config_b = BasicMemoryConfig(
+        env="test",
+        projects={"test-project": "/tmp/basic-memory-test"},
+        default_project="test-project",
+        semantic_search_enabled=True,
+        semantic_embedding_provider="fastembed",
+        semantic_embedding_threads=None,
+        semantic_embedding_parallel=None,
+    )
+
+    provider_a = create_embedding_provider(config_a)
+    provider_b = create_embedding_provider(config_b)
+
+    assert provider_a is provider_b
+
+
 @pytest.mark.asyncio
 async def test_openai_provider_runs_batches_concurrently_and_preserves_output_order(monkeypatch):
     """Concurrent request fan-out should keep batch order stable."""

tests/repository/test_semantic_search_base.py

@@ -12,6 +12,7 @@
 import pytest
 
 import basic_memory.repository.search_repository_base as search_repository_base_module
+from basic_memory.repository.fastembed_provider import FastEmbedEmbeddingProvider
 from basic_memory.repository.search_repository_base import (
     MAX_VECTOR_CHUNK_CHARS,
     SearchRepositoryBase,
@@ -702,3 +703,52 @@ async def _stub_flush(flush_jobs, entity_runtime, synced_entity_ids):
     assert histogram_names.count("vector_sync_write_seconds") == 2
     assert histogram_names.count("vector_sync_batch_total_seconds") == 1
     assert [name for name, _, _ in counter_calls].count("vector_sync_entities_total") == 1
+
+
+@pytest.mark.asyncio
+async def test_sync_entity_vectors_batch_logs_resolved_fastembed_runtime_settings(monkeypatch):
+    """Batch start should log the resolved FastEmbed knobs that shape this run."""
+    repo = _ConcreteRepo()
+    repo._semantic_enabled = True
+    repo._embedding_provider = FastEmbedEmbeddingProvider(
+        batch_size=128,
+        dimensions=384,
+        threads=4,
+        parallel=2,
+    )
+
+    async def _stub_prepare_window(entity_ids: list[int]):
+        return [
+            _PreparedEntityVectorSync(
+                entity_id=entity_id,
+                sync_start=0.0,
+                source_rows_count=1,
+                embedding_jobs=[],
+                entity_skipped=True,
+            )
+            for entity_id in entity_ids
+        ]
+
+    info_calls: list[tuple[str, dict]] = []
+
+    def _capture_info(message: str, **kwargs):
+        info_calls.append((message, kwargs))
+
+    monkeypatch.setattr(repo, "_prepare_entity_vector_jobs_window", _stub_prepare_window)
+    monkeypatch.setattr(search_repository_base_module.logger, "info", _capture_info)
+
+    result = await repo.sync_entity_vectors_batch([1])
+
+    assert result.entities_synced == 1
+    runtime_logs = [
+        kwargs
+        for message, kwargs in info_calls
+        if message.startswith("Vector batch runtime settings:")
+    ]
+    assert len(runtime_logs) == 1
+    assert runtime_logs[0]["model_name"] == "bge-small-en-v1.5"
+    assert runtime_logs[0]["provider_batch_size"] == 128
+    assert runtime_logs[0]["sync_batch_size"] == 64
+    assert runtime_logs[0]["threads"] == 4
+    assert runtime_logs[0]["configured_parallel"] == 2
+    assert runtime_logs[0]["effective_parallel"] == 2