Καρδιακοί παλμοί

Οι εργασίες μεγάλης διάρκειας και τα βήματα ροής δεδομένων μπορούν να στέλνουν καρδιακούς παλμούς για να επιβεβαιώνουν ότι παραμένουν ενεργά. Χωρίς αυτούς, ο ανιχνευτής παρωχημένων εργασιών μπορεί να ανακτήσει μια εργασία που εξακολουθεί να εκτελείται, επειδή δεν έχει ενημερωθεί η χρονική σήμανση reserved_at της.

Πώς λειτουργεί

Όταν ένας worker αρχίζει να επεξεργάζεται μια εργασία, καλεί το Heartbeat::init() για να ορίσει το τρέχον περιβάλλον εργασίας. Μέσα στον χειριστή σας, η κλήση του Heartbeat::beat() ενημερώνει τη στήλη reserved_at της εργασίας σε NOW(). Ο ανιχνευτής παρωχημένων εργασιών χρησιμοποιεί το reserved_at για να κρίνει αν μια εργασία έχει κολλήσει, οπότε η ανανέωσή του διατηρεί την εργασία ασφαλή.

Όταν ολοκληρωθεί η εργασία, ο worker καλεί το Heartbeat::clear() για να επαναφέρει το περιβάλλον.

Αποστολή καρδιακών παλμών

Καλέστε το Heartbeat::beat() μέσα από οποιονδήποτε χειριστή εργασίας ή βήμα ροής δεδομένων:

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

Δεδομένα προόδου

Η παράμετρος $progress είναι ένας προαιρετικός συσχετιστικός πίνακας. Αν δοθεί, κωδικοποιείται σε JSON και αποθηκεύεται στη στήλη heartbeat_data της εγγραφής εργασίας. Έτσι μπορείτε να παρακολουθείτε την πρόοδο έξω από τον worker:

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

Όταν καλείται χωρίς ορίσματα, ενημερώνεται μόνο το reserved_at:

Heartbeat::beat();

API καρδιακών παλμών

Η κλάση Queuety\Heartbeat είναι εξ ολοκλήρου στατική. Δεν δημιουργείτε στιγμιότυπό της.

Χρήση σε βήματα ροής δεδομένων

Ο worker στέλνει ήδη καρδιακό παλμό μετά από κάθε τμήμα που αποθηκεύει, με το heartbeat_data ορισμένο σε {"streaming_chunks": n}, οπότε μια ροή που αποδίδει τακτικά τμήματα παραμένει ενημερωμένη από μόνη της. Καλέστε το Heartbeat::beat() μόνοι σας όταν μπορεί να υπάρξουν μεγάλα διαστήματα μεταξύ των αποδόσεων, για παράδειγμα κατά την αναμονή για το πρώτο token. Τα δεδομένα που περνάτε αντικαθίστανται από τον επόμενο παλμό ανά τμήμα του worker:

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

Πότε να χρησιμοποιείτε καρδιακούς παλμούς

Χρησιμοποιήστε καρδιακούς παλμούς όταν μια εργασία ή ένα βήμα διαρκεί περισσότερο από QUEUETY_STALE_TIMEOUT δευτερόλεπτα (προεπιλογή: 600). Συνήθεις περιπτώσεις:

• Κλήσεις LLM API με αποκρίσεις σε ροή
• Λήψεις ή μεταφορτώσεις μεγάλων αρχείων
• Επαναλαμβανόμενη επεξεργασία σε παρτίδες μέσα σε ένα μόνο βήμα
• Κάθε εξωτερική κλήση API που μπορεί να χρειαστεί λεπτά για να απαντήσει

Αν το βήμα σας ολοκληρώνεται σε λιγότερο από ένα λεπτό, δεν χρειάζονται heartbeats. Ο ανιχνευτής παρωχημένων εργασιών ενεργοποιείται μόνο αφού παρέλθει το διαμορφωμένο χρονικό όριο.