Merge branch 'release/1.2.2'

README.md

@@ -43,7 +43,7 @@ LangGraph for Java. A library for building stateful, multi-agents applications w
 
 | Date         | Release        | info
 |--------------|----------------| ---
-| Jan 08, 2025 | `1.2.1` | official release
+| Jan 10, 2025 | `1.2.2` | official release
 
 
 ## Samples
@@ -73,7 +73,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.1</version>
+    <version>1.2.2</version>
 </dependency>
 ```
 

agent-executor/pom.xml

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

core/pom.xml

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

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

@@ -43,7 +43,7 @@ public enum StreamMode {
     final Map<String, EdgeValue<State>> edges = new LinkedHashMap<>();
 
     private int maxIterations = 25;
-    private final CompileConfig compileConfig;
+    protected final CompileConfig compileConfig;
 
     /**
      * Constructs a CompiledGraph with the given StateGraph.
@@ -53,9 +53,12 @@ public enum StreamMode {
     protected CompiledGraph(StateGraph<State> stateGraph, CompileConfig compileConfig ) {
         this.stateGraph = stateGraph;
         this.compileConfig = compileConfig;
-        stateGraph.nodes.forEach(n ->
-                nodes.put(n.id(), n.action())
-        );
+
+        for (var n : stateGraph.nodes) {
+            var factory = n.actionFactory();
+            Objects.requireNonNull(factory, format("action factory for node id '%s' is null!", n.id()) );
+            nodes.put(n.id(), factory.apply(compileConfig));
+        }
 
         stateGraph.edges.forEach(e ->
                 edges.put(e.sourceId(), e.target())
@@ -288,13 +291,9 @@ public AsyncGenerator<NodeOutput<State>> stream(Map<String,Object> inputs ) thro
      */
     public Optional<State> invoke(Map<String,Object> inputs, RunnableConfig config ) throws Exception {
 
-        Iterator<NodeOutput<State>> sourceIterator = stream(inputs, config).iterator();
-
-        java.util.stream.Stream<NodeOutput<State>> result = StreamSupport.stream(
-                Spliterators.spliteratorUnknownSize(sourceIterator, Spliterator.ORDERED),
-                false);
-
-        return  result.reduce((a, b) -> b).map( NodeOutput::state);
+       return stream(inputs, config).stream()
+                                        .reduce((a, b) -> b)
+                                        .map( NodeOutput::state);
     }
 
     /**

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

@@ -63,8 +63,8 @@ protected final <State extends AgentState> Context generate( StateGraph<State> s
 
         stateGraph.nodes
                 .forEach( n -> {
-                    if( n.action() instanceof SubgraphNodeAction ) {
-                        SubgraphNodeAction<State> subgraphNodeAction = (SubgraphNodeAction<State>) n.action();
+                    var action =  n.actionFactory().apply( CompileConfig.builder().build() );
+                    if( action instanceof SubgraphNodeAction<?>  subgraphNodeAction) {
                         Context subgraphCtx = generate( subgraphNodeAction.subGraph.stateGraph,
                                                         Context.builder()
                                                                 .title( n.id() )

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

@@ -1,49 +1,35 @@
 package org.bsc.langgraph4j;
 
-import lombok.Value;
-import lombok.experimental.Accessors;
-import org.bsc.langgraph4j.action.AsyncNodeAction;
 import org.bsc.langgraph4j.action.AsyncNodeActionWithConfig;
 import org.bsc.langgraph4j.state.AgentState;
 
+import java.util.Map;
 import java.util.Objects;
+import java.util.function.Function;
 
 /**
- * Represents a node in a graph with a unique identifier and an associated action.
+ * Represents a node in a graph, characterized by a unique identifier and a factory for creating
+ * actions to be executed by the node. This is a generic record where the state type is specified
+ * by the type parameter {@code State}.
  *
- * @param <State> the type of the state associated with the node
+ * @param <State> the type of the state associated with the node; it must extend {@link AgentState}.
+ * @param id the unique identifier for the node.
+ * @param actionFactory a factory function that takes a {@link CompileConfig} and returns an
+ *                      {@link AsyncNodeActionWithConfig} instance for the specified {@code State}.
  */
-@Value
-@Accessors(fluent = true)
-class Node<State extends AgentState> {
+record Node<State extends AgentState>(String id, Function<CompileConfig,AsyncNodeActionWithConfig<State>> actionFactory) {
 
-
     /**
-     * The unique identifier for the node.
-     */
-    String id;
-
-    /**
-     * The action to be performed asynchronously by the node.
+     * Constructor that accepts only the `id` and sets `actionFactory` to null.
+     *
+     * @param id the unique identifier for the node
      */
-    AsyncNodeActionWithConfig<State> action;
-
-    public Node( String id ) {
-        this.id = id;
-        this.action = null;
-
-    }
-    public Node( String id, AsyncNodeAction<State> action ) {
-        this.id = id;
-        this.action = AsyncNodeActionWithConfig.of(action);
-    }
-    public Node( String id, AsyncNodeActionWithConfig<State> action ) {
-        this.id = id;
-        this.action = action;
+    public Node(String id) {
+        this(id, null);
     }
+
     /**
      * Checks if this node is equal to another object.
-     *
      * @param o the object to compare with
      * @return true if this node is equal to the specified object, false otherwise
      */
@@ -64,5 +50,4 @@ public boolean equals(Object o) {
     public int hashCode() {
         return Objects.hash(id);
     }
-
-}
+}

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

@@ -201,17 +201,7 @@ public void setFinishPoint(String finishPoint) {
      * @throws GraphStateException if the node identifier is invalid or the node already exists
      */
     public StateGraph<State> addNode(String id, AsyncNodeAction<State> action) throws GraphStateException {
-        if (Objects.equals(id, END)) {
-            throw Errors.invalidNodeIdentifier.exception(END);
-        }
-        Node<State> node = new Node<State>(id, action);
-
-        if (nodes.contains(node)) {
-            throw Errors.duplicateNodeError.exception(id);
-        }
-
-        nodes.add(node);
-        return this;
+        return addNode( id,  AsyncNodeActionWithConfig.of(action) );
     }
 
     /**
@@ -225,7 +215,7 @@ public StateGraph<State> addNode(String id, AsyncNodeActionWithConfig<State> act
         if (Objects.equals(id, END)) {
             throw Errors.invalidNodeIdentifier.exception(END);
         }
-        Node<State> node = new Node<State>(id, actionWithConfig);
+        Node<State> node = new Node<>(id, ( config ) -> actionWithConfig);
 
         if (nodes.contains(node)) {
             throw Errors.duplicateNodeError.exception(id);
@@ -244,9 +234,43 @@ public StateGraph<State> addNode(String id, AsyncNodeActionWithConfig<State> act
      * @param subGraph the compiled subgraph to be added
      * @return this state graph instance
      * @throws GraphStateException if the node identifier is invalid or the node already exists
+     * @Deprecated This method is deprecated because since the subgraph's state with the parent graph, the compilation
+     *             must be done with the same compile config of the parent to avoid unintended side effects.
+     *             Use {@link #addSubgraph(String, StateGraph)} instead.
      */
+    @Deprecated
     public StateGraph<State> addSubgraph(String id, CompiledGraph<State> subGraph) throws GraphStateException {
-        return addNode(id, new SubgraphNodeAction<State>(subGraph) );
+        return this.addSubgraph( id, subGraph.stateGraph );
+    }
+
+    /**
+     * Adds a subgraph to the state graph by creating a node with the specified identifier.
+     * This implies that Subgraph share the same state with parent graph
+     *
+     * @param id the identifier of the node representing the subgraph
+     * @param subGraph the subgraph to be added. it will be compiled on compilation of the parent
+     * @return this state graph instance
+     * @throws GraphStateException if the node identifier is invalid or the node already exists
+     */
+    public StateGraph<State> addSubgraph(String id, StateGraph<State> subGraph) throws GraphStateException {
+        if (Objects.equals(id, END)) {
+            throw Errors.invalidNodeIdentifier.exception(END);
+        }
+        var node = new Node<>(id, ( config ) -> {
+            try {
+                return new SubgraphNodeAction<>(subGraph, config);
+            } catch (GraphStateException e) {
+                throw new RuntimeException(e);
+            }
+        });
+
+        if (nodes.contains(node)) {
+            throw Errors.duplicateNodeError.exception(id);
+        }
+
+        nodes.add(node);
+        return this;
+
     }
 
     /**
@@ -266,7 +290,7 @@ public StateGraph<State> addEdge(String sourceId, String targetId) throws GraphS
             return this;
         }
 
-        Edge<State> edge = new Edge<State>(sourceId, new EdgeValue<>(targetId, null));
+        Edge<State> edge = new Edge<>(sourceId, new EdgeValue<>(targetId, null));
 
         if (edges.contains(edge)) {
             throw Errors.duplicateEdgeError.exception(sourceId);

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

@@ -13,17 +13,20 @@ class SubgraphNodeAction<State extends AgentState> implements AsyncNodeActionWit
 
     final CompiledGraph<State> subGraph;
 
-    SubgraphNodeAction(CompiledGraph<State> subGraph ) {
-        this.subGraph = subGraph;
+    SubgraphNodeAction(StateGraph<State> subGraph, CompileConfig config ) throws GraphStateException {
+        this.subGraph = subGraph.compile(config);
     }
 
     @Override
     public CompletableFuture<Map<String, Object>> apply(State state, RunnableConfig config)  {
         CompletableFuture<Map<String, Object>> future = new CompletableFuture<>();
 
         try {
+            final Map<String,Object> input = ( subGraph.compileConfig.checkpointSaver().isPresent() ) ?
+                                            Map.of() :
+                                            state.data() ;
 
-            AsyncGenerator<NodeOutput<State>> generator = subGraph.stream( state.data(), config );
+            var generator = subGraph.stream( input, config );
 
             future.complete( mapOf( "_subgraph", generator ) );
 

core/src/main/java/org/bsc/langgraph4j/state/Channel.java

@@ -85,6 +85,7 @@ static <T> Channel<T> of( Reducer<T> reducer, Supplier<T> defaultProvider ) {
      * @param newValue the new value to be set
      * @return the new value of the state property
      */
+    @SuppressWarnings("unchecked")
     default Object update(String key, Object oldValue, Object newValue) {
         T _new = (T)newValue;
 

core/src/test/java/org/bsc/langgraph4j/StateGraphPersistenceTest.java

@@ -389,4 +389,55 @@ public void testPauseAndUpdatePastGraphState() throws Exception {
 
         System.out.println( results.get(2).state().lastMessage().get() );
     }
+
+    @Test
+    public void testWithSubgraph() throws Exception {
+
+        CompileConfig compileConfig = CompileConfig.builder().checkpointSaver(new MemorySaver()).build();
+        var childStep1 = node_async((MessagesState state) -> Map.of("messages", "child:step1"));
+
+        var childStep2 = node_async((MessagesState state) -> Map.of("messages", "child:step2"));
+
+        var childStep3 = node_async((MessagesState state) -> Map.of("messages", "child:step3"));
+
+
+        var workflowChild = new StateGraph<>(MessagesState.SCHEMA, MessagesState::new)
+                .addNode("child:step_1", childStep1)
+                .addNode("child:step_2", childStep2)
+                .addNode("child:step_3", childStep3)
+                .addEdge(START, "child:step_1")
+                .addEdge("child:step_1", "child:step_2")
+                .addEdge("child:step_2", "child:step_3")
+                .addEdge("child:step_3", END)
+                //.compile(compileConfig)
+                ;
+        var step1 = node_async((MessagesState state) -> Map.of("messages", "step1"));
+
+        var step2 = node_async((MessagesState state) -> Map.of("messages", "step2"));
+
+        var step3 = node_async((MessagesState state) -> Map.of("messages", "step3"));
+
+        var workflowParent = new StateGraph<>(MessagesState.SCHEMA, MessagesState::new)
+                .addNode("step_1", step1)
+                .addNode("step_2", step2)
+                .addNode("step_3", step3)
+                .addSubgraph("subgraph", workflowChild)
+                .addEdge(START, "step_1")
+                .addEdge("step_1", "step_2")
+                .addEdge("step_2", "subgraph")
+                .addEdge("subgraph", "step_3")
+                .addEdge("step_3", END)
+                .compile(compileConfig);
+
+
+        var result = workflowParent.stream(Map.of())
+                .stream()
+                .peek(System.out::println)
+                .reduce((a, b) -> b)
+                .map(NodeOutput::state);
+
+        assertTrue(result.isPresent());
+        assertIterableEquals(List.of("step1", "step2", "child:step1", "child:step2", "child:step3", "step3"), result.get().messages());
+
+    }
 }

core/src/test/java/org/bsc/langgraph4j/StateGraphTest.java

@@ -180,7 +180,8 @@ public void testWithSubgraph() throws Exception {
                 .addEdge("child:step_1", "child:step_2")
                 .addEdge("child:step_2", "child:step_3")
                 .addEdge("child:step_3", END)
-                .compile();
+                //.compile()
+                ;
         var step1 = node_async((MessagesState state) -> Map.of("messages", "step1"));
 
         var step2 = node_async((MessagesState state) -> Map.of("messages", "step2"));
@@ -199,21 +200,11 @@ public void testWithSubgraph() throws Exception {
                 .addEdge("step_3", END)
                 .compile();
 
-        var stream = workflowParent.stream(Map.of()).stream();
-
-
-        // NodeOutput{node=__START__, state={messages=[]}}
-        // NodeOutput{node=step_1, state={messages=[step1]}}
-        // NodeOutput{node=step_2, state={messages=[step1, step2]}}
-        // NodeOutput{node=__START__, state={messages=[step1, step2]}}
-        // NodeOutput{node=child:step_1, state={messages=[step1, step2, child:step1]}}
-        // NodeOutput{node=child:step_2, state={messages=[step1, step2, child:step1, child:step2]}}
-        // NodeOutput{node=child:step_3, state={messages=[step1, step2, child:step1, child:step2, child:step3]}}
-        // NodeOutput{node=__END__, state={messages=[step1, step2, child:step1, child:step2, child:step3]}}
-        // NodeOutput{node=subgraph, state={messages=[step1, step2, child:step1, child:step2, child:step3]}}
-        // NodeOutput{node=step_3, state={messages=[step1, step2, child:step1, child:step2, child:step3, step3]}}
-        // NodeOutput{node=__END__, state={messages=[step1, step2, child:step1, child:step2, child:step3, step3]}}
-        var result = stream.reduce((a, b) -> b).map(NodeOutput::state);
+        var result = workflowParent.stream(Map.of())
+                .stream()
+                .peek(System.out::println)
+                .reduce((a, b) -> b)
+                .map(NodeOutput::state);
 
         assertTrue(result.isPresent());
         assertIterableEquals(List.of("step1", "step2", "child:step1", "child:step2", "child:step3", "step3"), result.get().messages());