My messy adventures juggling multiple AI APIs (and how I tamed them)

I was building a feature that needed to ask the same question to three different AI models: GPT-4, Claude 3, and a local LLaMA instance. The goal? Cross-check answers and pick the most consistent response. Sounds simple, right?

It wasn’t.

Each API had its own request format, authentication method, rate limits, and error codes. My codebase quickly turned into a tangled mess of if-elif blocks and duplicated parsing logic. Every time I added a new model, I had to copy-paste the same boilerplate and pray I didn’t miss a subtle difference.

I was the fool who thought “I’ll just call each API directly, it’ll be fine.” Let me walk you through what I tried, why it failed, and the abstraction layer that finally let me sleep at night.

What I tried that didn’t work

My first approach was the naive one: write individual Python functions for each model.

# openai_caller.py
import openai

def ask_gpt4(prompt):
    resp = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return resp.choices[0].message.content

# claude_caller.py
import anthropic

def ask_claude(prompt):
    client = anthropic.Anthropic(api_key="...")
    resp = client.messages.create(
        model="claude-3-opus-20240229",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return resp.content[0].text

# local_llama.py
import requests

def ask_llama(prompt):
    resp = requests.post(
        "http://localhost:8080/completion",
        json={"prompt": prompt, "max_tokens": 500}
    )
    return resp.json()["choices"][0]["text"]

Then in the main orchestrator, I’d do something like:

if model == "gpt4":
    answer = ask_gpt4(prompt)
elif model == "claude":
    answer = ask_claude(prompt)
elif model == "llama":
    answer = ask_llama(prompt)

This worked for three models. When I added a fourth (Gemini), the pattern repeated. I also had to handle retries, timeouts, and logging separately for each. The code became a nightmare to maintain. Every error needed a custom wrapper. I was writing the same logic five times with slightly different keys.

Then came the hard part: I wanted to run all three models in parallel and collect their answers. But each API had a different timeout behaviour and error format. My async orchestration was a tangled mess of try-except inside asyncio.gather().

What eventually worked: a simple adapter pattern

I stepped back and asked: what is the common contract? Every model accepts a prompt string and returns an answer string (and maybe some metadata). The details of how they do that differ. So I defined a minimal interface:

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional

@dataclass
class AIResponse:
    text: str
    model_name: str
    tokens_used: Optional[int] = None
    error: Optional[str] = None

class AIModel(ABC):
    @abstractmethod
    def ask(self, prompt: str, **kwargs) -> AIResponse:
        pass

Then I implemented one adapter per model. Here’s the OpenAI version:

import openai
from openai import OpenAIError

class OpenAIModel(AIModel):
    def __init__(self, model_name: str = "gpt-4", api_key: str = None):
        self.client = openai.OpenAI(api_key=api_key or os.getenv("OPENAI_API_KEY"))
        self.model_name = model_name

    def ask(self, prompt: str, **kwargs) -> AIResponse:
        try:
            response = self.client.chat.completions.create(
                model=self.model_name,
                messages=[{"role": "user", "content": prompt}],
                **kwargs
            )
            text = response.choices[0].message.content
            tokens = response.usage.total_tokens if response.usage else None
            return AIResponse(text=text, model_name=self.model_name, tokens_used=tokens)
        except OpenAIError as e:
            return AIResponse(text="", model_name=self.model_name, error=str(e))

Similarly, I built adapters for Claude, local LLaMA, and later Gemini. The key insight: each adapter is responsible only for translation between the external API and my AIModel interface. Retry logic? I added a decorator. Logging? A wrapper. Rate limiting? A semaphore.

Now the orchestrator became clean:

import asyncio
from typing import List

async def query_all(models: List[AIModel], prompt: str) -> List[AIResponse]:
    async def ask_one(model: AIModel) -> AIResponse:
        # wrap synchronous call in executor if needed
        return await asyncio.to_thread(model.ask, prompt)

    tasks = [ask_one(m) for m in models]
    return await asyncio.gather(*tasks, return_exceptions=True)

No more if-elif. Adding a new model meant writing a new adapter class. That’s it.

Code you can copy-paste

Here’s a complete working skeleton you can adapt. I’ll include the abstract base, one concrete adapter (OpenAI), and the async orchestrator.

import os
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
import asyncio

@dataclass
class AIResponse:
    text: str
    model_name: str
    tokens_used: Optional[int] = None
    error: Optional[str] = None

class AIModel(ABC):
    @abstractmethod
    def ask(self, prompt: str, **kwargs) -> AIResponse:
        pass

# ---------------------------------------------------------------------------
# Concrete adapter for OpenAI
# ---------------------------------------------------------------------------
import openai

class OpenAIModel(AIModel):
    def __init__(self, model_name: str = "gpt-4", api_key: str = None):
        self.client = openai.OpenAI(
            api_key=api_key or os.getenv("OPENAI_API_KEY")
        )
        self.model_name = model_name

    def ask(self, prompt: str, **kwargs) -> AIResponse:
        try:
            response = self.client.chat.completions.create(
                model=self.model_name,
                messages=[{"role": "user", "content": prompt}],
                **kwargs
            )
            text = response.choices[0].message.content
            tokens = response.usage.total_tokens if response.usage else None
            return AIResponse(text=text, model_name=self.model_name, tokens_used=tokens)
        except Exception as e:
            return AIResponse(text="", model_name=self.model_name, error=str(e))

# ---------------------------------------------------------------------------
# Orchestrator
# ---------------------------------------------------------------------------

def parallel_query(models: list[AIModel], prompt: str) -> dict:
    async def _run():
        async def ask_one(model):
            return await asyncio.to_thread(model.ask, prompt)
        tasks = [ask_one(m) for m in models]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        # handle exceptions that were not caught inside adapters
        return [
            r if isinstance(r, AIResponse) else AIResponse(text="", model_name="unknown", error=str(r))
            for r in results
        ]
    return asyncio.run(_run())

# Usage:
# models = [OpenAIModel(), AnthropicModel(), LLaMAModel()]
# responses = parallel_query(models, "What is the capital of France?")
# for r in responses:
#     print(f"{r.model_name}: {r.text}")

This is the core. You can extend it with retries, caching, or logging by wrapping the adapter or using a decorator. I keep the adapter thin and handle cross-cutting concerns separately.

Lessons learned and trade-offs

What I gained:

• New models become cheap to add – I wrote a MockModel for testing in five minutes.
• Error handling is uniform: every adapter returns an AIResponse with an error field. No more try-except in the orchestrator.
• I can test the orchestrator with fake models without hitting real APIs.

What I lost:

• Some model-specific features are hard to abstract cleanly. For example, OpenAI supports functions/tools; Claude has a different streaming interface. If you need those, the adapter can’t hide all differences. I ended up adding optional supports_functions flag and a separate ask_with_tools() method in a subclass.
• Performance overhead: calling asyncio.to_thread for each synchronous call adds a tiny cost, but it’s negligible compared to network latency.
• Abstraction can leak: if you rely too much on a uniform interface, you might miss optimizations like batching (which some APIs support and others don’t).

When NOT to use this:

• If you only ever use one model, this is overkill. Just call the SDK directly.
• If your models are all OpenAI variants (gpt-3.5, gpt-4) the differences are small; a config dict might be simpler.
• If you need real-time streaming, this synchronous adapter pattern won’t work well. You’d need async streaming interfaces per model.

What I’d do differently next time

I’d start with the interface from day one, even for a single model. It costs almost nothing to define an ABC at the beginning. Also, I’d write integration tests that run against actual APIs, not just mocks – I caught a few subtle response format changes only when testing against the real endpoints.

Another thing: I initially put too much logic (like retry with exponential backoff) inside each adapter. That led to duplicated code when I wanted different retry strategies per model. Instead, I now use a generic RetryingAIModel wrapper that composes any AIModel. Decorator pattern wins again.

Final thoughts

This abstraction isn’t revolutionary – it’s textbook adapter pattern. But in the rush to ship AI features, I often see developers skip the design and pay for it later. A little upfront planning makes your codebase resilient to the chaos of the AI ecosystem.

What about you? How do you manage multiple AI providers in your projects? Do you use a library, a simple function, or something else? I’d love to hear what works (or doesn’t) for you.