Agentic RAG with LangGraph

Basic RAG (Retrieval-Augmented Generation) is a straight line: take the question, retrieve documents, generate an answer. Agentic RAG turns that line into a loop. The system checks what it retrieved, decides whether it is good enough, and can rewrite the query and try again. LangGraph is the LangChain-family library that lets you build this kind of loop as a graph of steps.

Overview: how agentic RAG works

A plain RAG chain has a weakness: it trusts the first retrieval. If the user asks “How do I get my money back?” but your documents say “refund”, the vector search may return weak matches, and the model will answer from poor context or invent something. A chain cannot notice this or recover.

Agentic RAG adds decisions to the pipeline. Typical decisions are:

• Grade the retrieved documents: are they relevant to the question?
• Rewrite the query if they are not, and retrieve again.
• Give up politely after a few attempts instead of looping forever.
• Route the question: answer directly, search a database, or search the web.

LangGraph models this as a state graph. You define three things:

• State: a shared dictionary-like object (often a TypedDict) holding the question, documents, answer and counters.
• Nodes: ordinary Python functions. Each reads the state and returns a partial update, for example {'docs': [...]}.
• Edges: connections between nodes. A normal edge always goes to the same next node. A conditional edge calls a routing function that looks at the state and returns the name of the next node.

Loops are allowed: an edge can point back to an earlier node. This is what separates LangGraph from a simple LCEL chain, which is a directed acyclic pipeline.

What happens internally

When you call compile(), LangGraph checks the graph and returns a runnable object. When you call invoke(), it starts at START, runs a node, merges the returned dictionary into the state, then follows the edges to the next node. It stops when it reaches END. By default, keys you return simply overwrite the old values for those keys; keys you do not return stay unchanged. LangGraph also has a recursion limit (a maximum number of steps) that raises an error if a graph loops too long. Do not rely on it as your retry logic; add your own counter.

Key concepts and syntax

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

class State(TypedDict):
    question: str

def my_node(state: State) -> dict:
    return {'question': state['question'].strip()}

def my_router(state: State) -> str:
    return 'other_node'

builder = StateGraph(State)
builder.add_node('my_node', my_node)
builder.add_edge(START, 'my_node')
builder.add_conditional_edges('my_node', my_router, ['other_node', END])
app = builder.compile()

Examples

Example 1: an offline retry loop (no API key)

First we build the full control flow with no LLM, so you can run it anywhere. Retrieval is a tiny keyword lookup, grading is a rule, and rewriting is a synonym replacement. The shape of the graph is exactly what a real system uses.

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

DOCS = {
    'refund': 'Refunds are issued within 7 days of a return request.',
    'shipping': 'Standard shipping takes 3 to 5 business days.',
}

class State(TypedDict):
    question: str
    docs: List[str]
    answer: str
    attempts: int

def retrieve(state: State) -> dict:
    words = state['question'].lower().replace('?', '').split()
    found = 
    return {'docs': found}

def grade(state: State) -> str:
    if state['docs']:
        return 'generate'
    if state['attempts'] >= 1:
        return 'give_up'
    return 'rewrite'

def rewrite(state: State) -> dict:
    q = state['question'].lower().replace('money back', 'refund')
    return {'question': q, 'attempts': state['attempts'] + 1}

def generate(state: State) -> dict:
    return {'answer': 'Based on the docs: ' + ' '.join(state['docs'])}

def give_up(state: State) -> dict:
    return {'answer': 'I could not find this in the knowledge base.'}

builder = StateGraph(State)
builder.add_node('retrieve', retrieve)
builder.add_node('rewrite', rewrite)
builder.add_node('generate', generate)
builder.add_node('give_up', give_up)
builder.add_edge(START, 'retrieve')
builder.add_conditional_edges('retrieve', grade, ['generate', 'rewrite', 'give_up'])
builder.add_edge('rewrite', 'retrieve')
builder.add_edge('generate', END)
builder.add_edge('give_up', END)
app = builder.compile()

for q in ['How do I get my money back?', 'What is your phone number?']:
    result = app.invoke({'question': q, 'docs': [], 'answer': '', 'attempts': 0})
    print('Q:', q)
    print('A:', result['answer'])
    print('Rewrites:', result['attempts'])

Output:

Q: How do I get my money back?
A: Based on the docs: Refunds are issued within 7 days of a return request.
Rewrites: 1
Q: What is your phone number?
A: I could not find this in the knowledge base.
Rewrites: 1

For the first question, the first retrieval finds nothing, because “money back” contains no key word. The router sends the flow to rewrite, which changes the text to “refund” and adds 1 to attempts. The edge from rewrite goes back to retrieve, which now succeeds, so the router picks generate. The second question fails twice, and because attempts is already 1, the router chooses give_up. Note that the router is a pure function of the state: it makes no changes, it only decides.

Example 2: the same graph with an LLM grader and rewriter

In a real system, the grader and rewriter are LLM calls and retrieval uses a vector store. This version needs an API key, read from an environment variable. Model names change often, so check your provider’s documentation for a current one. To follow along offline, replace OpenAIEmbeddings() with DeterministicFakeEmbedding(size=256) from langchain_core.embeddings; the retrieval will not be meaningful, but the graph will run. You would also need to replace the three LLM chains with stubs.

import os
from typing import TypedDict, List
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.vectorstores import InMemoryVectorStore
from langgraph.graph import StateGraph, START, END

assert os.environ.get('OPENAI_API_KEY'), 'Set OPENAI_API_KEY first'

llm = ChatOpenAI(model='gpt-4o-mini', temperature=0)
store = InMemoryVectorStore(OpenAIEmbeddings())
store.add_documents([
    Document(page_content='Refunds are issued within 7 days of a return request.'),
    Document(page_content='Standard shipping takes 3 to 5 business days.'),
    Document(page_content='Support is available Monday to Friday, 9am to 6pm IST.'),
])
retriever = store.as_retriever(search_kwargs={'k': 3})

class Grade(BaseModel):
    relevant: bool = Field(description='True if the text helps answer the question')

grader = ChatPromptTemplate.from_template(
    'Question: {question} Text: {text} Is the text relevant to the question?'
) | llm.with_structured_output(Grade)

rewriter = ChatPromptTemplate.from_template(
    'Rewrite this as a better search query. Return only the query: {question}'
) | llm | StrOutputParser()

answerer = ChatPromptTemplate.from_template(
    'Answer using only this context: {context} Question: {question}'
) | llm | StrOutputParser()

class State(TypedDict):
    question: str
    query: str
    docs: List[str]
    answer: str
    attempts: int

def retrieve(state: State) -> dict:
    found = retriever.invoke(state['query'])
    return {'docs': [d.page_content for d in found]}

def grade_docs(state: State) -> dict:
    keep = []
    for text in state['docs']:
        verdict = grader.invoke({'question': state['question'], 'text': text})
        if verdict.relevant:
            keep.append(text)
    return {'docs': keep}

def route(state: State) -> str:
    if state['docs']:
        return 'generate'
    if state['attempts'] >= 2:
        return 'give_up'
    return 'rewrite'

def rewrite(state: State) -> dict:
    new_query = rewriter.invoke({'question': state['question']})
    return {'query': new_query, 'attempts': state['attempts'] + 1}

def generate(state: State) -> dict:
    context = ' '.join(state['docs'])
    return {'answer': answerer.invoke({'context': context, 'question': state['question']})}

def give_up(state: State) -> dict:
    return {'answer': 'Sorry, I could not find a reliable answer.'}

builder = StateGraph(State)
builder.add_node('retrieve', retrieve)
builder.add_node('grade_docs', grade_docs)
builder.add_node('rewrite', rewrite)
builder.add_node('generate', generate)
builder.add_node('give_up', give_up)
builder.add_edge(START, 'retrieve')
builder.add_edge('retrieve', 'grade_docs')
builder.add_conditional_edges('grade_docs', route, ['generate', 'rewrite', 'give_up'])
builder.add_edge('rewrite', 'retrieve')
builder.add_edge('generate', END)
builder.add_edge('give_up', END)
app = builder.compile()

question = 'How long until I get my money back?'
result = app.invoke({
    'question': question, 'query': question, 'docs': [], 'answer': '', 'attempts': 0,
})
print(result['answer'])

Output (illustrative, the wording depends on the model):

Refunds are issued within 7 days of a return request.

Two design choices matter here. First, the original question is kept unchanged while the rewritten text goes into query. The grader and answerer always judge against what the user really asked, not a drifting rewrite. Second, grading is its own node and routing is a separate function, so you can test each piece alone.

Example 3: watching the loop with streaming

To see which node runs, use stream with stream_mode='updates'. Each item is a dictionary from node name to the update that node returned. Add this to Example 1:

inputs = {'question': 'How do I get my money back?', 'docs': [], 'answer': '', 'attempts': 0}
for step in app.stream(inputs, stream_mode='updates'):
    for node_name in step:
        print('ran:', node_name)

Output:

ran: retrieve
ran: rewrite
ran: retrieve
ran: generate

This trace is the best debugging tool for agentic RAG. If a question takes five trips through rewrite, you see it immediately.

Under the hood: one run, step by step

1. invoke receives the initial state and moves from START to retrieve.
2. retrieve returns {'docs': []}. LangGraph merges this into the state.
3. The conditional edge calls grade(state). It sees no docs and attempts == 0, so it returns 'rewrite'.
4. rewrite returns a new question and attempts = 1. The normal edge sends control back to retrieve.
5. The second retrieval finds a document, the router returns 'generate', and generate writes the answer.
6. The edge to END finishes the run, and invoke returns the final state.

Common Mistakes

1. A loop with no exit counter

def route(state):
    if state['docs']:
        return 'generate'
    return 'rewrite'   # always retries, even when no answer exists

If the knowledge base truly has no answer, this graph loops until LangGraph’s recursion limit raises an error, wasting tokens and money. Fix it by storing an attempts counter in the state, increasing it in the rewrite node, and routing to a give_up node when it reaches a limit, as in the examples above.

2. Returning the wrong thing from a node

def generate(state):
    state['answer'] = 'Hello'   # mutates in place
    return 'Hello'              # not a dictionary of updates

A node must return a dictionary of state updates, such as {'answer': 'Hello'}. Returning a plain string is an error, and relying on in-place mutation is fragile. Always return the update.

3. Overwriting the original question

If rewrite replaces question and the grader uses the rewritten text, the system may drift away from the user’s intent. Keep the original and the working query in separate keys, as in Example 2.

Best Practices

• Always cap retries (two or three is usually enough) and provide an honest “I don’t know” exit.
• Keep routing functions pure and cheap: they read state and return a node name, nothing else.
• Grade documents with a structured output schema (a Pydantic model) so the decision is a clean boolean, not free text you must parse.
• Use a small, fast model for grading and rewriting; save the larger model for the final answer.
• Grading each document costs one LLM call. For large k, grade in a batch, or only grade the top few results.
• Test nodes individually with plain function calls and fake data before connecting them.
• Stream updates during development and log node names in production so you can see how often rewrites happen.
• Do not add agentic steps unless you need them. Each loop adds latency and cost, so measure whether it improves answer quality on your own test questions.

Practice Exercises

1. Extend Example 1 with a second synonym: rewrite “delivery” to “shipping”. Test with “How long is delivery?” and confirm the trace shows one rewrite.
2. Change the retry limit in Example 1 to 2 and make rewrite try a different strategy each time (first the synonym, then removing question words such as “how” and “do”). Hint: use attempts to choose the strategy.
3. Add a route_question step at the start that sends greetings like “hello” straight to a chat node, skipping retrieval entirely. Hint: use a conditional edge from START.

Summary

• Agentic RAG adds decisions and loops to retrieval: grade, rewrite, retry, or give up.
• LangGraph models this with a shared state, nodes that return partial updates, and edges.
• Conditional edges use a routing function that returns the next node’s name.
• Always keep an attempts counter so loops end cleanly; do not rely on the recursion limit.
• Keep the original question separate from the rewritten query.
• Use stream(..., stream_mode='updates') to see which nodes ran.
• Build and test the control flow offline first, then swap in LLM-based graders and rewriters.