Kroky streamování

Zobrazit jako Markdown Upravit na GitHub Dokumentace na 2cc7b2b

Kroky streamování umožňují workflow postupně vytvářet výstup. Každá předaná část se ihned zapisuje do databáze, takže pokud krok během streamování selže, již uložené části zůstanou zachovány a budou k dispozici při opakování.

To se hodí pro streamované odpovědi LLM, zpracování velkých souborů a jakékoli situace, kdy se nemá ztratit částečný průběh.

Rozhraní StreamingStep

Krok streamování implementuje 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	Popis
`stream( $state, $existing_chunks )`	Generátor, který předává řetězcové části. Každé předání vyvolá okamžitý zápis do DB.
`on_complete( $chunks, $state )`	Volá se po dokončení streamu. Přijímá všechny části (předchozí i nové) a vrací data ke sloučení se stavem workflow.
`config()`	Volitelná konfigurace. Podporuje klíče `max_attempts` a `backoff`; `backoff` přijímá název strategie (`exponential`, `linear`, `fixed`), protože seznam prodlev se momentálně zahazuje (#16).

Jak se ukládají části

Pokaždé, když váš generátor předá hodnotu, Queuety zavolá ChunkStore::append_chunk() a zapíše ji do tabulky queuety_chunks. Žádné dávkování ani ukládání do vyrovnávací paměti neprobíhá. Jedno předání se rovná jednomu příkazu INSERT.

Řádek části ukládá hodnoty job_id, workflow_id, step_index, chunk_index (číslováno od nuly) a řetězec content.

Možnost obnovení

Při opakování kroku streamování po selhání načte pracovní proces všechny dříve uložené části a předá je jako parametr $existing_chunks. O způsobu obnovení rozhoduje vaše implementace:

Přeskočte již zpracované položky kontrolou počtu existujících částí
Posuňte se v stránkovaném API pomocí počtu částí jako posunu stránky
Začněte znovu od začátku úplným ignorováním $existing_chunks (části z předchozího pokusu jsou stále v DB a budou připojeny na začátek)

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

Pravidelné signály během streamování

Dlouhotrvající streamy by měly odesílat pravidelné signály, aby detektor nečinných úloh nepřevzal úlohu. Zavolejte Heartbeat::beat() uvnitř cyklu generátoru:

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 Heartbeat najdete v části Heartbeats API.

Čištění částí

Po úspěšném dokončení streamu a návratu on_complete() zavolá pracovní proces ChunkStore::clear_chunks() a odstraní všechny části dané úlohy. Sloučený výsledek z on_complete() je trvalým záznamem. Části jsou záměrně dočasné.

Pokud krok selže a opakuje se, části se neodstraní. Při dalším pokusu se předají zpět prostřednictvím $existing_chunks.

Kompletní příklad: streamování 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 ];
    }
}

Použijte jej ve workflow stejně jako jakýkoli jiný krok:

PHP

use Queuety\Queuety;

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

Kompletní příklad: zpracování souborů

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

Třída ChunkStore API

Třída ChunkStore spravuje ukládání částí. Získáte k ní přístup přes Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Metoda	Vrací	Popis
`get_chunks( int $job_id )`	`string[]`	Načte všechny části úlohy seřazené podle indexu části
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Přidá jednu část
`clear_chunks( int $job_id )`	`void`	Odstraní všechny části úlohy
`chunk_count( int $job_id )`	`int`	Spočítá počet částí úlohy
`get_accumulated( int $job_id )`	`string`	Spojí všechny části do jednoho řetězce

ChunkStore obvykle nemusíte volat přímo. Při zpracování StreamingStep zajišťuje pracovní proces ukládání i čištění automaticky. API je k dispozici pro pokročilé případy použití, například kontrolu právě zpracovávaných částí nebo ruční čištění.