Darbplūsmas straumēšanas darbības

Skatīt kā Markdown Rediģēt vietnē GitHub Dokumentācija vietnē 2cc7b2b

Straumēšanas darbības ļauj darbplūsmai pakāpeniski ģenerēt izvadi. Katrs nodotais fragments tiek uzreiz ierakstīts datubāzē, tāpēc, ja darbība neizdodas straumēšanas laikā, jau saglabātie fragmenti tiek saglabāti un ir pieejami atkārtotā mēģinājumā.

Tas noder LLM straumēšanas atbildēm, lielu failu apstrādei un jebkurā situācijā, kad nedrīkst zaudēt daļēju progresu.

StreamingStep saskarne

Straumēšanas darbība implementē 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;
}

Metode	Apraksts
`stream( $state, $existing_chunks )`	Ģenerators, kas nodod virknes fragmentus. Katra nodošana izraisa tūlītēju DB ierakstu.
`on_complete( $chunks, $state )`	Tiek izsaukta pēc straumes pabeigšanas. Saņem visus fragmentus (iepriekšējos un jaunos) un atgriež datus, ko apvienot ar darbplūsmas stāvokli.
`config()`	Neobligāta konfigurācija. Atbalsta atslēgas `max_attempts` un `backoff`; `backoff` pieņem stratēģijas nosaukumu (`exponential`, `linear`, `fixed`), jo aizkavju saraksts pašlaik tiek atmests (#16).

Kā tiek saglabāti fragmenti

Katru reizi, kad ģenerators nodod vērtību, Queuety izsauc ChunkStore::append_chunk(), lai ierakstītu to tabulā queuety_chunks. Netiek veikta pakešu apstrāde vai buferēšana. Viena nodošana atbilst vienam INSERT.

Fragmenta rindā tiek saglabātas vērtības job_id, workflow_id, step_index, chunk_index (sākot no nulles) un virkne content.

Izpildes atsākšana

Ja straumēšanas darbību mēģina atkārtoti pēc kļūmes, izpildītājs ielādē visus iepriekš saglabātos fragmentus un nodod tos kā parametru $existing_chunks. Jūs izvēlaties, kā atsākt darbu:

Izlaist jau apstrādātos vienumus, pārbaudot esošo fragmentu skaitu
Pāriet uz lappusēm sadalītā API nobīdi, izmantojot fragmentu skaitu kā lappuses nobīdi
Sākt no jauna, pilnībā ignorējot $existing_chunks (iepriekšējā mēģinājuma fragmenti joprojām ir DB un tiks pievienoti sākumā)

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

Sirdspuksti straumēšanas laikā

Ilgstošām straumēm jāsūta sirdspuksti, lai novērstu novecojušu darbu noteicēja mēģinājumus pārņemt darbu. Izsauciet Heartbeat::beat() ģeneratora ciklā:

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

Plašāku informāciju par Heartbeat API skatiet sadaļā Sirdspuksti.

Fragmentu tīrīšana

Kad straume tiek veiksmīgi pabeigta un on_complete() atgriežas, izpildītājs izsauc ChunkStore::clear_chunks(), lai dzēstu visus darba fragmentus. Apvienotais rezultāts no on_complete() ir pastāvīgais ieraksts. Fragmenti pēc konstrukcijas ir īslaicīgi.

Ja darbība neizdodas un tiek mēģināta atkārtoti, fragmenti netiek dzēsti. Nākamajā mēģinājumā tie tiek nodoti, izmantojot $existing_chunks.

Pilns piemērs: LLM straumēšana

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

Izmantojiet to darbplūsmā tāpat kā jebkuru citu darbību:

PHP

use Queuety\Queuety;

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

Pilns piemērs: failu apstrāde

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

Klase ChunkStore pārvalda fragmentu saglabāšanu. Tai var piekļūt, izmantojot Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Metode	Atgriež	Apraksts
`get_chunks( int $job_id )`	`string[]`	Iegūst visus darba fragmentus, sakārtotus pēc fragmenta indeksa
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Pievieno vienu fragmentu
`clear_chunks( int $job_id )`	`void`	Dzēš visus darba fragmentus
`chunk_count( int $job_id )`	`int`	Saskaita darba fragmentus
`get_accumulated( int $job_id )`	`string`	Savieno visus fragmentus vienā virknē

Parasti ChunkStore nav jāizsauc tieši. Apstrādājot StreamingStep, izpildītājs automātiski parūpējas par saglabāšanu un tīrīšanu. API ir pieejams īpašiem lietojumiem, piemēram, apstrādes gaitā esošo fragmentu pārbaudei vai manuālai tīrīšanai.