parallelWorkflow() works

Signed-off-by: Dmitrii Tikhomirov <[email protected]>

Author: treblereel

experimental/agentic/src/main/java/io/serverlessworkflow/impl/expressions/agentic/AgenticModelFactory.java

@@ -60,10 +60,7 @@ public WorkflowModel fromAny(WorkflowModel prev, Object obj) {
 
   @Override
   public WorkflowModel combine(Map<String, WorkflowModel> workflowVariables) {
-    // TODO: create a new agenticScope object in the AgenticScopeRegistryAssessor per branch
-    // TODO: Since we share the same agenticScope object, both branches are updating the same
-    // instance, so for now we return the first key.
-    return workflowVariables.values().iterator().next();
+    return newAgenticModel(workflowVariables);
   }
 
   @Override

experimental/fluent/agentic-langchain4j/src/test/java/io/serverlessworkflow/fluent/agentic/langchain4j/WorkflowAgentsIT.java

@@ -22,6 +22,7 @@
 import static io.serverlessworkflow.fluent.agentic.langchain4j.Agents.CreativeWriter;
 import static io.serverlessworkflow.fluent.agentic.langchain4j.Agents.StyleEditor;
 import static io.serverlessworkflow.fluent.agentic.langchain4j.Models.BASE_MODEL;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
@@ -32,14 +33,16 @@
 import dev.langchain4j.agentic.UntypedAgent;
 import dev.langchain4j.agentic.scope.AgenticScope;
 import dev.langchain4j.agentic.workflow.WorkflowAgentsBuilder;
+import io.serverlessworkflow.fluent.agentic.AgenticServices;
+import io.serverlessworkflow.fluent.agentic.AgentsUtils;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.stream.IntStream;
 
-import io.serverlessworkflow.fluent.agentic.AgenticServices;
-import io.serverlessworkflow.fluent.agentic.AgentsUtils;
 import org.junit.jupiter.api.Test;
 
 public class WorkflowAgentsIT {
@@ -49,38 +52,38 @@ void sequential_agents_tests() {
     WorkflowAgentsBuilder builder = new LC4JWorkflowBuilder();
 
     CreativeWriter creativeWriter =
-        spy(
-            dev.langchain4j.agentic.AgenticServices.agentBuilder(CreativeWriter.class)
-                .chatModel(BASE_MODEL)
-                .outputName("story")
-                .build());
+            spy(
+                    dev.langchain4j.agentic.AgenticServices.agentBuilder(CreativeWriter.class)
+                            .chatModel(BASE_MODEL)
+                            .outputName("story")
+                            .build());
 
     AudienceEditor audienceEditor =
-        spy(
-            dev.langchain4j.agentic.AgenticServices.agentBuilder(AudienceEditor.class)
-                .chatModel(BASE_MODEL)
-                .outputName("story")
-                .build());
+            spy(
+                    dev.langchain4j.agentic.AgenticServices.agentBuilder(AudienceEditor.class)
+                            .chatModel(BASE_MODEL)
+                            .outputName("story")
+                            .build());
 
     StyleEditor styleEditor =
-        spy(
-            dev.langchain4j.agentic.AgenticServices.agentBuilder(StyleEditor.class)
-                .chatModel(BASE_MODEL)
-                .outputName("story")
-                .build());
+            spy(
+                    dev.langchain4j.agentic.AgenticServices.agentBuilder(StyleEditor.class)
+                            .chatModel(BASE_MODEL)
+                            .outputName("story")
+                            .build());
 
     UntypedAgent novelCreator =
-        builder
-            .sequenceBuilder()
-            .subAgents(creativeWriter, audienceEditor, styleEditor)
-            .outputName("story")
-            .build();
+            builder
+                    .sequenceBuilder()
+                    .subAgents(creativeWriter, audienceEditor, styleEditor)
+                    .outputName("story")
+                    .build();
 
     Map<String, Object> input =
-        Map.of(
-            "topic", "dragons and wizards",
-            "style", "fantasy",
-            "audience", "young adults");
+            Map.of(
+                    "topic", "dragons and wizards",
+                    "style", "fantasy",
+                    "audience", "young adults");
 
     String story = (String) novelCreator.invoke(input);
     System.out.println(story);
@@ -96,10 +99,10 @@ public void sequenceHelperTest() {
     var audienceEditor = AgentsUtils.newAudienceEditor();
     var styleEditor = AgentsUtils.newStyleEditor();
 
-    NovelCreator novelCreator = AgenticServices.of(NovelCreator.class)
-            .flow(workflow("seqFlow")
-                    .sequence(creativeWriter, audienceEditor, styleEditor)
-            ).build();
+    NovelCreator novelCreator =
+            AgenticServices.of(NovelCreator.class)
+                    .flow(workflow("seqFlow").sequence(creativeWriter, audienceEditor, styleEditor))
+                    .build();
 
     String story = novelCreator.createNovel("dragons and wizards", "young adults", "fantasy");
     assertNotNull(story);
@@ -110,12 +113,25 @@ public void parallelWorkflow() {
     var foodExpert = AgentsUtils.newFoodExpert();
     var movieExpert = AgentsUtils.newMovieExpert();
 
-    EveningPlannerAgent eveningPlannerAgent = AgenticServices.of(EveningPlannerAgent.class)
-            .flow(workflow("parallelFlow")
-                    .parallel(foodExpert, movieExpert)
-            ).build();
+    Function<Map<String, List<String>>, List<EveningPlan>> planEvening =
+            input -> {
+              List<String> movies = input.getOrDefault("findMovie", List.of());
+              List<String> meals = input.getOrDefault("findMeal", List.of());
+              int max = Math.min(movies.size(), meals.size());
+
+              return IntStream.range(0, max)
+                      .mapToObj(i -> new EveningPlan(movies.get(i), meals.get(i)))
+                      .toList();
+            };
+
+    EveningPlannerAgent eveningPlannerAgent =
+            AgenticServices.of(EveningPlannerAgent.class)
+                    .flow(workflow("parallelFlow").parallel(foodExpert, movieExpert)
+                            .outputAs(planEvening))
+                    .build();
     List<EveningPlan> result = eveningPlannerAgent.plan("romantic");
-    assertTrue(result.size() > 0);
+    System.out.println(result);
+    assertEquals(3, result.size());
   }
 
   @Test
@@ -126,11 +142,10 @@ public void loopTest() {
 
     Predicate<AgenticScope> until = s -> s.readState("score", 0).doubleValue() >= 0.8;
 
-
-    StyledWriter styledWriter = AgenticServices.of(StyledWriter.class)
-            .flow(workflow("loopFlow").agent(creativeWriter)
-                    .loop(until, scorer, editor)
-            ).build();
+    StyledWriter styledWriter =
+            AgenticServices.of(StyledWriter.class)
+                    .flow(workflow("loopFlow").agent(creativeWriter).loop(until, scorer, editor))
+                    .build();
 
     String story = styledWriter.writeStoryWithStyle("dragons and wizards", "fantasy");
     assertNotNull(story);
@@ -140,25 +155,26 @@ public void loopTest() {
   public void humanInTheLoop() {
     var astrologyAgent = AgentsUtils.newAstrologyAgent();
 
-    var askSign = new Function<Map<String, Object>, Map<String, Object>>() {
-      @Override
-      public Map<String, Object> apply(Map<String, Object> holder) {
-        System.out.println("What's your star sign?");
-        //var sign = System.console().readLine();
-        holder.put("sign", "piscis");
-        return holder;
-      }
-    };
-
-    String result = AgenticServices.of(HoroscopeAgent.class)
-            .flow(workflow("humanInTheLoop")
-                    .tasks(tasks -> tasks.callFn(fn(askSign)))
-                    .agent(astrologyAgent))
-            .build()
-            .invoke("My name is Mario. What is my horoscope?");
+    var askSign =
+            new Function<Map<String, Object>, Map<String, Object>>() {
+              @Override
+              public Map<String, Object> apply(Map<String, Object> holder) {
+                System.out.println("What's your star sign?");
+                // var sign = System.console().readLine();
+                holder.put("sign", "piscis");
+                return holder;
+              }
+            };
+
+    String result =
+            AgenticServices.of(HoroscopeAgent.class)
+                    .flow(
+                            workflow("humanInTheLoop")
+                                    .tasks(tasks -> tasks.callFn(fn(askSign)))
+                                    .agent(astrologyAgent))
+                    .build()
+                    .invoke("My name is Mario. What is my horoscope?");
 
     assertNotNull(result);
-
   }
-
 }

experimental/fluent/agentic/src/main/java/io/serverlessworkflow/fluent/agentic/AgentTaskItemListBuilder.java

@@ -92,8 +92,8 @@ public AgentTaskItemListBuilder parallel(String name, Object... agents) {
           List<AgentExecutor> execs = AgentAdapters.toExecutors(agents);
           for (int i = 0; i < execs.size(); i++) {
             AgentExecutor ex = execs.get(i);
-            fork.branch(
-                "branch-" + i + "-" + name,
+            String agentName = ex.agentName() != null ? ex.agentName() : "branch-" + i + "-" + name;
+            fork.branch(agentName,
                 AgentAdapters.toFunction(ex),
                 DefaultAgenticScope.class);
           }

experimental/fluent/agentic/src/main/java/io/serverlessworkflow/fluent/agentic/AgenticServices.java

@@ -19,7 +19,6 @@
 import dev.langchain4j.service.V;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.WorkflowApplication;
-
 import java.lang.annotation.Annotation;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
@@ -48,12 +47,12 @@ public AgenticServices<T> flow(AgentWorkflowBuilder builder) {
   }
 
   public T build() {
-    Objects.requireNonNull(builder, "AgenticServices.flow(AgentWorkflowBuilder) must be called before build()");
+    Objects.requireNonNull(
+        builder, "AgenticServices.flow(AgentWorkflowBuilder) must be called before build()");
     Workflow workflow = builder.build();
     return AgenticServiceBuilder.create(agent, new AgentInvocationHandler(workflow));
   }
 
-
   private static class AgenticServiceBuilder {
 
     @SuppressWarnings("unchecked")
@@ -63,7 +62,7 @@ public static <T> T create(Class<T> runner, InvocationHandler h) {
       }
 
       ClassLoader cl = runner.getClassLoader();
-      Class<?>[] ifaces = new Class<?>[]{runner};
+      Class<?>[] ifaces = new Class<?>[] {runner};
       return (T) Proxy.newProxyInstance(cl, ifaces, h);
     }
   }
@@ -82,14 +81,15 @@ public Object invoke(Object proxy, Method method, Object[] args) {
         return switch (method.getName()) {
           case "toString" -> "AgentProxy(" + workflow.getDocument().getName() + ")";
           case "hashCode" -> System.identityHashCode(proxy);
-          case "equals"   -> proxy == args[0];
+          case "equals" -> proxy == args[0];
           default -> throw new IllegalStateException("Unexpected Object method: " + method);
         };
       }
 
       Agent agent = method.getAnnotation(Agent.class);
       if (agent == null) {
-        throw new IllegalStateException("Method " + method.getName() + " is not annotated with @Agent");
+        throw new IllegalStateException(
+            "Method " + method.getName() + " is not annotated with @Agent");
       }
 
       Annotation[][] annotations = method.getParameterAnnotations();
@@ -106,7 +106,12 @@ public Object invoke(Object proxy, Method method, Object[] args) {
           }
         }
         if (!found) {
-          throw new IllegalStateException("Parameter " + (i + 1) + " of method " + method.getName() + " is not annotated with @V");
+          throw new IllegalStateException(
+              "Parameter "
+                  + (i + 1)
+                  + " of method "
+                  + method.getName()
+                  + " is not annotated with @V");
         }
       }
 
@@ -117,5 +122,4 @@ public Object invoke(Object proxy, Method method, Object[] args) {
       }
     }
   }
-
 }

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseWorkflowBuilder.java

@@ -19,8 +19,12 @@
 import io.serverlessworkflow.api.types.Export;
 import io.serverlessworkflow.api.types.Input;
 import io.serverlessworkflow.api.types.Output;
+import io.serverlessworkflow.api.types.TaskItem;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.fluent.spec.spi.TransformationHandlers;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.UUID;
 import java.util.function.Consumer;
 
@@ -90,7 +94,9 @@ public SELF tasks(Consumer<DBuilder> doTaskConsumer) {
     if (this.workflow.getDo() == null) {
       this.workflow.setDo(doTaskBuilder.build().getDo());
     } else {
-      this.workflow.getDo().addAll(doTaskBuilder.build().getDo());
+      List<TaskItem> existingTasks = new ArrayList<>(this.workflow.getDo());
+      existingTasks.addAll(doTaskBuilder.build().getDo());
+      this.workflow.setDo(Collections.unmodifiableList(existingTasks));
     }
     return self();
   }