Passaggi di streaming

Visualizza come Markdown Modifica su GitHub Documentazione di 2cc7b2b

I passaggi di streaming consentono a un flusso di lavoro di produrre output in modo incrementale. Ogni blocco generato viene scritto subito nel database, così, se il passaggio si interrompe durante lo streaming, i blocchi già salvati restano disponibili al nuovo tentativo.

È utile per le risposte in streaming degli LLM, l'elaborazione di file di grandi dimensioni e ogni scenario in cui non si vuole perdere il lavoro parziale.

L'interfaccia StreamingStep

Un passaggio di streaming implementa 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;
}

Metodo	Descrizione
`stream( $state, $existing_chunks )`	Generatore che produce blocchi di stringhe. Ogni blocco generato attiva una scrittura immediata nel DB.
`on_complete( $chunks, $state )`	Chiamato al termine dello stream. Riceve tutti i blocchi (precedenti e nuovi) e restituisce i dati da unire allo stato del workflow.
`config()`	Configurazione opzionale. Supporta le chiavi `max_attempts` e `backoff`; `backoff` accetta il nome di una strategia (`exponential`, `linear`, `fixed`), poiché un elenco di ritardi viene attualmente ignorato (#16).

Come vengono salvati i blocchi

Ogni volta che il generatore produce un valore, Queuety chiama ChunkStore::append_chunk() per scriverlo nella tabella queuety_chunks. Non viene effettuato alcun raggruppamento o buffering. Ogni valore generato corrisponde a un INSERT.

La riga del blocco contiene job_id, workflow_id, step_index, chunk_index (con indice a partire da zero) e la stringa content.

Ripresa

Quando un passaggio di streaming viene ritentato dopo un errore, il worker carica tutti i blocchi salvati in precedenza e li passa come parametro $existing_chunks. Spetta alla vostra implementazione decidere come riprendere:

Saltare gli elementi già elaborati controllando il numero di blocchi esistenti
Usare il numero di blocchi come offset della pagina per scorrere un API paginato
Ricominciare da zero ignorando del tutto $existing_chunks (i blocchi del tentativo precedente sono ancora nel DB e verranno anteposti)

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 durante lo streaming

Gli stream di lunga durata dovrebbero inviare heartbeat per impedire al rilevatore di job obsoleti di riprendere il job. Chiamate Heartbeat::beat() all'interno del ciclo del generatore:

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

Per i dettagli, consultate Heartbeat del API.

Pulizia dei blocchi

Quando lo stream termina correttamente e on_complete() restituisce un risultato, il worker chiama ChunkStore::clear_chunks() per eliminare tutti i blocchi del job. Il risultato unito di on_complete() costituisce il record durevole. I blocchi sono temporanei per progettazione.

Se il passaggio fallisce e viene ritentato, i blocchi non vengono eliminati. Vengono passati di nuovo tramite $existing_chunks al tentativo successivo.

Esempio completo: streaming 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 ];
    }
}

Usatelo in un flusso di lavoro come qualsiasi altro passaggio:

PHP

use Queuety\Queuety;

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

Esempio completo: elaborazione di file

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

La classe ChunkStore gestisce il salvataggio dei blocchi. Vi si accede tramite Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Metodo	Restituisce	Descrizione
`get_chunks( int $job_id )`	`string[]`	Recupera tutti i blocchi di un job, ordinati per indice del blocco
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Aggiunge un singolo blocco
`clear_chunks( int $job_id )`	`void`	Elimina tutti i blocchi di un job
`chunk_count( int $job_id )`	`int`	Conta il numero di blocchi di un job
`get_accumulated( int $job_id )`	`string`	Concatena tutti i blocchi in un'unica stringa

Di norma non è necessario chiamare direttamente ChunkStore. Il worker gestisce automaticamente il salvataggio e la pulizia durante l'elaborazione di un StreamingStep. API è disponibile per casi d'uso avanzati, ad esempio per esaminare i blocchi in corso o eseguire una pulizia manuale.