Feature Request: Parallel node execution in Workflow (fork/join pattern)

[jelen07]

Context

First of all — great work on the async capabilities in Neuron! The chatAsync() support (v1.12) and especially the HttpClientInterface abstraction with the AMP adapter in v3 are exactly what's needed for building scalable AI applications in PHP. The event-driven Workflow redesign in v2 is also a much cleaner foundation than the previous graph-based approach.

This feature request builds on top of that existing async infrastructure and proposes extending it into the Workflow layer.

Use case

Many real-world workflows contain independent branches that don't depend on each other. For example, a document analysis pipeline:

StartEvent
  ├── ExtractTextNode (~3s LLM call)
  ├── AnalyzeImagesNode (~5s LLM call)
  └── ClassifyMetadataNode (~2s LLM call)
MergeResultsNode
StopEvent

Currently, this runs sequentially (~10s total). If the independent nodes could run in parallel, it would complete in ~5s — leveraging the same async I/O that already works beautifully at the agent level.

Current behavior

Workflow::execute() processes nodes one at a time in a while loop:

Node A → wait for completion → Node B → wait for completion → Node C

Even when using AmpHttpClient inside individual nodes, the orchestration layer itself runs them sequentially — only one node executes at a time.

Proposed enhancement

Allow a node to emit multiple events that trigger parallel execution, with a join mechanism to collect results:

// Conceptual API — just to illustrate the idea
class AnalyzeDocument extends Node {
    public function __invoke(StartEvent $event, WorkflowState $state): ParallelEvent {
        return new ParallelEvent([
            new ExtractTextEvent(),
            new AnalyzeImagesEvent(),
            new ClassifyMetadataEvent(),
        ], joinOn: MergeResultsEvent::class);
    }
}

This would integrate naturally with the existing event-driven architecture — parallel branches are just multiple events dispatched simultaneously, and the join is simply a node that waits for all expected events before proceeding.

Inspiration

Both LlamaIndex Workflows and LangGraph (Python) solve this elegantly within an event-driven model similar to Neuron's:

• LlamaIndex: @step(num_workers=3) + ctx.collect_events() for fork/join
• LangGraph: Implicit parallelism from graph topology + edge lists for joins

Related issues

• Add support of batch and async #149 — Async agent calls (solved in v1.12)
• Use PSR HTTP client. #436 — PSR HTTP client abstraction (solved in v3)
• Documentation for parallel execution is not up-to-date with the codebase. #528 — Parallel tool calls within a single agent (solved via ParallelToolNode)
• Support parallel execution for evaluation command #485 — Parallel execution for evaluation command (open)

These progressively extended async capabilities from agent → tools → evaluation. Workflow feels like the natural next step in that journey.

[ilvalerione]

Thank you for the feedback. I was already looking in this direction. After the release of Neuron v3 it was quite clear that this is the next step.

There a several architectural challenges that need to be addressed for making parallel or async branches possible.

I'm in the process of testing different approaches. To be transparent it will take time, and I'm not sure if it will require a mejor release, because braking changes here are highly probable. Anyway it's already on our radar.

[ilvalerione]

I want to announce the full support for parallel branches.

I would like to keep this new feature in its dedicated branch async for a bit, to give you and other interested developers the chance to test it. The async repository branch will be regurarly updated with fixes from the main branch. So it will be ready for an immediate merge after this rollout period.

Install the async branch

Add the repository section to your composer.json file to get the latest commit:

  "repositories": [
    {
      "type": "vcs",
      "url": "https://github.com/neuron-core/neuron-ai"
    }
  ],

Update the neuron-ai entry to ask for the async branch:

"neuron-core/neuron-ai": "dev-async",

WorkflowExecutor

To make this feature open to different implementations, I extracted the logic of the Workflow execution in a brand new component called WorkflowExecutor.

The default component just replicate the previous behaviour for backward compatibility and run branches not really "in-parallel", but sequentially until the end of all branches.

Since the executor has its own interface, I also created the AsyncExecutor implementation that uses Amp to run multiple branches asynchronously.

How to run Parallel Branches

Inpired by this conversation, when you want to call the execution of multiple branches in parallel you need to return the special event: ParallelEvent

use NeuronAI\Workflow\Events\ParallelEvent;

class DocumentProcessing extends Node
{
    public function __invoke(StartEvent $event, WorkflowState $state): DocumentParallelEvent
    {
		// Your node logic here...
	
		// Finally return a ParallelEvent
        return new ParallelEvent([
            'text' => new TextProcessEvent(),
            'image' => new ImageProcessEvent(),
        ]);
    }
}

The ParallelEvent must be constructed with an array of <branch_name> => <FirstInputEvent>:

new ParallelEvent([
	<branch_name> => <FirstInputEvent>
	...
])

Nodes handling the events you declare for each branch must be registered in the workflow:

$workflow = Workflow::make()
	->addNodes([
		new DocumentProcessing(),
		
		// "text" branch
		new DescriptionGenerationNode(), // Handle TextProcessEvent
		new TextRefactorNode(),
		
		// "image" branch
		new ImageProcessNode(), // Handle ImageProcessEvent
		new AddWatermarkNode(),
		
		new MergeNode(),
	]);

A branch can be just one node, or a list of multiple nodes always connected with events.

How to handle the END of branches execution

The last node in your branch must return the framework built-in StopEvent.

In the example above TextRefactorNode and AddWatermarkNode will declare the end of their branch returning StopEvent:

class AddWatermarkNode extends Node
{
    public function __invoke(TextProcessEvent $event, WorkflowState $state): StopEvent
    {
		// Node code here...
		
		// Returning StopEvent the branch ends
        return new StopEvent(result: 'Hello World!');
    }
}

Notice: StopEvent can also carry some results.

Handle the final result

The ParallelEvent returned by the DocumentProcessing node is basically waiting the end of branches execution before being forwarded to the next node.

In the example above, the MergeNode is in charge to finally handle the ParallelEvent:

class MergeNode extends Node
{
    public function __invoke(ParallelEvent $event, WorkflowState $state): StopEvent
    {
        $textBranchResult = $event->getResult('text');
        $imageBranchResult = $event->getResult('image');
        
        return new StopEvent();
    }
}

This node can read the final result of each branch with the getResult method passing the <branch_name>.

As usual the node can stop the workflow, or return other events moving the workflow forward.

Branch State Isolation

Branches will receive a copy of the Workflow state, so they can start with the same Workflow information, but every change will affect only the branch state. They are completely isolated.

The only way to send result back to the main workflow is adding result data to the StopEvent.

AsyncExecutor

You need to install Amp to use this component:

composer require amphp/amp

class MyAgent extends Workflow 
{
	/**
     * Get the executor instance.
     */
    protected function executor(): WorkflowExecutorInterface
    {
        return new AsyncExecutor();
    }
	
	protected function nodes(): array
	{
		return [...];
	}
}

This is particularly useful if you want to run multiple agentic tasks in parallel, since Neuron AI already provides the AmpHttpClient that you can inject into all components.

use NeuronAI\HttpClient\AmpHttpClient;

class DescriptionGenerationNode extends Node
{
    public function __invoke(TextProcessEvent $event, WorkflowState $state): StopEvent
    {
		$input = new UserMessage('Describe this image');
		$input->addContent(
			new ImageContent(...)
		);
	
		$response = Agent::make()
			->setAIProvider(
				new Anthropic(
					key: 'KEY',
					model: 'MODEL',
					httpClient: new AmpHttpClient() // Async http client
				)
			)
			->chat($input)
			->getMessage()
		
        return new StopEvent(result: $response);
    }
}

Compatibility

Branches support interruption and middleware like any other workflow node.

Monitoring

We recommend to configure the INSPECTOR_INGESTION_KEY in your environment file to be able to explore the Workflow execution timeline in the Inspector dashboard.

Conclusion

Feel free to explore this new implementation and provide here your feedback, considerations, or issues. You can also explore the unit tests to get inspiration from real code.