Asynchrónne odovzdávanie

Použite start_workflows(), keď pracovný postup potrebuje odovzdať objavenú prácu samostatným pracovným postupom najvyššej úrovne namiesto toho, aby všetko ponechal v jednom behu.

Ak výslovne modelujete behy agentov, start_agents() a wait_for_agents() sú sémantické aliasy s predvolenými nastaveniami prispôsobenými agentom.

Rozsiahlejšie príklady od začiatku do konca nájdete v téme Orchestrácia agentov.

Je to užitočné pri orchestrácii v štýle agentov:

• pracovný postup plánovača objaví úlohy počas behu
• každá úloha sa stane vlastným trvalým pracovným postupom
• nadradený postup uloží ID týchto pracovných postupov do stavu
• nadradený postup môže neskôr čakať na all, any, quorum alebo na pomenovanú spustenú skupinu

Metóda ->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 ] );

V tomto príklade:

1. PlanTopicsStep zapíše pole údajov tém do $state['topics']
2. start_workflows() spustí jeden pracovný postup agent_task pre každú tému
3. ID spustených pracovných postupov sa zapíšu do $state['child_workflow_ids']
4. wait_for_workflow_group() pozastaví beh, kým spustená skupina nesplní zvolenú podmienku dokončenia
5. SynthesizeBriefStep prijme dokončené stavy podriadených postupov pod $state['child_results']

Údaje podriadených postupov

Každá položka v items_key sa stane počiatočnými údajmi pre jeden spustený pracovný postup.

Ak je položka pole, jej verejné kľúče sa zlúčia s údajmi podriadeného postupu:

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

Ak je položka skalárna hodnota, Queuety ju uloží pod payload_key:

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

Dedenie stavu nadradeného postupu

Predvolene start_workflows() zlúči verejný stav nadradeného pracovného postupu do údajov každého podriadeného postupu.

To znamená, že zdieľaný kontext, napríklad brief_id, account_id alebo campaign, je automaticky dostupný v každom spustenom pracovnom postupe.

Zdroj items_key a result_key sú z dedenia vylúčené, aby každý podriadený postup nedostal celý zoznam úloh ani pole ID konečných pracovných postupov.

Nastavte inherit_state: false, keď má každý podriadený postup dostať iba údaje svojej položky.

Prečo nie run_workflow()?

run_workflow() vytvára vzťah nadradeného a podriadeného postupu, v ktorom sa nadradený postup v danom kroku pozastaví až do dokončenia podriadeného.

start_workflows() je voľnejšie prepojené:

• spustené pracovné postupy sú samostatné postupy najvyššej úrovne
• každý spustený postup možno samostatne kontrolovať, opakovať alebo exportovať
• nadradený postup sa neskôr rozhodne, či a ako na ne počká

Vďaka tomu sa start_workflows() lepšie hodí na asynchrónne odovzdávanie úloh agentom, zatiaľ čo run_workflow() je vhodnejšie na vnorené skladanie v rámci jedného stromu orchestrácie.

Čakanie neskôr

start_workflows() je navrhnuté tak, aby sa používalo spolu so závislosťami pracovných postupov Závislosti pracovných postupov:

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 zhromaždí všetky dokončené stavy podriadených postupov
• WaitMode::Any obnoví beh po prvom úspešnom podriadenom pracovnom postupe

Tak získate orchestráciu plánovač/vykonávateľ bez potreby DAG enginu, ktorého štruktúru možno meniť počas behu.

Aliasy vhodné pre agentov

start_agents() je aliasom pre start_workflows() s predvolenými nastaveniami prispôsobenými tokom typu plánovač/vykonávateľ:

• result_key má predvolenú hodnotu agent_workflow_ids
• payload_key má predvolenú hodnotu agent_task
• name má predvolenú hodnotu start_agents

wait_for_agents() je aliasom pre wait_for_workflows() s predvolenými nastaveniami prispôsobenými zhromažďovaniu výsledkov agentov:

• workflows má predvolenú hodnotu agent_workflow_ids
• result_key má predvolenú hodnotu agent_results
• name má predvolenú hodnotu wait_for_agents

wait_for_agent_group() je aliasom pre wait_for_workflow_group() s predvolenými nastaveniami prispôsobenými dávkam agentov.

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