LangGraph with Claude 3.5 Sonnet and GPT-4o in one graph

Most teams pick one model and optimize around it. But some tasks are genuinely better suited to different models — Claude 3.5 Sonnet excels at extended reasoning and code generation, while GPT-4o often wins on structured extraction and instruction following. This tutorial shows you how to build a langgraph claude 3.5 sonnet gpt-4o workflow that routes each node to the right model, shares state cleanly, and fails gracefully when a provider degrades.

Step 1: Set up the environment and dependencies

Create a fresh virtual environment and install the minimal set. We use langgraph for the graph runtime, langchain-anthropic and langchain-openai for the model wrappers, and python-dotenv for secrets.

python -m venv .venv
source .venv/bin/activate
pip install langgraph langchain-anthropic langchain-openai python-dotenv

Create a .env file with your API keys:

ANTHROPIC_API_KEY=sk-ant-...
OPENAI_API_KEY=sk-...

If you route through a gateway like n4n.ai, replace the base URLs in the client initialization below — the rest of the graph stays identical.

Step 2: Define the shared state schema

Every node in the graph reads from and writes to a single State object. Keep it flat and serializable; nested objects make checkpointing and debugging harder.

# state.py
from typing import TypedDict, Literal, Optional
from pydantic import BaseModel, Field

class TaskInput(BaseModel):
    prompt: str
    task_type: Literal["reasoning", "extraction", "coding", "general"]

class TaskOutput(BaseModel):
    result: str
    model_used: str
    tokens_in: int
    tokens_out: int

class State(TypedDict):
    input: TaskInput
    output: Optional[TaskOutput]
    error: Optional[str]
    retry_count: int

The task_type field drives routing. In production you’d classify this upstream (embed + classifier, or an LLM router), but for this tutorial we pass it explicitly.

Step 3: Initialize the two model clients

Wrap each provider in a thin function that returns a standardized response shape. This keeps the graph nodes clean and makes swapping providers trivial.

# models.py
import os
from langchain_anthropic import ChatAnthropic
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from state import TaskInput, TaskOutput

# Anthropic (Claude 3.5 Sonnet)
claude = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    temperature=0.2,
    max_tokens=4096,
)

# OpenAI (GPT-4o)
gpt4o = ChatOpenAI(
    model="gpt-4o-2024-08-06",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.2,
    max_tokens=4096,
)

def invoke_claude(task: TaskInput) -> TaskOutput:
    messages = [
        SystemMessage(content="You are a precise reasoning and coding assistant."),
        HumanMessage(content=task.prompt),
    ]
    resp = claude.invoke(messages)
    usage = resp.response_metadata.get("usage", {})
    return TaskOutput(
        result=resp.content,
        model_used="claude-3.5-sonnet",
        tokens_in=usage.get("input_tokens", 0),
        tokens_out=usage.get("output_tokens", 0),
    )

def invoke_gpt4o(task: TaskInput) -> TaskOutput:
    messages = [
        SystemMessage(content="You are a precise extraction and instruction-following assistant."),
        HumanMessage(content=task.prompt),
    ]
    resp = gpt4o.invoke(messages)
    usage = resp.response_metadata.get("token_usage", {})
    return TaskOutput(
        result=resp.content,
        model_used="gpt-4o",
        tokens_in=usage.get("prompt_tokens", 0),
        tokens_out=usage.get("completion_tokens", 0),
    )

Note the different usage key names — response_metadata shapes vary by provider. Normalize here so downstream nodes don’t need provider logic.

Step 4: Build the routing logic

The router is a pure function that inspects state["input"].task_type and returns the next node name. Keep it deterministic; stochastic routing belongs in a separate classifier node.

# router.py
from state import State
from typing import Literal

def route_task(state: State) -> Literal["claude_node", "gpt4o_node", "error_node"]:
    task_type = state["input"].task_type
    if task_type in ("reasoning", "coding"):
        return "claude_node"
    if task_type == "extraction":
        return "gpt4o_node"
    return "gpt4o_node"  # default fallback

Step 5: Implement the graph nodes

Each node is a thin wrapper around the model invocation with error handling and retry accounting. LangGraph passes the full State dict; return a partial dict to merge updates.

# nodes.py
from state import State, TaskOutput
from models import invoke_claude, invoke_gpt4o

MAX_RETRIES = 2

def claude_node(state: State) -> State:
    try:
        output = invoke_claude(state["input"])
        return {"output": output, "error": None}
    except Exception as e:
        return handle_error(state, e, "claude_node")

def gpt4o_node(state: State) -> State:
    try:
        output = invoke_gpt4o(state["input"])
        return {"output": output, "error": None}
    except Exception as e:
        return handle_error(state, e, "gpt4o_node")

def error_node(state: State) -> State:
    return {"error": "All models failed or task type unroutable", "output": None}

def handle_error(state: State, exc: Exception, node_name: str) -> State:
    retry_count = state.get("retry_count", 0) + 1
    if retry_count <= MAX_RETRIES:
        # In a real system you might switch models here
        return {"error": f"{node_name} failed: {exc}", "retry_count": retry_count}
    return {"error": f"{node_name} failed after {MAX_RETRIES} retries: {exc}", "retry_count": retry_count}

Step 6: Assemble the graph

Wire nodes and edges in a StateGraph. The conditional edge from START uses our router; each model node goes to END on success or error_node on failure (via the error field check).

# graph.py
from langgraph.graph import StateGraph, START, END
from state import State
from nodes import claude_node, gpt4o_node, error_node
from router import route_task

def should_retry(state: State) -> str:
    if state.get("error") and state.get("retry_count", 0) < 2:
        # Re-route to the other model on retry
        return "gpt4o_node" if "claude" in state["error"] else "claude_node"
    return "error_node" if state.get("error") else END

builder = StateGraph(State)

builder.add_node("claude_node", claude_node)
builder.add_node("gpt4o_node", gpt4o_node)
builder.add_node("error_node", error_node)

builder.add_conditional_edges(START, route_task)
builder.add_conditional_edges("claude_node", should_retry)
builder.add_conditional_edges("gpt4o_node", should_retry)
builder.add_edge("error_node", END)

graph = builder.compile()

The retry logic here is intentionally simple: on first failure, try the other model. For production you’d want exponential backoff, circuit breakers, and provider health checks.

Step 7: Add checkpointing for observability

LangGraph’s checkpointer persists every state transition. Use MemorySaver for local runs; swap to PostgresSaver or RedisSaver in production.

# run.py
from langgraph.checkpoint.memory import MemorySaver
from graph import graph
from state import TaskInput

checkpointer = MemorySaver()
app = graph.with_config(checkpointer=checkpointer)

# Example runs
tasks = [
    TaskInput(
        prompt="Refactor this async Python function to use structured concurrency with task groups.",
        task_type="coding",
    ),
    TaskInput(
        prompt="Extract all email addresses and phone numbers from this text: ...",
        task_type="extraction",
    ),
    TaskInput(
        prompt="Prove that the square root of 2 is irrational.",
        task_type="reasoning",
    ),
]

for i, task in enumerate(tasks):
    config = {"configurable": {"thread_id": f"demo-{i}"}}
    result = app.invoke({"input": task, "retry_count": 0}, config=config)
    print(f"=== Task {i+1} ({task.task_type}) ===")
    print(f"Model: {result['output'].model_used if result.get('output') else 'N/A'}")
    print(f"Tokens: {result['output'].tokens_in if result.get('output') else 0} in / {result['output'].tokens_out if result.get('output') else 0} out")
    print(f"Result: {result['output'].result[:200] if result.get('output') else result.get('error')}")
    print()

Run it:

python run.py

Step 8: Verify success and inspect checkpoints

Success criteria for each task:

output is present and error is null
model_used matches the expected routing (Claude for reasoning/coding, GPT-4o for extraction)
Token counts are non-zero and plausible for the prompt length

Inspect the full checkpoint history for a thread:

# inspect.py
from run import app

config = {"configurable": {"thread_id": "demo-0"}}
history = list(app.get_state_history(config))
for snapshot in history:
    print(f"Step: {snapshot.next_node}")
    print(f"  State keys: {list(snapshot.values.keys())}")
    if snapshot.values.get("output"):
        print(f"  Model: {snapshot.values['output'].model_used}")
    if snapshot.values.get("error"):
        print(f"  Error: {snapshot.values['error']}")

This shows every intermediate state — useful for debugging routing decisions and retry behavior.

Step 9: Add structured output for downstream consumers

Raw strings are brittle. Define a Pydantic schema for each task type and use with_structured_output on the model wrappers. This forces the model to emit valid JSON and gives you typed access in downstream nodes.

# structured.py
from pydantic import BaseModel, Field
from typing import List
from models import claude, gpt4o

class CodeRefactor(BaseModel):
    original_issues: List[str]
    refactored_code: str
    explanation: str

class ExtractionResult(BaseModel):
    emails: List[str]
    phones: List[str]

class ProofStep(BaseModel):
    statement: str
    justification: str

class ProofResult(BaseModel):
    theorem: str
    steps: List[ProofStep]
    conclusion: str

# Usage in nodes:
# structured_claude = claude.with_structured_output(CodeRefactor)
# result: CodeRefactor = structured_claude.invoke(messages)

Swap the node implementations to use the structured variants. The graph topology doesn’t change — only the node internals.

Step 10: Production hardening checklist

Before deploying this pattern, address these gaps:

Provider health awareness — Wrap each client with a circuit breaker that tracks error rates and latency percentiles. Route away from degraded providers before they fail requests.
Token budget enforcement — Add a pre-node check that estimates input tokens and rejects or truncates if the prompt exceeds the model’s context window minus a safety margin.
Cost attribution — Persist tokens_in, tokens_out, and model_used per request to your observability stack. Aggregate by task_type to validate routing decisions economically.
Idempotency keys — Generate a deterministic key from (task_type, prompt_hash) and pass it through the graph. Use it as the thread_id or a separate deduplication key so retries don’t double-charge.
Fallback ordering — Make the retry fallback configurable per task type. Some tasks (structured extraction) should never fall back to a model that hallucinates schema fields.
Streaming support — Replace invoke with astream in nodes that benefit from incremental output (long code generation). Update the state schema to accumulate chunks.

How to extend this pattern

The graph you built is a template. Common extensions:

Parallel fan-out: Send the same prompt to both models, then use a judge node (or deterministic comparison) to pick the better output.
Multi-step pipelines: Chain a reasoning node (Claude) → extraction node (GPT-4o) → formatting node (smaller/cheaper model).
Dynamic model selection: Replace the static router with an embedding-based classifier that maps arbitrary prompts to task_type at runtime.
Human-in-the-loop: Insert an interrupt before the final node for high-stakes tasks; resume after review.

The key insight: LangGraph makes the control flow explicit and inspectable. Model choice becomes a routing decision, not an architectural commitment.