Add bulk embeddings pool for ingest (via the pipeline settings singleton)

changelog.d/bulk-embeddings-pool.added.md

@@ -0,0 +1,13 @@
+- **Separate "bulk" embeddings microservice pool for ingest.** `MicroserviceEmbedder`
+  gained an optional `embeddings_microservice_url_bulk` setting (on its `Settings`
+  dataclass in `opencontractserver/pipeline/embedders/sent_transformer_microservice.py`,
+  seeded from the `EMBEDDINGS_MICROSERVICE_URL_BULK` env var via
+  `migrate_pipeline_settings`). The ingest Celery tasks in
+  `opencontractserver/tasks/embeddings_task.py` tag their embed calls with
+  `use_bulk_pool=True`; `_get_service_config` routes those to the bulk URL when one is
+  configured, while search queries (untagged) stay on `embeddings_microservice_url`.
+  This isolates search-query latency from batch-ingest load with no change to the
+  embedder client, base class, or any query call site. Opt-in: when the bulk URL is
+  unset the flag is a no-op and ingest stays on the query pool, so single-pool
+  deployments are unaffected. The bulk URL is configured through the same
+  `PipelineSettings` singleton as the query URL — no separate configuration pathway.

config/settings/base.py

@@ -1168,6 +1168,13 @@
 EMBEDDINGS_MICROSERVICE_URL = env(
     "EMBEDDINGS_MICROSERVICE_URL", default="http://vector-embedder:8000"
 )
+# Optional dedicated "bulk" pool for ingest embedding. This is the env binding
+# that seeds the MicroserviceEmbedder ``embeddings_microservice_url_bulk``
+# setting (via ``migrate_pipeline_settings``); the value is consumed through the
+# embedder's PipelineSettings singleton, NOT read directly here. When empty
+# (the default), batch ingest stays on EMBEDDINGS_MICROSERVICE_URL — set it only
+# to split ingest onto a separate pool while search queries stay warm.
+EMBEDDINGS_MICROSERVICE_URL_BULK = env("EMBEDDINGS_MICROSERVICE_URL_BULK", default="")
 VECTOR_EMBEDDER_API_KEY = env("VECTOR_EMBEDDER_API_KEY", default="")
 # CLIP embedder configuration (768-dimensional vectors)
 CLIP_EMBEDDER_URL = env("CLIP_EMBEDDER_URL", default="http://vector-embedder:8000")

docs/deployment/performance_tuning.md

@@ -187,6 +187,47 @@ gunicorn fleet).
 celery -A config.celery_app worker --concurrency=8 -Q ingest
 ```
 
+### Separate "bulk" embeddings pool for ingest
+
+Search queries embed one short string and need a *warm* microservice pod (a
+cold start adds seconds to every query). Batch ingest embeds thousands of
+strings and is happy to hit an autoscaled / scale-to-zero pool. Serving both
+from one URL forces a compromise.
+
+`MicroserviceEmbedder` supports a dedicated bulk pool without adding a parallel
+config pathway — the bulk URL is just another field on its `Settings`
+singleton (`embeddings_microservice_url_bulk`, seeded from the
+`EMBEDDINGS_MICROSERVICE_URL_BULK` env var via `migrate_pipeline_settings`,
+alongside `embeddings_microservice_url`). The ingest Celery tasks in
+`opencontractserver/tasks/embeddings_task.py` tag their embed calls with
+`use_bulk_pool=True`; `MicroserviceEmbedder._get_service_config` then routes
+those to the bulk URL while every (untagged) search query stays on
+`embeddings_microservice_url`. When no bulk URL is configured the flag is a
+no-op and ingest stays on the query pool, so single-pool deployments need no
+change.
+
+```bash
+# Point ingest at a separate autoscaled pool; queries stay on the warm pod.
+EMBEDDINGS_MICROSERVICE_URL=http://vector-embedder:8000          # warm queries
+EMBEDDINGS_MICROSERVICE_URL_BULK=http://vector-embedder-bulk:8000 # bulk ingest
+# then reseed the MicroserviceEmbedder singleton so it picks up the new URL:
+python manage.py migrate_pipeline_settings --component MicroserviceEmbedder --force
+```
+
+**Why `--force`:** `migrate_pipeline_settings` *preserves* existing
+`PipelineSettings` values on re-run unless `--force` is given
+(`migrate_pipeline_settings.py:233`). A deploy that ran the command once has
+already persisted `embeddings_microservice_url_bulk: ""` (its default), so a
+later plain `migrate_pipeline_settings` after setting the env var would keep the
+empty value and silently leave ingest on the query pool. `--component
+MicroserviceEmbedder` scopes the force to this one component so any hand-tuned
+DB settings on other components are left untouched.
+
+**Scope (v1):** the bulk pool reuses the query pool's `vector_embedder_api_key`
+and Cloud Run IAM auth — it is a separate URL, not a separately-credentialed
+deployment. If the bulk pool needs its own API key, that is not configurable
+yet.
+
 ## What's still slow (open work)
 
 The 80% per-doc overhead breaks down roughly as:

docs/sample_env_files/backend/local/.django

@@ -48,6 +48,12 @@ OPENAI_MODEL=gpt-4o
 # Microservice URLs
 # ------------------------------------------------------------------------------
 EMBEDDINGS_MICROSERVICE_URL=http://vector-embedder:8000
+# Optional: dedicated pool for batch ingest embedding. Seeds the embedder's
+# embeddings_microservice_url_bulk setting. After changing it, reseed with:
+#   python manage.py migrate_pipeline_settings --component MicroserviceEmbedder --force
+# (--force is required because the command otherwise preserves the existing DB
+# value). When unset, ingest uses EMBEDDINGS_MICROSERVICE_URL.
+# EMBEDDINGS_MICROSERVICE_URL_BULK=http://vector-embedder-bulk:8000
 DOCLING_PARSER_SERVICE_URL=http://docling-parser:8000/parse/
 
 # Docling

docs/sample_env_files/backend/production/.django

@@ -66,6 +66,13 @@ AUTH0_M2M_MANAGEMENT_GRANT_TYPE=client_credentials
 # bare metal, or managed service deployments, replace with the actual URLs.
 # ------------------------------------------------------------------------------
 EMBEDDINGS_MICROSERVICE_URL=http://vector-embedder:8000
+# Optional: dedicated pool for batch ingest embedding. Seeds the embedder's
+# embeddings_microservice_url_bulk setting. After changing it, reseed with:
+#   python manage.py migrate_pipeline_settings --component MicroserviceEmbedder --force
+# (--force is required because the command otherwise preserves the existing DB
+# value). When unset, ingest uses EMBEDDINGS_MICROSERVICE_URL. Set it to isolate
+# search-query latency from ingest load.
+# EMBEDDINGS_MICROSERVICE_URL_BULK=http://vector-embedder-bulk:8000
 DOCLING_PARSER_SERVICE_URL=http://docling-parser:8000/parse/
 
 # Docling

opencontractserver/pipeline/embedders/sent_transformer_microservice.py

@@ -148,6 +148,23 @@ class Settings:
                 )
             },
         )
+        embeddings_microservice_url_bulk: str = field(
+            default="",
+            metadata={
+                "pipeline_setting": PipelineSetting(
+                    setting_type=SettingType.OPTIONAL,
+                    required=False,
+                    description=(
+                        "Optional separate microservice URL for bulk ingest "
+                        "embedding. When set, batch ingest (calls tagged with "
+                        "use_bulk_pool=True) routes here while search queries "
+                        "stay on embeddings_microservice_url; falls back to "
+                        "embeddings_microservice_url when empty."
+                    ),
+                    env_var="EMBEDDINGS_MICROSERVICE_URL_BULK",
+                )
+            },
+        )
         vector_embedder_api_key: str = field(
             default="",
             metadata={
@@ -191,9 +208,21 @@ def _get_service_config(self, all_kwargs: dict) -> tuple[str, dict]:
         """
         s = self.settings if self.settings is not None else self.Settings()
 
-        service_url = all_kwargs.get(
+        query_url = all_kwargs.get(
             "embeddings_microservice_url", s.embeddings_microservice_url
         )
+        # Ingest tasks tag their calls with ``use_bulk_pool=True``. When a
+        # dedicated bulk URL is configured we route those to it; otherwise (and
+        # for every query call, which never sets the flag) we stay on the
+        # always-warm query URL, so single-pool deployments are unaffected. The
+        # bulk URL lives in the same PipelineSettings singleton as the query URL
+        # rather than a separate config pathway.
+        bulk_url = all_kwargs.get(
+            "embeddings_microservice_url_bulk", s.embeddings_microservice_url_bulk
+        )
+        use_bulk_pool = bool(all_kwargs.get("use_bulk_pool", False))
+        service_url = bulk_url if (use_bulk_pool and bulk_url) else query_url
+
         api_key = all_kwargs.get("vector_embedder_api_key", s.vector_embedder_api_key)
         use_cloud_run_iam_auth = bool(
             all_kwargs.get("use_cloud_run_iam_auth", s.use_cloud_run_iam_auth)

opencontractserver/tasks/embeddings_task.py

@@ -69,7 +69,10 @@ def _create_text_embedding(
         f"with embedder {embedder_path} (text length={len(text)})"
     )
 
-    vector = embedder.embed_text(text)
+    # Ingest routes through the dedicated bulk pool when one is configured
+    # (see MicroserviceEmbedder._get_service_config); embedders without a bulk
+    # URL ignore the flag and stay on their query URL.
+    vector = embedder.embed_text(text, use_bulk_pool=True)
 
     if vector is None:
         logger.error(
@@ -592,7 +595,8 @@ def _batch_embed_items(
 
     def _embed_one(chunk):
         texts_only = [text for _, text in chunk]
-        return chunk, embedder.embed_texts_batch(texts_only)
+        # Ingest routes through the dedicated bulk pool when configured.
+        return chunk, embedder.embed_texts_batch(texts_only, use_bulk_pool=True)
 
     # Map future -> chunk index for logging/sub-batch numbering.
     #
@@ -1050,7 +1054,8 @@ def _embed_relationship(
         embedder_path,
         len(text),
     )
-    vector = embedder.embed_text(text)
+    # Ingest routes through the dedicated bulk pool when configured.
+    vector = embedder.embed_text(text, use_bulk_pool=True)
     if vector is None:
         logger.error(
             "Embedder %s returned None for relationship %s",

opencontractserver/tests/test_batch_embedding.py

@@ -219,6 +219,35 @@ def test_basic_batch(self):
         self.assertEqual(result["failed"], 0)
         self.assertEqual(result["skipped"], 0)
 
+    def test_batch_tags_bulk_pool(self):
+        """Batch ingest tags embed_texts_batch with use_bulk_pool=True."""
+        captured: dict = {}
+
+        class RecordingBatchEmbedder(BaseEmbedder):
+            title = "Recording Batch"
+            description = "Test"
+            author = "Test"
+            dependencies = []
+            vector_size = 384
+            supported_file_types = [FileTypeEnum.TXT]
+
+            def _embed_text_impl(self, text, **all_kwargs):
+                return [0.1] * self.vector_size
+
+            def embed_texts_batch(self, texts, **kw):
+                captured.update(kw)
+                return [[0.1] * self.vector_size for _ in texts]
+
+        annots = [_make_mock_annotation(1, "Hello world")]
+        result = self._make_result()
+
+        _batch_embed_text_annotations(
+            annots, RecordingBatchEmbedder(), "test.RecordingBatchEmbedder", 50, result
+        )
+
+        self.assertEqual(result["succeeded"], 1)
+        self.assertTrue(captured.get("use_bulk_pool"))
+
     def test_empty_text_skipped(self):
         """Annotations with empty/whitespace text are skipped."""
         annots = [
@@ -711,10 +740,11 @@ def test_nan_values_handled_per_item(self, mock_post):
 class TestMicroserviceEmbedderSingleText(unittest.TestCase):
     """Test MicroserviceEmbedder._embed_text_impl and _get_service_config."""
 
-    def _make_embedder(self, api_key="", use_cloud_run=False):
+    def _make_embedder(self, api_key="", use_cloud_run=False, bulk_url=""):
         embedder = MicroserviceEmbedder()
         embedder._settings = MicroserviceEmbedder.Settings(
             embeddings_microservice_url="http://test-service:8080",
+            embeddings_microservice_url_bulk=bulk_url,
             vector_embedder_api_key=api_key,
             use_cloud_run_iam_auth=use_cloud_run,
         )
@@ -835,6 +865,24 @@ def test_get_service_config_fallback_to_settings(self):
         url, headers = embedder._get_service_config({})
         self.assertEqual(url, "http://test-service:8080")
 
+    def test_get_service_config_uses_bulk_pool_when_flagged(self):
+        """use_bulk_pool=True routes to the configured bulk URL (issue: ingest pool)."""
+        embedder = self._make_embedder(bulk_url="http://bulk-pool:9090")
+        url, _ = embedder._get_service_config({"use_bulk_pool": True})
+        self.assertEqual(url, "http://bulk-pool:9090")
+
+    def test_get_service_config_bulk_flag_without_bulk_url_falls_back(self):
+        """With the flag set but no bulk URL configured, stays on the query URL."""
+        embedder = self._make_embedder(bulk_url="")
+        url, _ = embedder._get_service_config({"use_bulk_pool": True})
+        self.assertEqual(url, "http://test-service:8080")
+
+    def test_get_service_config_ignores_bulk_url_without_flag(self):
+        """A configured bulk URL is never used for query calls (no flag)."""
+        embedder = self._make_embedder(bulk_url="http://bulk-pool:9090")
+        url, _ = embedder._get_service_config({})
+        self.assertEqual(url, "http://test-service:8080")
+
 
 class TestCalculateEmbeddingsForAnnotationBatch(unittest.TestCase):
     """Integration tests for the calculate_embeddings_for_annotation_batch task.

opencontractserver/tests/test_dual_embeddings.py

@@ -30,7 +30,7 @@ class MockEmbedder:
     is_multimodal = False
     supports_images = False
 
-    def embed_text(self, text: str) -> list[float]:
+    def embed_text(self, text: str, **kwargs) -> list[float]:
         """Return a mock embedding vector."""
         return [0.1] * 768
 
@@ -42,7 +42,7 @@ class MockCorpusEmbedder:
     is_multimodal = False
     supports_images = False
 
-    def embed_text(self, text: str) -> list[float]:
+    def embed_text(self, text: str, **kwargs) -> list[float]:
         """Return a different mock embedding vector."""
         return [0.2] * 768