Running the same LangGraph agent across four model providers

You want to run langgraph agent across multiple providers without rewriting your graph for each model’s API quirks. The trick is treating every provider as an OpenAI-compatible endpoint — same client, same tool-calling schema, same streaming interface. This tutorial builds a working ReAct agent in LangGraph and runs it against GPT-5, Claude 3.5 Sonnet, Gemini 1.5 Pro, and Llama 3.1 405B using one code path.

Prerequisites

• Python 3.10+
• langgraph>=0.2.0, langchain-openai>=0.1.0, langchain-core>=0.3.0
• Access to at least two model providers (API keys for OpenAI, Anthropic, Google AI Studio, and/or a Llama endpoint)
• Optional: an OpenRouter or n4n.ai key if you want a single endpoint that routes to all four

pip install langgraph langchain-openai langchain-core python-dotenv

Create a .env file with your keys:

# .env
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=...
# If using a unified gateway:
UNIFIED_BASE_URL=https://api.n4n.ai/v1
UNIFIED_API_KEY=n4n-...

The agent we’ll build

A minimal ReAct loop: the model decides whether to call a tool or respond, tools execute, results feed back, repeat until final answer. We’ll give it a calculator and a web search stub so you can see tool calling work across providers.

# agent.py
from typing import Annotated, Literal
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from pydantic import BaseModel
import os
from dotenv import load_dotenv

load_dotenv()

@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Error: {e}"

@tool
def web_search(query: str) -> str:
    """Stub search — replace with real implementation."""
    return f"[Search results for '{query}']: Sample result 1, Sample result 2"

tools = [calculator, web_search]

class AgentState(BaseModel):
    messages: Annotated[list[BaseMessage], add_messages]

def make_model(provider: Literal["openai", "anthropic", "google", "llama", "unified"]):
    """Return a ChatOpenAI client configured for the target provider."""
    configs = {
        "openai": {
            "model": "gpt-5",
            "api_key": os.getenv("OPENAI_API_KEY"),
            "base_url": "https://api.openai.com/v1",
        },
        "anthropic": {
            "model": "claude-3-5-sonnet-20241022",
            "api_key": os.getenv("ANTHROPIC_API_KEY"),
            "base_url": "https://api.anthropic.com/v1",
        },
        "google": {
            "model": "gemini-1.5-pro",
            "api_key": os.getenv("GOOGLE_API_KEY"),
            "base_url": "https://generativelanguage.googleapis.com/v1beta/openai",
        },
        "llama": {
            "model": "meta-llama/llama-3.1-405b-instruct",
            "api_key": os.getenv("LLAMA_API_KEY"),  # e.g., Together, Fireworks, Groq
            "base_url": os.getenv("LLAMA_BASE_URL", "https://api.together.xyz/v1"),
        },
        "unified": {
            "model": "auto",  # or specify "gpt-5", "claude-3.5-sonnet", etc.
            "api_key": os.getenv("UNIFIED_API_KEY"),
            "base_url": os.getenv("UNIFIED_BASE_URL", "https://api.n4n.ai/v1"),
        },
    }
    cfg = configs[provider]
    return ChatOpenAI(
        model=cfg["model"],
        api_key=cfg["api_key"],
        base_url=cfg["base_url"],
        temperature=0,
        max_tokens=4096,
    )

def should_continue(state: AgentState) -> Literal["tools", "end"]:
    last = state.messages[-1]
    return "tools" if last.tool_calls else "end"

def call_model(state: AgentState, model):
    response = model.invoke(state.messages)
    return {"messages": [response]}

def build_graph(provider: str):
    model = make_model(provider).bind_tools(tools)
    tool_node = ToolNode(tools)

    graph = StateGraph(AgentState)
    graph.add_node("agent", lambda s: call_model(s, model))
    graph.add_node("tools", tool_node)
    graph.set_entry_point("agent")
    graph.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
    graph.add_edge("tools", "agent")
    return graph.compile()

if __name__ == "__main__":
    import sys
    provider = sys.argv[1] if len(sys.argv) > 1 else "openai"
    app = build_graph(provider)
    
    # Test query requiring tool use
    result = app.invoke({
        "messages": [("human", "What's 37 * 42? Then search for 'LangGraph tutorial'.")]
    })
    for msg in result["messages"]:
        print(f"{msg.type.upper()}: {msg.content[:200]}")
        if msg.tool_calls:
            print(f"  TOOL CALLS: {msg.tool_calls}")

Run it:

python agent.py openai
python agent.py anthropic
python agent.py google
python agent.py llama
# Or with a unified gateway:
python agent.py unified

Expected output (OpenAI)

HUMAN: What's 37 * 42? Then search for 'LangGraph tutorial'.
AI: 
  TOOL CALLS: [{'name': 'calculator', 'args': {'expression': '37 * 42'}, 'id': 'call_123'}, {'name': 'web_search', 'args': {'query': 'LangGraph tutorial'}, 'id': 'call_456'}]
TOOL: 1554
TOOL: [Search results for 'LangGraph tutorial']: Sample result 1, Sample result 2
AI: 37 * 42 = 1,554. Here are the search results for 'LangGraph tutorial': Sample result 1, Sample result 2

The same graph produces identical structure on every provider. The only differences you’ll see are latency, token counts, and occasional formatting quirks in the final natural-language response.

Provider-specific gotchas

Tool calling format

All four providers now support OpenAI-style function calling, but the schema validation strictness varies.

If you hit Invalid tool call errors on Gemini or GPT-5, ensure your tool schemas are strict JSON Schema with no optional fields missing. LangChain’s @tool decorator handles this, but raw function definitions need strict: true in the function spec.

Context windows and output limits

# Adjust per provider in make_model()
context_limits = {
    "openai": 128_000,
    "anthropic": 200_000,
    "google": 1_000_000,
    "llama": 128_000,  # varies by host
}

Gemini’s 1M token window changes how you design state. You can stuff entire codebases into the prompt; on Llama you’ll need retrieval. The graph logic stays the same — only the preprocessing step changes.

Streaming differences

# Streaming works identically across providers
async for chunk in app.astream({"messages": [("human", "Calculate 123 * 456")]}):
    for node, update in chunk.items():
        if node == "agent" and update["messages"]:
            msg = update["messages"][-1]
            if msg.content:
                print(msg.content, end="", flush=True)

Claude streams tool calls as separate chunks; GPT-5 and Gemini bundle them. Your frontend should handle both: accumulate tool call deltas until tool_calls array is complete, then render.

Running the same agent in production

Unified routing with fallback

Instead of hardcoding providers, route through a gateway that handles failover. The graph code doesn’t change — only the base_url and model parameters.

# production_config.py
from dataclasses import dataclass
from typing import Optional

@dataclass
class ProviderConfig:
    name: str
    model: str
    base_url: str
    api_key: str
    priority: int
    max_retries: int = 2
    timeout: float = 30.0

PROVIDERS = [
    ProviderConfig("gpt-5", "gpt-5", "https://api.openai.com/v1", os.getenv("OPENAI_KEY"), 1),
    ProviderConfig("claude", "claude-3-5-sonnet", "https://api.anthropic.com/v1", os.getenv("ANTHROPIC_KEY"), 2),
    ProviderConfig("gemini", "gemini-1.5-pro", "https://generativelanguage.googleapis.com/v1beta/openai", os.getenv("GOOGLE_KEY"), 3),
    ProviderConfig("llama", "meta-llama/llama-3.1-405b", "https://api.together.xyz/v1", os.getenv("TOGETHER_KEY"), 4),
]

# Or use a unified endpoint that does this automatically
UNIFIED = ProviderConfig(
    "unified", "auto", 
    os.getenv("UNIFIED_BASE_URL", "https://api.n4n.ai/v1"),
    os.getenv("UNIFIED_KEY"), 1
)

A unified gateway like n4n.ai adds automatic fallback when a provider is rate-limited or degraded, per-token usage metering across all models, and forwards provider cache-control hints so you don’t pay for repeated prefixes.

Per-request routing directives

You can steer individual requests without changing the graph:

# In your request headers or body
extra_headers = {
    "x-n4n-model": "claude-3.5-sonnet",  # pin to specific model
    "x-n4n-fallback": "true",             # enable automatic fallback
    "x-n4n-cache-control": "no-store",    # disable caching for this request
}

Pass these via model.invoke(..., extra_headers=extra_headers) or configure them in the ChatOpenAI constructor.

Observability

Log the provider actually used for each request:

def call_model_with_logging(state: AgentState, model, provider_name: str):
    response = model.invoke(state.messages)
    usage = response.response_metadata.get("token_usage", {})
    print(f"[provider={provider_name}] input_tokens={usage.get('prompt_tokens')} output_tokens={usage.get('completion_tokens')}")
    return {"messages": [response]}

This lets you build cost dashboards per provider without changing your agent logic.

Testing checklist before switching providers

Run this matrix before declaring a provider production-ready:

# test_matrix.py
TEST_CASES = [
    ("simple_math", "What is 17 * 23?"),
    ("parallel_tools", "Calculate 12*34 and search for 'python asyncio'"),
    ("multi_turn", "What's the capital of France? Now what's its population?"),
    ("long_context", "Summarize this 50k token document: " + "x" * 50000),
    ("structured_output", "Return JSON: {\"name\": \"string\", \"age\": \"int\"} for a 30-year-old developer"),
]

def run_matrix(provider: str):
    app = build_graph(provider)
    for name, query in TEST_CASES:
        try:
            result = app.invoke({"messages": [("human", query)]})
            print(f"✓ {provider} {name}: {len(result['messages'])} messages")
        except Exception as e:
            print(f"✗ {provider} {name}: {e}")

for p in ["openai", "anthropic", "google", "llama", "unified"]:
    run_matrix(p)

Common failures:

• Llama on some hosts: tool call arguments arrive as stringified JSON instead of objects — add a parser middleware
• Gemini: safety filters trigger on benign math — adjust safety_settings in the request
• Claude: returns stop_reason: "max_tokens" mid-tool-call — increase max_tokens or handle continuation

When to use which provider

The agent graph doesn’t care. You swap one line — the provider identifier — and the same LangGraph compilation runs everywhere.

Next steps

1. Add real tools: Replace the search stub with Tavily, Exa, or your internal API
2. Persist state: Add SqliteSaver or PostgresSaver checkpointer for multi-turn conversations
3. Human-in-the-loop: Insert interrupt_before=["tools"] for approval gates
4. Evaluation: Build a test set with expected tool call sequences and assert parity across providers

The code in this tutorial is ~80 lines of graph definition plus provider config. That’s the entire surface area you maintain to run langgraph agent across multiple providers. Everything else — routing, fallback, metering, cache hints — lives in the infrastructure layer where it belongs.