Перейти к содержимому

Асинхронная передача задач

Просмотреть как Markdown Редактировать на GitHub Документация по 2cc7b2b

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

Если вы явно моделируете запуски агентов, start_agents() и wait_for_agents() — семантические псевдонимы со значениями по умолчанию для сценариев с агентами.

Примеры полного цикла см. в разделе Оркестрация агентов.

Это полезно для оркестрации в стиле агентов:

  • рабочий процесс планировщика обнаруживает задачи во время выполнения
  • каждая задача становится отдельным надёжным рабочим процессом
  • родительский процесс сохраняет идентификаторы этих рабочих процессов в состоянии
  • родительский процесс может позже дождаться all, any, quorum или именованной запущенной группы

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

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

  1. PlanTopicsStep записывает массив полезных нагрузок тем в $state['topics']
  2. start_workflows() запускает отдельный рабочий процесс agent_task для каждой темы
  3. Идентификаторы запущенных рабочих процессов записываются в $state['child_workflow_ids']
  4. wait_for_workflow_group() приостанавливается, пока запущенная группа не выполнит выбранное условие завершения
  5. SynthesizeBriefStep получает завершённые состояния дочерних процессов в $state['child_results']

Полезные нагрузки дочерних процессов

Каждый элемент в items_key становится исходной полезной нагрузкой для одного запущенного рабочего процесса.

Если элемент является массивом, его публичные ключи объединяются с полезной нагрузкой дочернего процесса:

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

Если элемент является скалярным значением, Queuety сохраняет его в payload_key:

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

Наследование состояния родительского процесса

По умолчанию start_workflows() объединяет публичное состояние родительского рабочего процесса с полезной нагрузкой каждого дочернего процесса.

Это означает, что общий контекст, например brief_id, account_id или campaign, автоматически доступен в каждом запущенном рабочем процессе.

Исходные данные items_key и result_key исключаются из наследования, чтобы каждый дочерний процесс не получал весь список задач или итоговый массив идентификаторов рабочих процессов.

Задайте inherit_state: false, если каждый дочерний процесс должен получать только полезную нагрузку своего элемента.

Почему не run_workflow()?

run_workflow() создаёт связь родительского и дочернего процессов: родитель приостанавливается на этом шаге до завершения дочернего процесса.

start_workflows() действует более независимо:

  • запущенные рабочие процессы являются независимыми процессами верхнего уровня
  • каждый из них можно отдельно проверять, повторять или экспортировать
  • родительский процесс позднее решает, нужно ли ждать их и каким образом

Поэтому start_workflows() лучше подходит для асинхронной передачи задач агентам, а run_workflow() — для вложенной композиции в одном дереве оркестрации.

Ожидание на более позднем этапе

start_workflows() предназначен для совместного использования с Зависимостями рабочего процесса:

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 собирает все завершённые состояния дочерних процессов
  • 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() со значениями по умолчанию для пакетов задач агентов.

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