Skip to content

Graph Basics — Nodes, Edges, Compile

1. Why this matters

If you understand state (Chapter 2) and this chapter, you can build 80% of LangGraph apps. Everything else — conditional routing, loops, persistence, agents — is built on this skeleton.

2. Mental model

Build → wire → compile → run.

flowchart LR
    Schema[Define State<br/>TypedDict] --> Build[builder = StateGraph state]
    Build --> Nodes[add_node 'name', fn]
    Nodes --> Edges[add_edge a, b<br/>add_conditional_edges ...]
    Edges --> Compile[graph = builder.compile]
    Compile --> Run[graph.invoke initial_state]

The compile step validates the graph and binds optional features (checkpointer, interrupts, debug). The runnable returned implements LCEL's interface, so it slots into bigger pipelines.

3. Architecture / Flow

A complete graph in pictures:

flowchart TD
    START((START)) --> A[Node A]
    A --> B[Node B]
    B --> R{Condition}
    R -->|yes| C[Node C]
    R -->|no| D[Node D]
    C --> ENDN((END))
    D --> ENDN

In code that becomes:

builder.add_edge(START, "A")
builder.add_edge("A", "B")
builder.add_conditional_edges("B", route_fn, {"yes": "C", "no": "D"})
builder.add_edge("C", END)
builder.add_edge("D", END)

4. Core concepts

  • StateGraph(StateSchema) — the builder. Pass the TypedDict (or Pydantic) class as the schema.
  • add_node("name", fn) — registers a function as a node. Name is how edges refer to it.
  • add_edge("from", "to") — static edge. After "from" runs, "to" always runs next.
  • add_conditional_edges("from", router_fn, mapping?) — router_fn(state) -> str picks the destination node name; optional mapping translates router output to actual node names.
  • START — pseudo-node where the run begins. Always add an edge from START.
  • END — pseudo-node where the run terminates. Nodes that reach END produce the final state.
  • set_entry_point("name") — shortcut for add_edge(START, "name").
  • compile(checkpointer=..., interrupt_before=..., debug=...) — finalize.
  • invoke(input, config) — run synchronously, return final state.
  • stream(input, config, stream_mode=...) — yield intermediate state.
  • get_graph().print_ascii() / .draw_mermaid() — visualize.

5. Code — minimal working example

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

class State(TypedDict):
    x: int
    y: int

def add_one(state: State):
    return {"y": state["x"] + 1}

builder = StateGraph(State)
builder.add_node("add_one", add_one)
builder.add_edge(START, "add_one")
builder.add_edge("add_one", END)

graph = builder.compile()
print(graph.invoke({"x": 41, "y": 0}))   # {'x': 41, 'y': 42}

6. Code — real-world pattern

Three-stage processing pipeline with logging via state:

from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class State(TypedDict):
    topic: str
    outline: str
    blog: str
    summary: str
    log: Annotated[list[str], add]    # accumulating audit log

def make_outline(state: State):
    out = llm.invoke([HumanMessage(f"Outline a blog about {state['topic']}.")])
    return {"outline": out.content, "log": ["outline_done"]}

def write_blog(state: State):
    out = llm.invoke([HumanMessage(f"Write the blog using this outline:\n{state['outline']}")])
    return {"blog": out.content, "log": ["blog_done"]}

def summarize(state: State):
    out = llm.invoke([HumanMessage(f"One-sentence TL;DR of:\n{state['blog']}")])
    return {"summary": out.content, "log": ["summary_done"]}

b = StateGraph(State)
b.add_node("outline", make_outline)
b.add_node("write",   write_blog)
b.add_node("sum",     summarize)

b.add_edge(START, "outline")
b.add_edge("outline", "write")
b.add_edge("write", "sum")
b.add_edge("sum", END)

graph = b.compile()

final = graph.invoke({
    "topic": "vector embeddings",
    "outline": "", "blog": "", "summary": "", "log": [],
})

print(final["summary"])
print("Audit:", final["log"])  # ['outline_done', 'blog_done', 'summary_done']

Visualize it:

graph.get_graph().print_ascii()
# Or in a notebook:
# graph.get_graph().draw_mermaid_png()

Stream intermediate state instead of waiting for the final value:

for snapshot in graph.stream(initial, stream_mode="updates"):
    # 'updates' yields {node_name: partial_update} per step
    print(snapshot)

7. Common pitfalls

  • ❗ Forgetting START and END edges. Compile fails with "no entry point" / "no terminal node". Every graph needs both.
  • ❗ Node name conflicts. Two nodes with the same name overwrite. Use clear, unique names.
  • ❗ Using a Python variable instead of a string in add_edge. Edges reference nodes by name string, not function object.
  • ❗ Conditional router returning a node name not in the mapping. Raises at runtime. Either avoid the mapping (return node names directly) or include all paths including END.
  • ❗ Calling .invoke() on the builder. You must compile() first — the builder isn't runnable.
  • ❗ Multiple nodes pointing to one downstream node without a join semantics decided. LangGraph runs parallel branches concurrently; reducers must handle the merge.

8. When to use vs not use

Use this When
add_edge (static) Step always follows step
add_conditional_edges Routing depends on state
set_entry_point Shorthand for the one START edge
compile() with checkpointer You need persistence (see Chapter 8)
compile() with interrupt_before Human-in-the-loop
compile() with debug=True Verbose logging during development

9. Cheatsheet

from langgraph.graph import StateGraph, START, END

# Build
builder = StateGraph(StateSchema)
builder.add_node("name1", fn1)
builder.add_node("name2", fn2)
builder.set_entry_point("name1")          # or add_edge(START, "name1")
builder.add_edge("name1", "name2")
builder.add_conditional_edges(
    "name2",
    router_fn,                            # state -> str
    {"branch_a": "name1", "branch_b": END},  # optional mapping
)
builder.set_finish_point("name2")         # or add_edge("name2", END)

# Compile
graph = builder.compile(
    checkpointer=...,
    interrupt_before=["risky_node"],      # HITL pause
    interrupt_after=[],
    debug=False,
)

# Run
graph.invoke(initial_state, config={"configurable": {"thread_id": "t1"}})
graph.stream(initial_state, config=..., stream_mode="updates")   # or "values" / "messages" / "debug"
graph.batch([state1, state2])

# Inspect
graph.get_graph().print_ascii()
graph.get_graph().draw_mermaid()          # returns Mermaid string
graph.get_state(config)                    # current snapshot (needs checkpointer)
list(graph.get_state_history(config))      # all snapshots
graph.update_state(config, {"key": "new"})

10. Q&A — recall test

  • Q: Two ways to set the entry of a graph? A: builder.add_edge(START, "first") or builder.set_entry_point("first"). They're equivalent.

  • Q: What does add_conditional_edges need that add_edge doesn't? A: A router function that takes state and returns a string (the next node's name). Optionally a mapping translating router output to node names.

  • Q: Why is compile() separate from building? A: Compile validates the graph (entry, terminal, no orphans), binds optional features (checkpointer, interrupts), and returns an immutable Runnable. Keeps build-time vs run-time concerns clean.

  • Q: What's the difference between stream_mode="values" and "updates"? A: "values" yields the full state after each step. "updates" yields the partial update each node returned. "messages" yields LLM tokens for streaming chat UIs.

  • Q: How do you visualize a compiled graph? A: graph.get_graph().print_ascii() (terminal) or graph.get_graph().draw_mermaid() (Mermaid source — drop into a Markdown file).

Practice

What does this print?

Expected: True

# A node is a function: state → state update
def my_node(state):
    return {"count": state["count"] + 1}
result = my_node({"count": 5})
print(result == {"count": 6})

Compile the graph before invoking it

Expected: True

is_compiled = False             # bug: must call .compile() before .invoke()
print(not is_compiled)

Quiz — Quick check

What you remember

Q1. What's a node in LangGraph?

  • A function that takes the current state and returns a partial state update
  • A class
  • An LLM
  • A tool

Why: Nodes are functions. They read state (first arg) and return a dict with the fields they want to update. LangGraph merges that into the global state using the schema's reducers.

Q2. What's the role of START and END in a graph?

  • Special sentinel nodes marking where execution begins (START) and terminates (END)
  • Required imports
  • Logging hooks
  • State fields

Why: Every graph needs at least one entry edge from START to a real node, and at least one edge ending at END. Without these, the graph can't execute or terminate.

Q3. Why must you .compile() a graph before invoking it?

  • Compilation validates the structure, sets up checkpointers, and creates the executable Runnable
  • To save it to disk
  • To save memory
  • Optional — works without

Why: Before compilation, the graph is just a builder. .compile() produces a CompiledGraph (a Runnable) with .invoke(), .stream(), etc. Pass options like checkpointer= to compile.

Common doubts

Can a node call another node?

No, not directly. Nodes communicate through state. To "call" another node, return state that routes the graph to it (via conditional edges). This separation keeps the graph predictable and debuggable.

How do I have multiple nodes share computation?

Compute once in a shared node early in the graph, store the result in state, then read it from subsequent nodes. Don't duplicate computation across nodes.

What happens to fields a node doesn't return?

They keep their previous value. Nodes return partial state updates — only the fields that changed. Unchanged fields are inherited from the previous state.