Auto PR - Bump SDK version to 3.5.13

README.md

@@ -120,7 +120,7 @@ With the application running, you can test the system using the following endpoi
 
 * Start a new session:
 ```shell
-curl -i -XPOST --location "http://localhost:9000/activities/alice" \
+curl -i -XPOST --location "http://localhost:9000/activities/alice/1" \
   --header "Content-Type: application/json" \
   --data '{"message": "I do not work tomorrow. I am in Madrid. What should I do? Beware of the weather"}'
 ```
@@ -129,15 +129,7 @@ The system will process the query, select the appropriate agents, and return a r
 
 * Retrieve the response for a specific session:
 ```shell
-curl -i -XGET --location "http://localhost:9000/activities/alice/{sessionId}"
-```
-
-Replace `{sessionId}` with the ID returned when the session was created. Example:
-
-```shell
-$ curl "http://localhost:9000/activities/alice/c1219e5a-abae-44c0-959b-ff76aa22cb2e"
-
-The weather in Madrid is rainy tomorrow, so you might want to explore indoor attractions like the Prado Museum or Reina Sofia Museum. Alternatively, you can visit local cafes and food markets, such as Mercado de San Miguel, to enjoy some culinary delights without getting wet. If you're up for something more active, you could also consider visiting an escape room or an indoor sports facility. 
+curl -i -XGET --location "http://localhost:9000/activities/alice/1"
 ```
 
 You can also retrieve for all previous suggestions for a user:

pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>io.akka</groupId>
     <artifactId>akka-javasdk-parent</artifactId>
-    <version>3.5.12</version>
+    <version>3.5.13</version>
   </parent>
 
   <groupId>io.akka.ai</groupId>

src/main/java/demo/multiagent/api/ActivityEndpoint.java

@@ -7,8 +7,11 @@
 import akka.javasdk.annotations.http.Post;
 import akka.javasdk.client.ComponentClient;
 import akka.javasdk.http.HttpResponses;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import demo.multiagent.application.ActivityView;
 import demo.multiagent.application.AgentTeamWorkflow;
+import demo.multiagent.application.AgentTeamWorkflow.AgentTeamNotification;
 import demo.multiagent.application.PreferencesEntity;
 import java.util.List;
 import java.util.UUID;
@@ -44,10 +47,8 @@ public ActivityEndpoint(ComponentClient componentClient) {
     this.componentClient = componentClient;
   }
 
-  @Post("/activities/{userId}")
-  public HttpResponse suggestActivities(String userId, Request request) {
-    var sessionId = UUID.randomUUID().toString();
-
+  @Post("/activities/{userId}/{sessionId}")
+  public HttpResponse suggestActivities(String userId, String sessionId, Request request) {
     var res = componentClient
       .forWorkflow(sessionId)
       .method(AgentTeamWorkflow::start)
@@ -88,4 +89,44 @@ public HttpResponse addPreference(String userId, AddPreference request) {
 
     return HttpResponses.created();
   }
+
+  @JsonTypeInfo(use = JsonTypeInfo.Id.NAME)
+  @JsonSubTypes(
+    {
+      @JsonSubTypes.Type(value = UpdateEvent.StatusUpdate.class, name = "S"),
+      @JsonSubTypes.Type(value = UpdateEvent.LlmResponseStart.class, name = "LS"),
+      @JsonSubTypes.Type(value = UpdateEvent.LlmResponseDelta.class, name = "LD"),
+      @JsonSubTypes.Type(value = UpdateEvent.LlmResponseEnd.class, name = "LE"),
+    }
+  )
+  public sealed interface UpdateEvent {
+    record StatusUpdate(String msg) implements UpdateEvent {}
+
+    record LlmResponseStart() implements UpdateEvent {}
+
+    record LlmResponseDelta(String response) implements UpdateEvent {}
+
+    record LlmResponseEnd() implements UpdateEvent {}
+  }
+
+  @Get("/updates/{sessionId}")
+  public HttpResponse updates(String sessionId) {
+    return HttpResponses.serverSentEvents(
+      componentClient
+        .forWorkflow(sessionId)
+        .notificationStream(AgentTeamWorkflow::updates)
+        .source()
+        .map(notification ->
+          switch (notification) {
+            case AgentTeamNotification.LlmResponseDelta llmResponseDelta -> new UpdateEvent.LlmResponseDelta(
+              llmResponseDelta.response()
+            );
+            case AgentTeamNotification.LlmResponseEnd __ -> new UpdateEvent.LlmResponseEnd();
+            case AgentTeamNotification.LlmResponseStart __ -> new UpdateEvent.LlmResponseStart();
+            case AgentTeamNotification.StatusUpdate statusUpdate -> new UpdateEvent.StatusUpdate(
+              statusUpdate.msg()
+            );
+          })
+    );
+  }
 }

src/main/java/demo/multiagent/api/UiEndpoint.java

@@ -0,0 +1,21 @@
+package demo.multiagent.api;
+
+import akka.http.javadsl.model.HttpResponse;
+import akka.javasdk.annotations.Acl;
+import akka.javasdk.annotations.http.Get;
+import akka.javasdk.annotations.http.HttpEndpoint;
+import akka.javasdk.http.HttpResponses;
+
+/**
+ * This Http endpoint returns the static UI page located under
+ * src/main/resources/static-resources/
+ */
+@HttpEndpoint
+@Acl(allow = @Acl.Matcher(principal = Acl.Principal.ALL))
+public class UiEndpoint {
+
+  @Get("/")
+  public HttpResponse index() {
+    return HttpResponses.staticResource("index.html"); // <1>
+  }
+}

src/main/java/demo/multiagent/application/AgentTeamWorkflow.java

@@ -3,14 +3,19 @@
 import static demo.multiagent.application.AgentTeamWorkflow.Status.COMPLETED;
 import static demo.multiagent.application.AgentTeamWorkflow.Status.FAILED;
 import static demo.multiagent.application.AgentTeamWorkflow.Status.STARTED;
+import static java.time.Duration.ofMillis;
 import static java.time.Duration.ofSeconds;
 
 import akka.Done;
+import akka.javasdk.NotificationPublisher;
 import akka.javasdk.annotations.Component;
 import akka.javasdk.annotations.StepName;
 import akka.javasdk.client.ComponentClient;
 import akka.javasdk.client.DynamicMethodRef;
 import akka.javasdk.workflow.Workflow;
+import akka.stream.Materializer;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import demo.multiagent.domain.AgentRequest;
 import demo.multiagent.domain.AgentSelection;
 import demo.multiagent.domain.Plan;
@@ -79,11 +84,37 @@ public State failed() {
   private static final Logger logger = LoggerFactory.getLogger(AgentTeamWorkflow.class);
 
   private final ComponentClient componentClient;
+  private final NotificationPublisher<AgentTeamNotification> notificationPublisher;
+  private final Materializer materializer;
 
-  public AgentTeamWorkflow(ComponentClient componentClient) {
+  public AgentTeamWorkflow(
+    ComponentClient componentClient,
+    NotificationPublisher<AgentTeamNotification> notificationPublisher,
+    Materializer materializer
+  ) {
     this.componentClient = componentClient;
+    this.notificationPublisher = notificationPublisher;
+    this.materializer = materializer;
   }
 
+  @JsonTypeInfo(use = JsonTypeInfo.Id.NAME)
+  @JsonSubTypes(
+    {
+      @JsonSubTypes.Type(value = AgentTeamNotification.StatusUpdate.class, name = "S"),
+      @JsonSubTypes.Type(value = AgentTeamNotification.LlmResponseStart.class, name = "LS"),
+      @JsonSubTypes.Type(value = AgentTeamNotification.LlmResponseDelta.class, name = "LD"),
+      @JsonSubTypes.Type(value = AgentTeamNotification.LlmResponseEnd.class, name = "LE"),
+    }
+  )
+  public sealed interface AgentTeamNotification {
+    record StatusUpdate(String msg) implements AgentTeamNotification {}
+
+    record LlmResponseStart() implements AgentTeamNotification {}
+
+    record LlmResponseDelta(String response) implements AgentTeamNotification {}
+
+    record LlmResponseEnd() implements AgentTeamNotification {}
+  }
 
   @Override
   public WorkflowSettings settings() {
@@ -140,6 +171,9 @@ private StepEffect selectAgentsStep() { // <2>
       .invoke(currentState().userQuery); // <4>
 
     logger.info("Selected agents: {}", selection.agents());
+    notificationPublisher.publish(
+      new AgentTeamNotification.StatusUpdate("Agents selected: " + selection.agents())
+    );
     if (selection.agents().isEmpty()) {
       var newState = currentState()
         .withFinalAnswer("Couldn't find any agent(s) able to respond to the original query.")
@@ -167,6 +201,11 @@ private StepEffect createPlanStep(AgentSelection agentSelection) { // <2>
       .invoke(new PlannerAgent.Request(currentState().userQuery, agentSelection)); // <6>
 
     logger.info("Execution plan: {}", plan);
+    notificationPublisher.publish(
+      new AgentTeamNotification.StatusUpdate(
+        "Execution plan formed. Number of steps: " + plan.steps().size()
+      )
+    );
     return stepEffects()
       .updateState(currentState().withPlan(plan))
       .thenTransitionTo(AgentTeamWorkflow::executePlanStep); // <7>
@@ -180,6 +219,9 @@ private StepEffect executePlanStep() { // <2>
       stepPlan.agentId(),
       stepPlan.query()
     );
+    notificationPublisher.publish(
+      new AgentTeamNotification.StatusUpdate("Calling: " + stepPlan.agentId())
+    );
     var agentResponse = callAgent(stepPlan.agentId(), stepPlan.query()); // <9>
     if (agentResponse.startsWith("ERROR")) {
       throw new RuntimeException(
@@ -220,11 +262,26 @@ private String callAgent(String agentId, String query) {
   @StepName("summarize")
   private StepEffect summarizeStep() { // <2>
     var agentsAnswers = currentState().agentResponses.values();
-    var finalAnswer = componentClient
+
+    var tokenSource = componentClient
       .forAgent()
       .inSession(sessionId())
-      .method(SummarizerAgent::summarize)
-      .invoke(new SummarizerAgent.Request(currentState().userQuery, agentsAnswers));
+      .tokenStream(SummarizerAgent::summarize)
+      .source(new SummarizerAgent.Request(currentState().userQuery, agentsAnswers));
+
+    notificationPublisher.publish(new AgentTeamNotification.LlmResponseStart());
+    var finalAnswer = notificationPublisher.publishTokenStream(
+      tokenSource,
+      10,
+      ofMillis(200),
+      AgentTeamNotification.LlmResponseDelta::new,
+      materializer
+    );
+
+    notificationPublisher.publish(new AgentTeamNotification.LlmResponseEnd());
+    notificationPublisher.publish(
+      new AgentTeamNotification.StatusUpdate("All steps completed!")
+    );
 
     return stepEffects()
       .updateState(currentState().withFinalAnswer(finalAnswer).complete())
@@ -239,6 +296,10 @@ private StepEffect interruptStep() {
     return stepEffects().updateState(currentState().failed()).thenEnd();
   }
 
+  public NotificationPublisher.NotificationStream<AgentTeamNotification> updates() {
+    return notificationPublisher.stream();
+  }
+
   private String sessionId() {
     return commandContext().workflowId();
   }

src/main/java/demo/multiagent/application/SummarizerAgent.java

@@ -29,13 +29,13 @@ private String buildSystemMessage(String userQuery) {
     """.formatted(userQuery);
   }
 
-  public Effect<String> summarize(Request request) {
+  public StreamEffect summarize(Request request) {
     var allResponses = request.agentsResponses
       .stream()
       .filter(response -> !response.startsWith("ERROR"))
       .collect(Collectors.joining("\n\n"));
 
-    return effects()
+    return streamEffects()
       .systemMessage(buildSystemMessage(request.originalQuery))
       .userMessage("Summarize the following: \n" + allResponses)
       .thenReply();