How It Works: Media Ingestion Flow#

Overview#

This document provides a comprehensive visual flow of the media ingestion process in the Multimodal DataPrep microservice, from initial upload through enrichment, embedding generation, and final storage in the vector database. It covers two media kinds:

  • Video — the deep-dive that follows (frame extraction → object detection → parallel batch embedding → storage), optimized for performance with parallel processing, batch operations, and the in-process embedding pipeline. See Detailed Video Ingestion Flow.

  • Image — a simpler, frame-less path that embeds the image directly (plus optional object-detection crops). See Image Ingestion Flow.

Both kinds converge on the same embedding model, the same shared vector collection, and the same storage/deduplication/metadata handling; records are distinguished by a content_type field (video / image). Storage and vector backends are pluggable (MinIO/local, VDMS/Milvus); the diagrams below label VDMS/MinIO as the default example.

High-Level Architecture#

        ---
config:
  theme: dark

---
graph TB
    subgraph "Entry Points"
        A1[POST /media/upload<br/>Direct File Upload]
        A2[POST /media/process<br/>Process from MinIO]
    end

    subgraph "Core Processing Pipeline"
        C[Frame Extraction]
        D[Object Detection]
        E[Batch Creation]
        F[Parallel Processing]
        G[Embedding Generation]
        H[Vector DB Storage]
    end

    subgraph "Storage Layer"
        I[(VDMS Vector Database)]
        J[(MinIO Object Storage)]
    end

    A1 --> J
    A1 --> C
    A2 --> J
    A2 --> C
    C --> D
    D --> E
    E --> F
    F --> G
    G --> H
    H --> I

Detailed Video Ingestion Flow#

Stage 1: Video Upload & Initial Processing#

        ---
config:
  theme: dark

---
flowchart TD
    START([Video Upload Request]) --> ENTRY{Entry Point?}

    ENTRY -->|Direct Upload| UPLOAD[POST /media/upload<br/>File: video.mp4<br/>Params: frame_interval, enable_detection]
    ENTRY -->|MinIO Reference| MINIO[POST /media/process<br/>Params: bucket_name, video_id]

    UPLOAD --> VALIDATE1[Validate File<br/>- Check MP4 format<br/>- Check size limit 500MB<br/>- Validate parameters]
    MINIO --> VALIDATE2[Validate MinIO Path<br/>- Check bucket exists<br/>- Verify video_id directory<br/>- Resolve stored video]

    VALIDATE1 --> STORE[Store Video to MinIO<br/>Path: bucket/video_id/filename.mp4]
    VALIDATE2 --> DOWNLOAD[Download from MinIO<br/>Get video content]

    STORE --> CONFIG[Load Configuration<br/>• frame_interval default: 15<br/>• enable_object_detection: true<br/>• detection_confidence: 0.85]
    DOWNLOAD --> CONFIG

    CONFIG --> FRAME_EXTRACT[Stage 2: Frame Extraction<br/>✓ In-process embeddings<br/>✓ Memory-based processing<br/>✓ OpenVINO optimized]

Key Decisions:

  • Entry Point Selection: Direct upload saves to MinIO first; MinIO processing retrieves from existing storage

  • Configuration Priority: Request params → Config file defaults → Service defaults

Performance Factors:

  • File validation is minimal overhead (~ms)

  • MinIO upload/download depends on video size and network

  • In-process embedding eliminates network latency

Stage 2: Frame Extraction & Metadata Creation#

        ---
config:
  theme: dark

---
flowchart TD
    START[Frame Extraction Stage] --> VIDEO_INFO[Read Video Information<br/>Using Decord VideoReader]

    VIDEO_INFO --> CALC[Calculate Video Metrics<br/>• Total Frames: len vr<br/>• FPS: vr.get_avg_fps<br/>• Duration: frames / fps]

    CALC --> INTERVAL[Calculate Frame Indices<br/>frame_indices = range 0, total_frames, frame_interval<br/><br/>Example: 900 frames, interval=15<br/>→ Extract 60 frames]

    INTERVAL --> EXTRACT_LOOP{For each<br/>frame_index}

    EXTRACT_LOOP --> SEEK[Seek to Frame<br/>vr at frame_idx]

    SEEK --> CONVERT[Convert Tensor to NumPy<br/>• Handle decord NDArray<br/>• Ensure uint8 format<br/>• Verify shape H,W,C]

    CONVERT --> META["Create Frame Metadata<br/>frame_metadata = <br/>• frame_id: video_id_framenum<br/>• frame_number<br/>• timestamp: frame_idx/fps<br/>• frame_type: 'full_frame'<br/>• content_type: 'video'<br/>• video_id, filename, bucket<br/>• video_url, video_rel_url<br/>• source_path (dir ingest)<br/>• tags, user metadata<br/>• fps, total_frames, duration"]

    META --> STORE_FRAME[Store Frame Array<br/>frames.append frame_numpy<br/>frames_metadata.append metadata]

    STORE_FRAME --> MORE{More frames?}

    MORE -->|Yes| EXTRACT_LOOP
    MORE -->|No| COMPLETE[Extraction Complete<br/>Result: List of numpy arrays + metadata]

    COMPLETE --> DETECT_STAGE[Stage 3: Object Detection]

Optimization Highlights:

  1. Decord Library: GPU-capable video reading (though currently using CPU context for reliability)

  2. Memory Efficiency: Frames stored as numpy arrays, not written to disk between stages

  3. Metadata Richness: Comprehensive frame metadata for search/retrieval

Performance Metrics:

  • Frame Extraction Time: Typically 0.5-2s for 60 frames from 30s video

  • Memory Usage: ~4MB per 1080p frame (uncompressed)

  • Extraction Rate: ~30-100 frames/second depending on video resolution

Stage 3: Object Detection (Optional)#

        ---
config:
  theme: dark

---
flowchart TD
    START[Object Detection Stage] --> CHECK{Object Detection<br/>Enabled?}

    CHECK -->|Disabled| SKIP[Skip Detection<br/>Use full frames only<br/>frames remain unchanged]
    CHECK -->|Enabled| INIT[Initialize YOLOX Detector<br/>• Model: yolox_s<br/>• Device: CPU/GPU<br/>• Confidence: 0.85<br/>• Input Size: 640x640<br/>• NMS Threshold: 0.45]

    INIT --> BATCH_DETECT[Create Detection Batches<br/>detection_batch_size = 32<br/><br/>Example: 60 frames → 2 batches]

    BATCH_DETECT --> PARALLEL_DETECT{Process Batches<br/>in Parallel}

    PARALLEL_DETECT --> BATCH_PROC[For Each Batch]

    BATCH_PROC --> FRAME_LOOP{For each frame<br/>in batch}

    FRAME_LOOP --> CONVERT_PIL[Convert Frame to PIL<br/>Image.fromarray frame_numpy]

    CONVERT_PIL --> DETECT[Run YOLOX Detection<br/>detector.detect frame_pil]

    DETECT --> ADD_FULL[Add Full Frame<br/>frame_type: full_frame<br/>Keep original frame metadata]

    ADD_FULL --> CROPS{Detections<br/>Found?}

    CROPS -->|Yes| CROP_LOOP{For each<br/>detection}

    CROP_LOOP --> EXTRACT_CROP[Extract Crop<br/>• Validate bbox coordinates<br/>• Crop: frame y1:y2, x1:x2<br/>• Convert to PIL Image<br/>• Min size: 10x10 pixels]

    EXTRACT_CROP --> CROP_META[Create Crop Metadata<br/>crop_metadata = frame_metadata + <br/>• frame_type: detected_crop<br/>• is_detected_crop: true<br/>• crop_index<br/>• detection_confidence<br/>• crop_bbox x1 y1 x2 y2<br/>• detected_class_id<br/>• detected_label]

    CROP_META --> ADD_CROP[Add Crop to Results<br/>all_images.append crop_pil<br/>all_metadata.append crop_meta]

    ADD_CROP --> MORE_CROPS{More crops?}
    MORE_CROPS -->|Yes| CROP_LOOP
    MORE_CROPS -->|No| MORE_FRAMES

    CROPS -->|No| MORE_FRAMES{More frames?}

    MORE_FRAMES -->|Yes| FRAME_LOOP
    MORE_FRAMES -->|No| BATCH_DONE[Batch Complete]

    BATCH_DONE --> MORE_BATCHES{More batches?}
    MORE_BATCHES -->|Yes| PARALLEL_DETECT
    MORE_BATCHES -->|No| DETECTION_DONE[Detection Complete<br/><br/>Example Expansion:<br/>60 frames → 240 items<br/>60 full frames + 180 crops<br/>avg 3 objects/frame]

    SKIP --> BATCH_CREATE
    DETECTION_DONE --> BATCH_CREATE[Stage 4: Batch Creation]

Object Detection Details:

Model Specifications:

  • Architecture: YOLOX-S (small variant)

  • Framework: OpenVINO IR format

  • Input Resolution: 640x640 (preprocessed)

  • Classes: 80 COCO categories (person, car, bicycle, etc.)

  • Download: Auto-downloaded from GitHub on first use

Detection Process:

  1. Preprocessing: Image resized to 640x640, normalized

  2. Inference: OpenVINO execution on CPU/GPU

  3. Postprocessing: NMS filtering with threshold 0.45

  4. Crop Extraction: Bounding boxes validated and extracted

Performance Characteristics:

  • Detection Speed: ~50-100ms per frame (CPU), ~10-20ms (GPU)

  • Parallel Batches: 2-4 detection workers typical

  • Expansion Factor: 1.5x to 5x (avg 3 objects/frame)

  • Memory Impact: +20-50% for crop storage

Optimization Strategy:

  • Batched detection reduces overhead

  • Parallel processing utilizes multi-core CPUs

  • Global detector instance reused (no reload per request)

  • Crops validated (min 10x10px, valid coordinates)

Stage 4: Batch Creation for Parallel Processing#

        ---
config:
  theme: dark

---
flowchart TD
    START[Batch Creation Stage] --> INPUT[Input: List of Images + Metadata<br/><br/>After Detection:<br/>• Full frames: 60<br/>• Detected crops: 180<br/>• Total items: 240]

    INPUT --> CONFIG[Get Pipeline Configuration<br/>Based on CPU cores and mode]

    CONFIG --> CALC_WORKERS[Calculate Worker Count<br/><br/>OpenVINO Mode:<br/>workers = max 1, cpu_cores // 4<br/>Limited by OV_NUM_STREAMS<br/><br/>PyTorch Mode:<br/>workers = cpu_cores // 16<br/>max 8 workers]

    CALC_WORKERS --> CALC_BATCH[Determine Batch Size<br/>batch_size = 32<br/>Optimal for embedding generation]

    CALC_BATCH --> CREATE_BATCHES[Create Processing Batches<br/>Split items into batches<br/><br/>Example:<br/>240 items ÷ 32 = 7.5<br/>→ 8 batches<br/>7 batches of 32 items<br/>1 batch of 16 items]

    CREATE_BATCHES --> BATCH_STRUCT[Batch Structure<br/>Each batch contains:<br/>• List of PIL Images<br/>• List of metadata dicts<br/>• Batch index<br/>• Total batch count]

    BATCH_STRUCT --> SUBMIT[Submit to Thread Pool<br/>ThreadPoolExecutor<br/>max_workers = pipeline_count<br/><br/>Example: 8 batches → 4 workers<br/>2 batches processed simultaneously]

    SUBMIT --> PARALLEL[Stage 5: Parallel Processing]

Pipeline Configuration Logic:

# Pseudo-code for worker calculation
cpu_cores = multiprocessing.cpu_count()

if use_openvino:
    base_workers = max(1, cpu_cores // 4)

    # Check OpenVINO environment variables
    ov_limit = check_env_vars([
        'OV_PERFORMANCE_HINT_NUM_REQUESTS',
        'PERFORMANCE_HINT_NUM_REQUESTS',
        'OV_NUM_STREAMS'
    ])

    if ov_limit:
        workers = min(base_workers, ov_limit)
    else:
        workers = base_workers
else:
    # PyTorch is more CPU-intensive
    workers = min(max(1, cpu_cores // 16), 8)

batch_size = 32  # Fixed optimal size

Configuration Examples:

CPU Cores

OpenVINO Mode

PyTorch Mode

8 cores

2 workers

1 worker

16 cores

4 workers

1 worker

32 cores

8 workers

2 workers

64 cores

16 workers

4 workers

96 cores

24 workers

6 workers

Batch Size Considerations:

  • 32 items per batch: Optimal balance for embedding models

  • Smaller batches: Lower memory, higher overhead

  • Larger batches: Higher memory, better throughput

  • Dynamic adjustment: Future enhancement based on available memory

Stage 5: Parallel Batch Processing Pipeline#

        ---
config:
  theme: dark

---
flowchart TD
    START[Parallel Processing Stage] --> POOL[Thread Pool Executor<br/>max_workers = pipeline_count<br/><br/>Example: 4 workers<br/>Processing 8 batches]

    POOL --> SUBMIT_ALL[Submit All Batches<br/>All batches queued<br/>Workers pick up jobs dynamically]

    SUBMIT_ALL --> WORKER1[Worker 1<br/>Process Batch]
    SUBMIT_ALL --> WORKER2[Worker 2<br/>Process Batch]
    SUBMIT_ALL --> WORKER3[Worker 3<br/>Process Batch]
    SUBMIT_ALL --> WORKER4[Worker 4<br/>Process Batch]

    WORKER1 --> BATCH_PROC1[Single Batch Processing]
    WORKER2 --> BATCH_PROC1
    WORKER3 --> BATCH_PROC1
    WORKER4 --> BATCH_PROC1

    subgraph "Single Batch Processing Pipeline"
        BATCH_PROC1[Start Batch Processing<br/>Input: 32 images + metadata]

        BATCH_PROC1 --> STEP1[Step 1: Validation<br/>Check model supports images<br/>Skip if text-only model]

        STEP1 --> STEP2[Step 2: Generate Embeddings<br/>In-process function call]

        STEP2 --> EMB_DETAIL[Embedding Generation Details]

        subgraph "Embedding Generation"
            EMB_DETAIL --> EMB_LOCAL_PROC[In-process Embedding<br/>• Use global embedding client<br/>• Thread-safe infer_new_request<br/>• No HTTP overhead<br/>• OpenVINO optimized<br/>• Batch processing inside model]

            EMB_LOCAL_PROC --> EMB_RESULT[Embedding Results<br/>Vector dimensions: 512/768/1024<br/>Format: List of float arrays]
        end

        EMB_RESULT --> STEP3[Step 3: Validate Embeddings<br/>Filter out null/failed embeddings<br/>Match embeddings to metadata]

        STEP3 --> STEP4[Step 4: Store to Vector DB<br/>Immediate batch storage<br/>Prevents OutOfJournalSpace]

        STEP4 --> STORAGE_DETAIL[VDMS Storage Details]
    end

    subgraph "VDMS Vector DB Storage"
        STORAGE_DETAIL --> VDB_PREP[Prepare Storage Request<br/>• Embeddings: List of vectors<br/>• Metadata: List of dicts<br/>• Collection: MM_DATAPREP_DB_COLLECTION]

        VDB_PREP --> VDB_BULK[Bulk Insert Operation<br/>AddEntity with AddDescriptor<br/>Batch operation more efficient]

        VDB_BULK --> VDB_INDEX[Vector Index Update<br/>VDMS updates HNSW index<br/>Enables similarity search]

        VDB_INDEX --> VDB_RETURN[Return Stored IDs<br/>List of entity IDs<br/>Used for verification]
    end

    VDB_RETURN --> BATCH_DONE[Batch Complete<br/>Return results:<br/>• embeddings_count<br/>• stored_ids<br/>• processing_time<br/>• detection_time<br/>• embedding_time<br/>• storage_time]

    BATCH_DONE --> COLLECTOR[Results Collector<br/>as_completed future]

    COLLECTOR --> MORE{More batches<br/>pending?}

    MORE -->|Yes| COLLECTOR
    MORE -->|No| AGGREGATE[Aggregate All Results<br/>Sum embeddings<br/>Combine stored_ids<br/>Calculate statistics]

    AGGREGATE --> FINAL[Stage 6: Final Results]

Parallel Processing Characteristics:

Thread Pool Execution:

  • Dynamic Work Distribution: Batches processed as workers become available

  • True Parallelism: Multiple batches processed simultaneously

  • Completion Order: Batches complete in any order (not sequential)

  • Timeout Protection: 300s per batch maximum

Embedding Generation Characteristics:

Aspect

In-process pipeline

Method

Direct function call

Network

None

Serialization

None (PIL objects)

Latency

~50-200ms/batch

Memory

Shared with service

Optimization

OpenVINO compiled when enabled

Vector DB Storage Strategy:

Why Immediate Batch Storage:

  • Prevents Memory Overflow: Storing after each batch prevents accumulation

  • Protects Against Failures: Partial results saved even if pipeline fails

  • Avoids VDMS Issues: Prevents OutOfJournalSpace errors

  • Progress Tracking: Stored IDs returned incrementally

Bulk Insert Benefits:

  • Reduced Overhead: Single VDMS transaction per batch

  • Index Efficiency: VDMS can optimize index updates

  • Faster Than Individual: ~10x faster than per-item inserts

Stage 6: Results Aggregation & Performance Metrics#

        ---
config:
  theme: dark

---
flowchart TD
    START[Results Aggregation] --> COLLECT[Collect All Batch Results<br/>From ThreadPoolExecutor.as_completed]

    COLLECT --> AGGREGATE[Aggregate Statistics]

    subgraph "Statistics Calculation"
        AGGREGATE --> COUNT[Total Embeddings<br/>Sum all batch counts<br/><br/>Example: 8 batches<br/>7×32 + 1×16 = 240 embeddings]

        COUNT --> IDS[Stored IDs<br/>Concatenate all ID lists<br/>Verify no duplicates]

        IDS --> TIMES[Processing Times<br/>Calculate per-stage statistics]

        TIMES --> TIME_DETAIL[Time Breakdown]

        subgraph "Time Statistics"
            TIME_DETAIL --> EXTRACT_TIME[Frame Extraction<br/>Total time: e.g., 1.2s<br/>Frames extracted: 60]

            EXTRACT_TIME --> DETECT_TIME["Object Detection<br/>Avg per batch: e.g., 0.8s<br/>Max per batch: e.g., 1.2s<br/>% of batch time: e.g., 25%"]

            DETECT_TIME --> EMBED_TIME["Embedding Generation<br/>Avg per batch: e.g., 2.1s<br/>Max per batch: e.g., 3.5s<br/>% of batch time: e.g., 65%"]

            EMBED_TIME --> STORE_TIME["Vector DB Storage<br/>Avg per batch: e.g., 0.3s<br/>Max per batch: e.g., 0.5s<br/>% of batch time: e.g., 10%"]

            STORE_TIME --> TOTAL_TIME[Total Pipeline Time<br/>Wall clock time: e.g., 8.5s<br/>Sequential would be: ~28s<br/>Speedup: 3.3x]
        end
    end

    TIME_DETAIL --> FRAME_COUNTS[Frame Count Summary]

    subgraph "Frame Flow Tracking"
        FRAME_COUNTS --> INPUT_FRAMES[Input Frames<br/>Extracted from video: 60]

        INPUT_FRAMES --> POST_DETECT[Post-Detection Items<br/>After object detection: 240<br/>Expansion: 4x]

        POST_DETECT --> STORED_EMBS["Stored Embeddings<br/>Successfully stored: 240<br/>Success rate: 100%"]
    end

    STORED_EMBS --> BATCH_STATS[Batch Statistics]

    subgraph "Batch Performance"
        BATCH_STATS --> BATCH_COUNT[Batches Processed<br/>Total: 8 batches]

        BATCH_COUNT --> AVG_BATCH[Average Batch Time<br/>e.g., 3.5s per batch]

        AVG_BATCH --> MAX_BATCH[Max Batch Time<br/>e.g., 4.2s slowest batch]

        MAX_BATCH --> EFFICIENCY["Processing Efficiency<br/>Parallel overhead: ~15%<br/>Resource utilization: 85%"]
    end

    EFFICIENCY --> RETURN_RESULT[Return Final Result]

    subgraph "API Response"
        RETURN_RESULT --> RESPONSE[HTTP Response<br/>Status: 201 CREATED<br/>Message: Embeddings created successfully]

        RESPONSE --> RESPONSE_BODY[Response Body Example]

        RESPONSE_BODY --> JSON[JSON Response:<br/>status: success<br/>message: Embeddings created<br/>total_embeddings: 240<br/>total_frames_processed: 60<br/>frame_interval: 15<br/>timing: extraction, parallel, storage<br/>frame_counts: extracted, detected, stored]
    end

    JSON --> COMPLETE([Processing Complete])

Performance Metrics Explained:

Time Statistics:

  1. Frame Extraction Time: Time to read video and extract frames

    • Depends on: Video size, resolution, codec, storage speed

    • Typical: 0.5-3s for 60 frames

  2. Detection Time per Batch: Time for object detection per batch

    • Depends on: Frame resolution, object count, CPU/GPU speed

    • Typical: 0.5-2s per batch of 32 items

  3. Embedding Time per Batch: Time to generate embeddings

    • Depends on: Model size, device, batch size

    • Typical: 1-3s per batch

  4. Storage Time per Batch: Time to store in VDMS

    • Depends on: Batch size, VDMS load, network latency

    • Typical: 0.2-0.8s per batch

  5. Pipeline Wall Time: Total end-to-end time

    • Benefits from parallelization

    • Typical speedup: 2-4x vs sequential

Frame Flow Tracking:

The system tracks three critical counts:

  1. Extracted Frames: Original frames from video (e.g., 60)

  2. Post-Detection Items: After adding crops (e.g., 240 = 60 + 180 crops)

  3. Stored Embeddings: Successfully stored in vector DB (e.g., 240)

Efficiency Calculations:

Expansion Factor = Post-Detection Items / Extracted Frames
                 = 240 / 60 = 4x

Success Rate = Stored Embeddings / Post-Detection Items
             = 240 / 240 = 100%

Theoretical Sequential Time = Batches × Max Batch Time
                            = 8 × 4.2s = 33.6s

Actual Parallel Time = 8.5s

Speedup = 33.6s / 8.5s = 3.95x

Efficiency = (Theoretical Sequential / Workers) / Actual
           = (33.6s / 4) / 8.5s = 8.4s / 8.5s = 98.8%

Image Ingestion Flow#

Images take a deliberately frame-less path: there is no decode loop and no frame sampling. A single image is embedded once, optionally augmented with object-detection crops, and stored in the same shared vector collection as video records — tagged with content_type="image".

Entry points and transports#

Endpoint

Content type

Image source

POST /media/upload

multipart/form-data

binary file bytes

POST /media/ingest

application/json

inline base64 (type=image_base64) or remote URL (type=image_url)

POST /media/ingest/batch

application/json

list of base64/URL image sources (async job)

POST /media/process

application/json

image already stored in object storage

For base64/URL inputs the decoded bytes are the trust boundary: the real format is sniffed from the bytes to derive the stored extension (the client-declared type is never trusted for the stored file), and the size is capped before any processing.

Stage flow#

        ---
config:
  theme: dark

---
graph TB
    subgraph "Entry"
        U1[POST /media/upload<br/>multipart bytes]
        U2[POST /media/ingest<br/>base64 / URL]
    end

    subgraph "Resolve & Validate"
        R1[Decode base64 / download URL]
        R2[Sniff real format<br/>derive extension]
        R3[Enforce size cap]
        R4{Dedup enabled?}
        R5[SHA-256 + hash marker<br/>409 on duplicate]
    end

    subgraph "Persist Asset"
        S1[(Object Storage<br/>MinIO / local)]
    end

    subgraph "Embed"
        E1[Full-image embed<br/>frame_type=full_frame]
        E2{Object detection?}
        E3[Per-crop embed]
    end

    subgraph "Store Vectors"
        V1[Attach metadata:<br/>content_type=image,<br/>download URL, tags]
        V2[(Vector DB<br/>VDMS / Milvus)]
    end

    U1 --> R2
    U2 --> R1 --> R2
    R2 --> R3 --> R4
    R4 -- yes --> R5 --> S1
    R4 -- no --> S1
    S1 --> E1
    E1 --> E2
    E2 -- enabled --> E3 --> V1
    E2 -- disabled --> V1
    E1 --> V1
    V1 --> V2

How it differs from the video flow#

Aspect

Video

Image

Decode / frame sampling

Yes — every Nth frame (MM_DATAPREP_FRAME_INTERVAL)

None — single image

Records produced

1 full frame + N crops per sampled frame

1 full image + optional crops

Timestamp metadata

Per-frame timestamp

Single (upload-time) reference

content_type

video

image

Base64 / URL transport

No

Yes (/media/ingest)

Parallel batch pipeline

Per-frame fan-out

Batched across images in a job

Object detection

Optional, per frame

Optional, per image

Storage / dedup / download

Shared

Shared (identical)

Because images reuse the same embedding model and vector contract as video frames, image and video records are directly comparable in a single similarity search, enabling cross-modal retrieval.

Complete End-to-End Flow Visualization#

        ---
config:
  theme: dark

---
graph TB
    subgraph "Stage 1: Video Upload"
        A[Video Upload<br/>POST /media/upload or /media/process] --> C[Memory Processing]
    end

    subgraph "Stage 2: Frame Extraction"
        C --> E[Extract Frames<br/>Decord VideoReader]
        E --> F[Frame List<br/>60 frames @ interval=15<br/>Time: 1.2s]
    end

    subgraph "Stage 3: Object Detection Optional"
        F --> G{Object Detection<br/>Enabled?}
        G -->|Yes| H[YOLOX Detection<br/>Parallel batches<br/>Time: 2.4s total]
        G -->|No| I[Skip Detection]
        H --> J[Expanded List<br/>240 items<br/>60 frames + 180 crops]
        I --> K[Original List<br/>60 items]
    end

    subgraph "Stage 4: Batch Creation"
        J --> L[Create Batches<br/>8 batches of 32 items<br/>batch_size=32]
        K --> L
        L --> M[Calculate Workers<br/>4 parallel workers<br/>OpenVINO optimized]
    end

    subgraph "Stage 5: Parallel Processing"
        M --> N[Submit to ThreadPool<br/>Process batches in parallel]

        N --> O1[Worker 1: Batch 1<br/>32 items → embeddings → store]
        N --> O2[Worker 2: Batch 2<br/>32 items → embeddings → store]
        N --> O3[Worker 3: Batch 3<br/>32 items → embeddings → store]
        N --> O4[Worker 4: Batch 4<br/>32 items → embeddings → store]

        O1 --> P[Collect Results<br/>as batches complete]
        O2 --> P
        O3 --> P
        O4 --> P
    end

    subgraph "Stage 6: Storage & Results"
        P --> Q[Aggregate Statistics<br/>240 embeddings stored<br/>Pipeline time: 8.5s]
        Q --> R[(VDMS Vector DB<br/>240 vectors indexed)]
        Q --> S[Return Response<br/>Status: 201 CREATED<br/>Details: timing + counts]
    end

Performance Optimization Summary#

Critical Optimizations#

  1. Parallel Processing

    • Implementation: ThreadPoolExecutor with dynamic worker count

    • Impact: 3-4x speedup vs sequential processing

    • Configuration: Auto-calculated based on CPU cores and runtime backend

  2. Batch Storage

    • Implementation: Store after each batch instead of all at end

    • Impact: Prevents memory overflow and VDMS journal errors

    • Benefit: Fault tolerance - partial results saved

  3. In-process embedding

    • Implementation: Direct function calls to the embedding package

    • Impact: Eliminates HTTP/network overhead

  4. Object Detection Optimization

    • Implementation: Global detector instance, parallel detection batches

    • Impact: Avoids model reload, utilizes multi-core CPUs

    • Caching: Detector initialized once, reused across requests

  5. Memory-Based Processing

    • Implementation: Process video directly from bytes in memory

    • Impact: Eliminates disk I/O overhead

    • Benefit: Lower latency, reduced disk wear

Configuration Parameters#

Parameter

Default

Impact

Tuning Guide

frame_interval

15

Frame extraction density

Lower = more frames (slower, more detail)

batch_size

32

Items per embedding batch

Fixed optimal value

pipeline_count

auto

Parallel workers

CPU cores ÷ 4 (OpenVINO) or ÷ 16 (PyTorch)

detection_confidence

0.85

Object detection threshold

Higher = fewer detections (faster)

enable_object_detection

true

Crop extraction

Disable for 4x fewer embeddings (faster)

Performance Expectations#

Example Video: 30 seconds, 30fps (900 frames), 1920x1080

Configuration

Frames Extracted

Items After Detection

Total Time

Speedup

interval=15, detection=ON, in-process

60

240 (avg 3 crops/frame)

8.5s

3.95x

interval=15, detection=OFF, in-process

60

60

4.2s

4.2x

interval=30, detection=ON, in-process

30

120

4.8s

3.8x

Timing Breakdown (Detection ON):

  • Frame Extraction: 1.2s (14%)

  • Object Detection: 2.4s (28%)

  • Embedding Generation: 4.2s (49%)

  • Vector DB Storage: 0.7s (8%)

  • Total Pipeline: 8.5s (100%)

System Architecture Context#

Component Interaction#

        ---
config:
  theme: dark

---
graph LR
    subgraph "Multimodal DataPrep Microservice"
        A[FastAPI Endpoints] --> B[Video Processing]
        B --> C[Frame Extraction]
        C --> D[Object Detection<br/>YOLOX]
        D --> E[Embedding Helper]
    end

    subgraph "Embedding Package"
        G[In-process Direct Functions]
    end

    subgraph "Storage Layer"
        H[(VDMS Vector DB)]
        I[(MinIO Object Storage)]
    end

    E --> G
    G --> H
    A --> I

Service Dependencies#

  1. VDMS Vector Database

    • Stores embeddings with metadata

    • Provides similarity search

    • HNSW index for fast retrieval

  2. MinIO Object Storage

    • Stores original videos

    • Bucket-based organization

    • Provides video download URLs

  3. Multimodal Embedding Package

    • Generates embeddings from images

    • OpenVINO or PyTorch backend

  4. YOLOX Object Detection

    • OpenVINO IR format

    • CPU/GPU inference

    • Auto-downloads model files

Troubleshooting & Monitoring#

Key Metrics to Monitor#

  1. Frame Extraction Rate: Frames/second during extraction

    • Target: >30 frames/second

    • Alert: <10 frames/second indicates I/O bottleneck

  2. Object Detection Time: Average seconds per batch

    • Target: <2s per batch of 32 items

    • Alert: >5s indicates CPU/GPU overload

  3. Embedding Generation Time: Average seconds per batch

    • Target: <3s per batch

    • Alert: >10s indicates model/network issues

  4. Storage Time: Average seconds per batch

    • Target: <1s per batch

    • Alert: >3s indicates VDMS performance issues

  5. Success Rate: Stored embeddings / Expected embeddings

    • Target: >95%

    • Alert: <90% indicates embedding failures

Common Performance Issues#

Issue

Symptom

Cause

Solution

Slow extraction

High frame_extraction_time

Large video, slow storage

Use faster storage, reduce resolution

Detection bottleneck

High detection_time

CPU overload

Enable GPU, reduce confidence threshold

Embedding slowdown

High embedding_time

Model overload

Increase workers, enable OpenVINO

Storage delays

High storage_time

VDMS overload

Check VDMS resources, reduce batch size

Memory errors

Process killed

Too many workers

Reduce pipeline_count

Logs to Check#

# Frame extraction progress
"Extracting X frames with interval Y"
"Frame extraction completed in Xs: Y frames"

# Object detection status
"Processing object detection in batches of X"
"Detection batch processed: Y frames → Z items"

# Parallel processing progress
"Processing X frames with Y maximum parallel workers"
"Batch N/M completed: X embeddings stored"

# Final results
"Embedding generation pipeline completed in Xs: Y embeddings across Z batches"
"Frame flow summary: extracted=X -> after_detection=Y -> stored=Z"

Conclusion#

The Multimodal DataPrep ingestion pipeline is a highly optimized system that efficiently processes both video and images for semantic search. The video path (detailed above) applies frame extraction and parallel batch embedding; the image path embeds directly. Both converge on the same embedding model and shared vector collection. Key achievements:

  • Parallel Processing: 3-4x speedup through multi-threaded execution

  • Batch Storage: Prevents memory overflow with incremental saves

  • In-process Optimization: Eliminates HTTP overhead

  • Object Detection: Expands search coverage with detected crops

  • Memory Efficiency: Direct memory processing between stages

  • Comprehensive Metrics: Detailed timing and statistics for optimization

Total Processing Time: 8-10 seconds for a 30-second video (detection ON)

Scalability: Handles videos up to 500MB, auto-configures workers based on available resources

References#

  • Source Code: /home/sdp/workbench/integration/edge-ai-libraries-mme-v2/microservices/visual-data-preparation-for-retrieval/multimodal-dataprep/

  • Configuration: src/config.yaml

  • API Endpoints: src/endpoints/video_processing/

  • Core Processing: src/core/embedding/embedding_helper.py

  • Object Detection: src/core/object_detection/detector.py

  • Video Utils: src/core/utils/video_utils.py