Langkah Streaming

Lihat sebagai Markdown Edit di GitHub Dokumentasi di 2cc7b2b

Langkah streaming memungkinkan alur kerja menghasilkan output secara bertahap. Setiap potongan yang dihasilkan langsung ditulis ke database, sehingga jika langkah gagal di tengah streaming, potongan yang sudah tersimpan tetap ada dan tersedia saat dicoba lagi.

Ini berguna untuk respons streaming LLM, pemrosesan file besar, dan skenario apa pun ketika progres parsial tidak boleh hilang.

Antarmuka StreamingStep

Langkah streaming mengimplementasikan 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	Deskripsi
`stream( $state, $existing_chunks )`	Generator yang menghasilkan potongan string. Setiap hasil langsung memicu penulisan ke DB.
`on_complete( $chunks, $state )`	Dipanggil setelah streaming selesai. Menerima semua potongan (sebelumnya + baru) dan mengembalikan data untuk digabungkan ke status alur kerja.
`config()`	Konfigurasi opsional. Mendukung kunci `max_attempts` dan `backoff`; `backoff` menerima nama strategi (`exponential`, `linear`, `fixed`), karena daftar jeda saat ini diabaikan (#16).

Cara potongan disimpan

Setiap kali generator Anda menghasilkan nilai, Queuety memanggil ChunkStore::append_chunk() untuk menulisnya ke tabel queuety_chunks. Tidak ada pengelompokan atau penyanggaan. Satu hasil berarti satu INSERT.

Baris potongan menyimpan job_id, workflow_id, step_index, chunk_index (dimulai dari nol), dan string content.

Kemampuan untuk dilanjutkan

Saat langkah streaming dicoba lagi setelah gagal, worker memuat semua potongan yang sebelumnya tersimpan dan meneruskannya sebagai parameter $existing_chunks. Implementasi Anda menentukan cara melanjutkan:

Lewati item yang sudah diproses dengan memeriksa jumlah potongan yang ada
Lanjutkan dari offset pada API berpaginasi dengan menggunakan jumlah potongan sebagai offset halaman
Mulai ulang dari awal dengan mengabaikan $existing_chunks sepenuhnya (potongan dari percobaan sebelumnya tetap ada di DB dan akan ditambahkan di awal)

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

Heartbeat selama streaming

Streaming yang berjalan lama harus mengirim heartbeat agar pendeteksi job usang tidak mengambil alih job. Panggil Heartbeat::beat() di dalam loop generator Anda:

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

Lihat Heartbeat untuk detail tentang API Heartbeat.

Pembersihan potongan

Saat streaming berhasil diselesaikan dan on_complete() mengembalikan hasil, worker memanggil ChunkStore::clear_chunks() untuk menghapus semua potongan untuk job tersebut. Hasil gabungan dari on_complete() menjadi catatan tahan lama. Potongan memang dirancang untuk bersifat sementara.

Jika langkah gagal dan dicoba lagi, potongan tidak dihapus. Potongan tersebut diteruskan kembali melalui $existing_chunks pada percobaan berikutnya.

Contoh lengkap: streaming 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 ];
    }
}

Gunakan dalam alur kerja seperti langkah lainnya:

PHP

use Queuety\Queuety;

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

Contoh lengkap: pemrosesan file

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

Kelas ChunkStore mengelola penyimpanan potongan. Akses melalui Queuety::chunk_store().

PHP

$store = Queuety::chunk_store();

Metode	Mengembalikan	Deskripsi
`get_chunks( int $job_id )`	`string[]`	Mengambil semua potongan untuk sebuah job, diurutkan berdasarkan indeks potongan
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Menambahkan satu potongan
`clear_chunks( int $job_id )`	`void`	Menghapus semua potongan untuk sebuah job
`chunk_count( int $job_id )`	`int`	Menghitung jumlah potongan untuk sebuah job
`get_accumulated( int $job_id )`	`string`	Menggabungkan semua potongan menjadi satu string

Biasanya Anda tidak perlu memanggil ChunkStore secara langsung. Worker menangani penyimpanan dan pembersihan secara otomatis saat memproses StreamingStep. API tersedia untuk kasus penggunaan tingkat lanjut, seperti memeriksa potongan yang sedang diproses atau melakukan pembersihan manual.