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 { 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: ```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 $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 ```php 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 ```php 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 ```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 Checkpoints cache expensive operations across interruptions: ```php 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. ```php use NeuronAI\Workflow\Exporter\MermaidExporter; $workflow->setExporter(new MermaidExporter()); $diagram = $workflow->export(); // Produces Mermaid flowchart showing event→node flow ``` ## CLI Generation ```bash 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 ```php class SequentialWorkflow extends Workflow { public static function make(WorkflowState $state): self { return parent::make($state) ->addNodes([ 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(); } } ``` ## 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 ```php 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')); } } ```