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: ```php 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 ```php 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 ```php 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 ```php 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: ```php $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 ```php 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 user@example.com' ), ], message: 'These operations require approval' )); foreach ($resumeRequest->actions as $action) { if ($action->decision === ActionDecision::Approved) { $this->executeAction($action->id); } } return new ResultEvent(...); } } ``` ### Conditional Interruption ```php 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 ```php 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. ```php 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 ```php 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 ```php // 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. ```php 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 ```php $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: ```php 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. ```php use NeuronAI\Workflow\Exporter\MermaidExporter; $workflow->setExporter(new MermaidExporter()); $diagram = $workflow->export(); // Produces Mermaid flowchart showing event→node flow ``` ## CLI Generation ```bash 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 ```php class SequentialWorkflow extends Workflow { /** * @return NodeInterface[] */ protected function nodes(): array { return [ new ValidationNode(), new ProcessingNode(), new OutputNode(), ]; } } ``` ### Branching Logic ```php 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 ```php 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