Starphix
Agentic Engineering / Sources
The Lab / Sources / Agentic Engineering / ModelArchitectures/Multimodal-Foundations.md

Multimodal Foundations

Vision-language, audio, and unified multimodal architectures for agents.


Multimodal Architecture Taxonomy

Multimodal Models
├── Two-Tower (Dual Encoder)
│   ├── CLIP, SigLIP, ALIGN
│   └── Contrastive pretraining
├── Early Fusion (Single Encoder)
│   ├── Flamingo, BLIP-2, LLaVA
│   └── Vision tokens → LLM
├── Late Fusion (Separate → Merge)
│   ├── ImageBind, AudioCLIP
│   └── Modality-specific encoders
├── Native/Unified (Single Transformer)
│   ├── GPT-4o, Gemini 1.5, Chameleon
│   └── Unified tokenizer for all modalities
└── Mixture of Experts (MoE)
    ├── MoE per modality
    └── Cross-modal routing

Vision-Language Models (VLMs)

CLIP Family (Contrastive)

# Two-tower: Image encoder + Text encoder
# Loss: Contrastive (match image-text pairs)

from transformers import CLIPModel, CLIPProcessor

model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14")

# Embedding
inputs = processor(text=["a photo of a cat"], images=image, return_tensors="pt")
outputs = model(**inputs)
image_emb = outputs.image_embeds
text_emb = outputs.text_embeds

Use cases: Retrieval, classification, zero-shot

Flamingo / BLIP-2 (Early Fusion)

# Vision encoder → Perceiver Resampler → LLM
# Frozen vision encoder + frozen LLM + trained adapter

from transformers import Blip2ForConditionalGeneration, Blip2Processor

model = Blip2ForConditionalGeneration.from_pretrained("Salesforce/blip2-opt-2.7b")
processor = Blip2Processor.from_pretrained("Salesforce/blip2-opt-2.7b")

inputs = processor(images=image, text="Question: what is this?", return_tensors="pt")
generated_ids = model.generate(**inputs, max_new_tokens=50)

LLaVA / LLaVA-NeXT (Instruction Tuned)

# Vicuna + CLIP-ViT + Projector
# Two-stage: Pretrain projector → Instruction tune

from llava.model.builder import load_pretrained_model
from llava.mm_utils import get_model_name_from_path

tokenizer, model, image_processor, context_len = load_pretrained_model(
    model_path="liuhaotian/llava-v1.5-7b",
    model_base=None,
    model_name=get_model_name_from_path("liuhaotian/llava-v1.5-7b")
)

# Inference
from llava.mm_utils import process_images, tokenizer_image_token

image_tensor = process_images([image], image_processor, model.config)
input_ids = tokenizer_image_token(prompt, tokenizer, IMAGE_TOKEN_INDEX, return_tensors="pt")

output_ids = model.generate(
    input_ids.unsqueeze(0).cuda(),
    images=image_tensor.half().cuda(),
    max_new_tokens=512
)

Native Multimodal (Unified)

GPT-4o / Omni

Single transformer trained on:
- Text tokens
- Image patches (ViT) → linear projection
- Audio spectrograms → encoder
- All interleaved in single sequence

Architecture:
Input → Unified Embedding → Transformer Layers → Unified Output
         ↑                      ↑
    Modality-specific      Modality-specific
    tokenizers             heads

Capabilities: - Real-time audio conversation (320ms latency) - Vision + audio + text interleaved - Native tool use across modalities

Gemini 1.5

Native multimodal from scratch:
- 1M/2M context window
- Mixture of Experts (MoE)
- Unified tokenizer: text, image, audio, video
- Long-context retrieval (needle in haystack)

Chameleon (Meta)

Early-fusion token-based:
- Discrete image tokens (VQ-VAE)
- Text + image tokens in single vocabulary
- 7B/34B parameters
- Transformer on mixed sequence

Unified Tokenization

# Discrete tokens for all modalities
class UnifiedTokenizer:
    def __init__(self):
        self.text_tokenizer = TextTokenizer(vocab_size=256000)
        self.image_tokenizer = VQVAETokenizer(vocab_size=8192)  # 8192 visual tokens
        self.audio_tokenizer = DACAudioTokenizer(vocab_size=8192)

    def encode(self, multimodal_input) -> list[int]:
        tokens = []
        for item in multimodal_input:
            if item.type == "text":
                tokens.extend(self.text_tokenizer.encode(item.content))
            elif item.type == "image":
                tokens.append(IMAGE_START)
                tokens.extend(self.image_tokenizer.encode(item.content))
                tokens.append(IMAGE_END)
            elif item.type == "audio":
                tokens.append(AUDIO_START)
                tokens.extend(self.audio_tokenizer.encode(item.content))
                tokens.append(AUDIO_END)
        return tokens

Audio Models

Whisper (ASR)

from transformers import WhisperForConditionalGeneration, WhisperProcessor

model = WhisperForConditionalGeneration.from_pretrained("openai/whisper-large-v3")
processor = WhisperProcessor.from_pretrained("openai/whisper-large-v3")

# Transcribe
inputs = processor(audio, sampling_rate=16000, return_tensors="pt")
predicted_ids = model.generate(inputs.input_features)
transcription = processor.batch_decode(predicted_ids, skip_special_tokens=True)

AudioLLM / Qwen-Audio / SALMONN

# Audio encoder + LLM
# Supports: ASR, AAC, speech translation, audio QA

from transformers import Qwen2AudioForConditionalGeneration, Qwen2AudioProcessor

model = Qwen2AudioForConditionalGeneration.from_pretrained("Qwen/Qwen2-Audio-7B")
processor = Qwen2AudioProcessor.from_pretrained("Qwen/Qwen2-Audio-7B")

inputs = processor(text="Transcribe this audio", audios=audio_array, return_tensors="pt")
generated_ids = model.generate(**inputs, max_new_tokens=256)

Moshi / Ultravox (Real-time Speech)

# Streaming, low-latency speech-to-speech
# End-to-end: Audio → Text → LLM → Audio

# Kyutai Moshi: 7B, streaming, 200ms latency
# Ultravox: Open-source, Llama 3 + Whisper

Video Understanding

Frame-Based (Sample + VLM)

class VideoVLM:
    def __init__(self, vlm_model):
        self.vlm = vlm_model

    def analyze(self, video_path, query, strategy="uniform", max_frames=16):
        frames = self.sample_frames(video_path, strategy, max_frames)

        # Prepare multimodal input
        content = [{"type": "text", "text": query}]
        for frame in frames:
            content.append({
                "type": "image_url",
                "image_url": {"url": self.encode_frame(frame)}
            })

        return self.vlm.generate(content)

Native Video Models

Model Approach Context
Video-LLaMA Frame sampling + LLM 16 frames
Valley Video encoder + LLM 64 frames
VideoChat Video encoder + Vicuna 100 frames
Gemini 1.5 Native video tokens 1M tokens (~1hr video)

Multimodal Training

Stage 1: Contrastive Pretraining

# Image-text pairs: contrastive loss
def clip_loss(image_emb, text_emb, temperature=0.07):
    logits = image_emb @ text_emb.T / temperature
    labels = torch.arange(len(image_emb))
    loss_i = F.cross_entropy(logits, labels)
    loss_t = F.cross_entropy(logits.T, labels)
    return (loss_i + loss_t) / 2

Stage 2: Adapter/Projector Training

# Freeze vision encoder + LLM, train projector
for param in vision_encoder.parameters():
    param.requires_grad = False
for param in llm.parameters():
    param.requires_grad = False
# Only projector learns

Stage 3: Instruction Tuning

# Multimodal instruction data
# Format: <image>Question? Answer.
# Loss: Next-token prediction on answer

Stage 4: RLHF / DPO

# Preference pairs for multimodal
# (image, prompt, chosen, rejected) → DPO loss

Multimodal for Agents

Vision Tools

class VisionTools:
    @function_tool
    async def analyze_screenshot(self, image: Image, query: str) -> str:
        """Analyze a screenshot/UI"""
        return await self.vlm.analyze(image, query)

    @function_tool
    async def read_document(self, pdf_path: str, query: str) -> str:
        """Extract info from PDF/document"""
        images = pdf_to_images(pdf_path)
        return await self.vlm.analyze_multi(images, query)

    @function_tool
    async def inspect_code_ui(self, image: Image) -> CodeAnalysis:
        """Analyze code from screenshot"""
        return await self.vlm.analyze(image, "Extract code, find bugs, suggest improvements")

Audio Tools

class AudioTools:
    @function_tool
    async def transcribe(self, audio: bytes) -> Transcript:
        return await self.asr.transcribe(audio)

    @function_tool
    async def speak(self, text: str) -> bytes:
        return await self.tts.synthesize(text)

    @function_tool
    async def analyze_audio(self, audio: bytes, query: str) -> str:
        return await self.audio_llm.analyze(audio, query)


Sources: CLIP (Radford et al.), Flamingo (Alayrac et al.), BLIP-2 (Li et al.), LLaVA (Liu et al.), GPT-4o system card, Gemini 1.5 report, Chameleon (Meta), Whisper (Radford et al.), Qwen-Audio, Moshi (Kyutai), Video-LLaMA, Valley, VideoChat, ImageBind, AudioCLIP, multimodal training surveys.

From the Agentic Engineering research vault · CC BY 4.0 · all notes · raw markdown