Merge branch 'release/1.3.0'

README.md

@@ -44,7 +44,7 @@ LangGraph for Java. A library for building stateful, multi-agents applications w
 
 | Date         | Release        | info
 |--------------|----------------| ---
-| Jan 23, 2025 | `1.2.4` | official release
+| Jan 30, 2025 | `1.3.0` | official release
 
 
 ## Samples
@@ -74,7 +74,7 @@ LangGraph for Java. A library for building stateful, multi-agents applications w
 <dependency>
     <groupId>org.bsc.langgraph4j</groupId>
     <artifactId>langgraph4j-core</artifactId>
-    <version>1.2.4</version>
+    <version>1.3.0</version>
 </dependency>
 ```
 

agent-executor/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>org.bsc.langgraph4j</groupId>
         <artifactId>langgraph4j-parent</artifactId>
-        <version>1.2.5</version>
+        <version>1.3.0</version>
         <relativePath>..</relativePath>
     </parent>
 

core/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>org.bsc.langgraph4j</groupId>
     <artifactId>langgraph4j-parent</artifactId>
-    <version>1.2.5</version>
+    <version>1.3.0</version>
   </parent>
 
   <artifactId>langgraph4j-core</artifactId>

core/src/main/java/org/bsc/langgraph4j/CompiledGraph.java

@@ -16,8 +16,9 @@
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
-import java.util.stream.StreamSupport;
+import java.util.stream.Stream;
 
 import static java.lang.String.format;
 import static java.util.concurrent.CompletableFuture.completedFuture;
@@ -60,9 +61,56 @@ protected CompiledGraph(StateGraph<State> stateGraph, CompileConfig compileConfi
             nodes.put(n.id(), factory.apply(compileConfig));
         }
 
-        stateGraph.edges.forEach(e ->
-                edges.put(e.sourceId(), e.target())
-        );
+        for( var e : stateGraph.edges ) {
+            var targets = e.targets();
+            if (targets.size() == 1) {
+                edges.put(e.sourceId(), targets.get(0));
+            }
+            else {
+                Supplier<Stream<EdgeValue<State>>> parallelNodeStream = () ->
+                        targets.stream().filter( target -> nodes.containsKey(target.id()) );
+
+                var parallelNodeEdges = parallelNodeStream.get()
+                        .map( target -> new Edge<State>(target.id()))
+                        .filter( ee -> stateGraph.edges.contains( ee ) )
+                        .map(  ee -> stateGraph.edges.indexOf( ee ) )
+                        .map( index -> stateGraph.edges.get(index) )
+                        .toList();
+
+                var  parallelNodeTargets = parallelNodeEdges.stream()
+                                                .map( ee -> ee.target().id() )
+                                                .collect(Collectors.toSet());
+
+                if( parallelNodeTargets.size() > 1  ) {
+
+                    var conditionalEdges = parallelNodeEdges.stream()
+                                                .filter( ee -> ee.target().value() != null )
+                                                .toList();
+                    if(!conditionalEdges.isEmpty()) {
+                        throw StateGraph.Errors.unsupportedConditionalEdgeOnParallelNode.exception(
+                                e.sourceId(),
+                                conditionalEdges.stream().map(Edge::sourceId).toList() );
+                    }
+                    throw StateGraph.Errors.illegalMultipleTargetsOnParallelNode.exception(e.sourceId(), parallelNodeTargets );
+                }
+
+                var actions = parallelNodeStream.get()
+                                    //.map( target -> nodes.remove(target.id()) )
+                                    .map( target -> nodes.get(target.id()) )
+                                    .toList();
+
+                var parallelNode = Node.parallel( e.sourceId(), actions, stateGraph.getChannels() );
+
+                nodes.put( parallelNode.id(), parallelNode.actionFactory().apply(compileConfig) );
+
+                edges.put( e.sourceId(), new EdgeValue<>( parallelNode.id(), null  ) );
+
+                edges.put( parallelNode.id(), new EdgeValue<>( parallelNodeTargets.iterator().next(), null  ));
+
+            }
+
+
+        }
     }
 
     public Collection<StateSnapshot<State>> getStateHistory( RunnableConfig config ) {
@@ -146,12 +194,12 @@ public RunnableConfig updateState( RunnableConfig config, Map<String,Object> val
         return updateState(config, values, null);
     }
 
-    @Deprecated
+    @Deprecated( forRemoval = true )
     public EdgeValue<State> getEntryPoint() {
         return stateGraph.getEntryPoint();
     }
 
-    @Deprecated
+    @Deprecated( forRemoval = true )
     public String getFinishPoint() {
         return stateGraph.getFinishPoint();
     }
@@ -204,7 +252,8 @@ private String nextNodeId(String nodeId, Map<String,Object> state) throws Except
     }
 
     private String getEntryPoint( Map<String,Object> state ) throws Exception {
-        return nextNodeId(stateGraph.getEntryPoint(), state, "entryPoint");
+        var entryPoint = this.edges.get(START);
+        return nextNodeId(entryPoint, state, "entryPoint");
     }
 
     private boolean shouldInterruptBefore(@NonNull String nodeId, String previousNodeId ) {
@@ -558,6 +607,5 @@ public Data<Output> next() {
         }
     }
 
-
-
 }
+

core/src/main/java/org/bsc/langgraph4j/DiagramGenerator.java

@@ -4,9 +4,10 @@
 import lombok.Builder;
 import lombok.Value;
 import lombok.experimental.Accessors;
-import org.bsc.langgraph4j.action.AsyncNodeActionWithConfig;
 import org.bsc.langgraph4j.state.AgentState;
 
+import java.util.Objects;
+
 import static java.lang.String.format;
 import static org.bsc.langgraph4j.StateGraph.START;
 
@@ -18,7 +19,6 @@ public abstract class DiagramGenerator {
 
     /**
      * Class that represents a context with various properties and methods.
-     *
      * This class is designed to store and manipulate context-specific data such as
      * a string builder, title, and conditional edge printing status. It also provides
      * a method for converting the title to snake case. The class is annotated with {@code @Value},
@@ -86,7 +86,6 @@ public String toString() {
     protected abstract void call( Context ctx, String from, String to, String description );
     /**
      * Declares a conditional element in the configuration or template.
-     *
      * This method is used to mark the start of a conditional section based on the provided {@code name}.
      * It takes a {@code Context} object that may contain additional parameters necessary for the declaration,
      * and a {@code name} which identifies the type or key associated with the conditional section.
@@ -171,39 +170,57 @@ protected final <State extends AgentState> Context generate( StateGraph<State> s
 
         final int[] conditionalEdgeCount = { 0 };
 
-        stateGraph.edges.forEach( e -> {
-            if( e.target().value() != null ) {
-                conditionalEdgeCount[0] += 1;
-                commentLine( ctx, !ctx.printConditionalEdge() );
-                declareConditionalEdge( ctx, conditionalEdgeCount[0] );
-            }
-        });
-
-        EdgeValue<State> entryPoint = stateGraph.getEntryPoint();
-        if( entryPoint.id() != null  ) {
-            call( ctx, START, entryPoint.id() );
+        stateGraph.edges.stream()
+            .filter( e -> !Objects.equals(e.sourceId(), START) )
+                .filter( e -> !e.isParallel() )
+            .forEach( e -> {
+                if( e.target().value() != null ) {
+                    conditionalEdgeCount[0] += 1;
+                    commentLine( ctx, !ctx.printConditionalEdge() );
+                    declareConditionalEdge( ctx, conditionalEdgeCount[0] );
+                }
+            });
+
+        var edgeStart = stateGraph.edges.stream()
+                .filter( e -> Objects.equals( e.sourceId(), START) )
+                .findFirst()
+                .orElseThrow();
+        if( edgeStart.isParallel() ) {
+            edgeStart.targets().forEach( target -> {
+                call( ctx, START, target.id() );
+            });
+        }
+        else if( edgeStart.target().id() != null  ) {
+            call( ctx, START, edgeStart.target().id() );
         }
-        else if( entryPoint.value() != null ) {
+        else if( edgeStart.target().value() != null ) {
             String conditionName = "startcondition";
             commentLine( ctx, !ctx.printConditionalEdge() );
             declareConditionalStart( ctx , conditionName );
-            edgeCondition( ctx, entryPoint.value(), START, conditionName ) ;
+            edgeCondition( ctx, edgeStart.target().value(), START, conditionName ) ;
         }
 
         conditionalEdgeCount[0] = 0; // reset
 
-        stateGraph.edges.forEach( v -> {
-            if( v.target().id() != null ) {
-                call(ctx, v.sourceId(), v.target().id());
-            }
-            else if( v.target().value() != null ) {
-                conditionalEdgeCount[0] += 1;
-                String conditionName = format("condition%d", conditionalEdgeCount[0]);
-
-                edgeCondition( ctx, v.target().value(), v.sourceId(), conditionName );
-
-            }
-        });
+        stateGraph.edges.stream()
+            .filter( e -> !Objects.equals(e.sourceId(), START) )
+            .forEach( v -> {
+
+                if( v.isParallel()) {
+                    v.targets().forEach( target -> {
+                        call(ctx, v.sourceId(), target.id());
+                    });
+                }
+                else if( v.target().id() != null ) {
+                    call(ctx, v.sourceId(), v.target().id());
+                }
+                else if( v.target().value() != null ) {
+                    conditionalEdgeCount[0] += 1;
+                    String conditionName = format("condition%d", conditionalEdgeCount[0]);
+
+                    edgeCondition( ctx, v.targets().get(0).value(), v.sourceId(), conditionName );
+                }
+            });
 
         appendFooter( ctx );
 

core/src/main/java/org/bsc/langgraph4j/Edge.java

@@ -1,30 +1,83 @@
 package org.bsc.langgraph4j;
 
-
-import lombok.Value;
-import lombok.experimental.Accessors;
+import lombok.NonNull;
 import org.bsc.langgraph4j.state.AgentState;
 
-import java.util.Objects;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static java.lang.String.format;
+import static org.bsc.langgraph4j.StateGraph.START;
 
 /**
  * Represents an edge in a graph with a source ID and a target value.
  *
  * @param <State> the type of the state associated with the edge
+ * @param sourceId The ID of the source node.
+ * @param targets The targets value associated with the edge.
  */
-@Value
-@Accessors(fluent = true)
-class Edge<State extends AgentState> {
+record Edge<State extends AgentState>(String sourceId, List<EdgeValue<State>> targets) {
 
-    /**
-     * The ID of the source node.
-     */
-    String sourceId;
+    public Edge(String sourceId, EdgeValue<State> target) {
+        this(sourceId, List.of(target));
+    }
 
-    /**
-     * The target value associated with the edge.
-     */
-    EdgeValue<State> target;
+    public Edge(String id) {
+        this(id, List.of());
+    }
+
+    public boolean isParallel() {
+        return targets.size() > 1;
+    }
+
+    public EdgeValue<State> target() {
+        if( isParallel() ) {
+            throw new IllegalStateException( format("Edge '%s' is parallel", sourceId));
+        }
+        return targets.get(0);
+    }
+
+    public void validate( @NonNull Collection<Node<State>> nodes) throws GraphStateException {
+        if ( !Objects.equals(sourceId(),START) && !nodes.contains(new Node<State>(sourceId()))) {
+            throw StateGraph.Errors.missingNodeReferencedByEdge.exception(sourceId());
+        }
+
+        if( isParallel() ) { // check for duplicates targets
+            Set<String> duplicates = targets.stream()
+                    .collect(Collectors.groupingBy(EdgeValue::id, Collectors.counting())) // Group by element and count occurrences
+                    .entrySet()
+                    .stream()
+                    .filter(entry -> entry.getValue() > 1) // Filter elements with more than one occurrence
+                    .map(Map.Entry::getKey)
+                    .collect(Collectors.toSet());
+            if( !duplicates.isEmpty() ) {
+                throw StateGraph.Errors.duplicateEdgeTargetError.exception(sourceId(), duplicates);
+            }
+        }
+
+        for( EdgeValue<State> target : targets ) {
+            validate(target, nodes);
+        }
+
+    }
+
+    private void validate( EdgeValue<State> target, Collection<Node<State>> nodes ) throws GraphStateException {
+        if (target.id() != null) {
+            if (!Objects.equals(target.id(), StateGraph.END) && !nodes.contains(new Node<State>(target.id()))) {
+                throw StateGraph.Errors.missingNodeReferencedByEdge.exception(target.id());
+            }
+        } else if (target.value() != null) {
+            for (String nodeId : target.value().mappings().values()) {
+                if (!Objects.equals(nodeId, StateGraph.END) && !nodes.contains(new Node<State>(nodeId))) {
+                    throw StateGraph.Errors.missingNodeInEdgeMapping.exception(sourceId(), nodeId);
+                }
+            }
+        } else {
+            throw StateGraph.Errors.invalidEdgeTarget.exception(sourceId());
+        }
+
+    }
 
     /**
      * Checks if this edge is equal to another object.
@@ -51,3 +104,4 @@ public int hashCode() {
     }
 
 }
+