Bucles y ramas

Workflow hace que la lógica de ramificación y bucles sea fácil de implementar gracias a su diseño orientado a eventos. Una vez que entiendes cómo los nodos pertenecen a los eventos, es fácil empezar a imaginar cómo puedes crear bucles y ramificaciones, que consiste simplemente en decidir qué evento debe devolverse en una "condición if" o cualquier otra lógica.

Bucles

Para crear un bucle, simplemente devuelve el evento de entrada de un nodo anterior como el evento de salida del nodo actual. También puedes usar el mismo evento de entrada como evento de salida del nodo actual para recorrer en bucle el nodo actual.

Echa un vistazo al ejemplo de abajo. El NodeOne puede tener dos eventos como tipo de retorno, FirstEvent y SecondEvent. Si el nodo devuelve FirstEvent, provocará otra ejecución del mismo nodo porque FirstEvent es manejado por sí mismo, creando un bucle.

Si el nodo devuelve SecondEvent finalmente avanzará la ejecución hacia otro nodo.

class NodeOne extends Node
{
    public function __invoke(FirstEvent $event, WorkflowState $state): FirstEvent|SecondEvent
    {
        echo "\n- ".$event->firstMsg;
        
        if (rand(0, 1) === 1) {
            // Al devolver FirstEvent se activará otra ejecución de NodeOne
            return new FirstEvent("Running a loop on NodeOne");
        }
        
        return new SecondEvent("NodeOne complete, move forward");
    }
}

Observa que el nodo ahora tiene dos tipos de retorno para el __invoke método: FirstEvent y SecondEvent. Debes declarar todos los eventos de retorno posibles en la firma del método para que Workflow construya la cadena de ejecución.

Devolver FirstEvent activará otra ejecución de NodeOne. Así que la salida final podría ser:

$state = Workflow::make()
    ->addNodes([
        new InitialNode(),
        new NodeOne(),
        new NodeTwo()
    ])
    ->init()
    ->run();

/*
- Manejo de StartEvent
- InitialNode complete
- Ejecutando un bucle en NodeOne
- Ejecutando un bucle en NodeOne
- NodeOne completo, avanza
- NodeTwo complete
*/

Puedes crear un bucle desde cualquier nodo hacia cualquier otro nodo del flujo de trabajo definiendo el evento de entrada apropiado y los eventos de retorno del método invoke.

El NodeOne incluso puede devolver un StartEvent para saltar directamente al primer nodo del Workflow. La arquitectura orientada a eventos te permite apuntar directamente a cualquier nodo del flujo de trabajo tanto hacia adelante como hacia atrás.

Ramas

Como ya has visto, puedes devolver condicionalmente distintos eventos desde un nodo para definir flujos de ejecución personalizados. En esta sección veremos un ejemplo de un flujo de trabajo que se ramifica en dos rutas diferentes.

Primero vamos a crear algunos eventos personalizados:

namespace App\Neuron;

class BrancheA1Event implements Event 
{
    public function __construct(protected string $firstMsg){}
}

class BrancheA2Event implements Event 
{
    public function __construct(protected string $secondMsg){}
}

class BrancheB1Event implements Event 
{
    public function __construct(protected string $secondMsg){}
}

class BrancheB2Event implements Event 
{
    public function __construct(protected string $secondMsg){}
}

En el nodo inicial del flujo de trabajo decidimos por qué rama queremos pasar. Recuerda definir siempre los tipos de retorno apropiados en la __invoke firma del método:

class InitialNode extends Node
{
    public function __invoke(StartEvent $event, WorkflowState $state): BrancheA1Event|BrancheB1Event
    {
        if (rand(0, 1) === 1) {
            // Al devolver FirstEvent se activará otra ejecución de NodeOne
            return new BrancheA1Event();
        }
        
        return new BrancheB1Event();
    }
}

Los otros nodos avanzarán secuencialmente.

$state = Workflow::make()
    ->addNodes([
        new InitialNode(),
        new A1Node(),
        new A2Node(),
        new B1Node(),
        new B2Node(),
    ])
    ->init()
    ->run();

Por supuesto, puedes combinar ramas y bucles en cualquier orden para satisfacer las necesidades de tu aplicación.

Ramas en paralelo

Cuando quieras invocar la ejecución de múltiples ramas en paralelo, necesitas devolver el evento especial ParallelEvent desde tu nodo.

use NeuronAI\Workflow\Events\ParallelEvent;

class DocumentProcessing extends Node
{
    public function __invoke(StartEvent $event, WorkflowState $state): ParallelEvent
    {
        // Lógica del nodo aquí...
	
        // Finalmente devuelve un ParallelEvent
        return new ParallelEvent([
            'text' => new TextProcessEvent(),
            'image' => new ImageProcessEvent(),
        ]);
    }
}

El ParallelEvent debe construirse con un array de <branch_name> => <FirstInputEvent>:

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

Los nodos que manejan los eventos que declares para cada rama deben registrarse en el flujo de trabajo:

class MyWorkflow extends Workflow
{
	protected function nodes(): array
	{
		return [
			new DocumentProcessing(),
			
			// rama "text"
			new DescriptionGenerationNode(), // Maneja TextProcessEvent
			new TextRefactorNode(),
			
			// rama "image"
			new ImageProcessNode(), // Maneja ImageProcessEvent
			new AddWatermarkNode(),
			
			new MergeNode(),
		];
	}
}

Una rama puede ser solo un nodo, o una lista de múltiples nodos siempre conectados con eventos.

Gestiona el FINAL de las ramas

El último nodo de tu rama debe devolver el StopEvent.

En el ejemplo anterior TextRefactorNode y AddWatermarkNode declararán el final de su rama devolviendo StopEvent:

class AddWatermarkNode extends Node
{
    public function __invoke(TextProcessEvent $event, WorkflowState $state): StopEvent
    {
        // Código del nodo aquí...
		
        // Al devolver StopEvent, la rama termina
        return new StopEvent(result: 'Hello World!');
    }
}

Nota: StopEvent también puede llevar algunos resultados.

Obtén el resultado de las ramas

El ParallelEvent devuelto por el DocumentProcessing nodo básicamente espera el final de la ejecución de las ramas antes de ser encaminado al siguiente nodo.

En el ejemplo anterior, el MergeNode se encarga finalmente de gestionar el ParallelEvent:

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

Este nodo puede leer el resultado final de cada rama con el getResult() método pasando el <branch_name>.

Como de costumbre, el nodo de fusión puede detener el flujo de trabajo o devolver otros eventos que lo hagan avanzar.

Aislamiento del estado de las ramas

Un detalle que vale la pena señalar: cada rama obtiene una copia aislada del estado del flujo de trabajo. Empiezan con la misma instantánea, pero las mutaciones dentro de una rama no se propagan a las ramas hermanas ni al flujo de trabajo principal. La única forma de devolver datos es a través del StopEvent resultado.

Esto es intencional, evita toda una clase de errores de concurrencia en los que las ramas interfieren con el estado de las demás.

Ejecutor asíncrono

También proporcionamos una implementación del ejecutor interno del flujo de trabajo que te permite ejecutar múltiples ramas de forma concurrente. Para usar el Ejecutor asíncrono necesitas instalar el Amp paquete:

composer require amphp/amp

class MyAgent extends Workflow 
{
    /**
     * Usa el AsyncExecutor
     */
    protected function executor(): WorkflowExecutorInterface
    {
        return new AsyncExecutor();
    }

    protected function nodes(): array
    {
        return [...];
    }
}

Esto es especialmente útil si quieres ejecutar múltiples tareas agenticas en paralelo, ya que Neuron AI ya proporciona el AmpHttpClient que puedes inyectar en todos los componentes.

use NeuronAI\HttpClient\AmpHttpClient;

class DescriptionGenerationNode extends Node
{
    public function __invoke(TextProcessEvent $event, WorkflowState $state): StopEvent
    {
        $input = new UserMessage('Describe esta imagen');
        $input->addContent(
            new ImageContent(...)
        );

        $response = AsyncAgent::make()
            ->chat($input)
            ->getMessage();

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

Monitorización y depuración

Muchas de las aplicaciones que construyas con Neuron contendrán varios pasos con múltiples invocaciones de llamadas a LLM. A medida que estas aplicaciones se vuelven cada vez más complejas, resulta crucial poder inspeccionar exactamente qué está ocurriendo dentro de tu sistema agéntico. La mejor manera de hacerlo es con Inspector.

Neuron AI | Inspectordocs.inspector.dev