diff --git a/.env.example b/.env.example index 52f1522..1574b42 100644 --- a/.env.example +++ b/.env.example @@ -99,6 +99,8 @@ VITE_REVERB_SCHEME=http LOCAL_WHISPER_URL=http://127.0.0.1:8090/v1 LOCAL_WHISPER_API_KEY=not-needed LOCAL_WHISPER_MODEL=Systran/faster-whisper-base +# curl streams large uploads off disk; http is for tests (Http::fake) +LOCAL_WHISPER_TRANSPORT=curl TRANSCRIPTION_TIMEOUT=600 # Must be greater than TRANSCRIPTION_TIMEOUT so long Whisper jobs are not re-queued mid-run DB_QUEUE_RETRY_AFTER=660 diff --git a/app/Services/LocalWhisperCurlClient.php b/app/Services/LocalWhisperCurlClient.php new file mode 100644 index 0000000..5abd258 --- /dev/null +++ b/app/Services/LocalWhisperCurlClient.php @@ -0,0 +1,120 @@ + 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); + } + } +} diff --git a/app/Services/TranscriptionService.php b/app/Services/TranscriptionService.php index 2dddf38..e8cd536 100644 --- a/app/Services/TranscriptionService.php +++ b/app/Services/TranscriptionService.php @@ -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|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).'…'; + } } diff --git a/config/ai.php b/config/ai.php index e2c9557..40ec538 100644 --- a/config/ai.php +++ b/config/ai.php @@ -29,6 +29,12 @@ return [ 'transcription_timeout' => (int) env('TRANSCRIPTION_TIMEOUT', 600), 'local_whisper_model' => env('LOCAL_WHISPER_MODEL', 'Systran/faster-whisper-base'), + /* + | "curl" streams large uploads off disk (default). "http" uses Laravel's + | HTTP client and is intended for Http::fake() in tests. + */ + 'local_whisper_transport' => env('LOCAL_WHISPER_TRANSPORT', 'curl'), + /* |-------------------------------------------------------------------------- | Caching diff --git a/phpunit.xml b/phpunit.xml index 628b887..c95f3bf 100644 --- a/phpunit.xml +++ b/phpunit.xml @@ -36,5 +36,6 @@ + diff --git a/tests/Feature/LocalWhisperCurlTransportTest.php b/tests/Feature/LocalWhisperCurlTransportTest.php new file mode 100644 index 0000000..3bc727e --- /dev/null +++ b/tests/Feature/LocalWhisperCurlTransportTest.php @@ -0,0 +1,119 @@ +markTestSkipped('curl extension required'); + } + + Storage::fake('local'); + Storage::disk('local')->put('recordings/large.mp3', str_repeat('A', 2_000_000)); + + $stub = base_path('tests/fixtures/whisper-sse-stub.php'); + $host = '127.0.0.1'; + $port = $this->reservePort($host); + $docRoot = base_path('tests/fixtures'); + + $server = Process::timeout(30) + ->path($docRoot) + ->start([ + PHP_BINARY, + '-S', + "{$host}:{$port}", + 'whisper-sse-stub.php', + ]); + + try { + $this->waitForServer($host, $port); + + config([ + 'ai.local_whisper_transport' => 'curl', + 'ai.providers.local-whisper.url' => "http://{$host}:{$port}/v1", + 'ai.providers.local-whisper.key' => 'test-key', + ]); + + $user = User::factory()->create(); + $recording = Recording::query()->create([ + 'user_id' => $user->id, + 'title' => 'Curl transport', + 'original_filename' => 'large.mp3', + 'file_path' => 'recordings/large.mp3', + 'file_size_bytes' => 2_000_000, + 'duration_seconds' => 40, + 'transcription_status' => 'processing', + 'transcription_driver' => 'local', + ]); + + $partials = []; + + $text = app(TranscriptionService::class)->transcribe( + $recording, + function (string $message, int $percent, ?string $partial = null) use (&$partials): void { + if ($partial !== null) { + $partials[] = $partial; + } + }, + ); + + $this->assertSame('Hello from whisper.', $text); + $this->assertContains('Hello ', $partials); + $this->assertContains('Hello from whisper.', $partials); + } finally { + $server->signal(SIGTERM); + $server->wait(); + } + } + + private function reservePort(string $host): int + { + $socket = stream_socket_server("tcp://{$host}:0"); + + if ($socket === false) { + $this->fail('Unable to reserve a local TCP port for the Whisper stub.'); + } + + $name = stream_socket_get_name($socket, false); + fclose($socket); + + if (! is_string($name) || ! str_contains($name, ':')) { + $this->fail('Unable to determine reserved port.'); + } + + return (int) substr($name, strrpos($name, ':') + 1); + } + + private function waitForServer(string $host, int $port): void + { + $deadline = microtime(true) + 5; + + while (microtime(true) < $deadline) { + $errno = 0; + $errstr = ''; + $fp = @fsockopen($host, $port, $errno, $errstr, 0.2); + + if (is_resource($fp)) { + fclose($fp); + + return; + } + + usleep(50_000); + } + + $this->fail("Whisper stub server did not start on {$host}:{$port}"); + } +} diff --git a/tests/fixtures/whisper-sse-stub.php b/tests/fixtures/whisper-sse-stub.php new file mode 100644 index 0000000..39ed6f8 --- /dev/null +++ b/tests/fixtures/whisper-sse-stub.php @@ -0,0 +1,32 @@ +