Multimodal Models

Text: "The dog jumps"
      → Tokenize → ["The", "dog", "jumps"]

Image: [Pixel Grid 224×224]
       → Vision Transformer / CNN
       → Image Tokens: [tok_1, tok_2, ..., tok_256]

Audio: [Waveform 16kHz]
       → Audio Encoder (Mel-Spectrogram)
       → Audio Tokens: [tok_1, tok_2, ..., tok_512]

Video: [30 frames]
       → Temporal Encoder
       → Video Tokens: [tok_1, tok_2, ..., tok_4096]

Then: All tokens → Transformer → LLM generates answer

response = client.messages.create(
    model="claude-3-5-sonnet",
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": "image/jpeg",
                    "data": base64_encoded_image,
                },
            },
            {
                "type": "text",
                "text": "What do you see? Describe details."
            }
        ],
    }]
)

# Vision
response = client.messages.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": [
            {"type": "image_url", "image_url": {"url": "https://..."}},
            {"type": "text", "text": "Explain this diagram"}
        ]
    }]
)

# Audio via Whisper first
transcript = client.audio.transcriptions.create(
    model="whisper-1",
    file=audio_file
)

from openai import OpenAI

client = OpenAI()
transcript = client.audio.transcriptions.create(
    model="whisper-1",
    file=open("audio.mp3", "rb"),
    language="en"
)

# Accuracy: 94-97%
# Cost: €0.02 per minute (expensive!)

Workflow:
1. Extract keyframes (every 10 frames)
2. Patch + embed each keyframe
3. Audio: Whisper transcription separately
4. Combine: [Video tokens] + [Audio] + [Transcript] → LLM

Result: Scene understanding, object detection, summaries

Input: 100-page scanned PDF
Process:
- PDF → Images (1 per page)
- Claude Vision: Analyze each page
- Extract: Text, tables, charts
- OCR handwriting

Cost: ~€0.03 per page
Accuracy: 95%+ for printed text

Input: Product photo
Output:
- Material: Cotton
- Color: Blue
- Size: Large
- Defects: None detected

Use: Automate 100+ catalogs per day

Input: 1h recorded meeting
Process:
1. Audio extract + Whisper
2. Key frames (every 30s)
3. Vision + Transcript combined
4. Summarize

Output: 5-min summary + action items
Cost: ~€0.50 per hour

Issue: Hallucinations on visual details
Fix: "Describe ONLY what you see, not interpretation"

Issue: Small text in images hard to read
Fix: Enlarge text area first, or OCR separately

Issue: Video is token-expensive (many frames)
Fix: Keyframe sampling, or key-scene detection

Issue: Audio not native in tokens
Fix: Whisper first → text → LLM

Image Input (224×224)
        ↓
Patch Embedding (16×16 patches)
        ↓
Position Encoding
        ↓
Transformer Encoder (12 layers)
        ↓
Visual Features: [cls_token, patch_1_feat, ..., patch_196_feat]
        ↓
[MERGED with Text Tokens]
        ↓
LLM Decoder
        ↓
Text Output

Scenario: Process 10,000 product images (1MB each)

Provider        | Cost per 1000 | Total Cost | Time |
----------------|---------------|-----------|------|
Claude 3.5      | €0.30         | €3.00     | 2h   |
GPT-4V          | €0.60         | €6.00     | 1.5h |
Gemini 2.0      | €0.075        | €0.75     | 3h   |
Local LLaVA     | Free (GPU)    | €0.50*    | 4h   |

*GPU cost amortized over 10K images

import base64
from anthropic import Anthropic

client = Anthropic()

def extract_text_from_image(image_path, include_handwriting=False):
    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")

    prompt = """Extract all text from this image.
    Format as markdown with structure preserved.
    Include tables, lists, headings."""

    if include_handwriting:
        prompt += "\nAlso transcribe any handwritten text."

    message = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=4096,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/jpeg",
                            "data": image_data,
                        },
                    },
                    {
                        "type": "text",
                        "text": prompt
                    }
                ],
            }
        ],
    )

    return message.content[0].text

# Usage
extracted = extract_text_from_image("invoice.jpg")
print(extracted)

import json
from pathlib import Path
from anthropic import Anthropic

def batch_process_documents(image_dir, output_file, batch_size=10):
    """Process multiple images, save results, resume on error."""
    client = Anthropic()
    results = []
    processed = set()

    # Load previous results to resume
    if Path(output_file).exists():
        with open(output_file) as f:
            existing = json.load(f)
            processed = {r['image'] for r in existing}
            results = existing

    images = sorted(Path(image_dir).glob("*.jpg"))
    images = [img for img in images if img.name not in processed]

    for i, image_path in enumerate(images):
        if i % batch_size == 0:
            print(f"Progress: {i}/{len(images)}")

        try:
            # Process image
            text = extract_text_from_image(str(image_path))
            results.append({
                "image": image_path.name,
                "text": text,
                "status": "success"
            })
        except Exception as e:
            results.append({
                "image": image_path.name,
                "error": str(e),
                "status": "error"
            })

        # Save after each image (checkpoint for resume)
        with open(output_file, 'w') as f:
            json.dump(results, f)

    return results

# Usage
batch_process_documents("./invoices/", "results.json")

import cv2
import json
from pathlib import Path

def extract_key_frames(video_path, fps_sampling=1, output_dir="frames"):
    """Extract frames at specified intervals."""
    cap = cv2.VideoCapture(video_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    video_fps = cap.get(cv2.CAP_PROP_FPS)

    frame_interval = int(video_fps / fps_sampling)  # 1 frame per second
    frame_num = 0
    extracted = []

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break

        if frame_num % frame_interval == 0:
            frame_path = f"{output_dir}/frame_{frame_num:06d}.jpg"
            cv2.imwrite(frame_path, frame)
            timestamp = frame_num / video_fps
            extracted.append({
                "frame": frame_path,
                "timestamp": f"{int(timestamp//60)}:{int(timestamp%60):02d}",
                "frame_num": frame_num
            })

        frame_num += 1

    cap.release()
    return extracted

def summarize_video(video_path, fps_sampling=1):
    """Extract frames and generate summary."""
    frames = extract_key_frames(video_path, fps_sampling=fps_sampling)

    summaries = []
    client = Anthropic()

    for frame_info in frames:
        # Process frame with Claude
        with open(frame_info["frame"], "rb") as f:
            image_data = base64.standard_b64encode(f.read()).decode("utf-8")

        message = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=200,
            messages=[{
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/jpeg",
                            "data": image_data,
                        },
                    },
                    {
                        "type": "text",
                        "text": "Describe what's happening in this video frame. Be concise (1-2 sentences)."
                    }
                ],
            }]
        )

        summaries.append({
            "timestamp": frame_info["timestamp"],
            "description": message.content[0].text
        })

    return summaries

# Usage
summaries = summarize_video("meeting.mp4", fps_sampling=0.5)  # 1 frame every 2 seconds
for s in summaries:
    print(f"{s['timestamp']}: {s['description']}")

from chromadb import Client
from pathlib import Path
import base64

def image_rag_pipeline(query_text, image_dir, query_image=None):
    """Retrieve similar images and documents for query."""

    client = Client()
    collection = client.create_collection(name="images")

    # 1. Add images to collection with descriptions
    for image_path in Path(image_dir).glob("*.jpg"):
        with open(image_path, "rb") as f:
            image_data = base64.b64encode(f.read()).decode("utf-8")

        # Generate image description
        description = get_image_description(image_data)

        collection.add(
            ids=[image_path.stem],
            documents=[description],
            metadatas=[{
                "image_path": str(image_path),
                "image_base64": image_data
            }]
        )

    # 2. Search by text query
    text_results = collection.query(
        query_texts=[query_text],
        n_results=3
    )

    # 3. Search by image query (if provided)
    if query_image:
        query_description = get_image_description(query_image)
        image_results = collection.query(
            query_texts=[query_description],
            n_results=3
        )
    else:
        image_results = {"ids": [], "documents": []}

    return text_results, image_results

def get_image_description(image_base64):
    """Get text description of image for embedding."""
    client = Anthropic()
    message = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=200,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/jpeg",
                        "data": image_base64,
                    },
                },
                {
                    "type": "text",
                    "text": "Describe this image in 2-3 sentences for a search database."
                }
            ],
        }]
    )
    return message.content[0].text

# Solution: Upscale before sending
from PIL import Image
import cv2

def upscale_image(image_path, scale_factor=2):
    img = cv2.imread(image_path)
    height, width = img.shape[:2]
    new_size = (width * scale_factor, height * scale_factor)
    upscaled = cv2.resize(img, new_size, interpolation=cv2.INTER_CUBIC)
    return upscaled

# Solution: Use explicit instruction
prompt = """Look at this image carefully.
Describe ONLY what you can actually see.
Do NOT infer or guess about details not visible."""

# Solution: Use consistent formatting/preprocessing
def preprocess_image(image_path, target_size=1024):
    img = Image.open(image_path)
    # Resize maintaining aspect ratio
    img.thumbnail((target_size, target_size), Image.Resampling.LANCZOS)
    # Convert to RGB if RGBA
    if img.mode == 'RGBA':
        img = img.convert('RGB')
    return img

# Solution: Adaptive frame sampling based on motion
def smart_frame_extraction(video_path, motion_threshold=10):
    """Extract frames only when scene changes significantly."""
    cap = cv2.VideoCapture(video_path)
    prev_frame = None
    frames = []
    frame_num = 0

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break

        if prev_frame is None:
            frames.append(frame_num)
        else:
            diff = cv2.absdiff(cv2.cvtColor(prev_frame, cv2.COLOR_BGR2GRAY),
                             cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY))
            if diff.mean() > motion_threshold:
                frames.append(frame_num)

        prev_frame = frame
        frame_num += 1

    cap.release()
    return frames  # Only significant frames