Asynchrone overdrachten

Bekijken als Markdown Bewerken op GitHub Documentatie op 2cc7b2b

Gebruik start_workflows() wanneer een workflow ontdekt werk moet overdragen aan onafhankelijke workflows op topniveau, in plaats van alles binnen één uitvoering te houden.

Als u agentuitvoeringen expliciet modelleert, zijn start_agents() en wait_for_agents() semantische aliassen met standaardinstellingen die op agents zijn gericht.

Zie voor grotere voorbeelden van begin tot eind Agentorkestratie.

Dit is nuttig voor orkestratie in agentstijl:

een plannerworkflow ontdekt taken tijdens runtime
elke taak wordt een eigen duurzame workflow
de bovenliggende workflow slaat de workflow-ID's op in de status
de bovenliggende workflow kan later wachten op all, any, quorum of een benoemde gestarte groep

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

In dat voorbeeld:

PlanTopicsStep schrijft een array met toppayloads naar $state['topics']
start_workflows() dispatcht één workflow agent_task per onderwerp
de gestarte workflow-ID's worden naar $state['child_workflow_ids'] geschreven
wait_for_workflow_group() pauzeert totdat aan de gekozen voltooiingsvoorwaarde voor die gestarte groep is voldaan
SynthesizeBriefStep ontvangt de voltooide onderliggende toestanden onder $state['child_results']

Payloads van onderliggende workflows

Elk item in items_key wordt de initiële payload voor één gestarte workflow.

Als het item een array is, worden de publieke sleutels samengevoegd met de payload van de onderliggende workflow:

PHP

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

Als het item een scalaire waarde is, slaat Queuety deze op onder payload_key:

PHP

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

Status van de bovenliggende workflow overnemen

Standaard voegt start_workflows() de publieke status van de bovenliggende workflow samen met elke payload van een onderliggende workflow.

Dat betekent dat gedeelde context, zoals brief_id, account_id of campaign, automatisch beschikbaar is in elke gestarte workflow.

De bron items_key en de result_key worden uitgesloten van overname, zodat elke onderliggende workflow niet de volledige takenlijst of de uiteindelijke array met workflow-ID's ontvangt.

Stel inherit_state: false in wanneer elke onderliggende workflow alleen de payload van het eigen item moet ontvangen.

Waarom geen `run_workflow()`?

run_workflow() maakt een relatie tussen bovenliggende en onderliggende workflows, waarbij de bovenliggende workflow bij die stap wacht totdat de onderliggende workflow is voltooid.

start_workflows() is minder strikt:

de gestarte workflows zijn onafhankelijke workflows op topniveau
elke gestarte workflow kan afzonderlijk worden geïnspecteerd, opnieuw geprobeerd of geëxporteerd
de bovenliggende workflow bepaalt later of en hoe erop wordt gewacht

Daardoor past start_workflows() beter bij de asynchrone overdracht van agenttaken, terwijl run_workflow() geschikter blijft voor geneste samenstelling binnen één orkestratiestructuur.

Later wachten

start_workflows() is bedoeld voor gebruik in combinatie met Workflowafhankelijkheden:

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 verzamelt alle voltooide onderliggende statussen
WaitMode::Any wordt hervat zodra de eerste onderliggende workflow succesvol is afgerond

Zo krijgt u planner-/uitvoerderorkestratie zonder een volledige, tijdens runtime bewerkbare DAG-engine nodig te hebben.

Aliassen voor agents

start_agents() is een alias voor start_workflows() met standaardinstellingen voor planner-/uitvoeringsflows:

result_key heeft standaard de waarde agent_workflow_ids
payload_key heeft standaard de waarde agent_task
name heeft standaard de waarde start_agents

wait_for_agents() is een alias voor wait_for_workflows() met standaardinstellingen voor het verzamelen van agentresultaten:

workflows heeft standaard de waarde agent_workflow_ids
result_key heeft standaard de waarde agent_results
name heeft standaard de waarde wait_for_agents

wait_for_agent_group() is een alias voor wait_for_workflow_group() met standaardinstellingen voor agentbatches.

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