Building Production-Ready AI Agents with LangGraph

Every team that ships an LLM-powered chain eventually hits the same wall: a linear sequence of prompt-parse-call steps cannot represent "try this, and if it fails, backtrack and ask the user a clarifying question." Chains are trees pretending to be lists. LangGraph fixes this by making the actual control flow - branches, loops, retries, human checkpoints - a first-class graph instead of something you fake with nested try/catch blocks.

Chains vs. Graphs

A LangChain chain is fundamentally step1 -> step2 -> step3. It works great for "summarize this document" or "classify this ticket." It falls apart the moment you need:

• Conditional branching based on intermediate output
• Loops (retry a tool call up to N times, refine an answer until it passes a check)
• Shared, mutable state across steps that isn't just "the output of the last step"
• A pause point where a human needs to approve before continuing

LangGraph models your agent as nodes and edges over an explicit state object, which is really just a typed dictionary that gets threaded through the graph and updated at each step.

from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    ticket_id: str
    resolution_attempts: int
    needs_human_review: bool

Defining Nodes and Edges

Each node is just a function that takes the current state and returns a partial update. Edges wire nodes together, and conditional edges let the graph branch based on the state.

from langgraph.graph import StateGraph, END

def classify_ticket(state: AgentState) -> dict:
    result = classifier_chain.invoke(state["messages"][-1])
    return {"messages": [result]}

def attempt_resolution(state: AgentState) -> dict:
    fix = resolution_agent.invoke(state)
    return {
        "messages": [fix],
        "resolution_attempts": state["resolution_attempts"] + 1,
    }

def route_after_attempt(state: AgentState) -> str:
    if state["resolution_attempts"] >= 3:
        return "escalate_to_human"
    if resolution_passed(state):
        return END
    return "attempt_resolution"

graph = StateGraph(AgentState)
graph.add_node("classify_ticket", classify_ticket)
graph.add_node("attempt_resolution", attempt_resolution)
graph.add_node("escalate_to_human", lambda s: {"needs_human_review": True})

graph.set_entry_point("classify_ticket")
graph.add_edge("classify_ticket", "attempt_resolution")
graph.add_conditional_edges("attempt_resolution", route_after_attempt)
graph.add_edge("escalate_to_human", END)

app = graph.compile()

Notice route_after_attempt isn't a hack bolted onto a chain - it's a real router function that inspects state and returns the name of the next node. This is the whole value proposition: control flow you can read, test, and reason about independently of the prompts.

Human-in-the-Loop Checkpoints

Production agents almost always need a point where a human approves a risky action - issuing a refund, deleting a record, sending an external email. LangGraph supports this natively with interrupt_before, which pauses graph execution and persists the state until you explicitly resume it.

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["issue_refund"],
)

# First invocation runs up to the interrupt and stops
result = app.invoke(initial_state, config={"configurable": {"thread_id": "ticket-4471"}})

# ... a human reviews the pending action in your UI ...

# Resume execution from exactly where it paused
app.invoke(None, config={"configurable": {"thread_id": "ticket-4471"}})

That thread_id is the key piece - it's how LangGraph keeps a persistent, resumable execution context per conversation or per task, separate from any single process's memory.

Persistence and Replay

LangGraph checkpointers snapshot state after every node execution. In production we back this with Postgres so a crashed worker, a deploy, or a multi-hour human review delay doesn't lose any progress.

from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string(
    "postgresql://agent_user:***@db.internal:5432/agent_state"
)
checkpointer.setup()

app = graph.compile(checkpointer=checkpointer)

Because every state transition is checkpointed, you also get replay for free - pull the full history of a thread and step through exactly what the agent saw and decided at each node, which turns "the agent did something weird in production" from a mystery into a five-minute debugging session.

for state in app.get_state_history(
    config={"configurable": {"thread_id": "ticket-4471"}}
):
    print(state.values["resolution_attempts"], state.next)

Subgraphs for Composability

Once an agent grows past a handful of nodes, compose it out of subgraphs rather than one flat graph. A "research" subgraph and a "drafting" subgraph can each be built, tested, and versioned independently, then wired together as a single node in the parent graph.

research_subgraph = build_research_graph().compile()

parent_graph.add_node("research", research_subgraph)

Common Pitfalls

• Unbounded loops - always cap retry counts in your routing function; an LLM that "almost" solves a problem will happily loop forever without a hard limit
• Overloaded state - keep the state object lean; stuffing entire document contents into it bloats every checkpoint write
• Skipping checkpointer setup in dev - test against the same Postgres-backed checkpointer you'll use in prod, not the in-memory default, or you'll miss serialization bugs

LangGraph doesn't make your agent smarter - it makes the thing surrounding the model honest about the fact that real workflows branch, loop, and sometimes need a human to press "approve." If your current chain has more than two if statements duct-taped around it, it's already a graph; LangGraph just lets you say so.