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
{
    public static function make(WorkflowState $state): self
    {
        return parent::make($state)
            ->addNodes([
                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
        $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 user@example.com'
                ),
            ],
            message: 'These operations require approval'
        ));

        // After resume, $actions contain user decisions
        $resumeRequest = $this->consumeResumeRequest();

        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
    $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');
$workflowId = 'workflow_' . uniqid();

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

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

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

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

Checkpoints cache expensive operations across interruptions:

class DataProcessingNode extends Node
{
    public function __invoke(ProcessEvent $event, WorkflowState $state): ResultEvent
    {
        // This expensive operation runs only once
        $data = $this->checkpoint('fetch_data', function() {
            return $this->fetchExpensiveData();
        });

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

        // When resumed, $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

php 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 expensive operations

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
{
    public static function make(WorkflowState $state): self
    {
        return parent::make($state)
            ->addNodes([
                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();
    }
}

Workflow vs Agent

Use Workflow when:

• You need complete control over execution flow
• Building custom orchestration patterns
• Need complex branching/looping logic
• Want to use individual components (providers, embeddings, etc.) independently

Use Agent when:

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

Testing Workflows

use PHPUnit\Framework\TestCase;

class MyWorkflowTest extends TestCase
{
    public function testWorkflowExecution(): void
    {
        $state = new WorkflowState(['input' => 'test']);
        $workflow = MyWorkflow::make($state);
        $finalState = $workflow->start()->run();

        $this->assertTrue($finalState->has('result'));
    }
}