LangGraph Chapter 2 — LangGraph Fundamentals: StateGraph, Nodes & Edges

Senior Architect Interview Series — LangGraph & Agentic AI

Navigation

← Chapter 1 — Why Agents? | Chapter 3 — AgentState →

2.0 What This Chapter Covers

Chapter 1 explained why agents and the ReAct pattern. This chapter explains how LangGraph implements it — the exact primitives that make your agent work.

Every LangGraph application is built from three things:

1. State — a shared data structure flowing through the graph
2. Nodes — Python functions that read and update state
3. Edges — connections that define the flow between nodes

Understanding these three primitives at a deep level is what lets you design and debug any LangGraph application.

2.1 The Core Mental Model

Think of a LangGraph application as a directed graph where:

         ┌─────────────────────────────────────────────────┐
         │                  StateGraph                     │
         │                                                 │
         │    [State]  ──────────────────────────────►     │
         │       │                                         │
         │       │    ┌──────────┐    ┌──────────┐        │
         │       └───►│  Node A  │───►│  Node B  │───►    │
         │            └──────────┘    └──────────┘        │
         │                  │                              │
         │                  └──── conditional ────►...     │
         │                                                 │
         └─────────────────────────────────────────────────┘

• State travels through the graph — each node can read it and write updates to it
• Nodes are pure Python functions: (state) → partial_state_update
• Edges are the connections — they can be fixed (A → B always) or conditional (A → B or C depending on state)

The graph is compiled once, then invoked with an initial state. It runs until it reaches END.

2.2 StateGraph — The Container

StateGraph is the LangGraph class that holds everything together.

from langgraph.graph import StateGraph, END

# 1. Create the graph, telling it the shape of the state
graph = StateGraph(AgentState)

# 2. Register nodes
graph.add_node("call_llm", call_llm)
graph.add_node("call_tools", call_tools)

# 3. Set entry point (which node runs first)
graph.set_entry_point("call_llm")

# 4. Define edges (flow)
graph.add_conditional_edges("call_llm", should_call_tools)
graph.add_edge("call_tools", "call_llm")

# 5. Compile into an executable
agent = graph.compile()

The StateGraph(AgentState) constructor takes your state schema — it knows what fields the state has, what types they are, and how to update them (via reducers, covered in Chapter 3).

Compilation

graph.compile() validates the graph structure (no dangling nodes, valid edges) and returns a CompiledGraph object that:

• Can be invoked: agent.invoke(initial_state)
• Can be streamed: agent.stream(initial_state)
• Can be introspected (draw the graph)
• Can be run asynchronously: await agent.ainvoke(initial_state)

2.3 Your Project's Graph — Visual Representation

# From agent/agent.py:
graph = StateGraph(AgentState)
graph.add_node("call_llm", call_llm)
graph.add_node("call_tools", call_tools)
graph.set_entry_point("call_llm")
graph.add_conditional_edges("call_llm", should_call_tools)
graph.add_edge("call_tools", "call_llm")
agent = graph.compile()

Visualized:

                    START
                      │
                      ▼
              ┌───────────────┐
              │   call_llm    │ ◄─────────┐
              └───────┬───────┘           │
                      │                   │
              should_call_tools()         │
                      │                   │
           ┌──────────┴──────────┐        │
           │                     │        │
     "call_tools"               END       │
           │                             │
           ▼                             │
    ┌──────────────┐                     │
    │  call_tools  │─────────────────────┘
    └──────────────┘
    (always back to call_llm)

This is the ReAct loop expressed as a graph. Two nodes, one conditional edge, one fixed edge, one cycle.

2.4 Nodes — The Building Blocks

A node is any Python callable with this signature:

def node_function(state: StateType) -> dict:
    # read from state
    # do work
    # return a PARTIAL update (only the fields you're changing)
    return {"field_name": new_value}

Critical rule: Nodes return partial state updates, not the full state. LangGraph merges your return value into the existing state using the reducer for each field.

Node Examples From Your Project

# call_llm node — reads messages, writes one new message
def call_llm(state: AgentState) -> AgentState:
    response = llm_with_tools.invoke(state["messages"])  # reads state
    return {"messages": [response]}                       # writes partial update
    # Returns only the new message — LangGraph handles appending via add_messages

# call_tools node — reads last message's tool_calls, writes ToolMessages
def call_tools(state: AgentState) -> AgentState:
    last_message = state["messages"][-1]   # read last AIMessage
    results = []
    for tool_call in last_message.tool_calls:
        tool_fn = tool_map[tool_call["name"]]
        output = tool_fn.invoke(tool_call["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=tool_call["id"]))
    return {"messages": results}            # returns list of ToolMessages to append

What You Can Do In a Node

Nodes are just Python functions — they can do anything:

def my_node(state: MyState) -> dict:
    # ✓ Call an LLM
    response = llm.invoke(state["messages"])
    
    # ✓ Call an API
    data = requests.get("https://api.example.com/data").json()
    
    # ✓ Query a database
    results = db.query(Item).all()
    
    # ✓ Execute code
    output = subprocess.run(["python", "script.py"], capture_output=True)
    
    # ✓ Sleep, retry, log
    time.sleep(1)
    logger.info("node executed")
    
    return {"field": new_value}

2.5 Edges — Controlling Flow

LangGraph has three types of edges:

Type 1 — Fixed Edge (add_edge)

Always goes from node A to node B, unconditionally.

graph.add_edge("call_tools", "call_llm")
# After call_tools runs → ALWAYS go to call_llm

Use when the next step is always the same regardless of state.

Type 2 — Conditional Edge (add_conditional_edges)

Routes to different nodes based on a routing function that reads the state.

graph.add_conditional_edges(
    "call_llm",          # from this node
    should_call_tools    # call this routing function
)

def should_call_tools(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "call_tools"   # → route to call_tools node
    return END                # → exit graph

The routing function must return a node name (string) or END. LangGraph uses the return value to decide where to go next.

Extended example with a mapping dict (explicit routing table):

graph.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {
        "rag_agent":  "rag_agent",    # return value → target node mapping
        "data_agent": "data_agent",
        "general":    "general_agent",
        END:          END
    }
)

The optional mapping dict makes routing explicit and readable.

Type 3 — Entry Point (set_entry_point)

Defines which node runs first when the graph is invoked.

graph.set_entry_point("call_llm")
# Equivalent to: graph.add_edge(START, "call_llm")

Special Constants: START and END

from langgraph.graph import START, END

# START → the special entry node (before any user-defined nodes)
# END   → the special exit node (graph stops here)

2.6 The END Sentinel

END is imported from langgraph.graph and signals that execution should stop.

from langgraph.graph import END

def should_call_tools(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "call_tools"
    return END   # ← tells LangGraph: we're done, return final state

When a conditional edge returns END, LangGraph:

1. Stops executing nodes
2. Returns the final state to the caller

The final state is what agent.invoke() returns — in your project, you then read final_state["messages"][-1].content to get the answer.

2.7 Graph Assembly Pattern — Complete Reference

Here is the canonical pattern for building any LangGraph agent:

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

# Step 1: Define state schema
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    # add more fields as needed: user_id, session_id, tool_results, etc.

# Step 2: Define node functions
def node_a(state: AgentState) -> dict:
    # ... do work ...
    return {"messages": [...]}   # partial update

def node_b(state: AgentState) -> dict:
    # ... do work ...
    return {"messages": [...]}

# Step 3: Define routing functions for conditional edges
def decide_next(state: AgentState) -> str:
    # read state, return node name or END
    if some_condition(state):
        return "node_b"
    return END

# Step 4: Assemble the graph
graph = StateGraph(AgentState)
graph.add_node("node_a", node_a)
graph.add_node("node_b", node_b)
graph.set_entry_point("node_a")
graph.add_conditional_edges("node_a", decide_next)
graph.add_edge("node_b", "node_a")   # loop back if needed

# Step 5: Compile
app = graph.compile()

# Step 6: Invoke
result = app.invoke({"messages": [HumanMessage("hello")]})

2.8 Invoking the Graph

Synchronous Invoke

# Returns the complete final state
final_state = agent.invoke({"messages": history})
answer = final_state["messages"][-1].content

Streaming (Token by Token)

# Streams state updates as each node completes
for chunk in agent.stream({"messages": history}):
    print(chunk)
# Output:
# {'call_llm': {'messages': [AIMessage(tool_calls=[...])]}}
# {'call_tools': {'messages': [ToolMessage("Agent Factory is...")]}}
# {'call_llm': {'messages': [AIMessage("Agent Factory is PepsiCo's...")]}}

Async Invoke

final_state = await agent.ainvoke({"messages": history})

With Config (Thread ID for Checkpointing)

config = {"configurable": {"thread_id": session_id}}
final_state = agent.invoke({"messages": history}, config=config)

2.9 Graph Introspection

LangGraph can describe its own structure:

# Print graph as ASCII diagram
print(agent.get_graph().draw_ascii())

# Draw as PNG (requires graphviz)
agent.get_graph().draw_png("agent_graph.png")

# List all nodes
print(agent.nodes)

# List all edges
print(agent.edges)

This is extremely useful for documentation and debugging in production.

2.10 Multiple Edges From One Node

A node can have multiple outgoing edges in a routing function:

def route_supervisor(state: SupervisorState) -> str:
    """Return which sub-agent to call."""
    route = state["route"]
    if route == "RAG":
        return "rag_agent"
    elif route == "DATA":
        return "data_agent"
    elif route == "GENERAL":
        return "general_agent"
    return END

graph.add_conditional_edges(
    "supervisor",
    route_supervisor,
    {
        "rag_agent":    "rag_agent",
        "data_agent":   "data_agent",
        "general_agent":"general_agent",
        END:            END
    }
)

This is exactly how your supervisor.py should plug into the multi-agent LangGraph system (Chapter 7).

2.11 Cycles (Loops) in the Graph

LangGraph explicitly supports cycles — a node can route back to a previous node. This is what makes the ReAct loop possible.

# This creates a cycle: call_llm → call_tools → call_llm → ...
graph.add_edge("call_tools", "call_llm")
graph.add_conditional_edges("call_llm", should_call_tools)
# should_call_tools can return "call_tools" (loop) or END (stop)

Without cycles, you'd need to hardcode the maximum number of tool calls. With cycles, the agent loops as many times as needed and stops when the LLM decides it has enough information.

Safety: Always ensure cycles have a path to END. Without it, the agent loops forever (infinite loop). The should_call_tools function always eventually returns END because the LLM will eventually produce a text answer.

2.12 LangGraph vs NetworkX vs Other Graph Libraries

LangGraph is NOT a generic graph library. It is purpose-built for LLM workflows.

2.13 Interview Q&A

Q: What is a StateGraph and how does it differ from a regular function pipeline?

A StateGraph is LangGraph's execution container — it manages a shared state object that flows through a directed graph of Python functions (nodes). Unlike a linear pipeline where each function calls the next, a StateGraph supports conditional branching and cycles. The state is shared and updated atomically between nodes using reducer functions. This makes it possible to implement loops (ReAct cycle), conditional routing (supervisor pattern), and complex state accumulation across multiple execution steps.

Q: What is the difference between add_edge and add_conditional_edges?

add_edge(A, B) is an unconditional connection — after node A runs, always go to node B. add_conditional_edges(A, routing_fn) calls the routing function with the current state after A runs, and routes to whichever node the function returns. In our agent, call_tools → call_llm is a fixed edge (always loop back after executing tools), while call_llm → should_call_tools is conditional (go to tools OR end depending on whether the LLM emitted tool calls).

Q: How does the graph know when to stop?

When a conditional routing function returns END (imported from langgraph.graph), LangGraph stops executing and returns the final state to the caller. In our agent, should_call_tools returns END when the last message is a plain AIMessage with no tool_calls — meaning the LLM has produced its final answer. The caller (run_agent) then reads final_state["messages"][-1].content to extract the answer string.

Q: What does graph.compile() do?

compile() validates the graph structure (checking for disconnected nodes, invalid edges, missing entry points), builds an internal execution plan, and returns a CompiledGraph object. The compiled graph is what you actually invoke. Compilation also allows you to attach optional components like a checkpointer for persistence (covered in the memory chapter) or an interrupt mechanism for human-in-the-loop. You compile once, then invoke as many times as needed.

Q: Can a LangGraph graph have cycles? How do you prevent infinite loops?

Yes, cycles are a first-class feature — they're what makes the ReAct loop possible. Infinite loops are prevented by the routing function: should_call_tools always has a path to END (when the LLM emits no tool calls). In production, you'd also add a safety limit: track iteration count in the state and force END after a maximum number of iterations (e.g., 10 tool calls). This prevents runaway agents caused by malformed tools or degenerate LLM behavior.

2.14 Key One-Liners to Memorize

"StateGraph = shared state + nodes (functions) + edges (flow control)."

"Nodes read state, do work, return a PARTIAL update — not the full state."

"add_edge: always go there. add_conditional_edges: routing function decides."

"END is not a node — it's a signal to stop execution and return final state."

"Cycles are features, not bugs — they're what makes the ReAct loop possible."

"compile() once, invoke() many times."

Next: Chapter 3 — AgentState & Reducers