How to stream progress from parallel branches¶

This guide runs three independent data sources in parallel. Each branch returns an AsyncGenerator, reports progress as chunks are produced, and finishes with a state update in Data.done(...). The updates are merged into one shared list with Channels.appender(ArrayList::new).

The example is deterministic and does not require an API key or network service.

Streaming semantics in a parallel node¶

A parallel node consumes each branch generator internally before it merges the branch states. The onChunk callback below therefore handles progress at the point where a chunk is produced; in an application it can update a UI, publish an event, or write a structured log. Because the outer graph stream waits for the parallel merge and does not forward inner StreamingOutput values, this callback is the only place to observe per-branch progress in real time.

Use a generator that retains its result value, such as the generator returned by AsyncGeneratorFlow.create(...). This lets the parallel path read the map supplied to Data.done(...). A bare AsyncGenerator.Base does not implement HasResultValue, so its final map cannot be recovered by the parallel node.

Setup¶

In [1]:

var userHomeDir = System.getProperty("user.home");
var localRepoUrl = "file://" + userHomeDir + "/.m2/repository/";
var langgraph4jVersion = "1.9.0";

var userHomeDir = System.getProperty("user.home"); var localRepoUrl = "file://" + userHomeDir + "/.m2/repository/"; var langgraph4jVersion = "1.9.0";

In [2]:

%dependency /add-repo local \{localRepoUrl} release|never snapshot|always
%dependency /add org.bsc.langgraph4j:langgraph4j-core:\{langgraph4jVersion}
%dependency /resolve

%dependency /add-repo local \{localRepoUrl} release|never snapshot|always %dependency /add org.bsc.langgraph4j:langgraph4j-core:\{langgraph4jVersion} %dependency /resolve

Repository local url: file:///Users/bsorrentino/.m2/repository/ added.
Adding dependency org.bsc.langgraph4j:langgraph4j-core:1.9.0
Solving dependencies
Resolved artifacts count: 3
Add to classpath: /Users/bsorrentino/Library/Jupyter/kernels/rapaio-jupyter-kernel/mima_cache/org/bsc/langgraph4j/langgraph4j-core/1.9.0/langgraph4j-core-1.9.0.jar
Add to classpath: /Users/bsorrentino/Library/Jupyter/kernels/rapaio-jupyter-kernel/mima_cache/org/bsc/async/async-generator/5.0.0/async-generator-5.0.0.jar
Add to classpath: /Users/bsorrentino/Library/Jupyter/kernels/rapaio-jupyter-kernel/mima_cache/org/slf4j/slf4j-api/2.0.9/slf4j-api-2.0.9.jar

Define the shared state¶

Every branch returns one final string under the same results key. The appender channel preserves all three updates instead of letting later branches overwrite earlier ones.

In [3]:

import org.bsc.langgraph4j.state.AgentState;
import org.bsc.langgraph4j.state.Channel;
import org.bsc.langgraph4j.state.Channels;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

class ResearchState extends AgentState {
    static final String RESULTS = "results";

    static final Map<String, Channel<?>> SCHEMA = Map.of(
            RESULTS, Channels.appender(ArrayList::new)
    );

    ResearchState(Map<String, Object> initData) {
        super(initData);
    }

    List<String> results() {
        return this.<List<String>>value(RESULTS).orElseGet(List::of);
    }
}

import org.bsc.langgraph4j.state.AgentState; import org.bsc.langgraph4j.state.Channel; import org.bsc.langgraph4j.state.Channels; import java.util.ArrayList; import java.util.List; import java.util.Map; class ResearchState extends AgentState { static final String RESULTS = "results"; static final Map<String, Channel<?>> SCHEMA = Map.of( RESULTS, Channels.appender(ArrayList::new) ); ResearchState(Map<String, Object> initData) { super(initData); } List<String> results() { return this.<List<String>>value(RESULTS).orElseGet(List::of); } }

Build a streaming node¶

The helper creates a node whose value is an AsyncGenerator. It sends each partial message to onChunk immediately and wraps the same message in StreamingOutput. When the source is finished, Data.done(...) carries the state update that the parallel node will merge.

In [4]:

import org.bsc.async.AsyncGenerator;
import org.bsc.async.v5.AsyncGeneratorFlow;
import org.bsc.langgraph4j.action.AsyncNodeAction;
import org.bsc.langgraph4j.streaming.StreamingOutput;

import java.util.function.Consumer;

AsyncNodeAction<ResearchState> streamingNode(
        String nodeId,
        List<String> chunks,
        String finalResult,
        Consumer<String> onChunk) {
    return state -> {
        var generator = AsyncGeneratorFlow.<StreamingOutput<ResearchState>>create(dispatcher -> {
            for (var chunk : chunks) {
                onChunk.accept("%s: %s".formatted(nodeId, chunk));
                dispatcher.dispatchAsync(AsyncGenerator.Data.of(
                        new StreamingOutput<>(chunk, nodeId, state, null)));
            }
            dispatcher.dispatchAsync(AsyncGenerator.Data.done(
                    Map.of(ResearchState.RESULTS, finalResult)));
        });

        return java.util.concurrent.CompletableFuture.completedFuture(
                Map.of(nodeId + "_stream", generator));
    };
}

import org.bsc.async.AsyncGenerator; import org.bsc.async.v5.AsyncGeneratorFlow; import org.bsc.langgraph4j.action.AsyncNodeAction; import org.bsc.langgraph4j.streaming.StreamingOutput; import java.util.function.Consumer; AsyncNodeAction<ResearchState> streamingNode( String nodeId, List<String> chunks, String finalResult, Consumer<String> onChunk) { return state -> { var generator = AsyncGeneratorFlow.<StreamingOutput<ResearchState>>create(dispatcher -> { for (var chunk : chunks) { onChunk.accept("%s: %s".formatted(nodeId, chunk)); dispatcher.dispatchAsync(AsyncGenerator.Data.of( new StreamingOutput<>(chunk, nodeId, state, null))); } dispatcher.dispatchAsync(AsyncGenerator.Data.done( Map.of(ResearchState.RESULTS, finalResult))); }); return java.util.concurrent.CompletableFuture.completedFuture( Map.of(nodeId + "_stream", generator)); }; }

Create the fork-join graph¶

The three edges from START form the parallel step. All branches then join at summarize; the explicit join ensures the merged state continues to END.

In [5]:

import org.bsc.langgraph4j.StateGraph;

import static org.bsc.langgraph4j.StateGraph.END;
import static org.bsc.langgraph4j.StateGraph.START;

Consumer<String> printChunk = message -> System.out.println("progress | " + message);

var workflow = new StateGraph<>(ResearchState.SCHEMA, ResearchState::new)
        .addNode("web_search", streamingNode(
                "web_search",
                List.of("searching docs", "ranking matches"),
                "web_search: LangGraph4j documentation",
                printChunk))
        .addNode("database", streamingNode(
                "database",
                List.of("opening connection", "reading rows"),
                "database: 3 matching records",
                printChunk))
        .addNode("api", streamingNode(
                "api",
                List.of("sending request", "parsing response"),
                "api: service healthy",
                printChunk))
        .addNode("summarize", AsyncNodeAction.<ResearchState>node_async(state -> Map.of()))
        .addEdge(START, "web_search")
        .addEdge(START, "database")
        .addEdge(START, "api")
        .addEdge("web_search", "summarize")
        .addEdge("database", "summarize")
        .addEdge("api", "summarize")
        .addEdge("summarize", END)
        .compile();

import org.bsc.langgraph4j.StateGraph; import static org.bsc.langgraph4j.StateGraph.END; import static org.bsc.langgraph4j.StateGraph.START; Consumer<String> printChunk = message -> System.out.println("progress | " + message); var workflow = new StateGraph<>(ResearchState.SCHEMA, ResearchState::new) .addNode("web_search", streamingNode( "web_search", List.of("searching docs", "ranking matches"), "web_search: LangGraph4j documentation", printChunk)) .addNode("database", streamingNode( "database", List.of("opening connection", "reading rows"), "database: 3 matching records", printChunk)) .addNode("api", streamingNode( "api", List.of("sending request", "parsing response"), "api: service healthy", printChunk)) .addNode("summarize", AsyncNodeAction.<ResearchState>node_async(state -> Map.of())) .addEdge(START, "web_search") .addEdge(START, "database") .addEdge(START, "api") .addEdge("web_search", "summarize") .addEdge("database", "summarize") .addEdge("api", "summarize") .addEdge("summarize", END) .compile();

Run the branches concurrently¶

Attach an executor to START, the source of the parallel edges. Progress lines may be interleaved because the branches run concurrently. Always shut down an application-owned executor when it is no longer needed.

In [6]:

import org.bsc.langgraph4j.GraphInput;
import org.bsc.langgraph4j.RunnableConfig;

import java.util.concurrent.Executors;

var executor = Executors.newFixedThreadPool(3);
try {
    var config = RunnableConfig.builder()
            .addParallelNodeExecutor(START, executor)
            .build();

    var result = workflow.invoke(GraphInput.noArgs(), config).orElseThrow();
    System.out.println("final results | " + result.results());
} finally {
    executor.shutdownNow();
}

import org.bsc.langgraph4j.GraphInput; import org.bsc.langgraph4j.RunnableConfig; import java.util.concurrent.Executors; var executor = Executors.newFixedThreadPool(3); try { var config = RunnableConfig.builder() .addParallelNodeExecutor(START, executor) .build(); var result = workflow.invoke(GraphInput.noArgs(), config).orElseThrow(); System.out.println("final results | " + result.results()); } finally { executor.shutdownNow(); }

SLF4J: No SLF4J providers were found.
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See https://www.slf4j.org/codes.html#noProviders for further details.

progress | web_search: searching docs
progress | api: sending request
progress | database: opening connection
progress | database: reading rows
progress | web_search: ranking matches
progress | api: parsing response
final results | [web_search: LangGraph4j documentation, database: 3 matching records, api: service healthy]

Key points¶

• Return the generator as a value in the node update map.
• Use AsyncGeneratorFlow.create(...) (or another HasResultValue generator) when Data.done(...) carries final state.
• Handle partial chunks in the branch callback while the generator is being consumed.
• Use an appender channel when multiple branches update the same list.
• Join all branches at one node so the merged state continues through the graph.