Sirdspuksti

Skatīt kā Markdown Rediģēt vietnē GitHub Dokumentācija vietnē 2cc7b2b

Ilgstoši izpildāmi uzdevumi un straumēšanas darbības var sūtīt sirdspukstus, lai apliecinātu, ka tie joprojām darbojas. Bez sirdspukstiem novecojušu uzdevumu noteicējs var pārņemt uzdevumu, kas joprojām tiek aktīvi apstrādāts, jo tā laikspiedols reserved_at nav atjaunināts.

Kā tas darbojas

Kad darba process sāk apstrādāt uzdevumu, tas izsauc Heartbeat::init(), lai iestatītu pašreizējā uzdevuma kontekstu. Apstrādātājā izsaucot Heartbeat::beat(), uzdevuma kolonnā reserved_at tiek iestatīts NOW(). Novecojušu uzdevumu noteicējs izmanto reserved_at, lai noteiktu, vai uzdevums ir iestrēdzis, tāpēc tā atsvaidzināšana pasargā uzdevumu.

Kad uzdevums tiek pabeigts, darba process izsauc Heartbeat::clear(), lai atiestatītu kontekstu.

Sirdspukstu sūtīšana

Izsauciet Heartbeat::beat() jebkura uzdevuma apstrādātāja vai straumēšanas darbības iekšienē:

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

Progresa dati

Parametrs $progress ir neobligāts asociatīvs masīvs. Ja tas ir norādīts, tas tiek kodēts formātā JSON un saglabāts uzdevuma rindas kolonnā heartbeat_data. Tas ļauj sekot progresam ārpus darba procesa:

PHP

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

Ja izsaukts bez argumentiem, tiek atjaunināts tikai reserved_at:

PHP

Heartbeat::beat();

Sirdspukstu API

Klase Queuety\Heartbeat ir pilnībā statiska. Jūs to neinstancējat.

Metode	Apraksts
`Heartbeat::init( int $job_id, Connection $conn )`	Iestata uzdevuma kontekstu. Darba process to izsauc automātiski pirms apstrādes.
`Heartbeat::beat( array $progress = [] )`	Nosūta sirdspukstu. Atjaunina `reserved_at` un neobligāti saglabā progresa datus.
`Heartbeat::clear()`	Notīra uzdevuma kontekstu. Darba process to izsauc automātiski pēc apstrādes.
`Heartbeat::current_job_id()`	Iegūst pašreizējā uzdevuma ID (noder testēšanai).

Izmantošana straumēšanas darbībās

Darba process jau nosūta sirdspukstu pēc katra saglabātā fragmenta, iestatot heartbeat_data uz {"streaming_chunks": n}, tāpēc straume, kas regulāri izvada datus, pati saglabājas aktīva. Izsauciet Heartbeat::beat() paši, ja starp izvades reizēm var būt ilgi pārtraukumi, piemēram, gaidot pirmo tokenu. Jūsu padotos datus pārraksta nākamais darba procesa sirdspuksts pēc fragmenta:

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

Kad izmantot sirdspukstus

Izmantojiet sirdspukstus, ja viena uzdevuma vai darbības izpilde ilgst vairāk par QUEUETY_STALE_TIMEOUT sekundēm (noklusējums: 600). Biežākie gadījumi:

LLM API izsaukumi ar straumētām atbildēm
Lielu failu lejupielāde vai augšupielāde
Pakešu apstrādes atkārtojumi vienā darbībā
Jebkurš ārējs API izsaukums, kura atbildi var nākties gaidīt vairākas minūtes

Ja darbība tiek pabeigta mazāk nekā minūtē, sirdspuksti nav nepieciešami. Novecojušu uzdevumu noteicējs sāk darboties tikai pēc konfigurētā taimauta beigām.