Kroky so streamovaním

Zobraziť ako Markdown Upraviť na GitHub Dokumentácia na 2cc7b2b

Kroky so streamovaním umožňujú pracovným postupom vytvárať výstup postupne. Každá odovzdaná časť sa okamžite zapíše do databázy, takže ak krok zlyhá uprostred streamovania, už uložené časti zostanú zachované a budú dostupné pri opakovanom pokuse.

Hodí sa to pri streamovaných odpovediach LLM, spracovaní veľkých súborov a v každom prípade, keď sa nesmie stratiť čiastočný pokrok.

Rozhranie StreamingStep

Krok so streamovaním 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;
}

Metóda	Opis
`stream( $state, $existing_chunks )`	Generátor, ktorý odovzdáva textové časti. Každé odovzdanie sa okamžite zapíše do DB.
`on_complete( $chunks, $state )`	Zavolá sa po skončení streamu. Dostane všetky časti (predchádzajúce aj nové) a vráti údaje, ktoré sa zlúčia so stavom pracovného postupu.
`config()`	Voliteľná konfigurácia. Podporuje kľúče `max_attempts` a `backoff`; `backoff` prijíma názov stratégie (`exponential`, `linear`, `fixed`), pretože zoznam oneskorení sa momentálne zahadzuje (#16).

Ako sa ukladajú časti

Vždy, keď váš generátor odovzdá hodnotu, Queuety zavolá ChunkStore::append_chunk() a zapíše ju do tabuľky queuety_chunks. Nedochádza k dávkovaniu ani vyrovnávacej pamäti. Jedno odovzdanie znamená jeden príkaz INSERT.

Záznam časti obsahuje job_id, workflow_id, step_index, chunk_index (od nuly) a textový reťazec content.

Možnosť obnovenia

Pri opakovanom spustení kroku so streamovaním po zlyhaní pracovník načíta všetky predtým uložené časti a odovzdá ich ako parameter $existing_chunks. Vaša implementácia určí, ako pokračovať:

Preskočiť už spracované položky kontrolou počtu existujúcich častí
Preskočiť na príslušnú pozíciu v stránkovanom API pomocou počtu častí ako posunu stránky
Začať od začiatku úplným ignorovaním $existing_chunks (časti z predchádzajúceho pokusu zostávajú v DB a pripoja sa na začiatok)

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

Signály aktivity počas streamovania

Dlho trvajúce streamy by mali odosielať signály aktivity, aby detektor neaktívnych úloh úlohu neprevzal. Zavolajte Heartbeat::beat() vo vnútri cyklu generátora:

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 rozhraní Heartbeat nájdete v téme Signály aktivity; pozrite si rozhranie API.

Čistenie častí

Keď sa stream úspešne dokončí a on_complete() vráti výsledok, pracovník zavolá ChunkStore::clear_chunks() a odstráni všetky časti úlohy. Zlúčený výsledok z on_complete() je trvalým záznamom. Časti sú zámerne dočasné.

Ak krok zlyhá a spustí sa znova, časti sa neodstránia. Pri ďalšom pokuse sa odovzdajú cez $existing_chunks.

Úplný príklad: streamovanie 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žite ho v pracovnom postupe ako ktorýkoľvek iný krok:

PHP

use Queuety\Queuety;

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

Úplný príklad: spracovanie súboru

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

API ChunkStore

Trieda ChunkStore spravuje ukladanie častí. Pristupujte k nej cez Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Metóda	Vráti	Opis
`get_chunks( int $job_id )`	`string[]`	Načíta všetky časti úlohy zoradené podľa indexu časti
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Pripojí jednu časť
`clear_chunks( int $job_id )`	`void`	Odstráni všetky časti úlohy
`chunk_count( int $job_id )`	`int`	Spočíta počet častí úlohy
`get_accumulated( int $job_id )`	`string`	Spojí všetky časti do jedného reťazca

ChunkStore zvyčajne nemusíte volať priamo. Worker sa pri spracovaní StreamingStep automaticky postará o uchovávanie údajov a čistenie. API je k dispozícii na pokročilé použitie, napríklad na kontrolu práve spracúvaných častí alebo manuálne čistenie.