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

Переглянути як 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 ] );

У цьому прикладі:

PlanTopicsStep записує масив даних тем до $state['topics']
start_workflows() запускає окремий робочий процес agent_task для кожної теми
ідентифікатори запущених процесів записуються до $state['child_workflow_ids']
wait_for_workflow_group() призупиняє виконання, доки запущена група не виконає вибрану умову завершення
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 ] );