Asynkrona överlämningar

Använd start_workflows() när ett arbetsflöde behöver lämna över upptäckt arbete till fristående arbetsflöden på toppnivå i stället för att hålla allt i en och samma körning.

Om du uttryckligen modellerar agentkörningar är start_agents() och wait_for_agents() semantiska alias med standardinställningar anpassade för agenter.

Se Agentorkestrering för fler exempel från början till slut.

Det här är användbart för agentliknande orkestrering:

• ett planeringsarbetsflöde upptäcker uppgifter under körning
• varje uppgift blir ett eget beständigt arbetsflöde
• det överordnade arbetsflödet sparar arbetsflödes-ID:na i tillståndet
• det överordnade arbetsflödet kan senare vänta på all, any, quorum eller en namngiven startad grupp

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

I det exemplet:

1. PlanTopicsStep skriver en array med nyttolaster för ämnen till $state['topics']
2. start_workflows() startar ett agent_task-arbetsflöde per ämne
3. ID:na för de startade arbetsflödena skrivs till $state['child_workflow_ids']
4. wait_for_workflow_group() pausar tills den startade gruppen uppfyller det valda slutföringsvillkoret
5. SynthesizeBriefStep tar emot de slutförda tillstånden från underordnade arbetsflöden i $state['child_results']

Nyttolaster för underordnade arbetsflöden

Varje objekt i items_key blir den inledande nyttolasten för ett startat arbetsflöde.

Om objektet är en array slås dess offentliga nycklar samman med nyttolasten för det underordnade arbetsflödet:

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

Om objektet är ett skalärvärde sparar Queuety det under payload_key:

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

Ärva tillstånd från överordnat arbetsflöde

Som standard slår start_workflows() samman det överordnade arbetsflödets offentliga tillstånd med varje nyttolast för underordnade arbetsflöden.

Det innebär att delad kontext, som brief_id, account_id eller campaign, automatiskt är tillgänglig i varje startat arbetsflöde.

Källan items_key och result_key undantas från arvet, så att varje underordnat arbetsflöde inte får hela uppgiftslistan eller den slutliga arrayen med arbetsflödes-ID:n.

Ange inherit_state: false när varje underordnat arbetsflöde bara ska få sin egen objektpayload.

Varför inte run_workflow()?

run_workflow() skapar en relation mellan överordnat och underordnat arbetsflöde, där det överordnade arbetsflödet stannar vid steget tills det underordnade är slutfört.

start_workflows() är mindre strikt:

• de startade arbetsflödena är fristående arbetsflöden på toppnivå
• varje startat arbetsflöde kan granskas, göras om eller exporteras separat
• det överordnade arbetsflödet avgör senare om och hur det ska vänta på dem

Det gör start_workflows() lämpligare för asynkron överlämning av agentuppgifter, medan run_workflow() passar bättre för nästlad sammansättning inom ett orkestreringsträd.

Vänta senare

start_workflows() är utformat för att användas tillsammans med Arbetsflödesberoenden:

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 samlar in alla slutförda tillstånd från underordnade arbetsflöden
• WaitMode::Any återupptas när det första underordnade arbetsflödet har slutförts utan fel

Det här ger dig orkestrering med planerare och utförare utan att du behöver en fullständig DAG-motor som kan redigeras under körning.

Alias anpassade för agenter

start_agents() är ett alias för start_workflows() med standardinställningar anpassade för flöden med planerare och utförare:

• result_key har standardvärdet agent_workflow_ids
• payload_key har standardvärdet agent_task
• name har standardvärdet start_agents

wait_for_agents() är ett alias för wait_for_workflows() med standardinställningar anpassade för att samla in agentresultat:

• workflows har standardvärdet agent_workflow_ids
• result_key har standardvärdet agent_results
• name har standardvärdet wait_for_agents

wait_for_agent_group() är ett alias för wait_for_workflow_group() med standardinställningar anpassade för agentbatchar.

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