Stream Whisper segments so the UI can show live transcript text.

Consume faster-whisper SSE instead of blocking on one JSON response, persist growing text and progress over Reverb, and drop the fake percent heartbeat.
2026-08-13 12:34:47 +02:00
parent 5cea5192c2
commit 4c73620458
9 changed files with 574 additions and 138 deletions
@@ -76,9 +76,11 @@ class TranscribeRecording implements ShouldQueue
        try {
            $text = $transcription->transcribe(
                $this->recording,
-                function (string $message, int $percent): void {
-                    $this->reportIfOwned($message, $percent);
+                function (string $message, int $percent, ?string $partialTranscript = null): void {
+                    $this->reportIfOwned($message, $percent, $partialTranscript);
                },
+                fn (): bool => Recording::query()->find($this->recording->id)
+                    ?->ownsTranscriptionRun($this->runStartedAt) ?? false,
            );

            if (! $this->claimSuccessfulTranscript($text)) {
@@ -170,12 +172,12 @@ class TranscribeRecording implements ShouldQueue
        return false;
    }

-    private function reportIfOwned(string $message, int $percent): void
+    private function reportIfOwned(string $message, int $percent, ?string $partialTranscript = null): void
    {
        if (! $this->recording->ownsTranscriptionRun($this->runStartedAt)) {
            return;
        }

-        $this->recording->reportProgress($message, $percent);
+        $this->recording->reportProgress($message, $percent, partialTranscript: $partialTranscript);
    }
 }
@@ -502,14 +502,20 @@ class Recording extends Model
    /**
     * Update the live progress fields shown in the UI.
     */
-    public function reportProgress(string $message, int $percent, string $status = 'processing'): void
+    public function reportProgress(string $message, int $percent, string $status = 'processing', ?string $partialTranscript = null): void
    {
-        $this->forceFill([
+        $attributes = [
            'transcription_status' => $status,
            'transcription_progress' => $message,
            'transcription_percent' => max(0, min(100, $percent)),
            'transcription_error' => null,
-        ])->save();
+        ];
+
+        if ($partialTranscript !== null) {
+            $attributes['transcript'] = $partialTranscript;
+        }
+
+        $this->forceFill($attributes)->save();

        RecordingTranscriptionUpdated::dispatch($this->fresh());
    }
@@ -4,9 +4,7 @@ namespace App\Services;

 use App\Models\Recording;
 use Closure;
-use GuzzleHttp\Promise\PromiseInterface;
 use Illuminate\Http\Client\PendingRequest;
-use Illuminate\Http\Client\Promises\LazyPromise;
 use Illuminate\Http\Client\Response;
 use Illuminate\Support\Facades\Http;
 use Laravel\Ai\Transcription;
@@ -14,14 +12,17 @@ use RuntimeException;

 class TranscriptionService
 {
+    public function __construct(private WhisperTranscriptionStream $stream) {}
+
    /**
     * Transcribe a recording with the local faster-whisper server.
     *
-     * @param  (Closure(string, int): void)|null  $onProgress
+     * @param  (Closure(string, int, ?string): void)|null  $onProgress
+     * @param  (Closure(): bool)|null  $shouldContinue
     */
-    public function transcribe(Recording $recording, ?Closure $onProgress = null): string
+    public function transcribe(Recording $recording, ?Closure $onProgress = null, ?Closure $shouldContinue = null): string
    {
-        $report = $onProgress ?? static fn (string $message, int $percent) => null;
+        $report = $onProgress ?? static fn (string $message, int $percent, ?string $partial = null) => null;
        $model = config('ai.local_whisper_model', 'Systran/faster-whisper-base');

        $report('Connecting to local faster-whisper server…', 30);
@@ -32,7 +33,7 @@ class TranscriptionService
                ->timeout((int) config('ai.transcription_timeout', 600))
                ->generate('local-whisper', $model);
        } else {
-            $transcript = $this->transcribeViaLocalWhisper($recording, $report, $model);
+            $transcript = $this->transcribeViaLocalWhisper($recording, $report, $model, $shouldContinue);
        }

        $report('Received transcript from local Whisper…', 85);
@@ -41,11 +42,12 @@ class TranscriptionService
    }

    /**
-     * Call local Whisper and pulse progress while waiting.
+     * Call local Whisper, streaming SSE when the server supports it.
     *
-     * @param  Closure(string, int): void  $report
+     * @param  Closure(string, int, ?string): void  $report
+     * @param  (Closure(): bool)|null  $shouldContinue
     */
-    private function transcribeViaLocalWhisper(Recording $recording, Closure $report, string $model): string
+    private function transcribeViaLocalWhisper(Recording $recording, Closure $report, string $model, ?Closure $shouldContinue): string
    {
        $path = $recording->absolutePath();

@@ -55,31 +57,165 @@ class TranscriptionService

        $message = "Transcribing locally with {$model} (audio stays on this machine)…";
        $filename = $recording->original_filename ?: basename($path);
-        $request = $this->localWhisperRequest($filename, $path);

-        if ($this->shouldTranscribeSynchronously()) {
-            return $this->extractTranscriptText($request->post('audio/transcriptions', [
-                'model' => $model,
-                'response_format' => 'json',
-            ]));
-        }
-
-        /** @var LazyPromise $promise */
-        $promise = $request
-            ->async()
+        $response = $this->localWhisperRequest($filename, $path)
+            ->withOptions(['stream' => true])
            ->post('audio/transcriptions', [
                'model' => $model,
                'response_format' => 'json',
+                'stream' => 'true',
+                'without_timestamps' => 'false',
            ]);

-        $promise->buildPromise();
+        if (! $response->successful()) {
+            throw new RuntimeException(
+                'Local transcription failed (HTTP '.$response->status().'): '.$response->body()
+            );
+        }

-        $this->pulseProgressWhilePending($promise, $report, $message);
+        $contentType = strtolower((string) $response->header('Content-Type'));

-        /** @var Response $response */
-        $response = $promise->wait();
+        if (! str_contains($contentType, 'text/event-stream')) {
+            return $this->extractTranscriptText($response);
+        }

-        return $this->extractTranscriptText($response);
+        return $this->consumeWhisperStream($response, $report, $message, $recording->duration_seconds, $shouldContinue);
+    }
+
+    /**
+     * @param  Closure(string, int, ?string): void  $report
+     * @param  (Closure(): bool)|null  $shouldContinue
+     */
+    private function consumeWhisperStream(
+        Response $response,
+        Closure $report,
+        string $message,
+        ?int $durationSeconds,
+        ?Closure $shouldContinue,
+    ): string {
+        $body = $response->toPsrResponse()->getBody();
+        $buffer = '';
+        $accumulated = '';
+        $lastPercent = 50;
+        $eventsWithoutTimestamp = 0;
+        $lastFlushAt = 0.0;
+        $pending = false;
+
+        try {
+            while (! $body->eof()) {
+                if ($shouldContinue !== null && ! $shouldContinue()) {
+                    $body->close();
+
+                    break;
+                }
+
+                $chunk = $body->read(8192);
+
+                if ($chunk === '') {
+                    break;
+                }
+
+                $buffer .= $chunk;
+
+                foreach ($this->stream->extractPayloads($buffer) as $payload) {
+                    [$accumulated, $lastPercent, $eventsWithoutTimestamp, $pending] = $this->ingestEvent(
+                        $payload,
+                        $accumulated,
+                        $lastPercent,
+                        $eventsWithoutTimestamp,
+                        $durationSeconds,
+                        $report,
+                        $message,
+                        $lastFlushAt,
+                        $pending,
+                    );
+                }
+            }
+
+            foreach ($this->stream->flushBuffer($buffer) as $payload) {
+                [$accumulated, $lastPercent, $eventsWithoutTimestamp, $pending] = $this->ingestEvent(
+                    $payload,
+                    $accumulated,
+                    $lastPercent,
+                    $eventsWithoutTimestamp,
+                    $durationSeconds,
+                    $report,
+                    $message,
+                    $lastFlushAt,
+                    $pending,
+                );
+            }
+
+            if ($pending && $accumulated !== '') {
+                $report($message, $lastPercent, $accumulated);
+            }
+        } finally {
+            $response->close();
+        }
+
+        if ($accumulated === '') {
+            throw new RuntimeException('Local transcription returned an empty transcript.');
+        }
+
+        return $accumulated;
+    }
+
+    /**
+     * @param  Closure(string, int, ?string): void  $report
+     * @return array{0: string, 1: int, 2: int, 3: bool}
+     */
+    private function ingestEvent(
+        string $payload,
+        string $accumulated,
+        int $lastPercent,
+        int $eventsWithoutTimestamp,
+        ?int $durationSeconds,
+        Closure $report,
+        string $message,
+        float &$lastFlushAt,
+        bool $pending,
+    ): array {
+        $event = $this->stream->parseEvent($payload);
+
+        if ($event === null) {
+            return [$accumulated, $lastPercent, $eventsWithoutTimestamp, $pending];
+        }
+
+        $wasEmpty = $accumulated === '';
+        $accumulated = $this->stream->applyEvent($accumulated, $event);
+
+        if ($accumulated === '') {
+            return [$accumulated, $lastPercent, $eventsWithoutTimestamp, $pending];
+        }
+
+        $percent = $this->percentForEvent($event, $durationSeconds, $lastPercent, $eventsWithoutTimestamp);
+        $lastPercent = max($lastPercent, $percent);
+
+        $now = microtime(true);
+        $shouldFlush = $wasEmpty || $event['done'] || ($now - $lastFlushAt) >= 1.0;
+
+        if ($shouldFlush) {
+            $report($message, $lastPercent, $accumulated);
+            $lastFlushAt = $now;
+
+            return [$accumulated, $lastPercent, $eventsWithoutTimestamp, false];
+        }
+
+        return [$accumulated, $lastPercent, $eventsWithoutTimestamp, true];
+    }
+
+    /**
+     * @param  array{append: ?string, replace: ?string, end: ?float, done: bool, legacy: bool}  $event
+     */
+    private function percentForEvent(array $event, ?int $durationSeconds, int $lastPercent, int &$eventsWithoutTimestamp): int
+    {
+        if ($event['end'] !== null && $durationSeconds !== null && $durationSeconds > 0) {
+            return (int) min(99, max(50, round(100 * $event['end'] / $durationSeconds)));
+        }
+
+        $eventsWithoutTimestamp++;
+
+        return min(84, max($lastPercent, 50 + $eventsWithoutTimestamp));
    }

    private function localWhisperRequest(string $filename, string $path): PendingRequest
@@ -91,25 +227,12 @@ class TranscriptionService
        return Http::baseUrl($baseUrl)
            ->withHeaders(['Authorization' => 'Bearer '.($config['key'] ?? 'not-needed')])
            ->timeout($timeout)
+            ->connectTimeout(10)
            ->attach('file', fopen($path, 'r'), $filename);
    }

-    /**
-     * Laravel's HTTP fake does not resolve async promises; use sync in tests.
-     */
-    private function shouldTranscribeSynchronously(): bool
-    {
-        return app()->runningUnitTests();
-    }
-
    private function extractTranscriptText(Response $response): string
    {
-        if (! $response->successful()) {
-            throw new RuntimeException(
-                'Local transcription failed (HTTP '.$response->status().'): '.$response->body()
-            );
-        }
-
        $text = $response->json('text');

        if (! is_string($text) || $text === '') {
@@ -118,32 +241,4 @@ class TranscriptionService

        return $text;
    }
-
-    /**
-     * Broadcast incremental progress while Whisper is working.
-     *
-     * @param  Closure(string, int): void  $report
-     */
-    private function pulseProgressWhilePending(LazyPromise $promise, Closure $report, string $message, int $floor = 50, int $ceiling = 84): void
-    {
-        $lastPercent = $floor;
-        $lastPulseAt = microtime(true);
-        $startedAt = microtime(true);
-
-        while ($promise->getState() === PromiseInterface::PENDING) {
-            $now = microtime(true);
-
-            if ($now - $lastPulseAt >= 2.0) {
-                $elapsed = $now - $startedAt;
-                $percent = $floor + (int) floor(($ceiling - $floor) * (1 - exp(-$elapsed / 90)));
-                $percent = max($lastPercent + 1, min($ceiling, $percent));
-
-                $report($message, $percent);
-                $lastPercent = $percent;
-                $lastPulseAt = $now;
-            }
-
-            usleep(250_000);
-        }
-    }
 }
@@ -0,0 +1,161 @@
+<?php
+
+namespace App\Services;
+
+class WhisperTranscriptionStream
+{
+    /**
+     * Pull complete SSE `data:` payloads out of a rolling buffer.
+     *
+     * @return list<string>
+     */
+    public function extractPayloads(string &$buffer): array
+    {
+        $frames = explode("\n\n", str_replace("\r\n", "\n", $buffer));
+        $buffer = array_pop($frames) ?? '';
+        $payloads = [];
+
+        foreach ($frames as $frame) {
+            foreach ($this->dataLines($frame) as $payload) {
+                if ($payload === '[DONE]' || $payload === '') {
+                    continue;
+                }
+
+                $payloads[] = $payload;
+            }
+        }
+
+        return $payloads;
+    }
+
+    /**
+     * Flush a trailing incomplete frame at end-of-stream.
+     *
+     * @return list<string>
+     */
+    public function flushBuffer(string &$buffer): array
+    {
+        $trimmed = trim($buffer);
+
+        if ($trimmed === '') {
+            $buffer = '';
+
+            return [];
+        }
+
+        $buffer .= "\n\n";
+
+        return $this->extractPayloads($buffer);
+    }
+
+    /**
+     * Parse one SSE JSON payload into append/replace/end/done fields.
+     *
+     * @return array{append: ?string, replace: ?string, end: ?float, done: bool, legacy: bool}|null
+     */
+    public function parseEvent(string $json): ?array
+    {
+        $data = json_decode($json, true);
+
+        if (! is_array($data)) {
+            return null;
+        }
+
+        $type = $data['type'] ?? null;
+
+        if ($type === 'transcript.text.delta') {
+            $delta = $data['delta'] ?? '';
+
+            if (! is_string($delta) || $delta === '') {
+                return null;
+            }
+
+            return [
+                'append' => $delta,
+                'replace' => null,
+                'end' => $this->nullableFloat($data['end'] ?? null),
+                'done' => false,
+                'legacy' => false,
+            ];
+        }
+
+        if ($type === 'transcript.text.done') {
+            $text = $data['text'] ?? '';
+
+            return [
+                'append' => null,
+                'replace' => is_string($text) ? $text : '',
+                'end' => $this->nullableFloat($data['end'] ?? null),
+                'done' => true,
+                'legacy' => false,
+            ];
+        }
+
+        if ($type === null && isset($data['text']) && is_string($data['text']) && $data['text'] !== '') {
+            return [
+                'append' => $data['text'],
+                'replace' => null,
+                'end' => $this->nullableFloat($data['end'] ?? null),
+                'done' => false,
+                'legacy' => true,
+            ];
+        }
+
+        return null;
+    }
+
+    /**
+     * Apply a parsed event to the accumulated transcript.
+     *
+     * @param  array{append: ?string, replace: ?string, end: ?float, done: bool, legacy: bool}  $event
+     */
+    public function applyEvent(string $accumulated, array $event): string
+    {
+        if (is_string($event['replace'])) {
+            return $event['replace'];
+        }
+
+        $chunk = $event['append'] ?? '';
+
+        if ($chunk === '') {
+            return $accumulated;
+        }
+
+        if ($event['legacy']) {
+            $chunk = trim($chunk);
+
+            if ($accumulated === '') {
+                return $chunk;
+            }
+
+            return rtrim($accumulated).' '.$chunk;
+        }
+
+        return $accumulated.$chunk;
+    }
+
+    /**
+     * @return list<string>
+     */
+    private function dataLines(string $frame): array
+    {
+        $payloads = [];
+
+        foreach (explode("\n", $frame) as $line) {
+            if (str_starts_with($line, 'data:')) {
+                $payloads[] = ltrim(substr($line, 5));
+            }
+        }
+
+        return $payloads;
+    }
+
+    private function nullableFloat(mixed $value): ?float
+    {
+        if (! is_numeric($value)) {
+            return null;
+        }
+
+        return (float) $value;
+    }
+}