neuron-core/neuron-workflow-architect

Neuron AI Workflow Architect

This skill helps you build custom event-driven workflows in Neuron AI. Workflows are the foundation of the entire framework - Agent and RAG are built on top of Workflow.

Core Concepts

Event-Driven Architecture

Workflows operate through events flowing between nodes:

StartEvent → Node1 → Event2 → Node2 → Event3 → Node3 → StopEvent

Each node:

1. Receives a typed Event
2. Processes it
3. Returns a new Event (or StopEvent to complete)

The Node Pattern

Nodes extend the Node base class:

use NeuronAI\Workflow\Node;
use NeuronAI\Workflow\Event;
use NeuronAI\Workflow\StartEvent;
use NeuronAI\Workflow\StopEvent;
use NeuronAI\Workflow\WorkflowState;

class ValidationNode extends Node
{
    // The __invoke signature determines which event this node handles
    public function __invoke(StartEvent $event, WorkflowState $state): ProcessEvent
    {
        $input = $state->get('input');
        $validated = $this->validate($input);
        $state->set('validated', $validated);
        return new ProcessEvent($validated);
    }

    private function validate(mixed $input): array
    {
        // Validation logic
        return ['valid' => true, 'data' => $input];
    }
}

Key Pattern: The workflow automatically maps events to nodes based on the first parameter type of __invoke().

Defining Custom Events

use NeuronAI\Workflow\Event;

class UserValidatedEvent implements Event
{
    public function __construct(
        public readonly string $userId,
        public readonly array $userData
    ) {}
}

class ProcessCompleteEvent implements Event
{
    public function __construct(
        public readonly string $result
    ) {}
}

Events should:

• Implement the Event interface
• Use readonly properties for immutability
• Contain all data needed by the handling node

Creating a Workflow

Basic Workflow

use NeuronAI\Workflow\Workflow;
use NeuronAI\Workflow\WorkflowState;
use NeuronAI\Workflow\StartEvent;
use NeuronAI\Workflow\StopEvent;

$state = new WorkflowState([
    'input' => $userData,
]);

$workflow = Workflow::make($state)
    ->addNodes([
        new ValidationNode(),
        new ProcessingNode(),
        new OutputNode(),
    ]);

$handler = $workflow->start();
$finalState = $handler->run();
$result = $finalState->get('result');

Using the Static Constructor

class MyWorkflow extends Workflow
{
    /**
     * @return NodeInterface[]
     */
    protected function nodes(): array
    {
        return [
            new ValidationNode(),
            new ProcessingNode(),
        ];
    }
}

Workflow State

WorkflowState is a shared state container that persists across all nodes:

$state = new WorkflowState();

// Set values
$state->set('user_id', 123);
$state->set('data', ['key' => 'value']);

// Get values
$userId = $state->get('user_id');
$default = $state->get('missing_key', 'default_value');

// Check existence
if ($state->has('data')) {
    // Data exists
}

// Get subset of state
$subset = $state->only(['user_id', 'data']);

// Delete value
$state->delete('data');

// Get all state
$all = $state->all();

Human-in-the-Loop Patterns

Workflows support interruption for human intervention at any point.

Interrupting a Node

use NeuronAI\Workflow\Interrupt\ApprovalRequest;
use NeuronAI\Workflow\Interrupt\Action;

class DangerousOperationNode extends Node
{
    public function __invoke(ProcessEvent $event, WorkflowState $state): ResultEvent
    {
        // Interrupt for approval
        $resumeRequest = $this->interrupt(new ApprovalRequest(
            actions: [
                new Action(
                    id: 'delete_files',
                    name: 'Delete Files',
                    description: 'Delete all files in /tmp/uploads'
                ),
                new Action(
                    id: 'send_email',
                    name: 'Send Notification',
                    description: 'Send email to [email protected]'
                ),
            ],
            message: 'These operations require approval'
        ));

        foreach ($resumeRequest->actions as $action) {
            if ($action->decision === ActionDecision::Approved) {
                $this->executeAction($action->id);
            }
        }

        return new ResultEvent(...);
    }
}

Conditional Interruption

public function __invoke(ProcessEvent $event, WorkflowState $state): ResultEvent
{
    $cost = $state->get('estimated_cost');

    // Only interrupt if cost exceeds threshold
    $resumeRequest = $this->interruptIf(
        $cost > 1000,
        new ApprovalRequest(
            actions: [/* ... */],
            message: "Operation costs $${cost}. Approval required."
        )
    );

    return new ResultEvent(...);
}

Persistence for Interruptions

use NeuronAI\Workflow\Persistence\FilePersistence;

$persistence = new FilePersistence('/tmp/workflows');

$workflow = Workflow::make($persistence)
    ->addNodes([...]);

try {
    $handler = $workflow->start();
    $result = $handler->run();
} catch (WorkflowInterrupt $interrupt) {
    // Present to user
    $request = $interrupt->getRequest();
    $workflowId = $interrupt->getWorkflowId();

    // After user makes decisions:
    $resumeRequest = $this->getUserDecisions($request);
    $result = $workflow->init($resumeRequest)->run();
}

Checkpoints

Nodes can use checkpoints to cache operations happening before the interruption point.

class DataProcessingNode extends Node
{
    public function __invoke(ProcessEvent $event, WorkflowState $state): ResultEvent
    {
        // When resumed,
        // $data is retrieved from checkpoint
        $data = $this->checkpoint('fetch_data', function() {
            return $this->fetchExpensiveData();
        });

        // Might interrupt here
        $resumeRequest = $this->interruptIf($needsApproval, new ApprovalRequest(...));

        if ($resumeRequest->getAction('check')->isApproved()) {
            return new ResultEvent($data);
        }

        return new AnotherEvent();
    }
}

Middleware System

Middleware wraps node execution for cross-cutting concerns.

Creating Custom Middleware

use NeuronAI\Workflow\Middleware\WorkflowMiddleware;
use NeuronAI\Workflow\NodeInterface;
use NeuronAI\Workflow\Event;

class LoggingMiddleware implements WorkflowMiddleware
{
    public function __construct(private \Psr\Log\LoggerInterface $logger) {}

    public function before(NodeInterface $node, Event $event, WorkflowState $state): void
    {
        $this->logger->info("Executing: " . $node::class);
    }

    public function after(NodeInterface $node, Event $event, Event|Generator $result, WorkflowState $state): void
    {
        $this->logger->info("Completed: " . $node::class);
    }
}

Registering Middleware

// Node-specific middleware
$workflow->middleware(ProcessingNode::class, new LoggingMiddleware($logger));

// Multiple middleware on one node
$workflow->middleware(ProcessingNode::class, [
    new ValidationMiddleware(),
    new LoggingMiddleware(),
]);

// Global middleware (runs on all nodes)
$workflow->globalMiddleware(new PerformanceMiddleware());

Execution Order

before() calls → Node execution → after() calls

All before() methods execute in registration order, then the node, then all after() methods.

Streaming Support

Nodes can return Generator to yield intermediate results.

class ProcessingNode extends Node
{
    public function __invoke(ProcessEvent $event, WorkflowState $state): \Generator
    {
        yield new ProgressEvent("Starting process...");

        $result = $this->longRunningOperation();

        yield new ProgressEvent("Completed!");

        return new ResultEvent($result);
    }
}

Consuming Streams

$handler = $workflow->start();

foreach ($handler->events() as $event) {
    if ($event instanceof ProgressEvent) {
        echo $event->message . PHP_EOL;
    }
}

$finalState = $handler->run();

Checkpoint System

Checkpoint cache operation results across interruptions:

class DataProcessingNode extends Node
{
    public function __invoke(ProcessEvent $event, WorkflowState $state): ResultEvent
    {
        // When resumed, $data is retrieved from checkpoint
        $data = $this->checkpoint('fetch_data', function() {
            return $this->fetchExpensiveData();
        });

        // Might interrupt here
        $resumeRequest = $this->interruptIf($needsApproval, new ApprovalRequest(...));

        if (!$resumeRequest->isApproved()) {
            // ...
        }

        // $data is retrieved from checkpoint
        $result = $this->process($data);

        return new ResultEvent($result);
    }
}

Workflow Export

Export workflows to diagram formats for visualization.

use NeuronAI\Workflow\Exporter\MermaidExporter;

$workflow->setExporter(new MermaidExporter());
$diagram = $workflow->export();

// Produces Mermaid flowchart showing event→node flow

CLI Generation

vendor/bin/neuron make:workflow DataProcessingWorkflow

Best Practices

Node Design

• Keep nodes focused and single-purpose
• Use typed events for input/output
• Make nodes testable in isolation
• Use checkpoints for operations before interruption points

State Management

• Store shared data in WorkflowState, not node properties
• Use descriptive keys for state data
• Clean up state that's no longer needed

Middleware

• Use middleware for cross-cutting concerns
• Order matters - register in logical sequence
• Prefer node-specific middleware over global

Interruptions

• ALWAYS configure persistence when using interruptions
• Provide clear, actionable descriptions in InterruptRequest
• Use checkpoints to avoid re-running expensive operations

Common Patterns

Sequential Processing

class SequentialWorkflow extends Workflow
{
    /**
     * @return NodeInterface[]
     */
    protected function nodes(): array
    {
        return [
            new ValidationNode(),
            new ProcessingNode(),
            new OutputNode(),
        ];
    }
}

Branching Logic

class RouterNode extends Node
{
    public function __invoke(ProcessEvent $event, WorkflowState $state): Event
    {
        if ($state->get('priority') === 'high') {
            return new HighPriorityEvent($event->data);
        }
        return new LowPriorityEvent($event->data);
    }
}

Loop Pattern

class LoopNode extends Node
{
    public function __invoke(ProcessEvent $event, WorkflowState $state): Event
    {
        $items = $state->get('items');
        $current = $state->get('current_index', 0);

        if ($current < count($items)) {
            $state->set('current_item', $items[$current]);
            $state->set('current_index', $current + 1);
            return new ProcessItemEvent($items[$current]);
        }

        return new StopEvent();
    }
}

Parallel Execution

When a node needs to run multiple sub-tasks concurrently (e.g. extracting structured data from an image while also generating a description), use ParallelEvent to fork execution into parallel branches.

How It Works

ForkNode → ParallelEvent([branch1 => EventA, branch2 => EventB])
              ├─ BranchA → NodeA → StopEvent(resultA)
              └─ BranchB → NodeB → StopEvent(resultB)
           → JoinNode (reads results from ParallelEvent) → StopEvent

1. A fork node returns a ParallelEvent subclass with branch-starting events.
2. The executor runs each branch independently until StopEvent.
3. Each branch's StopEvent::getResult() is collected into the ParallelEvent.
4. A join node (whose __invoke() accepts the ParallelEvent subclass) reads the results.

Step 1 — Define a ParallelEvent Subclass

use NeuronAI\Workflow\Events\ParallelEvent;

class ImageAnalysisParallelEvent extends ParallelEvent {}

Step 2 — Create the Branch Events

use NeuronAI\Workflow\Events\Event;

class ExtractStructuredDataEvent implements Event
{
    public function __construct(public readonly string $imageUrl) {}
}

class GenerateDescriptionEvent implements Event
{
    public function __construct(public readonly string $imageUrl) {}
}

Step 3 — Create the Fork Node

use NeuronAI\Workflow\Events\StartEvent;
use NeuronAI\Workflow\Node;
use NeuronAI\Workflow\WorkflowState;

class AnalyzeImageForkNode extends Node
{
    public function __invoke(StartEvent $event, WorkflowState $state): ImageAnalysisParallelEvent
    {
        $imageUrl = $state->get('image_url');

        return new ImageAnalysisParallelEvent([
            'structured' => new ExtractStructuredDataEvent($imageUrl),
            'description' => new GenerateDescriptionEvent($imageUrl),
        ]);
    }
}

Branch IDs come from the array keys ('structured', 'description'). If you pass a sequential array, IDs are auto-derived from each event's short class name.

Step 4 — Create Branch Nodes (Each Ends with StopEvent)

use NeuronAI\Agent;
use NeuronAI\Providers\OpenAI\OpenAI;
use NeuronAI\HttpClient\AmpHttpClient;
use NeuronAI\Workflow\Events\StopEvent;
use NeuronAI\Workflow\Node;
use NeuronAI\Workflow\WorkflowState;

class ExtractStructuredDataNode extends Node
{
    public function __invoke(ExtractStructuredDataEvent $event, WorkflowState $state): StopEvent
    {
        $agent = Agent::make()
            ->setProvider(
                (new OpenAI(getenv('OPENAI_API_KEY'), 'gpt-4o'))
                    ->setHttpClient(new AmpHttpClient())
            )
            ->setTools([/* ... */])
            ->addSystemTip('Extract structured data from the image.');

        $result = $agent->structured(/* your structured output class */);

        return new StopEvent(result: $result);
    }
}

class GenerateDescriptionNode extends Node
{
    public function __invoke(GenerateDescriptionEvent $event, WorkflowState $state): StopEvent
    {
        $agent = Agent::make()
            ->setProvider(
                (new OpenAI(getenv('OPENAI_API_KEY'), 'gpt-4o'))
                    ->setHttpClient(new AmpHttpClient())
            )
            ->addSystemTip('Describe the image in detail.');

        $description = $agent->chat($event->imageUrl);

        return new StopEvent(result: $description);
    }
}

Step 5 — Create the Join Node

class MergeAnalysisNode extends Node
{
    public function __invoke(ImageAnalysisParallelEvent $event, WorkflowState $state): StopEvent
    {
        $structuredData = $event->getResult('structured');
        $description = $event->getResult('description');

        $state->set('analysis', [
            'data' => $structuredData,
            'description' => $description,
        ]);

        return new StopEvent();
    }
}

Step 6 — Wire Up the Workflow

$workflow = Workflow::make()
    ->addNodes([
        new AnalyzeImageForkNode(),
        new ExtractStructuredDataNode(),
        new GenerateDescriptionNode(),
        new MergeAnalysisNode(),
    ]);

$state = $workflow->init(new WorkflowState(['image_url' => 'https://example.com/photo.jpg']));
$result = $state->run();

Sequential vs Concurrent Execution

By default, WorkflowExecutor runs branches sequentially (one after another). For true concurrency, use AsyncExecutor:

use NeuronAI\Workflow\Executor\AsyncExecutor;
use NeuronAI\Workflow\Workflow;

$workflow = Workflow::make()
    ->setExecutor(new AsyncExecutor())
    ->addNodes([
        new AnalyzeImageForkNode(),
        new ExtractStructuredDataNode(),
        new GenerateDescriptionNode(),
        new MergeAnalysisNode(),
    ]);

AsyncExecutor is a drop-in replacement — it runs branches as concurrent Amp futures while keeping linear (non-parallel) nodes sequential as usual.

AsyncWorkflow with AmpHttpClient

For fully asynchronous execution where branches make HTTP calls to AI providers concurrently, combine AsyncExecutor with AmpHttpClient:

• AsyncExecutor runs parallel branches as concurrent Amp fibers (non-blocking).
• AmpHttpClient is the async HTTP client built on amphp/http-client. Inject it on the provider via ->setHttpClient(new AmpHttpClient()) to ensure HTTP calls inside each branch are non-blocking.

Without AmpHttpClient, each branch's HTTP call would block its fiber, negating the concurrency benefit. With it, all branches make their API calls truly in parallel — a workflow that extracts structured data and generates a description simultaneously completes in the time of the slower branch, not the sum of both.

use NeuronAI\HttpClient\AmpHttpClient;
use NeuronAI\Providers\OpenAI\OpenAI;

$provider = (new OpenAI(getenv('OPENAI_API_KEY'), 'gpt-4o'))
    ->setHttpClient(new AmpHttpClient());

Parallel Branches with Interruptions

Parallel branches fully support human-in-the-loop. If any branch calls $this->interrupt(), the executor throws a WorkflowInterrupt with parallel context:

use NeuronAI\Workflow\Interrupt\WorkflowInterrupt;

try {
    $result = $workflow->init()->run();
} catch (WorkflowInterrupt $interrupt) {
    if ($interrupt->isParallelInterrupt()) {
        // $interrupt->getBranchId() — which branch interrupted
        // $interrupt->getCompletedBranchResults() — results from branches that finished
        // Present interrupt to user...
    }
}

// After user responds:
$handler = $workflow->init($interrupt->getRequest());
$result = $handler->run();
// Resuming skips already-completed branches, only re-runs the interrupted one.

Use Checkpoint inside branch nodes for expensive operations that should not re-run after resume:

class ExtractStructuredDataNode extends Node
{
    public function __invoke(ExtractStructuredDataEvent $event, WorkflowState $state): StopEvent
    {
        $data = $this->checkpoint('fetch_image', fn() => $this->fetchExpensiveImageData());

        $resumeRequest = $this->interruptIf(
            $this->needsApproval($data),
            new ApprovalRequest(actions: [...], message: 'Review extracted data')
        );

        return new StopEvent(result: $data);
    }
}

Workflow vs Agent

Use Workflow when:

• You need complete control over the execution flow
• Building custom orchestration patterns
• Need complex branching/looping logic
• Want to run multiple agents in parallel for heavy tasks
• Want to use individual components (audio providers, embeddings, etc.) independently

Use Agent when:

• Building chat-based applications
• Need tool calling
• Want built-in features (chat history, streaming, structured output)
• Following common conversational patterns