Asinchroninis perdavimas

Peržiūrėti kaip Markdown Redaguoti per GitHub Dokumentacija: 2cc7b2b

Naudokite start_workflows(), kai darbo eiga turi perduoti aptiktą darbą nepriklausomoms aukščiausio lygio darbo eigoms, o ne laikyti viską viename vykdyme.

Jei aiškiai modeliuojate agentų vykdymus, start_agents() ir wait_for_agents() yra semantiniai sinonimai su agentams pritaikytomis numatytosiomis reikšmėmis.

Išsamesnių pavyzdžių nuo pradžios iki pabaigos ieškokite puslapyje Agentų orkestravimas.

Tai naudinga agentų tipo orkestravimui:

planuotojo darbo eiga vykdymo metu aptinka užduotis
kiekviena užduotis tampa atskira patvaria darbo eiga
pirminė eiga išsaugo šių darbo eigų ID būsenoje
vėliau pirminė eiga gali laukti all, any, quorum arba įvardytos pradėtos grupės

Metodas `->start_workflows()`

PHP

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

Šiame pavyzdyje:

PlanTopicsStep įrašo temų duomenų rinkinių masyvą į $state['topics']
start_workflows() kiekvienai temai paleidžia po vieną agent_task darbo eigą
Pradėtų darbo eigų ID įrašomi į $state['child_workflow_ids']
wait_for_workflow_group() pristabdo vykdymą, kol pradėta grupė įvykdo pasirinktą užbaigimo sąlygą
SynthesizeBriefStep gauna užbaigtas antrines būsenas, pateiktas per $state['child_results']

Antriniai duomenų rinkiniai

Kiekvienas items_key elementas tampa pradine vienos paleistos darbo eigos duomenų apkrova.

Jei elementas yra masyvas, jo viešieji raktai sujungiami su antrinės darbo eigos duomenų apkrova:

PHP

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

Jei elementas yra skaliaras, Queuety išsaugo jį per payload_key:

PHP

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

Pirminės būsenos paveldėjimas

Pagal numatytąsias nuostatas start_workflows() sujungia pirminės darbo eigos viešąją būseną su kiekvienos antrinės darbo eigos duomenų apkrova.

Tai reiškia, kad bendras kontekstas, pavyzdžiui, brief_id, account_id arba campaign, automatiškai pasiekiamas kiekvienoje pradėtoje darbo eigoje.

Šaltinis items_key ir result_key neįtraukiami į paveldėjimą, kad kiekviena antrinė eiga negautų viso užduočių sąrašo ar galutinio darbo eigų ID masyvo.

Nustatykite inherit_state: false, kai kiekviena antrinė eiga turi gauti tik savo elemento duomenų apkrovą.

Kodėl ne `run_workflow()`?

run_workflow() sukuria pirminės ir antrinės eigos ryšį, kuriame pirminė eiga sustoja ties tuo žingsniu, kol baigiama antrinė eiga.

start_workflows() suteikia daugiau laisvės:

pradėtos darbo eigos yra nepriklausomos aukščiausio lygio darbo eigos
kiekvieną pradėtą darbo eigą galima atskirai apžiūrėti, pakartoti arba eksportuoti
pirminė eiga vėliau nusprendžia, ar ir kaip jų laukti

Todėl start_workflows() geriau tinka asinchroniniam agentų užduočių perdavimui, o run_workflow() labiau tinka įdėtinei kompozicijai viename orkestravimo medyje.

Laukimas vėliau

start_workflows() skirtas naudoti kartu su Darbo eigos priklausomybėmis:

PHP

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 surenka visas užbaigtas antrines būsenas
WaitMode::Any tęsiamas užbaigus pirmą sėkmingą antrinę darbo eigą

Taip galima orkestruoti planuotoją ir vykdytoją be visaverčio vykdymo metu redaguojamo DAG variklio.

Agentams pritaikyti sinonimai

start_agents() yra start_workflows() sinonimas su planuotojo / vykdytojo srautams pritaikytomis numatytosiomis reikšmėmis:

result_key numatytoji reikšmė yra agent_workflow_ids
payload_key numatytoji reikšmė yra agent_task
name numatytoji reikšmė yra start_agents

wait_for_agents() yra wait_for_workflows() sinonimas su agentų rezultatams rinkti pritaikytomis numatytosiomis reikšmėmis:

workflows numatytoji reikšmė yra agent_workflow_ids
result_key numatytoji reikšmė yra agent_results
name numatytoji reikšmė yra wait_for_agents

wait_for_agent_group() yra wait_for_workflow_group() sinonimas su agentų paketams pritaikytomis numatytosiomis reikšmėmis.

PHP

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