Βήματα ροής δεδομένων

Προβολή ως Markdown Επεξεργασία στο GitHub Τεκμηρίωση στο 2cc7b2b

Τα βήματα ροής δεδομένων επιτρέπουν σε μια ροή εργασιών να παράγει έξοδο σταδιακά. Κάθε τμήμα που παράγεται γράφεται αμέσως στη βάση δεδομένων, ώστε αν το βήμα αποτύχει κατά τη ροή, τα ήδη αποθηκευμένα τμήματα να διατηρούνται και να είναι διαθέσιμα στην επανάληψη.

Αυτό είναι χρήσιμο για αποκρίσεις ροής LLM, επεξεργασία μεγάλων αρχείων και κάθε περίπτωση όπου δεν πρέπει να χάνεται η μερική πρόοδος.

Η διεπαφή StreamingStep

Ένα βήμα ροής δεδομένων υλοποιεί το Queuety\Contracts\StreamingStep:

PHP

namespace Queuety\Contracts;

interface StreamingStep {
    public function stream( array $state, array $existing_chunks = [] ): \Generator;
    public function on_complete( array $chunks, array $state ): array;
    public function config(): array;
}

Μέθοδος	Περιγραφή
`stream( $state, $existing_chunks )`	Γεννήτρια που παράγει τμήματα συμβολοσειράς. Κάθε παραγωγή προκαλεί άμεση εγγραφή στη DB.
`on_complete( $chunks, $state )`	Καλείται μετά την ολοκλήρωση της ροής. Λαμβάνει όλα τα τμήματα (προηγούμενα και νέα) και επιστρέφει δεδομένα για συγχώνευση με την κατάσταση της ροής εργασιών.
`config()`	Προαιρετική διαμόρφωση. Υποστηρίζει κλειδιά `max_attempts` και `backoff`· το `backoff` δέχεται όνομα στρατηγικής (`exponential`, `linear`, `fixed`), καθώς μια λίστα καθυστερήσεων απορρίπτεται προς το παρόν (#16).

Πώς αποθηκεύονται τα τμήματα

Κάθε φορά που η γεννήτριά σας παράγει μια τιμή, το Queuety καλεί το ChunkStore::append_chunk() για να την εγγράψει στον πίνακα queuety_chunks. Δεν υπάρχει ομαδοποίηση ή προσωρινή αποθήκευση. Μία παραγωγή ισοδυναμεί με μία INSERT.

Η γραμμή του τμήματος αποθηκεύει τα job_id, workflow_id, step_index, chunk_index (με αρίθμηση από το μηδέν) και τη συμβολοσειρά content.

Δυνατότητα συνέχισης

Όταν ένα βήμα ροής δεδομένων επαναλαμβάνεται μετά από αποτυχία, ο worker φορτώνει όλα τα τμήματα που είχαν αποθηκευτεί και τα περνά ως παράμετρο $existing_chunks. Η υλοποίησή σας αποφασίζει πώς θα συνεχιστεί:

Παράλειψη στοιχείων που έχουν ήδη υποβληθεί σε επεξεργασία ελέγχοντας τον αριθμό των υπαρχόντων τμημάτων
Μετάβαση στη σωστή θέση ενός σελιδοποιημένου API χρησιμοποιώντας τον αριθμό των τμημάτων ως μετατόπιση σελίδας
Επανεκκίνηση από την αρχή αγνοώντας εντελώς το $existing_chunks (τα τμήματα από την προηγούμενη προσπάθεια παραμένουν στη DB και προστίθενται στην αρχή)

PHP

public function stream( array $state, array $existing_chunks = [] ): \Generator {
    $offset = count( $existing_chunks );

    $pages = $this->client->paginate( $state['query'], offset: $offset );

    foreach ( $pages as $page ) {
        yield json_encode( $page );
    }
}

Καρδιακοί παλμοί κατά τη ροή δεδομένων

Οι ροές μεγάλης διάρκειας πρέπει να στέλνουν καρδιακούς παλμούς για να μην ανακτήσει την εργασία ο ανιχνευτής παρωχημένων εργασιών. Καλέστε το Heartbeat::beat() μέσα στον βρόχο της γεννήτριάς σας:

PHP

use Queuety\Heartbeat;

public function stream( array $state, array $existing_chunks = [] ): \Generator {
    foreach ( $this->client->stream( $state['prompt'] ) as $i => $chunk ) {
        Heartbeat::beat( [ 'chunks' => $i + 1 ] );
        yield $chunk;
    }
}

Για λεπτομέρειες σχετικά με τη διεπαφή Heartbeat, δείτε την ενότητα Καρδιακοί παλμοί API.

Εκκαθάριση τμημάτων

Όταν η ροή ολοκληρωθεί με επιτυχία και επιστρέψει το on_complete(), ο worker καλεί το ChunkStore::clear_chunks() για να διαγράψει όλα τα τμήματα της εργασίας. Το συγχωνευμένο αποτέλεσμα από το on_complete() είναι η ανθεκτική εγγραφή. Τα τμήματα είναι προσωρινά εκ σχεδιασμού.

Αν το βήμα αποτύχει και επαναληφθεί, τα τμήματα δεν διαγράφονται. Περνούν ξανά μέσω του $existing_chunks στην επόμενη προσπάθεια.

Πλήρες παράδειγμα: ροή LLM

PHP

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

class StreamLLMResponse implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        // If we have existing chunks from a previous attempt, we can skip
        // re-requesting. The on_complete method will concatenate everything.
        if ( ! empty( $existing_chunks ) ) {
            return;
        }

        $response = OpenAI::chat()->createStreamed( [
            'model'    => 'gpt-4',
            'messages' => $state['messages'],
        ] );

        foreach ( $response as $i => $chunk ) {
            $text = $chunk->choices[0]->delta->content ?? '';
            if ( '' !== $text ) {
                Heartbeat::beat( [ 'tokens' => $i ] );
                yield $text;
            }
        }
    }

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

    public function config(): array {
        return [ 'max_attempts' => 5 ];
    }
}

Χρησιμοποιήστε το σε μια ροή εργασιών όπως οποιοδήποτε άλλο βήμα:

PHP

use Queuety\Queuety;

Queuety::workflow( 'ai_chat' )
    ->then( BuildPromptHandler::class )
    ->then( StreamLLMResponse::class )
    ->then( SaveResponseHandler::class )
    ->dispatch( [ 'user_input' => $input ] );

Πλήρες παράδειγμα: επεξεργασία αρχείων

PHP

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

class ProcessCSVRows implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        $file   = fopen( $state['csv_path'], 'r' );
        $offset = count( $existing_chunks );
        $row    = 0;

        // Skip the header.
        fgetcsv( $file );

        while ( $line = fgetcsv( $file ) ) {
            $row++;

            // Skip rows already processed in a previous attempt.
            if ( $row <= $offset ) {
                continue;
            }

            $result = $this->process_row( $line );
            Heartbeat::beat( [ 'row' => $row ] );
            yield json_encode( $result );
        }

        fclose( $file );
    }

    public function on_complete( array $chunks, array $state ): array {
        $results = array_map( 'json_decode', $chunks );
        return [
            'processed_count' => count( $results ),
            'results'         => $results,
        ];
    }

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

    private function process_row( array $line ): array {
        // Your row processing logic here.
        return [ 'id' => $line[0], 'status' => 'ok' ];
    }
}

Αποθήκη τμημάτων API

Η κλάση ChunkStore διαχειρίζεται την αποθήκευση τμημάτων. Αποκτήστε πρόσβαση μέσω του Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Μέθοδος	Επιστρέφει	Περιγραφή
`get_chunks( int $job_id )`	`string[]`	Λαμβάνει όλα τα τμήματα μιας εργασίας, ταξινομημένα κατά δείκτη τμήματος
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Προσθέτει ένα τμήμα
`clear_chunks( int $job_id )`	`void`	Διαγράφει όλα τα τμήματα μιας εργασίας
`chunk_count( int $job_id )`	`int`	Μετρά τον αριθμό τμημάτων μιας εργασίας
`get_accumulated( int $job_id )`	`string`	Συνενώνει όλα τα τμήματα σε μία συμβολοσειρά

Κανονικά δεν χρειάζεται να καλέσετε απευθείας το ChunkStore. Ο worker χειρίζεται αυτόματα την αποθήκευση και την εκκαθάριση κατά την επεξεργασία ενός StreamingStep. Το API είναι διαθέσιμο για πιο προχωρημένες περιπτώσεις, όπως η επιθεώρηση τμημάτων υπό επεξεργασία ή η χειροκίνητη εκκαθάριση.