Akış Adımları

Markdown olarak görüntüle GitHub üzerinde düzenle Dokümantasyon: 2cc7b2b

Akış adımları, bir iş akışının çıktıyı parça parça üretmesini sağlar. Oluşturulan her parça hemen veritabanına yazılır; böylece adım akış sırasında başarısız olursa önceden kalıcılaştırılan parçalar korunur ve yeniden denemede kullanılabilir.

Bu özellik, LLM akış yanıtları, büyük dosya işleme ve kısmi ilerlemenin kaybolmaması gereken her senaryo için kullanışlıdır.

StreamingStep arayüzü

Bir akış adımı Queuety\Contracts\StreamingStep arayüzünü uygular:

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

Yöntem	Açıklama
`stream( $state, $existing_chunks )`	Metin parçaları üreten bir üreteçtir. Her yield anında bir DB yazma işlemi başlatır.
`on_complete( $chunks, $state )`	Akış tamamlandıktan sonra çağrılır. Tüm parçaları (öncekiler + yeniler) alır ve iş akışı durumuyla birleştirilecek verileri döndürür.
`config()`	İsteğe bağlı yapılandırma. `max_attempts` ve `backoff` anahtarlarını destekler; `backoff` bir strateji adı (`exponential`, `linear`, `fixed`) alır; çünkü gecikme listesi şu anda yok sayılır (#16).

Parçaların kalıcılaştırılması

Üreteciniz bir değer ürettiğinde Queuety, değeri queuety_chunks tablosuna yazmak için ChunkStore::append_chunk() çağrısını yapar. Gruplama veya arabelleğe alma yoktur. Her yield bir INSERT işlemine karşılık gelir.

Parça satırı job_id, workflow_id, step_index, sıfır tabanlı chunk_index değerini ve content metnini saklar.

Devam ettirilebilirlik

Akış adımı bir başarısızlıktan sonra yeniden denendiğinde çalışan, önceden kalıcılaştırılmış tüm parçaları yükler ve $existing_chunks parametresi olarak iletir. Uygulamanız devam etme biçimini belirler:

Mevcut parçaların sayısını kontrol ederek önceden işlenmiş öğeleri atlayın
Parça sayısını sayfa konumu olarak kullanıp sayfalandırılmış bir API içinde ilerleyin
$existing_chunks değerini tamamen yok sayarak baştan başlayın (önceki denemedeki parçalar hâlâ DB'dedir ve başa eklenir)

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

Akış sırasında kalp atışları

Uzun süren akışlar, güncel olmayan iş algılayıcısının işi geri almasını önlemek için kalp atışları göndermelidir. Üreteç döngünüzün içinde Heartbeat::beat() çağrısını yapın:

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

Ayrıntılar için Kalp Atışları bölümüne ve Heartbeat API arayüzüne bakın.

Parça temizliği

Akış başarıyla tamamlanıp on_complete() döndürdüğünde çalışan, işe ait tüm parçaları silmek için ChunkStore::clear_chunks() çağrısını yapar. on_complete() ile birleştirilen sonuç kalıcı kayıttır. Parçalar tasarım gereği geçicidir.

Adım başarısız olup yeniden denendiğinde parçalar silinmez. Sonraki denemede $existing_chunks aracılığıyla geri iletilirler.

Tam örnek: LLM akışı

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

Diğer adımlar gibi bunu da bir iş akışında kullanın:

PHP

use Queuety\Queuety;

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

Tam örnek: dosya işleme

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

ChunkStore sınıfı, parçaların kalıcılaştırılmasını yönetir. Erişmek için Queuety::chunk_store() kullanın.

PHP

$store = Queuety::chunk_store();

Yöntem	Döndürdüğü değer	Açıklama
`get_chunks( int $job_id )`	`string[]`	Bir işin tüm parçalarını getirir ve parça dizinine göre sıralar
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Tek bir parça ekler
`clear_chunks( int $job_id )`	`void`	Bir işe ait tüm parçaları siler
`chunk_count( int $job_id )`	`int`	Bir işin parça sayısını döndürür
`get_accumulated( int $job_id )`	`string`	Tüm parçaları tek bir metinde birleştirir

Normalde ChunkStore yöntemini doğrudan çağırmanız gerekmez. Çalışan, bir StreamingStep işlerken kalıcılaştırma ve temizleme işlemlerini otomatik olarak yönetir. API, devam eden parçaları inceleme veya elle temizleme gibi gelişmiş kullanım durumları için kullanılabilir.