This langchain multi-provider setup tutorial walks through standing up a single LangChain client abstraction that can call OpenAI, Anthropic, and n4n.ai without duplicating orchestration logic. You get copy-paste config, a fallback router, and the sharp edges we’ve hit shipping this in production.
Why a single client matters
Scattering ChatOpenAI and ChatAnthropic calls across your codebase forces you to reimplement retries, logging, and token accounting everywhere. A unified client centralizes those concerns. You also gain the ability to shift traffic when one provider throws 429s, without touching business logic.
The tradeoff is a thin indirection layer you must maintain. Done right, it’s a few dozen lines. Done wrong, it becomes a leaky abstraction that hides provider quirks until 2 a.m.
Prerequisites
- Python 3.10+
langchain-openai,langchain-anthropic,httpx- API keys for each provider (or just one if you route everything through the gateway)
pip install langchain-openai langchain-anthropic httpx
Export keys:
export OPENAI_API_KEY=sk-...
export ANTHROPIC_API_KEY=sk-ant-...
export N4N_API_KEY=sk-n4n-...
Step 1: Define provider configurations
Keep base URLs and model defaults in one dict. This makes rotating models or endpoints a one-line change and keeps environment-specific overrides out of your call sites.
import os
CONFIG = {
"openai": {
"api_key": os.environ["OPENAI_API_KEY"],
"base_url": None, # default OpenAI endpoint
"default_model": "gpt-4o-mini",
},
"anthropic": {
"api_key": os.environ["ANTHROPIC_API_KEY"],
"default_model": "claude-3-5-sonnet-20240620",
},
"n4n": {
"api_key": os.environ["N4N_API_KEY"],
"base_url": "https://api.n4n.ai/v1",
"default_model": "openai/gpt-4o",
},
}
Environment-based overrides
In staging you may want to force every call through the gateway to cap cost. Patch the dict at load time:
if os.environ.get("ENV") == "staging":
CONFIG["openai"]["base_url"] = CONFIG["n4n"]["base_url"]
CONFIG["openai"]["api_key"] = CONFIG["n4n"]["api_key"]
Step 2: Build a unified wrapper
LangChain’s ChatOpenAI and ChatAnthropic expose the same invoke/stream surface, but they are different classes. Wrap them behind a method that selects on model name prefix. Use with_config to swap models per call instead of instantiating new clients.
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
class UnifiedChat:
def __init__(self, config):
self.openai = ChatOpenAI(
api_key=config["openai"]["api_key"],
base_url=config["openai"]["base_url"],
model=config["openai"]["default_model"],
)
self.anthropic = ChatAnthropic(
api_key=config["anthropic"]["api_key"],
model=config["anthropic"]["default_model"],
)
self.n4n = ChatOpenAI(
api_key=config["n4n"]["api_key"],
base_url=config["n4n"]["base_url"],
model=config["n4n"]["default_model"],
)
def pick(self, model: str | None):
if model and model.startswith("claude"):
return self.anthropic.with_config({"model": model})
if model and "/" in model:
# gateway namespacing: "openai/...", "anthropic/...", "n4n/..."
return self.n4n.with_config({"model": model.split("/", 1)[-1]})
return self.openai
The pick method returns a runnable bound to a specific model. You avoid socket churn and keep the hot path cheap.
Step 3: Add fallback and routing
A simple ordered retry covers most degradation scenarios. Try the primary, then the gateway, then the other native provider. Because n4n.ai performs automatic fallback when a backing provider is rate-limited or degraded, using it as the middle slot often recovers the request without your code handling provider-specific errors.
import logging
log = logging.getLogger(__name__)
def chat_with_fallback(uc: UnifiedChat, messages, primary="openai"):
order = [primary, "n4n", "anthropic"] if primary != "n4n" else ["n4n", "openai"]
last_err = None
for slot in order:
try:
if slot == "n4n":
client = uc.pick("n4n/openai/gpt-4o")
else:
client = uc.pick(slot if slot == "anthropic" else "openai/gpt-4o-mini")
return client.invoke(messages)
except Exception as e:
log.warning("provider %s failed: %s", slot, e)
last_err = e
raise RuntimeError("all providers unavailable") from last_err
For production, replace the bare except Exception with typed catches and add exponential backoff. The skeleton above is intentional.
Step 4: Collapse providers with the OpenAI-compatible gateway
If you point ChatOpenAI at n4n.ai’s endpoint, you can address 240+ models—including Anthropic’s—with one client class. This removes the ChatAnthropic dependency entirely for many teams.
gateway_client = ChatOpenAI(
api_key=CONFIG["n4n"]["api_key"],
base_url=CONFIG["n4n"]["base_url"],
model="anthropic/claude-3-5-sonnet-20240620",
temperature=0.1,
)
resp = gateway_client.invoke([{"role": "user", "content": "Summarize this log"}])
The gateway honors client routing directives and forwards provider cache-control hints, so prompt caching works as expected. You still get per-token usage metering in the response headers, which simplifies cost attribution when you mix models.
Step 5: Streaming and tool calls
Streaming works uniformly through the stream method, but chunk shapes differ slightly between native Anthropic and OpenAI-compatible routes. LangChain normalizes content, but tool call deltas need care.
for chunk in gateway_client.stream(messages):
if chunk.tool_call_chunks:
print("tool delta:", chunk.tool_call_chunks)
else:
print(chunk.content, end="")
When using native ChatAnthropic, set stream_usage=True if you need token counts mid-stream. The gateway returns usage only at the end of the stream unless you pass the same hint. Test tool schemas against each target model—Anthropic rejects certain JSON schema constructs that OpenAI tolerates.
Cache-control hints
To use prompt caching on Anthropic via the gateway, pass the standard Anthropic cache breakpoint in your message metadata. LangChain forwards it; the gateway passes it through.
messages = [{
"role": "system",
"content": "Long static context...",
"metadata": {"anthropic": {"cache_control": {"type": "ephemeral"}}}
}]
Common pitfalls and tradeoffs
Model name drift. OpenAI uses gpt-4o, Anthropic uses claude-3-5-sonnet-20240620, and the gateway may namespace them as openai/gpt-4o. Keep a mapping table; don’t hardcode strings at call sites.
Latency. Routing through a gateway adds a proxy hop. Measure p99 before committing. For latency-sensitive paths, call native endpoints directly and use the gateway only for long-tail fallback.
Error taxonomy. ChatAnthropic raises AnthropicAPIStatusError; ChatOpenAI raises APIStatusError. Catch the base Exception in the fallback loop, but log the type to tune retries.
Token accounting. Native providers return usage in different fields. If you use the gateway, per-token metering is consistent across all models, which simplifies billing code.
Tool schema divergence. Anthropic accepts only certain JSON schema subsets. LangChain converts, but complex nested unions can break on one side. Test tool calls against each target model.
Timeout tuning. Default httpx timeout in LangChain is 60s. For long completions, raise it on the client constructor: ChatOpenAI(timeout=120).
Observability
Wrap invoke with a decorator that emits span attributes: provider used, model, token estimate, latency. If you use the gateway, record the x-n4n-model response header to know which backing model actually served the request after its internal fallback.
def traced(uc, messages, **kw):
import time
t0 = time.monotonic()
out = uc.pick(kw.get("model")).invoke(messages)
log.info("llm latency=%.2f model=%s", time.monotonic()-t0, kw.get("model"))
return out
Testing your setup
Mock the HTTP layer with httpx.MockTransport or LangChain’s FakeListChatModel for unit tests. For integration, run a nightly job that hits each provider with a tiny prompt and asserts response shape.
def test_fallback(monkeypatch):
uc = UnifiedChat(CONFIG)
monkeypatch.setattr(uc.openai, "invoke",
lambda *a, **k: (_ for _ in ()).throw(ValueError("mock")))
out = chat_with_fallback(uc, [{"role": "user", "content": "hi"}], primary="openai")
assert out is not None
The langchain multi-provider setup tutorial above gives you a working skeleton. Extend the UnifiedChat.pick method to support weight-based routing or cost-aware model selection once the basics are stable.
When to skip the abstraction
If you only use one model from one provider, this pattern is overhead. Adopt it when you have at least two providers or need compliance routing. The moment you start writing if provider == "anthropic" in three files, centralize.
We’ve run this design under sustained traffic and the maintenance cost is low. The biggest win is deleting provider-specific retry code from eight services and letting the gateway absorb provider degradation. This langchain multi-provider setup tutorial is the distillation of that cleanup.