Alur kerja

Lihat sebagai Markdown Edit di GitHub Dokumentasi di 2cc7b2b

Alur kerja memecah pekerjaan yang berjalan lama menjadi langkah-langkah terpisah. Setiap langkah menyimpan output-nya ke dalam wadah status bersama di basis data. Jika PHP mati di tengah langkah, worker mencoba ulang langkah tersebut dengan seluruh status sebelumnya tetap utuh.

Apa itu alur kerja?

Alur kerja adalah urutan langkah yang tersusun. Setiap langkah merupakan kelas PHP yang menerima status terakumulasi dari semua langkah sebelumnya dan mengembalikan data baru untuk digabungkan ke dalam status. Alur kerja berpindah satu langkah setiap kali, dengan setiap transisi dibungkus dalam transaksi MySQL atomik.

PHP

use Queuety\Queuety;

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

Akumulasi status

Payload awal yang Anda berikan ke dispatch() menjadi status awal. Kunci yang dikembalikan oleh setiap langkah digabungkan ke dalam status. Kunci yang diawali dengan _ dicadangkan untuk nilai kontrol seperti _next_step dan tidak digabungkan ke status publik:

Kode

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'}

Setiap langkah melihat seluruh status yang terakumulasi. Dengan begitu, Anda dapat meneruskan data antar-langkah dengan mudah tanpa penyimpanan eksternal.

Kemampuan untuk dilanjutkan

Alur kerja bersifat tahan lama karena status disimpan ke basis data setelah setiap langkah. Jika worker mogok selama langkah 1, alur kerja dimulai ulang dari langkah 1 dengan output langkah 0 yang sudah ada dalam status.

Batas langkah berupa satu transaksi MySQL yang secara atomik:

Memperbarui status alur kerja dengan output langkah tersebut
Menandai tugas langkah saat ini sebagai selesai
Mengantrekan tugas langkah berikutnya

Ini menjamin bahwa status tidak akan hilang, bahkan saat terjadi crash, timeout, atau kesalahan fatal PHP.

Siklus hidup alur kerja

Kode

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)

Status	Deskripsi
`running`	Sedang memproses langkah-langkah
`completed`	Semua langkah berhasil diselesaikan
`failed`	Langkah gagal setelah seluruh percobaan ulang habis
`paused`	Dijeda secara manual. Pekerjaan yang sedang berjalan dapat diselesaikan, tetapi langkah berikutnya tidak diantrekan hingga dilanjutkan
`waiting_for_signal`	Menunggu sinyal eksternal melalui `Queuety::signal()`
`waiting_for_workflows`	Menunggu satu atau beberapa alur kerja lain selesai melalui `wait_for_workflow()` atau `wait_for_workflows()`
`cancelled`	Dibatalkan secara eksplisit melalui `Queuety::cancel_workflow()`. Tugas yang tertunda dikubur dan handler pembersihan opsional dijalankan.

Mengelola alur kerja

Periksa status:

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

Jeda dan lanjutkan:

PHP

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

Coba ulang alur kerja yang gagal dari langkah yang gagal:

PHP

Queuety::retry_workflow( $workflow_id );

Batalkan alur kerja dan jalankan handler pembersihan:

PHP

Queuety::cancel_workflow( $workflow_id );

Lihat Pembatalan untuk mengetahui detail tentang handler pembersihan dan apa yang terjadi saat alur kerja dibatalkan.

Opsi alur kerja

Atur antrean, prioritas, dan perilaku percobaan ulang untuk semua langkah dalam alur kerja:

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

Metode on_cancel() mendaftarkan handler pembersihan yang dijalankan saat alur kerja dibatalkan. Lihat Pembatalan.

Metode prune_state_after() mengaktifkan penghapusan otomatis output langkah lama agar status alur kerja tetap terbatas ukurannya. Lihat Pemangkasan Status.

Gunakan compensate_with() untuk memasang handler kompensasi per langkah. Gunakan compensate_on_failure() jika Anda ingin kompensasi langkah yang telah selesai dijalankan secara otomatis setelah kegagalan alur kerja, bukan hanya saat pembatalan eksplisit. Lihat Kompensasi.

Gunakan batas pengaman alur kerja saat Anda memerlukan orkestrasi jangka panjang yang lebih aman: version() memberi definisi tag versi tingkat aplikasi, Queuety juga menyimpan hash definisi deterministik untuk setiap eksekusi, idempotency_key() membuat dispatch tahan lama terhadap percobaan ulang atau permintaan duplikat, dan max_transitions(), max_for_each_items(), serta max_state_bytes() menggagalkan eksekusi yang tak terkendali sebelum memburuk. Lihat Batas Pengaman.

Jenis langkah

Queuety mendukung sebelas jenis langkah dalam alur kerja:

Rangkaian berurutan menggunakan ->then() untuk eksekusi satu per satu
Grup paralel menggunakan ->parallel() untuk eksekusi bersamaan
Perulangan dinamis untuk setiap item menggunakan ->for_each() untuk pekerjaan cabang yang ditemukan saat runtime dan semantik penyelesaian eksplisit
Serah terima asinkron menggunakan ->start_workflows() atau ->start_agents() untuk mengirim alur kerja anak independen dari item yang ditemukan saat runtime
Percabangan bersyarat menggunakan langkah bernama dan _next_step
Pengulangan menggunakan ->repeat_until() dan ->repeat_while() untuk kembali secara tahan lama ke langkah bernama sebelumnya
Jalankan Alur Kerja menggunakan ->run_workflow() untuk komposisi bertingkat
Penundaan tahan lama menggunakan ->delay() untuk penundaan berbasis waktu
Sinyal menggunakan ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input(), dan ->wait_for_decision() untuk pemicu peristiwa eksternal
Dependensi alur kerja menggunakan ->wait_for_workflow(), ->wait_for_workflows(), dan ->wait_for_agents() untuk koordinasi lintas alur kerja
Langkah streaming menggunakan antarmuka StreamingStep untuk output berbasis potongan yang tahan lama

Pola umum

Jika Anda membangun sistem otonom atau semi-otonom, mulailah dengan Orkestrasi Agen. Panduan ini menunjukkan bagaimana for_each(), start_agents(), wait_for_agents(), wait_for_workflow(), dan wait_for_decision() berpadu dalam alur planner/executor dan tinjauan manusia yang nyata.