Allow to configure more than one AI service

langstream-core/src/main/java/ai/langstream/impl/agents/ai/GenAIToolKitFunctionAgentProvider.java

@@ -280,6 +280,11 @@ public void generateSteps(
                 }
             };
     private static final Map<String, StepConfigurationInitializer> STEP_TYPES;
+    protected static final String SERVICE_VERTEX = "vertex-configuration";
+    protected static final String SERVICE_HUGGING_FACE = "hugging-face-configuration";
+    protected static final String SERVICE_OPEN_AI = "open-ai-configuration";
+    protected static final List<String> AI_SERVICES =
+            List.of(SERVICE_VERTEX, SERVICE_HUGGING_FACE, SERVICE_OPEN_AI);
 
     static {
         final Map<String, StepConfigurationInitializer> steps = new HashMap<>();
@@ -369,33 +374,47 @@ interface DataSourceConfigurationGenerator {
 
     private void generateAIProvidersConfiguration(
             Application applicationInstance,
+            Map<String, Object> originalConfiguration,
             Map<String, Object> configuration,
             ComputeClusterRuntime clusterRuntime,
             PluginsRegistry pluginsRegistry) {
         // let the user force the provider or detect it automatically
-        String service = (String) configuration.remove("service");
-        for (Resource resource : applicationInstance.getResources().values()) {
-            Map<String, Object> configurationCopy =
-                    clusterRuntime.getResourceImplementation(resource, pluginsRegistry);
-            switch (resource.type()) {
-                case "vertex-configuration":
-                    if (service == null || service.equals("vertex")) {
-                        configuration.put("vertex", configurationCopy);
-                    }
-                    configuration.put("vertex", configurationCopy);
-                    break;
-                case "hugging-face-configuration":
-                    if (service == null || service.equals("hugging-face")) {
-                        configuration.put("huggingface", configurationCopy);
-                    }
-                    break;
-                case "open-ai-configuration":
-                    if (service == null || service.equals("open-ai")) {
-                        configuration.put("openai", configurationCopy);
-                    }
-                    break;
-                default:
-                    // ignore
+        String resourceId = (String) originalConfiguration.get("ai-service");
+        if (resourceId != null) {
+            Resource resource = applicationInstance.getResources().get(resourceId);
+            log.info("Generating ai provider configuration for {}", resourceId);
+            if (resource != null) {
+                if (!AI_SERVICES.contains(resource.type())) {
+                    throw new IllegalArgumentException(
+                            "Resource " + resourceId + " is not in types: " + AI_SERVICES);
+                }
+                Map<String, Object> resourceImplementation =
+                        clusterRuntime.getResourceImplementation(resource, pluginsRegistry);
+                final String configKey =
+                        switch (resource.type()) {
+                            case SERVICE_VERTEX -> "vertex";
+                            case SERVICE_HUGGING_FACE -> "huggingface";
+                            case SERVICE_OPEN_AI -> "openai";
+                            default -> throw new IllegalStateException();
+                        };
+                configuration.put(configKey, resourceImplementation);
+            } else {
+                throw new IllegalArgumentException("Resource " + resourceId + " not found");
+            }
+        } else {
+            for (Resource resource : applicationInstance.getResources().values()) {
+                Map<String, Object> configurationCopy =
+                        clusterRuntime.getResourceImplementation(resource, pluginsRegistry);
+                final String configKey =
+                        switch (resource.type()) {
+                            case SERVICE_VERTEX -> "vertex";
+                            case SERVICE_HUGGING_FACE -> "huggingface";
+                            case SERVICE_OPEN_AI -> "openai";
+                            default -> null;
+                        };
+                if (configKey != null) {
+                    configuration.put(configKey, configurationCopy);
+                }
             }
         }
     }
@@ -443,7 +462,11 @@ protected Map<String, Object> computeAgentConfiguration(
         Map<String, Object> configuration = new HashMap<>();
 
         generateAIProvidersConfiguration(
-                executionPlan.getApplication(), configuration, clusterRuntime, pluginsRegistry);
+                executionPlan.getApplication(),
+                originalConfiguration,
+                configuration,
+                clusterRuntime,
+                pluginsRegistry);
 
         generateSteps(
                 module,

langstream-core/src/main/java/ai/langstream/impl/parser/ModelBuilder.java

@@ -298,7 +298,7 @@ private static void parseConfiguration(String content, Application application)
                             id = r.name();
                         }
                         if (id == null) {
-                            throw new RuntimeException("Resource 'name' is required");
+                            throw new RuntimeException("Resource 'name' or 'id' is required");
                         }
                         application.getResources().put(id, r);
                     });

langstream-core/src/main/java/ai/langstream/impl/resources/AIProvidersResourceProvider.java

@@ -36,6 +36,7 @@ public class AIProvidersResourceProvider implements ResourceNodeProvider {
 
     private static final Set<String> SUPPORTED_TYPES =
             Set.of("open-ai-configuration", "hugging-face-configuration", "vertex-configuration");
+    protected static final ObjectMapper MAPPER = new ObjectMapper();
 
     @Override
     public Map<String, Object> createImplementation(
@@ -77,7 +78,7 @@ private void validateVertexConfigurationResource(Resource resource) {
         }
         if (!serviceAccountJson.isEmpty()) {
             try {
-                new ObjectMapper().readValue(serviceAccountJson, Map.class);
+                MAPPER.readValue(serviceAccountJson, Map.class);
             } catch (Exception e) {
                 throw new IllegalArgumentException(
                         "Invalid JSON for field serviceAccountJson in " + describe(resource).get(),

langstream-e2e-tests/src/test/java/ai/langstream/tests/TextCompletionsIT.java

@@ -36,13 +36,24 @@ public class TextCompletionsIT extends BaseEndToEndTest {
 
     @BeforeAll
     public static void checkCredentials() {
-        appEnv =
+        try {
+            appEnv =
+                    getAppEnvMapFromSystem(
+                            List.of("OPEN_AI_ACCESS_KEY", "OPEN_AI_URL", "OPEN_AI_PROVIDER"));
+        } catch (Throwable t) {
+            // no openai - try vertex
+            appEnv =
+                    getAppEnvMapFromSystem(
+                            List.of(
+                                    "VERTEX_AI_URL",
+                                    "VERTEX_AI_TOKEN",
+                                    "VERTEX_AI_REGION",
+                                    "VERTEX_AI_PROJECT"));
+        }
+
+        appEnv.putAll(
                 getAppEnvMapFromSystem(
-                        List.of(
-                                "OPEN_AI_ACCESS_KEY",
-                                "OPEN_AI_URL",
-                                "OPEN_AI_TEXT_COMPLETIONS_MODEL",
-                                "OPEN_AI_PROVIDER"));
+                        List.of("TEXT_COMPLETIONS_MODEL", "TEXT_COMPLETIONS_SERVICE")));
     }
 
     @Test

langstream-e2e-tests/src/test/java/ai/langstream/tests/util/BaseEndToEndTest.java

@@ -617,8 +617,12 @@ private static void installLangStream(boolean authentication) {
                         : (LANGSTREAM_TAG.equals("latest-dev")
                                 ? "langstream"
                                 : "ghcr.io/langstream");
+
         final String imagePullPolicy =
-                LANGSTREAM_TAG.equals("latest-dev") ? "Never" : "IfNotPresent";
+                switch (LANGSTREAM_TAG) {
+                    case "latest-dev" -> "Never";
+                    default -> "IfNotPresent";
+                };
         final Map<String, String> controlPlaneConfig = new HashMap<>();
         if (authentication) {
             KeyPairGenerator kpg = KeyPairGenerator.getInstance("RSA");
@@ -756,28 +760,27 @@ private static void awaitDeploymentReady(String deploymentName) {
                             if (pod == null) {
                                 return false;
                             }
-                            final boolean ready = Readiness.getInstance().isReady(pod);
-                            if (!ready) {
-                                String podLogs;
-                                try {
-                                    podLogs =
-                                            getPodLogs(
-                                                    pod.getMetadata().getName(),
-                                                    pod.getMetadata().getNamespace(),
-                                                    15);
-                                } catch (Throwable e) {
-                                    podLogs = "failed to get pod logs: " + e.getMessage();
-                                }
-                                log.info(
-                                        "pod {} not ready, logs:\n{}",
-                                        pod.getMetadata().getName(),
-                                        podLogs);
-                                return false;
-                            }
-                            return true;
+                            return checkPodReadiness(pod);
                         });
     }
 
+    private static boolean checkPodReadiness(Pod pod) {
+        final boolean ready = Readiness.getInstance().isReady(pod);
+        if (!ready) {
+            String podLogs;
+            try {
+                podLogs =
+                        getPodLogs(
+                                pod.getMetadata().getName(), pod.getMetadata().getNamespace(), 30);
+            } catch (Throwable e) {
+                podLogs = "failed to get pod logs: " + e.getMessage();
+            }
+            log.info("pod {} not ready, logs:\n{}", pod.getMetadata().getName(), podLogs);
+            return false;
+        }
+        return true;
+    }
+
     @SneakyThrows
     private static void awaitApiGatewayReady() {
         log.info("waiting for api gateway to be ready");
@@ -792,8 +795,10 @@ protected static String getPodLogs(String podName, String namespace, int tailing
                 namespace,
                 tailingLines,
                 (container, logs) -> {
-                    sb.append("container: ").append(container).append("\n");
-                    sb.append(logs).append("\n");
+                    if (!logs.isBlank()) {
+                        sb.append("container: ").append(container).append("\n");
+                        sb.append(logs).append("\n");
+                    }
                 });
         return sb.toString();
     }
@@ -1122,7 +1127,9 @@ private static void deployLocalApplicationAndAwaitReady(
 
         awaitApplicationReady(applicationId, expectedNumExecutors);
         Awaitility.await()
-                .atMost(1, TimeUnit.MINUTES)
+                .atMost(2, TimeUnit.MINUTES)
+                .pollDelay(Duration.ZERO)
+                .pollInterval(5, TimeUnit.SECONDS)
                 .untilAsserted(
                         () -> {
                             log.info("waiting new executors to be ready");
@@ -1148,7 +1155,7 @@ private static void deployLocalApplicationAndAwaitReady(
                             Assertions.assertNotEquals(podUids, currentUids);
                             for (Pod pod : pods) {
                                 log.info("checking pod readiness {}", pod.getMetadata().getName());
-                                assertTrue(Readiness.getInstance().isReady(pod));
+                                assertTrue(checkPodReadiness(pod));
                             }
                         });
     }

langstream-e2e-tests/src/test/java/ai/langstream/tests/util/kafka/RemoteKafkaProvider.java

@@ -63,12 +63,10 @@ public void cleanup() {
         final List<String> topics = getTopics();
         if (!topics.isEmpty()) {
             log.info("Deleting topics: {}", topics);
-            for (String topic : topics) {
-                admin.deleteTopics(List.of(topic)).all().get();
-            }
-
-            // admin.deleteTopics(topics).all().get();
+            admin.deleteTopics(topics).all().get();
+            log.info("Deleted topics: {}", topics);
         }
+        log.info("Topics after cleanup: {}", getTopics());
     }
 
     @Override

langstream-e2e-tests/src/test/resources/apps/text-completions/configuration.yaml

@@ -23,3 +23,11 @@ configuration:
         url: "{{ secrets.open-ai.url }}"
         access-key: "{{ secrets.open-ai.access-key }}"
         provider: "{{ secrets.open-ai.provider }}"
+    - type: "vertex-configuration"
+      name: "Google Vertex AI configuration"
+      id: "vertex"
+      configuration:
+        url: "{{ secrets.vertex-ai.url }}"
+        token: "{{ secrets.vertex-ai.token }}"
+        region: "{{ secrets.vertex-ai.region }}"
+        project: "{{ secrets.vertex-ai.project }}"

langstream-e2e-tests/src/test/resources/apps/text-completions/pipeline.yaml

@@ -31,12 +31,12 @@ pipeline:
     type: "ai-text-completions"
     output: "ls-test-history-topic"
     configuration:
-      model: "{{{secrets.open-ai.text-completions-model}}}"
+      ai-service: "{{{secrets.text-completions.service}}}"
+      model: "{{{secrets.text-completions.model}}}"
       completion-field: "value.answer"
       log-field: "value.prompt"
       stream-to-topic: "ls-test-output-topic"
       stream-response-completion-field: "value"
       min-chunks-per-message: 20
-      messages:
-        - role: user
-          content: "{{% value.question}}"
+      prompt:
+        - "{{% value.question}}"

langstream-e2e-tests/src/test/resources/secrets/secret1.yaml

@@ -20,6 +20,12 @@ secrets:
     id: secret1
     data:
       value-key: "${SECRET1_VK}"
+  - name: text-completions
+    id: text-completions
+    data:
+      model: "${TEXT_COMPLETIONS_MODEL}"
+      service: "${TEXT_COMPLETIONS_SERVICE}"
+
   - name: cassandra
     id: cassandra
     data:
@@ -28,12 +34,17 @@ secrets:
       port: "${CASSANDRA_PORT}"
   - id: open-ai
     data:
-      access-key: "${OPEN_AI_ACCESS_KEY}"
-      url: "${OPEN_AI_URL}"
-      provider: "${OPEN_AI_PROVIDER}"
-      embeddings-model: "${OPEN_AI_EMBEDDINGS_MODEL}"
-      chat-completions-model: "${OPEN_AI_CHAT_COMPLETIONS_MODEL}"
-      text-completions-model: "${OPEN_AI_TEXT_COMPLETIONS_MODEL}"
+      access-key: "${OPEN_AI_ACCESS_KEY:-}"
+      url: "${OPEN_AI_URL:-}"
+      provider: "${OPEN_AI_PROVIDER:-openai}"
+      embeddings-model: "${OPEN_AI_EMBEDDINGS_MODEL:-}"
+      chat-completions-model: "${OPEN_AI_CHAT_COMPLETIONS_MODEL:-}"
+  - id: vertex-ai
+    data:
+      url: "${VERTEX_AI_URL:-}"
+      token: "${VERTEX_AI_TOKEN:-}"
+      region: "${VERTEX_AI_REGION:-}"
+      project: "${VERTEX_AI_PROJECT:-}"
   - id: astra
     data:
       clientId: "${ASTRA_CLIENT_ID}"