Stream Whisper uploads with curl so large audio does not OOM.

2026-08-13 20:29:59 +02:00
parent 1898be4def
commit e77a18c49f
7 changed files with 488 additions and 42 deletions
@@ -0,0 +1,120 @@
+<?php
+
+namespace App\Services;
+
+use Closure;
+use CURLFile;
+use RuntimeException;
+
+/**
+ * Streams large audio uploads to Whisper with curl (disk → network),
+ * and delivers response body chunks without buffering the request in PHP.
+ *
+ * Laravel Http's stream => true uses Guzzle's StreamHandler, which casts the
+ * multipart body to a string and OOMs on big recordings.
+ */
+class LocalWhisperCurlClient
+{
+    /**
+     * POST /audio/transcriptions and stream the response body.
+     *
+     * @param  Closure(string): bool  $onBodyChunk  Return false to abort the transfer.
+     * @param  Closure(string): void|null  $onContentType  Invoked when the response Content-Type header arrives.
+     * @return array{status: int, content_type: string}
+     */
+    public function streamTranscription(
+        string $url,
+        string $apiKey,
+        string $path,
+        string $filename,
+        string $mimeType,
+        string $model,
+        int $timeout,
+        Closure $onBodyChunk,
+        ?Closure $onContentType = null,
+    ): array {
+        if (! function_exists('curl_init')) {
+            throw new RuntimeException('The curl PHP extension is required for local Whisper transcription.');
+        }
+
+        $contentType = '';
+        $abort = false;
+
+        $handle = curl_init($url);
+
+        if ($handle === false) {
+            throw new RuntimeException('Unable to initialize curl for local Whisper.');
+        }
+
+        try {
+            curl_setopt_array($handle, [
+                CURLOPT_POST => true,
+                CURLOPT_HTTPHEADER => [
+                    'Authorization: Bearer '.$apiKey,
+                    'Accept: application/json, text/event-stream',
+                ],
+                CURLOPT_POSTFIELDS => [
+                    'file' => new CURLFile($path, $mimeType, $filename),
+                    'model' => $model,
+                    'response_format' => 'verbose_json',
+                    'stream' => 'true',
+                    'without_timestamps' => 'false',
+                    'timestamp_granularities[]' => 'word',
+                ],
+                CURLOPT_RETURNTRANSFER => false,
+                CURLOPT_HEADER => false,
+                CURLOPT_TIMEOUT => max(1, $timeout),
+                CURLOPT_CONNECTTIMEOUT => 10,
+                CURLOPT_HEADERFUNCTION => static function ($ch, string $header) use (&$contentType, $onContentType): int {
+                    if (stripos($header, 'Content-Type:') === 0) {
+                        $contentType = trim(substr($header, strlen('Content-Type:')));
+
+                        if ($onContentType !== null) {
+                            $onContentType($contentType);
+                        }
+                    }
+
+                    return strlen($header);
+                },
+                CURLOPT_WRITEFUNCTION => static function ($ch, string $chunk) use ($onBodyChunk, &$abort): int {
+                    if ($abort) {
+                        return 0;
+                    }
+
+                    if ($onBodyChunk($chunk) === false) {
+                        $abort = true;
+
+                        return 0;
+                    }
+
+                    return strlen($chunk);
+                },
+            ]);
+
+            $ok = curl_exec($handle);
+            $errno = curl_errno($handle);
+            $error = curl_error($handle);
+            $status = (int) curl_getinfo($handle, CURLINFO_RESPONSE_CODE);
+
+            if ($abort) {
+                return [
+                    'status' => $status > 0 ? $status : 499,
+                    'content_type' => $contentType,
+                ];
+            }
+
+            if ($ok === false && $errno !== 0) {
+                throw new RuntimeException(
+                    'Local transcription request failed: '.$error.' (curl '.$errno.')'
+                );
+            }
+
+            return [
+                'status' => $status,
+                'content_type' => $contentType,
+            ];
+        } finally {
+            curl_close($handle);
+        }
+    }
+}
@@ -5,7 +5,6 @@ namespace App\Services;
 use App\Models\Recording;
 use Closure;
 use Illuminate\Http\Client\PendingRequest;
-use Illuminate\Http\Client\Response;
 use Illuminate\Support\Facades\Http;
 use Illuminate\Support\Facades\Log;
 use Laravel\Ai\Transcription;
@@ -13,7 +12,10 @@ use RuntimeException;

 class TranscriptionService
 {
-    public function __construct(private WhisperTranscriptionStream $stream) {}
+    public function __construct(
+        private WhisperTranscriptionStream $stream,
+        private LocalWhisperCurlClient $curl,
+    ) {}

    /**
     * Transcribe a recording with the local faster-whisper server.
@@ -56,8 +58,31 @@ class TranscriptionService
        $message = "Transcribing locally with {$model} (audio stays on this machine)…";
        $filename = $recording->original_filename ?: basename($path);

+        if ($this->usesHttpTransport()) {
+            return $this->transcribeViaHttp($recording, $report, $message, $model, $filename, $path, $shouldContinue);
+        }
+
+        return $this->transcribeViaCurl($recording, $report, $message, $model, $filename, $path, $shouldContinue);
+    }
+
+    /**
+     * Laravel Http client path (used in tests via Http::fake).
+     *
+     * Does not set Guzzle stream => true, which would string-cast the multipart upload.
+     *
+     * @param  Closure(string, int, ?string, ?array): void  $report
+     * @param  (Closure(): bool)|null  $shouldContinue
+     */
+    private function transcribeViaHttp(
+        Recording $recording,
+        Closure $report,
+        string $message,
+        string $model,
+        string $filename,
+        string $path,
+        ?Closure $shouldContinue,
+    ): string {
        $response = $this->localWhisperRequest($filename, $path)
-            ->withOptions(['stream' => true])
            ->post('audio/transcriptions', [
                'model' => $model,
                'response_format' => 'verbose_json',
@@ -68,7 +93,7 @@ class TranscriptionService

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

@@ -78,54 +103,92 @@ class TranscriptionService
            Log::warning('Whisper did not stream SSE; falling back to a single JSON body', [
                'recording_id' => $recording->id,
                'content_type' => $contentType,
+                'transport' => 'http',
            ]);

-            return $this->extractTranscriptText($response, $report, $message);
+            return $this->extractTranscriptText($response->json(), $report, $message);
        }

-        return $this->consumeWhisperStream($response, $report, $message, $recording, $shouldContinue);
+        return $this->consumeSseBody($response->body(), $report, $message, $recording, $shouldContinue);
    }

    /**
+     * Curl path for production: streams the file upload and SSE response chunks.
+     *
     * @param  Closure(string, int, ?string, ?array): void  $report
     * @param  (Closure(): bool)|null  $shouldContinue
     */
-    private function consumeWhisperStream(
-        Response $response,
+    private function transcribeViaCurl(
+        Recording $recording,
        Closure $report,
        string $message,
-        Recording $recording,
+        string $model,
+        string $filename,
+        string $path,
        ?Closure $shouldContinue,
    ): string {
-        $body = $response->toPsrResponse()->getBody();
+        $config = config('ai.providers.local-whisper');
+        $baseUrl = rtrim((string) ($config['url'] ?? 'http://127.0.0.1:8090/v1'), '/');
+        $timeout = (int) config('ai.transcription_timeout', 600);
+        $apiKey = (string) ($config['key'] ?? 'not-needed');
+
        $buffer = '';
+        $rawBody = '';
        $accumulated = '';
        $lastPercent = 0;
-        $idleReads = 0;
+        $isSse = null;

-        try {
-            while (! $body->eof()) {
+        $result = $this->curl->streamTranscription(
+            $baseUrl.'/audio/transcriptions',
+            $apiKey,
+            $path,
+            $filename,
+            $recording->audioMimeType(),
+            $model,
+            $timeout,
+            function (string $chunk) use (
+                &$buffer,
+                &$rawBody,
+                &$accumulated,
+                &$lastPercent,
+                &$isSse,
+                $report,
+                $message,
+                $recording,
+                $shouldContinue,
+            ): bool {
                if ($shouldContinue !== null && ! $shouldContinue()) {
-                    $body->close();
-
-                    break;
+                    return false;
                }

-                $chunk = $body->read(8192);
+                if ($isSse === null) {
+                    $rawBody .= $chunk;

-                if ($chunk === '') {
-                    $idleReads++;
+                    return true;
+                }

-                    if ($idleReads >= 40) {
-                        break;
+                if ($isSse === false) {
+                    $rawBody .= $chunk;
+
+                    return true;
+                }
+
+                if ($rawBody !== '') {
+                    $buffer .= $rawBody;
+                    $rawBody = '';
+
+                    foreach ($this->stream->extractPayloads($buffer) as $payload) {
+                        [$accumulated, $lastPercent] = $this->ingestEvent(
+                            $payload,
+                            $accumulated,
+                            $lastPercent,
+                            $recording,
+                            $report,
+                            $message,
+                        );
                    }
-
-                    usleep(50_000);
-
-                    continue;
                }

-                $idleReads = 0;
                $buffer .= $chunk;

                foreach ($this->stream->extractPayloads($buffer) as $payload) {
@@ -138,20 +201,113 @@ class TranscriptionService
                        $message,
                    );
                }
-            }

-            foreach ($this->stream->flushBuffer($buffer) as $payload) {
-                [$accumulated, $lastPercent] = $this->ingestEvent(
-                    $payload,
-                    $accumulated,
-                    $lastPercent,
-                    $recording,
-                    $report,
-                    $message,
-                );
-            }
-        } finally {
-            $response->close();
+                return true;
+            },
+            function (string $contentType) use (&$isSse): void {
+                $isSse = str_contains(strtolower($contentType), 'text/event-stream');
+            },
+        );
+
+        if ($result['status'] >= 400 || $result['status'] === 0) {
+            throw new RuntimeException(
+                'Local transcription failed (HTTP '.$result['status'].'): '.$this->truncateBody($rawBody)
+            );
+        }
+
+        $contentType = strtolower($result['content_type']);
+        $isSse ??= str_contains($contentType, 'text/event-stream');
+
+        if (! $isSse) {
+            Log::warning('Whisper did not stream SSE; falling back to a single JSON body', [
+                'recording_id' => $recording->id,
+                'content_type' => $contentType,
+                'transport' => 'curl',
+            ]);
+
+            $json = json_decode($rawBody, true);
+
+            return $this->extractTranscriptText(is_array($json) ? $json : null, $report, $message);
+        }
+
+        if ($rawBody !== '') {
+            $buffer .= $rawBody;
+            $rawBody = '';
+        }
+
+        foreach ($this->stream->extractPayloads($buffer) as $payload) {
+            [$accumulated, $lastPercent] = $this->ingestEvent(
+                $payload,
+                $accumulated,
+                $lastPercent,
+                $recording,
+                $report,
+                $message,
+            );
+        }
+
+        foreach ($this->stream->flushBuffer($buffer) as $payload) {
+            [$accumulated, $lastPercent] = $this->ingestEvent(
+                $payload,
+                $accumulated,
+                $lastPercent,
+                $recording,
+                $report,
+                $message,
+            );
+        }
+
+        if ($accumulated === '') {
+            throw new RuntimeException('Local transcription returned an empty transcript.');
+        }
+
+        return $accumulated;
+    }
+
+    private function usesHttpTransport(): bool
+    {
+        return config('ai.local_whisper_transport', 'curl') === 'http';
+    }
+
+    /**
+     * @param  Closure(string, int, ?string, ?array): void  $report
+     * @param  (Closure(): bool)|null  $shouldContinue
+     */
+    private function consumeSseBody(
+        string $body,
+        Closure $report,
+        string $message,
+        Recording $recording,
+        ?Closure $shouldContinue,
+    ): string {
+        if ($shouldContinue !== null && ! $shouldContinue()) {
+            throw new RuntimeException('Local transcription returned an empty transcript.');
+        }
+
+        $buffer = $body;
+        $accumulated = '';
+        $lastPercent = 0;
+
+        foreach ($this->stream->extractPayloads($buffer) as $payload) {
+            [$accumulated, $lastPercent] = $this->ingestEvent(
+                $payload,
+                $accumulated,
+                $lastPercent,
+                $recording,
+                $report,
+                $message,
+            );
+        }
+
+        foreach ($this->stream->flushBuffer($buffer) as $payload) {
+            [$accumulated, $lastPercent] = $this->ingestEvent(
+                $payload,
+                $accumulated,
+                $lastPercent,
+                $recording,
+                $report,
+                $message,
+            );
        }

        if ($accumulated === '') {
@@ -227,12 +383,11 @@ class TranscriptionService
    }

    /**
+     * @param  array<string, mixed>|null  $json
     * @param  Closure(string, int, ?string, ?array): void  $report
     */
-    private function extractTranscriptText(Response $response, ?Closure $report = null, string $message = ''): string
+    private function extractTranscriptText(?array $json, ?Closure $report = null, string $message = ''): string
    {
-        $json = $response->json();
-
        if (! is_array($json)) {
            throw new RuntimeException('Local transcription returned an empty transcript.');
        }
@@ -251,4 +406,15 @@ class TranscriptionService

        return $text;
    }
+
+    private function truncateBody(string $body): string
+    {
+        $body = trim($body);
+
+        if (strlen($body) <= 2000) {
+            return $body;
+        }
+
+        return substr($body, 0, 2000).'…';
+    }
 }