Route LangChain calls between GPT-4o and Claude 3.5

Routing LangChain calls between GPT-4o and Claude 3.5 lets you optimize for cost, latency, and capability per request. This langchain routing gpt-4o claude tutorial walks through three patterns: simple fallback, capability-based routing, and a production gateway that handles provider failures automatically. You’ll end up with runnable code and a verification checklist.

Prerequisites

• Python 3.10+
• OpenAI API key with GPT-4o access
• Anthropic API key with Claude 3.5 Sonnet access
• Optional: n4n.ai API key if you want a single endpoint for both (covers 240+ models with automatic fallback)

Install dependencies:

pip install langchain langchain-openai langchain-anthropic python-dotenv

Create a .env file:

OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
# N4N_API_KEY=n4n-...  # optional, for Step 4

Step 1: Initialize both chat models

LangChain’s ChatOpenAI and ChatAnthropic classes share the BaseChatModel interface, so you can swap them without changing downstream code.

# models.py
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.language_models import BaseChatModel
from dotenv import load_dotenv
import os

load_dotenv()

def get_gpt4o(**kwargs) -> BaseChatModel:
    return ChatOpenAI(
        model="gpt-4o",
        temperature=0,
        api_key=os.getenv("OPENAI_API_KEY"),
        **kwargs
    )

def get_claude_35(**kwargs) -> BaseChatModel:
    return ChatAnthropic(
        model="claude-3-5-sonnet-20241022",
        temperature=0,
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        **kwargs
    )

Verify both work independently:

# test_models.py
from models import get_gpt4o, get_claude_35

for name, factory in [("gpt-4o", get_gpt4o), ("claude-3.5", get_claude_35)]:
    llm = factory()
    resp = llm.invoke("Reply with only the model name.")
    print(f"{name}: {resp.content.strip()}")

Run it. You should see each model identify itself.

Step 2: Build a fallback chain

The simplest langchain routing gpt-4o claude pattern: try GPT-4o first, fall back to Claude on any error (rate limit, timeout, 5xx). LangChain’s RunnableWithFallbacks handles this.

# fallback_chain.py
from langchain_core.runnables import RunnableWithFallbacks
from models import get_gpt4o, get_claude_35

primary = get_gpt4o()
fallback = get_claude_35()

chain = primary.with_fallbacks([fallback])

# Test: force a failure by using an invalid key temporarily
# or just verify the chain structure
print(chain.get_graph().print_ascii())

The graph shows ChatOpenAI -> fallback -> ChatAnthropic. Now invoke it:

resp = chain.invoke("Summarize the plot of The Matrix in two sentences.")
print(resp.content)

If GPT-4o succeeds, you get its response. If it throws, Claude answers transparently. The caller sees one Runnable interface.

Verification: Temporarily invalidate OPENAI_API_KEY in .env, re-run, and confirm Claude responds. Restore the key afterward.

Step 3: Route by capability, not just failure

Fallback is reactive. Proactive routing picks the model suited to the task: GPT-4o for structured output and tool use, Claude for long-context reasoning and coding. Implement a router that inspects the request.

# router.py
from typing import Literal
from langchain_core.runnables import Runnable, RunnableLambda, RunnablePassthrough
from langchain_core.messages import BaseMessage
from models import get_gpt4o, get_claude_35

gpt4o = get_gpt4o()
claude = get_claude_35()

def pick_model(input_data: dict) -> Runnable:
    """
    input_data keys: 'messages' (list[BaseMessage]), 'task_type' (str)
    """
    task = input_data.get("task_type", "general")
    
    # Extend this logic as needed
    if task in ("json_mode", "function_calling", "vision"):
        return gpt4o
    if task in ("long_context", "code_review", "analysis"):
        return claude
    # Default: cheaper/faster model
    return gpt4o

router = RunnableLambda(pick_model)

# Usage
chain = (
    RunnablePassthrough.assign(model=router)
    | RunnableLambda(lambda x: x["model"].invoke(x["messages"]))
)

# Test cases
test_cases = [
    {"messages": [("human", "Output JSON: {\"name\": \"Alice\", \"age\": 30}")], "task_type": "json_mode"},
    {"messages": [("human", "Review this 500-line Python file for bugs...")], "task_type": "code_review"},
    {"messages": [("human", "What's 2+2?")], "task_type": "general"},
]

for tc in test_cases:
    resp = chain.invoke(tc)
    print(f"Task: {tc['task_type']} -> {resp.content[:80]}...")

Verification: Each task type routes to the intended model. Add logging inside pick_model to print the selected model name during development.

Step 4: Unify behind a single gateway (optional but recommended)

Managing two API keys, two SDKs, and two rate-limit buckets gets messy. n4n.ai exposes one OpenAI-compatible endpoint that addresses 240+ models — including GPT-4o and Claude 3.5 — and handles automatic fallback when a provider is rate-limited or degraded. You keep the same LangChain code; only the base URL and model name change.

# gateway_models.py
from langchain_openai import ChatOpenAI
from langchain_core.language_models import BaseChatModel
from dotenv import load_dotenv
import os

load_dotenv()

N4N_BASE_URL = "https://api.n4n.ai/v1"  # OpenAI-compatible
N4N_KEY = os.getenv("N4N_API_KEY")

def get_gateway_model(model_name: str, **kwargs) -> BaseChatModel:
    return ChatOpenAI(
        model=model_name,
        temperature=0,
        api_key=N4N_KEY,
        base_url=N4N_BASE_URL,
        **kwargs
    )

# Usage
gpt4o = get_gateway_model("openai/gpt-4o")
claude = get_gateway_model("anthropic/claude-3.5-sonnet")

The router from Step 3 works unchanged — just import gpt4o and claude from here. Per-token usage metering comes back in the response headers, and the gateway forwards provider cache-control hints so you can reason about cached vs. fresh tokens.

Verification: Call both models through the gateway. Check response headers for x-n4n-usage and x-n4n-provider to confirm routing.

Step 5: Add observability and guardrails

Production routing needs visibility. Wrap the chain with LangSmith or a lightweight callback that logs model selection, latency, token counts, and errors.

# observed_chain.py
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from typing import Any, Dict, List
import time
import json

class RoutingLogger(BaseCallbackHandler):
    def __init__(self):
        self.start_times: Dict[str, float] = {}
    
    def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs):
        run_id = kwargs.get("run_id")
        if run_id:
            self.start_times[str(run_id)] = time.time()
        model = serialized.get("kwargs", {}).get("model_name", "unknown")
        print(f"[ROUTE] Starting {model} | prompt chars: {sum(len(p) for p in prompts)}")
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        run_id = kwargs.get("run_id")
        if run_id and str(run_id) in self.start_times:
            latency = time.time() - self.start_times.pop(str(run_id))
            usage = response.llm_output.get("token_usage", {}) if response.llm_output else {}
            print(f"[ROUTE] Completed in {latency:.2f}s | usage: {usage}")

# Attach to any chain
from router import chain as base_chain
observed_chain = base_chain.with_config(callbacks=[RoutingLogger()])

# Test
observed_chain.invoke({
    "messages": [("human", "Write a haiku about distributed systems.")],
    "task_type": "general"
})

Output shows which model handled the request and how long it took. Extend RoutingLogger to push to your metrics backend (Datadog, Prometheus, etc.).

Step 6: Handle streaming and tool calls

Streaming works across both providers, but tool calling schemas differ. GPT-4o uses OpenAI’s tools format; Claude uses Anthropic’s tools format (similar but not identical). LangChain normalizes this via bind_tools, but only if the underlying model supports it.

# streaming_tools.py
from langchain_core.tools import tool
from models import get_gpt4o, get_claude_35

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"{city}: 72°F, sunny"

gpt4o = get_gpt4o().bind_tools([get_weather])
claude = get_claude_35().bind_tools([get_weather])

# Test tool calling
for name, model in [("gpt-4o", gpt4o), ("claude", claude)]:
    print(f"\n--- {name} ---")
    resp = model.invoke("What's the weather in Tokyo?")
    print(f"Tool calls: {resp.tool_calls}")
    if resp.tool_calls:
        for tc in resp.tool_calls:
            result = get_weather.invoke(tc)
            print(f"Tool result: {result}")

Verification: Both models should emit a get_weather tool call with {"city": "Tokyo"}. If one fails, check the model’s tool-calling support in the LangChain version you’re using.

For streaming, wrap the final chain:

from router import chain

for chunk in chain.stream({
    "messages": [("human", "Count to 10 slowly.")],
    "task_type": "general"
}):
    print(chunk.content, end="", flush=True)
print()

Both providers stream tokens through the same iterator.

Step 7: Verify end-to-end with a test suite

Create a pytest file that exercises every route and the fallback path.

# test_routing.py
import pytest
from router import chain as router_chain
from fallback_chain import chain as fallback_chain
from gateway_models import get_gateway_model

@pytest.mark.parametrize("task_type,expected_model", [
    ("json_mode", "gpt-4o"),
    ("code_review", "claude-3.5-sonnet"),
    ("general", "gpt-4o"),
])
def test_router_selects_correct_model(task_type, expected_model, monkeypatch):
    # Monkeypatch the model factories to return spies
    # This test structure assumes you refactor router.py to accept model instances
    pass  # Implement based on your DI approach

def test_fallback_triggers_on_failure(monkeypatch):
    # Force primary to raise, assert fallback responds
    pass

def test_gateway_returns_provider_header():
    if not get_gateway_model("openai/gpt-4o"):
        pytest.skip("N4N_API_KEY not set")
    model = get_gateway_model("openai/gpt-4o")
    resp = model.invoke("ping")
    # In real test, inspect raw response headers via httpx client
    assert resp.content

Run with pytest -v. Fill in the monkeypatch logic once you extract model factories into a config module.

Production checklist

Before deploying:

• Secrets management: API keys in vault, not .env
• Rate-limit handling: Gateway fallback covers provider limits; add client-side exponential backoff for gateway-level limits
• Cost tracking: Log token_usage per model per request; aggregate daily
• Latency SLAs: Route latency-sensitive tasks to the faster model (usually GPT-4o for short prompts)
• Data residency: Confirm provider regions match compliance requirements
• Model version pinning: Use dated model IDs (gpt-4o-2024-08-06, claude-3-5-sonnet-20241022) to avoid surprise regressions
• Fallback testing: Schedule monthly chaos tests that disable each provider

What to avoid

• Don’t route based on prompt keywords alone — use explicit task_type from your application logic
• Don’t assume both models support identical tool schemas — test each integration
• Don’t hardcode model names in business logic — centralize in a config module
• Don’t skip streaming tests — token-by-token behavior differs between providers

Next steps

• Add a cost-aware router that checks real-time pricing and picks the cheaper model for equivalent quality
• Implement semantic routing with an embedding classifier for fuzzy task types
• Build a canary deployment that shifts 5% of traffic to a new model version
• Explore n4n.ai’s routing directives (x-n4n-route: preferred=anthropic,fallback=openai) for declarative control without code changes

The patterns here scale from a single script to a platform serving millions of requests. Start with the fallback chain, add capability routing when you have clear task categories, and move to a gateway when operational overhead outweighs the integration cost.