Multi-Modal AI Pipelines | The Garnet Wiki

Multi-modal AI pipelines process and correlate data from multiple input types—text, images, audio, video, sensor streams, and structured tabular data—within a single, cohesive computational workflow. They are essential when a single modality cannot capture the full context of a problem: for example, diagnosing a medical condition from patient records, X-rays, and voice recordings; or understanding a video presentation with spoken commentary and on-screen text. The pipeline orchestrates feature extraction, alignment, fusion, and inference across modalities, enabling models that reason across inputs with shared semantics.

Feature Extraction Across Modalities

Each modality requires distinct preprocessing and embedding extraction. For image data, use a pre-trained vision transformer such as ViT-B/16 via the torchvision library:

import torch
from torchvision import transforms
from PIL import Image

transform = transforms.Compose([
    transforms.Resize(224),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])

image = Image.open("scan.png")
input_tensor = transform(image).unsqueeze(0)  # Add batch dimension
with torch.no_grad():
    features = vision_model(input_tensor)  # torch.Size([1, 768])

For audio, use torchaudio to extract Mel-frequency cepstral coefficients (MFCCs) from a WAV file:

import torchaudio

waveform, sample_rate = torchaudio.load("interview.wav")
mfccs = torchaudio.transforms.MFCC(
    sample_rate=sample_rate,
    n_mfcc=40,
    log_mels=True,
    n_fft=2048,
    hop_length=512,
    n_mels=128,
)(waveform)

Text is processed with sentence-transformers using all-MiniLM-L6-v2:

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')
sentences = [
    "Patient reports chest pain for three days.",
    "ECG shows ST elevation in leads II, III, and aVF.",
    "Echo reveals reduced ejection fraction."
]
embeddings = model.encode(sentences)  # Shape: [3, 384]

All modalities must produce features of the same dimensionality and time alignment. A common failure is mismatched sequence lengths: audio MFCCs may be sampled at 40 Hz, while video frames arrive at 30 Hz. A silent error occurs when fusion layers assume synchronized inputs but receive misaligned tensors.

Temporal and Spatial Alignment

When modalities arrive at different rates, temporal alignment is critical. Use resample in torchaudio to align audio to video:

# Resample audio to 30 Hz to match video frame rate
resampler = torchaudio.transforms.Resample(orig_freq=sample_rate, new_freq=30)
aligned_audio = resampler(mfccs)  # [1, 40, T], T now matches video

For spatial alignment—e.g., aligning image regions to text spans—use bounding box annotations in JSON:

{
  "image_id": "scan-001",
  "objects": [
    {
      "label": "heart",
      "bbox": [120, 80, 300, 250],
      "text": "Cardiomegaly observed in the left atrium."
    },
    {
      "label": "lung",
      "bbox": [400, 100, 600, 300],
      "text": "Interstitial markings in the right lower lobe."
    }
  ]
}

Each object’s bounding box is used to crop a region from the image, and that crop is fed into a vision model alongside the full image. The alignment fails silently when bounding boxes are in pixel coordinates but the model expects normalized coordinates (0–1), leading to misaligned features.

Fusion combines features from multiple modalities into a joint representation. The simplest is concatenation:

# Assume features have shape [B, D] for each modality
image_features = vision_model(image_tensor)  # [B, 768]
text_features = sentence_model(sentences)     # [B, 384]
audio_features = mfcc_model(waveform)         # [B, 40, 128] → [B, 5120] via pooling

# Align audio to 768D via 1D convolution
audio_pooled = torch.nn.AdaptiveAvgPool1d(768)(audio_features.transpose(1, 2))  # [B, 768]
fusion = torch.cat([image_features, text_features, audio_pooled], dim=1)  # [B, 1936]

A more robust approach uses a cross-attention module, where one modality attends to the other:

class CrossModalAttention(nn.Module):
    def __init__(self, d_model=768, n_heads=8):
        super().__init__()
        self.attention = nn.MultiheadAttention(d_model, n_heads, batch_first=True)

    def forward(self, query, key, value):
        # query: [B, T_q, D], key/value: [B, T_k, D]
        out, _ = self.attention(query, key, value)
        return out

Here, the image features act as keys and values, and the text features are queries. The output is a sequence of cross-attended features where each text token attends to image regions.

A common error: using query = text, key = image, but forgetting to apply a modality-specific projection layer to align the query and key dimensions. The model fails silently when the text features (384D) are fed directly as queries into a 768D attention head without projection.

Pipeline Orchestration with Task Graphs

Orchestrate the full pipeline using a DAG of tasks with dependencies. Define tasks in a tasks.json file:

{
  "tasks": [
    {
      "name": "extract_image_features",
      "input": ["images/*.png"],
      "output": "features/image_features.pkl",
      "command": "python extract_image_features.py --input-dir images --output features/image_features.pkl",
      "depends": [],
      "resources": { "gpu": 1, "memory": "8GB" }
    },
    {
      "name": "extract_audio_features",
      "input": ["audio/*.wav"],
      "output": "features/audio_features.pkl",
      "command": "python extract_audio_features.py --input-dir audio --output features/audio_features.pkl",
      "depends": [],
      "resources": { "cpu": 4, "memory": "4GB" }
    },
    {
      "name": "fuse_features",
      "input": [
        "features/image_features.pkl",
        "features/audio_features.pkl",
        "features/text_features.pkl"
      ],
      "output": "features/fused_features.pt",
      "command": "python fuse_features.py --input features --output features/fused_features.pt",
      "depends": [
        "extract_image_features",
        "extract_audio_features",
        "extract_text_features"
      ],
      "resources": { "gpu": 1, "memory": "16GB" }
    }
  ]
}

Run the pipeline using a simple orchestrator:

# Run all tasks in dependency order
for task in $(jq -r '.tasks[].name' tasks.json); do
  task_json=$(jq -c --arg t "$task" '.tasks[] | select(.name == $t)' tasks.json)
  depends=$(echo "$task_json" | jq -r '.depends[]')
  if ! all_deps_completed $depends; then
    echo "Skipping $task: dependencies not ready"
    continue
  fi

  cmd=$(echo "$task_json" | jq -r '.command')
  $cmd
done

Failure modes:

A task’s input list contains files that do not exist—extract_image_features fails silently when images/ is empty.
The output file is written to a path that does not exist—mkdir -p must be part of the task.
resources are ignored: a GPU task runs on CPU because the orchestrator does not pass --gpu to python.

Distributed Execution with Shared Storage

When processing large-scale data—e.g., 100,000 medical scans and 50,000 audio interviews—use a shared file system (e.g., NFS, S3, or HDFS) as the pipeline’s data backbone.

# Mount S3 bucket as local storage
mount -t nfs s3://my-pipelines/data /data
# Or use AWS CLI with s3:// prefixes
aws s3 cp s3://my-pipelines/data/images/ ./images/

Each task reads from and writes to this shared location. The extract_image_features.py script reads from s3://my-pipelines/data/images/ and writes s3://my-pipelines/features/image_features.pkl.

Use rsync to keep outputs synchronized:

# After running a task
rsync -avz --delete /data/features/ s3://my-pipelines/features/

A common failure: a task reads from /data/images/ but the pipeline expects images/ in the current directory. The --input-dir flag is ignored, and the pipeline processes only the files in the working directory, not the mounted volume.

Error Handling and Monitoring

Each pipeline stage should emit structured logs:

{
  "timestamp": "2024-04-15T08:30:45Z",
  "task": "extract_audio_features",
  "status": "success",
  "input_files": 124,
  "output_file": "features/audio_features.pkl",
  "duration_seconds": 38.2,
  "metrics": {
    "audio_duration_seconds": 2540.1,
    "mean_mfcc_energy": 0.72,
    "feature_shape": [124, 40, 63]
  },
  "errors": []
}

If a task fails, the log includes an errors array:

{
  "task": "fuse_features",
  "status": "failed",
  "errors": [
    {
      "type": "DimensionMismatch",
      "message": "Image features (768D) and audio features (5120D) have incompatible dimensions.",
      "details": "Expected image features to be 5120D, but got 768D.",
      "traceback": "File \"fuse_features.py\", line 42, in forward\n  self.linear(image_features)",
      "timestamp": "2024-04-15T08:31:12Z"
    }
  ]
}

Monitor the pipeline using systemctl to run a health checker:

# /etc/systemd/system/pipeline-monitor.service
[Unit]
Description=Multi-modal pipeline monitor
After=network.target

[Service]
ExecStart=/usr/local/bin/pipeline-monitor --config /etc/pipeline-monitor.json
Restart=on-failure
User=ai-pipeline

[Install]
WantedBy=multi-user.target

The monitor polls the tasks.json file every 30 seconds, checks if all tasks have completed, and alerts on missing or failed outputs. It logs to journalctl:

journalctl -u pipeline-monitor.service -f

Common Pitfalls and Silent Failures

Feature dimension mismatch: A model expects 768D features, but the image encoder outputs 384D. No error is raised; the downstream model runs but with degraded accuracy. The fix requires a Linear(384, 768) layer, but it’s often omitted.
Misaligned time series: Audio MFCCs are 40 Hz, video frames 30 Hz. Without resampling, the fusion step treats each audio frame as a separate time step, but the model expects one feature per video frame.
Missing feature normalization: Features from different modalities are not normalized across the dataset. Text features have mean 0.5, image features have mean 0.1, and audio features are unnormalized. The fusion layer fails to combine them effectively.
Incomplete output paths: A task writes to features/output.pkl, but the orchestrator expects features/fused_features.pt. The pipeline runs, but the final model uses the wrong file.
Inconsistent metadata format: The pipeline expects image_id in the JSON metadata, but some files use img_id. The alignment step fails silently when joining image features with text.

Best Practices

Use a standard feature format—HDF5 or Parquet—to store and share features across tasks.
Define a schema for metadata:

  {
    "image_id": "string",
    "modality": "image|audio|text",
    "sample_rate": "float",
    "frame_rate": "float",
    "duration_seconds": "float",
    "features": "array of float"
  }

Ensure versioning of feature extractors: each extract_*_features.py script includes a --version flag and logs the version of the underlying model.
Use checksums to verify data integrity: compute sha256 of output files and store them in a checksums.json file.

These practices make the pipeline robust to changes in data, models, and infrastructure, ensuring that multi-modal AI systems remain reliable under real-world conditions.

This page was rewritten on 17 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.

Core Patterns in Multi-Modal Processing