Quick Guide to LangGraph 1.0: Core Concepts, Nodes, and Edges

Core abstractions

Data → State : mutable information that flows through the graph.

Function → Node : a Python function that receives a State instance and returns an updated State.

Control flow → Edge : determines the execution order; can be static, parallel, or conditional.

Storage → Checkpoint/Memory : persistence mechanism for pausing, resuming, or long‑term memory.

Interrupt → Human‑in‑the‑Loop : manual review or intervention at designated points.

State definition

State is declared as a TypedDict (or a Pydantic model). Example:

class State(TypedDict):
    nlist: List[str]

The graph itself is stateless; the State instance is shared by all nodes and can be checkpointed for recovery after failures.

Node definition

A node is a pure function that operates on the state. Minimal example that appends a message:

def node_a(state: State):
    print(f"node_a received {state['nlist']}")
    return State(nlist=["Hello, I am node a"])

Building a simple graph

Creating a single‑node graph:

from langgraph.graph import START, END, StateGraph

builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_edge(START, "a")
builder.add_edge("a", END)
graph = builder.compile()

initial_state = State(nlist=["Hello Node a, how are you?"])
print(graph.invoke(initial_state))

The output shows the node receiving the initial state and returning the updated state.

Edge types

Ordinary (static) edges : fixed execution order; support serial chains (START → A → B → END) and parallel branches where a node has multiple outgoing edges, causing downstream nodes to run concurrently.

Conditional edges : a routing function inspects the current state and returns the name of the next node, enabling dynamic branching based on runtime data.

MapReduce edges : a node can spawn a variable number of downstream nodes, each receiving a distinct piece of data.

Reducer mechanism

When parallel nodes update the same state field, a conflict arises. By default later updates overwrite earlier ones. A Reducer is a merge function that defines how to combine concurrent updates. Built‑in reducers include: operator.add (int, float) – accumulate numeric values (e.g., counters, scores). operator.extend (list[T]) – extend a list (e.g., collect multiple results). operator.or_ (set[T]) – union of sets with deduplication (e.g., collect unique IDs). update_dict (dict) – new keys overwrite old ones (e.g., merge configurations).

Custom reducer example

A reducer that merges two lists while preserving order and removing duplicates:

def deduplicate_merge(old_list: List[str], new_list: List[str]) -> List[str]:
    combined = old_list + new_list
    return list(dict.fromkeys(combined))  # keep order, dedupe

Use it in a TypedDict field:

class MyState(TypedDict):
    unique_items: Annotated[List[str], deduplicate_merge]

Parallel execution with shared global state

Example graph where nodes B and C run concurrently after A, each appending a token to nlist. Because the state is globally shared, downstream nodes BB and CC see the combined updates (e.g., ['Initial', 'A', 'B', 'C']). A reducer such as operator.add ensures the list is concatenated rather than overwritten.

import operator
from typing import TypedDict, List, Annotated

class State(TypedDict):
    nlist: Annotated[List[str], operator.add]

def node_a(state: State) -> State:
    return State(nlist=["A"])

def node_b(state: State) -> State:
    return State(nlist=["B"])

def node_c(state: State) -> State:
    return State(nlist=["C"])

def node_bb(state: State) -> State:
    return State(nlist=["BB"])

def node_cc(state: State) -> State:
    return State(nlist=["CC"])

def node_d(state: State) -> State:
    return State(nlist=["D"])

builder = StateGraph(State)
for name, fn in {
    "a": node_a,
    "b": node_b,
    "c": node_c,
    "bb": node_bb,
    "cc": node_cc,
    "d": node_d,
}.items():
    builder.add_node(name, fn)

builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "bb")
builder.add_edge("c", "cc")
builder.add_edge("bb", "d")
builder.add_edge("cc", "d")
builder.add_edge("d", END)

graph = builder.compile()

initial_state = State(nlist=["Initial"])
print(graph.invoke(initial_state))

Conditional edge example

Routing function that selects the next node based on the last element of nlist:

def conditional_edge(state: State) -> Literal['b', 'c', END]:
    select = state["nlist"][-1]
    if select == "b":
        return 'b'
    elif select == "c":
        return 'c'
    else:
        return END

Attach the conditional edge to node a:

builder.add_conditional_edges("a", conditional_edge)

Running the graph with user input "b", "c", or "q" routes execution to the corresponding node or directly to END, demonstrating dynamic control flow.