Koraki pretakanja

Prikaži kot Markdown Uredi na GitHub Dokumentacija za 2cc7b2b

Koraki pretakanja omogočajo, da potek dela ustvarja izhod sproti. Vsak oddani kos se takoj zapiše v zbirko podatkov, zato se ob napaki koraka med pretakanjem že shranjeni kosi ohranijo in so na voljo pri ponovnem poskusu.

To je uporabno za pretočne odgovore LLM-jev, obdelavo velikih datotek in vse primere, v katerih se delni napredek ne sme izgubiti.

Vmesnik StreamingStep

Korak pretakanja implementira 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;
}

Metoda	Opis
`stream( $state, $existing_chunks )`	Generator, ki oddaja nize v kosih. Vsak oddani kos sproži takojšen zapis v DB.
`on_complete( $chunks, $state )`	Pokliče se po koncu pretakanja. Prejme vse kose (prejšnje in nove) ter vrne podatke za združitev s stanjem poteka dela.
`config()`	Izbirna konfiguracija. Podpira ključe `max_attempts` in `backoff`; `backoff` sprejme ime strategije (`exponential`, `linear`, `fixed`), saj je seznam zakasnitev trenutno zavržen (#16).

Kako se kosi shranjujejo

Vsakič, ko generator odda vrednost, Queuety pokliče ChunkStore::append_chunk(), da jo zapiše v tabelo queuety_chunks. Združevanja ali medpomnjenja ni. Ena oddaja pomeni en INSERT.

Vrstica kosa shrani job_id, workflow_id, step_index, chunk_index (od nič naprej) in niz content.

Možnost nadaljevanja

Ko se korak pretakanja po napaki ponovi, izvajalnik naloži vse prej shranjene kose in jih posreduje kot parameter $existing_chunks. Vaša implementacija določa, kako se nadaljuje:

Preskočite že obdelane elemente tako, da preverite število obstoječih kosov
Pomaknite se na ustrezno mesto v straničenem API tako, da uporabite število kosov kot odmik strani
Začnite znova od začetka tako, da v celoti prezrete $existing_chunks (kosi iz prejšnjega poskusa so še vedno v DB in bodo dodani na začetek)

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

Srčni utripi med pretakanjem

Dolgotrajni tokovi bi morali pošiljati srčne utripe, da zastarela opravila ne prevzame zaznavalnik. V zanki generatorja pokličite 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;
    }
}

Podrobnosti o vmesniku Heartbeat API najdete v razdelku Srčni utripi.

Čiščenje kosov

Ko se pretakanje uspešno zaključi in on_complete() vrne vrednost, izvajalnik pokliče ChunkStore::clear_chunks(), da izbriše vse kose za opravilo. Združeni rezultat iz on_complete() je trajni zapis. Kosi so po zasnovi začasni.

Če korak odpove in se ponovi, se kosi ne izbrišejo. Pri naslednjem poskusu se znova posredujejo prek $existing_chunks.

Celoten primer: pretakanje 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 ];
    }
}

Uporabite ga v poteku dela kot kateri koli drug korak:

PHP

use Queuety\Queuety;

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

Celoten primer: obdelava datoteke

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

Razred ChunkStore upravlja shranjevanje kosov. Dostopajte do njega prek Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Metoda	Vrne	Opis
`get_chunks( int $job_id )`	`string[]`	Pridobi vse kose za opravilo, urejene po indeksu kosa
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Doda en kos
`clear_chunks( int $job_id )`	`void`	Izbriše vse kose za opravilo
`chunk_count( int $job_id )`	`int`	Prešteje število kosov za opravilo
`get_accumulated( int $job_id )`	`string`	Združi vse kose v en niz

Običajno vam metode ChunkStore ni treba klicati neposredno. Izvajalnik samodejno poskrbi za shranjevanje in čiščenje pri obdelavi StreamingStep. API je na voljo za naprednejše primere uporabe, kot sta pregledovanje še nedokončanih kosov ali ročno čiščenje.