Streamingtrin

Vis som Markdown Rediger på GitHub Dokumentation hos 2cc7b2b

Streamingtrin lader et workflow producere output løbende. Hver udsendt chunk skrives straks til databasen, så hvis trinnet fejler midt i streamen, bevares de allerede lagrede chunks og er tilgængelige ved et nyt forsøg.

Det er nyttigt til LLM-streamingsvar, behandling af store filer og alle situationer, hvor delvise fremskridt ikke må gå tabt.

StreamingStep-grænsefladen

Et streamingtrin implementerer 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;
}

Metode	Beskrivelse
`stream( $state, $existing_chunks )`	Generator, der udsender string-chunks. Hver udsendelse udløser straks en skrivning til databasen.
`on_complete( $chunks, $state )`	Kaldes, når streamen er færdig. Modtager alle chunks (tidligere og nye) og returnerer data, der flettes ind i workflowtilstanden.
`config()`	Valgfri konfiguration. Understøtter nøglerne `max_attempts` og `backoff`; `backoff` tager et strateginavn (`exponential`, `linear`, `fixed`), da en liste med forsinkelser i øjeblikket kasseres (#16).

Sådan lagres chunks

Hver gang din generator udsender en værdi, kalder Queuety ChunkStore::append_chunk() for at skrive den til tabellen queuety_chunks. Der er ingen batching eller buffering. Én udsendelse svarer til én INSERT.

Chuncrækken gemmer job_id, workflow_id, step_index, chunk_index (nulbaseret) og strengen content.

Mulighed for genoptagelse

Når et streamingtrin forsøges igen efter en fejl, indlæser workeren alle tidligere lagrede chunks og sender dem som parameteren $existing_chunks. Din implementering afgør, hvordan den skal fortsætte:

Spring allerede behandlede elementer over ved at kontrollere antallet af eksisterende chunks
Spring frem i et sideinddelt API ved at bruge chunckantallet som sideforskydning
Start forfra ved helt at ignorere $existing_chunks (chunks fra det tidligere forsøg ligger stadig i databasen og føjes foran)

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 under streaming

Langvarige streams bør sende heartbeats for at forhindre, at detektoren for forældede jobs overtager jobbet. Kald Heartbeat::beat() inde i din generatorløkke:

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

Se Heartbeats for detaljer om Heartbeat API.

Oprydning af chunks

Når streamen afsluttes uden fejl, og on_complete() returnerer, kalder workeren ChunkStore::clear_chunks() for at slette alle chunks for jobbet. Det flettede resultat fra on_complete() er den permanente registrering. Chunks er midlertidige af design.

Hvis trinnet fejler og forsøges igen, slettes chunks ikke. De sendes tilbage via $existing_chunks ved næste forsøg.

Fuldt eksempel: 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 ];
    }
}

Brug det i et workflow som ethvert andet trin:

PHP

use Queuety\Queuety;

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

Fuldt eksempel: filbehandling

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

ChunkStore API

Klassen ChunkStore håndterer lagring af chunks. Få adgang til den via Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Metode	Returnerer	Beskrivelse
`get_chunks( int $job_id )`	`string[]`	Henter alle chunks for et job, sorteret efter chunkindeks
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Tilføjer én enkelt chunk
`clear_chunks( int $job_id )`	`void`	Sletter alle chunks for et job
`chunk_count( int $job_id )`	`int`	Tæller antallet af chunks for et job
`get_accumulated( int $job_id )`	`string`	Sammenkæder alle chunks til én streng

Du behøver normalt ikke kalde ChunkStore direkte. Workeren håndterer automatisk lagring og oprydning, når den behandler et StreamingStep. API er tilgængelig til avancerede anvendelser som inspektion af igangværende chunks eller manuel oprydning.