Към съдържанието

Оркестриране на агенти

Преглед като 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 );

Тези контроли дават отговор на важните за продукционна среда практически въпроси: коя версия на дефиницията е създала дадено изпълнение, могат ли дублирани външни заявки да го изпратят два пъти, може ли планировчикът да създаде твърде много клонове и може ли състоянието на работния поток да нараства без ограничение. При повечето агентски системи започнете с тези предпазни мерки, преди да настройвате броя на работниците или топологията на опашките.

Вижте също