Passaggi asincroni

Usate start_workflows() quando un workflow deve affidare il lavoro individuato a workflow indipendenti di primo livello, invece di mantenere tutto in un'unica esecuzione.

Se state modellando esplicitamente esecuzioni di agenti, start_agents() e wait_for_agents() sono alias semantici con impostazioni predefinite orientate agli agenti.

Per esempi più ampi che coprono l'intero processo, consultate Orchestrazione degli agenti.

È utile per l'orchestrazione in stile agente:

• un workflow di pianificazione individua attività durante l'esecuzione
• ogni attività diventa un workflow persistente indipendente
• il processo padre salva gli ID dei workflow nello stato
• in seguito, il processo padre può attendere all, any, quorum o un gruppo avviato con un nome

Il metodo ->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 ] );

In questo esempio:

1. PlanTopicsStep scrive un array di payload tematici in $state['topics']
2. start_workflows() avvia un workflow agent_task per ogni argomento
3. gli ID dei workflow avviati vengono scritti in $state['child_workflow_ids']
4. wait_for_workflow_group() mette in pausa l'esecuzione finché il gruppo avviato non soddisfa la condizione di completamento scelta
5. SynthesizeBriefStep riceve gli stati completati dei processi figli in $state['child_results']

Payload dei processi figli

Ogni elemento in items_key diventa il payload iniziale di un workflow avviato.

Se l'elemento è un array, le sue chiavi pubbliche vengono unite al payload del processo figlio:

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

Se l'elemento è scalare, Queuety lo salva in payload_key:

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

Eredità dello stato del processo padre

Per impostazione predefinita, start_workflows() unisce lo stato pubblico del workflow padre a ogni payload dei processi figli.

Ciò rende automaticamente disponibile in ogni workflow avviato il contesto condiviso, come brief_id, account_id o campaign.

L'origine items_key e result_key sono escluse dall'ereditarietà, così ciascun processo figlio non riceve l'intero elenco di attività né l'array finale degli ID dei workflow.

Impostate inherit_state: false quando ogni processo figlio deve ricevere solo il proprio payload.

Perché non run_workflow()?

run_workflow() crea una relazione padre-figlio in cui il processo padre resta in attesa su quel passaggio finché il processo figlio non termina.

start_workflows() è più flessibile:

• i workflow avviati sono workflow indipendenti di primo livello
• ciascun workflow avviato può essere ispezionato, ripetuto o esportato autonomamente
• in seguito, il processo padre decide se e come attenderli

Per questo start_workflows() è più adatto al passaggio asincrono di attività tra agenti, mentre run_workflow() è più adatto alla composizione annidata all'interno di un unico albero di orchestrazione.

Attesa successiva

start_workflows() è progettato per essere usato insieme a Dipendenze dei workflow:

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 raccoglie tutti gli stati completati dei processi figli
• WaitMode::Any riprende l'esecuzione al completamento del primo workflow figlio riuscito

In questo modo potete realizzare un'orchestrazione con pianificazione ed esecuzione senza dover creare un motore DAG completamente modificabile durante l'esecuzione.

Alias orientati agli agenti

start_agents() è un alias di start_workflows() con impostazioni predefinite adatte ai flussi di pianificazione ed esecuzione:

• result_key usa agent_workflow_ids come valore predefinito
• payload_key usa agent_task come valore predefinito
• name usa start_agents come valore predefinito

wait_for_agents() è un alias di wait_for_workflows() con impostazioni predefinite adatte alla raccolta dei risultati degli agenti:

• workflows usa agent_workflow_ids come valore predefinito
• result_key usa agent_results come valore predefinito
• name usa wait_for_agents come valore predefinito

wait_for_agent_group() è un alias di wait_for_workflow_group() con impostazioni predefinite adatte ai batch di agenti.

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 ] );