LangStream · eolivelli · Sep 15, 2023 · Sep 15, 2023 · Sep 15, 2023 · Sep 15, 2023
diff --git a/dev/stop-port-forward.sh b/dev/stop-port-forward.sh
@@ -0,0 +1,20 @@
+#
+#
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# Port forward LangStream control plane and Gateway
+pkill -f "kubectl port-forward svc/langstream-control-plane 8090:8090" || true
+pkill -f "kubectl port-forward svc/langstream-api-gateway 8091:8091" || true
diff --git a/langstream-agents/langstream-ai-agents/pom.xml b/langstream-agents/langstream-ai-agents/pom.xml
@@ -105,11 +105,6 @@
       <artifactId>netty-common</artifactId>
       <version>${netty.version}</version>
     </dependency>
-    <dependency>
-      <groupId>io.netty</groupId>
-      <artifactId>netty-handler</artifactId>
-      <version>${netty.version}</version>
-    </dependency>
     <dependency>
       <groupId>io.netty</groupId>
       <artifactId>netty-handler-proxy</artifactId>

diff --git a/...i-agents/src/main/java/ai/langstream/ai/agents/services/impl/OpenAICompletionService.java b/...i-agents/src/main/java/ai/langstream/ai/agents/services/impl/OpenAICompletionService.java
@@ -59,9 +59,9 @@ public CompletableFuture<ChatCompletions> getChatCompletions(
                                         .map(
                                                 message ->
                                                         new com.azure.ai.openai.models.ChatMessage(
-                                                                        ChatRole.fromString(
-                                                                                message.getRole()))
-                                                                .setContent(message.getContent()))
+                                                                ChatRole.fromString(
+                                                                        message.getRole()),
+                                                                message.getContent()))
                                         .collect(Collectors.toList()))
                         .setMaxTokens(getInteger("max-tokens", null, options))
                         .setTemperature(getDouble("temperature", null, options))

diff --git a/...eam-ai-agents/src/main/java/com/datastax/oss/streaming/ai/util/TransformFunctionUtil.java b/...eam-ai-agents/src/main/java/com/datastax/oss/streaming/ai/util/TransformFunctionUtil.java
@@ -17,8 +17,8 @@
 
 import com.azure.ai.openai.OpenAIClient;
 import com.azure.ai.openai.OpenAIClientBuilder;
-import com.azure.ai.openai.models.NonAzureOpenAIKeyCredential;
 import com.azure.core.credential.AzureKeyCredential;
+import com.azure.core.credential.KeyCredential;
 import com.azure.core.http.HttpClient;
 import com.azure.core.http.HttpHeaders;
 import com.azure.core.http.HttpPipeline;
@@ -107,8 +107,7 @@ public static OpenAIClient buildOpenAIClient(OpenAIConfig openAIConfig) {
         if (openAIConfig.getProvider() == OpenAIProvider.AZURE) {
             openAIClientBuilder.credential(new AzureKeyCredential(openAIConfig.getAccessKey()));
         } else {
-            openAIClientBuilder.credential(
-                    new NonAzureOpenAIKeyCredential(openAIConfig.getAccessKey()));
+            openAIClientBuilder.credential(new KeyCredential(openAIConfig.getAccessKey()));
         }
         if (openAIConfig.getUrl() != null) {
             openAIClientBuilder.endpoint(openAIConfig.getUrl());

diff --git a/...eam-ai-agents/src/test/java/ai/langstream/ai/agents/services/impl/OpenAIProviderTest.java b/...eam-ai-agents/src/test/java/ai/langstream/ai/agents/services/impl/OpenAIProviderTest.java
@@ -42,7 +42,7 @@ class OpenAIProviderTest {
     void testStreamingCompletion(WireMockRuntimeInfo vmRuntimeInfo) throws Exception {
 
         stubFor(
-                post("/openai/deployments/gpt-35-turbo/chat/completions?api-version=2023-03-15-preview")
+                post("/openai/deployments/gpt-35-turbo/chat/completions?api-version=2023-08-01-preview")
                         .willReturn(
                                 okJson(
                                         """

diff --git a/...tream-ai-agents/src/test/java/com/datastax/oss/pulsar/functions/transforms/GenAITest.java b/...tream-ai-agents/src/test/java/com/datastax/oss/pulsar/functions/transforms/GenAITest.java
@@ -291,7 +291,7 @@ void testChatCompletionsWithLogField() throws Exception {
         assertEquals("result", valueAvroRecord.get("completion").toString());
         assertEquals(
                 valueAvroRecord.get("log").toString(),
-                "{\"options\":{\"max_tokens\":null,\"temperature\":null,\"top_p\":null,\"logit_bias\":null,\"user\":null,\"n\":null,\"stop\":null,\"presence_penalty\":null,\"frequency_penalty\":null,\"stream\":true,\"model\":\"test-model\",\"min-chunks-per-message\":20},\"messages\":[{\"role\":\"user\",\"content\":\"value1 key2\"}],\"model\":\"test-model\"}");
+                "{\"options\":{\"max_tokens\":null,\"temperature\":null,\"top_p\":null,\"logit_bias\":null,\"user\":null,\"n\":null,\"stop\":null,\"presence_penalty\":null,\"frequency_penalty\":null,\"stream\":true,\"model\":\"test-model\",\"functions\":null,\"function_call\":null,\"dataSources\":null,\"min-chunks-per-message\":20},\"messages\":[{\"role\":\"user\",\"content\":\"value1 key2\"}],\"model\":\"test-model\"}");
     }
 
     @Test

diff --git a/langstream-core/src/main/java/ai/langstream/impl/deploy/ApplicationDeployer.java b/langstream-core/src/main/java/ai/langstream/impl/deploy/ApplicationDeployer.java
@@ -60,10 +60,14 @@ public ExecutionPlan createImplementation(
         StreamingClusterRuntime streamingClusterRuntime =
                 registry.getStreamingClusterRuntime(
                         applicationInstance.getInstance().streamingCluster());
-        log.info("Building execution plan for application {}", applicationInstance);
+        if (log.isDebugEnabled()) {
+            log.debug("Building execution plan for application {}", applicationInstance);
+        }
         final Application resolvedApplicationInstance =
                 ApplicationPlaceholderResolver.resolvePlaceholders(applicationInstance);
-        log.info("After resolving the placeholders {}", resolvedApplicationInstance);
+        if (log.isDebugEnabled()) {
+            log.debug("After resolving the placeholders {}", resolvedApplicationInstance);
+        }
         return clusterRuntime.buildExecutionPlan(
                 applicationId,
                 resolvedApplicationInstance,

diff --git a/...-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/ChatCompletionsIT.java b/...-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/ChatCompletionsIT.java
@@ -61,7 +61,7 @@ public void testChatCompletionWithStreaming() throws Exception {
         String model = "gpt-35-turbo";
 
         stubFor(
-                post("/openai/deployments/gpt-35-turbo/chat/completions?api-version=2023-03-15-preview")
+                post("/openai/deployments/gpt-35-turbo/chat/completions?api-version=2023-08-01-preview")
                         .willReturn(
                                 okJson(
                                         """
@@ -182,7 +182,7 @@ tenant, appId, application, buildInstanceYaml(), expectedAgents)) {
                         waitForMessages(
                                 consumer,
                                 List.of(
-                                        "{\"question\":\"the car\",\"session-id\":\"2139847128764192\",\"answer\":\"A car is a vehicle\",\"prompt\":\"{\\\"options\\\":{\\\"max_tokens\\\":null,\\\"temperature\\\":null,\\\"top_p\\\":null,\\\"logit_bias\\\":null,\\\"user\\\":null,\\\"n\\\":null,\\\"stop\\\":null,\\\"presence_penalty\\\":null,\\\"frequency_penalty\\\":null,\\\"stream\\\":true,\\\"model\\\":\\\"gpt-35-turbo\\\",\\\"min-chunks-per-message\\\":3},\\\"messages\\\":[{\\\"role\\\":\\\"user\\\",\\\"content\\\":\\\"What can you tell me about the car ?\\\"}],\\\"model\\\":\\\"gpt-35-turbo\\\"}\"}"));
+                                        "{\"question\":\"the car\",\"session-id\":\"2139847128764192\",\"answer\":\"A car is a vehicle\",\"prompt\":\"{\\\"options\\\":{\\\"max_tokens\\\":null,\\\"temperature\\\":null,\\\"top_p\\\":null,\\\"logit_bias\\\":null,\\\"user\\\":null,\\\"n\\\":null,\\\"stop\\\":null,\\\"presence_penalty\\\":null,\\\"frequency_penalty\\\":null,\\\"stream\\\":true,\\\"model\\\":\\\"gpt-35-turbo\\\",\\\"functions\\\":null,\\\"function_call\\\":null,\\\"dataSources\\\":null,\\\"min-chunks-per-message\\\":3},\\\"messages\\\":[{\\\"role\\\":\\\"user\\\",\\\"content\\\":\\\"What can you tell me about the car ?\\\"}],\\\"model\\\":\\\"gpt-35-turbo\\\"}\"}"));
                 ConsumerRecord record = mainOutputRecords.get(0);
 
                 assertNull(record.headers().lastHeader("stream-id"));

diff --git a/...untime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/ComputeEmbeddingsIT.java b/...untime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/ComputeEmbeddingsIT.java
@@ -125,7 +125,7 @@ private static Stream<Arguments> providers() {
                                         .formatted(wireMockRuntimeInfo.getHttpBaseUrl()),
                                 () ->
                                         stubFor(
-                                                post("/openai/deployments/text-embedding-ada-002/embeddings?api-version=2023-03-15-preview")
+                                                post("/openai/deployments/text-embedding-ada-002/embeddings?api-version=2023-08-01-preview")
                                                         .willReturn(
                                                                 okJson(
                                                                         """
@@ -278,7 +278,7 @@ public void testComputeBatchEmbeddings() throws Exception {
                             wireMockRuntimeInfo.getWireMock().removeStubMapping(stubMapping);
                         });
         stubFor(
-                post("/openai/deployments/text-embedding-ada-002/embeddings?api-version=2023-03-15-preview")
+                post("/openai/deployments/text-embedding-ada-002/embeddings?api-version=2023-08-01-preview")
                         .willReturn(
                                 okJson(
                                         """

diff --git a/...eam-runtime-tester/src/main/java/ai/langstream/runtime/tester/LocalApplicationRunner.java b/...eam-runtime-tester/src/main/java/ai/langstream/runtime/tester/LocalApplicationRunner.java
@@ -16,6 +16,7 @@
 package ai.langstream.runtime.tester;
 
 import ai.langstream.api.model.Application;
+import ai.langstream.api.runner.assets.AssetManagerRegistry;
 import ai.langstream.api.runner.topics.TopicConnectionsRuntimeRegistry;
 import ai.langstream.api.runtime.ClusterRuntimeRegistry;
 import ai.langstream.api.runtime.ExecutionPlan;
@@ -75,11 +76,14 @@ public LocalApplicationRunner(Path agentsDirectory, Path codeDirectory) throws E
                 new TopicConnectionsRuntimeRegistry();
         narFileHandler.scan();
         topicConnectionsRuntimeRegistry.setPackageLoader(narFileHandler);
+        AssetManagerRegistry assetManagerRegistry = new AssetManagerRegistry();
+        assetManagerRegistry.setAssetManagerPackageLoader(narFileHandler);
         this.applicationDeployer =
                 ApplicationDeployer.builder()
                         .registry(new ClusterRuntimeRegistry())
                         .pluginsRegistry(new PluginsRegistry())
                         .topicConnectionsRuntimeRegistry(topicConnectionsRuntimeRegistry)
+                        .assetManagerRegistry(assetManagerRegistry)
                         .build();
     }
 

diff --git a/...tream-runtime/langstream-runtime-tester/src/main/resources/gateway.application.properties b/...tream-runtime/langstream-runtime-tester/src/main/resources/gateway.application.properties
@@ -1,2 +1,3 @@
 application.storage.apps.type=memory
 server.port=8091
+application.gateways.code.path=/app/agents
diff --git a/pom.xml b/pom.xml
@@ -35,7 +35,8 @@
     <confluent.avro>7.4.0</confluent.avro>
     <!-- Must match version in pulsar -->
     <avro.version>1.10.2</avro.version>
-    <azure-ai-openai.version>1.0.0-beta.2</azure-ai-openai.version>
+    <azure-ai-openai.version>1.0.0-beta.4</azure-ai-openai.version>
+    <azure-sdk-bom.version>1.2.16</azure-sdk-bom.version>
     <!-- same version of Kafka -->
     <netty.version>4.1.97.Final</netty.version>
     <netty.boringssl.version>2.0.61.Final</netty.boringssl.version>
@@ -98,6 +99,14 @@
         <version>${netty.version}</version>
         <type>pom</type>
       </dependency>
+      <dependency>
+        <groupId>com.azure</groupId>
+        <artifactId>azure-sdk-bom</artifactId>
+        <version>${azure-sdk-bom.version}</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+
       <dependency>
         <groupId>io.netty.incubator</groupId>
         <artifactId>netty-incubator-transport-native-io_uring</artifactId>