Streamingstappen

Bekijken als Markdown Bewerken op GitHub Documentatie op 2cc7b2b

Met streamingstappen kan een workflow incrementeel output produceren. Elke opgeleverde chunk wordt direct naar de database geschreven, zodat de al opgeslagen chunks behouden blijven en beschikbaar zijn bij een nieuwe poging als de stap halverwege de stream mislukt.

Dit is nuttig voor streamingreacties van LLM's, het verwerken van grote bestanden en elk scenario waarin gedeeltelijke voortgang niet verloren mag gaan.

De StreamingStep-interface

Een streamingstap implementeert 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;
}

Methode	Beschrijving
`stream( $state, $existing_chunks )`	Generator die stringchunks oplevert. Elke yield leidt direct tot een DB-schrijfactie.
`on_complete( $chunks, $state )`	Wordt aangeroepen nadat de stream is voltooid. Ontvangt alle chunks (vorige + nieuwe) en retourneert gegevens om samen te voegen met de workflowstatus.
`config()`	Optionele configuratie. Ondersteunt de sleutels `max_attempts` en `backoff`; `backoff` verwacht een strategienaam (`exponential`, `linear`, `fixed`), omdat een lijst met vertragingen momenteel wordt genegeerd (#16).

Hoe chunks worden opgeslagen

Elke keer dat je generator een waarde oplevert, roept Queuety ChunkStore::append_chunk() aan om deze naar de tabel queuety_chunks te schrijven. Er wordt niets gebundeld of gebufferd. Eén yield staat gelijk aan één INSERT.

De chunkregel slaat de job_id, workflow_id, step_index, chunk_index (vanaf nul) en de string content op.

Hervatbaarheid

Wanneer een streamingstap na een fout opnieuw wordt geprobeerd, laadt de worker alle eerder opgeslagen chunks en geeft die door als de parameter $existing_chunks. Je implementatie bepaalt hoe het hervatten verloopt:

Sla al verwerkte items over door het aantal bestaande chunks te controleren
Verspring in een gepagineerde API met het aantal chunks als pagina-offset
Begin helemaal opnieuw door $existing_chunks volledig te negeren (chunks van de vorige poging staan nog in de DB en worden ervoor geplaatst)

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

Heartbeats tijdens het streamen

Langlopende streams moeten heartbeats versturen om te voorkomen dat de detector voor verouderde taken de taak opnieuw claimt. Roep Heartbeat::beat() aan in je generatorlus:

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

Zie Heartbeats voor meer informatie over de Heartbeat API.

Chunks opruimen

Wanneer de stream succesvol is voltooid en on_complete() retourneert, roept de worker ChunkStore::clear_chunks() aan om alle chunks voor de taak te verwijderen. Het samengevoegde resultaat van on_complete() is het duurzame record. Chunks zijn bewust tijdelijk.

Als de stap mislukt en opnieuw wordt geprobeerd, worden de chunks niet verwijderd. Bij de volgende poging worden ze opnieuw doorgegeven via $existing_chunks.

Volledig voorbeeld: LLM-streaming

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

Gebruik dit in een workflow zoals elke andere stap:

PHP

use Queuety\Queuety;

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

Volledig voorbeeld: bestandsverwerking

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

De API-klasse

De klasse ChunkStore beheert de opslag van chunks. Je krijgt er toegang toe via Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Methode	Retourneert	Beschrijving
`get_chunks( int $job_id )`	`string[]`	Haalt alle chunks voor een taak op, gesorteerd op chunkindex
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Voegt één chunk toe
`clear_chunks( int $job_id )`	`void`	Verwijdert alle chunks voor een taak
`chunk_count( int $job_id )`	`int`	Telt het aantal chunks voor een taak
`get_accumulated( int $job_id )`	`string`	Voegt alle chunks samen tot één string

Normaal gesproken hoef je ChunkStore niet rechtstreeks aan te roepen. De worker regelt het opslaan en opruimen automatisch bij het verwerken van een StreamingStep. API is beschikbaar voor geavanceerde gebruikssituaties, zoals het inspecteren van chunks die nog worden verwerkt of handmatig opruimen.