LangGraph — when you outgrow LCEL

LCEL composes Runnables into DAGs. That covers a lot of ground — extract-then-summarize, RAG, fan-out / fan-in. What it doesn’t handle gracefully:

• Loops (call a tool, check the result, decide whether to call again).
• Conditional re-entry (the model’s answer fails validation; go back to the planning step).
• Pauses for a human to approve.
• Persistent state that outlives a single invocation.

These are the everyday shapes of real agents. LangGraph is LangChain’s answer for them — explicit state graphs (a graph where every node shares a common typed state object) built on top of the same Runnable interface you already know.

Why a graph and not a linear chain? A chain is a pipeline: data flows through once, front to back. A graph can have cycles — a node (a function that reads state and returns an update) can point back to an earlier node. That lets you express “loop until done,” “retry on bad output,” and “wait for human input.” The linear chain has no way to represent those shapes; the graph does.

Put a name to that contrast. A pipeline — an LCEL chain, an Airflow job, a Spark plan — is a DAG: a directed acyclic graph. “Directed” means each edge has a direction; “acyclic” is the load-bearing word — no path is ever allowed to loop back to a node that already ran. LangGraph keeps “directed graph” and deliberately drops “acyclic.” That single change — letting an edge point backward — is the entire reason it exists. So when people say “LangGraph is not just a DAG,” this is what they mean: a DAG can’t loop, and looping is exactly what an agent does.

The mental model

A LangGraph workflow has three parts:

1. State — a typed dict that flows through the graph. Every node reads it and produces a partial update.
2. Nodes — functions of state ((State) -> dict). Each node returns a partial dict of updates that LangGraph merges back.
3. Edges — wiring between nodes. Edges can be unconditional (A → B) or conditional (a router function inspects state and picks the next node: A → B or C).

That’s the whole abstraction. A graph is “starting from START, run node, apply update, follow edge, repeat until you hit END.”

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class State(TypedDict):
    question: str
    answer: str

def think(state: State) -> dict:
    return {"answer": f"thinking about: {state['question']}"}

graph = StateGraph(State)
graph.add_node("think", think)
graph.add_edge(START, "think")
graph.add_edge("think", END)

app = graph.compile()
app.invoke({"question": "what is a JOIN?"})
# → {"question": "what is a JOIN?", "answer": "thinking about: what is a JOIN?"}

Five concepts: StateGraph, add_node, add_edge, compile, invoke. Everything else is variation on this skeleton.

A three-node graph

Let’s build something less trivial: a classify → process → format flow. The first node decides whether the user’s input is a question or a command, the second handles it, the third packages the output.

# A mocked LangGraph 3-node graph. The shape (StateGraph, add_node,
# add_edge, compile, invoke) matches real LangGraph code.

from typing import TypedDict

# --- mocked LangGraph runtime ---
START, END = "__start__", "__end__"

class StateGraph:
    def __init__(self, schema):
        self.schema = schema
        self.nodes = {}
        self.edges = {}            # plain edges: src -> dst
        self.cond_edges = {}       # conditional edges: src -> (router, mapping)
    def add_node(self, name, fn):
        self.nodes[name] = fn
    def add_edge(self, src, dst):
        self.edges[src] = dst
    def add_conditional_edges(self, src, router, mapping):
        self.cond_edges[src] = (router, mapping)
    def compile(self):
        return CompiledGraph(self)

class CompiledGraph:
    def __init__(self, graph):
        self.g = graph
    def invoke(self, state):
        current = START
        while True:
            if current in self.g.cond_edges:
                router, mapping = self.g.cond_edges[current]
                key = router(state)
                current = mapping[key]
                continue
            if current in self.g.edges:
                current = self.g.edges[current]
                if current == END:
                    return state
                update = self.g.nodes[current](state)
                state = {**state, **update}
            else:
                return state

# --- the actual graph (this part is the LangGraph shape you'd write) ---
class State(TypedDict):
    input: str
    kind: str
    result: str
    formatted: str

def classify(state: State) -> dict:
    text = state["input"].lower()
    return {"kind": "question" if "?" in text else "command"}

def process(state: State) -> dict:
    if state["kind"] == "question":
        return {"result": f"Here's an answer to: {state['input']}"}
    else:
        return {"result": f"Executed: {state['input']}"}

def format_output(state: State) -> dict:
    return {"formatted": f"[{state['kind'].upper()}] {state['result']}"}

graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("process", process)
graph.add_node("format", format_output)

graph.add_edge(START, "classify")
graph.add_edge("classify", "process")
graph.add_edge("process", "format")
graph.add_edge("format", END)

app = graph.compile()

print(app.invoke({"input": "What is a SQL JOIN?", "kind": "", "result": "", "formatted": ""}))
print(app.invoke({"input": "delete table users", "kind": "", "result": "", "formatted": ""}))

{'input': 'What is a SQL JOIN?', 'kind': 'question', 'result': "Here's an answer to: What is a SQL JOIN?", 'formatted': "[QUESTION] Here's an answer to: What is a SQL JOIN?"}
{'input': 'delete table users', 'kind': 'command', 'result': 'Executed: delete table users', 'formatted': '[COMMAND] Executed: delete table users'}

Follow the state through. It enters as a dict with three empty fields; classify fills kind (the ? makes the first input a question, the second a command), process fills result off that kind, and format assembles formatted from both. Each node returned only the one key it changed, and the runtime merged it back. The interesting thing isn’t this particular graph — it’s the shape. You’ll write hundreds of these; the same pattern scales from three nodes to thirty.

Conditional edges — the branching primitive

Most real graphs have branches. add_conditional_edges lets a node hand off to one of several next nodes based on the state.

def route(state) -> str:
    return "tool" if state["needs_tool"] else "respond"

graph.add_conditional_edges(
    "agent_node",
    route,
    {"tool": "tool_node", "respond": "format_node"},
)

The router function returns a key; the mapping points that key to the next node. This is how you implement “if the LLM emitted a tool call, go run the tool; otherwise, return the answer”:

When to choose LangGraph over LCEL

A decision flow:

• One LLM call → no framework needed.
• Two-to-five steps, no branching → LCEL.
• DAG with fan-out / fan-in → LCEL.
• Cycles, retries, branching back to a previous step → LangGraph.
• Long-running with persistence → LangGraph.
• Human-in-the-loop pauses → LangGraph.

The honest test: draw your workflow on a whiteboard. If there’s a cycle, an “if/else that loops back,” or a “wait for human” step, LangGraph is what you want.

Nodes can wrap LCEL chains

You don’t throw out your LCEL chains when you move to LangGraph. A node is just a function over state — if you already have a prompt | model | parser chain, wrap it in a node.

analyze_chain = prompt | model | parser  # LCEL

def analyze_node(state: State) -> dict:
    answer = analyze_chain.invoke({"text": state["input"]})
    return {"analysis": answer}

graph.add_node("analyze", analyze_node)

This is the right migration pattern: keep your LCEL chains, drop them into LangGraph nodes, wire branching and loops at the graph level.

The four things you still need to learn

This lesson covered the skeleton — StateGraph, nodes, edges, compile. The four remaining ideas show up in the next lessons:

• State updates with reducers — what happens when two nodes update the same key? (Default: overwrite. Often: append.)
• Checkpointers — saving state to disk so a graph can be paused and resumed.
• Streaming — watching intermediate state flow as the graph runs.
• Interrupts — pausing a graph for human approval before continuing.

Those are the next three lessons.

In one breath

• LCEL handles DAGs; LangGraph keeps “directed graph” but drops acyclic — that single back-edge is the whole reason it exists, because looping is what agents do.
• A workflow is three parts: state (a typed dict that flows through), nodes (functions of state returning a partial update), edges (transitions, plain or conditional).
• The skeleton is five calls: StateGraph → add_node → add_edge → compile → invoke; run from START, apply each update, follow edges to END.
• add_conditional_edges is the branching primitive — a router reads state and picks the next node (the basis of “tool call → run tool, else answer”).
• Reach for it on cycles, retries, persistence, or human-in-the-loop — nodes can wrap existing LCEL chains, so migration is cheap; but don’t over-graph a simple linear flow.

Quick check

Next

Nodes return partial state updates. What if two nodes update the same key? The next lesson covers state and reducers — how LangGraph merges updates and why this matters for message lists.