Asinhrona nodošana

Izmantojiet start_workflows(), ja darbplūsmai atrastais darbs jānodod neatkarīgām augstākā līmeņa darbplūsmām, nevis jāatstāj viss vienā izpildē.

Ja īpaši modelējat aģentu izpildes, start_agents() un wait_for_agents() ir semantiski aizstājvārdi ar aģentiem pielāgotiem noklusējuma iestatījumiem.

Lielākus piemērus no sākuma līdz beigām skatiet sadaļā Aģentu orķestrēšana.

Tas noder aģentiem paredzētā orķestrēšanā:

• plānotāja darbplūsma izpildes laikā atklāj uzdevumus
• katrs uzdevums kļūst par atsevišķu noturīgu darbplūsmu
• vecākdarbplūsma saglabā šo darbplūsmu ID stāvoklī
• vecākdarbplūsma vēlāk var gaidīt all, any, quorum vai nosauktu uzsākto grupu

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

Šajā piemērā:

1. PlanTopicsStep ieraksta tēmu datu kopu masīvu $state['topics']
2. start_workflows() palaiž vienu agent_task darbplūsmu katrai tēmai
3. Uzsākto darbplūsmu ID tiek ierakstīti $state['child_workflow_ids']
4. wait_for_workflow_group() aptur darbību, līdz uzsāktā grupa atbilst izvēlētajam pabeigšanas nosacījumam
5. SynthesizeBriefStep saņem pabeigtos pakārtoto darbplūsmu stāvokļus zem $state['child_results']

Pakārtoto darbplūsmu dati

Katrs vienums items_key kļūst par sākotnējo datu kopu vienai uzsāktajai darbplūsmai.

Ja vienums ir masīvs, tā publiskās atslēgas tiek apvienotas ar pakārtotās darbplūsmas datu kopu:

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

Ja vienums ir skalārs, Queuety to saglabā zem payload_key:

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

Vecākdarbplūsmas stāvokļa pārmantošana

Pēc noklusējuma start_workflows() apvieno vecākdarbplūsmas publisko stāvokli ar katras pakārtotās darbplūsmas datu kopu.

Tas nozīmē, ka koplietotais konteksts, piemēram, brief_id, account_id vai campaign, ir automātiski pieejams katrā uzsāktajā darbplūsmā.

Avots items_key un result_key tiek izslēgti no pārmantošanas, lai katra pakārtotā darbplūsma nesaņemtu visu uzdevumu sarakstu vai gala darbplūsmu ID masīvu.

Iestatiet inherit_state: false, ja katrai pakārtotajai darbplūsmai jāsaņem tikai tās vienuma dati.

Kāpēc ne run_workflow()?

run_workflow() izveido vecāk- un pakārtotās darbplūsmas attiecības, kurās vecākdarbplūsma gaida šajā darbībā, līdz pakārtotā darbplūsma tiek pabeigta.

start_workflows() ir brīvāks risinājums:

• uzsāktās darbplūsmas ir neatkarīgas augstākā līmeņa darbplūsmas
• katru uzsākto darbplūsmu var pārbaudīt, atkārtot vai eksportēt atsevišķi
• vecākdarbplūsma vēlāk izvēlas, vai un kā tās gaidīt

Tādēļ start_workflows() ir piemērotāks asinhronai aģenta uzdevumu nodošanai, savukārt run_workflow() — ligzdotai kompozīcijai vienā orķestrēšanas kokā.

Gaidīšana vēlāk

start_workflows() ir paredzēts lietošanai kopā ar Darbplūsmu atkarības:

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 apkopo visus pabeigtos pakārtoto darbplūsmu stāvokļus
• WaitMode::Any atsākas pēc pirmās sekmīgās pakārtotās darbplūsmas

Tādējādi var orķestrēt plānotāju un izpildītāju, neizveidojot pilnu izpildlaikā rediģējamu DAG dzinēju.

Aģentiem piemēroti aizstājvārdi

start_agents() ir start_workflows() aizstājvārds ar iestatījumiem, kas pielāgoti plānotāja un izpildītāja plūsmām:

• Pēc noklusējuma result_key ir agent_workflow_ids
• Pēc noklusējuma payload_key ir agent_task
• Pēc noklusējuma name ir start_agents

wait_for_agents() ir wait_for_workflows() aizstājvārds ar iestatījumiem, kas pielāgoti aģentu rezultātu apkopošanai:

• Pēc noklusējuma workflows ir agent_workflow_ids
• Pēc noklusējuma result_key ir agent_results
• Pēc noklusējuma name ir wait_for_agents

wait_for_agent_group() ir wait_for_workflow_group() aizstājvārds ar iestatījumiem, kas pielāgoti aģentu grupām.

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