Pași de transmitere în flux

Vizualizați ca Markdown Editați pe GitHub Documentație la 2cc7b2b

Pașii de transmitere în flux permit unui flux de lucru să producă rezultate treptat. Fiecare fragment generat este scris imediat în baza de date, astfel încât, dacă pasul eșuează în timpul transmiterii, fragmentele deja persistate se păstrează și sunt disponibile la reîncercare.

Acest lucru este util pentru răspunsurile transmise în flux de LLM, procesarea fișierelor mari și orice situație în care progresul parțial nu trebuie pierdut.

Interfața StreamingStep

Un pas de transmitere în flux implementează 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ă	Descriere
`stream( $state, $existing_chunks )`	Generator care produce fragmente de text. Fiecare valoare generată declanșează imediat o scriere în DB.
`on_complete( $chunks, $state )`	Se apelează după încheierea fluxului. Primește toate fragmentele (anterioare + noi) și returnează date de îmbinat cu starea fluxului de lucru.
`config()`	Configurare opțională. Acceptă cheile `max_attempts` și `backoff`; `backoff` primește un nume de strategie (`exponential`, `linear`, `fixed`), deoarece o listă de întârzieri este eliminată în prezent (#16).

Cum se persistă fragmentele

De fiecare dată când generatorul dumneavoastră produce o valoare, Queuety apelează ChunkStore::append_chunk() pentru a o scrie în tabelul queuety_chunks. Nu există grupare sau tamponare. O valoare generată înseamnă o instrucțiune INSERT.

Rândul fragmentului stochează job_id, workflow_id, step_index, chunk_index (indexare de la zero) și șirul content.

Posibilitatea de reluare

Când un pas de transmitere în flux este reîncercat după o eroare, lucrătorul încarcă toate fragmentele persistate anterior și le transmite ca parametru $existing_chunks. Implementarea dumneavoastră decide cum să reia:

Omiteți elementele deja procesate verificând numărul fragmentelor existente
Avansați într-un API paginat folosind numărul fragmentelor drept decalajul paginii
Reporniți de la zero ignorând complet $existing_chunks (fragmentele din încercarea anterioară sunt încă în DB și vor fi adăugate la început)

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

Semnale de puls în timpul transmiterii în flux

Fluxurile de lungă durată ar trebui să trimită semnale de puls pentru a împiedica detectorul joburilor învechite să recupereze jobul. Apelați Heartbeat::beat() în bucla generatorului:

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

Consultați Semnale de puls pentru detalii despre API de puls.

Curățarea fragmentelor

Când fluxul se încheie cu succes și on_complete() returnează un rezultat, lucrătorul apelează ChunkStore::clear_chunks() pentru a șterge toate fragmentele jobului. Rezultatul îmbinat din on_complete() este înregistrarea durabilă. Fragmentele sunt temporare prin proiectare.

Dacă pasul eșuează și este reîncercat, fragmentele nu sunt șterse. Ele sunt transmise din nou prin $existing_chunks la următoarea încercare.

Exemplu complet: transmitere în flux pentru 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 ];
    }
}

Folosiți-l într-un flux de lucru ca pe orice alt pas:

PHP

use Queuety\Queuety;

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

Exemplu complet: procesarea fișierelor

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

Metodele ChunkStore API

Clasa ChunkStore gestionează persistența fragmentelor. Accesați-o prin Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Metodă	Returnează	Descriere
`get_chunks( int $job_id )`	`string[]`	Preia toate fragmentele unui job, ordonate după indexul fragmentului
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Adaugă un singur fragment
`clear_chunks( int $job_id )`	`void`	Șterge toate fragmentele unui job
`chunk_count( int $job_id )`	`int`	Numără fragmentele unui job
`get_accumulated( int $job_id )`	`string`	Concatenează toate fragmentele într-un singur șir

În mod normal, nu este nevoie să apelați direct ChunkStore. Lucrătorul gestionează automat persistența și curățarea când procesează un StreamingStep. API este disponibil pentru cazuri avansate, precum inspectarea fragmentelor în curs sau curățarea manuală.