Munkafolyamatok

A munkafolyamatok különálló lépésekre bontják a hosszú ideig futó feladatokat. Minden lépés kimenete egy közös állapotobjektumba kerül az adatbázisban. Ha a PHP leáll egy lépés közben, a feldolgozó az összes korábbi állapot megőrzésével újrapróbálja azt a lépést.

Mik azok a munkafolyamatok?

A munkafolyamat lépések rendezett sorozata. Minden lépés egy PHP osztály, amely megkapja az összes korábbi lépésből összeállt állapotot, és új adatokat ad vissza, amelyeket az állapotba kell egyesíteni. A munkafolyamat lépésenként halad előre, minden átmenetet egy atomi MySQL tranzakció foglal magában.

use Queuety\Queuety;

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

Az állapot felhalmozása

A dispatch() számára átadott kezdeti adatcsomag lesz a kiinduló állapot. Minden lépés visszaadott kulcsai egyesülnek az állapottal. A _ karakterrel kezdődő kulcsok vezérlőértékek, például _next_step számára vannak fenntartva, ezért nem kerülnek be a nyilvános állapotba:

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

Minden lépés a teljes, addig felhalmozott állapotot látja. Így könnyedén továbbíthatók az adatok a lépések között külső tároló nélkül.

Folytathatóság

A munkafolyamatok azért tartósak, mert az állapot minden lépés után az adatbázisba kerül. Ha a feldolgozó az 1. lépés közben összeomlik, a munkafolyamat az 1. lépéssel indul újra, miközben a 0. lépés kimenete már az állapotban van.

A lépések határa egyetlen MySQL tranzakció, amely atomikusan:

1. Frissíti a munkafolyamat állapotát a lépés kimenetével
2. Befejezettként jelöli az aktuális lépés feladatát
3. Sorba állítja a következő lépés feladatát

Ez garantálja, hogy soha nem vész el állapot, még összeomlás, időtúllépés vagy PHP végzetes hiba esetén sem.

A munkafolyamat életciklusa

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)

Munkafolyamatok kezelése

Állapot ellenőrzése:

$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

Szüneteltetés és folytatás:

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

Sikertelen munkafolyamat újrapróbálása a sikertelen lépéstől:

Queuety::retry_workflow( $workflow_id );

Munkafolyamat megszakítása és takarító kezelők futtatása:

Queuety::cancel_workflow( $workflow_id );

A takarító kezelőkről és a munkafolyamat megszakításakor történő eseményekről a Megszakítás oldalon olvashat részletesen.

Munkafolyamat-beállítások

A munkafolyamat összes lépésére beállítható a sor, a prioritás és az újrapróbálkozás viselkedése:

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

A on_cancel() metódus takarító kezelőt regisztrál, amely a munkafolyamat megszakításakor fut le. Lásd: Megszakítás.

A prune_state_after() metódus engedélyezi a régi lépések kimenetének automatikus eltávolítását, hogy a munkafolyamat állapota korlátok között maradjon. Lásd: Állapot ritkítása.

A compensate_with() segítségével lépésenkénti kompenzációs kezelőt csatolhat. Akkor használja a compensate_on_failure() lehetőséget, ha azt szeretné, hogy a befejezett lépések kompenzációi automatikusan lefussanak a munkafolyamat hibája után, ne csak kifejezett megszakításkor. Lásd: Kompenzáció.

Biztonságosabb, hosszan futó vezényléshez használja a munkafolyamatok korlátait: a version() alkalmazásszintű verzióval lát el egy definíciót, a Queuety emellett minden futáshoz determinisztikus definícióhash-t is tárol, a idempotency_key() tartóssá teszi az indítást újrapróbálkozások és ismételt kérések esetén is, a max_transitions(), max_for_each_items() és max_state_bytes() pedig még az elszabaduló futásokat is leállítja, mielőtt kezelhetetlenné válnának. Lásd: Korlátok.

Lépéstípusok

A Queuety tizenegyféle lépést támogat egy munkafolyamaton belül:

• Szekvenciális láncok a ->then() használatával az egymás utáni végrehajtáshoz
• Párhuzamos csoportok a ->parallel() használatával az egyidejű végrehajtáshoz
• Dinamikus for-each a ->for_each() használatával futás közben felfedezett ágak feldolgozásához és a befejezés egyértelmű meghatározásához
• Aszinkron átadások a ->start_workflows() vagy a ->start_agents() használatával független gyermek-munkafolyamatok indításához a futás közben felfedezett elemekből
• Feltételes elágazás névvel ellátott lépésekkel és a _next_step használatával
• Ismétlések a ->repeat_until() és a ->repeat_while() használatával a korábbi, névvel ellátott lépésekhez vezető tartós visszaugrásokhoz
• Munkafolyamatok futtatása a ->run_workflow() használatával a beágyazott összeállításhoz
• Tartós késleltetések a ->delay() használatával az időalapú várakozásokhoz
• Jelek a ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() és ->wait_for_decision() használatával külső események indításához
• Munkafolyamat-függőségek a ->wait_for_workflow(), ->wait_for_workflows() és ->wait_for_agents() használatával a munkafolyamatok közötti koordinációhoz
• Folyamatos lépések a StreamingStep felület használatával a darabonkénti kimenet tartós kezeléséhez

Gyakori minták

Ha autonóm vagy részben autonóm rendszereket épít, kezdje az Ügynökök vezénylése oldallal. Bemutatja, hogyan illeszkedik egymáshoz a for_each(), a start_agents(), a wait_for_agents(), a wait_for_workflow() és a wait_for_decision() a valós tervező/végrehajtó és emberi felülvizsgálati folyamatokban.