Aller au contenu principal

F-03 Traitement vidéo

Brique F de la plateforme Enterprise AI. Technologie : FFmpeg + PyAV dans des workers média asynchrones (file NATS JetStream). Version anglaise : ../../en/bricks/F-03-video-processing.md.

Rôle et responsabilités​

  • Analyser, transcoder, découper, extraire des images (cadence fixe ou changement de scène), extraire l'audio, créer miniatures et proxys.
  • Alimenter la vision F-01 (images) et le STT F-05 (audio) pour rendre une vidéo analysable et indexable.

Position et interactions​

  • Chemin d'exécution : batch et événement (l'événement media.uploaded déclenche le traitement).
  • Chemin événementiel : media.uploaded sur NATS -> worker -> mises à jour du job -> événement media.processed.
  • Les vidéos longues sont traitées par segments avec points de reprise ; des workflows Temporal orchestrent les pipelines multi-étapes.

Part open source et part développée​

  • Open source réutilisé : FFmpeg, PyAV, ffprobe.
  • Développé dans la plateforme : Catalogue d'opérations, segmentation, limites de ressources par job, sorties en médias enfants.

Modèle de données​

Tables du Control Plane (schéma eai) : eai.media, eai.multimodal_jobs, eai.usage_events.

Chaque table de tenant a tenant_id uuid NOT NULL et une politique de sécurité au niveau des lignes sur current_setting('app.tenant_id', true)::uuid. ClickHouse usage_events est la réplique analytique de eai.usage_events (ReplacingMergeTree par event_id).

Contrat : contracts/v1/multimodal-contract.schema.json (schema_version 1.0). L'usage est toujours remonté avec contracts/v1/usage-event.schema.json, l'audit avec contracts/v1/audit-event.schema.json.

Endpoints d'API​

MéthodeChemin
POST/v1/multimodal/video/process
GET/v1/multimodal/jobs
GET/v1/multimodal/jobs/{id}

POST /v1/multimodal/video/process​

Requête :

{
  "media_id": "med_VID1",
  "operations": [
    "frames",
    "audio"
  ],
  "fps": 1,
  "sample_rate": 16000
}

Réponse :

{
  "job_id": "mmj_V01",
  "status": "queued",
  "status_url": "/v1/multimodal/jobs/mmj_V01"
}

GET /v1/multimodal/jobs​

Réponse :

{
  "items": [
    {
      "job_id": "mmj_V01",
      "operation": "video_process",
      "status": "running",
      "progress": 0.4
    }
  ],
  "next_cursor": null
}

GET /v1/multimodal/jobs/{id}​

Réponse :

{
  "job_id": "mmj_V01",
  "operation": "video_process",
  "status": "succeeded",
  "outputs": {
    "frames": 118,
    "audio_media_id": "med_AUD7"
  },
  "attempts": 1
}

Exemples SDK​

Python​

from enterprise_ai import EnterpriseAI

client = EnterpriseAI(base_url="<control-plane-url>", api_key="eai_live_xxx", tenant="acme", locale="fr")
client.multimodal.video.process("med_VID1", operations=["frames", "audio"])
client.multimodal.jobs(status="running", limit=50)
client.multimodal.job(job_id="mmj_V01")

TypeScript​

import { EnterpriseAI } from "@enterprise-ai/sdk";

const client = new EnterpriseAI({ baseUrl: "<control-plane-url>", apiKey: process.env.EAI_API_KEY, tenant: "acme", locale: "fr" });
await client.multimodal.processVideo({ media_id: "med_VID1", operations: ["frames", "audio"] });
await client.multimodal.listJobs({ status: "running", limit: 50 });
await client.multimodal.getJob("mmj_V01");

Java​

EnterpriseAIClient client = EnterpriseAIClient.builder()
    .baseUrl("<control-plane-url>")
    .apiKey(System.getenv("EAI_API_KEY"))
    .tenant("acme")
    .locale(Locale.FRENCH)
    .build();
client.multimodal().processVideo(Map.of("media_id", "med_VID1", "operations", List.of("frames", "audio")));
client.multimodal().jobs(Map.of("status", "running", "limit", 50));
client.multimodal().job("mmj_V01");

Go​

client, err := eai.NewClient(eai.WithBaseURL("<control-plane-url>"), eai.WithAPIKey(os.Getenv("EAI_API_KEY")), eai.WithTenant("acme"), eai.WithLocale("fr"))
if err != nil { log.Fatal(err) }
ctx := context.Background()
_, err = client.Multimodal.ProcessVideo(ctx, map[string]any{"media_id": "med_VID1", "operations": []string{"frames", "audio"}})
_, err = client.Multimodal.Jobs(ctx, eai.WithQuery(map[string]string{"status": "running", "limit": "50"}))
_, err = client.Multimodal.Job(ctx, "mmj_V01")

// worker package: run a media worker consuming multimodal jobs
w, err := worker.New(worker.Config{Source: jobs, Handler: handle, Client: client, Concurrency: 2})
if err != nil { log.Fatal(err) }
err = w.Run(ctx)

Configuration​

Variables d'environnement : EAI_MEDIA_MAX_BYTES, EAI_NATS_URL.

Clés de valeurs Helm (deploy/helm/enterprise-ai/values.yaml) : workers.items[name=media].replicas, workers.nats.stream.

Sécurité​

  • FFmpeg tourne sans réseau, système de fichiers racine en lecture seule, limites CPU et mémoire, runtime gVisor.
  • Seule une liste blanche de conteneurs et de codecs est acceptée.
  • Tenant déduit de l'identifiant d'accès, jamais du corps ; sécurité au niveau des lignes PostgreSQL sur chaque table de tenant.
  • Chaque mutation et décision de politique écrit un événement d'audit chaîné par hachage (G-11).
  • Les erreurs utilisent l'enveloppe standard avec message_i18n en et fr.

Observabilité​

  • Spans de job liés à la trace d'upload via le traceparent stocké sur le job.
  • traceparent W3C propagé, X-Request-Id renvoyé, les spans portent eai.tenant_id, eai.project, eai.agent_id, eai.run_id.

Unités FinOps​

  • Enregistré en cpu_seconds, bytes_transferred et storage_gb_hours sur le média ; video_seconds traitées dans labels.

Modes de défaillance et résilience​

  • Fichier corrompu -> job en échec avec validation_error, original conservé pour l'audit.
  • Crash de worker -> message relivré par JetStream, sorties idempotentes par id de job et segment.

Adoption autonome​

Un client peut utiliser seuls les workers média pour un prétraitement vidéo conforme avant ses propres analyses.

Tests d'acceptation​

  1. Un MP4 de 10 minutes produit 600 images à 1 fps et un média audio.
  2. Tuer un worker en cours de job termine le job une seule fois, sans médias enfants dupliqués.

Feuille de route d'implémentation