LangChain LCEL chains with multi-provider model routing

Building langchain lcel multi-provider routing on top of a single OpenAI-compatible endpoint removes the headache of juggling separate SDKs for each model vendor. You get one ChatOpenAI client, a unified message format, and the ability to switch models inside a chain based on cost, latency, or task complexity.

Step 1: Install dependencies and configure the gateway client

Install the LangChain OpenAI integration and core runnables:

pip install langchain-openai langchain-core

Set environment variables for your gateway key and base URL. If you use a gateway such as n4n.ai, one OpenAI-compatible endpoint fronts 240+ models and automatically fails over when a provider is rate-limited or degraded. That matters because your LCEL code never needs to know whether anthropic/claude-3-5-sonnet is served directly or via a proxy.

import os

os.environ["OPENAI_API_KEY"] = "sk-gateway-xxx"  # your gateway key
BASE_URL = "https://api.n4n.ai/v1"  # OpenAI-compatible gateway base

from langchain_openai import ChatOpenAI

def make_client(model: str) -> ChatOpenAI:
    return ChatOpenAI(
        model=model,
        base_url=BASE_URL,
        api_key=os.environ["OPENAI_API_KEY"],
        temperature=0.2,
        max_retries=2,
    )

Keep BASE_URL consistent across all clients. The gateway translates the model field into the correct upstream provider call, normalizes auth, and returns OpenAI-shaped response objects. This is what makes langchain lcel multi-provider routing feasible without custom LLM subclasses.

Step 2: Define a routing function for langchain lcel multi-provider routing

Routing logic belongs in a plain Python function wrapped as a RunnableLambda. Inspect the input and return a model identifier your gateway understands. Use provider-prefixed names to avoid ambiguity.

from langchain_core.runnables import RunnableLambda

def select_model(inp: dict) -> str:
    q = inp["question"]
    # Short, low-stakes queries go to a cheap model.
    if len(q) < 60 and "code" not in q.lower():
        return "openai/gpt-4o-mini"
    # Long or code-related prompts use a stronger model.
    if "code" in q.lower() or len(q) > 200:
        return "anthropic/claude-3-5-sonnet"
    # Default mid-tier
    return "meta-llama/llama-3-70b-instruct"

router = RunnableLambda(select_model)

The heuristic above is deliberately dumb. In production you might call a tiny classifier first, but the LCEL shape stays identical: a runnable that emits a string. This is the core of langchain lcel multi-provider routing—a deterministic selector that maps request shape to a model string, decoupling routing from execution.

Step 3: Pre-instantiate model clients and pick at runtime

Creating a ChatOpenAI instance per request is wasteful and can leak connections. Build a small registry keyed by model name at startup.

MODEL_REGISTRY = {
    "openai/gpt-4o-mini": make_client("openai/gpt-4o-mini"),
    "anthropic/claude-3-5-sonnet": make_client("anthropic/claude-3-5-sonnet"),
    "meta-llama/llama-3-70b-instruct": make_client("meta-llama/llama-3-70b-instruct"),
}

def pick_client(model_name: str) -> ChatOpenAI:
    return MODEL_REGISTRY[model_name]

model_picker = router | RunnableLambda(pick_client)

Compose the picker with the prompt using RunnableAssign so the downstream step sees both the question and the selected client.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableAssign

prompt = ChatPromptTemplate.from_messages([("user", "{question}")])

chain = (
    RunnableAssign({"client": model_picker})
    | RunnableLambda(
        lambda d: d["client"].invoke(
            prompt.format_messages(question=d["question"])
        )
    )
)

The registry pattern is thread-safe for sync invocation because ChatOpenAI wraps a shared HTTP client. If you need async, the same registry works with .ainvoke.

Step 4: Add explicit fallbacks for client-side resilience

The gateway already handles provider-side degradation, but you should still guard against malformed model names or unexpected upstream errors. Wrap the invocation in RunnableWithFallbacks.

from langchain_core.runnables import RunnableWithFallbacks

fallback_client = make_client("openai/gpt-4o-mini")

safe_chain = chain.with_fallbacks(
    [RunnableLambda(lambda d: fallback_client.invoke(
        prompt.format_messages(question=d["question"])
    ))]
)

Now if the routed model raises, the chain retries on the cheap fallback model. This complements gateway-level automatic fallback with a predictable local behavior. You can stack multiple fallbacks by passing more runnables; LangChain tries them in order.

Step 5: Run the chain and verify routing, usage, and cache hints

Invoke with a few distinct inputs to confirm the router picks different models.

queries = [
    {"question": "What's the capital of France?"},
    {"question": "Write a Python function to parse ISO timestamps with timezone support and explain it."},
]

for q in queries:
    resp = safe_chain.invoke(q)
    print(resp.content)
    print("Usage:", resp.usage_metadata)

Verification checklist:

• The short query should route to openai/gpt-4o-mini (assert by calling select_model(q) in a unit test).
• The code query should route to anthropic/claude-3-5-sonnet.
• resp.usage_metadata shows input_tokens and output_tokens — the gateway meters per-token usage even across providers.
• No exception is raised; if you temporarily pass an unknown model, the fallback returns a valid response.

If your gateway honors client routing directives and forwards provider cache-control hints, you can pass cache settings via default_headers on the client:

cached_client = ChatOpenAI(
    model="anthropic/claude-3-5-sonnet",
    base_url=BASE_URL,
    api_key=os.environ["OPENAI_API_KEY"],
    default_headers={"x-cache-control": "max-age=3600"},
)

This only works if the gateway propagates the hint to the upstream provider; verify against your gateway docs. For streaming, swap .invoke for .stream and iterate resp.content chunks—the same routing and fallback logic applies.

Step 6: Production notes for langchain lcel multi-provider routing

In production, externalize the routing table to configuration so you can shift traffic without code changes. A simple JSON file works:

{
  "routes": {
    "short": "openai/gpt-4o-mini",
    "code": "anthropic/claude-3-5-sonnet",
    "default": "meta-llama/llama-3-70b-instruct"
  },
  "thresholds": {"short_max_len": 60}
}

Load it at startup and feed the values into select_model. Because the LCEL chain is just runnables, you can also branch on streaming vs non-streaming by swapping the invoked method (stream instead of invoke).

Keep the gateway base URL and key in secret storage. The per-token metering from the gateway gives you an auditable cost breakdown per route, which you can aggregate by model name in your observability stack. If you run multiple regions, pin the gateway URL per deployment.

Langchain lcel multi-provider routing is not magic: it is a function that returns a string and a registry of clients. The win is that the rest of your LCEL pipeline—prompts, output parsers, retries—stays identical regardless of which vendor serves the token. When you add a new provider, you add one line to the registry and one route key; no prompt changes required.

Test the chain in CI with mocked clients to assert the router picks the expected model for representative inputs. That catches regressions when you change thresholds or introduce a new model class. Once verified, the pattern scales to dozens of models behind one endpoint without touching chain composition.