Merge pull request #2 from andreadimaio/tools-execution-model

Removed emitOn on default worker pool for watsonx.ai
quarkiverse · Nov 6, 2024 · 790932f · 790932f
2 parents bc7d6c0 + 612241f
commit 790932f
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 9 deletions.
diff --git a/...oyment/src/test/java/io/quarkiverse/langchain4j/watsonx/deployment/AiChatServiceTest.java b/...oyment/src/test/java/io/quarkiverse/langchain4j/watsonx/deployment/AiChatServiceTest.java
@@ -45,6 +45,7 @@
 import io.quarkiverse.langchain4j.watsonx.runtime.config.ChatModelConfig;
 import io.quarkiverse.langchain4j.watsonx.runtime.config.LangChain4jWatsonxConfig;
 import io.quarkus.test.QuarkusUnitTest;
+import io.smallrye.common.annotation.Blocking;
 import io.smallrye.mutiny.Multi;
 
 public class AiChatServiceTest extends WireMockAbstract {
@@ -106,6 +107,7 @@ interface AIServiceWithTool {
     @Singleton
     static class Calculator {
         @Tool("Execute the sum of two numbers")
+        @Blocking
         public int sum(int first, int second) {
             return first + second;
         }

diff --git a/...rs/watsonx/runtime/src/main/java/io/quarkiverse/langchain4j/watsonx/WatsonxChatModel.java b/...rs/watsonx/runtime/src/main/java/io/quarkiverse/langchain4j/watsonx/WatsonxChatModel.java
@@ -32,7 +32,6 @@
 import io.quarkiverse.langchain4j.watsonx.bean.TextStreamingChatResponse;
 import io.quarkiverse.langchain4j.watsonx.bean.TokenizationRequest;
 import io.smallrye.mutiny.Context;
-import io.smallrye.mutiny.infrastructure.Infrastructure;
 
 public class WatsonxChatModel extends Watsonx implements ChatLanguageModel, StreamingChatLanguageModel, TokenCountEstimator {
 
@@ -110,14 +109,8 @@ public void generate(List<ChatMessage> messages, List<ToolSpecification> toolSpe
         context.put(TOOLS_CONTEXT, new ArrayList<StreamingToolFetcher>());
         context.put(COMPLETE_MESSAGE_CONTEXT, new StringBuilder());
 
-        var mutiny = client.streamingChat(request, version);
-        if (tools != null) {
-            // Today Langchain4j doesn't allow to use the async operation with tools.
-            // One idea might be to give to the developer the possibility to use the VirtualThread.
-            mutiny.emitOn(Infrastructure.getDefaultWorkerPool());
-        }
-
-        mutiny.subscribe()
+        client.streamingChat(request, version)
+                .subscribe()
                 .with(context,
                         new Consumer<TextStreamingChatResponse>() {
                             @Override