Transkription: eingebaute Instanz läuft auf dem PC im Tailnet; ist sie nicht erreichbar, wartet der Auftrag und versucht es alle 10 Minuten neu (höchstens zwei Tage); Whisper-Server kann CUDA

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
mrwnslz 2026-11-09 11:17:23 +01:00
parent e8bb70feb0
commit 55239f71a0
4 changed files with 125 additions and 12 deletions

View file

@ -12,12 +12,19 @@ import subprocess
import threading import threading
import time import time
import numpy as np # Windows with a GPU: where the CUDA libraries (cuBLAS, cuDNN) are, separated by ';'.
from fastapi import FastAPI, File, Form, HTTPException, UploadFile for _d in filter(None, os.environ.get('WHISPER_DLL_DIRS', '').split(';')):
from faster_whisper import WhisperModel os.add_dll_directory(_d)
os.environ['PATH'] = _d + os.pathsep + os.environ['PATH']
import numpy as np # noqa: E402
from fastapi import FastAPI, File, Form, HTTPException, UploadFile # noqa: E402
from faster_whisper import WhisperModel # noqa: E402
MODEL = os.environ.get('WHISPER_MODEL', 'large-v3-turbo') MODEL = os.environ.get('WHISPER_MODEL', 'large-v3-turbo')
THREADS = int(os.environ.get('WHISPER_THREADS', '4')) THREADS = int(os.environ.get('WHISPER_THREADS', '4'))
DEVICE = os.environ.get('WHISPER_DEVICE', 'cpu') # cpu or cuda
COMPUTE = os.environ.get('WHISPER_COMPUTE', 'int8') # int8_float16 on a GPU
IDLE = int(os.environ.get('WHISPER_IDLE', '600')) IDLE = int(os.environ.get('WHISPER_IDLE', '600'))
CACHE = os.environ.get('WHISPER_CACHE', '/models') CACHE = os.environ.get('WHISPER_CACHE', '/models')
@ -28,7 +35,7 @@ state = {'model': None, 'last': 0.0}
def model(): def model():
if state['model'] is None: if state['model'] is None:
state['model'] = WhisperModel(MODEL, device='cpu', compute_type='int8', cpu_threads=THREADS, state['model'] = WhisperModel(MODEL, device=DEVICE, compute_type=COMPUTE, cpu_threads=THREADS,
download_root=CACHE) download_root=CACHE)
return state['model'] return state['model']
@ -56,7 +63,7 @@ def decode(data: bytes) -> np.ndarray:
@app.get('/health') @app.get('/health')
def health(): def health():
return {'ok': True, 'model': MODEL, 'loaded': state['model'] is not None} return {'ok': True, 'model': MODEL, 'device': DEVICE, 'loaded': state['model'] is not None}
@app.post('/v1/audio/transcriptions') @app.post('/v1/audio/transcriptions')

View file

@ -19,8 +19,8 @@ only way to start a transcription is the `POST` below, sent when the user presse
The **server** transcribes. It already downloads and decodes episodes for skipping silence The **server** transcribes. It already downloads and decodes episodes for skipping silence
(`podcasts/silence.py`), and browsers and the WebView can't fetch podcast audio (CORS). It cuts (`podcasts/silence.py`), and browsers and the WebView can't fetch podcast audio (CORS). It cuts
the audio at pauses and sends the pieces in parallel to a Whisper instance with an the audio at pauses and sends the pieces in parallel to a Whisper instance with an
OpenAI-compatible API (`POST /v1/audio/transcriptions`): the built-in one by default (a container OpenAI-compatible API (`POST /v1/audio/transcriptions`): the built-in one by default (`DIORA_WHISPER_URL`),
beside diora-web, `DIORA_WHISPER_URL`), or any instance the user enters. YouTube episodes are or any instance the user enters. YouTube episodes are
not transcribed: the server never fetches their sound (docs/youtube.md). not transcribed: the server never fetches their sound (docs/youtube.md).
The **device** writes the article with the user's own key (docs/ai.md). The key never reaches the The **device** writes the article with the user's own key (docs/ai.md). The key never reaches the
@ -242,3 +242,21 @@ using the metadata, and don't quote garbled passages.
``` ```
Prompt (user turn): the metadata block, then `Transkript:` and the text. Prompt (user turn): the metadata block, then `Transkript:` and the text.
## The built-in instance (since 2026-11-09)
Whisper does not fit on the Strato server: with about 2 GB of memory free, large-v3-turbo needed
more than 2.2 GB and even `small` broke a 1.2 GB limit, and earlyoom then killed other services.
So the built-in instance runs on the user's PC (multi11, GTX 1660 Ti) and diora-prod reaches it over
the tailnet: `DIORA_WHISPER_URL=http://100.64.0.3:8766/v1` in `~/docker-compose/diora-prod/.env`.
It is `deploy/whisper/server.py` (a small OpenAI-compatible server around faster-whisper,
large-v3-turbo, int8_float16 on CUDA; one request at a time; the model is loaded on demand and freed
after 10 idle minutes), started at logon from `F:\diora-whisper` (start.cmd via start-hidden.vbs,
shortcut in the user's Startup folder; log `F:\diora-whisper\server.log`), listening only on the
tailnet address. One hour of speech takes about five minutes.
When the PC is off, a job does not fail: the server checks the instance before downloading, and a
connection error (there or while pieces are out) puts the job back into the queue with
`error = "Whisper-Instanz nicht erreichbar, neuer Versuch um HH:MM"`, tried again every ten
minutes (`WAIT_RETRY`) for up to two days after the button was pressed (`WAIT_MAX`). Only the
built-in instance waits like this; a custom instance's connection errors fail the job as before.

View file

@ -299,6 +299,41 @@ class RunTests(PipelineBase):
self.assertIn('Whisper', job.error) self.assertIn('Whisper', job.error)
self.assertEqual(self.fake.calls, []) self.assertEqual(self.fake.calls, [])
def test_unreachable_builtin_waits_and_tries_again(self):
import socket
sock = socket.socket()
sock.bind(('127.0.0.1', 0))
port = sock.getsockname()[1]
sock.close() # nothing listens there now
with override_settings(WHISPER_URL=f'http://127.0.0.1:{port}/v1'):
self.configure()
job = self.job()
transcribe.run(job.pk)
job = self.reload(job)
self.assertEqual(job.state, 'queued')
self.assertIn('nicht erreichbar', job.error)
self.assertGreater(transcribe._not_before[job.pk], time.time())
self.assertEqual(self.fake.calls, []) # nothing downloaded while the instance is away
self.assertEqual(transcribe._next_job(), transcribe.WAIT)
transcribe._not_before[job.pk] = 0 # due again: the instance is back
with override_settings(WHISPER_URL=self.fake.whisper):
transcribe.run(job.pk)
job = self.reload(job)
self.assertEqual((job.state, job.error), ('done', ''))
transcribe._not_before.clear()
def test_waiting_gives_up_after_two_days(self):
from datetime import timedelta
from django.utils import timezone
with override_settings(WHISPER_URL='http://127.0.0.1:9/v1'):
self.configure()
job = self.job()
TranscriptionJob.objects.filter(pk=job.pk).update(created_at=timezone.now() - timedelta(days=3))
transcribe.run(job.pk)
job = self.reload(job)
self.assertEqual(job.state, 'error')
self.assertIn('zwei Tage', job.error)
@override_settings(WHISPER_URL='') @override_settings(WHISPER_URL='')
def test_retries_then_succeeds(self): def test_retries_then_succeeds(self):
self.fake.fail_first = 2 self.fake.fail_first = 2

View file

@ -11,6 +11,7 @@ taken oldest first (by id). A job found `downloading` or `transcribing` when the
just started (a restart killed it) is queued again. Cancelling deletes the job's row: the just started (a restart killed it) is queued again. Cancelling deletes the job's row: the
worker notices when it next writes progress and stops. worker notices when it next writes progress and stops.
""" """
import datetime
import logging import logging
import os import os
import subprocess import subprocess
@ -42,10 +43,15 @@ DECODE_TIMEOUT = 3600
OPUS_BITRATE = '32k' OPUS_BITRATE = '32k'
MAX_PROMPT = 500 # characters MAX_PROMPT = 500 # characters
ALIVE_CHECK = 5 # seconds between looks whether the job was cancelled ALIVE_CHECK = 5 # seconds between looks whether the job was cancelled
WAIT_RETRY = 600 # seconds between tries while the built-in instance can't be reached
WAIT_MAX = 48 * 3600 # after this long since the button was pressed a waiting job gives up
WAIT_IDLE = 30 # seconds the runner sleeps while every queued job is waiting
_lock = threading.Lock() _lock = threading.Lock()
_runner = None # the thread working through the queue, or None _runner = None # the thread working through the queue, or None
_resumed = False # whether jobs left over from before a restart were queued again _resumed = False # whether jobs left over from before a restart were queued again
_not_before = {} # job id -> time.time() before which a waiting job isn't tried again
WAIT = 'wait' # _next_job: jobs are queued, but none is due yet
class Cancelled(Exception): class Cancelled(Exception):
@ -56,6 +62,11 @@ class Failed(Exception):
"""The job can't succeed; the message is shown to the user.""" """The job can't succeed; the message is shown to the user."""
class Unreachable(Exception):
"""The built-in instance can't be reached (e.g. the machine it runs on is off): the job waits
and tries again later instead of failing."""
class Permanent(Exception): class Permanent(Exception):
"""A request the instance refused for good (a client error): not worth a retry.""" """A request the instance refused for good (a client error): not worth a retry."""
@ -85,11 +96,14 @@ def kick():
def _next_job(): def _next_job():
global _runner global _runner
with _lock: with _lock:
job_id = (TranscriptionJob.objects.filter(state=TranscriptionJob.QUEUED) queued = list(TranscriptionJob.objects.filter(state=TranscriptionJob.QUEUED)
.order_by('id').values_list('id', flat=True).first()) .order_by('id').values_list('id', flat=True))
if job_id is None: if not queued:
_runner = None # decided under the lock, so a job queued right now starts a new runner _runner = None # decided under the lock, so a job queued right now starts a new runner
return job_id return None
now = time.time()
due = next((i for i in queued if _not_before.get(i, 0) <= now), None)
return due if due is not None else WAIT
def _loop(): def _loop():
@ -100,6 +114,9 @@ def _loop():
job_id = _next_job() job_id = _next_job()
if job_id is None: if job_id is None:
return return
if job_id == WAIT:
time.sleep(WAIT_IDLE)
continue
try: try:
run(job_id) run(job_id)
except Exception: # run() reports its own failures; this keeps the queue going except Exception: # run() reports its own failures; this keeps the queue going
@ -124,6 +141,8 @@ def run(job_id):
_write(job_id, state=TranscriptionJob.DONE, data=data, error='', finished_at=timezone.now()) _write(job_id, state=TranscriptionJob.DONE, data=data, error='', finished_at=timezone.now())
except Cancelled: except Cancelled:
log.info('transcription job %s cancelled', job_id) log.info('transcription job %s cancelled', job_id)
except Unreachable as e:
_wait(job_id, str(e))
except Failed as e: except Failed as e:
_fail(job_id, str(e)) _fail(job_id, str(e))
except Exception as e: except Exception as e:
@ -131,6 +150,31 @@ def run(job_id):
_fail(job_id, f'{type(e).__name__}: {e}') _fail(job_id, f'{type(e).__name__}: {e}')
def _wait(job_id, reason):
"""Back into the queue to try again in WAIT_RETRY seconds, or give up when it waited too long."""
job = TranscriptionJob.objects.filter(pk=job_id).only('created_at').first()
if job is None:
return
if (timezone.now() - job.created_at).total_seconds() > WAIT_MAX:
_not_before.pop(job_id, None)
_fail(job_id, 'Whisper-Instanz zwei Tage lang nicht erreichbar')
return
_not_before[job_id] = time.time() + WAIT_RETRY
again = timezone.localtime(timezone.now() + datetime.timedelta(seconds=WAIT_RETRY)).strftime('%H:%M')
log.info('transcription job %s waits for the instance (%s)', job_id, reason)
TranscriptionJob.objects.filter(pk=job_id).update(
state=TranscriptionJob.QUEUED, chunks_done=0, chunks_total=0, started_at=None,
error=f'Whisper-Instanz nicht erreichbar, neuer Versuch um {again}')
def _probe(url):
"""Unreachable unless something answers at the built-in instance's address."""
try:
requests.get(url.rsplit('/audio/transcriptions', 1)[0], timeout=(5, 10)).close()
except (requests.ConnectionError, requests.Timeout) as e:
raise Unreachable(e.__class__.__name__)
def _fail(job_id, message): def _fail(job_id, message):
TranscriptionJob.objects.filter(pk=job_id).update( TranscriptionJob.objects.filter(pk=job_id).update(
state=TranscriptionJob.ERROR, error=(message or 'failed')[:500], finished_at=timezone.now()) state=TranscriptionJob.ERROR, error=(message or 'failed')[:500], finished_at=timezone.now())
@ -173,6 +217,8 @@ def _transcribe(job_id):
episode = job.episode episode = job.episode
conf = for_user(episode.feed.user) conf = for_user(episode.feed.user)
url, headers, builtin = _target(conf) url, headers, builtin = _target(conf)
if builtin:
_probe(url) # before downloading: is the instance there at all?
options = {'model': conf.model or DEFAULT_MODEL, 'response_format': 'verbose_json'} options = {'model': conf.model or DEFAULT_MODEL, 'response_format': 'verbose_json'}
if conf.language: if conf.language:
options['language'] = conf.language options['language'] = conf.language
@ -312,7 +358,10 @@ def _post(url, headers, builtin, name, mime, body, options):
try: try:
if builtin: if builtin:
# The operator's own instance (DIORA_WHISPER_URL) may sit on a private network. # The operator's own instance (DIORA_WHISPER_URL) may sit on a private network.
r = requests.post(url, headers=headers, files=files, data=options, timeout=UPLOAD_TIMEOUT) try:
r = requests.post(url, headers=headers, files=files, data=options, timeout=UPLOAD_TIMEOUT)
except requests.ConnectionError as e:
raise Unreachable(e.__class__.__name__)
if r.status_code >= 400: if r.status_code >= 400:
if r.status_code < 500 and r.status_code not in (408, 429): if r.status_code < 500 and r.status_code not in (408, 429):
raise Permanent(f'HTTP {r.status_code}') raise Permanent(f'HTTP {r.status_code}')
@ -343,6 +392,10 @@ def _send(url, headers, builtin, name, mime, body, options, stop):
return _post(url, headers, builtin, name, mime, body, options) return _post(url, headers, builtin, name, mime, body, options)
except Permanent as e: except Permanent as e:
raise Failed(f'Whisper: {e}') raise Failed(f'Whisper: {e}')
except Unreachable:
if attempt >= RETRIES:
raise
time.sleep(RETRY_PAUSES[min(attempt, len(RETRY_PAUSES) - 1)])
except RuntimeError as e: except RuntimeError as e:
if attempt >= RETRIES: if attempt >= RETRIES:
raise Failed(f'Whisper: {e}') raise Failed(f'Whisper: {e}')