Асинхронно предаване

Използвайте start_workflows(), когато работният поток трябва да предаде откритата работа на независими работни потоци от най-високо ниво, вместо да задържа всичко в рамките на едно изпълнение.

Ако изрично моделирате изпълнения на агенти, start_agents() и wait_for_agents() са семантични псевдоними с настройки по подразбиране, ориентирани към агенти.

За по-обширни примери от начало до край вижте Оркестриране на агенти.

Това е полезно за оркестриране от агентски тип:

• работният поток на планировчика открива задачи по време на изпълнение
• всяка задача става отделен устойчив работен поток
• родителското изпълнение съхранява ID-тата на тези работни потоци в състоянието
• по-късно родителското изпълнение може да изчака all, any, quorum или именувана група стартирали изпълнения

Методът ->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 ] );

В този пример:

1. PlanTopicsStep записва масив от данни за темите в $state['topics']
2. start_workflows() изпраща по един работен поток agent_task за всяка тема
3. ID-тата на стартираните работни потоци се записват в $state['child_workflow_ids']
4. wait_for_workflow_group() поставя изпълнението на пауза, докато стартираната група не изпълни избраното условие за завършване
5. SynthesizeBriefStep получава завършените състояния на дъщерните изпълнения в $state['child_results']

Данни на дъщерните изпълнения

Всеки елемент в items_key става началните данни за един стартиран работен поток.

Ако елементът е масив, публичните му ключове се обединяват с данните на дъщерното изпълнение:

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

Ако елементът е скаларна стойност, Queuety я съхранява под ключ payload_key:

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

Наследяване на родителското състояние

По подразбиране start_workflows() обединява публичното състояние на родителския работен поток с данните на всяко дъщерно изпълнение.

Това означава, че споделен контекст, като brief_id, account_id или campaign, е автоматично наличен във всеки стартиран работен поток.

Изходният items_key и result_key се изключват от наследяването, така че всяко дъщерно изпълнение да не получава целия списък със задачи или крайния масив с ID-тата на работните потоци.

Задайте inherit_state: false, когато всяко дъщерно изпълнение трябва да получи само данните за своя елемент.

Защо не run_workflow()?

run_workflow() създава връзка родител-дете, при която родителското изпълнение спира на тази стъпка, докато дъщерното не завърши.

start_workflows() е по-свободно обвързан подход:

• стартираните работни потоци са независими изпълнения от най-високо ниво
• всяко стартирано изпълнение може да бъде проверено, повторено или експортирано самостоятелно
• родителското изпълнение решава по-късно дали и как да ги изчака

Това прави start_workflows() по-подходящ за предаване на задачи между асинхронни агенти, докато run_workflow() остава по-подходящ за вложено композиране в рамките на едно дърво за оркестрация.

Изчакване на по-късен етап

start_workflows() е създаден да работи съвместно с Зависимости между работни потоци:

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 събира състоянията на всички завършили дъщерни работни потоци
• WaitMode::Any продължава при първия успешно завършил дъщерен работен поток

Така можете да организирате изпълнението между планировчик и изпълнител, без да ви е необходим пълноценен механизъм за DAG, който може да се редактира по време на изпълнение.

Псевдоними за агенти

start_agents() е псевдоним на start_workflows() с настройки по подразбиране за потоци между планировчик и изпълнител:

• result_key е agent_workflow_ids по подразбиране
• payload_key е agent_task по подразбиране
• name е start_agents по подразбиране

wait_for_agents() е псевдоним на wait_for_workflows() с настройки по подразбиране за събиране на резултати от агенти:

• workflows е agent_workflow_ids по подразбиране
• result_key е agent_results по подразбиране
• name е wait_for_agents по подразбиране

wait_for_agent_group() е псевдоним на wait_for_workflow_group() с настройки по подразбиране за партиди от агенти.

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