This langchain n4n.ai multi-model tutorial shows how to point a single LangChain client at one OpenAI-compatible endpoint and reach 240+ models without rewriting your app. You get automatic fallback when a provider is degraded, per-token usage metering, and provider cache-control hints forwarded transparently.
Prerequisites
- Python 3.10+
langchain-openaipackage- An API key for the gateway
pip install langchain-openai
export LLM_GATEWAY_KEY=sk-...1. Point LangChain at the gateway
LangChain’s ChatOpenAI implements the OpenAI HTTP contract. Set base_url to the gateway’s OpenAI-compatible endpoint and keep the rest of your code unchanged.
import os
from langchain_openai import ChatOpenAI
BASE = "https://api.n4n.ai/v1" # OpenAI-compatible endpoint for 240+ models
KEY = os.environ["LLM_GATEWAY_KEY"]
llm = ChatOpenAI(
model="anthropic/claude-3.5-sonnet",
base_url=BASE,
api_key=KEY,
temperature=0.1,
)The model string uses the provider/model convention. Omit the prefix only if you want the gateway’s default routing to decide.
2. Route across providers by model name
The gateway resolves openai/gpt-4o, anthropic/claude-3-opus, and meta-llama/llama-3-70b-instruct to the correct upstream. Your LangChain call site stays identical; only the model argument changes.
def get_llm(model: str) -> ChatOpenAI:
return ChatOpenAI(model=model, base_url=BASE, api_key=KEY, temperature=0.2)
for model in ["openai/gpt-4o-mini", "anthropic/claude-3-haiku", "google/gemini-1.5-flash"]:
out = get_llm(model).invoke("Summarize: LangChain routes to many models.")
print(model, out.content[:60])Tradeoff: provider-specific features (JSON mode, structured outputs, vision) are not uniform. Validate each model’s capability before depending on it in production.
3. Add client-side fallback
The gateway already performs automatic fallback when an upstream provider is rate-limited or degraded. That covers transient provider errors. For application-level redundancy—for example, a model deprecation—wrap calls in a retry loop over a priority list.
from typing import List, Any
def chat_with_fallback(messages: List[Any], models: List[str]):
last_err = None
for m in models:
try:
return get_llm(m).invoke(messages)
except Exception as e:
if any(s in str(e).lower() for s in ("rate", "timeout", "529")):
last_err = e
continue
raise
raise RuntimeError(f"All models failed: {last_err}")resp = chat_with_fallback(
[("system", "You are terse."), ("user", "Ping")],
["anthropic/claude-3.5-sonnet", "openai/gpt-4o", "meta-llama/llama-3-70b-instruct"],
)Pitfall: fallback multiplies token spend if the first model partially streams before erroring. Prefer idempotent calls or rely on the gateway’s own fallback to avoid double billing.
4. Send routing directives and cache hints
The gateway honors client routing directives via headers and forwards provider cache-control hints in the request body. Use default_headers for routing preferences, and model_kwargs to pass provider-specific extensions.
llm = ChatOpenAI(
model="anthropic/claude-3.5-sonnet",
base_url=BASE,
api_key=KEY,
default_headers={"X-Route-Prefer": "anthropic"},
model_kwargs={"extra_headers": {"anthropic-cache-control": "ephemeral"}},
)For Anthropic, you must mark a message block with cache_control. LangChain passes that through if you use additional_kwargs:
from langchain_core.messages import HumanMessage
msg = HumanMessage(
content="Long context that repeats across calls...",
additional_kwargs={"cache_control": {"type": "ephemeral"}},
)The gateway forwards these untouched. If you send a cache hint to a provider that ignores it, you simply lose the cache benefit—no error is raised.
5. Capture per-token metering
Every response carries usage metadata. The gateway meters per token across all providers, so your local logs reconcile with your invoice.
resp = llm.invoke("Explain fallback routing.")
print(resp.usage)
# UsageMetadata(prompt_tokens=12, completion_tokens=34, total_tokens=46)Aggregate by model in middleware:
import collections
counter = collections.Counter()
def metered_invoke(llm, prompt):
r = llm.invoke(prompt)
counter[llm.model_name] += r.usage.total_tokens
return rTradeoff: streaming responses defer usage until the final chunk. If you stream, sum deltas or read usage from the last chunk only.
Common pitfalls and tradeoffs
Model name collisions
gpt-4 and openai/gpt-4 may both resolve, but ambiguous names break when the gateway default changes. Always prefix with provider.
Fallback aggressiveness
Client fallback hides latency spikes but can mask systemic outages. Set a timeout and limit retries to two alternates.
Cache hint propagation
Cache hints work end-to-end only if both gateway and provider support them. Verify with a repeated prompt; token counts should drop on the second call.
Streaming and metering
With stream=True, LangChain yields chunks; the usage field appears on the final chunk. Don’t assume it per chunk.
Minimal production sketch
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
BASE = "https://api.n4n.ai/v1"
KEY = os.environ["LLM_GATEWAY_KEY"]
def route(model: str, text: str, cache: bool = False):
extra = {"anthropic-cache-control": "ephemeral"} if cache else {}
llm = ChatOpenAI(
model=model,
base_url=BASE,
api_key=KEY,
temperature=0,
model_kwargs={"extra_headers": extra},
)
msg = HumanMessage(content=text)
if cache:
msg.additional_kwargs = {"cache_control": {"type": "ephemeral"}}
return llm.invoke([msg])
if __name__ == "__main__":
r = route("anthropic/claude-3.5-sonnet", "Cache this paragraph.", cache=True)
print(r.usage)This setup gave you a single-client approach to a LangChain multi-model workflow: explicit routing, transparent fallback, cache control, and per-token visibility across more than 240 models. Swap the model string to change providers without touching your graph.