Signály aktivity

Zobraziť ako Markdown Upraviť na GitHub Dokumentácia na 2cc7b2b

Dlho bežiace úlohy a kroky streamovania môžu odosielať signály aktivity, ktorými potvrdzujú, že sú stále aktívne. Bez nich môže detektor neaktívnych úloh prevziať úlohu, ktorá sa stále spracúva, pretože jej časová značka reserved_at sa neaktualizovala.

Ako to funguje

Keď pracovník začne spracúvať úlohu, zavolá Heartbeat::init() a nastaví aktuálny kontext úlohy. Volanie Heartbeat::beat() v obslužnej rutine aktualizuje stĺpec reserved_at úlohy na hodnotu NOW(). Detektor neaktívnych úloh používa reserved_at na určenie, či sa úloha zasekla, takže jej obnovenie ju udržiava v bezpečí.

Po dokončení úlohy pracovník zavolá Heartbeat::clear() a obnoví kontext.

Odosielanie signálov aktivity

Zavolajte Heartbeat::beat() v obslužnej rutine ľubovoľnej úlohy alebo kroku streamovania:

PHP

use Queuety\Heartbeat;

class LongRunningStep implements \Queuety\Step {

    public function handle( array $state ): array {
        $items = $state['items'];

        foreach ( $items as $i => $item ) {
            $this->process( $item );
            Heartbeat::beat( [ 'processed' => $i + 1, 'total' => count( $items ) ] );
        }

        return [ 'done' => true ];
    }

    public function config(): array {
        return [];
    }

    private function process( $item ): void {
        // Expensive work here.
    }
}

Údaje o priebehu

Parameter $progress je voliteľné asociatívne pole. Ak ho zadáte, zakóduje sa vo formáte JSON a uloží do stĺpca heartbeat_data v riadku úlohy. Vďaka tomu môžete sledovať priebeh aj mimo pracovníka:

PHP

Heartbeat::beat( [
    'processed' => 450,
    'total'     => 1000,
    'percent'   => 45,
] );

Ak funkciu zavoláte bez argumentov, aktualizuje sa iba reserved_at:

PHP

Heartbeat::beat();

Trieda API

Trieda Queuety\Heartbeat je výhradne statická. Nevytvárajte jej inštanciu.

Metóda	Opis
`Heartbeat::init( int $job_id, Connection $conn )`	Nastaví kontext úlohy. Pracovník ju zavolá automaticky pred spracovaním.
`Heartbeat::beat( array $progress = [] )`	Odošle signál aktivity. Aktualizuje `reserved_at` a voliteľne uloží údaje o priebehu.
`Heartbeat::clear()`	Vymaže kontext úlohy. Pracovník ju zavolá automaticky po spracovaní.
`Heartbeat::current_job_id()`	Vráti aktuálne ID úlohy (užitočné pri testovaní).

Použitie v krokoch streamovania

Pracovník už odosiela signál aktivity po každej uloženej časti a nastavuje heartbeat_data na {"streaming_chunks": n}, takže stream, ktorý pravidelne odovzdáva výstup, zostáva aktívny aj bez ďalších zásahov. Zavolajte Heartbeat::beat() sami, ak medzi odovzdaniami môžu vzniknúť dlhé prestávky, napríklad pri čakaní na prvý token. Údaje, ktoré odovzdáte, prepíše nasledujúci signál pracovníka po spracovaní ďalšej časti:

PHP

use Queuety\Contracts\StreamingStep;
use Queuety\Heartbeat;

class StreamAPIResponse implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        $response = $this->client->stream( $state['prompt'] );

        foreach ( $response as $i => $chunk ) {
            Heartbeat::beat( [ 'chunks' => $i + 1 ] );
            yield $chunk;
        }
    }

    public function on_complete( array $chunks, array $state ): array {
        return [ 'response' => implode( '', $chunks ) ];
    }

    public function config(): array {
        return [];
    }
}

Kedy používať signály aktivity

Signály aktivity používajte, keď jedna úloha alebo krok trvá dlhšie ako QUEUETY_STALE_TIMEOUT sekúnd (predvolene: 600). Bežné prípady:

Volania API s odpoveďami streamovanými z LLM
Sťahovanie alebo nahrávanie veľkých súborov
Opakované dávkové spracovanie v rámci jedného kroku
Ľubovoľné externé volanie API, ktorého odpoveď môže trvať niekoľko minút

Ak sa krok dokončí za menej než minútu, signály aktivity nie sú potrebné. Detektor neaktívnych úloh sa aktivuje až po uplynutí nastaveného časového limitu.