Strömningssteg

Visa som Markdown Redigera på GitHub Dokumentation på 2cc7b2b

Strömningssteg låter ett arbetsflöde producera utdata stegvis. Varje del som avges skrivs omedelbart till databasen, så om steget misslyckas mitt i strömningen finns de redan lagrade delarna kvar och är tillgängliga vid omförsök.

Detta är användbart för strömmade LLM-svar, bearbetning av stora filer och alla scenarier där partiella framsteg inte får gå förlorade.

Gränssnittet StreamingStep

Ett strömningssteg implementerar 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;
}

Metod	Beskrivning
`stream( $state, $existing_chunks )`	Generator som avger strängdelar. Varje avropad yield utlöser omedelbar skrivning till DB.
`on_complete( $chunks, $state )`	Anropas när strömningen är klar. Tar emot alla delar (tidigare + nya) och returnerar data som ska slås samman med arbetsflödestillståndet.
`config()`	Valfri konfiguration. Stöder nycklarna `max_attempts` och `backoff`; `backoff` tar ett strateginamn (`exponential`, `linear`, `fixed`), eftersom en lista med fördröjningar för närvarande tas bort (#16).

Så lagras delar

Varje gång din generator avger ett värde anropar Queuety ChunkStore::append_chunk() för att skriva det till tabellen queuety_chunks. Ingen samling eller buffring sker. En yield motsvarar en INSERT.

Raden för delen lagrar job_id, workflow_id, step_index, chunk_index (index från noll) och strängen content.

Återupptagning

När ett strömningssteg görs om efter ett misslyckande läser arbetaren in alla tidigare lagrade delar och skickar dem som parametern $existing_chunks. Din implementation bestämmer hur strömningen ska återupptas:

Hoppa över redan bearbetade objekt genom att kontrollera antalet befintliga delar
Hoppa till en sida i en paginerad API genom att använda antalet delar som sidförskjutning
Börja om från början genom att ignorera $existing_chunks helt (delar från föregående försök finns fortfarande i DB och läggs till först)

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

Hjärtljud under strömning

Långvariga strömningar bör skicka hjärtljud för att förhindra att detektorn för inaktuella jobb återtar jobbet. Anropa Heartbeat::beat() i generatorslingan:

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 Hjärtljud för information om gränssnittet Heartbeat API.

Rensa delar

När strömningen slutförs utan fel och on_complete() returnerar anropar arbetaren ChunkStore::clear_chunks() för att ta bort alla delar för jobbet. Det sammanfogade resultatet från on_complete() är den beständiga posten. Delarna är avsiktligt tillfälliga.

Om steget misslyckas och görs om tas delarna inte bort. De skickas tillbaka via $existing_chunks vid nästa försök.

Fullständigt exempel: LLM-strömning

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

Använd det i ett arbetsflöde som vilket annat steg som helst:

PHP

use Queuety\Queuety;

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

Fullständigt exempel: filbearbetning

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 hanterar lagringen av delar. Kom åt den via Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Metod	Returnerar	Beskrivning
`get_chunks( int $job_id )`	`string[]`	Hämtar alla delar för ett jobb, sorterade efter delindex
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Lägger till en enskild del
`clear_chunks( int $job_id )`	`void`	Tar bort alla delar för ett jobb
`chunk_count( int $job_id )`	`int`	Räknar antalet delar för ett jobb
`get_accumulated( int $job_id )`	`string`	Slår samman alla delar till en sträng

Du behöver vanligtvis inte anropa ChunkStore direkt. Arbetaren hanterar lagring och rensning automatiskt när den bearbetar en StreamingStep. API är tillgänglig för avancerade användningsfall, till exempel granskning av delar under pågående strömning eller manuell rensning.