Add Conversation AI to Java SDK

examples/src/main/java/io/dapr/examples/conversation/DemoConversationAI.java

@@ -0,0 +1,35 @@
+package io.dapr.examples.conversation;
+
+import io.dapr.client.DaprClientBuilder;
+import io.dapr.client.DaprPreviewClient;
+import io.dapr.client.domain.ConversationInput;
+import io.dapr.client.domain.ConversationRequest;
+import io.dapr.client.domain.ConversationResponse;
+import reactor.core.publisher.Mono;
+
+import java.util.ArrayList;
+import java.util.Collections;
+
+public class DemoConversationAI {
+    /**
+     * The main method to start the client.
+     *
+     * @param args Input arguments (unused).
+     */
+    public static void main(String[] args) {
+        try (DaprPreviewClient client = new DaprClientBuilder().buildPreviewClient()) {
+            ConversationInput daprConversationInput = new ConversationInput("Hello How are you ?");
+
+            // Component name is the name provided in the metadata block of the conversation.yaml file.
+            Mono<ConversationResponse> instanceId = client.converse(new ConversationRequest("openai", new ArrayList<>(Collections.singleton(daprConversationInput)))
+                .setContextId("contextId")
+                .setScrubPii(true).setTemperature(1.1d));
+            System.out.printf("Started a new chaining model workflow with instance ID: %s%n", instanceId);
+            ConversationResponse response = instanceId.block();
+
+            System.out.println(response);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

pom.xml

@@ -17,7 +17,7 @@
     <grpc.version>1.69.0</grpc.version>
     <protobuf.version>3.25.5</protobuf.version>
     <protocCommand>protoc</protocCommand>
-    <dapr.proto.baseurl>https://raw.githubusercontent.com/dapr/dapr/v1.14.4/dapr/proto</dapr.proto.baseurl>
+    <dapr.proto.baseurl>https://raw.githubusercontent.com/dapr/dapr/v1.15.2/dapr/proto</dapr.proto.baseurl>
     <dapr.sdk.version>1.15.0-SNAPSHOT</dapr.sdk.version>
     <dapr.sdk.alpha.version>0.15.0-SNAPSHOT</dapr.sdk.alpha.version>
     <os-maven-plugin.version>1.7.1</os-maven-plugin.version>

sdk-tests/src/test/java/io/dapr/it/testcontainers/DaprConversationIT.java

@@ -0,0 +1,119 @@
+package io.dapr.it.testcontainers;
+
+import io.dapr.client.DaprPreviewClient;
+import io.dapr.client.domain.ConversationInput;
+import io.dapr.client.domain.ConversationRequest;
+import io.dapr.client.domain.ConversationResponse;
+import io.dapr.testcontainers.Component;
+import io.dapr.testcontainers.DaprContainer;
+import io.dapr.testcontainers.DaprLogLevel;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+
+@SpringBootTest(
+        webEnvironment = WebEnvironment.RANDOM_PORT,
+        classes = {
+                TestDaprConversationConfiguration.class,
+                TestConversationApplication.class
+        }
+)
+@Testcontainers
+@Tag("testcontainers")
+public class DaprConversationIT {
+
+    private static final Network DAPR_NETWORK = Network.newNetwork();
+    private static final Random RANDOM = new Random();
+    private static final int PORT = RANDOM.nextInt(1000) + 8000;
+
+    @Container
+    private static final DaprContainer DAPR_CONTAINER = new DaprContainer("daprio/daprd:1.15.2")
+            .withAppName("conversation-dapr-app")
+            .withComponent(new Component("echo", "conversation.echo", "v1", new HashMap<>()))
+            .withNetwork(DAPR_NETWORK)
+            .withDaprLogLevel(DaprLogLevel.DEBUG)
+            .withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
+            .withAppChannelAddress("host.testcontainers.internal")
+            .withAppPort(PORT);
+
+    /**
+     * Expose the Dapr ports to the host.
+     *
+     * @param registry the dynamic property registry
+     */
+    @DynamicPropertySource
+    static void daprProperties(DynamicPropertyRegistry registry) {
+        registry.add("dapr.http.endpoint", DAPR_CONTAINER::getHttpEndpoint);
+        registry.add("dapr.grpc.endpoint", DAPR_CONTAINER::getGrpcEndpoint);
+        registry.add("server.port", () -> PORT);
+    }
+
+    @Autowired
+    private DaprPreviewClient daprPreviewClient;
+
+    @BeforeEach
+    public void setUp(){
+        org.testcontainers.Testcontainers.exposeHostPorts(PORT);
+        // Ensure the subscriptions are registered
+    }
+
+    @Test
+    public void testConversationSDKShouldHaveSameOutputAndInput() {
+        ConversationInput conversationInput = new ConversationInput("input this");
+        List<ConversationInput> conversationInputList = new ArrayList<>();
+        conversationInputList.add(conversationInput);
+
+        ConversationResponse response =
+                this.daprPreviewClient.converse(new ConversationRequest("echo", conversationInputList)).block();
+
+        Assertions.assertEquals("", response.getContextId());
+        Assertions.assertEquals("input this", response.getConversationOutpus().get(0).getResult());
+    }
+
+    @Test
+    public void testConversationSDKShouldScrubPIIEntirelyWhenScrubPIIIsSetInRequestBody() {
+        List<ConversationInput> conversationInputList = new ArrayList<>();
+        conversationInputList.add(new ConversationInput("input this [email protected]"));
+        conversationInputList.add(new ConversationInput("input this +12341567890"));
+
+        ConversationResponse response =
+                this.daprPreviewClient.converse(new ConversationRequest("echo", conversationInputList)
+                        .setScrubPii(true)).block();
+
+        Assertions.assertEquals("", response.getContextId());
+        Assertions.assertEquals("input this <EMAIL_ADDRESS>",
+                response.getConversationOutpus().get(0).getResult());
+        Assertions.assertEquals("input this <PHONE_NUMBER>",
+                response.getConversationOutpus().get(1).getResult());
+    }
+
+    @Test
+    public void testConversationSDKShouldScrubPIIOnlyForTheInputWhereScrubPIIIsSet() {
+        List<ConversationInput> conversationInputList = new ArrayList<>();
+        conversationInputList.add(new ConversationInput("input this [email protected]"));
+        conversationInputList.add(new ConversationInput("input this +12341567890").setScrubPii(true));
+
+        ConversationResponse response =
+                this.daprPreviewClient.converse(new ConversationRequest("echo", conversationInputList)).block();
+
+        Assertions.assertEquals("", response.getContextId());
+        Assertions.assertEquals("input this [email protected]",
+                response.getConversationOutpus().get(0).getResult());
+        Assertions.assertEquals("input this <PHONE_NUMBER>",
+                response.getConversationOutpus().get(1).getResult());
+    }
+}

sdk-tests/src/test/java/io/dapr/it/testcontainers/TestConversationApplication.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2024 The Dapr Authors
+ * 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.
+*/
+
+package io.dapr.it.testcontainers;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class TestConversationApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(TestConversationApplication.class, args);
+    }
+
+}

sdk-tests/src/test/java/io/dapr/it/testcontainers/TestDaprConversationConfiguration.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2025 The Dapr Authors
+ * 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.
+*/
+
+package io.dapr.it.testcontainers;
+
+import io.dapr.client.DaprClientBuilder;
+import io.dapr.client.DaprPreviewClient;
+import io.dapr.config.Properties;
+import io.dapr.config.Property;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.Map;
+
+@Configuration
+public class TestDaprConversationConfiguration {
+
+    @Bean
+    public DaprPreviewClient daprPreviewClient(
+            @Value("${dapr.http.endpoint}") String daprHttpEndpoint,
+            @Value("${dapr.grpc.endpoint}") String daprGrpcEndpoint
+    ){
+        Map<Property<?>, String> overrides = Map.of(
+                Properties.HTTP_ENDPOINT, daprHttpEndpoint,
+                Properties.GRPC_ENDPOINT, daprGrpcEndpoint
+        );
+
+        return new DaprClientBuilder().withPropertyOverrides(overrides).buildPreviewClient();
+    }
+}

sdk/src/main/java/io/dapr/client/DaprClientImpl.java

@@ -14,6 +14,7 @@
 package io.dapr.client;
 
 import com.google.common.base.Strings;
+import com.google.protobuf.Any;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Empty;
 import io.dapr.client.domain.ActorMetadata;
@@ -26,6 +27,10 @@
 import io.dapr.client.domain.CloudEvent;
 import io.dapr.client.domain.ComponentMetadata;
 import io.dapr.client.domain.ConfigurationItem;
+import io.dapr.client.domain.ConversationInput;
+import io.dapr.client.domain.ConversationOutput;
+import io.dapr.client.domain.ConversationRequest;
+import io.dapr.client.domain.ConversationResponse;
 import io.dapr.client.domain.DaprMetadata;
 import io.dapr.client.domain.DeleteStateRequest;
 import io.dapr.client.domain.ExecuteStateTransactionRequest;
@@ -1402,6 +1407,79 @@ public Mono<DaprMetadata> getMetadata() {
             });
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public Mono<ConversationResponse> converse(ConversationRequest conversationRequest) {
+
+    try {
+      validateConversationRequest(conversationRequest);
+
+      DaprProtos.ConversationRequest.Builder protosConversationRequestBuilder = DaprProtos.ConversationRequest
+          .newBuilder().setTemperature(conversationRequest.getTemperature())
+          .setScrubPII(conversationRequest.isScrubPii())
+          .setName(conversationRequest.getLlmName());
+
+      if (conversationRequest.getContextId() != null) {
+        protosConversationRequestBuilder.setContextID(conversationRequest.getContextId());
+      }
+
+      for (ConversationInput input : conversationRequest.getConversationInputs()) {
+        if (input.getContent() == null || input.getContent().isEmpty()) {
+          throw new IllegalArgumentException("Conversation input content cannot be null or empty.");
+        }
+
+        DaprProtos.ConversationInput.Builder conversationInputOrBuilder = DaprProtos.ConversationInput.newBuilder()
+            .setContent(input.getContent())
+            .setScrubPII(input.isScrubPii());
+
+        if (input.getRole() != null) {
+          conversationInputOrBuilder.setRole(input.getRole().toString());
+        }
+
+        protosConversationRequestBuilder.addInputs(conversationInputOrBuilder.build());
+      }
+
+      Mono<DaprProtos.ConversationResponse> conversationResponseMono = Mono.deferContextual(
+          context -> this.createMono(
+              it -> intercept(context, asyncStub)
+                  .converseAlpha1(protosConversationRequestBuilder.build(), it)
+          )
+      );
+
+      return conversationResponseMono.map(conversationResponse -> {
+
+        List<ConversationOutput> conversationOutputs = new ArrayList<>();
+        for (DaprProtos.ConversationResult conversationResult : conversationResponse.getOutputsList()) {
+          Map<String, byte[]> parameters = new HashMap<>();
+          for (Map.Entry<String, Any> entrySet : conversationResult.getParametersMap().entrySet()) {
+            parameters.put(entrySet.getKey(), entrySet.getValue().toByteArray());
+          }
+
+          ConversationOutput conversationOutput =
+              new ConversationOutput(conversationResult.getResult(), parameters);
+          conversationOutputs.add(conversationOutput);
+        }
+
+        return new ConversationResponse(conversationResponse.getContextID(), conversationOutputs);
+      });
+    } catch (Exception ex) {
+      return DaprException.wrapMono(ex);
+    }
+  }
+
+  private void validateConversationRequest(ConversationRequest conversationRequest) {
+    if ((conversationRequest.getLlmName() == null) || (conversationRequest.getLlmName().trim().isEmpty())) {
+      throw new IllegalArgumentException("LLM name cannot be null or empty.");
+    }
+
+    if ((conversationRequest.getConversationInputs() == null) || (conversationRequest
+        .getConversationInputs().isEmpty())) {
+      throw new IllegalArgumentException("Conversation inputs cannot be null or empty.");
+    }
+  }
+
   private DaprMetadata buildDaprMetadata(DaprProtos.GetMetadataResponse response) throws IOException {
     String id = response.getId();
     String runtimeVersion = response.getRuntimeVersion();

sdk/src/main/java/io/dapr/client/DaprPreviewClient.java

@@ -17,6 +17,9 @@
 import io.dapr.client.domain.BulkPublishRequest;
 import io.dapr.client.domain.BulkPublishResponse;
 import io.dapr.client.domain.BulkPublishResponseFailedEntry;
+import io.dapr.client.domain.ConversationInput;
+import io.dapr.client.domain.ConversationRequest;
+import io.dapr.client.domain.ConversationResponse;
 import io.dapr.client.domain.LockRequest;
 import io.dapr.client.domain.QueryStateRequest;
 import io.dapr.client.domain.QueryStateResponse;
@@ -268,4 +271,12 @@ <T> Mono<BulkPublishResponse<T>> publishEvents(String pubsubName, String topicNa
    */
   <T> Subscription subscribeToEvents(
       String pubsubName, String topic, SubscriptionListener<T> listener, TypeRef<T> type);
+
+  /**
+   * Converse with an LLM.
+   *
+   * @param conversationRequest request to be passed to the LLM.
+   * @return {@link ConversationResponse}.
+   */
+  Mono<ConversationResponse> converse(ConversationRequest conversationRequest);
 }