Skip to main content

F-03 Video processing

Brick F of the Enterprise AI Platform. Technology: FFmpeg + PyAV in asynchronous media workers (NATS JetStream queue). French version: ../../fr/bricks/F-03-video-processing.md.

Role and responsibilities​

  • Probe, transcode, cut, extract frames (fixed rate or scene change), extract audio tracks, build thumbnails and proxies.
  • Feed F-01 vision (frames) and F-05 STT (audio) to make a video searchable and analyzable.

Position and interactions​

  • Execution path: batch and event (media.uploaded event triggers processing).
  • Event path: media.uploaded on NATS -> worker -> job status updates -> media.processed event.
  • Long videos are processed in segments with checkpointing; Temporal workflows orchestrate multi-step pipelines.

Open source and developed parts​

  • Open source reused: FFmpeg, PyAV, ffprobe.
  • Developed in the platform: Operation catalog, segmenting, resource limits per job, outputs as child media.

Data model​

Control Plane tables (schema eai): eai.media, eai.multimodal_jobs, eai.usage_events.

Every tenant table has tenant_id uuid NOT NULL and a row level security policy on current_setting('app.tenant_id', true)::uuid. ClickHouse usage_events is the analytics replica of eai.usage_events (ReplacingMergeTree keyed by event_id).

Contract: contracts/v1/multimodal-contract.schema.json (schema_version 1.0). Usage is always reported with contracts/v1/usage-event.schema.json, audit with contracts/v1/audit-event.schema.json.

API endpoints​

MethodPath
POST/v1/multimodal/video/process
GET/v1/multimodal/jobs
GET/v1/multimodal/jobs/{id}

POST /v1/multimodal/video/process​

Request:

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

Response:

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

GET /v1/multimodal/jobs​

Response:

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

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

Response:

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

SDK examples​

Python​

from enterprise_ai import EnterpriseAI

client = EnterpriseAI(base_url="<control-plane-url>", api_key="eai_live_xxx", tenant="acme", locale="en")
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: "en" });
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.ENGLISH)
    .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("en"))
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​

Environment variables: EAI_MEDIA_MAX_BYTES, EAI_NATS_URL.

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

Security​

  • FFmpeg runs without network, read-only root filesystem, CPU and memory limits, gVisor runtime class.
  • Only an allowlist of containers and codecs is accepted.
  • Tenant resolved from the credential, never from the body; PostgreSQL row level security on every tenant table.
  • Every mutation and policy decision writes a hash chained audit event (G-11).
  • Errors use the standard envelope with message_i18n en and fr.

Observability​

  • Job spans linked to the upload trace through traceparent stored on the job.
  • W3C traceparent propagated, X-Request-Id returned, spans carry eai.tenant_id, eai.project, eai.agent_id, eai.run_id.

FinOps units​

  • Recorded as cpu_seconds, bytes_transferred and storage_gb_hours on the media; video_seconds processed in labels.

Failure modes and resilience​

  • Corrupted file -> job failed with validation_error, original kept for audit.
  • Worker crash -> message redelivered by JetStream, idempotent outputs keyed by job id and segment.

Standalone adoption​

A client can use the media workers alone for compliant video preprocessing before its own analytics.

Acceptance tests​

  1. A 10 minute MP4 produces 600 frames at 1 fps and one audio media.
  2. Killing a worker mid-job completes the job once, without duplicate child media.

Implementation roadmap