Implement an embedding abstraction layer with provider switching

Introduction

When you hardcode a single embedding provider's SDK directly into your pipeline, the next Voyage outage takes your retrieval system down with it, a Matryoshka-dimension experiment requires touching every call site, and the next API revision turns into a multi-day migration. Teams that lack an abstraction layer end up shipping vendor lock-in alongside their RAG features, and pay for it during every incident. By the end of this lesson you'll be able to define a unified embedding interface, implement provider adapters that normalize incompatible SDK shapes behind it, and compose them into a fallback chain so a single provider failure becomes a transparent retry rather than a dropped request.

Key Terminology

• EmbeddingProvider — the abstract base class that defines the contract every provider adapter must satisfy (an embed() method plus capability discovery), so pipeline code in this lesson can call any provider without conditional logic.
• EmbeddingResponse — the standardized dataclass each adapter returns (embeddings, model, dimensions, token count, latency); a uniform response shape is what lets the router swap providers without changing downstream code.
• Provider adapter — a concrete subclass of EmbeddingProvider (e.g. OpenAIAdapter, VoyageAdapter below) that wraps one vendor SDK, translating its parameter names and response shape into the unified contract.
• EmbeddingRouter — the composition object holding a primary provider plus an ordered fallback list; it walks them in sequence on failure, turning a single point of failure into a degraded-but-available path.

Concepts

Embedding APIs across providers are gratuitously incompatible: OpenAI accepts input= and exposes response.data[*].embedding with usage.total_tokens, Voyage takes positional texts plus an input_type argument and returns result.embeddings with result.total_tokens, and dimension-truncation support is opt-in only on Matryoshka-capable models. Hard-coding any one of these shapes into your pipeline couples it to a vendor and a current API revision — both will change.

The abstraction layer inverts that dependency in three layers. First, a dataclass response (EmbeddingResponse) and an abstract base class (EmbeddingProvider) define a contract — embeddings, model, dimensions, token count, latency — that pipeline code consumes. Second, one adapter per provider translates between that contract and the vendor SDK, measuring latency at the boundary so observability is uniform. Third, a router composes adapters into a fallback chain so a Voyage outage routes to OpenAI without manual intervention or dropped requests. The result is that switching providers, running A/B tests across models, or absorbing a provider incident becomes a configuration change rather than a code change.

Code Walkthrough

Defining the Unified Interface

The unified interface defines a contract that all provider adapters must implement, ensuring consistent behavior regardless of which embedding provider handles a given request. The EmbeddingResponse dataclass below standardizes the return format with embeddings, model identifier, dimensions, token count, and latency, while the EmbeddingProvider abstract base class requires three methods: embed() for generating embeddings, max_dimensions() for capability discovery, and provider_name() for logging. This abstraction lets your pipeline code work with any provider without conditional logic.

1from abc import ABC, abstractmethod
2from dataclasses import dataclass
3from typing import Optional
4
5@dataclass
6class EmbeddingResponse:
7    embeddings: list[list[float]]
8    model: str
9    dimensions: int
10    token_count: int
11    latency_ms: float
12
13class EmbeddingProvider(ABC):
14    @abstractmethod
15    def embed(
16        self, texts: list[str], dimensions: Optional[int] = None
17    ) -> EmbeddingResponse:
18        pass
19
20    @abstractmethod
21    def max_dimensions(self) -> int:
22        pass
23
24    @abstractmethod
25    def provider_name(self) -> str:
26        pass

Implementing Provider Adapters and the Fallback Router

Each provider adapter encapsulates the provider-specific client library, API calling conventions, and response parsing behind the standard EmbeddingProvider interface. The OpenAIAdapter below wraps OpenAI's embeddings API with optional Matryoshka dimension support and normalized response mapping, while the VoyageAdapter handles Voyage's different client library and input_type conventions. Both adapters measure latency internally and extract token counts from provider-specific response objects, producing identical EmbeddingResponse outputs regardless of the underlying provider. The EmbeddingRouter at the bottom composes those adapters into a fallback chain, attempting each provider in sequence until one succeeds and raising a RuntimeError with the last error if all of them fail — so your pipeline survives transient API errors, sustained outages, and rate-limit exhaustion on any single provider.

Code snippetpython
1import time
2from openai import OpenAI
3
4class OpenAIAdapter(EmbeddingProvider):
5    def __init__(self, model: str = "text-embedding-3-large"):
6        self.client = OpenAI()
7        self.model = model
8
9    def embed(self, texts: list[str], dimensions: Optional[int] = None) -> EmbeddingResponse:
10        start = time.perf_counter()
11        kwargs = {"model": self.model, "input": texts}
12        if dimensions:
13            kwargs["dimensions"] = dimensions
14        response = self.client.embeddings.create(**kwargs)
15        elapsed = (time.perf_counter() - start) * 1000
16
17        return EmbeddingResponse(
18            embeddings=[item.embedding for item in response.data],
19            model=self.model,
20            dimensions=len(response.data[0].embedding),
21            token_count=response.usage.total_tokens,
22            latency_ms=elapsed,
23        )
24
25    def max_dimensions(self) -> int:
26        return 3072
27
28    def provider_name(self) -> str:
29        return "openai"
30
31class VoyageAdapter(EmbeddingProvider):
32    def __init__(self, model: str = "voyage-4"):
33        import voyageai
34        self.client = voyageai.Client()
35        self.model = model
36
37    def embed(self, texts: list[str], dimensions: Optional[int] = None) -> EmbeddingResponse:
38        start = time.perf_counter()
39        result = self.client.embed(texts, model=self.model, input_type="document")
40        elapsed = (time.perf_counter() - start) * 1000
41        return EmbeddingResponse(
42            embeddings=result.embeddings,
43            model=self.model,
44            dimensions=len(result.embeddings[0]),
45            token_count=result.total_tokens,
46            latency_ms=elapsed,
47        )
48
49    def max_dimensions(self) -> int:
50        return 2048
51
52    def provider_name(self) -> str:
53        return "voyage"
54
55class EmbeddingRouter:
56    def __init__(self, primary: EmbeddingProvider, fallbacks: list[EmbeddingProvider]):
57        self.primary = primary
58        self.fallbacks = fallbacks
59
60    def embed(self, texts: list[str], dimensions: Optional[int] = None) -> EmbeddingResponse:
61        providers = [self.primary] + self.fallbacks
62        last_error = None
63
64        for provider in providers:
65            try:
66                return provider.embed(texts, dimensions)
67            except Exception as e:
68                last_error = e
69                continue
70
71        raise RuntimeError(f"All providers failed. Last error: {last_error}")
72
73router = EmbeddingRouter(
74    primary=VoyageAdapter(),
75    fallbacks=[OpenAIAdapter()],
76)

• OpenAIAdapter: passes the optional dimensions parameter for Matryoshka truncation; the response is normalized into the standard EmbeddingResponse format.
• VoyageAdapter: uses a different client library but produces the same response shape — provider-specific details (input_type, token counting) stay encapsulated inside the adapter.
• EmbeddingRouter: tries the primary provider first, then each fallback in order, handling both transient API errors and sustained outages without manual intervention. Configure with Voyage as primary (best retrieval quality) and OpenAI as fallback (broad availability, reasonable quality).

This abstraction layer decouples your pipeline from any single embedding provider, enabling seamless switching, A/B testing, and resilient production operation.

Do's and Don'ts

Now that you have worked through the implementation, the practices below separate a durable approach from a fragile one.

Do's

1. ✓Keep adapters thin — translate inputs and outputs only; do not bake in retry policy, caching, or business logic that belongs in the router or upstream callers.
2. ✓Measure latency inside each adapter at the SDK call boundary so EmbeddingResponse.latency_ms is consistent across providers and usable for routing decisions.
3. ✓Verify dimension support before forwarding the dimensions argument — OpenAI's text-embedding-3-* accepts it via Matryoshka truncation, but most other providers will error on an unknown parameter.

Don'ts

1. ✗Don't leak provider-specific types (e.g. an OpenAI CreateEmbeddingResponse) through the interface; callers should only see EmbeddingResponse, otherwise the abstraction buys nothing.
2. ✗Don't swallow exceptions inside an adapter to "make fallback simpler" — let the router catch and decide; an adapter that silently returns empty embeddings on failure corrupts downstream indexes.
3. ✗Don't order the fallback chain by cost alone; order by retrieval-quality compatibility first, since a chain that silently degrades from a 2048-dim Voyage embedding to a 768-dim fallback breaks any index sized to the primary.