Suoratoistovaiheet

Näytä Markdown-muodossa Muokkaa palvelussa GitHub Dokumentaatio: 2cc7b2b

Suoratoistovaiheet antavat työnkulun tuottaa tulostetta vähitellen. Jokainen tuotettu osa kirjoitetaan heti tietokantaan, joten jos vaihe epäonnistuu suoratoiston aikana, jo tallennetut osat säilyvät ja ovat käytettävissä uudelleenyrityksessä.

Tästä on hyötyä LLM-suoratoistovastauksissa, suurten tiedostojen käsittelyssä ja kaikissa tilanteissa, joissa osittaista edistymistä ei pidä menettää.

StreamingStep-rajapinta

Suoratoistovaihe toteuttaa kohteen 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;
}

Metodi	Kuvaus
`stream( $state, $existing_chunks )`	Generaattori, joka tuottaa merkkijono-osia. Jokainen tuotto käynnistää välittömän tietokantakirjoituksen.
`on_complete( $chunks, $state )`	Kutsutaan suoratoiston päätyttyä. Saa kaikki osat (aiemmat ja uudet) ja palauttaa tiedot yhdistettäviksi työnkulun tilaan.
`config()`	Valinnaiset asetukset. Tukee avaimia `max_attempts` ja `backoff`; `backoff` saa strategiannimen (`exponential`, `linear`, `fixed`), sillä viiveiden luettelo hylätään tällä hetkellä (#16).

Osien tallennus

Aina kun generaattorisi tuottaa arvon, Queuety kutsuu kohdetta ChunkStore::append_chunk() kirjoittaakseen sen taulukkoon queuety_chunks. Eräkirjoitusta tai puskurointia ei ole. Yksi tuotto vastaa yhtä INSERT-lauseketta.

Osariville tallennetaan arvot job_id, workflow_id, step_index, chunk_index (nollasta alkava) ja merkkijono content.

Jatkaminen

Kun suoratoistovaihe yritetään uudelleen epäonnistumisen jälkeen, työntekijä lataa kaikki aiemmin tallennetut osat ja välittää ne parametrina $existing_chunks. Toteutuksesi päättää, miten suoritus jatkuu:

Ohita jo käsitellyt kohteet tarkistamalla olemassa olevien osien määrä
Siirry sivutetussa kohteessa API käyttämällä osien määrää sivun siirtymänä
Aloita alusta ohittamalla arvo $existing_chunks kokonaan (aiemman yrityksen osat ovat edelleen tietokannassa ja lisätään alkuun)

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

Sydämenlyönnit suoratoiston aikana

Pitkien suoratoistojen aikana tulee lähettää sydämenlyöntejä, jotta vanhentuneiden työtehtävien tunnistin ei ota tehtävää haltuunsa uudelleen. Kutsu kohdetta Heartbeat::beat() generaattorisilmukan sisällä:

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

Katso lisätietoja kohdasta Sydämenlyönnit ja Heartbeat-rajapinnasta API.

Osien siivous

Kun suoratoisto valmistuu onnistuneesti ja on_complete() palauttaa arvon, työntekijä kutsuu kohdetta ChunkStore::clear_chunks() poistaakseen kaikki työn tehtävän osat. Kohteen on_complete() yhdistetty tulos on pysyvä tietue. Osat ovat tarkoituksella väliaikaisia.

Jos vaihe epäonnistuu ja sitä yritetään uudelleen, osia ei poisteta. Ne välitetään seuraavalle yritykselle kohteen $existing_chunks kautta.

Täydellinen esimerkki: LLM-suoratoisto

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

Käytä sitä työnkulussa kuten mitä tahansa muuta vaihetta:

PHP

use Queuety\Queuety;

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

Täydellinen esimerkki: tiedoston käsittely

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

Luokka ChunkStore hallinnoi osien tallennusta. Käytä sitä kohteen Queuety::chunk_store() kautta.

PHP

$store = Queuety::chunk_store();

Metodi	Palauttaa	Kuvaus
`get_chunks( int $job_id )`	`string[]`	Hakee kaikki työn tehtävän osat osan indeksin mukaisessa järjestyksessä
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Lisää yhden osan
`clear_chunks( int $job_id )`	`void`	Poistaa kaikki työn tehtävän osat
`chunk_count( int $job_id )`	`int`	Laskee työn tehtävän osien määrän
`get_accumulated( int $job_id )`	`string`	Yhdistää kaikki osat yhdeksi merkkijonoksi

Yleensä kohdetta ChunkStore ei tarvitse kutsua suoraan. Työntekijä hoitaa tallennuksen ja siivouksen automaattisesti käsitellessään kohdetta StreamingStep. API on käytettävissä edistyneempiin käyttötapauksiin, kuten keskeneräisten osien tarkasteluun tai manuaaliseen siivoukseen.