Semnale periodice

Activitățile de lungă durată și pașii de streaming pot trimite semnale periodice pentru a confirma că sunt încă active. Fără acestea, detectorul de activități învechite poate revendica o activitate care încă se procesează, deoarece marcajul temporal reserved_at nu a fost actualizat.

Cum funcționează

Când un worker începe să proceseze o activitate, apelează Heartbeat::init() pentru a seta contextul activității curente. În handler, apelarea Heartbeat::beat() actualizează coloana reserved_at a activității cu NOW(). Detectorul de activități învechite folosește reserved_at pentru a stabili dacă o activitate este blocată, așa că reîmprospătarea acesteia protejează activitatea.

Când activitatea se încheie, worker-ul apelează Heartbeat::clear() pentru a reseta contextul.

Trimiterea semnalelor periodice

Apelați Heartbeat::beat() din interiorul oricărui handler de activitate sau pas de streaming:

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

Date despre progres

Parametrul $progress este un array asociativ opțional. Când este furnizat, este codificat în JSON și stocat în coloana heartbeat_data din rândul activității. Astfel, puteți urmări progresul din afara worker-ului:

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

Dacă apelul nu are argumente, se actualizează doar reserved_at:

Heartbeat::beat();

API pentru semnale periodice

Clasa Queuety\Heartbeat este complet statică. Nu creați o instanță a acesteia.

Utilizarea în pașii de streaming

Worker-ul trimite deja un semnal periodic după fiecare fragment pe care îl persistă, cu heartbeat_data setat la {"streaming_chunks": n}, așa că un flux care generează regulat fragmente rămâne actualizat automat. Apelați chiar dvs. Heartbeat::beat() când pot exista intervale mari între fragmente, de exemplu în timp ce așteptați primul token. Următorul semnal trimis de worker după un fragment suprascrie datele furnizate:

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 [];
    }
}

Când să folosiți semnale periodice

Folosiți semnale periodice când o singură activitate sau un pas durează mai mult de QUEUETY_STALE_TIMEOUT secunde (implicit: 600). Cazuri obișnuite:

• Apeluri LLM API cu răspunsuri în flux
• Descărcări sau încărcări de fișiere mari
• Repetări ale procesării în loturi într-un singur pas
• Orice apel extern API care poate dura câteva minute până răspunde

Dacă pasul se încheie în mai puțin de un minut, semnalele heartbeat nu sunt necesare. Detectorul de procese inactive se activează numai după expirarea timpului de așteptare configurat.