n4nAI
Open console →
AI agents & automationTutorialAgentic RAG

Agentic RAG with LangGraph: a hands-on walkthrough

Step-by-step agentic RAG LangGraph tutorial: implement a retrieval agent with tool routing, reflection, and fallback using Python and LangChain.

n4n Team2 min read518 words

Audio narration

Coming soon — every post will get a voice note here.

Most RAG pipelines stop at a single vector lookup and a prompt stuffed with context. This agentic RAG LangGraph tutorial shows how to build a stateful agent that decides when to retrieve, when to call a tool, and when to answer from memory—without hand-rolling a brittle linear script.

Prerequisites

  • Python 3.10 or newer.
  • Install the dependencies:
pip install langgraph langchain langchain-openai langchain-community faiss-cpu python-dotenv
  • An API key for an OpenAI-compatible chat model. You can point LangChain at any gateway; for example, n4n.ai exposes one endpoint covering 240+ models with automatic fallback when a provider is rate-limited or degraded.

This agentic RAG LangGraph tutorial assumes you can run a Python script and have used LangChain expressions before.

Project setup

Load your key from .env and instantiate the model clients.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

load_dotenv()
os.environ["OPENAI_API_BASE"] = os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

Keep temperature=0 for routing decisions. Non-deterministic grading nodes waste debug time.

Build a retriever over fixed documents

Skip the crawler. Three facts are enough to demonstrate the loop.

from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document

seed_docs = [
    Document(page_content="The Fermi paradox asks why no aliens have contacted us despite the age of the galaxy."),
    Document(page_content="A black hole's event horizon is the boundary beyond which light cannot escape."),
    Document(page_content="Quantum entanglement correlates the states of particles regardless of distance."),
]
vectorstore = FAISS.from_documents(seed_docs, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

Sanity check the retriever:

hits = retriever.invoke("alien silence")
print([d.page_content[:30] for d in hits])

Expected output:

['The Fermi paradox asks why n', 'Quantum entanglement correl']

Define agent state

LangGraph flows a typed state object through nodes. We track the question, retrieved docs, message history, and a retry counter.

from typing import TypedDict, List, Annotated
from langchain_core.documents import Document
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    question: str
    documents: List[Document]
    messages: Annotated[List[BaseMessage], lambda x, y: x + y]
    attempts: int

The Annotated reducer appends messages instead of overwriting them—critical for multi-turn reasoning.

Node: retrieve

A thin wrapper around the retriever. It just fetches and increments the attempt counter.

def retrieve(state: AgentState) -> dict:
    docs = retriever.invoke(state["question"])
    return {"documents": docs, "attempts": state["attempts"] + 1}

Node: grade documents

Agentic retrieval needs a relevance gate. We ask the model to judge if the retrieved context can answer the question.

from langchain_core.messages import HumanMessage, SystemMessage

def grade(state: AgentState) -> dict:
    if not state["documents"]:
        return {"documents": []}
    ctx = "\n".join(d.page_content for d in state["documents"])
    judge_prompt = SystemMessage(content="Reply 'YES' if the context answers the question, else 'NO'.")
    user_prompt = HumanMessage(content=f"Question: {state['question']}\nContext: {ctx}")
    verdict = llm.invoke([judge_prompt, user_prompt]).content.strip()
    if verdict == "YES":
        return {"documents": state["documents"]}
    return {"documents": []}

Checkpoint the grader on a good match:

out = grade({"question": "alien contact", "documents": seed_docs[:1], "messages": [], "attempts": 1})
print(out)

Expected: {'documents': [Document(...)]} (the same doc returned, not cleared).

Node: generate answer

If graded relevant, synthesize. If not, the router will send us elsewhere.

def generate(state: AgentState) -> dict:
    ctx = "\n".join(d.page_content for d in state["documents"])
    prompt = HumanMessage(content=f"Answer concisely.\nQuestion: {state['question']}\nContext: {ctx}")
    answer = llm.invoke([prompt])
    return {"messages": [answer]}

Node: calculator tool (fallback)

When retrieval fails repeatedly, delegate arithmetic to a stub tool.

import re
from langchain_core.messages import AIMessage

def calculator(state: AgentState) -> dict:
    nums = [int(x) for x in re.findall(r"\d+", state["question"])]
    result = sum(nums)
    msg = AIMessage(content=f"Tool result: {result}")
    return {"messages": [msg], "documents": []}

Wire the graph

Conditional edges implement the agentic loop: retrieve → grade → (generate | retrieve again | calculator).

from langgraph.graph import StateGraph, END

def route(state: AgentState) -> str:
    if state["documents"]:
        return "generate"
    if state["attempts"] >= 3:
        return "calculator"
    return "retrieve"

workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade)
workflow.add_node("generate", generate)
workflow.add_node("calculator", calculator)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", route,
    {"generate": "generate", "retrieve": "retrieve", "calculator": "calculator"})
workflow.add_edge("generate", END)
workflow.add_edge("calculator", END)

app = workflow.compile()

The route function enforces a max of three retrieval attempts before bailing to the tool. That bound prevents infinite loops on empty indexes.

Run the agent

Invoke with a question the seed docs cover:

result = app.invoke({
    "question": "Why haven't aliens contacted us?",
    "documents": [],
    "messages": [],
    "attempts": 0
})
print(result["messages"][-1].content)

Expected output (model phrasing may vary):

The Fermi paradox highlights the contradiction between the high probability of alien life and the lack of contact.

Now a question that triggers the calculator after failed retrieval:

result = app.invoke({
    "question": "What is 12 plus 30?",
    "documents": [],
    "messages": [],
    "attempts": 0
})
print(result["messages"][-1].content)

Because the retriever returns unrelated docs, grade clears them, route hits the attempt limit, and calculator returns Tool result: 42.

Stream intermediate steps

Use stream to see each node fire—invaluable when the loop misbehaves.

for chunk in app.stream({"question": "alien life", "documents": [], "messages": [], "attempts": 0}):
    print(chunk)

You’ll see state deltas after retrieve, grade, and generate.

Common failure modes

Empty retriever results cause tight loops. Always bound attempts.

LLM grading is flaky. For production, use a smaller classifier or logprob thresholds instead of free-text “YES/NO”.

State shape drift breaks reducers. Keep the TypedDict strict and never mutate nested lists in place.

Where to take this next

The pattern above is a skeleton. Replace the stub calculator with a real ToolNode from langgraph.prebuilt, swap FAISS for a managed vector DB, and add a rewriter node that rephrases the question when grade fails. The state machine stays the same; you only swap node logic.

If you need to run this against multiple providers without rewriting the LLM client, an OpenAI-compatible gateway with per-token metering and cache-control forwarding keeps the ChatOpenAI lines untouched.

Keep the graph small. Debugging a 12-node agent is harder than debugging a 4-node one. This agentic RAG LangGraph tutorial deliberately stops at four.

Tagsagentic-raglanggraphtutorial

Written by

n4n Team

The team building n4n — a single OpenAI-compatible API in front of 240+ models, with automatic fallback, load balancing and pay-per-token metering.

More from n4n Team →
AI agents & automationHow to build an agentic RAG pipeline step by stepStep-by-step tutorial to build agentic RAG pipeline with routing, retrieval, and tool use. Runnable Python code and expected outputs included.AI agents & automationHow agentic RAG decides when to retrieveEngineer's guide to agentic RAG retrieval decision: building state machines, tool use, confidence scoring, and routing to cut unnecessary fetches.AI agents & automationHow much does agentic RAG cost per queryAnalyze the real token and model drivers behind agentic RAG cost per query, with concrete examples and engineering controls to keep expenses predictable.AI agents & automationHow to add query rewriting to an agentic RAG systemAdd query rewriting to an agentic RAG system with this hands-on guide: implement a rewriter, parallelize retrieval, and verify retrieval gains.AI agents & automationHow to evaluate an agentic RAG pipelineA practical guide to building an evaluation harness for agentic RAG pipelines: measure trajectories, retrieval quality, and judge scoring with runnable code.AI agents & automationHow to reduce hallucinations in agentic RAG systemsPractical steps to reduce hallucinations agentic RAG systems using citation constraints, output validation, critic loops, and resilient model routing.

All agentic rag posts →