Module 2: Vision + LLMs

8. Project: Multi-Provider Image Analyzer

Description

This project closes Module 2 by integrating everything you learned: OpenAI (capsule 02), Anthropic (capsule 03) and Google (capsule 04) vision APIs, the comparison between providers (capsule 05), the analysis patterns — description, OCR, classification, extraction, Q&A — (capsule 06) and handling multiple images (capsule 07).

You're going to build an Image Analyzer — a system that takes an image, chooses the analysis pattern, routes to the optimal provider with automatic fallback, calculates real costs and returns a structured result with all the metadata.

Why it matters: In production, no provider has 100% uptime. If your system depends on a single provider and that provider fails at 3am, your pipeline stops. An analyzer with automatic fallback, cost tracking and logging is the difference between a prototype and an operable system.

Connection to the guide: In Module 3 (Document Understanding), this analyzer becomes the vision engine of the Document Analyzer. In Module 8 (Final Project), the routing and fallback logic is reused to handle multi-page documents with different providers per page.

Technical Specifications

Input

Parameter	Type	Description
`image_path`	`str`	Local path to image (PNG, JPEG, WebP, GIF)
`pattern`	`str`	Analysis pattern: `description`, `ocr`, `classification`, `extraction`, `qa`
`provider`	`str \| None`	Preferred provider. `None` = auto-selection
`priority`	`str`	`"quality"`, `"cost"` or `"balanced"`
`question`	`str \| None`	Question for the `qa` pattern

Output

@dataclass
class AnalysisResult:
    response: str
    pattern: str
    provider: str
    model: str
    latency_seconds: float
    input_tokens: int
    output_tokens: int
    estimated_cost_usd: float
    warnings: list[str]
    fallback_log: list[str]

Requirements

Support the 3 providers with real code (no stubs)
5 analysis patterns with dedicated prompts
Automatic fallback with a configurable chain
Real cost calculation based on tokens consumed
Logging of each attempt (success and failure)
Image validation before sending

Step 1: Configuration and Constants

import os
import time
import math
import base64
import logging
from enum import Enum
from pathlib import Path
from dataclasses import dataclass, field

logger = logging.getLogger("image_analyzer")
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")


class AnalysisPattern(Enum):
    DESCRIPTION = "description"
    OCR = "ocr"
    CLASSIFICATION = "classification"
    EXTRACTION = "extraction"
    QA = "qa"


@dataclass
class AnalysisResult:
    response: str = ""
    pattern: str = ""
    provider: str = ""
    model: str = ""
    latency_seconds: float = 0.0
    input_tokens: int = 0
    output_tokens: int = 0
    estimated_cost_usd: float = 0.0
    warnings: list[str] = field(default_factory=list)
    fallback_log: list[str] = field(default_factory=list)


PROVIDER_CONFIG = {
    "openai": {
        "model_quality": "gpt-4o",
        "model_cost": "gpt-4o-mini",
        "pricing": {
            "gpt-4o": {"input": 2.50, "output": 10.00},
            "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        },
        "max_image_mb": 20,
        "supported_formats": {".jpg", ".jpeg", ".png", ".gif", ".webp"},
    },
    "anthropic": {
        "model_quality": "claude-3-5-sonnet-latest",
        "model_cost": "claude-3-haiku-20240307",
        "pricing": {
            "claude-3-5-sonnet-latest": {"input": 3.00, "output": 15.00},
            "claude-3-haiku-20240307": {"input": 0.25, "output": 1.25},
        },
        "max_image_mb": 5,
        "supported_formats": {".jpg", ".jpeg", ".png", ".gif", ".webp"},
    },
    "google": {
        "model_quality": "gemini-1.5-pro",
        "model_cost": "gemini-2.0-flash",
        "pricing": {
            "gemini-1.5-pro": {"input": 1.25, "output": 5.00},
            "gemini-2.0-flash": {"input": 0.10, "output": 0.40},
        },
        "max_image_mb": 20,
        "supported_formats": {".jpg", ".jpeg", ".png", ".gif", ".webp"},
    },
}

MIME_TYPES = {
    ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png",
    ".gif": "image/gif", ".webp": "image/webp",
}

Step 2: Prepare Image per Provider

Each provider expects the image in a different format. OpenAI accepts data URIs (Base64 with a MIME prefix), Anthropic needs the raw Base64 with a separate media_type, and Gemini receives raw bytes.

def validate_image(image_path: str) -> tuple[Path, list[str]]:
    """Validates the image's existence, format and size."""
    warnings = []
    path = Path(image_path)

    if not path.exists():
        raise FileNotFoundError(f"Image not found: {image_path}")

    ext = path.suffix.lower()
    if ext not in MIME_TYPES:
        raise ValueError(f"Unsupported format: {ext}")

    size_mb = path.stat().st_size / (1024 * 1024)
    if size_mb > 20:
        raise ValueError(f"Image too large: {size_mb:.1f} MB (maximum 20 MB)")
    if size_mb > 4:
        warnings.append(f"Large image ({size_mb:.1f} MB) — may be slow with Anthropic (5 MB limit)")

    return path, warnings


def prepare_image_openai(path: Path) -> dict:
    """Data URI for OpenAI."""
    mime = MIME_TYPES[path.suffix.lower()]
    encoded = base64.b64encode(path.read_bytes()).decode()
    return {
        "type": "image_url",
        "image_url": {"url": f"data:{mime};base64,{encoded}", "detail": "high"},
    }


def prepare_image_anthropic(path: Path) -> dict:
    """Base64 with separate media_type for Anthropic."""
    mime = MIME_TYPES[path.suffix.lower()]
    encoded = base64.b64encode(path.read_bytes()).decode()
    return {
        "type": "image",
        "source": {"type": "base64", "media_type": mime, "data": encoded},
    }


def prepare_image_gemini(path: Path) -> dict:
    """Raw bytes with MIME for Gemini."""
    mime = MIME_TYPES[path.suffix.lower()]
    return {"mime_type": mime, "data": path.read_bytes()}

Step 3: Generate Prompt per Pattern

Each analysis pattern produces a specialized prompt. These come directly from capsule 06.

PATTERN_PROMPTS = {
    AnalysisPattern.DESCRIPTION: (
        "Describe this image objectively.\n"
        "Include: main scene, visible objects, dominant colors, visible text (if any).\n"
        "Maximum 100 words."
    ),
    AnalysisPattern.OCR: (
        "Extract all the visible text in this image.\n"
        "Keep the structure: paragraphs, lists, tables.\n"
        "If there are tables, use markdown format.\n"
        "Do not invent text that is not visible."
    ),
    AnalysisPattern.CLASSIFICATION: (
        "Classify this image into exactly one of the following categories:\n"
        "document, photo, diagram, screenshot, art, meme, other.\n"
        "Respond ONLY with the category name."
    ),
    AnalysisPattern.EXTRACTION: (
        "Extract all the structured information visible in this image.\n"
        "Respond ONLY with valid JSON.\n"
        "Include all the fields you can identify: dates, numbers, names, amounts.\n"
        "Use null for ambiguous fields."
    ),
    AnalysisPattern.QA: (
        "Answer the following question based ONLY on what is visible in the image.\n"
        "If the information is not in the image, respond 'Not visible in the image'.\n\n"
        "Question: {question}"
    ),
}


def get_prompt(pattern: AnalysisPattern, question: str | None = None) -> str:
    """Returns the prompt for the given pattern."""
    prompt = PATTERN_PROMPTS[pattern]
    if pattern == AnalysisPattern.QA:
        if not question:
            raise ValueError("The 'qa' pattern requires a question")
        prompt = prompt.format(question=question)
    return prompt

Step 4: Call Each Provider

Each function sends the image with the prompt to the corresponding provider and returns a partial AnalysisResult with the response, tokens and latency.

OpenAI

def _call_openai(path: Path, prompt: str, model: str) -> AnalysisResult:
    from openai import OpenAI

    client = OpenAI()
    image_block = prepare_image_openai(path)

    start = time.time()
    response = client.chat.completions.create(
        model=model,
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": prompt},
                image_block,
            ],
        }],
        max_tokens=1024,
    )
    latency = time.time() - start

    usage = response.usage
    return AnalysisResult(
        response=response.choices[0].message.content,
        provider="openai",
        model=model,
        latency_seconds=round(latency, 2),
        input_tokens=usage.prompt_tokens,
        output_tokens=usage.completion_tokens,
    )

Anthropic

def _call_anthropic(path: Path, prompt: str, model: str) -> AnalysisResult:
    import anthropic

    size_mb = path.stat().st_size / (1024 * 1024)
    if size_mb > PROVIDER_CONFIG["anthropic"]["max_image_mb"]:
        raise ValueError(f"Image exceeds Anthropic limit: {size_mb:.1f} MB > 5 MB")

    client = anthropic.Anthropic()
    image_block = prepare_image_anthropic(path)

    start = time.time()
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": [image_block, {"type": "text", "text": prompt}],
        }],
    )
    latency = time.time() - start

    return AnalysisResult(
        response=response.content[0].text,
        provider="anthropic",
        model=model,
        latency_seconds=round(latency, 2),
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
    )

Google Gemini

def _call_gemini(path: Path, prompt: str, model: str) -> AnalysisResult:
    import google.generativeai as genai

    genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
    gemini_model = genai.GenerativeModel(model)
    image_part = prepare_image_gemini(path)

    start = time.time()
    response = gemini_model.generate_content([prompt, image_part])
    latency = time.time() - start

    input_tokens = 0
    output_tokens = 0
    if hasattr(response, "usage_metadata") and response.usage_metadata:
        input_tokens = getattr(response.usage_metadata, "prompt_token_count", 0)
        output_tokens = getattr(response.usage_metadata, "candidates_token_count", 0)

    return AnalysisResult(
        response=response.text,
        provider="google",
        model=model,
        latency_seconds=round(latency, 2),
        input_tokens=input_tokens,
        output_tokens=output_tokens,
    )

Step 5: Router with Fallback

The router tries providers in order. If the first one fails, it moves to the next and logs the failure.

CALLER_MAP = {
    "openai": _call_openai,
    "anthropic": _call_anthropic,
    "google": _call_gemini,
}

DEFAULT_CHAINS = {
    "quality": ["openai", "anthropic", "google"],
    "cost": ["google", "openai", "anthropic"],
    "balanced": ["openai", "google", "anthropic"],
}


def route_with_fallback(
    path: Path,
    prompt: str,
    priority: str = "balanced",
    preferred_provider: str | None = None,
) -> AnalysisResult:
    """Tries providers in order with automatic fallback."""
    chain = list(DEFAULT_CHAINS.get(priority, DEFAULT_CHAINS["balanced"]))

    if preferred_provider and preferred_provider in CALLER_MAP:
        chain.remove(preferred_provider) if preferred_provider in chain else None
        chain.insert(0, preferred_provider)

    fallback_log = []

    for provider_name in chain:
        config = PROVIDER_CONFIG[provider_name]
        model = config["model_quality"] if priority == "quality" else config["model_cost"]
        caller = CALLER_MAP[provider_name]

        try:
            logger.info(f"Trying {provider_name} ({model})...")
            result = caller(path, prompt, model)
            result.fallback_log = fallback_log
            if fallback_log:
                result.warnings.append(f"Primary provider failed, fallback used: {provider_name}")
            return result
        except Exception as e:
            msg = f"{provider_name} ({model}) failed: {type(e).__name__}: {e}"
            logger.warning(msg)
            fallback_log.append(msg)
            continue

    error_result = AnalysisResult(
        response="",
        warnings=["All providers failed"],
        fallback_log=fallback_log,
    )
    raise RuntimeError(f"All providers failed. Log: {fallback_log}")

Step 6: Cost Calculation

The real cost is calculated from the tokens reported by each provider, using the price table per million tokens.

def calculate_cost(result: AnalysisResult) -> float:
    """Calculates real cost based on tokens consumed."""
    config = PROVIDER_CONFIG.get(result.provider)
    if not config:
        return 0.0

    pricing = config["pricing"].get(result.model)
    if not pricing:
        return 0.0

    input_cost = (result.input_tokens / 1_000_000) * pricing["input"]
    output_cost = (result.output_tokens / 1_000_000) * pricing["output"]
    return round(input_cost + output_cost, 6)


def estimate_cost_before_call(
    image_path: str, priority: str, provider: str | None = None
) -> dict:
    """Estimates cost before calling the API (without making the call)."""
    path = Path(image_path)
    size_mb = path.stat().st_size / (1024 * 1024)
    image_tokens = max(85, int(size_mb * 1000))
    output_tokens_est = 200

    estimates = {}
    for name, config in PROVIDER_CONFIG.items():
        model = config["model_quality"] if priority == "quality" else config["model_cost"]
        pricing = config["pricing"][model]
        cost = (image_tokens / 1_000_000) * pricing["input"] + (output_tokens_est / 1_000_000) * pricing["output"]
        estimates[name] = {"model": model, "estimated_cost": round(cost, 6)}

    return estimates

Step 7: Main Function — Integration

def analyze_image(
    image_path: str,
    pattern: str = "description",
    provider: str | None = None,
    priority: str = "balanced",
    question: str | None = None,
) -> AnalysisResult:
    """Analyzes an image with the given pattern and provider.

    Args:
        image_path: Local path to the image.
        pattern: Analysis pattern (description, ocr, classification, extraction, qa).
        provider: Preferred provider. None = auto-selection based on priority.
        priority: "quality", "cost" or "balanced".
        question: Question for the qa pattern.

    Returns:
        AnalysisResult with response, metadata, costs and fallback log.
    """
    path, warnings = validate_image(image_path)

    analysis_pattern = AnalysisPattern(pattern)
    prompt = get_prompt(analysis_pattern, question)

    result = route_with_fallback(path, prompt, priority, provider)

    result.pattern = analysis_pattern.value
    result.estimated_cost_usd = calculate_cost(result)
    result.warnings.extend(warnings)

    logger.info(
        f"Analysis completed: pattern={result.pattern} provider={result.provider} "
        f"model={result.model} tokens={result.input_tokens}+{result.output_tokens} "
        f"cost=${result.estimated_cost_usd:.6f} latency={result.latency_seconds}s"
    )

    return result

Demo Code

def print_result(result: AnalysisResult):
    print(f"\n{'='*60}")
    print(f"  Pattern:    {result.pattern}")
    print(f"  Provider:   {result.provider} ({result.model})")
    print(f"  Latency:    {result.latency_seconds}s")
    print(f"  Tokens:     {result.input_tokens} in / {result.output_tokens} out")
    print(f"  Cost:       ${result.estimated_cost_usd:.6f}")
    if result.warnings:
        for w in result.warnings:
            print(f"  ⚠ {w}")
    if result.fallback_log:
        print(f"  Fallback:   {len(result.fallback_log)} failed attempts")
    print(f"  Response:   {result.response[:200]}...")
    print(f"{'='*60}")


# --- Demo ---
if __name__ == "__main__":
    # Description with automatic provider selection
    r1 = analyze_image("./sample.jpg", pattern="description", priority="balanced")
    print_result(r1)

    # OCR prioritizing quality
    r2 = analyze_image("./invoice.png", pattern="ocr", priority="quality")
    print_result(r2)

    # Classification prioritizing cost
    r3 = analyze_image("./screenshot.png", pattern="classification", priority="cost")
    print_result(r3)

    # Q&A with a specific provider
    r4 = analyze_image(
        "./diagram.png",
        pattern="qa",
        provider="anthropic",
        question="How many nodes does the diagram have?",
    )
    print_result(r4)

    # Structured extraction
    r5 = analyze_image("./receipt.jpg", pattern="extraction", priority="quality")
    print_result(r5)

Expected output (schematic):

============================================================
  Pattern:    description
  Provider:   openai (gpt-4o-mini)
  Latency:    1.34s
  Tokens:     1150 in / 87 out
  Cost:       $0.000225
  Response:   The image shows an urban landscape at sunset...
============================================================

Extension 1: Batch Analysis

Process multiple images with accumulated cost tracking and progress reporting.

@dataclass
class BatchResult:
    total: int = 0
    successful: int = 0
    failed: int = 0
    total_cost_usd: float = 0.0
    total_latency_seconds: float = 0.0
    by_provider: dict = field(default_factory=dict)
    results: list[AnalysisResult] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)


def analyze_batch(
    image_paths: list[str],
    pattern: str = "description",
    priority: str = "cost",
    budget: float | None = None,
) -> BatchResult:
    """Analyzes multiple images with cost and progress tracking."""
    batch = BatchResult()

    for i, img_path in enumerate(image_paths):
        batch.total += 1
        logger.info(f"[{i+1}/{len(image_paths)}] Processing: {img_path}")

        if budget and batch.total_cost_usd >= budget:
            batch.errors.append(f"Budget exhausted (${budget:.4f}) at image {i+1}")
            break

        try:
            result = analyze_image(img_path, pattern=pattern, priority=priority)
            batch.successful += 1
            batch.total_cost_usd += result.estimated_cost_usd
            batch.total_latency_seconds += result.latency_seconds
            batch.by_provider[result.provider] = batch.by_provider.get(result.provider, 0) + 1
            batch.results.append(result)
        except Exception as e:
            batch.failed += 1
            batch.errors.append(f"{img_path}: {e}")

    return batch


Usage:

```python
images = ["./img1.jpg", "./img2.png", "./img3.webp"]
batch = analyze_batch(images, pattern="classification", priority="cost", budget=0.05)
print(f"Successful: {batch.successful}/{batch.total}, Cost: ${batch.total_cost_usd:.6f}")

Extension 2: Consensus Voting

Sends the same image to all 3 providers and combines the results. Useful for classification tasks where you want high confidence.

@dataclass
class ConsensusResult:
    individual_results: list[AnalysisResult] = field(default_factory=list)
    consensus_response: str = ""
    agreement_score: float = 0.0
    total_cost_usd: float = 0.0
    total_latency_seconds: float = 0.0


def analyze_with_consensus(
    image_path: str,
    pattern: str = "classification",
    question: str | None = None,
) -> ConsensusResult:
    """Sends the image to all 3 providers and votes by consensus."""
    path, warnings = validate_image(image_path)
    prompt = get_prompt(AnalysisPattern(pattern), question)
    consensus = ConsensusResult()

    providers = [
        ("openai", PROVIDER_CONFIG["openai"]["model_cost"]),
        ("anthropic", PROVIDER_CONFIG["anthropic"]["model_cost"]),
        ("google", PROVIDER_CONFIG["google"]["model_cost"]),
    ]

    for provider_name, model in providers:
        caller = CALLER_MAP[provider_name]
        try:
            result = caller(path, prompt, model)
            result.pattern = pattern
            result.estimated_cost_usd = calculate_cost(result)
            consensus.individual_results.append(result)
            consensus.total_cost_usd += result.estimated_cost_usd
            consensus.total_latency_seconds += result.latency_seconds
        except Exception as e:
            logger.warning(f"Consensus: {provider_name} failed: {e}")

    if not consensus.individual_results:
        raise RuntimeError("No provider responded for consensus")

    responses = [r.response.strip().lower() for r in consensus.individual_results]
    from collections import Counter
    vote_counts = Counter(responses)
    winner, count = vote_counts.most_common(1)[0]
    consensus.consensus_response = winner
    consensus.agreement_score = count / len(responses)

    return consensus

Project Troubleshooting

Problem 1: Anthropic rejects large images

Symptom: anthropic.BadRequestError with images larger than ~4 MB.

Cause: Anthropic has a 5 MB limit per image, but Base64 encoding increases the size ~33%. A 4 MB image generates ~5.3 MB in Base64.

Solution: Check the size before sending and resize if necessary:

def resize_if_needed(path: Path, max_raw_mb: float = 3.5) -> bytes:
    from PIL import Image
    import io

    raw_bytes = path.read_bytes()
    if len(raw_bytes) / (1024 * 1024) <= max_raw_mb:
        return raw_bytes

    with Image.open(path) as img:
        img = img.convert("RGB")
        quality = 85
        while quality >= 30:
            buffer = io.BytesIO()
            img.save(buffer, "JPEG", quality=quality, optimize=True)
            if len(buffer.getvalue()) / (1024 * 1024) <= max_raw_mb:
                return buffer.getvalue()
            quality -= 10
        img = img.resize((img.width // 2, img.height // 2))
        buffer = io.BytesIO()
        img.save(buffer, "JPEG", quality=80, optimize=True)
        return buffer.getvalue()

Problem 2: Gemini doesn't report tokens

Symptom: input_tokens and output_tokens are 0 after calling Gemini.

Cause: Not all Gemini models include usage_metadata in the response.

Solution: The code already handles this with getattr and defaults to 0. To estimate tokens when they're not reported:

if result.input_tokens == 0 and result.provider == "google":
    size_kb = path.stat().st_size / 1024
    result.input_tokens = max(258, int(size_kb * 1.5))
    result.output_tokens = len(result.response.split()) * 2
    result.warnings.append("Estimated tokens (Gemini did not report usage)")

Problem 3: Rate limit with all providers simultaneously

Symptom: In consensus voting, all 3 providers fail due to rate limit if you send many images quickly.

Solution: Add a delay between calls in the consensus loop:

import time

for provider_name, model in providers:
    try:
        result = caller(path, prompt, model)
        consensus.individual_results.append(result)
    except Exception as e:
        logger.warning(f"{provider_name} failed: {e}")
    time.sleep(0.5)

Problem 4: `FileNotFoundError` with relative paths

Symptom: The path "./images/photo.jpg" fails even though the file exists.

Cause: The script's working directory is not what you expect.

Solution: Use absolute paths or resolve relative ones:

path = Path(image_path).resolve()

Problem 5: Inconsistent responses between providers for OCR

Symptom: OpenAI extracts text correctly, but Gemini omits sections or invents text.

Cause: The models have different OCR performance. GPT-4o and Claude 3.5 Sonnet are superior at accurate OCR.

Solution: For OCR in production, prioritize OpenAI or Anthropic. Use analyze_image(path, pattern="ocr", priority="quality", provider="openai").

Completeness Checklist

Core functionality:

analyze_image() accepts the 5 parameters (image_path, pattern, provider, priority, question)
Returns AnalysisResult with all fields populated
Supports the 5 patterns: description, ocr, classification, extraction, qa
The qa pattern validates that question is not None

Providers:

_call_openai() sends the image as a data URI and parses usage
_call_anthropic() sends as Base64 with media_type and validates size
_call_gemini() sends as raw bytes and handles missing usage_metadata
All 3 providers return AnalysisResult with tokens and latency

Fallback:

Router tries providers in order based on priority
If the preferred one fails, it moves to the next
Each failure is logged in fallback_log
If all fail, it raises RuntimeError with the full log

Cost calculation:

calculate_cost() uses real tokens and per-provider/model prices
Cost is included in the final result
estimate_cost_before_call() works without making an API call

Validation and errors:

Nonexistent image → FileNotFoundError
Unsupported format → ValueError
Image > 20 MB → ValueError
Large image generates a warning (not an error)

Extensions:

Batch analysis with budget and reporting
Consensus voting with agreement_score

Exercises

Exercise 1: Analysis cache (Medium)

Implement a cache layer that avoids re-analyzing the same combination of image + pattern + provider. Use a hash of the image content (not the path) as the key. The cache must persist to disk as JSON.

Requirements:

SHA-256 hash of the image content
Composite key: {image_hash}_{pattern}_{provider}
Storage in ./cache/analysis_cache.json
get() and set() methods in an AnalysisCache class
Configurable TTL (default 24 hours)

See solution

import json
import hashlib
from datetime import datetime, timedelta


class AnalysisCache:
    def __init__(self, cache_dir: str = "./cache", ttl_hours: int = 24):
        self.cache_path = Path(cache_dir) / "analysis_cache.json"
        self.cache_path.parent.mkdir(parents=True, exist_ok=True)
        self.ttl = timedelta(hours=ttl_hours)
        self._cache = self._load()

    def _load(self) -> dict:
        if self.cache_path.exists():
            return json.loads(self.cache_path.read_text())
        return {}

    def _save(self):
        self.cache_path.write_text(json.dumps(self._cache, indent=2, default=str))

    def _image_hash(self, image_path: str) -> str:
        return hashlib.sha256(Path(image_path).read_bytes()).hexdigest()[:16]

    def _key(self, image_path: str, pattern: str, provider: str) -> str:
        return f"{self._image_hash(image_path)}_{pattern}_{provider}"

    def get(self, image_path: str, pattern: str, provider: str) -> AnalysisResult | None:
        key = self._key(image_path, pattern, provider)
        entry = self._cache.get(key)
        if not entry:
            return None
        cached_at = datetime.fromisoformat(entry["cached_at"])
        if datetime.now() - cached_at > self.ttl:
            del self._cache[key]
            self._save()
            return None
        data = entry["result"]
        result = AnalysisResult(**{k: v for k, v in data.items() if k in AnalysisResult.__dataclass_fields__})
        result.warnings.append("Result from cache")
        return result

    def set(self, image_path: str, result: AnalysisResult):
        key = self._key(image_path, result.pattern, result.provider)
        self._cache[key] = {
            "cached_at": datetime.now().isoformat(),
            "result": {
                "response": result.response,
                "pattern": result.pattern,
                "provider": result.provider,
                "model": result.model,
                "latency_seconds": result.latency_seconds,
                "input_tokens": result.input_tokens,
                "output_tokens": result.output_tokens,
                "estimated_cost_usd": result.estimated_cost_usd,
            },
        }
        self._save()


cache = AnalysisCache()

def analyze_image_cached(image_path: str, pattern: str = "description", **kwargs) -> AnalysisResult:
    provider = kwargs.get("provider", "auto")
    cached = cache.get(image_path, pattern, provider)
    if cached:
        logger.info(f"Cache hit: {pattern}/{provider}")
        return cached
    result = analyze_image(image_path, pattern=pattern, **kwargs)
    cache.set(image_path, result)
    return result

Exercise 2: Automatic quality evaluation (Hard)

Implement a system that sends the same image to all 3 providers and then uses an LLM to evaluate the quality of each response with a score from 1 to 10 on: precision, completeness and clarity.