Transferuri asincrone

Folosiți start_workflows() când un flux de lucru trebuie să transfere activitatea descoperită către fluxuri independente de nivel superior, în loc să păstreze totul într-o singură execuție.

Dacă modelați în mod explicit execuțiile agenților, start_agents() și wait_for_agents() sunt aliasuri semantice cu valori implicite adaptate agenților.

Pentru exemple mai ample, de la început până la sfârșit, consultați Orchestrarea agenților.

Acest lucru este util pentru orchestrarea de tip agent:

• un flux de lucru al planificatorului descoperă sarcini în timpul execuției
• fiecare sarcină devine propriul flux de lucru durabil
• fluxul părinte stochează ID-urile acestor fluxuri în stare
• ulterior, fluxul părinte poate aștepta all, any, quorum sau un grup denumit, deja pornit

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

În acest exemplu:

1. PlanTopicsStep scrie o listă de date despre subiecte în $state['topics']
2. start_workflows() pornește câte un flux de lucru agent_task pentru fiecare subiect
3. ID-urile fluxurilor de lucru pornite sunt scrise în $state['child_workflow_ids']
4. wait_for_workflow_group() se pune pe pauză până când grupul pornit îndeplinește condiția de finalizare aleasă
5. SynthesizeBriefStep primește stările finalizate ale copiilor în $state['child_results']

Datele copiilor

Fiecare element din items_key devine setul inițial de date pentru un flux de lucru pornit.

Dacă elementul este o listă, cheile sale publice sunt incluse în datele copilului:

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

Dacă elementul este scalar, Queuety îl stochează în payload_key:

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

Moștenirea stării părintelui

În mod implicit, start_workflows() include starea publică a fluxului de lucru părinte în datele fiecărui copil.

Astfel, contextul partajat, precum brief_id, account_id sau campaign, este disponibil automat în fiecare flux de lucru pornit.

Sursa items_key și result_key sunt excluse de la moștenire, astfel încât fiecare copil să nu primească întreaga listă de sarcini sau lista finală de ID-uri ale fluxurilor de lucru.

Setați inherit_state: false când fiecare copil trebuie să primească doar datele elementului său.

De ce nu run_workflow()?

run_workflow() creează o relație părinte-copil în care părintele rămâne în așteptare la acel pas până când copilul se finalizează.

start_workflows() oferă mai multă libertate:

• fluxurile de lucru pornite sunt independente și de nivel superior
• fiecare flux de lucru pornit poate fi inspectat, reîncercat sau exportat separat
• părintele decide ulterior dacă și cum să le aștepte

Astfel, start_workflows() se potrivește mai bine pentru transferul asincron al sarcinilor către agenți, în timp ce run_workflow() este mai potrivit pentru compoziția imbricată într-un singur arbore de orchestrare.

Așteptarea ulterioară

start_workflows() este conceput să fie folosit împreună cu Dependențe între fluxuri de lucru:

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 colectează toate stările finalizate ale copiilor
• WaitMode::Any se reia la primul flux de lucru copil finalizat cu succes

Astfel, puteți orchestra un planificator și un executor fără să aveți nevoie de un motor DAG complet, editabil în timpul execuției.

Aliasuri adaptate agenților

start_agents() este un alias pentru start_workflows(), cu valori implicite adaptate fluxurilor planificator/executor:

• valoarea implicită pentru result_key este agent_workflow_ids
• valoarea implicită pentru payload_key este agent_task
• valoarea implicită pentru name este start_agents

wait_for_agents() este un alias pentru wait_for_workflows(), cu valori implicite adaptate colectării rezultatelor agenților:

• valoarea implicită pentru workflows este agent_workflow_ids
• valoarea implicită pentru result_key este agent_results
• valoarea implicită pentru name este wait_for_agents

wait_for_agent_group() este un alias pentru wait_for_workflow_group(), cu valori implicite adaptate loturilor de agenți.

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