refactor ParallelizationWorkflow

Author: alexcheng1982

core/src/main/java/com/javaaidev/agenticpatterns/core/AbstractAgenticWorkflow.java

@@ -4,25 +4,25 @@
 import com.javaaidev.agenticpatterns.core.observation.WorkflowExecutionObservationContext;
 import com.javaaidev.agenticpatterns.core.observation.WorkflowExecutionObservationDocumentation;
 import io.micrometer.observation.ObservationRegistry;
+import java.util.Objects;
 import org.jspecify.annotations.Nullable;
 
 public abstract class AbstractAgenticWorkflow<Request, Response> implements
     AgenticWorkflow<Request, Response> {
 
-  @Nullable
   protected String name;
   @Nullable
   protected ObservationRegistry observationRegistry;
 
   protected AbstractAgenticWorkflow(@Nullable String name,
       @Nullable ObservationRegistry observationRegistry) {
-    this.name = name;
+    this.name = Objects.requireNonNullElseGet(name, () -> this.getClass().getSimpleName());
     this.observationRegistry = observationRegistry;
   }
 
   @Override
   public String getName() {
-    return this.name != null ? this.name : this.getClass().getSimpleName();
+    return this.name;
   }
 
   @Override

examples/src/main/java/com/javaaidev/agenticpatterns/examples/parallelizationworkflow/AlgorithmArticleGenerationAgent.java

@@ -1,21 +1,91 @@
 package com.javaaidev.agenticpatterns.examples.parallelizationworkflow;
 
+import com.javaaidev.agenticpatterns.core.AgentUtils;
+import com.javaaidev.agenticpatterns.parallelizationworkflow.DefaultResponseAssembler;
+import com.javaaidev.agenticpatterns.parallelizationworkflow.ParallelizationWorkflow;
+import com.javaaidev.agenticpatterns.parallelizationworkflow.SubtaskCreationRequest;
+import com.javaaidev.agenticpatterns.taskexecution.TaskExecutionAgent;
 import io.micrometer.observation.ObservationRegistry;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 import org.springframework.ai.chat.client.ChatClient;
 import org.springframework.ai.chat.client.advisor.SimpleLoggerAdvisor;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.util.CollectionUtils;
 
 @Configuration
 public class AlgorithmArticleGenerationConfiguration {
 
   @Bean
-  public AlgorithmArticleGenerationAgent algorithmArticleGenerationAgent(
+  @Qualifier("algorithmArticleGenerationWorkflow")
+  public ParallelizationWorkflow<AlgorithmArticleGenerationRequest, AlgorithmArticleGenerationResponse> algorithmArticleGenerationWorkflow(
       ChatClient.Builder chatClientBuilder,
       SimpleLoggerAdvisor simpleLoggerAdvisor,
-      ObservationRegistry observationRegistry) {
-    return new AlgorithmArticleGenerationAgent(
-        chatClientBuilder.defaultAdvisors(simpleLoggerAdvisor).build(),
-        observationRegistry);
+      ObservationRegistry observationRegistry
+  ) {
+    var chatClient = chatClientBuilder.defaultAdvisors(simpleLoggerAdvisor).build();
+    return ParallelizationWorkflow.<AlgorithmArticleGenerationRequest, AlgorithmArticleGenerationResponse>builder()
+        .subtasksCreator(request -> {
+          var languages = AgentUtils.safeGet(request,
+              AlgorithmArticleGenerationRequest::languages, List.<String>of());
+          if (CollectionUtils.isEmpty(languages)) {
+            return List.of();
+          }
+          return languages.stream().map(
+              language -> new SubtaskCreationRequest<>(language,
+                  TaskExecutionAgent.<SampleCodeGenerationRequest, AlgorithmArticleGenerationResponse>defaultBuilder()
+                      .promptTemplate("""
+                          Write {language} code to meet the requirement.
+                          {description}
+                          """
+                      )
+                      .responseType(SampleCodeGenerationResponse.class)
+                      .name("CodeGeneration_" + language)
+                      .chatClient(chatClient)
+                      .observationRegistry(observationRegistry)
+                      .build(),
+                  (AlgorithmArticleGenerationRequest req) -> new SampleCodeGenerationRequest(
+                      language,
+                      "Implement algorithm " + req.algorithm()))).toList();
+        })
+        .responseAssembler(
+            DefaultResponseAssembler.<AlgorithmArticleGenerationRequest, AlgorithmArticleGenerationResponse>builder()
+                .promptTemplate(
+                    """
+                        Goal: Write an article about {algorithm}.
+                        
+                        Requirements:
+                        - Start with a brief introduction.
+                        - Include only sample code listed below.
+                        - Output the article in markdown.
+                        
+                        {sample_code}
+                        """)
+                .responseType(AlgorithmArticleGenerationResponse.class)
+                .promptTemplateContextProvider(input -> {
+                  var request = input.request();
+                  var results = input.results();
+                  var algorithm = AgentUtils.safeGet(request,
+                      AlgorithmArticleGenerationRequest::algorithm, "");
+                  var sampleCode = results.allSuccessfulResults().entrySet().stream()
+                      .map(entry -> """
+                          Language: %s
+                          Code:
+                          %s
+                          """.formatted(entry.getKey(),
+                          ((SampleCodeGenerationResponse) entry.getValue()).code()))
+                      .collect(Collectors.joining("==========\n", "\n----------\n", "=========="));
+                  return Map.of("algorithm", algorithm, "sample_code", sampleCode);
+                })
+                .name("AlgorithmArticleGeneration")
+                .chatClient(chatClient)
+                .observationRegistry(observationRegistry)
+                .build())
+        .name("algorithmArticleGenerationWorkflow")
+        .observationRegistry(observationRegistry)
+        .build();
   }
 }

examples/src/main/java/com/javaaidev/agenticpatterns/examples/parallelizationworkflow/AlgorithmArticleGenerationConfiguration.java

@@ -1,5 +1,7 @@
 package com.javaaidev.agenticpatterns.examples.parallelizationworkflow;
 
+import com.javaaidev.agenticpatterns.parallelizationworkflow.ParallelizationWorkflow;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -9,15 +11,17 @@
 @RequestMapping("/algorithm_article")
 public class AlgorithmArticleGenerationController {
 
-  private final AlgorithmArticleGenerationAgent agent;
+  private final ParallelizationWorkflow<AlgorithmArticleGenerationRequest, AlgorithmArticleGenerationResponse> workflow;
 
-  public AlgorithmArticleGenerationController(AlgorithmArticleGenerationAgent agent) {
-    this.agent = agent;
+  public AlgorithmArticleGenerationController(
+      @Qualifier("algorithmArticleGenerationWorkflow") ParallelizationWorkflow<AlgorithmArticleGenerationRequest, AlgorithmArticleGenerationResponse> workflow) {
+    this.workflow = workflow;
   }
 
+
   @PostMapping
   public AlgorithmArticleGenerationResponse generateAlgorithmArticle(
       @RequestBody AlgorithmArticleGenerationRequest request) {
-    return agent.call(request);
+    return workflow.execute(request);
   }
 }

examples/src/main/java/com/javaaidev/agenticpatterns/examples/parallelizationworkflow/AlgorithmArticleGenerationController.java

@@ -0,0 +1,5 @@
+package com.javaaidev.agenticpatterns.examples.parallelizationworkflow;
+
+public record SampleCodeGenerationRequest(String language, String description) {
+
+}

examples/src/main/java/com/javaaidev/agenticpatterns/examples/parallelizationworkflow/SampleCodeGenerationAgent.java

@@ -0,0 +1,5 @@
+package com.javaaidev.agenticpatterns.examples.parallelizationworkflow;
+
+public record SampleCodeGenerationResponse(String code) {
+
+}

examples/src/main/java/com/javaaidev/agenticpatterns/examples/parallelizationworkflow/SampleCodeGenerationRequest.java

@@ -0,0 +1,59 @@
+package com.javaaidev.agenticpatterns.parallelizationworkflow;
+
+import com.javaaidev.agenticpatterns.parallelizationworkflow.DefaultResponseAssembler.AssemblingInput;
+import com.javaaidev.agenticpatterns.taskexecution.AbstractTaskExecutionAgentBuilder;
+import com.javaaidev.agenticpatterns.taskexecution.TaskExecutionAgent;
+import io.micrometer.observation.ObservationRegistry;
+import java.lang.reflect.Type;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import org.jspecify.annotations.Nullable;
+import org.springframework.ai.chat.client.ChatClient;
+import org.springframework.ai.chat.client.ChatClient.ChatClientRequestSpec;
+
+public class DefaultResponseAssembler<Request, Response> extends
+    TaskExecutionAgent<AssemblingInput<Request>, Response> implements
+    ResponseAssembler<Request, Response> {
+
+  public DefaultResponseAssembler(ChatClient chatClient,
+      String promptTemplate, @Nullable Type responseType,
+      @Nullable Function<AssemblingInput<Request>, Map<String, Object>> promptTemplateContextProvider,
+      @Nullable Consumer<ChatClientRequestSpec> chatClientRequestSpecUpdater,
+      @Nullable String name,
+      @Nullable ObservationRegistry observationRegistry) {
+    super(chatClient, promptTemplate, responseType, promptTemplateContextProvider,
+        chatClientRequestSpecUpdater, name, observationRegistry);
+  }
+
+  @Override
+  public Response assemble(@Nullable Request request, TaskExecutionResults results) {
+    return this.call(new AssemblingInput<>(request, results));
+  }
+
+  public static <Request, Response> Builder<Request, Response> builder() {
+    return new Builder<>();
+  }
+
+  public record AssemblingInput<Request>(@Nullable Request request,
+                                         TaskExecutionResults results) {
+
+  }
+
+  public static class Builder<Request, Response> extends
+      AbstractTaskExecutionAgentBuilder<AssemblingInput<Request>, Response, Builder<Request, Response>> {
+
+    @Override
+    public DefaultResponseAssembler<Request, Response> build() {
+      return new DefaultResponseAssembler<>(
+          chatClient,
+          promptTemplate,
+          responseType,
+          promptTemplateContextProvider,
+          chatClientRequestSpecUpdater,
+          name,
+          observationRegistry
+      );
+    }
+  }
+}