Transferencias asíncronas

Use start_workflows() cuando un flujo de trabajo necesite transferir el trabajo descubierto a flujos de trabajo independientes de nivel superior, en lugar de mantenerlo todo dentro de una sola ejecución.

Si está modelando explícitamente ejecuciones de agentes, start_agents() y wait_for_agents() son alias semánticos con valores predeterminados orientados a agentes.

Para ver ejemplos más completos de principio a fin, consulte Orquestación de agentes.

Esto resulta útil para la orquestación de agentes:

• un flujo de trabajo planificador descubre tareas en tiempo de ejecución
• cada tarea se convierte en su propio flujo de trabajo duradero
• el flujo principal guarda los ID de esos flujos de trabajo en el estado
• más adelante, el flujo principal puede esperar a all, any, quorum o a un grupo iniciado con nombre

El método ->start_workflows()

use Queuety\Enums\WaitMode;
use Queuety\Queuety;

$agent_task = Queuety::workflow( 'agent_task' )
    ->then( ResearchTopicStep::class )
    ->then( SummarizeTopicStep::class );

Queuety::workflow( 'brief_research' )
    ->then( PlanTopicsStep::class )
    ->start_workflows(
        items_key: 'topics',
        workflow_builder: $agent_task,
        result_key: 'child_workflow_ids',
        group_key: 'researchers',
    )
    ->wait_for_workflow_group( 'researchers', WaitMode::All, result_key: 'child_results' )
    ->then( SynthesizeBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

En ese ejemplo:

1. PlanTopicsStep escribe una matriz de cargas útiles de temas en $state['topics']
2. start_workflows() inicia un flujo de trabajo agent_task por tema
3. Los ID de los flujos de trabajo iniciados se escriben en $state['child_workflow_ids']
4. wait_for_workflow_group() pausa la ejecución hasta que el grupo iniciado cumpla la condición de finalización elegida
5. SynthesizeBriefStep recibe los estados de los procesos secundarios completados en $state['child_results']

Cargas útiles de los procesos secundarios

Cada elemento de items_key se convierte en la carga útil inicial de un flujo de trabajo iniciado.

Si el elemento es una matriz, sus claves públicas se combinan con la carga útil del proceso secundario:

[
    [ 'topic' => 'pricing', 'source' => 'competitors' ],
    [ 'topic' => 'reviews', 'source' => 'customers' ],
]

Si el elemento es escalar, Queuety lo guarda en payload_key:

->start_workflows(
    items_key: 'topic_names',
    workflow_builder: $agent_task,
    payload_key: 'topic',
)

Heredar el estado del flujo principal

De forma predeterminada, start_workflows() combina el estado público del flujo de trabajo principal con la carga útil de cada proceso secundario.

Esto significa que el contexto compartido, como brief_id, account_id o campaign, está disponible automáticamente en cada flujo de trabajo iniciado.

Se excluyen de la herencia el items_key de origen y result_key, para que cada proceso secundario no reciba la lista completa de tareas ni la matriz final de ID de flujos de trabajo.

Establezca inherit_state: false cuando cada proceso secundario deba recibir únicamente la carga útil de su elemento.

¿Por qué no run_workflow()?

run_workflow() crea una relación entre el proceso principal y el secundario, en la que el principal queda en espera en ese paso hasta que se completa el secundario.

start_workflows() ofrece mayor flexibilidad:

• los flujos de trabajo iniciados son flujos independientes de nivel superior
• cada flujo de trabajo iniciado se puede inspeccionar, reintentar o exportar por separado
• el flujo principal decide más adelante si espera a que terminen y cómo hacerlo

Por eso, start_workflows() se adapta mejor a la transferencia asíncrona de tareas de agentes, mientras que run_workflow() sigue siendo más adecuado para la composición anidada dentro de un único árbol de orquestación.

Esperar más adelante

start_workflows() está diseñado para combinarse con Dependencias entre flujos de trabajo:

use Queuety\Enums\WaitMode;

Queuety::workflow( 'review_board' )
    ->start_workflows( 'review_tasks', $review_task_workflow, 'review_workflow_ids' )
    ->wait_for_workflows( 'review_workflow_ids', WaitMode::Any, 'winner' )
    ->then( FinalizeDecisionStep::class )
    ->dispatch( $payload );

• WaitMode::All recopila todos los estados de los procesos secundarios completados
• WaitMode::Any se reanuda cuando termina correctamente el primer flujo de trabajo secundario

Así puede crear una orquestación de planificador y ejecutor sin necesitar un motor DAG completamente editable en tiempo de ejecución.

Alias orientados a agentes

start_agents() es un alias de start_workflows() con valores predeterminados adaptados a los flujos de planificador y ejecutor:

• result_key tiene como valor predeterminado agent_workflow_ids
• payload_key tiene como valor predeterminado agent_task
• name tiene como valor predeterminado start_agents

wait_for_agents() es un alias de wait_for_workflows() con valores predeterminados adaptados a la recopilación de resultados de agentes:

• workflows tiene como valor predeterminado agent_workflow_ids
• result_key tiene como valor predeterminado agent_results
• name tiene como valor predeterminado wait_for_agents

wait_for_agent_group() es un alias de wait_for_workflow_group() con valores predeterminados adaptados a lotes de agentes.

use Queuety\Enums\WaitMode;

$agent_run = Queuety::workflow( 'agent_run' )
    ->then( ResearchTopicStep::class )
    ->then( SummarizeTopicStep::class );

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