Load balance LLM calls in LangChain with n4n.ai

If you’re building production LLM applications, you’ve hit the reliability wall: a single provider goes down, rate limits throttle your traffic, or a model degrades on specific tasks. This langchain load balancing llm tutorial shows you how to route requests across multiple models with automatic fallback, using n4n.ai as a unified OpenAI-compatible endpoint that handles 240+ models behind one API.

Step 1: Set up your environment

Install the dependencies you need. We’ll use LangChain’s OpenAI integration since n4n.ai speaks the OpenAI API spec.

pip install langchain-openai langchain-core python-dotenv

Create a .env file with your n4n.ai credentials:

N4N_API_KEY=your-api-key-here
N4N_BASE_URL=https://api.n4n.ai/v1

The base URL is the only configuration change from a standard OpenAI setup. Everything else — chat completions, embeddings, streaming, function calling — works identically.

Step 2: Initialize the LangChain chat model

Create a minimal wrapper that points at the n4n.ai endpoint. This single client gives you access to every model n4n.ai routes to — GPT-4o, Claude 3.5 Sonnet, Llama 3.1, Gemini, and dozens more — without changing your code.

import os
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(
    model="gpt-4o-mini",  # logical model name; n4n.ai resolves to a provider
    api_key=os.getenv("N4N_API_KEY"),
    base_url=os.getenv("N4N_BASE_URL"),
    temperature=0.2,
    max_tokens=1024,
)

Verify it works:

response = llm.invoke("Reply with exactly: OK")
print(response.content)  # Should print: OK

Step 3: Configure model fallback with routing hints

n4n.ai honors client-side routing directives via extra headers. You can specify an ordered list of models to try, and the gateway will automatically fall back if the primary is rate-limited, degraded, or returns an error.

from langchain_openai import ChatOpenAI
import os

llm_with_fallback = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    base_url=os.getenv("N4N_BASE_URL"),
    temperature=0.2,
    max_tokens=1024,
    default_headers={
        "x-n4n-routing": '["gpt-4o-mini", "claude-3-5-sonnet", "llama-3.1-70b"]'
    },
)

The x-n4n-routing header accepts a JSON array of model identifiers. The gateway tries each in order until one succeeds. This is your load balancing primitive — no custom retry logic, no circuit breaker library, no provider-specific SDKs.

Test fallback behavior:

# Force a fallback by requesting a model that doesn't exist
# n4n.ai will skip to the next model in your routing list
llm_test = ChatOpenAI(
    model="nonexistent-model",
    api_key=os.getenv("N4N_API_KEY"),
    base_url=os.getenv("N4N_BASE_URL"),
    default_headers={
        "x-n4n-routing": '["nonexistent-model", "gpt-4o-mini"]'
    },
)

response = llm_test.invoke("Say hello")
print(response.content)  # Returns gpt-4o-mini's response

Step 4: Implement weighted load balancing across models

For true load distribution — not just fallback — specify weights in your routing directive. This splits traffic across healthy providers, reducing blast radius when one degrades.

import json

# 50% gpt-4o-mini, 30% claude-3-5-sonnet, 20% llama-3.1-70b
routing_config = {
    "strategy": "weighted",
    "models": [
        {"model": "gpt-4o-mini", "weight": 50},
        {"model": "claude-3-5-sonnet", "weight": 30},
        {"model": "llama-3.1-70b", "weight": 20},
    ],
}

llm_balanced = ChatOpenAI(
    model="gpt-4o-mini",  # default; overridden by routing header
    api_key=os.getenv("N4N_API_KEY"),
    base_url=os.getenv("N4N_BASE_URL"),
    temperature=0.2,
    max_tokens=1024,
    default_headers={
        "x-n4n-routing": json.dumps(routing_config)
    },
)

Run a quick distribution test:

from collections import Counter

results = Counter()
for _ in range(100):
    resp = llm_balanced.invoke("Reply with the model name only")
    # n4n.ai returns the actual model used in response headers
    # Access via response.response_metadata if available
    results[resp.response_metadata.get("model_name", "unknown")] += 1

print(results)
# Expected: roughly 50/30/20 split across the three models

Step 5: Handle streaming with fallback

Streaming works transparently across fallbacks. If the primary model fails mid-stream, n4n.ai switches to the next model and continues the stream — your code doesn’t change.

from langchain_core.messages import HumanMessage

for chunk in llm_with_fallback.stream([HumanMessage(content="Write a haiku about load balancing")]):
    print(chunk.content, end="", flush=True)
print()

Verify streaming fallback: Temporarily add a failing model to the front of your routing list and confirm the stream completes without raising an exception.

llm_stream_test = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    base_url=os.getenv("N4N_BASE_URL"),
    default_headers={
        "x-n4n-routing": '["invalid-model", "gpt-4o-mini"]'
    },
)

# This should stream successfully from gpt-4o-mini after the first model fails
for chunk in llm_stream_test.stream("Count to five"):
    print(chunk.content, end="")

Step 6: Add per-request routing overrides

Different tasks need different models. Code generation might route to Claude; classification to a smaller, faster model. Override routing per-call without creating new client instances.

def route_to_model(llm: ChatOpenAI, model_preference: list[str], prompt: str):
    """Invoke with a one-off routing directive."""
    return llm.invoke(
        prompt,
        headers={"x-n4n-routing": json.dumps(model_preference)}
    )

# Use a coding-optimized model
code_result = route_to_model(
    llm,
    ["claude-3-5-sonnet", "gpt-4o", "deepseek-coder"],
    "Write a Python function that validates email addresses"
)

# Use a fast, cheap model for classification
classify_result = route_to_model(
    llm,
    ["gpt-4o-mini", "llama-3.1-8b", "gemini-1.5-flash"],
    "Classify: 'I love this product!' as positive/negative/neutral"
)

This pattern lets you build a routing policy layer in your application — map task types to model preferences in a config file, then apply the right header at call time.

Step 7: Observe usage and costs with response metadata

n4n.ai returns per-token usage and the actual model served in response metadata. Hook into LangChain’s callbacks or inspect response_metadata for logging, cost tracking, and alerting.

from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict

class UsageLogger(BaseCallbackHandler):
    def on_llm_end(self, response: Any, **kwargs: Any) -> None:
        metadata = response.llm_output or {}
        usage = metadata.get("token_usage", {})
        model = response.generations[0][0].generation_info.get("model_name", "unknown")
        
        print(f"Model: {model}")
        print(f"  Prompt tokens: {usage.get('prompt_tokens', 0)}")
        print(f"  Completion tokens: {usage.get('completion_tokens', 0)}")
        print(f"  Total tokens: {usage.get('total_tokens', 0)}")

llm_observed = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    base_url=os.getenv("N4N_BASE_URL"),
    callbacks=[UsageLogger()],
    default_headers={
        "x-n4n-routing": '["gpt-4o-mini", "claude-3-5-sonnet"]'
    },
)

llm_observed.invoke("Explain load balancing in one sentence")

Sample output:

Model: gpt-4o-mini
  Prompt tokens: 12
  Completion tokens: 18
  Total tokens: 30

If a fallback occurs, model_name reflects the model that actually served the request. Your observability pipeline sees the truth without extra instrumentation.

Step 8: Build a resilient chain with automatic retries

Combine LangChain’s RunnableWithFallbacks with n4n.ai’s gateway-level fallback for defense in depth. The gateway handles provider failures; the runnable handles transient network errors.

from langchain_core.runnables import RunnableWithFallbacks
from langchain_openai import ChatOpenAI

primary = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    base_url=os.getenv("N4N_BASE_URL"),
    default_headers={"x-n4n-routing": '["gpt-4o-mini", "claude-3-5-sonnet"]'},
)

# Separate client with different routing for the fallback chain
fallback_client = ChatOpenAI(
    model="llama-3.1-70b",
    api_key=os.getenv("N4N_API_KEY"),
    base_url=os.getenv("N4N_BASE_URL"),
    default_headers={"x-n4n-routing": '["llama-3.1-70b", "gemini-1.5-pro"]'},
)

resilient_chain = RunnableWithFallbacks(
    runnable=primary,
    fallbacks=[fallback_client],
    exception_key="error",  # optional: capture error info
)

# This tries primary (with its internal fallback), then fallback_client (with its internal fallback)
result = resilient_chain.invoke("Summarize the benefits of multi-model routing")
print(result.content)

Step 9: Verify end-to-end in a test script

Create a verification script you can run in CI or as a smoke test after deployments.

# verify_routing.py
import os
import json
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

def test_basic_invocation():
    llm = ChatOpenAI(
        model="gpt-4o-mini",
        api_key=os.getenv("N4N_API_KEY"),
        base_url=os.getenv("N4N_BASE_URL"),
    )
    resp = llm.invoke("Reply with: SUCCESS")
    assert "SUCCESS" in resp.content.upper()
    print("✓ Basic invocation works")

def test_fallback_routing():
    llm = ChatOpenAI(
        model="gpt-4o-mini",
        api_key=os.getenv("N4N_API_KEY"),
        base_url=os.getenv("N4N_BASE_URL"),
        default_headers={"x-n4n-routing": '["invalid-model", "gpt-4o-mini"]'},
    )
    resp = llm.invoke("Reply with: FALLBACK_OK")
    assert "FALLBACK_OK" in resp.content.upper()
    print("✓ Fallback routing works")

def test_streaming():
    llm = ChatOpenAI(
        model="gpt-4o-mini",
        api_key=os.getenv("N4N_API_KEY"),
        base_url=os.getenv("N4N_BASE_URL"),
    )
    chunks = list(llm.stream("Count: one, two"))
    assert len(chunks) > 1
    print("✓ Streaming works")

def test_usage_metadata():
    llm = ChatOpenAI(
        model="gpt-4o-mini",
        api_key=os.getenv("N4N_API_KEY"),
        base_url=os.getenv("N4N_BASE_URL"),
    )
    resp = llm.invoke("Hi")
    usage = resp.response_metadata.get("token_usage", {})
    assert usage.get("total_tokens", 0) > 0
    print("✓ Usage metadata present")

if __name__ == "__main__":
    test_basic_invocation()
    test_fallback_routing()
    test_streaming()
    test_usage_metadata()
    print("\nAll verification tests passed.")

Run it:

python verify_routing.py

Step 10: Production hardening checklist

Before shipping, address these operational concerns:

Timeouts and retries: Configure HTTP timeouts on the client. n4n.ai’s gateway fallback typically completes within 2-3 seconds, but network variability exists.

import httpx

http_client = httpx.Client(timeout=httpx.Timeout(30.0, connect=5.0))

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    base_url=os.getenv("N4N_BASE_URL"),
    http_client=http_client,
)

Circuit breaking at the application layer: If your routing list exhausts, the gateway returns a 5xx. Wrap calls in a circuit breaker (e.g., pybreaker) to fail fast and alert.

Model capability matching: Don’t route a 7B model a task that needs 70B reasoning. Maintain a capability matrix in your routing config:

ROUTING_POLICIES = {
    "coding": ["claude-3-5-sonnet", "gpt-4o", "deepseek-coder"],
    "reasoning": ["gpt-4o", "claude-3-5-sonnet", "gemini-1.5-pro"],
    "classification": ["gpt-4o-mini", "llama-3.1-8b", "gemini-1.5-flash"],
    "chat": ["gpt-4o-mini", "claude-3-5-haiku", "llama-3.1-70b"],
}

Cache control: n4n.ai forwards provider cache-control headers. For idempotent prompts (embeddings, classifications), enable caching at your load balancer or CDN layer to reduce latency and cost.

You now have a LangChain pipeline that load balances across dozens of models, falls back automatically when providers degrade, streams transparently through failures, and emits the observability data you need to debug and optimize. The gateway handles the provider complexity; your application code stays clean.