LangGraph Fundamentals and State Graphs

Intermediate⏱45 min📖4 min readLesson 1 of 5

LangGraph Fundamentals and State Graphs

Learn what LangGraph is, how StateGraph differs from Graph, and how to define state, nodes, edges, compile, and run a state graph in Python.

LangGraph is a framework from LangChain for building stateful, multi-actor applications using graphs as the core abstraction. Each node modifies a shared state and edges define the flow.

What is LangGraph?

LangGraph extends LangChain by modeling agent logic as a directed graph. The graph carries a typed state object that persists across nodes, enabling complex loops, branching, and memory.

Key concepts:

StateGraph: Recommended class for stateful graphs
Graph: Simpler, stateless alternative
Nodes: Python functions that receive and mutate state
Edges: Directed connections between nodes

⚠️Warning

LangGraph is not a workflow DAG tool. Nodes can be revisited, loops can form, and state is preserved across cycles. This is what makes it suitable for agentic systems.

Mermaid: Full Execution Cycle

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

The Definition phase builds the graph topology. The Execution phase runs nodes sequentially or in parallel, each reading and writing to the shared state.

StateGraph vs Graph

Feature	StateGraph	Graph
Typed state	Yes (TypedDict or dataclass)	No (bare values only)
Conditional edges	Yes	Yes
Checkpointing	Built-in via MemorySaver	Not supported
Parallel branching	Yes	Limited
Re-entrant nodes	Yes	No
Production readiness	High (PostgresSaver, etc.)	Low
Loop support	Yes	No
Human-in-the-loop	Via interrupt()	Not supported
Subgraph composition	Yes	No

💡Tip

Always default to StateGraph unless you have a very simple stateless pipeline. The overhead is minimal and you gain checkpointing, branching, and production features for free.

Mermaid: StateGraph API Class Diagram

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

The StateGraph builder pattern collects nodes and edges, then .compile() produces a CompiledGraph that can be invoked with state and config.

Defining State with TypedDict

python

from typing import TypedDict, List
from langgraph.graph import StateGraph

# Define the shared state schema
class AgentState(TypedDict):
    messages: List[str]   # conversation so far
    next_step: str        # which node to run next
    metadata: dict        # arbitrary metadata

# Instantiate a StateGraph with the schema
builder = StateGraph(AgentState)

ℹ️Note

StateGraph supports three schema definition approaches: TypedDict (lightweight, no validation), dataclass (mutable, Pythonic), and pydantic.BaseModel (validation, serialization). Choose BaseModel for production when you need runtime type checking.

Comparison: State Definition Approaches

Approach	Validation	Serialization	Boilerplate	Use Case
`TypedDict`	None	Manual	Minimal	Prototyping, simple agents
`dataclass`	None	Via dataclasses.asdict()	Low	Internal tools
`BaseModel`	Pydantic full validation	Built-in .dict()/.json()	Moderate	Production systems

Nodes and Edges

python

# Node: a function that receives state and returns updates
def node_a(state: AgentState) -> dict:
    print("--- Node A ---")
    return {"messages": state["messages"] + ["Hello from A"]}

def node_b(state: AgentState) -> dict:
    print("--- Node B ---")
    return {"messages": state["messages"] + ["Hello from B"]}

# Register nodes
builder.add_node("a", node_a)
builder.add_node("b", node_b)

# Add edges: a -> b
builder.add_edge("a", "b")

# Set entry and exit points
builder.set_entry_point("a")
builder.set_finish_point("b")

💡Tip

Node functions must return a dict (or None). The returned values are merged into the shared state via a shallow update. Keys not returned retain their previous value; this is how state persists between nodes.

Parallel Node Execution

python

def node_a(state: AgentState) -> dict:
    return {"messages": state["messages"] + ["A"]}

def node_b(state: AgentState) -> dict:
    return {"messages": state["messages"] + ["B"]}

def node_c(state: AgentState) -> dict:
    return {"messages": state["messages"] + ["C"]}

builder = StateGraph(AgentState)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_node("c", node_c)

# Fan-out: a triggers b and c simultaneously
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", END)
builder.add_edge("c", END)

When two edges leave the same node, both targets execute in parallel using Python threads. Each branch gets a copy of the state and writes merges at completion.

Error Handling in Nodes

python

import traceback

def safe_node(state: AgentState) -> dict:
    try:
        result = risky_operation(state["messages"][-1])
        return {"messages": state["messages"] + [result]}
    except Exception as e:
        # Log the error and continue with a fallback
        return {
            "messages": state["messages"] + [f"[ERROR]: {str(e)}"],
            "errors": state.get("errors", []) + [traceback.format_exc()]
        }

Wrap fallible node logic in try/except to prevent the entire graph from crashing. Store errors in state for downstream handling or human review.

Compiling and Running

python

# Compile the graph into a runnable object
app = builder.compile()

# Invoke with initial state
result = app.invoke({
    "messages": [],
    "next_step": "start",
    "metadata": {}
})

print(result["messages"])
# Output: ['Hello from A', 'Hello from B']

📌Important

The .compile() method freeze the graph definition. After compilation, you cannot add nodes or edges; you must rebuild the builder. For dynamic topologies, see Lesson 4 on dynamic graph updates.

Streaming Results

python

# Stream updates as each node completes
for event in app.stream({"messages": [], "next_step": "start", "metadata": {}}):
    for node_name, output in event.items():
        if node_name != "__end__":
            print(f"[{node_name}] -> {output}")
# Output:
# [a] -> {'messages': ['Hello from A']}
# [b] -> {'messages': ['Hello from A', 'Hello from B']}

Use .stream() instead of .invoke() when you want to observe intermediate states. Each emitted event is keyed by node name with the partial state update.

Mermaid: Basic State Graph

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

State flows down edges; each node can read and write to the shared AgentState.

Mermaid State Diagram: Node Lifecycle

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Each node transitions through read → execute → write. The graph orchestrates the sequence, passing state along edges.

Debugging Graphs with LangSmith

💡Tip

When your graph behaves unexpectedly, trace execution with LangSmith. Set LANGCHAIN_TRACING_V2=true and LANGCHAIN_API_KEY=your_key to get full trace logs showing every node's input, output, and timing.

bash

# Enable LangSmith tracing
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_PROJECT=my-agent

Practice Question

How to answer