Aller au contenu principal

Workers multimodaux GPU

Le Control Plane exécute une opération multimodale en ligne quand un adaptateur est disponible dans son propre processus. Sinon le job reste dans eai.multimodal_jobs au statut queued (« no adapter available, queued for GPU workers ») et c'est un worker GPU, python -m eai.media_worker, qui le traite.

Lancement​

python -m eai.media_worker --kind stt [--kind diarization] [--stream EAI_MULTIMODAL] [--consumer stt-workers]
python -m eai.media_worker --kind tts --once          # au plus un job, puis sortie
python -m eai.media_worker --kind stt --max-jobs 10   # sortie après 10 jobs
--kindCapacité d'adaptateurAdaptateurs locaux (extra workers)
sttsttfaster-whisper, ou EAI_STT_URL
diarizationdiarizationpyannote.audio (EAI_HF_TOKEN)
ttsttspiper (piper-tts), ou EAI_TTS_URL
image_generation, image_editimage_generationpoint d'accès compatible OpenAI
vision_understandingvisionpoint d'accès compatible OpenAI, ou métadonnées et OCR locaux
video_generationvideo_generationpoint d'accès de génération vidéo

Au démarrage, le worker refuse de tourner (code de sortie 2, message explicite) si aucun adaptateur n'est disponible pour l'un des --kind demandés. L'image deploy/docker/worker.Dockerfile installe l'extra workers de services/control-plane/pyproject.toml (faster-whisper, pyannote.audio, piper-tts, silero-vad, torch, diffusers ; le client NATS nats-py fait partie des dépendances de base) ; une installation qui échoue fait échouer la construction de l'image.

Traitement d'un job​

  1. Réservation : une seule requête dans le contexte système, FOR UPDATE SKIP LOCKED, le job queued le plus ancien des opérations demandées passe à running, adapter = worker:<hostname>, attempts + 1. Deux workers ne réservent jamais le même job.
  2. Exécution dans une transaction limitée au tenant du job, avec le principal système de ce tenant : la RLS, l'événement d'usage FinOps et l'audit appartiennent au tenant du job.
  3. La charge utile de l'adaptateur est reconstruite par adapter_payload() (module eai.domains.multimodal.service) à partir de l'entrée du job et du média déchiffré. Le chemin en ligne utilise la même fonction : les deux exécutions ne peuvent pas diverger. L'entrée du job ne contient jamais de secret ; un média fourni en base64 est stocké chiffré au moment de la mise en file.
  4. Le post-traitement (DLP sur le texte extrait, stockage des médias produits) et finish_job sont ceux du chemin en ligne.

Erreurs : une AdapterError met le job en failed avec un message tronqué à 2000 caractères ; toute autre exception le met en failed avec le code générique worker_error, et le journal ne garde que le type de l'exception et sa position (jamais le contenu, qui peut contenir des données personnelles).

Arrêt : SIGTERM ou SIGINT arrête les réservations ; le job en cours se termine, un job réservé mais pas encore démarré est rendu (queued), puis le processus sort avec le code 0.

Bail et remise en file​

Pendant l'exécution, le worker rafraîchit updated_at du job (le bail) toutes les EAI_MEDIA_WORKER_HEARTBEAT_SECONDS. Si le worker meurt entre la réservation et la fin, le job périodique media_requeue de python -m eai.worker remet en file les jobs running réservés par un worker (adapter commençant par worker:) dont updated_at dépasse EAI_MEDIA_JOB_LEASE_SECONDS, et les passe en failed (lease_expired_max_attempts) quand attempts atteint EAI_MEDIA_JOB_MAX_ATTEMPTS. Chaque remise en file est auditée (multimodal.job.requeue).

NATS JetStream et KEDA​

Avec EAI_NATS_URL, la mise en file d'un job écrit aussi {job_id, tenant_id, operation} sur le sujet multimodal.jobs.<operation> dans l'outbox transactionnelle (eai.event_outbox) : l'événement est validé avec le job. Le job outbox de python -m eai.worker le relaie vers JetStream, qui préfixe le tenant : le sujet réel est eai.<tenant_id>.multimodal.jobs.<operation>.

Flux à créer (une fois, par l'équipe plateforme) :

nats stream add EAI_MULTIMODAL --subjects 'eai.*.multimodal.jobs.>' --storage file --retention work \
  --replicas 3 --discard old --max-age 7d

Le worker crée ou met à jour son consommateur durable --consumer sur --stream, filtré sur eai.*.multimodal.jobs.<kind> pour chacun de ses --kind (accusé explicite, ack_wait égal au bail). Chaque message est un réveil : le worker réserve et traite le job le plus ancien, puis accuse le message après le traitement, ce qui donne à KEDA (ScaledObject nats-jetstream du chart, workers.items[].lagThreshold) un retard qui compte les jobs non traités. Les permissions NATS de deploy/data/nats/values.yaml sont alignées : l'utilisateur worker publie eai.*.multimodal.jobs.>, l'utilisateur multimodal s'y abonne.

Sans NATS (ou sans le client nats-py), le worker interroge la base toutes les EAI_MEDIA_WORKER_POLL_SECONDS.

Déduplication côté consommateur (inbox)​

Chaque message reçu passe par la boîte de réception du consommateur (eai.inbox, table eai.inbox_events, migration V166) avant son traitement. Il y est inscrit sous la clé unique (tenant, --consumer, identifiant du message) ; l'identifiant est l'en-tête Nats-Msg-Id s'il existe, sinon <flux>:<séquence du flux>, identique à chaque relivraison. Le tenant vient du sujet eai.<tenant_id>.<sujet> ou du champ tenant_id du message ; la ligne est donc soumise à la RLS du tenant.

  • Un message déjà traité (relivraison après un accusé perdu, publication en double) est accusé sans nouvelle réservation de job.
  • Un traitement en échec passe la ligne en failed (code d'erreur court, jamais le contenu), le message est refusé (nak) et la livraison suivante le rejoue (attempts + 1).
  • Le contenu du message n'est jamais copié dans la table, seulement ses identifiants.
  • Le job inbox_purge de python -m eai.worker supprime les lignes processed plus anciennes que EAI_INBOX_RETENTION_DAYS ; les lignes received et failed restent pour le rejeu et l'analyse.

Déploiement​

  • Helm : workers.enabled: true et une entrée par opération dans workers.items (name, kind, consumer, gpus, minReplicas, maxReplicas, lagThreshold). Les pods lisent le ConfigMap et le Secret du Control Plane (base, clés de chiffrement, stockage objet, NATS) et gardent le durcissement : non root, système de fichiers racine en lecture seule, drop: [ALL], seccomp RuntimeDefault, pas de jeton de compte de service. La politique réseau multimodal-workers n'ouvre aucune entrée ; en sortie seulement PostgreSQL, NATS, le stockage objet (médias chiffrés), OpenBao, OTLP et le DNS du cluster.
  • Compose : service media-worker du profil gpu (docker compose --profile gpu up media-worker).

Paramètres​

VariableDéfautRôle
EAI_MEDIA_WORKER_POLL_SECONDS2intervalle d'interrogation de la base sans NATS
EAI_MEDIA_WORKER_HEARTBEAT_SECONDS60rafraîchissement du bail pendant un job
EAI_MEDIA_JOB_LEASE_SECONDS900âge de updated_at au-delà duquel un job running est remis en file
EAI_MEDIA_JOB_MAX_ATTEMPTS3réservations avant l'échec définitif
EAI_WORKER_MEDIA_REQUEUE_SECONDS60intervalle du job media_requeue de python -m eai.worker
EAI_INBOX_RETENTION_DAYS7conservation des lignes traitées de eai.inbox_events
EAI_WORKER_INBOX_PURGE_SECONDS3600intervalle du job inbox_purge de python -m eai.worker
EAI_NATS_URL(vide)active les réveils JetStream