Why we moved from the raw OpenAI SDK to LangGraph

We started with the raw OpenAI SDK because it’s the path of least resistance: client.chat.completions.create() and you’re done. Six months later, our inference layer had become a sprawling collection of ad-hoc orchestration logic — retry loops wrapped around provider fallbacks, conversation state scattered across Redis keys, and a “human review” flow that required a separate service just to pause execution. The SDK doesn’t solve orchestration; it pretends orchestration doesn’t exist. Migrating to LangGraph forced us to model our LLM workflows as explicit state machines, and that structural shift eliminated entire categories of bugs we’d been chasing for quarters.

The breaking point of raw SDK calls

The OpenAI SDK is excellent at what it does: serializing requests, handling auth, parsing streaming responses, and exposing a clean async interface. It is not, and never claims to be, a workflow engine. But once you move past single-turn chat, you start building one anyway.

Our first workaround was a run_with_fallback helper that tried OpenAI, then Anthropic, then a local model. Then we needed to preserve conversation history across provider switches, so we added a ConversationManager class. Then came the need to branch: if the model calls a tool, execute it and feed the result back; if it refuses, escalate to a human. That logic lived in a 400-line orchestrate.py that nobody wanted to touch.

The problems compounded silently:

No checkpointing: A failure at step 4 of 6 meant restarting from scratch, re-paying for tokens already generated.
Implicit state: Context lived in Python lists passed between functions. Debugging meant adding print() statements to reconstruct what the model “knew” at each step.
No streaming at the workflow level: We could stream tokens from the model, but couldn’t stream progress — “executing tool,” “waiting for human,” “retrying provider” — to the frontend.
Testing was theater: Unit tests mocked chat.completions.create. Integration tests hit real APIs. Nothing exercised the actual control flow.

The SDK encourages you to treat each model call as independent. Real LLM applications are sequences of dependent calls with branching, loops, and external side effects. That mismatch is why migrate openai sdk to langgraph becomes a real question, not a hypothetical one.

What LangGraph actually gives you

LangGraph is a state machine library that happens to integrate tightly with LangChain’s model abstractions. Its core primitives are nodes (functions that read/write state), edges (deterministic or conditional transitions), and a checkpointing layer that persists state after every node.

This maps directly to how LLM workflows actually work:

# Before: implicit flow in a single function
async def handle_request(user_msg: str, conversation_id: str):
    history = await redis.get(conversation_id)
    messages = history + [{"role": "user", "content": user_msg}]
    
    response = await client.chat.completions.create(
        model="gpt-4o", messages=messages, tools=tools
    )
    
    if response.choices[0].message.tool_calls:
        for call in response.choices[0].message.tool_calls:
            result = await execute_tool(call)
            messages.append({"role": "tool", "content": result, "tool_call_id": call.id})
        # Recursive call — no checkpoint, no visibility
        return await handle_request("", conversation_id)
    
    await redis.set(conversation_id, messages)
    return response.choices[0].message.content

# After: explicit graph with checkpointing
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # auto-accumulates
    next_step: str

def call_model(state: AgentState):
    response = client.chat.completions.create(
        model="gpt-4o", messages=state["messages"], tools=tools
    )
    return {"messages": [response.choices[0].message]}

def execute_tools(state: AgentState):
    tool_calls = state["messages"][-1].tool_calls
    results = []
    for call in tool_calls:
        result = execute_tool_sync(call)  # or async variant
        results.append({"role": "tool", "content": result, "tool_call_id": call.id})
    return {"messages": results}

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

graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", execute_tools)
graph.add_edge("tools", "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
graph.set_entry_point("agent")

# Compile with checkpointing — every node execution is a recoverable snapshot
app = graph.compile(checkpointer=SqliteSaver.from_conn_string("checkpoints.db"))

# Run with a thread_id — resumable, inspectable, streamable
config = {"configurable": {"thread_id": conversation_id}}
for chunk in app.stream({"messages": [HumanMessage(content=user_msg)]}, config):
    print(chunk)  # yields node outputs as they complete

The difference isn’t syntactic — it’s architectural. The graph is the documentation. A new engineer can read should_continue and understand the branching logic without tracing through nested conditionals. The checkpointer means a crash at the tool node lets you resume from that exact point with app.invoke(None, config) — same thread, same state, no duplicate tokens.

Where the migration paid off immediately

1. Provider fallback as a graph edge, not a try/except block

We route 240+ models through a single gateway. When a provider returns 429 or 5xx, we want to retry with exponential backoff, then fail over to the next preferred model — without losing the conversation context. In the raw SDK world, this lived in a wrapper function that knew too much about provider internals.

With LangGraph, fallback is a conditional edge:

def route_provider(state: AgentState) -> str:
    last_error = state.get("last_error")
    if last_error and last_error.provider == "openai":
        return "anthropic"
    return "openai"

graph.add_node("openai", call_openai)
graph.add_node("anthropic", call_anthropic)
graph.add_conditional_edges("openai", route_provider, {"anthropic": "anthropic", "openai": "openai"})

The gateway we use (n4n.ai) handles provider-level fallback automatically, but application-level fallback — switching model families when one degrades — still lives in our graph. The checkpoint means the failover picks up mid-conversation, not at the beginning.

2. Human-in-the-loop without a separate queue service

Our content moderation flow requires human approval for flagged outputs. Previously: write to a Postgres queue, spin up a React admin panel, poll for approval, resume. Now:

def needs_review(state: AgentState) -> bool:
    return moderate(state["messages"][-1].content) == "flagged"

graph.add_node("human_review", interrupt_before=["human_review"])
graph.add_edge("human_review", "agent")

# In the API handler:
config = {"configurable": {"thread_id": thread_id}}
# First run — pauses at interrupt_before
result = app.invoke(input_data, config)

# Later, when human approves via API:
app.invoke(Command(resume={"approved": True}), config)

interrupt_before persists state and returns control to the caller. The frontend polls /threads/{id}/state to show “awaiting review.” No queue, no separate worker, no serialization logic. The graph is the queue.

3. Streaming workflow progress, not just tokens

Frontend teams want to show “Thinking…”, “Searching docs…”, “Generating response…” — not just token deltas. With the raw SDK, we’d hack this by yielding custom events from the orchestration function. LangGraph’s stream() yields node-level events natively:

for event in app.stream(input_data, config, stream_mode="updates"):
    for node_name, output in event.items():
        if node_name == "retrieve":
            yield {"status": "searching", "query": output["query"]}
        elif node_name == "agent":
            yield {"status": "generating", "tokens": output["messages"][-1].content}

The graph structure makes this trivial: each node maps to a UI state. No ad-hoc event emission scattered through business logic.

The tradeoffs we accepted

Learning curve is real

LangGraph has its own mental model: state reducers, channel semantics, interrupt/resume patterns, subgraph compilation. Two engineers on our team took ~3 weeks to become productive. The documentation assumes familiarity with LangChain’s Runnable interface. If your team has never used LangChain, budget for onboarding.

Abstraction leakage at the edges

LangGraph nodes are just Python functions. That’s a feature — you can do anything — but it means the graph doesn’t enforce purity. We’ve caught bugs where a node mutated state directly instead of returning updates, breaking checkpoint consistency. The fix is discipline (always return new state dicts) and linting, not framework guardrails.

Debugging requires new tools

print() debugging doesn’t work well when execution is distributed across graph invocations. We rely on:

LangSmith (or self-hosted equivalent) for trace visualization
The checkpointer’s SQLite DB for manual state inspection: SELECT * FROM checkpoints WHERE thread_id = '...'
app.get_state(config) to snapshot mid-flight

It’s better than the old print() chaos, but it’s a different workflow.

Not everything belongs in the graph

We kept pure utility functions (token counting, PII redaction, prompt templating) outside the graph. They’re stateless, testable in isolation, and don’t benefit from checkpointing. Forcing them into nodes adds serialization overhead and graph complexity without value. The heuristic: if it needs to resume after failure, it’s a node. If it’s a pure transform, it’s a function.

When not to migrate

If your LLM usage is:

Single-turn request/response
No tool use, no branching, no human review
Stateless by design (e.g., embedding generation, classification)

…the raw SDK (or LiteLLM, or direct HTTP) is the right choice. LangGraph adds compilation overhead (~50ms cold start), a dependency chain, and cognitive load. Don’t pay it unless you have stateful, multi-step workflows.

Also: if you’re already invested in a different orchestration layer (Temporal, Prefect, custom state machines), LangGraph duplicates that investment. It shines when the LLM calls themselves are the workflow steps — not when you’re orchestrating microservices that happen to call LLMs.

The decisive takeaway

We migrated because the raw OpenAI SDK treats every model call as an independent event. Our application treats them as a continuous, interruptible, auditable process. LangGraph closes that gap by making the process explicit: state, transitions, checkpoints, and streaming are first-class, not afterthoughts.

The migration cost was real — three engineers, six weeks, ~2,000 lines of orchestration code replaced by ~400 lines of graph definition. But we deleted the ConversationManager, the fallback wrapper, the human-review queue service, and the custom streaming event bus. The remaining code is about the workflow, not plumbing the workflow.

If you’re building anything beyond chat-with-history, the raw SDK will eventually force you to build a bad workflow engine. LangGraph is a good one. Use it.