Build a vision analysis API with GPT-4o + Gemini concurrently

Introduction

When you ship a vision feature behind a single model provider, you inherit that provider's blind spots and outage windows — GPT-4o misses chart details Gemini catches, and a one-vendor incident takes every image-aware endpoint down with it. Teams that fan out to two providers concurrently get richer outputs and survive a single-vendor failure without doubling wall-clock latency.

By the end of this lesson you'll be able to wire a POST /api/v1/vision/analyze endpoint that dispatches the same image to GPT-4o and Gemini 2.5 Flash in parallel, forces both into a shared Pydantic schema via Instructor, merges their structured outputs, and returns a unified response your downstream consumers can trust.

Key Terminology

• Dual-provider fan-out: dispatching the same image+prompt payload to GPT-4o and Gemini 2.5 Flash concurrently via asyncio.gather so total latency tracks the slower provider rather than the sum.
• Instructor patching: wrapping litellm.acompletion with instructor.from_litellm(..., mode=instructor.Mode.JSON) so each provider returns a validated VisionResult Pydantic object instead of raw JSON.
• Consensus objects: object labels emitted by both providers in their DetectedObject lists, computed by _merge_results and surfaced on DualVisionResponse.consensus_objects as a cross-provider agreement signal.

Concepts

Resilience and Production Considerations

The return_exceptions=True pattern in asyncio.gather is the minimum viable resilience strategy—it prevents one provider's failure from killing the request, but it does not address timeouts, retries, or circuit breaking. In production, wrap each _call_provider call with asyncio.wait_for to enforce a per-provider timeout (typically 10–15 seconds for vision models processing high-resolution images). If a provider consistently times out, a circuit breaker library like aiobreaker can temporarily remove it from the fan-out pool, reducing wasted compute and improving median latency.

Instructor's automatic retry mechanism handles transient JSON parsing failures—when a model returns output that does not validate against the Pydantic schema, Instructor re-sends the request with the validation error appended to the prompt, giving the model a chance to self-correct. Configure max_retries=2 explicitly in production to avoid unbounded retry loops that inflate costs. Set temperature=0.1 (not 0.0) because some providers treat exact zero as a special case that disables sampling entirely, which can cause non-deterministic behavior depending on the provider's implementation.

For the content safety dimension, the raw image bytes flowing through this endpoint should pass through Llama Guard 4 classification (covered later in this chapter) before reaching the vision models. Inject safety middleware at the service layer—between the analyze method's image encoding step and the fan-out dispatch—so that policy-violating images are rejected with a 422 response before incurring any model API costs. This placement ensures that neither GPT-4o nor Gemini 2.5 Flash ever processes a flagged image, which is both a cost optimization and a compliance requirement for applications handling user-uploaded content.

Code Walkthrough

Architecture Overview

Before examining code, study how the request flows through the system. The FastAPI endpoint receives the image and prompt, hands them to VisionAnalysisService, which fans out to both providers concurrently. Each provider call is wrapped by Instructor to enforce the same VisionResult Pydantic schema. Results merge at the service layer and return as a unified DualVisionResponse.

• Participants: Client, the FastAPI route POST /api/v1/vision/analyze, VisionAnalysisService, and the two LiteLLM-fronted models (GPT-4o, Gemini 2.5 Flash).
• Initial dispatch: The client uploads an image and prompt; FastAPI hands off to VisionAnalysisService.analyze.
• Concurrent fan-out: Inside the par/and/end block, both acompletion calls run in parallel; Instructor (patched once in __init__) enforces the VisionResult schema on each provider response.
• Merge and respond: The service computes consensus objects and merged tags, then returns DualVisionResponse, which FastAPI serializes to JSON.

This fan-out pattern keeps total latency close to the slower of the two providers rather than the sum—typically under 4 seconds for a 1 MB image when both providers respond within their respective SLAs.

Defining the Response Schema

The schema design is the most consequential decision in the entire pipeline. Both providers must produce identical output shapes so the service layer can merge results without provider-specific deserialization logic. The following code defines the VisionResult Pydantic model that Instructor will enforce on both GPT-4o and Gemini 2.5 Flash responses, along with the DualVisionResponse wrapper that the API endpoint returns. Note how DetectedObject uses a confidence field bounded between 0 and 1 via Pydantic's Field constraints, and how VisionResult includes a provider literal so the client always knows which model produced each sub-result.

1from pydantic import BaseModel, Field
2from typing import Literal
3
4class DetectedObject(BaseModel):
5    """A single object or region identified in the image."""
6    label: str = Field(
7        ..., description="Canonical name of the detected object"
8    )
9    confidence: float = Field(
10        ..., ge=0.0, le=1.0,
11        description="Model confidence score between 0 and 1"
12    )
13    description: str = Field(
14        ..., description="One-sentence description of the object in context"
15    )
16
17class VisionResult(BaseModel):
18    """Structured analysis from a single vision model provider."""
19    provider: Literal["gpt-4o", "gemini-2.5-flash"]
20    summary: str = Field(
21        ..., description="2-3 sentence overall image description"
22    )
23    objects: list[DetectedObject] = Field(
24        default_factory=list,
25        description="Objects detected in the image"
26    )
27    text_content: str | None = Field(
28        None, description="Any text extracted via OCR, or None"
29    )
30    scene_tags: list[str] = Field(
31        default_factory=list,
32        description="High-level scene classification tags"
33    )
34
35class DualVisionResponse(BaseModel):
36    """Merged response returned to the API consumer."""
37    results: list[VisionResult] = Field(
38        ..., min_length=1, max_length=2
39    )
40    merged_tags: list[str] = Field(
41        default_factory=list,
42        description="Union of scene_tags from all providers"
43    )
44    consensus_objects: list[str] = Field(
45        default_factory=list,
46        description="Object labels found by both providers"
47    )

Service Layer and FastAPI Endpoint

VisionAnalysisService encapsulates all provider interaction and is consumed by a thin FastAPI route. The constructor patches litellm.acompletion with Instructor once so every subsequent call returns a validated VisionResult. analyze base64-encodes the image, builds the multi-modal payload both OpenAI and Gemini accept through LiteLLM's normalized interface, and dispatches the two _call_provider coroutines through asyncio.gather(..., return_exceptions=True) so one provider failure cannot crash the request. The analyze_image route validates content_type and payload size before delegating, then returns the DualVisionResponse directly — FastAPI serializes it via Pydantic.

Code snippetpython
1import asyncio
2import base64
3import instructor
4import litellm
5from fastapi import APIRouter, UploadFile, File, Form, HTTPException
6
7class VisionAnalysisService:
8    def __init__(self):
9        self.aclient = instructor.from_litellm(
10            litellm.acompletion, mode=instructor.Mode.JSON
11        )
12        self.models = {
13            "gpt-4o": "gpt-4o-2024-11-20",
14            "gemini-2.5-flash": "gemini/gemini-2.5-flash-preview-05-20",
15        }
16
17    def _build_messages(
18        self, image_b64: str, prompt: str, mime: str
19    ) -> list[dict]:
20        return [
21            {
22                "role": "user",
23                "content": [
24                    {"type": "text", "text": prompt},
25                    {
26                        "type": "image_url",
27                        "image_url": {
28                            "url": f"data:{mime};base64,{image_b64}"
29                        },
30                    },
31                ],
32            }
33        ]
34
35    async def _call_provider(
36        self, model_key: str, messages: list[dict]
37    ) -> VisionResult:
38        return await self.aclient(
39            model=self.models[model_key],
40            messages=messages,
41            response_model=VisionResult,
42            max_tokens=1024,
43            temperature=0.1,
44        )
45
46    async def analyze(
47        self, image_bytes: bytes, prompt: str, mime_type: str = "image/png"
48    ) -> DualVisionResponse:
49        image_b64 = base64.b64encode(image_bytes).decode("utf-8")
50        messages = self._build_messages(image_b64, prompt, mime_type)
51
52        tasks = [self._call_provider(key, messages) for key in self.models]
53        outcomes = await asyncio.gather(*tasks, return_exceptions=True)
54
55        results = [r for r in outcomes if isinstance(r, VisionResult)]
56        if not results:
57            raise RuntimeError("Both vision providers failed")
58
59        all_tags: set[str] = set()
60        label_counts: dict[str, int] = {}
61        for result in results:
62            all_tags.update(result.scene_tags)
63            for obj in result.objects:
64                label_counts[obj.label] = label_counts.get(obj.label, 0) + 1
65
66        consensus = [lbl for lbl, cnt in label_counts.items() if cnt > 1]
67
68        return DualVisionResponse(
69            results=results,
70            merged_tags=sorted(all_tags),
71            consensus_objects=sorted(consensus),
72        )
73
74router = APIRouter(prefix="/api/v1/vision", tags=["vision"])
75service = VisionAnalysisService()
76ALLOWED_MIMES = {"image/png", "image/jpeg", "image/webp", "image/gif"}
77
78@router.post("/analyze", response_model=DualVisionResponse)
79async def analyze_image(
80    image: UploadFile = File(..., description="Image to analyze"),
81    prompt: str = Form(
82        default="Describe this image in detail",
83        description="Analysis prompt sent to both vision models",
84    ),
85):
86    if image.content_type not in ALLOWED_MIMES:
87        raise HTTPException(
88            status_code=415,
89            detail=f"Unsupported media type: {image.content_type}",
90        )
91
92    image_bytes = await image.read()
93    if len(image_bytes) > 20 * 1024 * 1024:
94        raise HTTPException(status_code=413, detail="Image exceeds 20 MB")
95
96    return await service.analyze(
97        image_bytes, prompt, mime_type=image.content_type
98    )

• instructor.from_litellm(..., mode=Mode.JSON): Patches the LiteLLM async client once at startup so response_model=VisionResult flows down to both providers; the gemini/ model prefix tells LiteLLM to route through the Google AI Studio provider.
• _build_messages: Constructs the multi-modal chat message with a text block plus an image_url block; the data:{mime};base64,... URI is the standard inline-image encoding both OpenAI and Gemini accept through LiteLLM's normalization layer.
• Concurrent fan-out: asyncio.gather(*tasks, return_exceptions=True) runs both _call_provider coroutines in parallel; the comprehension that filters isinstance(r, VisionResult) silently drops a failed provider so the request still succeeds if at least one model responded.
• Consensus computation: An object label is consensus when it appears in more than one provider's output; sorted tags and consensus labels make the response deterministic, which matters for snapshot tests and downstream caching.
• Endpoint guardrails: The MIME allow-list set gives O(1) membership testing and rejects PDFs/videos with HTTP 415 before any model cost is incurred; the 20 MB cap returns HTTP 413 so a malicious upload can't waste bandwidth or trigger a provider-side payload error.

You'll know it works when an httpx client posting a 1 MB PNG to /api/v1/vision/analyze returns HTTP 200 with a results array of length 2, both entries having distinct provider values, and consensus_objects containing at least one label that appears in both results[i].objects lists.

Do's and Don'ts

Do's

1. ✓Do patch litellm.acompletion with instructor.from_litellm exactly once in VisionAnalysisService.__init__ — a single shared Instructor-wrapped client guarantees every _call_provider coroutine in the concurrent fan-out enforces the same VisionResult schema; re-patching per invocation creates independent wrappers that can diverge and adds serialization overhead on every request.
2. ✓Do pass return_exceptions=True to asyncio.gather when dispatching both _call_provider coroutines — with that flag, a GPT-4o 5xx or Gemini 2.5 Flash timeout lands as an exception object in the result list rather than an uncaught raise, so _merge_results can still construct a valid DualVisionResponse from the surviving provider's VisionResult, which min_length=1 on results permits.
3. ✓Do declare provider: Literal["gpt-4o", "gemini-2.5-flash"] in VisionResult — Instructor embeds the exact allowed values in the function-calling schema it sends to each model, so both providers self-identify in their structured output; without it, DualVisionResponse.results is a list of structurally identical blobs and tracing which model produced a divergent summary or a missed object requires guesswork.

Don'ts

1. ✗Don't write provider-specific deserialization branches for GPT-4o and Gemini responses — the shared VisionResult schema enforced by Instructor on both acompletion calls exists precisely so _merge_results can compute consensus_objects and merged_tags without branching on the source model; separate parsers reintroduce the coupling the unified schema was designed to eliminate.
2. ✗Don't define DualVisionResponse.results without min_length=1 — asyncio.gather(..., return_exceptions=True) ensures the fan-out never raises, but if both providers fail simultaneously the result list is empty; without the min_length=1 constraint, Pydantic validates that empty list and the endpoint silently returns a well-formed 200 with no vision data, masking a complete dual-provider outage.
3. ✗Don't omit ge=0.0, le=1.0 Field constraints on DetectedObject.confidence — vision models occasionally return raw logit values or percentage-style scores outside the unit interval; without those Pydantic bounds, an out-of-range score passes the Instructor boundary silently and corrupts any downstream threshold logic that assumes normalized confidence values when computing consensus_objects.