Srautiniai veiksmai

Peržiūrėti kaip Markdown Redaguoti per GitHub Dokumentacija: 2cc7b2b

Srautiniai veiksmai leidžia darbo sekai kurti išvestį dalimis. Kiekviena pateikta dalis iškart įrašoma į duomenų bazę, todėl, jei veiksmas nepavyksta srautiniam perdavimui dar nepasibaigus, jau išsaugotos dalys išlieka ir jas galima naudoti bandant dar kartą.

Tai naudinga LLM srautinio atsakymo, didelių failų apdorojimo ir bet kokiais kitais atvejais, kai dalinės pažangos nereikėtų prarasti.

StreamingStep sąsaja

Srautinis veiksmas įgyvendina 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;
}

Metodas	Aprašymas
`stream( $state, $existing_chunks )`	Generatorius, pateikiantis eilutės dalis. Kiekviena pateikta dalis iškart įrašoma į DB.
`on_complete( $chunks, $state )`	Iškviečiamas srautiniam perdavimui pasibaigus. Gauna visas dalis (ankstesnes ir naujas) ir grąžina duomenis, kurie sujungiami su darbo sekos būsena.
`config()`	Pasirenkama konfigūracija. Palaiko raktus `max_attempts` ir `backoff`; `backoff` priima strategijos pavadinimą (`exponential`, `linear`, `fixed`), nes delsų sąrašas šiuo metu atmetamas (#16).

Kaip išsaugomos dalys

Kiekvieną kartą, kai generatorius pateikia reikšmę, Queuety iškviečia ChunkStore::append_chunk() ir įrašo ją į lentelę queuety_chunks. Duomenys negrupuojami ir nekaupiami buferyje. Viena pateikta dalis atitinka vieną INSERT.

Dalos eilutėje saugomos reikšmės job_id, workflow_id, step_index, chunk_index (skaičiuojant nuo nulio) ir eilutė content.

Tęsimas

Pakartotinai vykdant srautinį veiksmą po nesėkmės, darbuotojas įkelia visas anksčiau išsaugotas dalis ir perduoda jas kaip parametrą $existing_chunks. Jūsų realizacija nusprendžia, kaip tęsti:

Praleiskite jau apdorotus elementus tikrindami esamų dalių skaičių
Nurodykite vietą puslapiu suskaidytame API naudodami dalių skaičių kaip puslapio poslinkį
Pradėkite iš naujo visiškai ignoruodami $existing_chunks (ankstesnio bandymo dalys vis tiek lieka DB ir bus pateiktos pradžioje)

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

Širdies plakimo signalai perduodant srautą

Ilgai veikiantys srautai turėtų siųsti širdies plakimo signalus, kad pasenusių užduočių aptikimo priemonė neatimtų užduoties. Iškvieskite Heartbeat::beat() generatoriaus cikle:

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

Daugiau informacijos rasite puslapyje Širdies plakimo signalai apie metodą Heartbeat API.

Dalių išvalymas

Sėkmingai baigus srautinį perdavimą ir grąžinus on_complete(), darbuotojas iškviečia ChunkStore::clear_chunks(), kad ištrintų visas užduoties dalis. Sujungtas on_complete() rezultatas yra patikimas įrašas. Dalys pagal paskirtį yra laikinos.

Jei veiksmas nepavyksta ir paleidžiamas iš naujo, dalys neištrinamos. Kito bandymo metu jos perduodamos per $existing_chunks.

Visas pavyzdys: LLM srautinis perdavimas

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

Naudokite tai darbo sekoje kaip ir bet kurį kitą veiksmą:

PHP

use Queuety\Queuety;

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

Visas pavyzdys: failų apdorojimas

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

Klasė ChunkStore tvarko dalių išsaugojimą. Ją pasieksite per Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Metodas	Grąžina	Aprašymas
`get_chunks( int $job_id )`	`string[]`	Grąžina visas užduoties dalis, surikiuotas pagal dalies indeksą
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Prideda vieną dalį
`clear_chunks( int $job_id )`	`void`	Ištrina visas užduoties dalis
`chunk_count( int $job_id )`	`int`	Suskaičiuoja užduoties dalis
`get_accumulated( int $job_id )`	`string`	Sujungia visas dalis į vieną eilutę

Paprastai ChunkStore tiesiogiai kviesti nereikia. Darbinis procesas automatiškai pasirūpina duomenų išsaugojimu ir išvalymu, kai apdoroja StreamingStep. API galima naudoti pažangesniais atvejais, pavyzdžiui, norint peržiūrėti dar apdorojamas dalis arba išvalyti jas rankiniu būdu.