The LangGraph Masterclass

A. The State ($S$)

The State is the "memory" of your application. In a standard Python script, variables are passed between functions locally and disappear when the function returns. In LangGraph, the State is a shared, persistent schema that all components read from and write to.

Intuition: Think of the State as a Shared Whiteboard in a meeting room. Every worker (Node) looks at the whiteboard, does some work, and writes their results back onto the whiteboard for the next person to see. We typically use TypedDict to define the structure of this whiteboard, ensuring strict typing and validation.

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

# The Schema
class AgentState(TypedDict):
    # 'messages': The conversational memory
    # 'add_messages': A special reducer function.
    # It tells LangGraph: "When a node returns a message, APPEND it to this list; 
    # do not overwrite the existing list."
    messages: Annotated[list, add_messages]
    
    # You can add custom keys for business logic
    # documents: list[str]
    # user_intent: str

B. Nodes & Edges

The graph is composed of two primary elements:

• Nodes ($V$): The "workers." These are standard Python functions. A node receives the current State, performs an action (e.g., calling an LLM, searching a database, formatting data), and returns an update to the State.
• Edges ($E$): The "traffic controllers." They define the control flow rules.
  • Normal Edges: "After Node A is done, always go to Node B."
  • Conditional Edges: "After Node A, check the State. If result is 'Good', go to End. If 'Bad', loop back to Node A."

from langgraph.graph import StateGraph, START, END

# 1. Define the Node Logic
def chatbot_node(state: AgentState):
    # Logic: Read state -> Call LLM -> Return Update
    return {"messages": ["Hello World"]}

# 2. Initialize the Graph
workflow = StateGraph(AgentState)

# 3. Add Nodes
workflow.add_node("chatbot", chatbot_node)

# 4. Add Edges
workflow.add_edge(START, "chatbot") # Entry Point
workflow.add_edge("chatbot", END)   # Exit Point

# 5. Compile (Freezes the structure)
app = workflow.compile()

An enterprise agent shouldn't just talk; it should do. If the user asks "What is the weather in Tokyo?", the agent should decide to use a weather API, not just hallucinate an answer.

This requires a Conditional Edge. We check the output of the LLM. If the LLM requested a "Tool Call," we route execution to a "Tools" node. If it just replied with text, we route to "END."

from langgraph.prebuilt import ToolNode, tools_condition

# 1. Bind tools to the model
# This "teaches" the LLM that it has these functions available.
llm_with_tools = llm.bind_tools([web_search, send_email])

def agent_node(state):
    # Invoking the LLM with the tool definitions
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 2. Build Graph with Loops
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", ToolNode([web_search]))

workflow.add_edge(START, "agent")

# 3. Conditional Edge
# 'tools_condition' is a prebuilt function.
# Logic: If the last message has 'tool_calls', go to 'tools'. Else go to END.
workflow.add_conditional_edges(
    "agent",
    tools_condition,
)

# 4. The Cycle (The most important line!)
# After the tool executes, we MUST go back to the agent so it can 
# read the tool output and formulate an answer.
workflow.add_edge("tools", "agent")

In a standard web server architecture, the application forgets the variable state as soon as the HTTP request ends. However, users expect long-running conversations (multi-turn chat) or background tasks that last for hours.

LangGraph solves this by decoupling logic from storage. By passing a checkpointer (like MemorySaver for testing, or PostgresSaver/RedisSaver for production), LangGraph automatically snapshots the state at every single step. You can pause a graph today and resume it next week.

from langgraph.checkpoint.memory import MemorySaver

# Initialize the storage backend
memory = MemorySaver()

# Compile the graph with the checkpointer
app = workflow.compile(checkpointer=memory)

# Usage: We MUST provide a 'thread_id' to identify the session
config = {"configurable": {"thread_id": "session_123"}}

# Turn 1: User introduces themselves
app.invoke({"messages": [("user", "My name is Gemini")]}, config=config)

# Turn 2: We ask a question. 
# Note: We do NOT need to pass the history back in. LangGraph loads it from memory.
result = app.invoke({"messages": [("user", "What is my name?")]}, config=config)
print(result["messages"][-1].content) 
# Output: "Your name is Gemini."

In enterprise settings, autonomy is dangerous. You cannot let an agent autonomously execute high-stakes actions like "Refund $5,000" or "Deploy to Production" without oversight.

LangGraph allows you to set breakpoints. You can tell the graph: "Run until you reach the 'Tools' node, then STOP and wait for permission." This transforms the agent from a "Black Box" into a "Co-pilot."

# 1. Compile with interrupt
# This tells LangGraph to pause RIGHT BEFORE entering the 'tools' node.
app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["tools"] 
)

# 2. Run Phase 1
# The agent will reason, generate the tool call, and then PAUSE.
app.invoke(inputs, config=config)

# 3. Inspection (Your Admin UI)
print("Graph paused. Waiting for approval...")

# 4. Resume Phase 2
# Passing 'None' tells LangGraph: "I have no new input, just continue 
# from the saved checkpoint where you left off."
app.invoke(None, config=config)

We discussed "Human-in-the-Loop" where you pause and approve. Time Travel is the advanced version where you modify the state before resuming. This is effectively an "Undo/Edit" button for AI.

The Scenario: The Agent decides to search for "LangGraph" (too vague). You pause it. Instead of just rejecting the action, you reach into the state, edit the tool argument to "LangGraph Python Documentation", and then resume. The agent executes the corrected action, believing it came up with it itself.

# 1. Inspect the paused state
snapshot = app.get_state(config)
last_msg = snapshot.values["messages"][-1]

print(last_msg.tool_calls[0]["args"]) 
# Output: {"query": "LangGraph"} (Too vague!)

# 2. Modify State (Inject new reality)
from copy import deepcopy
new_msg = deepcopy(last_msg)
new_msg.tool_calls[0]["args"]["query"] = "LangGraph Python Docs"

# update_state() overwrites the existing message history with our new version
app.update_state(config, {"messages": [new_msg]})

# 3. Resume with Corrected Brain
# The agent now executes the 'web_search' tool with the SPECIFIC query.
app.invoke(None, config=config)

The Concept: Fractal Agency. As your graph grows, putting 50 nodes (Research, Coding, QA, HR, Legal) into one flat graph becomes unmanageable ("Spaghetti Graph").

The solution is to encapsulate logic. You build a "Research Agent" graph and compile it. Then, you treat that entire compiled graph as a single Node inside a larger "Company" graph. This allows you to build modular, reusable agents that can be plugged into any workflow.

# 1. Build the Child Graph (The Specialist)
research_builder = StateGraph(ResearchState)
research_builder.add_node("search", search_node)
research_builder.add_edge(START, "search")
# Compile it into a Runnable
research_app = research_builder.compile()

# 2. Build the Parent Graph (The Supervisor)
parent_builder = StateGraph(CompanyState)
parent_builder.add_node("manager", manager_node)

# 3. Add the Child Graph as a Node
# We add the compiled 'research_app' directly as if it were a function.
parent_builder.add_node("research_team", research_app)

# 4. Connect them
parent_builder.add_edge("manager", "research_team")

Standard loops are sequential (Step A → Step B). Sometimes you need speed. LangGraph supports Fan-out (triggering multiple nodes simultaneously) and Fan-in (waiting for all of them to finish before proceeding).

Use Case: You want to write a blog post. Instead of writing Section 1, then Section 2, then Section 3, you hire 3 writers to write all sections in parallel. Then an "Editor" node collects and stitches them together.