İş akışları

Markdown olarak görüntüle GitHub üzerinde düzenle Dokümantasyon: 2cc7b2b

İş akışları uzun süren işleri ayrı adımlara böler. Her adım, çıktısını veritabanındaki paylaşılan bir durum alanına kaydeder. PHP adımın ortasında çökerse çalışan, önceki tüm durum korunarak o adımı yeniden dener.

İş akışları nedir?

İş akışı, sıralı adımlardan oluşan bir dizidir. Her adım, önceki tüm adımlardan biriken durumu alan ve duruma eklenecek yeni veriler döndüren bir PHP sınıfıdır. İş akışı her seferinde bir adım ilerler ve her geçiş atomik bir MySQL işlemi içine alınır.

PHP

use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

Durum biriktirme

dispatch() öğesine ilettiğiniz ilk veri, başlangıç durumu olur. Her adımın döndürdüğü anahtarlar duruma eklenir. _ ile başlayan anahtarlar, _next_step gibi denetim değerleri için ayrılmıştır ve herkese açık duruma eklenmez:

Kod

dispatch(['user_id' => 42])

Step 0: FetchDataHandler   -> returns ['user_name' => 'Alice']
  state: {user_id: 42, user_name: 'Alice'}

Step 1: CallLLMHandler     -> returns ['llm_response' => '...']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...'}

Step 2: FormatOutputHandler -> returns ['report_url' => '/reports/42.pdf']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...', report_url: '/reports/42.pdf'}

Her adım, birikmiş durumun tamamını görür. Bu sayede harici depolama kullanmadan adımlar arasında veri aktarmak kolaylaşır.

Devam ettirilebilirlik

Durum her adımdan sonra veritabanına kaydedildiği için iş akışları kalıcıdır. Çalışan 1. adım sırasında çökerse iş akışı, 0. adımın çıktısı zaten durumdayken 1. adımdan yeniden başlar.

Adım sınırı, atomik olarak şunları yapan tek bir MySQL işlemidir:

Adımın çıktısıyla iş akışı durumunu günceller
Geçerli adımın işini tamamlandı olarak işaretler
Sonraki adımın işini kuyruğa alır

Bu, çökmeler, zaman aşımları veya PHP kritik hatalar karşısında bile hiçbir durumun kaybolmamasını sağlar.

İş akışı yaşam döngüsü

Kod

dispatch() -> running -> step 0 -> step 1 -> ... -> step N -> completed
                  |                   |
                  |                   +-> failed -> retry -> running (from failed step)
                  |
                  +-> paused -> resumed -> running
                  |
                  +-> waiting_for_signal -> signal received -> running
                  |
                  +-> waiting_for_workflows -> dependency settles -> running
                  |
                  +-> cancelled (via cancel_workflow)

Durum	Açıklama
`running`	Adımlar etkin olarak işleniyor
`completed`	Tüm adımlar başarıyla tamamlandı
`failed`	Bir adım, yeniden deneme hakkı tükendikten sonra başarısız oldu
`paused`	Elle duraklatıldı. Devam eden işler tamamlanabilir, ancak sürdürülene kadar sonraki adımlar kuyruğa alınmaz
`waiting_for_signal`	`Queuety::signal()` aracılığıyla harici sinyal bekleniyor
`waiting_for_workflows`	`wait_for_workflow()` veya `wait_for_workflows()` aracılığıyla tamamlanması için başka bir iş akışı ya da iş akışları bekleniyor
`cancelled`	`Queuety::cancel_workflow()` aracılığıyla açıkça iptal edildi. Bekleyen işler arşivlenir ve isteğe bağlı bir temizleme işleyicisi çalışır.

İş akışlarını yönetme

Durumu kontrol edin:

PHP

$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value;   // 'running'
echo $state->current_step;    // 1
echo $state->total_steps;     // 3
print_r( $state->waiting_for ); // ['42'] while waiting on another workflow
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info

Duraklatma ve sürdürme:

PHP

Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id );

Başarısız bir iş akışını başarısız olduğu adımdan yeniden deneyin:

PHP

Queuety::retry_workflow( $workflow_id );

Bir iş akışını iptal edin ve temizleme işleyicilerini çalıştırın:

PHP

Queuety::cancel_workflow( $workflow_id );

Temizleme işleyicileri ve bir iş akışı iptal edildiğinde neler olduğu hakkında ayrıntılar için İptal bölümüne bakın.

İş akışı seçenekleri

İş akışındaki tüm adımlar için kuyruk, öncelik ve yeniden deneme davranışını ayarlayın:

PHP

use Queuety\Enums\Priority;

Queuety::workflow( 'important_pipeline' )
    ->on_queue( 'pipelines' )
    ->with_priority( Priority::High )
    ->max_attempts( 5 )
    ->on_cancel( PipelineCleanupHandler::class )
    ->prune_state_after( 3 )
    ->then( StepA::class )
    ->then( StepB::class )
    ->dispatch( $payload );

on_cancel() yöntemi, iş akışı iptal edildiğinde çalışan bir temizleme işleyicisi kaydeder. Bkz. İptal.

prune_state_after() yöntemi, iş akışı durumunun sınırlandırılmış kalması için eski adım çıktılarını otomatik olarak kaldırmayı sağlar. Bkz. Durum Budama.

Adım başına telafi işleyicisi eklemek için compensate_with() kullanın. Tamamlanan adımların telafi işlemlerinin yalnızca açık iptal durumunda değil, iş akışı başarısız olduğunda da otomatik çalışmasını istiyorsanız compensate_on_failure() kullanın. Bkz. Telafi.

Uzun süren orkestrasyonları daha güvenli hâle getirmek için iş akışı koruyucu sınırlarını kullanın: version() bir tanıma uygulama düzeyinde sürüm ekler, Queuety her çalıştırma için deterministik bir tanım karması da saklar, idempotency_key() yeniden denemelerde veya yinelenen isteklerde dağıtımı kalıcı hâle getirir; max_transitions(), max_for_each_items() ve max_state_bytes() ise kontrolden çıkan çalıştırmaları büyümeden başarısız kılar. Bkz. Koruyucu Sınırlar.

Adım türleri

Queuety, bir iş akışında on bir adım türünü destekler:

Sıralı zincirler: teker teker yürütme için ->then() kullanır
Paralel gruplar: eşzamanlı yürütme için ->parallel() kullanır
Dinamik for-each: çalışma zamanında keşfedilen dallar ve açık tamamlanma kuralları için ->for_each() kullanır
Asenkron devirler: çalışma zamanında keşfedilen öğelerden bağımsız alt iş akışları dağıtmak için ->start_workflows() veya ->start_agents() kullanır
Koşullu dallanma: adlandırılmış adımlar ve _next_step kullanır
Tekrarlar: daha önceki adlandırılmış adımlara kalıcı geri bağlantılar kurmak için ->repeat_until() ve ->repeat_while() kullanır
İş Akışlarını Çalıştırma: iç içe oluşturma için ->run_workflow() kullanır
Kalıcı gecikmeler: zamana dayalı gecikmeler için ->delay() kullanır
Sinyaller: harici olay tetikleyicileri için ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() ve ->wait_for_decision() kullanır
İş akışı bağımlılıkları: iş akışları arası koordinasyon için ->wait_for_workflow(), ->wait_for_workflows() ve ->wait_for_agents() kullanır
Akış adımları: parçalara dayalı kalıcı çıktı için StreamingStep arayüzünü kullanır

Yaygın Örüntüler

Otonom veya yarı otonom sistemler geliştiriyorsanız Aracı Orkestrasyonu ile başlayın. Gerçek planlayıcı/yürütücü ve insan incelemesi akışlarında for_each(), start_agents(), wait_for_agents(), wait_for_workflow() ve wait_for_decision() öğelerinin nasıl bir araya geldiğini gösterir.