Asynchronní předání

Zobrazit jako Markdown Upravit na GitHub Dokumentace na 2cc7b2b

Použijte start_workflows(), když workflow potřebuje předat objevenou práci samostatným workflow nejvyšší úrovně, místo aby vše ponechalo v jediném běhu.

Pokud výslovně modelujete běhy agentů, start_agents() a wait_for_agents() jsou významové aliasy s výchozím nastavením pro agenty.

Rozsáhlejší příklady od začátku do konce najdete v části Orchestrace agentů.

To se hodí pro orchestraci ve stylu agentů:

workflow plánovače za běhu objeví úlohy
každá úloha se stane vlastním trvalým workflow
nadřazené workflow ukládá ID těchto workflow do stavu
nadřazené workflow může později čekat na all, any, quorum nebo pojmenovanou skupinu spuštěných workflow

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

V tomto příkladu:

PlanTopicsStep zapíše pole dat k jednotlivým tématům do $state['topics']
start_workflows() odešle jedno workflow agent_task pro každé téma
ID spuštěných workflow se zapíší do $state['child_workflow_ids']
wait_for_workflow_group() pozastaví běh, dokud spuštěná skupina nesplní zvolenou podmínku dokončení
SynthesizeBriefStep obdrží dokončené stavy podřízených běhů v $state['child_results']

Data podřízených workflow

Každá položka v items_key se stane počátečními daty jednoho spuštěného workflow.

Pokud je položkou pole, jeho veřejné klíče se sloučí s daty podřízeného workflow:

PHP

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

Pokud je položka skalární hodnota, Queuety ji uloží pod klíčem payload_key:

PHP

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

Zdědění stavu nadřazeného workflow

Ve výchozím nastavení start_workflows() sloučí veřejný stav nadřazeného workflow s daty každého podřízeného workflow.

To znamená, že sdílený kontext, jako například brief_id, account_id nebo campaign, je automaticky dostupný v každém spuštěném workflow.

Zdrojové items_key a result_key jsou ze zděděných dat vyloučeny, aby každé podřízené workflow nedostalo celý seznam úloh ani pole ID výsledných workflow.

Nastavte inherit_state: false, když má každé podřízené workflow obdržet pouze data své položky.

Proč nepoužít `run_workflow()`?

run_workflow() vytváří vztah rodič–potomek, ve kterém nadřazené workflow zůstane na daném kroku pozastavené, dokud se podřízené workflow nedokončí.

start_workflows() je volnější:

spuštěná workflow jsou samostatná workflow nejvyšší úrovně
každé spuštěné workflow lze samostatně kontrolovat, opakovat nebo exportovat
nadřazené workflow později rozhodne, zda a jak na ně počká

Díky tomu se start_workflows() lépe hodí pro předávání úloh mezi asynchronními agenty, zatímco run_workflow() je vhodnější pro vnořené skládání v rámci jednoho orchestračního stromu.

Pozdější čekání

start_workflows() je navržené tak, aby se používalo společně s Závislostmi workflow:

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 shromažďuje stavy všech dokončených podřízených workflow
WaitMode::Any pokračuje po dokončení prvního úspěšného podřízeného workflow

Díky tomu můžete koordinovat plánovač a vykonavatele bez potřeby plnohodnotného enginu DAG upravitelného za běhu.

Aliasy vhodné pro agenty

start_agents() je alias pro start_workflows() s výchozím nastavením přizpůsobeným tokům plánovače a vykonavatele:

result_key má jako výchozí hodnotu agent_workflow_ids
payload_key má jako výchozí hodnotu agent_task
name má jako výchozí hodnotu start_agents

wait_for_agents() je alias pro wait_for_workflows() s výchozím nastavením přizpůsobeným shromažďování výsledků agentů:

workflows má jako výchozí hodnotu agent_workflow_ids
result_key má jako výchozí hodnotu agent_results
name má jako výchozí hodnotu wait_for_agents

wait_for_agent_group() je alias pro wait_for_workflow_group() s výchozím nastavením přizpůsobeným dávkám agentů.

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