Asynchrone Übergaben

Verwende start_workflows(), wenn ein Workflow erkannte Aufgaben an unabhängige Workflows auf oberster Ebene übergeben soll, statt alles in einem Lauf zu behalten.

Wenn du ausdrücklich Agentenläufe modellierst, sind start_agents() und wait_for_agents() semantische Aliasse mit agentenspezifischen Standardwerten.

Weitere umfassende Beispiele findest du unter Agenten-Orchestrierung.

Das ist nützlich für eine Orchestrierung im Stil von Agenten:

• Ein Planer-Workflow ermittelt Aufgaben zur Laufzeit
• Jede Aufgabe wird zu einem eigenen dauerhaften Workflow
• Der übergeordnete Workflow speichert diese Workflow-IDs im Zustand
• Der übergeordnete Workflow kann später auf all, any, quorum oder eine benannte gestartete Gruppe warten

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

In diesem Beispiel gilt:

1. PlanTopicsStep schreibt ein Array von Themen-Nutzlasten nach $state['topics']
2. start_workflows() startet für jedes Thema einen agent_task-Workflow
3. Die IDs der gestarteten Workflows werden nach $state['child_workflow_ids'] geschrieben
4. wait_for_workflow_group() pausiert, bis die gestartete Gruppe die gewählte Abschlussbedingung erfüllt
5. SynthesizeBriefStep erhält die abgeschlossenen Zustände der untergeordneten Workflows unter $state['child_results']

Nutzlasten untergeordneter Workflows

Jedes Element in items_key wird zur initialen Nutzlast eines gestarteten Workflows.

Wenn das Element ein Array ist, werden seine öffentlichen Schlüssel mit der Nutzlast des untergeordneten Workflows zusammengeführt:

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

Wenn das Element ein Skalar ist, speichert Queuety es unter payload_key:

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

Zustand des übergeordneten Workflows übernehmen

Standardmäßig führt start_workflows() den öffentlichen Zustand des übergeordneten Workflows mit jeder Nutzlast der untergeordneten Workflows zusammen.

Das bedeutet, dass gemeinsamer Kontext wie brief_id, account_id oder campaign automatisch in jedem gestarteten Workflow verfügbar ist.

Die Quelle items_key und das result_key werden von der Übernahme ausgeschlossen, damit jeder untergeordnete Workflow nicht die gesamte Aufgabenliste oder das endgültige Workflow-ID-Array erhält.

Setze inherit_state: false, wenn jeder untergeordnete Workflow nur seine Element-Nutzlast erhalten soll.

Warum nicht run_workflow()?

run_workflow() erstellt eine Eltern-Kind-Beziehung, bei der der übergeordnete Workflow bei diesem Schritt wartet, bis der untergeordnete Workflow abgeschlossen ist.

start_workflows() ist weniger eng gekoppelt:

• Die gestarteten Workflows sind unabhängige Workflows auf oberster Ebene
• Jeder gestartete Workflow kann separat geprüft, wiederholt oder exportiert werden
• Der übergeordnete Workflow entscheidet später, ob und wie er auf sie wartet

Damit eignet sich start_workflows() besser für die asynchrone Übergabe von Agentenaufgaben, während run_workflow() besser für verschachtelte Komposition innerhalb eines Orchestrierungsbaums geeignet ist.

Später warten

start_workflows() ist für die Kombination mit Workflow-Abhängigkeiten vorgesehen:

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 sammelt alle abgeschlossenen Zustände der untergeordneten Workflows
• WaitMode::Any wird fortgesetzt, sobald der erste untergeordnete Workflow erfolgreich abgeschlossen ist

So erhältst du eine Planer-Ausführer-Orchestrierung, ohne eine vollständig zur Laufzeit bearbeitbare DAG-Engine zu benötigen.

Aliasse für Agenten

start_agents() ist ein Alias für start_workflows() mit Standardwerten, die auf Planer-Ausführer-Abläufe abgestimmt sind:

• result_key ist standardmäßig agent_workflow_ids
• payload_key ist standardmäßig agent_task
• name ist standardmäßig start_agents

wait_for_agents() ist ein Alias für wait_for_workflows() mit Standardwerten, die auf das Sammeln von Agentenergebnissen abgestimmt sind:

• workflows ist standardmäßig agent_workflow_ids
• result_key ist standardmäßig agent_results
• name ist standardmäßig wait_for_agents

wait_for_agent_group() ist ein Alias für wait_for_workflow_group() mit Standardwerten, die auf Agenten-Batches abgestimmt sind.

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