Eşzamansız Devirler

Bir iş akışının keşfedilen işleri tek bir çalıştırmanın içinde tutmak yerine bağımsız üst düzey iş akışlarına devretmesi gerektiğinde start_workflows() kullanın.

Ajan çalıştırmalarını özellikle modelliyorsanız, start_agents() ve wait_for_agents() ajanlara yönelik varsayılanlara sahip anlamsal takma adlardır.

Daha kapsamlı uçtan uca örnekler için Ajan Orkestrasyonu bölümüne bakın.

Bu, ajan tarzı orkestrasyon için kullanışlıdır:

• bir planlayıcı iş akışı çalışma zamanında görevleri keşfeder
• her görev kendi kalıcı iş akışına dönüşür
• üst iş akışı bu iş akışlarının kimliklerini duruma kaydeder
• üst iş akışı daha sonra all, any, quorum veya adı belirtilmiş bir başlatılmış grubu bekleyebilir

->start_workflows() yöntemi

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

Bu örnekte:

1. PlanTopicsStep, konu yüklerinden oluşan bir diziyi $state['topics'] konumuna yazar
2. start_workflows(), her konu için bir agent_task iş akışı başlatır
3. Başlatılan iş akışlarının kimlikleri $state['child_workflow_ids'] konumuna yazılır
4. wait_for_workflow_group(), başlatılan grup seçilen tamamlanma koşulunu karşılayana kadar duraklar
5. SynthesizeBriefStep, tamamlanan alt iş akışı durumlarını $state['child_results'] altında alır

Alt iş akışı yükleri

items_key içindeki her öğe, başlatılan bir iş akışının ilk yükü olur.

Öğe bir diziyse, genel anahtarları alt iş akışının yüküne birleştirilir:

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

Öğe skaler ise, Queuety onu payload_key altında saklar:

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

Üst iş akışı durumunu devralma

Varsayılan olarak start_workflows(), üst iş akışının genel durumunu her alt iş akışının yüküyle birleştirir.

Bu, brief_id, account_id veya campaign gibi paylaşılan bağlamın başlatılan her iş akışında otomatik olarak kullanılabileceği anlamına gelir.

Kaynak items_key ile result_key devralmaya dahil edilmez; böylece her alt iş akışı görev listesinin tamamını veya nihai iş akışı kimlikleri dizisini almaz.

Her alt iş akışının yalnızca kendi öğe yükünü alması gerekiyorsa inherit_state: false ayarlayın.

Neden run_workflow() değil?

run_workflow(), üst iş akışının alt iş akışı tamamlanana kadar o adımda beklediği bir üst-alt ilişkisi oluşturur.

start_workflows() daha esnektir:

• başlatılan iş akışları bağımsız üst düzey iş akışlarıdır
• başlatılan her iş akışı kendi başına incelenebilir, yeniden denenebilir veya dışa aktarılabilir
• üst iş akışı, bunları bekleyip beklemeyeceğine ve nasıl bekleyeceğine daha sonra karar verir

Bu nedenle start_workflows(), ajan görevlerini eşzamansız olarak devretmeye daha uygundur; run_workflow() ise tek bir orkestrasyon ağacı içindeki iç içe bileşimler için daha uygundur.

Daha sonra bekleme

start_workflows(), İş Akışı Bağımlılıkları ile birlikte kullanılmak üzere tasarlanmıştır:

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, tamamlanan tüm alt iş akışı durumlarını toplar
• WaitMode::Any, ilk başarılı alt iş akışında devam eder

Böylece çalışma zamanında düzenlenebilen tam bir DAG motoruna ihtiyaç duymadan planlayıcı/yürütücü orkestrasyonu kurabilirsiniz.

Ajanlara uygun takma adlar

start_agents(), planlayıcı/yürütücü akışlarına göre ayarlanmış varsayılanlara sahip bir start_workflows() takma adıdır:

• result_key varsayılan olarak agent_workflow_ids değerini alır
• payload_key varsayılan olarak agent_task değerini alır
• name varsayılan olarak start_agents değerini alır

wait_for_agents(), ajan sonuçlarını toplamaya göre ayarlanmış varsayılanlara sahip bir wait_for_workflows() takma adıdır:

• workflows varsayılan olarak agent_workflow_ids değerini alır
• result_key varsayılan olarak agent_results değerini alır
• name varsayılan olarak wait_for_agents değerini alır

wait_for_agent_group(), ajan gruplarına göre ayarlanmış varsayılanlara sahip bir wait_for_workflow_group() takma adıdır.

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