Tentative to call the workflow

Signed-off-by: Ricardo Zanini <ricardozanini@gmail.com>

Author: ricardozanini

fluent/agentic-langchain4j/src/main/java/io/serverlessworkflow/fluent/agentic/langchain4j/AbstractAgentService.java

@@ -37,12 +37,8 @@ public abstract class AbstractAgentService<T, S> implements WorkflowDefinitionBu
   protected final Class<T> agentServiceClass;
 
   protected AbstractAgentService(Class<T> agentServiceClass) {
-    this("", agentServiceClass);
-  }
-
-  protected AbstractAgentService(String name, Class<T> agentServiceClass) {
     this.workflowBuilder =
-        AgentWorkflowBuilder.workflow(name)
+        AgentWorkflowBuilder.workflow()
             .outputAs(DEFAULT_OUTPUT_FUNCTION)
             .input(i -> i.from(DEFAULT_INPUT_FUNCTION));
     this.agentServiceClass = agentServiceClass;
@@ -60,25 +56,27 @@ public T build() {
 
   @SuppressWarnings("unchecked")
   public S beforeCall(Consumer<Cognisphere> beforeCall) {
-    // TODO: Our runner must know that our input can be a cognisphere object and extract it from the
-    // context, so that we can accept the consumer.
-    // TODO: For now, we can add the consumer to inputFrom
-    this.workflowBuilder.input(i -> i.from(beforeCall));
+    this.workflowBuilder.inputFrom(
+        cog -> {
+          beforeCall.accept(cog);
+          return cog;
+        },
+        Cognisphere.class);
     return (S) this;
   }
 
   @SuppressWarnings("unchecked")
   public S outputName(String outputName) {
     Function<Cognisphere, Object> outputFunction = cog -> cog.readState(outputName);
-    this.workflowBuilder.outputAs(outputFunction);
+    this.workflowBuilder.outputAs(outputFunction, Cognisphere.class);
     this.workflowBuilder.document(
         d -> d.metadata(m -> m.metadata(META_KEY_OUTPUTNAME, outputName)));
     return (S) this;
   }
 
   @SuppressWarnings("unchecked")
   public S output(Function<Cognisphere, Object> output) {
-    this.workflowBuilder.outputAs(output);
+    this.workflowBuilder.outputAs(output, Cognisphere.class);
     return (S) this;
   }
 

fluent/agentic-langchain4j/src/main/java/io/serverlessworkflow/fluent/agentic/langchain4j/WorkflowExecutor.java

@@ -15,15 +15,26 @@
  */
 package io.serverlessworkflow.fluent.agentic.langchain4j;
 
+import dev.langchain4j.agentic.UntypedAgent;
 import dev.langchain4j.agentic.cognisphere.Cognisphere;
 import dev.langchain4j.agentic.cognisphere.CognisphereAccess;
 import dev.langchain4j.agentic.cognisphere.CognisphereKey;
 import dev.langchain4j.agentic.cognisphere.CognisphereRegistry;
+import dev.langchain4j.agentic.cognisphere.ResultWithCognisphere;
 import dev.langchain4j.agentic.internal.AgentInstance;
+import dev.langchain4j.agentic.internal.AgentSpecification;
 import dev.langchain4j.agentic.internal.CognisphereOwner;
+import dev.langchain4j.service.MemoryId;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowModel;
+
 import java.lang.reflect.Method;
+import java.lang.reflect.Parameter;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 
 public class WorkflowInvocationHandler implements CognisphereOwner {
 
@@ -37,6 +48,38 @@ public class WorkflowInvocationHandler implements CognisphereOwner {
     this.workflowApplicationBuilder = workflowApplicationBuilder;
   }
 
+  @SuppressWarnings("unchecked")
+  private static void writeCognisphereState(Cognisphere cognisphere, Method method, Object[] args) {
+    if (method.getDeclaringClass() == UntypedAgent.class) {
+      cognisphere.writeStates((Map<String, Object>) args[0]);
+    } else {
+      Parameter[] parameters = method.getParameters();
+      for (int i = 0; i < parameters.length; i++) {
+        int index = i;
+        AgentSpecification.optionalParameterName(parameters[i])
+            .ifPresent(argName -> cognisphere.writeState(argName, args[index]));
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static void writeWorkflowInputState(final Map<String, Object> input, Method method, Object[] args) {
+    if (method.getDeclaringClass() == UntypedAgent.class) {
+      input.putAll(((Map<String, Object>) args[0]));
+    } else {
+      Parameter[] parameters = method.getParameters();
+      for (int i = 0; i < parameters.length; i++) {
+        int index = i;
+          AgentSpecification.optionalParameterName(parameters[i])
+                .ifPresent(argName -> input.put(argName, args[index]));
+      }
+    }
+  }
+
+  private String agentId() {
+    return workflow.getDocument().getName();
+  }
+
   @Override
   public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
     // outputName
@@ -69,24 +112,73 @@ public Object invoke(Object proxy, Method method, Object[] args) throws Throwabl
     if (method.getDeclaringClass() == CognisphereAccess.class) {
       return switch (method.getName()) {
         case "getCognisphere" ->
-            CognisphereRegistry.get(
-                new CognisphereKey(this.workflow.getDocument().getName(), args[0]));
+            CognisphereRegistry.get(new CognisphereKey(this.agentId(), args[0]));
         case "evictCognisphere" ->
-            CognisphereRegistry.evict(
-                new CognisphereKey(this.workflow.getDocument().getName(), args[0]));
+            CognisphereRegistry.evict(new CognisphereKey(this.agentId(), args[0]));
         default ->
             throw new UnsupportedOperationException(
                 "Unknown method on CognisphereAccess class : " + method.getName());
       };
     }
 
     // invoke
-    return null;
+    return executeWorkflow(method, args);
+  }
+
+  private Object executeWorkflow(Method method, Object[] args) {
+    // TODO: actually, we must own the Cognisphere object creation upon calling the workflow
+
+    //writeCognisphereState(cognisphere, method, args);
+
+    Object input;
+    if (args == null || args.length == 0) {
+      input = new HashMap<>();
+    } else if (args.length == 1) {
+      input = args[0];
+    } else {
+      Map<String, Object> inputMap = new HashMap<>();
+      writeWorkflowInputState(inputMap, method, args);
+      input = inputMap;
+    }
+
+    try (WorkflowApplication app = workflowApplicationBuilder.build()) {
+      CompletableFuture<WorkflowModel> workflowInstance = app.workflowDefinition(workflow).instance(input).start();
+
+      if (method.getReturnType().equals(ResultWithCognisphere.class)) {
+        return workflowInstance.get().as(ResultWithCognisphere.class);
+      } else {
+        return workflowInstance.get().asJavaObject();
+      }
+    } catch (ExecutionException | InterruptedException e) {
+      throw new RuntimeException(
+          "Failed to execute workflow: " + agentId() + " - Cognisphere: " + cognisphere, e);
+    }
   }
 
   @Override
   public CognisphereOwner withCognisphere(Cognisphere cognisphere) {
     this.cognisphere = cognisphere;
     return this;
   }
+
+  private Cognisphere currentCognisphere(Method method, Object[] args) {
+    if (cognisphere != null) {
+      return cognisphere;
+    }
+
+    Object memoryId = memoryId(method, args);
+    return memoryId != null
+        ? CognisphereRegistry.getOrCreate(new CognisphereKey(this.agentId(), memoryId))
+        : CognisphereRegistry.createEphemeralCognisphere();
+  }
+
+  private Object memoryId(Method method, Object[] args) {
+    Parameter[] parameters = method.getParameters();
+    for (int i = 0; i < parameters.length; i++) {
+      if (parameters[i].getAnnotation(MemoryId.class) != null) {
+        return args[i];
+      }
+    }
+    return null;
+  }
 }