LangGraph Series, Part 2: Core Building Blocks — State, Nodes, Edges, and Reducers

Part 2 of 7. Part 1 covered why LangGraph exists. This post covers the four concepts everything else builds on: State, Nodes, Edges, and reducers. Every pattern in Parts 3–5 is a composition of these four things.

📦 Companion Code: part2_basic_chatbot.py in the langgraph-ai repository.

A note on the model provider: the code in this series uses OpenRouter as the LLM provider, so the same graph code runs against any model available there by changing one environment variable. Each file defines a get_llm() helper that reads your OPENROUTER_API_KEY and a model slug from the environment — see the repository README for setup. All parts default to z-ai/glm-5.3-flash.

State: the shared object every node reads from and writes to

State is a TypedDict — a plain Python typed dictionary — that every node in the graph both reads from and writes updates to. It's the graph's "working memory" for a single run.

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]
    retry_count: int

Why a single shared state object, instead of passing custom arguments between functions like normal code? Because a graph's nodes don't call each other directly — the graph runtime decides what runs next, sometimes based on conditions, sometimes in a loop you didn't write by hand. If Node A had to know exactly what arguments Node B expects, you couldn't rearrange the graph without rewriting every node's signature. A shared state object decouples nodes from each other: each node only needs to know the shape of the state, not which specific node ran before it. This is the same reasoning behind why event-driven systems pass a shared message/event object rather than direct function arguments — it's what makes the pieces swappable.

Reducers: how updates actually get applied

By default, when a node returns {"retry_count": 3}, LangGraph overwrites that key in the state. But for messages, you don't want to overwrite the whole conversation history every time — you want to append. That's what Annotated[list, add_messages] does: it tells LangGraph "when a node updates this field, run it through the add_messages function instead of just overwriting."

Why does this matter enough to be a core concept? Because it's the most common source of subtle bugs for people new to LangGraph: forgetting to mark a list field with an appropriate reducer, then watching conversation history silently disappear because each node's return value replaced the previous list instead of extending it. Understanding reducers early saves you a confusing debugging session later. You can also write your own reducer for custom merge logic — e.g., a field that should keep the maximum of two values (useful for tracking things like "highest risk score seen so far" across nodes) rather than the default overwrite or the messages-style append.

Nodes: plain functions with one job

A node is a function: (state) -> partial_state_update. That's the entire contract.

def validate_input(state: State) -> dict:
    if not state["messages"]:
        return {"retry_count": state["retry_count"] + 1}
    return {}

Why keep nodes this dumb on purpose? Because a function that only reads state in and returns state out is trivially unit-testable — you don't need to spin up the whole graph, mock an LLM, or set up a database connection to test validate_input; you just call it with a fake state dict and check the return value. This is a deliberate design choice that pays off enormously in maintainability (more in Part 6): the alternative — nodes that reach out and mutate global variables, write directly to a database mid-function, or depend on some other node having run first in a way that isn't visible from the function signature — is exactly the kind of code that becomes unmaintainable as a team grows past one person.

Edges: normal and conditional

A normal edge is fixed: after Node A, always run Node B.

graph_builder.add_edge("validate_input", "process")

A conditional edge inspects the state and picks the next node dynamically:

def route(state: State) -> str:
    if state["retry_count"] >= 3:
        return "escalate_to_human"
    return "process"

graph_builder.add_conditional_edges("validate_input", route)

Why separate "normal" and "conditional" edges instead of just always using a routing function? Performance and readability aren't the main reason — it's about making the fixed parts of your business logic visually and structurally undeniable. If a refund must always go through a specific compliance-check node no matter what, encoding that as a plain edge means nothing — not even a bug in a routing function — can accidentally skip it. Conditional edges are reserved for genuine decision points. This separation is what lets a non-engineer reviewing your graph diagram (most tracing tools render the graph visually) understand at a glance which paths are guaranteed and which are decided at runtime — a distinction that matters a lot when a compliance or risk team is asking "can this ever skip the approval step?"

The Graph: START, END, and compilation

from langgraph.graph import StateGraph, START, END

graph_builder = StateGraph(State)
graph_builder.add_node("validate_input", validate_input)
graph_builder.add_node("process", process)
graph_builder.add_edge(START, "validate_input")
graph_builder.add_conditional_edges("validate_input", route)
graph_builder.add_edge("process", END)

graph = graph_builder.compile()

START and END are just fixed markers for "entry point" and "exit point." Compilation is a deliberate separate step — graph_builder is a mutable definition you can still add nodes/edges to, while graph (the compiled result) is the executable, immutable object you actually run. Why split these? So that graph construction (which might be conditional on config, feature flags, or environment) is clearly separated from graph execution — you build the shape once, validate it (LangGraph checks for things like unreachable nodes at compile time), and then run that exact validated shape repeatedly, rather than re-validating structure on every single invocation.

A working chatbot — the full code, explained

Here is the complete runnable chatbot from part2_basic_chatbot.py:

import os
from typing import Annotated
from typing_extensions import TypedDict
from dotenv import load_dotenv
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

load_dotenv()


def get_llm(
    model_env_var: str = "OPENROUTER_MODEL",
    default_model: str = "z-ai/glm-5.3-flash",  # cheap, fast model good for demos
    **kwargs,
) -> ChatOpenAI:
    return ChatOpenAI(
        model=os.environ.get(model_env_var, default_model),
        api_key=os.environ["OPENROUTER_API_KEY"],
        base_url="https://openrouter.ai/api/v1",
        default_headers={
            "HTTP-Referer": "https://example.com",
            "X-Title": "LangGraph Series Examples",
        },
        **kwargs,
    )


# --------------- State ---------------
class State(TypedDict):
    messages: Annotated[list, add_messages]
    # add_messages reducer: new messages are APPENDED, not overwritten.
    # Without this annotation, every node return would replace the full history.


# --------------- LLM + Node ---------------
llm = get_llm()


def chatbot(state: State) -> dict:
    # Invoke the LLM with the current conversation history.
    # Returns a dict; LangGraph merges it into state using the reducers.
    return {"messages": [llm.invoke(state["messages"])]}


# --------------- Graph ---------------
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# MemorySaver: keeps state in RAM, scoped by thread_id.
# Swap for SqliteSaver / PostgresSaver in production (see Part 4).
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)


def main() -> None:
    # thread_id is the conversation's unique key in the checkpointer.
    # Same thread_id → history is loaded and appended.
    # Different thread_id → fresh conversation, no shared memory.
    config = {"configurable": {"thread_id": "demo-user-1"}}

    print("--- Turn 1 ---")
    result1 = graph.invoke({"messages": [("user", "Hi, I'm Sam.")]}, config)
    print(result1["messages"][-1].content)
    # Output:
    # Hi Sam! I'm GLM, an AI assistant trained by Z.ai. How are you doing today?
    # I'm here to help with any questions or topics you'd like to discuss.
    # What can I do for you?

    print("\n--- Turn 2 (same thread_id, should remember the name) ---")
    result2 = graph.invoke({"messages": [("user", "What's my name?")]}, config)
    print(result2["messages"][-1].content)
    # Output:
    # Your name is Sam! You mentioned it when you first said hello.
    # Is there anything I can help you with, Sam?

    print("\n--- Turn 3 (different thread_id, should NOT know the name) ---")
    other_config = {"configurable": {"thread_id": "demo-user-2"}}
    result3 = graph.invoke({"messages": [("user", "What's my name?")]}, other_config)
    print(result3["messages"][-1].content)
    # Output:
    # I don't have access to your personal information unless you've shared it
    # with me in our conversation. I'm GLM, a large language model trained by
    # Z.ai, and I don't store your personal data between conversations.
    # If you'd like to tell me your name, I'm happy to remember it for the
    # duration of our chat! How else can I assist you today?


if __name__ == "__main__":
    main()

Why MemorySaver and thread_id here, and why does this matter beyond "it remembers things"? The checkpointer decouples conversation identity from your application code. Nothing in the chatbot function knows or cares that there's history being tracked — the graph runtime handles loading the right history for thread_id: "demo-user-1" before running the node, and saving the updated state after. This means the exact same node function works whether you're running one conversation or ten thousand concurrent ones; the isolation is handled entirely by which thread_id you pass in.

Notice Turn 3: passing a different thread_id ("demo-user-2") gives a completely fresh conversation — the model has no memory of "Sam." This is the thread_id-as-isolation mechanism at work: the same compiled graph, zero code changes, different scope. This is also why, in Part 4, swapping MemorySaver for PostgresSaver for production requires zero changes to any node — the persistence backend is a plug-in choice at compile time, not something baked into your business logic.

Common mistake this section prevents

A very common first bug: defining state as a plain dict type hint (or forgetting Annotated[..., add_messages] on the messages field) and then being confused why the chatbot "forgets" what was just said one node later, inside the same run — not even across turns. This happens because two nodes both returning {"messages": [...]} without a reducer causes the second one to silently overwrite the first's contribution instead of appending to it. If you remember one thing from this post, make it: any state field that should accumulate across nodes (lists, running totals, logs) needs an explicit reducer; anything that should just be replaced (a status flag, a single classification result) is fine with the default overwrite behavior.

Sample Output

Running part2_basic_chatbot.py produces the following (live run, 2026-08-31 ✅):

--- Turn 1 ---
Hi Sam! I'm GLM, an AI assistant trained by Z.ai. How are you doing today? I'm here
to help with any questions or topics you'd like to discuss. What can I do for you?

--- Turn 2 (same thread_id, should remember the name) ---
Your name is Sam! You mentioned it when you first said hello. Is there anything I can
help you with, Sam?

--- Turn 3 (different thread_id, should NOT know the name) ---
I don't have access to your personal information unless you've shared it with me in
our conversation. I'm GLM, a large language model trained by Z.ai, and I don't store
your personal data between conversations.

If you'd like to tell me your name, I'm happy to remember it for the duration of our
chat! How else can I assist you today?

• Turn 1 — LLM greeted user by name ✅
• Turn 2 — Same thread_id → memory retained, name recalled ✅
• Turn 3 — Different thread_id → no memory of name ✅

What's next

Part 3 builds a tool-using agent — the ReAct loop — and gets into why loop-based agent control beats a hardcoded sequence of tool calls, plus how to handle tool failures without the whole graph breaking.