# Akış Adımları

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](https://github.com/inline0/queuety/issues/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ı](https://queuety.inline0.dev/tr/docs/features/heartbeats) 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.