fix embed stage ignoring pipeline endpoint (#1459)

Co-authored-by: Charles Blackmon-Luca <20627856+charlesbluca@users.noreply.github.com>

Author: edknv

api/api_tests/internal/transform/test_embed_text.py

@@ -201,7 +201,6 @@ def _local_embedder(texts):
         return [[0.9, 0.8]] * len(list(texts))
 
     task_cfg = {
-        "endpoint_url": None,  # Explicitly disable remote endpoint
         "embedder": _local_embedder,
         "local_batch_size": 2,
     }

api/src/nv_ingest_api/internal/transform/embed_text.py

@@ -717,18 +717,11 @@ def transform_create_text_embeddings_internal(
             - The updated DataFrame with embeddings applied.
             - A dictionary with trace information.
     """
-    # Allow task_config to explicitly override values with None by checking key presence.
-    api_key = task_config["api_key"] if "api_key" in task_config else transform_config.api_key
-    endpoint_url = (
-        task_config["endpoint_url"] if "endpoint_url" in task_config else transform_config.embedding_nim_endpoint
-    )
-    model_name = task_config["model_name"] if "model_name" in task_config else transform_config.embedding_model
-    custom_content_field = (
-        task_config["custom_content_field"]
-        if "custom_content_field" in task_config
-        else transform_config.custom_content_field
-    )
-    dimensions = task_config["dimensions"] if "dimensions" in task_config else transform_config.dimensions
+    api_key = task_config.get("api_key") or transform_config.api_key
+    endpoint_url = task_config.get("endpoint_url") or transform_config.embedding_nim_endpoint
+    model_name = task_config.get("model_name") or transform_config.embedding_model
+    custom_content_field = task_config.get("custom_content_field") or transform_config.custom_content_field
+    dimensions = task_config.get("dimensions") or transform_config.dimensions
 
     endpoint_url = endpoint_url.strip() if isinstance(endpoint_url, str) else endpoint_url
     if isinstance(endpoint_url, str) and not endpoint_url:
@@ -868,7 +861,13 @@ def _content_type_getter(row):
             else:
                 modality_batches = None
 
-            if endpoint_url:
+            if callable(embedder):
+                content_embeddings = _callable_runner(
+                    filtered_content_batches,
+                    embedder=embedder,
+                    batch_size=local_batch_size,
+                )
+            elif endpoint_url:
                 content_embeddings = _async_runner(
                     filtered_content_batches,
                     api_key,
@@ -881,12 +880,6 @@ def _content_type_getter(row):
                     modalities=modality_batches,
                     dimensions=dimensions,
                 )
-            elif callable(embedder):
-                content_embeddings = _callable_runner(
-                    filtered_content_batches,
-                    embedder=embedder,
-                    batch_size=local_batch_size,
-                )
             else:
                 raise ValueError(
                     "No embedding endpoint configured (endpoint_url/embedding_nim_endpoint are empty) "
@@ -926,7 +919,13 @@ def _content_type_getter(row):
             custom_content_list = extracted_custom_content[valid_custom_content_mask].to_list()
             custom_content_batches = _generate_batches(custom_content_list, batch_size=transform_config.batch_size)
 
-            if endpoint_url:
+            if callable(embedder):
+                custom_content_embeddings = _callable_runner(
+                    custom_content_batches,
+                    embedder=embedder,
+                    batch_size=local_batch_size,
+                )
+            elif endpoint_url:
                 custom_content_embeddings = _async_runner(
                     custom_content_batches,
                     api_key,
@@ -938,12 +937,6 @@ def _content_type_getter(row):
                     False,
                     dimensions=dimensions,
                 )
-            elif callable(embedder):
-                custom_content_embeddings = _callable_runner(
-                    custom_content_batches,
-                    embedder=embedder,
-                    batch_size=local_batch_size,
-                )
             else:
                 raise ValueError(
                     "No embedding endpoint configured (endpoint_url/embedding_nim_endpoint are empty) "