5. Comparison: GPT-4 Vision vs Claude Vision vs Gemini Vision

Description

In capsules 02, 03 and 04 you learned to use GPT-4 Vision, Claude 3 Vision and Gemini Vision individually. Each provider has a different API, models with different capabilities, and its own cost structure. But in production you don't pick one provider and stay with it forever — you pick the right provider for each task.

A high-volume product image classification system needs a cheap, fast model. An OCR pipeline over legal contracts needs maximum accuracy. Video analysis needs a provider that supports video as native input. Same domain (vision), but the optimal decision changes in each case.

This capsule gives you the tools to make that decision: comparison tables with real data, benchmarks you can run with your own images, a programmatic decision tree, cost calculation per scenario, and fallback patterns for production.

What you're going to build: A multi-provider benchmark, an automatic provider selection function, a cost estimator, and a router with fallback.

Detailed Comparison Table

This table compares the most relevant vision models from each provider. Prices correspond to the official APIs at the time of writing — always check the pricing pages before estimating production costs.

Reading the table

• Cost: Gemini Flash is ~33x cheaper than GPT-4o on input. For tasks where quality is "good enough", the cost difference dominates the decision.
• Context window: Gemini offers 1M tokens. If you need to analyze 50+ page documents in a single call, Gemini is the only viable option without chunking.
• Video: Only Gemini supports video as native input. The other providers require extracting frames and sending them as individual images.
• URLs: OpenAI accepts direct image URLs. Anthropic and Google require Base64 or bytes. This affects your pipeline architecture.

Benchmark: Same Image, Three Providers

The best way to compare providers is to send the same image with the same prompt to all three and measure the results. This benchmark measures latency, tokens and real cost.

import time
import base64
from pathlib import Path
from dataclasses import dataclass
from openai import OpenAI
import anthropic
import google.generativeai as genai


@dataclass
class BenchmarkResult:
    provider: str
    model: str
    response_text: str
    latency_ms: float
    input_tokens: int
    output_tokens: int
    estimated_cost_usd: float


PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "claude-3-5-sonnet-latest": {"input": 3.00, "output": 15.00},
    "claude-3-haiku-20240307": {"input": 0.25, "output": 1.25},
    "gemini-1.5-flash": {"input": 0.075, "output": 0.30},
    "gemini-1.5-pro": {"input": 1.25, "output": 5.00},
}


def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    prices = PRICING[model]
    return (input_tokens * prices["input"] + output_tokens * prices["output"]) / 1_000_000


def _timed_call(fn):
    start = time.perf_counter()
    result = fn()
    return result, (time.perf_counter() - start) * 1000


def benchmark_openai(image_b64: str, prompt: str, model: str = "gpt-4o") -> BenchmarkResult:
    client = OpenAI()
    response, latency = _timed_call(lambda: client.chat.completions.create(
        model=model, max_tokens=500,
        messages=[{"role": "user", "content": [
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}}
        ]}]
    ))
    u = response.usage
    return BenchmarkResult("OpenAI", model, response.choices[0].message.content,
        round(latency, 1), u.prompt_tokens, u.completion_tokens,
        estimate_cost(model, u.prompt_tokens, u.completion_tokens))


def benchmark_anthropic(image_b64: str, prompt: str, model: str = "claude-3-5-sonnet-latest") -> BenchmarkResult:
    client = anthropic.Anthropic()
    response, latency = _timed_call(lambda: client.messages.create(
        model=model, max_tokens=500,
        messages=[{"role": "user", "content": [
            {"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": image_b64}},
            {"type": "text", "text": prompt}
        ]}]
    ))
    u = response.usage
    return BenchmarkResult("Anthropic", model, response.content[0].text,
        round(latency, 1), u.input_tokens, u.output_tokens,
        estimate_cost(model, u.input_tokens, u.output_tokens))


def benchmark_gemini(image_bytes: bytes, prompt: str, model: str = "gemini-1.5-flash") -> BenchmarkResult:
    gmodel = genai.GenerativeModel(model)
    response, latency = _timed_call(
        lambda: gmodel.generate_content([prompt, {"mime_type": "image/jpeg", "data": image_bytes}])
    )
    u = response.usage_metadata
    return BenchmarkResult("Google", model, response.text,
        round(latency, 1), u.prompt_token_count, u.candidates_token_count,
        estimate_cost(model, u.prompt_token_count, u.candidates_token_count))


def benchmark_vision(image_path: str, prompt: str) -> list[BenchmarkResult]:
    raw_bytes = Path(image_path).read_bytes()
    image_b64 = base64.b64encode(raw_bytes).decode()

    results = [
        benchmark_openai(image_b64, prompt, "gpt-4o"),
        benchmark_anthropic(image_b64, prompt, "claude-3-5-sonnet-latest"),
        benchmark_gemini(raw_bytes, prompt, "gemini-1.5-flash"),
    ]

    print(f"\n{'Provider':<12} {'Model':<28} {'Latency':>10} {'Tokens In':>10} {'Tokens Out':>11} {'Cost':>10}")
    print("-" * 85)
    for r in results:
        print(f"{r.provider:<12} {r.model:<28} {r.latency_ms:>8.0f}ms {r.input_tokens:>10,} {r.output_tokens:>11,} ${r.estimated_cost_usd:>8.6f}")
    return results


results = benchmark_vision("invoice.jpg", "Extract all the data from this invoice in JSON format.")

Example output:

Provider     Model                           Latency  Tokens In  Tokens Out       Cost
-------------------------------------------------------------------------------------
OpenAI       gpt-4o                           2340ms      1,285         312 $0.006332
Anthropic    claude-3-5-sonnet-latest         2780ms      1,412         298 $0.008706
Google       gemini-1.5-flash                  890ms        980         275 $0.000156

Gemini Flash was ~3x faster and ~40x cheaper. But response quality may vary — the benchmark gives you the data, you decide which trade-off your use case accepts.

Comparison by Task

Each type of task has an optimal provider. This table summarizes the recommendations based on benchmarks and practical experience.

When the recommendation changes

The recommendations above assume ideal conditions. In practice, other factors can change the decision:

• You already have an integration with a provider: If your system already uses OpenAI for text, adding vision with GPT-4o has a lower integration cost than adding Anthropic from scratch.
• Regulation and compliance: Some sectors require that data not leave certain regions. Review each provider's policies.
• Rate limits for your tier: If you're on the free tier with Google but tier-2 with OpenAI, rate limits may favor a provider that is theoretically more expensive.
• Quality for your specific domain: These rankings are general. Always run the benchmark with your real images.

Decision Tree

Text version

Does your task involve video?
├── YES → Gemini 1.5 Pro
└── NO → Does the document have more than 50 pages?
    ├── YES → Gemini 1.5 Pro (1M context)
    └── NO → Is cost priority #1?
        ├── YES → Gemini 1.5 Flash
        └── NO → Do you need maximum reasoning quality?
            ├── YES → Claude 3.5 Sonnet
            └── NO → Volume > 1000 images/day?
                ├── YES → GPT-4o-mini or Gemini Flash
                └── NO → GPT-4o (general balance)

Code version

def select_provider(
    task: str,
    budget: str = "medium",
    context_length: int = 0,
    needs_video: bool = False,
    volume_per_day: int = 0
) -> dict:
    """Selects the optimal provider based on the use case parameters.

    Args:
        task: Task type (ocr_simple, ocr_complex, classification, reasoning, description, json_extraction)
        budget: Budget constraint (low, medium, high)
        context_length: Estimated document tokens (0 if image only)
        needs_video: Whether the task requires video processing
        volume_per_day: Estimated number of requests per day

    Returns:
        dict with provider, model and reason
    """
    if needs_video:
        return {"provider": "Google", "model": "gemini-1.5-pro", "reason": "Only one with native video support"}

    if context_length > 200_000:
        return {"provider": "Google", "model": "gemini-1.5-pro", "reason": f"Context of {context_length:,} tokens requires a 1M window"}

    if budget == "low":
        if volume_per_day > 1000:
            return {"provider": "Google", "model": "gemini-1.5-flash", "reason": "Minimum cost for high volume"}
        return {"provider": "Google", "model": "gemini-1.5-flash", "reason": "Cheapest option"}

    task_map = {
        "ocr_simple": {"provider": "Google", "model": "gemini-1.5-flash", "reason": "Sufficient OCR, minimal cost"},
        "ocr_complex": {"provider": "Anthropic", "model": "claude-3-5-sonnet-latest", "reason": "Superior at handwriting and complex documents"},
        "classification": {"provider": "OpenAI", "model": "gpt-4o-mini", "reason": "Sufficient quality for classification, low cost"},
        "reasoning": {"provider": "Anthropic", "model": "claude-3-5-sonnet-latest", "reason": "Best detailed visual reasoning"},
        "description": {"provider": "OpenAI", "model": "gpt-4o-mini", "reason": "Good descriptions at low cost"},
        "json_extraction": {"provider": "OpenAI", "model": "gpt-4o", "reason": "Best adherence to JSON schemas"},
    }

    if task in task_map:
        return task_map[task]

    if volume_per_day > 1000:
        return {"provider": "OpenAI", "model": "gpt-4o-mini", "reason": "Cost/quality balance for volume"}

    return {"provider": "OpenAI", "model": "gpt-4o", "reason": "Balanced default for general tasks"}


print(select_provider("ocr_complex", budget="medium"))
print(select_provider("classification", budget="low", volume_per_day=5000))
print(select_provider("reasoning", needs_video=True))

{'provider': 'Anthropic', 'model': 'claude-3-5-sonnet-latest', 'reason': 'Superior at handwriting and complex documents'}
{'provider': 'Google', 'model': 'gemini-1.5-flash', 'reason': 'Minimum cost for high volume'}
{'provider': 'Google', 'model': 'gemini-1.5-pro', 'reason': 'Only one with native video support'}

Real Costs: Compared Scenarios

Prices per million tokens are abstract. What matters is how much your concrete use case costs. This function estimates the cost of a specific scenario for each provider.

SCENARIOS = [
    {"name": "Single image analysis",      "imgs": 1,     "in_tok": 1_000,  "out_tok": 300},
    {"name": "100 product images",          "imgs": 100,   "in_tok": 800,    "out_tok": 200},
    {"name": "30-page document",            "imgs": 1,     "in_tok": 30_000, "out_tok": 2_000},
    {"name": "1,000 screenshots",           "imgs": 1_000, "in_tok": 1_200,  "out_tok": 150},
]

MODELS = ["gpt-4o", "gpt-4o-mini", "claude-3-5-sonnet-latest",
          "claude-3-haiku-20240307", "gemini-1.5-flash", "gemini-1.5-pro"]

DISPLAY_NAMES = {
    "gpt-4o": "GPT-4o",
    "gpt-4o-mini": "GPT-4o-mini",
    "claude-3-5-sonnet-latest": "Claude Sonnet",
    "claude-3-haiku-20240307": "Claude Haiku",
    "gemini-1.5-flash": "Gemini Flash",
    "gemini-1.5-pro": "Gemini Pro",
}

def calculate_scenario_costs(scenarios: list[dict]) -> None:
    header = f"{'Scenario':<30}" + "".join(f"{DISPLAY_NAMES[m]:>16}" for m in MODELS)
    print(header)
    print("-" * len(header))
    for s in scenarios:
        total_in, total_out = s["imgs"] * s["in_tok"], s["imgs"] * s["out_tok"]
        costs = [(total_in * PRICING[m]["input"] + total_out * PRICING[m]["output"]) / 1_000_000 for m in MODELS]
        min_cost = min(costs)
        row = f"{s['name']:<30}" + "".join(
            f"${c:>13.4f}{' ★' if c == min_cost else '  '}" for c in costs
        )
        print(row)

calculate_scenario_costs(SCENARIOS)

Output:

Scenario                                GPT-4o     GPT-4o-mini   Claude Sonnet    Claude Haiku    Gemini Flash      Gemini Pro
------------------------------------------------------------------------------------------------------------------------------
Single image analysis         $       0.0055  $       0.0003  $       0.0075  $       0.0006  $       0.0002 ★$       0.0027
100 product images            $       0.4000  $       0.0240  $       0.5400  $       0.0450  $       0.0120 ★$       0.2000
30-page document              $       0.0950  $       0.0057  $       0.1200  $       0.0100  $       0.0029 ★$       0.0475
1,000 screenshots             $       4.5000  $       0.2700  $       5.8500  $       0.4875  $       0.1350 ★$       2.2500

Interpreting the numbers

• Gemini Flash wins on cost in every scenario. If your task tolerates "good" (not "excellent") quality, the economic decision is obvious.
• GPT-4o-mini is the second cheapest option and offers better quality than Gemini Flash on many tasks.
• Claude Sonnet and GPT-4o have similar costs. The choice between them depends on the task, not the price.
• 1,000 screenshots with GPT-4o cost $4.50. With Gemini Flash, $0.14. In production at scale, that difference is thousands of dollars per month.

Fallback Strategy

In production, APIs fail. Rate limits, timeouts, internal provider errors, scheduled maintenance. If your system depends on a single provider, one 503 from OpenAI stops everything. A VisionRouter with fallback keeps your system running.

Principles

1. Define a priority order based on your task (quality first, or cost first).
2. If the primary provider fails, try the next one automatically.
3. Log every attempt to monitor provider health.

import time
import logging
from dataclasses import dataclass, field
from typing import Callable

logger = logging.getLogger(__name__)


@dataclass
class ProviderAttempt:
    provider: str
    success: bool
    latency_ms: float
    error: str | None = None


@dataclass
class VisionRouter:
    providers: list[dict]
    max_retries: int = 2
    timeout_ms: float = 15_000
    attempts_log: list[ProviderAttempt] = field(default_factory=list)
    total_cost_usd: float = 0.0

    def call(self, image_b64: str, prompt: str) -> dict:
        """Tries each provider in order until it gets a successful response."""
        last_error = None

        for provider_config in self.providers:
            name = provider_config["name"]
            call_fn: Callable = provider_config["fn"]

            for attempt in range(self.max_retries):
                try:
                    start = time.perf_counter()
                    result = call_fn(image_b64, prompt)
                    latency = (time.perf_counter() - start) * 1000

                    self.attempts_log.append(ProviderAttempt(name, True, latency))
                    self.total_cost_usd += result.get("cost", 0)

                    logger.info(f"{name} responded in {latency:.0f}ms (attempt {attempt + 1})")
                    return result

                except Exception as e:
                    latency = (time.perf_counter() - start) * 1000
                    self.attempts_log.append(ProviderAttempt(name, False, latency, str(e)))
                    last_error = e
                    logger.warning(f"{name} failed (attempt {attempt + 1}): {e}")

        raise RuntimeError(f"All providers failed. Last error: {last_error}")

    def health_report(self) -> dict:
        """Generates a success/failure report per provider."""
        from collections import Counter
        success = Counter()
        failure = Counter()
        for a in self.attempts_log:
            if a.success:
                success[a.provider] += 1
            else:
                failure[a.provider] += 1
        return {
            "success_counts": dict(success),
            "failure_counts": dict(failure),
            "total_cost_usd": round(self.total_cost_usd, 6),
            "total_attempts": len(self.attempts_log),
        }

Usage

Each provider is wrapped in a function with the signature (image_b64: str, prompt: str) -> dict that returns {"text": ..., "cost": ...}. Reuse the functions from capsules 02-04.

def call_openai(image_b64: str, prompt: str) -> dict:
    client = OpenAI()
    r = client.chat.completions.create(model="gpt-4o", max_tokens=500,
        messages=[{"role": "user", "content": [
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}}
        ]}])
    return {"text": r.choices[0].message.content,
            "cost": estimate_cost("gpt-4o", r.usage.prompt_tokens, r.usage.completion_tokens)}

def call_anthropic(image_b64: str, prompt: str) -> dict:
    client = anthropic.Anthropic()
    r = client.messages.create(model="claude-3-5-sonnet-latest", max_tokens=500,
        messages=[{"role": "user", "content": [
            {"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": image_b64}},
            {"type": "text", "text": prompt}
        ]}])
    return {"text": r.content[0].text,
            "cost": estimate_cost("claude-3-5-sonnet-latest", r.usage.input_tokens, r.usage.output_tokens)}

router = VisionRouter(providers=[
    {"name": "OpenAI", "fn": call_openai},
    {"name": "Anthropic", "fn": call_anthropic},
])
result = router.call(image_b64="...", prompt="Describe this image")
print(result["text"])
print(router.health_report())

Multi-Provider Strategy in Production

Fallback solves availability: if one fails, use another. But there are scenarios where you want to use multiple providers simultaneously.

Consensus Voting

You send the same image to 2-3 providers and compare responses. If two out of three agree, confidence is high. Useful for critical tasks where an error is costly (medical diagnosis, legal compliance).

import json
from concurrent.futures import ThreadPoolExecutor


def consensus_vote(image_b64: str, prompt: str, providers: list[Callable]) -> dict:
    with ThreadPoolExecutor(max_workers=len(providers)) as executor:
        futures = [executor.submit(fn, image_b64, prompt) for fn in providers]
        responses = [f.result() for f in futures]

    texts = [r["text"] for r in responses]
    total_cost = sum(r["cost"] for r in responses)

    return {
        "responses": texts,
        "total_cost": total_cost,
        "agreement": len(set(texts)) == 1,
    }

When to use consensus

• Critical binary classification (e.g. "Does this document contain personal information?") — if two out of three say "yes", proceed.
• High-value data extraction — if two providers extract the same invoice number, trust the result.
• Do not use it for open-ended descriptions — the responses will always differ in wording even when they're correct.

The cost of consensus

Multiplying by N providers isn't always unviable. If you use Gemini Flash ($0.075/1M) + GPT-4o-mini ($0.15/1M), consensus costs $0.225/1M — still cheaper than a single call to GPT-4o ($2.50/1M) or Claude Sonnet ($3.00/1M).

Multi-Provider Troubleshooting

Inconsistent response formats

Each provider structures text differently. If you ask for JSON, GPT-4o tends to return clean JSON, Claude sometimes adds text before/after, and Gemini may include markdown.

import json
import re


def extract_json_from_response(text: str) -> dict:
    """Extracts JSON from a response that may contain extra text."""
    json_match = re.search(r'\{[\s\S]*\}', text)
    if json_match:
        try:
            return json.loads(json_match.group())
        except json.JSONDecodeError:
            pass
    raise ValueError(f"No valid JSON found in the response: {text[:200]}")

Qualitatively different results

The same prompt may produce results with a different level of detail. Mitigation:

• Use explicit prompts with the exact expected format.
• Include an example schema in the prompt.
• Add "Respond ONLY with the JSON, no extra text" at the end.

API key management

With three providers, you have three keys to rotate, monitor and protect.

import os


REQUIRED_KEYS = {
    "OpenAI": "OPENAI_API_KEY",
    "Anthropic": "ANTHROPIC_API_KEY",
    "Google": "GOOGLE_API_KEY",
}


def validate_api_keys() -> dict[str, bool]:
    """Checks which providers have an API key configured."""
    return {
        provider: bool(os.environ.get(env_var))
        for provider, env_var in REQUIRED_KEYS.items()
    }


print(validate_api_keys())

Cross-provider rate limits

Each provider has its own rate limits. A multi-provider system needs independent tracking:

• OpenAI: headers x-ratelimit-remaining-requests and x-ratelimit-remaining-tokens
• Anthropic: header anthropic-ratelimit-requests-remaining
• Google: limits per model and per minute, varying by tier

If a provider returns 429, your router should automatically try the next one — which is exactly what the VisionRouter from the previous section does.

Exercises

Exercise 1 (Easy): Improved provider selector

Implement select_provider_v2() that takes these parameters and returns the optimal provider:

• task: task type
• budget_per_1000_images: maximum budget in USD for 1000 images
• min_context_tokens: minimum required context window tokens
• requires_video: whether video support is needed

The function must filter out providers that don't meet the requirements and pick the cheapest of the rest.

PROVIDER_SPECS = {
    "gpt-4o": {"context": 128_000, "video": False, "cost_per_1k_imgs": 2.50},
    "gpt-4o-mini": {"context": 128_000, "video": False, "cost_per_1k_imgs": 0.15},
    "claude-3-5-sonnet-latest": {"context": 200_000, "video": False, "cost_per_1k_imgs": 3.00},
    "claude-3-haiku-20240307": {"context": 200_000, "video": False, "cost_per_1k_imgs": 0.25},
    "gemini-1.5-flash": {"context": 1_000_000, "video": True, "cost_per_1k_imgs": 0.075},
    "gemini-1.5-pro": {"context": 1_000_000, "video": True, "cost_per_1k_imgs": 1.25},
}

QUALITY_RANKING = {
    "ocr_complex": ["claude-3-5-sonnet-latest", "gpt-4o", "gemini-1.5-pro"],
    "reasoning": ["claude-3-5-sonnet-latest", "gpt-4o", "gemini-1.5-pro"],
    "classification": ["gpt-4o-mini", "gemini-1.5-flash", "claude-3-haiku-20240307"],
    "ocr_simple": ["gemini-1.5-flash", "gpt-4o-mini", "claude-3-haiku-20240307"],
}


def select_provider_v2(
    task: str,
    budget_per_1000_images: float = 10.0,
    min_context_tokens: int = 0,
    requires_video: bool = False,
) -> dict:
    candidates = []
    for model, specs in PROVIDER_SPECS.items():
        if requires_video and not specs["video"]:
            continue
        if specs["context"] < min_context_tokens:
            continue
        if specs["cost_per_1k_imgs"] > budget_per_1000_images:
            continue
        candidates.append(model)

    if not candidates:
        return {"model": None, "reason": "No provider meets all the requirements"}

    if task in QUALITY_RANKING:
        for preferred in QUALITY_RANKING[task]:
            if preferred in candidates:
                return {"model": preferred, "reason": f"Best quality for {task} within budget"}

    cheapest = min(candidates, key=lambda m: PROVIDER_SPECS[m]["cost_per_1k_imgs"])
    return {"model": cheapest, "reason": "Cheapest option that meets the requirements"}


print(select_provider_v2("ocr_complex", budget_per_1000_images=5.0))
print(select_provider_v2("classification", requires_video=True))
print(select_provider_v2("reasoning", min_context_tokens=500_000))

Exercise 2 (Medium): Cost table generator

Create a function generate_cost_report() that takes a list of custom scenarios and generates a formatted comparison table. Each scenario has: name, number of images, average input tokens per image, and average output tokens per image. The function must:

1. Calculate the total cost per scenario for each model.
2. Mark the cheapest model with ★.
3. Mark the most expensive model with ✗.
4. At the end, show a summary with the total savings of always using the cheapest vs always using the most expensive.

def generate_cost_report(scenarios: list[dict]) -> None:
    models = list(PRICING.keys())
    total_min = 0.0
    total_max = 0.0

    header = f"{'Scenario':<35}" + "".join(f"{m:>22}" for m in models)
    print(header)
    print("=" * len(header))

    for s in scenarios:
        total_in = s["images"] * s["input_tokens"]
        total_out = s["images"] * s["output_tokens"]

        costs = {}
        for model in models:
            p = PRICING[model]
            costs[model] = (total_in * p["input"] + total_out * p["output"]) / 1_000_000

        min_cost = min(costs.values())
        max_cost = max(costs.values())
        total_min += min_cost
        total_max += max_cost

        row = f"{s['name']:<35}"
        for model in models:
            c = costs[model]
            marker = " ★" if c == min_cost else (" ✗" if c == max_cost else "  ")
            row += f"  ${c:>16.4f}{marker}"
        print(row)

    print("=" * len(header))
    savings = total_max - total_min
    pct = (savings / total_max * 100) if total_max > 0 else 0
    print(f"\nTotal savings (always cheapest vs always most expensive): ${savings:.4f} ({pct:.1f}%)")


my_scenarios = [
    {"name": "Onboarding: 50 ID documents", "images": 50, "input_tokens": 1500, "output_tokens": 400},
    {"name": "Catalog: 500 products", "images": 500, "input_tokens": 800, "output_tokens": 200},
    {"name": "Audit: 10 long contracts", "images": 10, "input_tokens": 50_000, "output_tokens": 5_000},
    {"name": "Monitoring: 2000 screenshots/day", "images": 2000, "input_tokens": 1000, "output_tokens": 100},
]

generate_cost_report(my_scenarios)

Exercise 3 (Hard): Complete VisionRouter with metrics

Extend the VisionRouter to implement:

1. Retry with exponential backoff: The first retry waits 1s, the second 2s, the third 4s.
2. Structured logging: Every attempt is logged with timestamp, provider, latency, success/failure, and error code if applicable.
3. Accumulated metrics: get_metrics() returns a dict with: total requests, success rate per provider, p50 and p95 latency per provider, total accumulated cost.
4. Dynamic selection: If a provider has >50% failures in the last 10 calls, it is skipped automatically.

import time, statistics, logging
from datetime import datetime
from dataclasses import dataclass, field
from typing import Callable
from collections import deque

logger = logging.getLogger(__name__)

@dataclass
class AttemptRecord:
    timestamp: str
    provider: str
    success: bool
    latency_ms: float
    cost_usd: float
    error_code: str | None = None

@dataclass
class SmartVisionRouter:
    providers: list[dict]
    max_retries: int = 3
    base_delay_s: float = 1.0
    history: list[AttemptRecord] = field(default_factory=list)
    recent_by_provider: dict[str, deque] = field(default_factory=dict)
    total_cost: float = 0.0

    def __post_init__(self):
        for p in self.providers:
            self.recent_by_provider[p["name"]] = deque(maxlen=10)

    def _is_healthy(self, name: str) -> bool:
        recent = self.recent_by_provider[name]
        if len(recent) < 5:
            return True
        return sum(1 for r in recent if not r.success) / len(recent) <= 0.5

    def _record(self, name, success, latency, cost=0, error=None):
        rec = AttemptRecord(datetime.now().isoformat(), name, success, round(latency, 1), cost, error)
        self.history.append(rec)
        self.recent_by_provider[name].append(rec)
        if success:
            self.total_cost += cost
        return rec

    def call(self, image_b64: str, prompt: str) -> dict:
        last_error = None
        for prov in self.providers:
            name, call_fn = prov["name"], prov["fn"]
            if not self._is_healthy(name):
                logger.warning(f"Skipping {name}: failure rate > 50%")
                continue
            for attempt in range(self.max_retries):
                if attempt > 0:
                    time.sleep(self.base_delay_s * (2 ** attempt))
                start = time.perf_counter()
                try:
                    result = call_fn(image_b64, prompt)
                    latency = (time.perf_counter() - start) * 1000
                    self._record(name, True, latency, result.get("cost", 0))
                    return result
                except Exception as e:
                    latency = (time.perf_counter() - start) * 1000
                    self._record(name, False, latency, error=type(e).__name__)
                    last_error = e
                    logger.warning(f"{name} attempt {attempt+1}/{self.max_retries}: {e}")
        raise RuntimeError(f"All providers failed. Last error: {last_error}")

    def get_metrics(self) -> dict:
        metrics = {}
        for name in self.recent_by_provider:
            recs = [r for r in self.history if r.provider == name]
            if not recs:
                continue
            ok = [r for r in recs if r.success]
            lats = sorted(r.latency_ms for r in ok)
            metrics[name] = {
                "total_calls": len(recs),
                "success_rate": len(ok) / len(recs),
                "latency_p50_ms": round(statistics.median(lats), 1) if lats else None,
                "latency_p95_ms": round(lats[int(len(lats) * 0.95)], 1) if len(lats) >= 2 else (round(lats[0], 1) if lats else None),
            }
        return {"total_requests": len(self.history), "total_cost_usd": round(self.total_cost, 6), "providers": metrics}

Summary

• There is no universally best provider — the choice depends on the task, the budget and the technical requirements.
• Gemini Flash dominates on cost. Claude Sonnet dominates on reasoning. GPT-4o is the general balance.
• Run benchmarks with your own images — general rankings don't always apply to specific domains.
• In production, implement fallback between providers to guarantee availability.
• Calculate costs with real scenarios before committing to a provider.

Additional Resources

1. OpenAI Vision Pricing — Up-to-date costs for GPT-4o and GPT-4o-mini
2. Anthropic Pricing — Costs for Claude 3.5 Sonnet, Opus and Haiku
3. Google AI Pricing — Costs for Gemini Flash and Pro
4. Artificial Analysis LLM Benchmarks — Independent quality and speed benchmarks
5. LMSYS Chatbot Arena — Model rankings based on human evaluations