LangGraph fundamentals: StateGraph, nodes, edges, routing

Why this matters

LangGraph 1.x is the production standard for agent orchestration in 2026 — used by Klarna, Uber, LinkedIn, BlackRock, JPMorgan, Replit, Cisco. "Multi-agent orchestration in LangGraph" pays a +$30k to $60k US salary premium because most candidates have only built linear LCEL chains.

This lesson teaches the core: state, nodes, edges, conditional routing. The next lesson covers HIL, persistence, sub-graphs.

Learning objectives

Build a StateGraph from scratch.
Define typed state with TypedDict + reducers.
Add nodes (Python functions) and edges (transitions).
Use conditional edges for routing decisions.
Stream node-by-node output to a UI.
Use the prebuilt create_agent for ReAct out of the box.

1. Mental model

A LangGraph graph is:

State — a single TypedDict that flows through the graph.
Nodes — pure-ish functions: def node(state) -> dict_of_updates.
Edges — directed transitions; can be conditional.
Compile — turn the builder into a runnable.
Checkpointer (optional) — persist state across runs.

Picture a state machine where every node reads state, returns updates, and the runtime merges them.

2. Hello world

python

# uv add langgraph langchain langchain-openai
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

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

def answer_node(state: S) -> dict:
    return {"answer": f"You asked: {state['question']}"}

builder = StateGraph(S)
builder.add_node("answer", answer_node)
builder.add_edge(START, "answer")
builder.add_edge("answer", END)
graph = builder.compile()

print(graph.invoke({"question": "Hi"}))
# {'question': 'Hi', 'answer': 'You asked: Hi'}

That is the pattern. Memorise it.

3. State reducers (when nodes "append" instead of overwrite)

By default, when two nodes write the same key, the later one overwrites. For chat-style histories you want append semantics.

python

from typing import Annotated
from operator import add

class ChatState(TypedDict):
    messages: Annotated[list[dict], add]   # reducer: concat lists
    step: int                                # default: overwrite

Or use the helper add_messages which handles dedup of message ids:

python

from langgraph.graph.message import add_messages
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]

4. Nodes that call LLMs and tools

python

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

llm = ChatOpenAI(model="gpt-4.1-mini")

def llm_node(state: ChatState) -> dict:
    sys = SystemMessage(content="You are concise.")
    response = llm.invoke([sys] + state["messages"])
    return {"messages": [response]}

builder = StateGraph(ChatState)
builder.add_node("chat", llm_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
graph = builder.compile()

print(graph.invoke({"messages": [HumanMessage("What is RAG?")]}))

Each node returns only the keys it wants to update. The runtime merges with the reducer for each key.

5. Conditional edges (the routing primitive)

Where LangGraph leaves LCEL behind: routing nodes by inspecting state.

python

def classify(state: S) -> str:
    """Returns the name of the next node."""
    q = state["question"].lower()
    if "billing" in q: return "billing"
    if "tech"    in q: return "tech"
    return "general"

builder.add_node("billing", billing_node)
builder.add_node("tech",    tech_node)
builder.add_node("general", general_node)

builder.add_conditional_edges(
    source="router",
    path=classify,                    # function returning next node name
    path_map={"billing": "billing", "tech": "tech", "general": "general"},
)
builder.add_edge("billing", END)
builder.add_edge("tech",    END)
builder.add_edge("general", END)

path can also be the literal return value if it matches a node name. path_map is documented for clarity.

6. Visualising your graph

python

graph.get_graph().draw_mermaid()    # returns a Mermaid string
# or save:
graph.get_graph().draw_mermaid_png(output_file_path="graph.png")

Drop the PNG in your README. It is one of the cheapest "looks senior" moves.

For interactive debugging install LangGraph Studio (free, electron app). Open the project; live trace and time-travel debug.

7. Streaming events from a graph

python

for event in graph.stream(
    {"messages": [HumanMessage("Tell me about RAG")]},
    stream_mode="updates",
):
    print(event)

stream_mode options (memorise):

"values" — full state after each step.
"updates" — only what changed (the deltas).
"messages" — token-by-token from LLM nodes.
"debug" — every internal event (verbose).

For UIs you usually want messages (token streaming) plus updates (so you can show "Calling tool: X").

8. The prebuilt `create_agent` (ReAct in 1 line)

You can write your own ReAct loop, but for many cases the prebuilt create_agent is faster. (Note: create_react_agent from langgraph.prebuilt was renamed to create_agent and now lives in the langchain package as of LangGraph 1.x.)

python

# uv add langchain langgraph langchain-openai
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a given city."""
    return f"It's 32°C in {city}"

@tool
def calc(expr: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expr))

agent = create_agent(
    model="openai:gpt-4.1-mini",
    tools=[get_weather, calc],
    prompt="You are a helpful assistant. Use tools when useful.",
)

result = agent.invoke({"messages": [{"role":"user", "content":"weather in Mumbai squared"}]})
print(result["messages"][-1].content)

Under the hood it is a StateGraph with two nodes (agent and tools) and a conditional edge that loops until no more tool calls. You get checkpointing, streaming, and HIL for free.

Reach for create_agent when you have a single agent with a fixed tool set. Build your own graph when you need multi-agent topologies, custom routing, or non-trivial state.

9. A real example: classify → retrieve → answer

python

class State(TypedDict):
    question: str
    intent: str
    sources: list[str]
    answer: str

def classify(state: State) -> dict:
    intent = router_llm.invoke(state["question"])  # returns 'kb' or 'web'
    return {"intent": intent}

def retrieve_kb(state: State) -> dict:
    return {"sources": kb_retriever.invoke(state["question"])}

def retrieve_web(state: State) -> dict:
    return {"sources": tavily_search(state["question"])}

def answer(state: State) -> dict:
    ctx = "\n".join(state["sources"])
    a = llm.invoke([
        SystemMessage("Answer using only the sources."),
        HumanMessage(f"Sources:\n{ctx}\n\nQ: {state['question']}"),
    ]).content
    return {"answer": a}

g = StateGraph(State)
g.add_node("classify",     classify)
g.add_node("retrieve_kb",  retrieve_kb)
g.add_node("retrieve_web", retrieve_web)
g.add_node("answer",       answer)

g.add_edge(START, "classify")
g.add_conditional_edges(
    "classify",
    lambda s: s["intent"],
    {"kb": "retrieve_kb", "web": "retrieve_web"},
)
g.add_edge("retrieve_kb",  "answer")
g.add_edge("retrieve_web", "answer")
g.add_edge("answer", END)

app = g.compile()
print(app.invoke({"question": "Latest news on Anthropic models"}))

This graph branches by intent, retrieves from the right source, then synthesises. Visualise it to check edges. Add tracing (next lesson) and you can debug each step.

Hands-on lab (5 hours)

Build a smart-help bot as a LangGraph:

State:

python

class S(TypedDict):
    question: str
    intent: Literal["billing", "technical", "general", "abuse"]
    sources: list[str]
    answer: str
    refused: bool

Nodes:

guardrail_in — flags abuse / off-topic; sets refused if so.
classify_intent — maps question to one of the 3 specialist intents.
retrieve_kb — vector + BM25 + rerank.
billing_specialist, technical_specialist, general_specialist — each with its own system prompt.
format_answer — adds citations, polite tone.

Edges:

START → guardrail_in.
guardrail_in → END if refused else classify_intent.
classify_intent → retrieve_kb (always).
Conditional edge from retrieve_kb to one of the 3 specialists by intent.
Each specialist → format_answer → END.

Acceptance:

A 50-question eval set. Accuracy ≥ 0.85 on intent classification.
Streaming UI in Streamlit shows current node ("Routing → Retrieving → Answering").
README has the mermaid diagram.

Common pitfalls

Forgetting reducers — chat history gets overwritten instead of appended.
Returning the whole state — only return updated keys.
Conditional edge function returning a node name that does not exist — runtime error.
Mixing path and path_map — pick one style and document it.
Hard-coding the next node — use conditional edges; static add_edge is for true straight-line transitions only.

Self-check

What is the difference between stream_mode="values" and "updates"?
When do you need a reducer on a state field?
What does add_conditional_edges do that add_edge cannot?
When do you choose create_agent over a hand-built StateGraph?
How do you draw a mermaid diagram of your graph for the README?

Why this matters

Learning objectives

1. Mental model

2. Hello world

3. State reducers (when nodes "append" instead of overwrite)

4. Nodes that call LLMs and tools

5. Conditional edges (the routing primitive)

6. Visualising your graph

7. Streaming events from a graph

8. The prebuilt create_agent (ReAct in 1 line)

9. A real example: classify → retrieve → answer

Hands-on lab (5 hours)

Common pitfalls

Self-check

References

8. The prebuilt `create_agent` (ReAct in 1 line)