Dependencias entre flujos de trabajo

Use esperas por dependencias entre flujos de trabajo cuando un flujo de trabajo de nivel superior solo deba continuar después de que se hayan completado otros flujos de trabajo. Así se coordinan flujos de trabajo sin tener que reunirlo todo en una única definición enorme.

Estas esperas combinan especialmente bien con las transferencias asíncronas, en las que un flujo de trabajo inicia primero flujos de trabajo secundarios independientes y espera por ellos más adelante.

Para ver ejemplos más completos de planificador y ejecutor, consulte Orquestación de agentes.

Queuety ofrece tres métodos de constructor:

• wait_for_workflow() para un flujo de trabajo
• wait_for_workflows() para varios flujos de trabajo con semántica all, any o quorum
• wait_for_workflow_group() cuando los ID de los flujos de trabajo proceden de un grupo iniciado anteriormente

Esperar a un flujo de trabajo

$research_id = Queuety::workflow( 'research_run' )
    ->then( PlanResearchStep::class )
    ->for_each( 'tasks', ExecuteResearchTask::class, 'results' )
    ->dispatch( [ 'brief_id' => 42 ] );

$editorial_id = Queuety::workflow( 'editorial_run' )
    ->wait_for_workflow( $research_id, 'research' )
    ->then( DraftBriefStep::class )
    ->dispatch();

Cuando se completa research_run, editorial_run se reanuda. Si se establece result_key, el estado de la dependencia completada se almacena bajo esa clave:

$state['research']; // public state from the completed research workflow

Si no se especifica result_key, el estado público del flujo de trabajo dependiente se combina con el estado de nivel superior del flujo de trabajo en espera.

Esperar a varios flujos de trabajo

use Queuety\Enums\WaitMode;

Queuety::workflow( 'synthesis_run' )
    ->wait_for_workflows(
        [ $research_id, $editorial_id, $legal_id ],
        WaitMode::All,
        'dependencies'
    )
    ->then( PublishPacketStep::class )
    ->dispatch();

• WaitMode::All se reanuda solo cuando se han completado todos los flujos de trabajo indicados.
• WaitMode::Any se reanuda cuando se completa el primero de los flujos de trabajo indicados.
• WaitMode::Quorum se reanuda cuando se ha completado correctamente la cantidad requerida de flujos de trabajo.

Queuety::workflow( 'synthesis_run' )
    ->wait_for_workflows(
        [ $research_id, $editorial_id, $legal_id ],
        WaitMode::Quorum,
        'dependencies',
        quorum: 2,
    )
    ->then( PublishPacketStep::class )
    ->dispatch();

Con result_key, las esperas por varios flujos de trabajo almacenan el estado de las dependencias completadas, indexado por ID de flujo de trabajo:

[
    '42' => [ 'summary' => '...' ],
    '43' => [ 'summary' => '...' ],
]

Si no se especifica result_key, el estado de las dependencias completadas se combina con el estado de nivel superior del flujo de trabajo en espera según el orden de las dependencias.

Resolver los ID de dependencias desde el estado del flujo de trabajo

Puede aplazar la lista de dependencias hasta el tiempo de ejecución leyendo los ID de los flujos de trabajo del estado público:

Queuety::workflow( 'agent_follow_up' )
    ->wait_for_workflow( 'parent_workflow_id', 'parent' )
    ->then( FinalizeAgentRunStep::class )
    ->dispatch( [ 'parent_workflow_id' => $workflow_id ] );

Para varias dependencias:

Queuety::workflow( 'aggregator' )
    ->wait_for_workflows( 'dependency_ids', WaitMode::Any, 'winner' )
    ->dispatch( [ 'dependency_ids' => [ 12, 18, 22 ] ] );

Esperar a un grupo de flujos de trabajo iniciado

Si un paso anterior start_workflows() o start_agents() almacenó una clave de grupo duradera, puede esperar directamente a ese grupo en lugar de pasar los ID de los flujos de trabajo por el estado público.

Queuety::workflow( 'brief_research' )
    ->then( PlanTopicsStep::class )
    ->start_agents( 'agent_tasks', $agent_run, group_key: 'researchers' )
    ->wait_for_agent_group( 'researchers', WaitMode::Quorum, 2, 'agent_results' )
    ->then( SynthesizeBriefStep::class )
    ->dispatch();

Qué ocurre mientras se espera

Cuando un flujo de trabajo llega a uno de estos pasos, pasa al estado waiting_for_workflows. Ningún worker permanece vinculado a él mientras espera.

$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value; // 'waiting_for_workflows'
echo $state->wait_type;     // 'workflow'
print_r( $state->waiting_for ); // ['42', '43']

Si una dependencia ya se completó antes de que se alcance el paso de espera, el flujo de trabajo continúa de inmediato.

Semántica de los errores

Las esperas por dependencias entre flujos de trabajo se centran en los resultados satisfactorios:

• wait_for_workflow() y WaitMode::All fallan si falla o se cancela cualquier dependencia obligatoria.
• WaitMode::Any falla solo cuando todas las dependencias han alcanzado un estado terminal y ninguna se ha completado correctamente.
• WaitMode::Quorum falla cuando los flujos de trabajo activos restantes ya no pueden satisfacer el quórum requerido.

Esto hace que las esperas de flujos de trabajo sean útiles en sistemas de planificación y ejecución, donde un flujo de trabajo puede encargarse de cada tarea y otro puede esperar los resultados de all o la primera rama satisfactoria que termine.