Transferts asynchrones

Utilisez start_workflows() lorsqu’un workflow doit confier le travail découvert à des workflows indépendants de premier niveau au lieu de tout conserver dans une seule exécution.

Si vous modélisez explicitement des exécutions d’agents, start_agents() et wait_for_agents() sont des alias sémantiques avec des valeurs par défaut adaptées aux agents.

Pour des exemples plus complets de bout en bout, consultez Orchestration d’agents.

Cette approche est utile pour l’orchestration de type agent :

• un workflow planificateur découvre des tâches à l’exécution
• chaque tâche devient un workflow durable
• le parent enregistre les ID de ces workflows dans l’état
• le parent peut ensuite attendre all, any, quorum ou un groupe nommé démarré

La méthode ->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 ] );

Dans cet exemple :

1. PlanTopicsStep écrit un tableau de charges utiles de sujets dans $state['topics']
2. start_workflows() lance un workflow agent_task par sujet
3. les ID des workflows démarrés sont écrits dans $state['child_workflow_ids']
4. wait_for_workflow_group() met le processus en pause jusqu’à ce que le groupe démarré satisfasse la condition d’achèvement choisie
5. SynthesizeBriefStep reçoit les états des enfants terminés sous $state['child_results']

Charges utiles des enfants

Chaque élément de items_key devient la charge utile initiale d’un workflow démarré.

Si l’élément est un tableau, ses clés publiques sont fusionnées dans la charge utile de l’enfant :

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

Si l’élément est scalaire, Queuety le stocke sous payload_key :

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

Hériter de l’état du parent

Par défaut, start_workflows() fusionne l’état public du workflow parent dans la charge utile de chaque enfant.

Le contexte partagé, comme brief_id, account_id ou campaign, est ainsi automatiquement disponible dans chaque workflow démarré.

La source items_key et le result_key sont exclus de l’héritage afin que chaque enfant ne reçoive ni toute la liste de tâches ni le tableau final d’ID de workflows.

Définissez inherit_state: false lorsque chaque enfant ne doit recevoir que la charge utile de son élément.

Pourquoi ne pas utiliser run_workflow() ?

run_workflow() crée une relation parent-enfant dans laquelle le parent reste en attente à cette étape jusqu’à l’achèvement de l’enfant.

start_workflows() est plus souple :

• les workflows démarrés sont des workflows indépendants de premier niveau
• chaque workflow démarré peut être inspecté, relancé ou exporté séparément
• le parent décide plus tard s’il doit les attendre et de quelle manière

start_workflows() convient donc mieux au transfert asynchrone de tâches d’agents, tandis que run_workflow() convient mieux à la composition imbriquée au sein d’un même arbre d’orchestration.

Attendre ultérieurement

start_workflows() est conçu pour être associé aux dépendances de workflow :

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 rassemble tous les états des enfants terminés
• WaitMode::Any reprend à la première exécution réussie d’un workflow enfant

Vous obtenez ainsi une orchestration planificateur/exécuteur sans avoir besoin d’un moteur DAG entièrement modifiable à l’exécution.

Alias adaptés aux agents

start_agents() est un alias de start_workflows() avec des valeurs par défaut adaptées aux flux planificateur/exécuteur :

• result_key prend agent_workflow_ids comme valeur par défaut
• payload_key prend agent_task comme valeur par défaut
• name prend start_agents comme valeur par défaut

wait_for_agents() est un alias de wait_for_workflows() avec des valeurs par défaut adaptées à la collecte des résultats d’agents :

• workflows prend agent_workflow_ids comme valeur par défaut
• result_key prend agent_results comme valeur par défaut
• name prend wait_for_agents comme valeur par défaut

wait_for_agent_group() est un alias de wait_for_workflow_group() avec des valeurs par défaut adaptées aux groupes d’agents.

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