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

Оркестрация агентов

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

Queuety хорошо подходит для систем агентов, которым нужна адаптивная работа без потери надёжного состояния, повторов, возможности проверки и явной семантики ожидания.

Для большинства сценариев работы агентов не нужен механизм DAG с возможностью менять структуру во время выполнения. Обычно достаточно небольшого набора надёжных операций оркестрации. for_each() распределяет работу внутри одного запуска рабочего процесса. start_agents() и wait_for_agents() передают обнаруженные во время выполнения задачи в независимые рабочие процессы верхнего уровня. wait_for_decision() и wait_for_input() приостанавливают выполнение для участия человека. wait_for_workflow(), wait_for_workflows() и wait_for_agent_group() позволяют одному запуску верхнего уровня дождаться другого запуска или запущенной группы, достигшей кворума.

Полезная модель для понимания

Оркестрацию агентов проще всего рассматривать по уровням. Планировщик определяет, какая работа нужна. Queuety превращает её в надёжные задания или рабочие процессы. Запуски специализированных агентов формируют результаты и сохраняют артефакты. Контролирующий этап объединяет результаты и решает, что делать дальше. В любой момент человек может одобрить, отклонить или скорректировать запуск, прежде чем он продолжится.

Ключевое различие в том, что агент решает, какую работу выполнять, а Queuety по-прежнему определяет, как эта работа сохраняется, повторяется, возобновляется и объединяется. Если цель оркестрации — долгоживущий сеанс, а не ограниченный по времени запуск, используйте вместе с Конечными автоматами.

Шаблон 1: планировщик и исполнитель с независимыми запусками агентов

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

PHP
use Queuety\Enums\WaitMode;
use Queuety\Queuety;

$agent_run = Queuety::workflow( 'agent_run' )
    ->then( FetchSourcesStep::class )
    ->then( DraftFindingStep::class );

Queuety::workflow( 'brief_research' )
    ->version( 'brief-research.v1' )
    ->idempotency_key( 'brief:42:research' )
    ->max_transitions( 20 )
    ->then( PlanResearchTasksStep::class )
    ->start_agents( 'agent_tasks', $agent_run, group_key: 'researchers' )
    ->wait_for_agent_group( 'researchers', WaitMode::Quorum, 2, 'agent_results' )
    ->then( SynthesizeBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

В этом шаблоне PlanResearchTasksStep записывает обнаруженные во время выполнения элементы в $state['agent_tasks'], start_agents() запускает отдельный надёжный рабочий процесс для каждой задачи, а wait_for_agent_group() приостанавливает родительский процесс до выполнения выбранного условия завершения. Затем SynthesizeBriefStep получает завершённые состояния дочерних процессов в $state['agent_results'].

Это лучший отправной шаблон для систем с планировщиком и исполнителем: каждый запуск агента можно проверять и повторять отдельно.

Шаблон 2: проверка человеком перед продолжением запуска

Используйте wait_for_decision(), когда рабочий процесс должен остановиться для принятия решения «да» или «нет», и wait_for_input(), когда оператору нужно передать структурированную обратную связь обратно в запуск.

PHP
use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'brief_review' )
    ->then( DraftBriefStep::class )
    ->wait_for_decision( result_key: 'review' )
    ->then( HandleReviewDecisionStep::class )
    ->wait_for_input( result_key: 'revision_notes' )
    ->then( ApplyRevisionNotesStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

Если проверяющий одобряет:

PHP
Queuety::approve_workflow(
    $workflow_id,
    [
        'reviewer' => 'editor@example.com',
        'notes' => 'Looks good, publish after legal check',
    ],
    'approved'
);

Если проверяющий отклоняет:

PHP
Queuety::reject_workflow(
    $workflow_id,
    [
        'reviewer' => 'editor@example.com',
        'reason' => 'Needs citations and a stronger conclusion',
    ],
    'rejected'
);

После wait_for_decision() рабочий процесс получает структурированную полезную нагрузку с итогом, сигналом, возобновившим запуск, и данными проверяющего, если они были переданы. Так создаётся полноценная контрольная точка с участием человека, при этом процесс PHP не остаётся открытым во время ожидания.

Шаблон 3: один рабочий процесс верхнего уровня ожидает другой

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

PHP
use Queuety\Enums\WaitMode;
use Queuety\Queuety;

$research_id = Queuety::workflow( 'research_brief' )
    ->then( PlanResearchTasksStep::class )
    ->start_agents( 'agent_tasks', $agent_run )
    ->wait_for_agents( mode: WaitMode::All, result_key: 'agent_results' )
    ->then( SynthesizeBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

Queuety::workflow( 'publish_brief' )
    ->wait_for_workflow( $research_id, 'research' )
    ->wait_for_decision( result_key: 'editor_review' )
    ->then( PublishBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

Это создаёт надёжную границу между этапами исследования и публикации. У каждого запуска свой операционный жизненный цикл, но второй рабочий процесс всё равно возобновляется автоматически после завершения зависимости.

Выбор подходящего примитива

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

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

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

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

Рекомендуемые соглашения для состояний

Системы агентов проще отлаживать, если ключи состояния предсказуемы. Обычно agent_tasks удобно использовать для результатов планировщика, agent_workflow_ids — для запущенных процессов верхнего уровня, agent_results — для состояния завершённых дочерних рабочих процессов, а review или decision — для данных одобрения или отклонения. Большие результаты, например черновики, цитаты или манифесты исследований, обычно лучше хранить в хранилище артефактов, а не в основном наборе состояния рабочего процесса.

Ограничения особенно полезны для рабочих процессов агентов, поскольку их поведение динамично по своей природе.

PHP
Queuety::workflow( 'agent_research' )
    ->version( 'agent-research.v2' )
    ->idempotency_key( 'brief:42:agent-research' )
    ->max_transitions( 20 )
    ->max_for_each_items( 12 )
    ->max_state_bytes( 32768 );

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

См. также