LangGraph Tutorial: Build a Multi-Node AI Agent in Python

A RAG pipeline that routes queries through HyDE, CRAG, and multi-query decomposition cannot be expressed as a linear chain. It is a directed graph. LangGraph makes that graph explicit — with typed state, named nodes, and conditional edges that hold all routing logic.

The Valid Failure

ShopBot's Book 2 pipeline was a sequential chain:

chunks = retrieve(query)
answer = generate(query, chunks)

By Book 3, it needed to:

• Classify whether the query was vague
• If vague → HyDE
• If specific → standard retrieval
• Grade the retrieved chunks (CRAG)
• If PARTIAL → expand retrieval
• If IRRELEVANT → refuse
• If compound query → decompose first

Seven code paths through a single chat() function. Seven if/elif blocks nested inside each other. When a bug appeared — the wrong route being taken — no one could draw a diagram of what the code actually did.

LangGraph replaced the nested if/elif with a graph that could be drawn on a whiteboard.

The Core Concepts

State — a typed dictionary shared by all nodes. Every node reads from state and returns an update to state.

Nodes — functions that take state and return state. One node = one operation.

Edges — connections between nodes. Regular edges are unconditional. Conditional edges call a function to decide which node to go to next.

Step 1: Define the State Schema

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END

class ShopBotState(TypedDict):
    query: str
    is_vague: bool
    sub_queries: list[str]
    chunks: list[dict]
    crag_verdict: str   # RELEVANT | PARTIAL | IRRELEVANT
    answer: str

Every field the pipeline might need lives in the state. Nodes cannot pass data to each other directly — only through state.

Step 2: Write the Nodes

def classify_node(state: ShopBotState) -> ShopBotState:
    """Classify query as vague or specific."""
    state["is_vague"] = is_vague_query(state["query"])
    return state

def hyde_node(state: ShopBotState) -> ShopBotState:
    """For vague queries: generate hypothetical, retrieve against it."""
    hypothetical = generate_hypothetical(state["query"])
    state["chunks"] = retrieve_by_text(hypothetical)
    return state

def retrieve_node(state: ShopBotState) -> ShopBotState:
    """For specific queries: standard retrieval."""
    state["chunks"] = retrieve(state["query"])
    return state

def crag_grade_node(state: ShopBotState) -> ShopBotState:
    """Grade the top retrieved chunk."""
    if not state["chunks"]:
        state["crag_verdict"] = "IRRELEVANT"
    else:
        verdict = grade_chunk(state["query"], state["chunks"][0]["text"])
        state["crag_verdict"] = verdict
    return state

def expand_retrieve_node(state: ShopBotState) -> ShopBotState:
    """For PARTIAL verdict: add more chunks."""
    extra = retrieve(state["query"], top_k=5)
    existing_texts = {c["text"] for c in state["chunks"]}
    new_chunks = [c for c in extra if c["text"] not in existing_texts]
    state["chunks"] = state["chunks"] + new_chunks
    return state

def generate_node(state: ShopBotState) -> ShopBotState:
    """Generate answer from chunks."""
    state["answer"] = generate(state["query"], state["chunks"])
    return state

def refuse_node(state: ShopBotState) -> ShopBotState:
    """No evidence — refuse rather than hallucinate."""
    state["answer"] = "I don't have enough information about that in our catalog."
    return state

Step 3: Define the Routing Functions

def route_by_vagueness(state: ShopBotState) -> str:
    return "hyde" if state["is_vague"] else "retrieve"

def route_by_crag_verdict(state: ShopBotState) -> str:
    verdict = state["crag_verdict"]
    if verdict == "RELEVANT":
        return "generate"
    elif verdict == "PARTIAL":
        return "expand_retrieve"
    else:
        return "refuse"

Routing functions read state and return the name of the next node as a string. All routing logic lives in these functions — not scattered through the node code.

Step 4: Build the Graph

graph = StateGraph(ShopBotState)

# Add nodes
graph.add_node("classify", classify_node)
graph.add_node("hyde", hyde_node)
graph.add_node("retrieve", retrieve_node)
graph.add_node("crag_grade", crag_grade_node)
graph.add_node("expand_retrieve", expand_retrieve_node)
graph.add_node("generate", generate_node)
graph.add_node("refuse", refuse_node)

# Entry point
graph.set_entry_point("classify")

# Conditional routing after classify
graph.add_conditional_edges("classify", route_by_vagueness, {
    "hyde": "hyde",
    "retrieve": "retrieve",
})

# Both retrieval paths go to grading
graph.add_edge("hyde", "crag_grade")
graph.add_edge("retrieve", "crag_grade")

# Conditional routing after grading
graph.add_conditional_edges("crag_grade", route_by_crag_verdict, {
    "generate": "generate",
    "expand_retrieve": "expand_retrieve",
    "refuse": "refuse",
})

# Expand retrieval goes to generate (not back to grading)
graph.add_edge("expand_retrieve", "generate")

# Terminal nodes
graph.add_edge("generate", END)
graph.add_edge("refuse", END)

# Compile
app = graph.compile()

Step 5: Run the Graph

def chat(query: str) -> str:
    initial_state: ShopBotState = {
        "query": query,
        "is_vague": False,
        "sub_queries": [],
        "chunks": [],
        "crag_verdict": "",
        "answer": "",
    }
    final_state = app.invoke(initial_state)
    return final_state["answer"]

What the Graph Looks Like

          [classify]
         /          \
      [hyde]      [retrieve]
         \          /
        [crag_grade]
       /      |      \
[expand]  [generate]  [refuse]
   |
[generate]

Every routing decision is in the conditional edges. Every operation is in a node. The graph can be drawn on one whiteboard.

Inspecting the Final State for Debugging

The final state records every decision the pipeline made:

final_state = app.invoke(initial_state)

print(f"Was vague: {final_state['is_vague']}")
print(f"CRAG verdict: {final_state['crag_verdict']}")
print(f"Chunks used: {len(final_state['chunks'])}")
print(f"Answer: {final_state['answer']}")

When a query produces the wrong answer, the state tells you exactly which route was taken and what evidence was used — without re-running the query.

The Valid Knowledge

• One node = one operation: nodes that do two things are two nodes waiting to happen. Small nodes make routing failures easy to identify.
• All routing logic in edge functions: a node that calls if is_vague: retrieve_hyde() is a routing decision inside a node — this hides the graph structure. Put it in a conditional edge.
• State is the single source of truth: nodes communicate only through state. If a node needs a value, it reads it from state. If a node produces a value, it writes it to state.
• The final state is your debugging tool: when an answer is wrong, read the final state. The routing path and the evidence used are both there.
• LangGraph vs LangChain chains: chains are linear and implicit. LangGraph is graphical and explicit. For pipelines with conditional routing, LangGraph is the correct tool — chains cannot express branches.