8. Project: Multimodal Classifier

Description

This project closes Module 1 by integrating everything you learned: modalities (vision, audio), formats (Base64, URLs), APIs, costs, rate limits and troubleshooting. You're going to build a Multimodal Classifier — a system that takes any input (text, an image path, an audio path, a URL), detects the modality, validates the format, recommends the optimal model and provider, estimates the cost, and prepares the input in the right format for the API.

This isn't a toy example. It's the pattern real content-processing systems use: an intelligent router that decides which pipeline to apply before making the API call.

Why it matters: In production, inputs arrive in any format. An endpoint receives a JPEG image, an MP3 audio, or plain text, and it has to decide what to do with each one. Without a classifier up front, you end up with fragile if/elif chains that break on the first unexpected format.

Connection with the module: Every component of the classifier comes from an earlier capsule:

• Modality detection → Capsules 02 (vision), 03 (audio), 04 (combinations)
• Model selection → Capsule 05 (model landscape)
• Format and encoding validation → Capsule 06 (formats and APIs)
• Error handling → Capsule 07 (troubleshooting)

Connection with the guide: In later modules, this classifier evolves. In Module 2, the vision pipeline uses similar logic to choose between GPT-4 Vision, Claude 3 and Gemini. In Module 8 (Document Analyzer), the system detects the document type and routes to the right pipeline.

Technical Specifications

Input

The classifier accepts a single argument input_value: str that can be:

Output

@dataclass
class ClassificationResult:
    modality: str              # "text" | "image" | "audio"
    source_type: str           # "raw" | "local_file" | "url"
    recommended_model: str     # "gpt-4o" | "gpt-4o-mini" | "whisper-1" | etc.
    recommended_provider: str  # "openai" | "anthropic" | "google"
    estimated_cost_usd: float  # Estimated cost of processing this input
    format_valid: bool         # Whether the format passed validation
    metadata: dict             # Size, format, dimensions, duration, etc.
    warnings: list[str]        # Warnings (large file, suboptimal format)
    api_payload: dict | None   # Payload ready to send to the API

Functional requirements

1. Detect the modality by extension, URL pattern, or content
2. Validate the format before classifying (the real format, not just the extension)
3. Recommend a model and provider based on modality, size and the cost/quality balance
4. Estimate the cost based on image tokens or audio minutes
5. Prepare the payload in the provider's correct format (Base64 for Anthropic, URL or Base64 for OpenAI)
6. Handle errors with clear messages (file doesn't exist, invalid format, exceeds the size limit)

Step 1: Constants and Configuration

from dataclasses import dataclass, field
from pathlib import Path
import base64, re, math

IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".mp4", ".mpeg", ".webm", ".ogg", ".flac"}
MIME_TYPES = {
    ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png",
    ".gif": "image/gif", ".webp": "image/webp",
    ".mp3": "audio/mpeg", ".wav": "audio/wav", ".m4a": "audio/mp4",
    ".ogg": "audio/ogg", ".flac": "audio/flac",
}

SIZE_LIMITS_MB = {
    "openai": {"image": 20, "audio": 25},
    "anthropic": {"image": 5, "audio": 0},
    "google": {"image": 20, "audio": 0},
}

MODEL_PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "whisper-1": {"per_minute": 0.006},
    "claude-3.5-sonnet": {"input": 3.00, "output": 15.00},
    "gemini-1.5-flash": {"input": 0.075, "output": 0.30},
}


@dataclass
class ClassificationResult:
    modality: str = "unknown"
    source_type: str = "unknown"
    recommended_model: str = ""
    recommended_provider: str = ""
    estimated_cost_usd: float = 0.0
    format_valid: bool = False
    metadata: dict = field(default_factory=dict)
    warnings: list[str] = field(default_factory=list)
    api_payload: dict | None = None
    error: str | None = None

Separating constants from logic lets you change prices or supported formats without touching the classification logic.

Step 2: Detect the Modality

Detection follows a priority order: first it checks whether it's a URL, then whether it's a local file, and as a fallback it assumes text.

def detect_modality(input_value: str) -> tuple[str, str]:
    """Detect the modality and the source type.

    Returns:
        (modality, source_type) where modality is "text"|"image"|"audio"
        and source_type is "raw"|"local_file"|"url"
    """
    if re.match(r"https?://", input_value):
        ext = _extract_extension_from_url(input_value)
        if ext in IMAGE_EXTENSIONS:
            return "image", "url"
        if ext in AUDIO_EXTENSIONS:
            return "audio", "url"
        return "text", "url"

    path = Path(input_value)
    if path.exists() and path.is_file():
        ext = path.suffix.lower()
        if ext in IMAGE_EXTENSIONS:
            return "image", "local_file"
        if ext in AUDIO_EXTENSIONS:
            return "audio", "local_file"

    return "text", "raw"


def _extract_extension_from_url(url: str) -> str:
    """Extract the extension from a URL, ignoring query params."""
    clean = url.split("?")[0].split("#")[0]
    path = Path(clean)
    return path.suffix.lower()

Design decision: URLs with no recognized extension are classified as text. This is intentional — if you can't determine the modality without downloading the content, it's safer not to assume. The caller can force the modality if they have extra information.

Step 3: Collect Metadata

The metadata varies depending on the modality and the source type:

def collect_metadata(input_value: str, modality: str, source_type: str) -> dict:
    """Collect relevant metadata according to the modality."""
    if modality == "text":
        return {
            "length_chars": len(input_value),
            "length_words": len(input_value.split()),
            "estimated_tokens": len(input_value) // 4,
        }

    if source_type == "url":
        return {"url": input_value, "format": _extract_extension_from_url(input_value)[1:]}

    path = Path(input_value)
    size_bytes = path.stat().st_size
    meta = {
        "file_path": str(path),
        "file_size_kb": round(size_bytes / 1024, 1),
        "file_size_mb": round(size_bytes / (1024 * 1024), 2),
        "format": path.suffix[1:].lower(),
    }

    if modality == "image":
        meta.update(_get_image_metadata(path))
    elif modality == "audio":
        meta.update(_get_audio_metadata(path, size_bytes))

    return meta


def _get_image_metadata(path: Path) -> dict:
    """Get the image dimensions if PIL is available."""
    try:
        from PIL import Image
        with Image.open(path) as img:
            return {"width": img.width, "height": img.height, "mode": img.mode}
    except ImportError:
        return {"width": None, "height": None, "note": "PIL not installed"}
    except Exception:
        return {"width": None, "height": None, "note": "Could not read the dimensions"}


def _get_audio_metadata(path: Path, size_bytes: int) -> dict:
    """Estimate the audio duration from the size (heuristic, no pydub)."""
    format_bitrates_kbps = {"mp3": 128, "wav": 1411, "m4a": 128, "ogg": 112, "flac": 800}
    ext = path.suffix[1:].lower()
    bitrate = format_bitrates_kbps.get(ext, 128)
    estimated_seconds = (size_bytes * 8) / (bitrate * 1000)
    return {
        "estimated_duration_seconds": round(estimated_seconds, 1),
        "estimated_duration_minutes": round(estimated_seconds / 60, 2),
    }

For images we try to get the dimensions with PIL (optional). For audio we estimate the duration from the size — a heuristic that works reasonably well without depending on pydub or ffprobe.

Step 4: Validate the Format

Validation checks that the file really is what the extension says and that it doesn't exceed the provider's limits:

def validate_format(
    input_value: str, modality: str, source_type: str, metadata: dict, provider: str
) -> tuple[bool, list[str]]:
    """Validate the format and return (is_valid, warnings)."""
    warnings = []

    if source_type == "url":
        warnings.append("URL not validated locally — errors will surface when calling the API")
        return True, warnings

    if source_type == "raw":
        return True, warnings

    path = Path(input_value)
    if not path.exists():
        return False, [f"File not found: {input_value}"]

    size_mb = metadata.get("file_size_mb", 0)
    limit = SIZE_LIMITS_MB.get(provider, {}).get(modality, 20)

    if limit == 0 and modality == "audio":
        return False, [f"{provider} doesn't support audio directly"]

    if size_mb > limit:
        return False, [f"File exceeds {provider}'s limit: {size_mb:.1f} MB > {limit} MB"]

    if size_mb > limit * 0.8:
        warnings.append(f"File close to the limit ({size_mb:.1f}/{limit} MB) — consider resizing")

    if modality == "image":
        _validate_image_format(path, warnings)

    return True, warnings


def _validate_image_format(path: Path, warnings: list[str]):
    """Check that the image content matches the extension."""
    try:
        from PIL import Image
        with Image.open(path) as img:
            img.verify()
    except ImportError:
        warnings.append("PIL not installed — image integrity was not verified")
    except Exception as e:
        warnings.append(f"Image possibly corrupt: {e}")

Key point: Validation is conservative with URLs (it doesn't download them just to validate) and strict with local files (it verifies integrity with PIL when available).

Step 5: Recommend a Model and Provider

The recommendation considers modality, size, and the cost/quality balance:

def recommend_model(
    modality: str, metadata: dict, priority: str = "balanced"
) -> tuple[str, str, str]:
    """Recommend a model and provider.

    Args:
        priority: "quality" | "cost" | "balanced"

    Returns:
        (model, provider, reason)
    """
    if modality == "text":
        if priority == "quality":
            return "gpt-4o", "openai", "Complex text: high-capability model"
        return "gpt-4o-mini", "openai", "Text: economical model with good quality"

    if modality == "audio":
        return "whisper-1", "openai", "Audio: Whisper is the standard for STT"

    if modality == "image":
        return _recommend_vision_model(metadata, priority)

    return "gpt-4o-mini", "openai", "Fallback: unrecognized modality"


def _recommend_vision_model(metadata: dict, priority: str) -> tuple[str, str, str]:
    """Select a vision model based on size and priority."""
    size_mb = metadata.get("file_size_mb", 0)

    if priority == "cost":
        if size_mb > 5:
            return "gemini-1.5-flash", "google", "Large image + cost priority: Gemini Flash"
        return "gpt-4o-mini", "openai", "Image: gpt-4o-mini is 17x cheaper than gpt-4o"

    if priority == "quality":
        return "gpt-4o", "openai", "Image + quality priority: GPT-4 Vision"

    if size_mb > 10:
        return "gpt-4o-mini", "openai", "Large image (>10MB): economical model to reduce cost"

    return "gpt-4o-mini", "openai", "Standard image: good quality at low cost"

Design decision: The priority parameter lets the caller tune the recommendation. It defaults to "balanced" (gpt-4o-mini for most cases), but the caller can ask for "quality" for critical tasks or "cost" for large batches.

Step 6: Estimate the Cost

def estimate_cost(modality: str, model: str, metadata: dict) -> float:
    """Estimate the cost of processing the input with the recommended model."""
    pricing = MODEL_PRICING.get(model)
    if not pricing:
        return 0.0

    if modality == "audio":
        minutes = metadata.get("estimated_duration_minutes", 1.0)
        return round(minutes * pricing["per_minute"], 6)

    if modality == "image":
        image_tokens = _estimate_image_tokens(metadata)
        prompt_tokens = 50
        output_tokens = 150
        input_cost = ((image_tokens + prompt_tokens) / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        return round(input_cost + output_cost, 6)

    if modality == "text":
        input_tokens = metadata.get("estimated_tokens", 100)
        output_tokens = min(input_tokens, 500)
        input_cost = (input_tokens / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        return round(input_cost + output_cost, 6)

    return 0.0


def _estimate_image_tokens(metadata: dict) -> int:
    """Estimate image tokens using OpenAI's tile system."""
    width = metadata.get("width")
    height = metadata.get("height")

    if not width or not height:
        return 765  # default: 1024x1024 → 4 tiles (capsule 06)

    if max(width, height) <= 512:
        return 85  # low detail

    scale = min(2048 / max(width, height), 1.0)
    w, h = int(width * scale), int(height * scale)

    scale_short = 768 / min(w, h)
    if scale_short < 1:
        w, h = int(w * scale_short), int(h * scale_short)

    tiles_w = math.ceil(w / 512)
    tiles_h = math.ceil(h / 512)
    return 85 + (170 * tiles_w * tiles_h)

This token calculation replicates exactly the logic from capsule 06: scale to 2048px, then to 768px on the short side, and count 512×512 tiles.

Step 7: Prepare the API Payload

def prepare_api_payload(
    input_value: str, modality: str, source_type: str, provider: str
) -> dict | None:
    """Prepare the payload in the provider's format."""
    if modality == "text":
        return {"type": "text", "text": input_value}

    if modality == "audio":
        if source_type == "local_file":
            return {"type": "audio", "file_path": input_value, "model": "whisper-1"}
        return None

    if modality == "image":
        return _prepare_image_payload(input_value, source_type, provider)

    return None


def _prepare_image_payload(input_value: str, source_type: str, provider: str) -> dict:
    """Prepare the image payload according to the provider."""
    if source_type == "url":
        if provider == "anthropic":
            return {
                "type": "image",
                "note": "Anthropic requires Base64 — download the URL first",
                "url": input_value,
            }
        return {"type": "image_url", "image_url": {"url": input_value}}

    path = Path(input_value)
    with open(path, "rb") as f:
        encoded = base64.b64encode(f.read()).decode("utf-8")
    mime = MIME_TYPES.get(path.suffix.lower(), "image/jpeg")

    if provider == "anthropic":
        return {
            "type": "image",
            "source": {"type": "base64", "media_type": mime, "data": encoded},
        }

    data_uri = f"data:{mime};base64,{encoded}"
    return {"type": "image_url", "image_url": {"url": data_uri}}

The payload respects the differences between providers: Anthropic needs media_type + data separately; OpenAI accepts data URIs.

Step 8: Main Function — Integration

def classify_multimodal(
    input_value: str,
    priority: str = "balanced",
    provider: str | None = None,
    prepare_payload: bool = True,
) -> ClassificationResult:
    """Classify a multimodal input and produce a complete result.

    Args:
        input_value: Text, a file path, or a URL
        priority: "quality" | "cost" | "balanced"
        provider: Force a specific provider. If None, it's auto-selected.
        prepare_payload: If True, generate the payload ready for the API.
    """
    result = ClassificationResult()

    modality, source_type = detect_modality(input_value)
    result.modality = modality
    result.source_type = source_type

    result.metadata = collect_metadata(input_value, modality, source_type)

    model, rec_provider, reason = recommend_model(modality, result.metadata, priority)
    result.recommended_model = model
    result.recommended_provider = provider or rec_provider

    is_valid, warnings = validate_format(
        input_value, modality, source_type, result.metadata, result.recommended_provider
    )
    result.format_valid = is_valid
    result.warnings = warnings

    if not is_valid:
        result.error = warnings[0] if warnings else "Invalid format"
        return result

    result.estimated_cost_usd = estimate_cost(modality, model, result.metadata)

    if prepare_payload and source_type != "url":
        try:
            result.api_payload = prepare_api_payload(
                input_value, modality, source_type, result.recommended_provider
            )
        except Exception as e:
            result.warnings.append(f"Could not prepare the payload: {e}")

    return result

Complete Code

The complete code integrates every previous step into a single runnable file. What follows is the usage interface and a demo:

def print_classification(result: ClassificationResult):
    """Print the classification result in a readable way."""
    status = "VALID" if result.format_valid else "INVALID"
    print(f"\n{'='*60}")
    print(f"  Modality:    {result.modality}")
    print(f"  Source:      {result.source_type}")
    print(f"  Format:      {status}")
    print(f"  Model:       {result.recommended_model} ({result.recommended_provider})")
    print(f"  Est. cost:   ${result.estimated_cost_usd:.6f}")

    if result.metadata:
        print(f"  Metadata:")
        for k, v in result.metadata.items():
            print(f"    {k}: {v}")

    if result.warnings:
        print(f"  Warnings:")
        for w in result.warnings:
            print(f"    ⚠ {w}")

    if result.error:
        print(f"  ERROR: {result.error}")

    if result.api_payload:
        payload_type = result.api_payload.get("type", "?")
        print(f"  Payload:     {payload_type} (ready for the API)")

    print(f"{'='*60}")


# Demo
inputs = [
    "Analyze the Q4 sales trends",
    "./photos/laptop_product.jpg",
    "./recordings/customer_call.mp3",
    "https://upload.wikimedia.org/wikipedia/commons/3/3a/Cat03.jpg",
    "./nonexistent_file.png",
]

for inp in inputs:
    result = classify_multimodal(inp, priority="balanced")
    print_classification(result)

Expected output (schematic):

============================================================
  Modality:    text
  Source:      raw
  Format:      VALID
  Model:       gpt-4o-mini (openai)
  Est. cost:   $0.000005
  Metadata:
    length_chars: 27
    length_words: 5
    estimated_tokens: 6
  Payload:     text (ready for the API)
============================================================

============================================================
  Modality:    image
  Source:      local_file
  Format:      VALID
  Model:       gpt-4o-mini (openai)
  Est. cost:   $0.000212
  Metadata:
    file_path: ./photos/laptop_product.jpg
    file_size_kb: 245.3
    file_size_mb: 0.24
    format: jpg
    width: 1024
    height: 768
  Payload:     image_url (ready for the API)
============================================================

Extension 1: Batch Classification

To process multiple inputs, add accumulated cost tracking:

@dataclass
class BatchReport:
    total: int = 0
    by_modality: dict = field(default_factory=lambda: {"text": 0, "image": 0, "audio": 0})
    valid: int = 0
    invalid: int = 0
    total_cost_usd: float = 0.0
    results: list[ClassificationResult] = field(default_factory=list)


def classify_batch(
    inputs: list[str], priority: str = "balanced", budget: float | None = None
) -> BatchReport:
    """Classify multiple inputs with cost tracking."""
    report = BatchReport()

    for inp in inputs:
        result = classify_multimodal(inp, priority=priority)
        report.total += 1
        report.by_modality[result.modality] = report.by_modality.get(result.modality, 0) + 1

        if result.format_valid:
            report.valid += 1
        else:
            report.invalid += 1

        if budget and (report.total_cost_usd + result.estimated_cost_usd) > budget:
            result.warnings.append(f"Budget exceeded: ${report.total_cost_usd:.4f}/{budget}")
            result.api_payload = None

        report.total_cost_usd += result.estimated_cost_usd
        report.results.append(result)

    return report


def print_batch_report(report: BatchReport):
    print(f"\n{'='*60}")
    print(f"  BATCH REPORT")
    print(f"{'='*60}")
    print(f"  Total inputs:  {report.total}")
    print(f"  Valid:         {report.valid}")
    print(f"  Invalid:       {report.invalid}")
    print(f"  By modality:   {report.by_modality}")
    print(f"  Total cost:    ${report.total_cost_usd:.6f}")
    print(f"{'='*60}")

Extension 2: Multi-Provider Comparison Mode

To decide between providers, classify the same input with each one and compare:

def compare_providers(
    input_value: str, providers: list[str] = None
) -> list[dict]:
    """Compare the classification across providers."""
    providers = providers or ["openai", "anthropic", "google"]
    comparisons = []

    for provider in providers:
        result = classify_multimodal(input_value, provider=provider)
        comparisons.append({
            "provider": provider,
            "model": result.recommended_model,
            "cost": result.estimated_cost_usd,
            "valid": result.format_valid,
            "warnings": len(result.warnings),
        })

    comparisons.sort(key=lambda x: x["cost"])
    return comparisons

Usage:

for c in compare_providers("./photos/product.jpg"):
    status = "✓" if c["valid"] else "✗"
    print(f"  {status} {c['provider']:<12} {c['model']:<20} ${c['cost']:.6f}")

Project Troubleshooting

Problem 1: Image file with the wrong extension

Symptom: A .txt image that is really a JPEG gets classified as text.

Solution: Add magic-byte detection as a second level:

MAGIC_BYTES = {
    b"\xff\xd8\xff": "image",      # JPEG
    b"\x89PNG": "image",            # PNG
    b"GIF8": "image",              # GIF
    b"RIFF": "audio",             # WAV
    b"ID3": "audio",              # MP3
    b"\xff\xfb": "audio",         # MP3 (without ID3)
}

def detect_by_magic_bytes(path: Path) -> str | None:
    """Detect the modality from the file's first bytes."""
    try:
        with open(path, "rb") as f:
            header = f.read(8)
        for magic, modality in MAGIC_BYTES.items():
            if header.startswith(magic):
                return modality
    except Exception:
        pass
    return None

Integrate this into detect_modality as a fallback when the extension isn't recognized but the file exists.

Problem 2: A URL with query params gets classified wrong

Symptom: https://images.com/photo.jpg?w=800&h=600 works, but https://api.example.com/image?id=123 has no extension and gets classified as text.

Solution: The current design is correct — with no recognizable extension, we don't assume a modality. The caller can use the provider parameter or pass the URL downloaded locally.

Problem 3: Incorrect cost estimate for images without PIL

Symptom: Without PIL installed, every image is estimated at 765 tokens (the default).

Solution: Install pillow (pip install Pillow). Without PIL, the estimate is conservative but imprecise. Add a warning:

if not width or not height:
    warnings.append("No PIL: cost estimated with 765 tokens (default). Install Pillow for accuracy.")

Problem 4: Long audio estimates the duration badly

Symptom: A 2-hour podcast in variable-bitrate MP3 gets an incorrect duration estimate.

Solution: The size heuristic assumes a constant bitrate. For real accuracy, use pydub:

from pydub import AudioSegment
audio = AudioSegment.from_file(path)
duration_minutes = len(audio) / (1000 * 60)

But pydub requires ffmpeg. The heuristic is enough for cost estimates (error ≤20%).

Problem 5: The Base64 payload is too large for Anthropic

Symptom: A 4 MB image generates a ~5.3 MB Base64 payload, exceeding Anthropic's limit (5 MB).

Solution: The validate_format function already catches this by checking against SIZE_LIMITS_MB. If you need to send the image to Anthropic, resize first:

from PIL import Image
import io

def resize_for_anthropic(image_path: str, max_mb: float = 3.5) -> bytes:
    """Resize so that the Base64 fits in 5 MB."""
    with Image.open(image_path) as img:
        img = img.convert("RGB")
        quality = 90
        while True:
            buffer = io.BytesIO()
            img.save(buffer, "JPEG", quality=quality, optimize=True)
            encoded_size = len(buffer.getvalue()) * 4 / 3
            if encoded_size <= max_mb * 1024 * 1024:
                return buffer.getvalue()
            quality -= 10
            if quality < 30:
                img = img.resize((img.width // 2, img.height // 2))
                quality = 85

Completeness Checklist

Before considering the project done, check every point:

Core functionality:

• Detects text, local image, local audio, image URL, audio URL
• Returns a ClassificationResult with all the fields populated
• Metadata includes size, format, and dimensions/duration when possible
• The model recommendation is coherent with the modality and priority
• The cost estimate uses the tile system for images and per-minute for audio

Validation:

• Nonexistent file → format_valid=False with a clear message
• File exceeding the limit → format_valid=False with the provider's limit
• File close to the limit → warning
• Corrupt image → warning (if PIL is available)

Payload:

• Local image → Base64 with a data URI (OpenAI) or media_type+data (Anthropic)
• Image by URL → direct URL (OpenAI/Google), download note (Anthropic)
• Audio → file_path for Whisper
• Text → dict with type "text"

Extensions (optional):

• Batch classification with cost tracking
• Multi-provider comparison
• Magic-byte detection

Exercises

Exercise 1: Classifier tests (Easy)

Create a test_classifier() function that tests the 5 input types (text, local image, local audio, image URL, audio URL) with asserts. You don't need real files — for local files, create temporary ones with tempfile.

import tempfile, os

def test_classifier():
    r = classify_multimodal("Hello world")
    assert r.modality == "text"
    assert r.source_type == "raw"
    assert r.format_valid is True

    with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
        f.write(b"\xff\xd8\xff\xe0" + b"\x00" * 100)
        img_path = f.name
    r = classify_multimodal(img_path)
    assert r.modality == "image"
    assert r.source_type == "local_file"
    os.unlink(img_path)

    with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
        f.write(b"ID3" + b"\x00" * 100)
        audio_path = f.name
    r = classify_multimodal(audio_path)
    assert r.modality == "audio"
    assert r.source_type == "local_file"
    os.unlink(audio_path)

    r = classify_multimodal("https://example.com/photo.png")
    assert r.modality == "image"
    assert r.source_type == "url"

    r = classify_multimodal("https://example.com/audio.mp3")
    assert r.modality == "audio"
    assert r.source_type == "url"

    r = classify_multimodal("./does_not_exist.jpg")
    assert r.modality == "text"  # the file doesn't exist → fallback to text

    print("All the tests passed")

test_classifier()

Exercise 2: Add video support (Medium)

Extend the classifier to support video files (.mp4, .mov, .avi, .mkv). The recommended model for video must be gemini-1.5-flash (Google is the only one that supports video directly). Add a cost estimate based on the video's estimated duration.

VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm"}
VIDEO_BITRATES_KBPS = {"mp4": 2500, "mov": 3000, "avi": 2000, "mkv": 2500, "webm": 1500}

def detect_modality_v2(input_value: str) -> tuple[str, str]:
    if re.match(r"https?://", input_value):
        ext = _extract_extension_from_url(input_value)
        if ext in VIDEO_EXTENSIONS:
            return "video", "url"
        # ... (the rest is the same)

    path = Path(input_value)
    if path.exists() and path.is_file():
        ext = path.suffix.lower()
        if ext in VIDEO_EXTENSIONS:
            return "video", "local_file"
        # ... (the rest is the same)

    return "text", "raw"

def recommend_model_v2(modality, metadata, priority="balanced"):
    if modality == "video":
        return "gemini-1.5-flash", "google", "Video: Gemini supports native video"
    return recommend_model(modality, metadata, priority)

MODEL_PRICING["gemini-1.5-flash-video"] = {"per_minute": 0.015}

Exercise 3: Batch dashboard with accumulated costs (Medium)

Using classify_batch, process a list of 20 mixed inputs and generate a report showing: distribution by modality (a text chart), the top 3 most expensive inputs, the total cost, and the percentage of the budget used.

def batch_dashboard(inputs: list[str], budget: float = 1.0):
    report = classify_batch(inputs, budget=budget)

    print("\n📊 CLASSIFICATION DASHBOARD")
    print(f"{'='*50}")

    total = report.total
    for mod, count in report.by_modality.items():
        bar = "█" * int(count / total * 30) if total > 0 else ""
        pct = count / total * 100 if total > 0 else 0
        print(f"  {mod:<8} {bar:<30} {count:>3} ({pct:.0f}%)")

    print(f"\n  Valid:   {report.valid}/{total}")
    print(f"  Invalid: {report.invalid}/{total}")

    top_costly = sorted(report.results, key=lambda r: r.estimated_cost_usd, reverse=True)[:3]
    print(f"\n  Top 3 most expensive:")
    for i, r in enumerate(top_costly, 1):
        path = r.metadata.get("file_path", r.metadata.get("url", "text"))
        print(f"    {i}. {path}: ${r.estimated_cost_usd:.6f} ({r.recommended_model})")

    pct_budget = (report.total_cost_usd / budget * 100) if budget > 0 else 0
    print(f"\n  Total cost: ${report.total_cost_usd:.6f} ({pct_budget:.1f}% of the budget)")
    print(f"{'='*50}")

Summary

In this project you built a complete Multimodal Classifier that:

• Detects the modality by file extension, URL pattern, or content (text as a fallback)
• Validates the format by checking existence, size against the provider's limits, and image integrity
• Recommends a model and provider based on modality, size and priority (quality vs cost)
• Estimates the cost using OpenAI's tile system for images and per-minute cost for audio
• Prepares payloads in each provider's correct format (data URI for OpenAI, media_type+data for Anthropic)
• Handles errors with clear messages and warnings for ambiguous situations

This classifier is the foundation of the pipeline routers you'll build in later modules. In Module 2 (Vision), the model-selection logic expands to include GPT-4 Vision, Claude 3 and Gemini with their specific capabilities. In Module 8 (Document Analyzer), modality detection integrates with processing PDFs, invoices and contracts.

Next module: Module 2 — Vision with LLMs. You already know what multimodal is; now you're going to master image analysis with the main providers.

Additional Resources

1. OpenAI Vision Guide — Input formats and limits
2. OpenAI Pricing — Up-to-date prices for estimating costs
3. Anthropic Vision Docs — The Base64 format Claude requires
4. Google Gemini Vision — Native video support
5. Python Pathlib — Handling paths and extensions
6. Pillow Documentation — Image validation and manipulation
7. Python dataclasses — Data structures for results