Agentic RAG for Dummies

Build a modular Agentic RAG system with LangGraph, conversation memory, and human-in-the-loop query clarification

Overview • How It Works • LLM Providers • Implementation • Installation & Usage • Troubleshooting

If you like this project, a star ⭐️ would mean a lot :)

Overview

This repository demonstrates how to build an Agentic RAG (Retrieval-Augmented Generation) system using LangGraph with minimal code. Most RAG tutorials show basic concepts but lack guidance on building modular, agent-driven systems — this project bridges that gap by providing both learning materials and an extensible architecture.

What's inside

| Feature | Description | |---|---| | 🗂️ Hierarchical Indexing | Search small chunks for precision, retrieve large Parent chunks for context | | 🧠 Conversation Memory | Maintains context across questions for natural dialogue | | ❓ Query Clarification | Rewrites ambiguous queries or pauses to ask the user for details | | 🤖 Agent Orchestration | LangGraph coordinates the full retrieval and reasoning workflow | | 🔀 Multi-Agent Map-Reduce | Decomposes complex queries into parallel sub-queries | | ✅ Self-Correction | Re-queries automatically if initial results are insufficient | | 🗜️ Context Compression | Keeps working memory lean across long retrieval loops | | 🔍 Observability | Track LLM calls, tool usage, and graph execution with Langfuse | | 📊 Evaluation | Evaluate retrieval and answer quality with RAGAS metrics |

🎯 Two Ways to Use This Repo

1️⃣ Learning Path: Interactive Notebook

Step-by-step tutorial perfect for understanding core concepts. Start here if you're new to Agentic RAG or want to experiment quickly.

2️⃣ Building Path: Modular Project

Flexible architecture where each component can be independently adapted — LLM provider, embedding model, PDF converter, and agent workflow. The runnable app is Ollama-first, and it can be adapted to any chat model provider supported by LangChain. Examples are included for Anthropic, OpenAI, and Google.

See Modular Architecture and Installation & Usage to get started.

How It Works

Document Preparation: Hierarchical Indexing

Before queries can be processed, documents are split twice for optimal retrieval:

• Parent Chunks: Bounded large sections based on Markdown headers (H1, H2, H3)
• Child Chunks: Small, fixed-size pieces derived from parents

Optional: 🐿️ Chunky is an open-source toolkit for reliable RAG pipelines: convert PDFs to Markdown, clean documents, inspect chunks, compare chunking strategies, and enrich metadata before building the vector store.

This combines the precision of small chunks for search with the contextual richness of large chunks for answer generation.

Query Processing: Four-Stage Intelligent Workflow

User Query → Conversation Summary → Query Rewriting → Query Clarification →
Parallel Agent Reasoning → Aggregation → Final Response

Stage 1 — Conversation Understanding: Maintains a rolling summary and recent conversation history to preserve continuity without indefinitely increasing context size.

Stage 2 — Query Clarification: Resolves references ("How do I update it?" → "How do I update SQL?"), splits multi-part questions into focused sub-queries, detects unclear inputs, and rewrites queries for optimal retrieval. Pauses for human input when clarification is needed.

Stage 3 — Intelligent Retrieval (Multi-Agent Map-Reduce): Spawns parallel agent subgraphs — one per sub-query. Each agent searches child chunks, fetches parent chunks for context, self-corrects if results are insufficient, compresses context to avoid redundant fetches, and falls back gracefully if the search budget is exhausted.

Example: "What is JavaScript? What is Python?" → 2 parallel agents execute simultaneously.

Stage 4 — Response Generation: Aggregates all agent responses into a single coherent answer.

LLM Provider Configuration

This system is provider-agnostic: the runnable app uses Ollama by default, and the chat model initialization can be adapted to any LLM provider available in LangChain. The examples below cover the most common options, but the same pattern applies to any other supported provider.

Note: Model names change frequently. Always check the official documentation for the latest available models and their identifiers before deploying.

Ollama (Local)

# Install Ollama from https://ollama.com
ollama pull granite4.1:8b

from langchain_ollama import ChatOllama
llm = ChatOllama(model="granite4.1:8b", temperature=0, seed=42)

⚠️ For reliable tool calling and instruction following, prefer models 8B+. Smaller models may ignore retrieval instructions or hallucinate. See Troubleshooting.

Cloud Providers

pip install -qU langchain-openai

from langchain_openai import ChatOpenAI import os

os.environ["OPENAIAPIKEY"] = "your-api-key-here" llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

pip install -qU langchain-anthropic

from langchain_anthropic import ChatAnthropic import os

os.environ["ANTHROPICAPIKEY"] = "your-api-key-here" llm = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)

pip install -qU langchain-google-genai

import os from langchaingooglegenai import ChatGoogleGenerativeAI

os.environ["GOOGLEAPIKEY"] = "your-api-key-here" llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0)

Implementation

Additional details, extended explanations, and Langfuse observability are available in the notebook and full project. The companion evaluation notebook scores the final answers and the actual child/parent tool outputs used by the agent with direct RAGAS metric calls.

| Step | Description | |------|-------------| | 1 | Initial Setup and Configuration | | 2 | Configure Vector Database | | 3 | PDFs to Markdown | | 4 | Hierarchical Document Indexing | | 5 | Define Agent Tools | | 6 | Define System Prompts | | 7 | Define State and Data Models | | 8 | Agent Configuration | | 9 | Build Graph Node and Edge Functions | | 10 | Build the LangGraph Graphs | | 11 | Create Chat Interface |

Step 1: Initial Setup and Configuration

Define paths and initialize core components.

import os
from pathlib import Path
from langchain_huggingface import HuggingFaceEmbeddings
from langchainqdrant.fastembedsparse import FastEmbedSparse
from qdrant_client import QdrantClient
DOCS_DIR = "docs"  # Directory containing your pdf files MARKDOWNDIR = "markdowndocs" # Directory containing the pdfs converted to markdown PARENTSTOREPATH = "parent_store"  # Directory for parent chunk JSON files CHILDCOLLECTION = "documentchild_chunks" DEFAULTRETRIEVALK = 7 CHILDCHUNKSEPARATOR = "\n\n<CHILDCHUNKBOUNDARY>\n\n"
os.makedirs(DOCSDIR, existok=True) os.makedirs(MARKDOWNDIR, existok=True) os.makedirs(PARENTSTOREPATH, exist_ok=True)
from langchain_ollama import ChatOllama llm = ChatOllama(model="granite4.1:8b", temperature=0, seed=42)
denseembeddings = HuggingFaceEmbeddings(modelname="Qwen/Qwen3-Embedding-0.6B") sparseembeddings = FastEmbedSparse(modelname="Qdrant/bm25")
client = QdrantClient(path="qdrant_db")

Step 2: Configure Vector Database

Set up Qdrant to store child chunks with hybrid search capabilities.

from qdrant_client.http import models as qmodels
from langchain_qdrant import QdrantVectorStore
from langchain_qdrant.qdrant import RetrievalMode
embeddingdimension = len(denseembeddings.embed_query("test"))
def ensurecollection(collectionname):     if not client.collectionexists(collectionname):         client.create_collection(             collectionname=collectionname,             vectors_config=qmodels.VectorParams(                 size=embedding_dimension,                 distance=qmodels.Distance.COSINE             ),             sparsevectorsconfig={                 "sparse": qmodels.SparseVectorParams()             },         )

Step 3: PDFs to Markdown

Convert the PDFs to Markdown. For more details about other techniques use this companion notebook.

import os
import pymupdf.layout
import pymupdf4llm
from pathlib import Path
import glob
os.environ["TOKENIZERS_PARALLELISM"] = "false"
def pdftomarkdown(pdfpath, outputdir):     doc = pymupdf.open(pdf_path)     md = pymupdf4llm.tomarkdown(doc, header=False, footer=False, pageseparators=True, ignoreimages=True, writeimages=False, image_path=None)     md_cleaned = md.encode('utf-8', errors='surrogatepass').decode('utf-8', errors='ignore')     outputpath = Path(outputdir) / Path(doc.name).stem     Path(outputpath).withsuffix(".md").writebytes(mdcleaned.encode('utf-8'))
def pdfstomarkdowns(path_pattern, overwrite: bool = False):     outputdir = Path(MARKDOWNDIR)     outputdir.mkdir(parents=True, existok=True)
for pdfpath in map(Path, glob.glob(pathpattern)):         mdpath = (outputdir / pdfpath.stem).withsuffix(".md")         if overwrite or not md_path.exists():             pdftomarkdown(pdfpath, outputdir)
pdfstomarkdowns(f"{DOCS_DIR}/*.pdf")

Step 4: Hierarchical Document Indexing

Process documents with the Parent/Child splitting strategy.

import os import glob import json from pathlib import Path from langchaintextsplitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

def merge_metadata(target, source, prepend=False):
    for key, value in source.items():
        if key not in target:
            target[key] = value
        else:
            first, second = (value, target[key]) if prepend else (target[key], value)
            values = [
                item.strip()
                for raw in (first, second)
                for item in str(raw).split(" -> ")
                if item.strip()
            ]
            target[key] = " -> ".join(dict.fromkeys(values))
def mergesmallparents(chunks, min_size):     if not chunks:         return []
merged, current = [], None
for chunk in chunks:         if current is None:             current = chunk         else:             current.pagecontent += "\n\n" + chunk.pagecontent             merge_metadata(current.metadata, chunk.metadata)
if len(current.pagecontent) >= minsize:             merged.append(current)             current = None
if current:         if merged:             merged[-1].pagecontent += "\n\n" + current.pagecontent             merge_metadata(merged[-1].metadata, current.metadata)         else:             merged.append(current)
return merged
def splitlargeparents(chunks, max_size, overlap):     split_chunks = []
for chunk in chunks:         if len(chunk.pagecontent) <= maxsize:             split_chunks.append(chunk)         else:             large_splitter = RecursiveCharacterTextSplitter(                 chunksize=maxsize,                 chunk_overlap=overlap             )             subchunks = largesplitter.split_documents([chunk])             splitchunks.extend(subchunks)
return split_chunks
def rebalancepair(first, second, minsize, max_size):     combined = first.pagecontent.rstrip() + "\n\n" + second.pagecontent.lstrip()     lower = max(1, len(combined) - max_size)     upper = min(max_size, len(combined) - 1)     if len(combined) >= 2 * min_size:         lower = max(lower, min_size)         upper = min(upper, len(combined) - min_size)     preferred = min(max(len(combined) // 2, lower), upper)
split_at = preferred     for separator in ("\n\n", "\n", " "):         before = combined.rfind(separator, lower, preferred + 1)         after = combined.find(separator, preferred, upper + 1)         if before >= lower:             split_at = before             break         if after != -1:             split_at = after             break
lefttext = combined[:splitat].rstrip()     righttext = combined[splitat:].lstrip()     if len(combined) >= 2 * minsize and (len(lefttext) < minsize or len(righttext) < min_size):         split_at = preferred         lefttext, righttext = combined[:splitat], combined[splitat:]     if not lefttext or not righttext:         return first, second
metadata = dict(first.metadata)     merge_metadata(metadata, second.metadata)     first.pagecontent, first.metadata = lefttext, dict(metadata)     second.pagecontent, second.metadata = righttext, dict(metadata)     return first, second
def cleansmallchunks(chunks, minsize, maxsize):     cleaned = []
for i, chunk in enumerate(chunks):         if len(chunk.pagecontent) < minsize:             if cleaned and len(cleaned[-1].pagecontent) + 2 + len(chunk.pagecontent) <= max_size:                 cleaned[-1].pagecontent += "\n\n" + chunk.pagecontent                 merge_metadata(cleaned[-1].metadata, chunk.metadata)             elif i < len(chunks) - 1 and len(chunk.pagecontent) + 2 + len(chunks[i + 1].pagecontent) <= max_size:                 chunks[i + 1].pagecontent = chunk.pagecontent + "\n\n" + chunks[i + 1].page_content                 merge_metadata(chunks[i + 1].metadata, chunk.metadata, prepend=True)             else:                 cleaned.append(chunk)         else:             cleaned.append(chunk)
for i, chunk in enumerate(cleaned):         if len(chunk.pagecontent) >= minsize or len(cleaned) == 1:             continue         if i < len(cleaned) - 1:             cleaned[i], cleaned[i + 1] = rebalancepair(chunk, cleaned[i + 1], minsize, max_size)         else:             cleaned[i - 1], cleaned[i] = rebalancepair(cleaned[i - 1], chunk, minsize, max_size)
return cleaned

if client.collectionexists(CHILDCOLLECTION):
    client.deletecollection(CHILDCOLLECTION)
    ensurecollection(CHILDCOLLECTION)
else:
    ensurecollection(CHILDCOLLECTION)
childvectorstore = QdrantVectorStore(     client=client,     collectionname=CHILDCOLLECTION,     embedding=dense_embeddings,     sparseembedding=sparseembeddings,     retrieval_mode=RetrievalMode.HYBRID,     sparsevectorname="sparse" )
def index_documents():     headerstosplit_on = [("#", "H1"), ("##", "H2"), ("###", "H3")]     parentsplitter = MarkdownHeaderTextSplitter(headerstospliton=headerstospliton, stripheaders=False)     childchunksize = 500     childchunkoverlap = 100     minparentsize = 2000     maxparentsize = 4000     if minparentsize <= 0 or maxparentsize < minparentsize:         raise ValueError("Parent chunk sizes must be positive and minparentsize <= maxparentsize.")     if not 0 <= childchunkoverlap < childchunksize:         raise ValueError("childchunkoverlap must be smaller than childchunksize.")     child_splitter = RecursiveCharacterTextSplitter(         chunksize=childchunk_size,         chunkoverlap=childchunk_overlap,     )
allparentpairs, allchildchunks = [], []     mdfiles = sorted(glob.glob(os.path.join(MARKDOWNDIR, "*.md")))
if not md_files:         return
for docpathstr in md_files:         docpath = Path(docpath_str)         try:             with open(doc_path, "r", encoding="utf-8") as f:                 md_text = f.read()         except Exception as e:             continue
parentchunks = parentsplitter.splittext(mdtext)         mergedparents = mergesmallparents(parentchunks, minparentsize)         splitparents = splitlargeparents(mergedparents, maxparentsize, childchunkoverlap)         cleanedparents = cleansmallchunks(splitparents, minparentsize, maxparentsize)         if any(len(chunk.pagecontent) > maxparentsize for chunk in cleanedparents):             raise ValueError("Parent chunking produced an oversized chunk.")
for i, pchunk in enumerate(cleanedparents):             parentid = f"{docpath.stem}_p{i}"             pchunk.metadata.update({"source": docpath.stem + ".pdf", "parentid": parentid})             allparentpairs.append((parentid, pchunk))             children = childsplitter.splitdocuments([p_chunk])             allchildchunks.extend(children)
if not allchildchunks:         return
try:         childvectorstore.adddocuments(allchild_chunks)     except Exception as e:         return
for item in os.listdir(PARENTSTOREPATH):         os.remove(os.path.join(PARENTSTOREPATH, item))
for parentid, doc in allparent_pairs:         docdict = {"pagecontent": doc.page_content, "metadata": doc.metadata}         filepath = os.path.join(PARENTSTOREPATH, f"{parent_id}.json")         with open(filepath, "w", encoding="utf-8") as f:             json.dump(docdict, f, ensureascii=False, indent=2)
index_documents()

Step 5: Define Agent Tools

Create the retrieval tools the agent will use.

import json
from typing import List
from langchain_core.tools import tool
RETRIEVALSCORETHRESHOLD = 0.4
@tool def searchchildchunks(query: str, limit: int = DEFAULTRETRIEVALK) -> str:     """Search document excerpts for evidence related to the user question.
Use this as the first retrieval step. Results include parent IDs, file     names, and short child-chunk excerpts. If excerpts are relevant but too     fragmented to answer confidently, call retrieveparentchunks with the     returned parent_id.
Args:         query: Focused search query with concrete keywords from the question.         limit: Maximum number of child chunks to return.     """     try:         results = childvectorstore.similarity_search(             query,             k=limit,             scorethreshold=RETRIEVALSCORE_THRESHOLD,         )         if not results:             return "NORELEVANTCHUNKS"
return CHILDCHUNKSEPARATOR.join([             f"Parent ID: {doc.metadata.get('parent_id', '')}\n"             f"File Name: {doc.metadata.get('source', '')}\n"             f"Content: {doc.page_content.strip()}"             for doc in results         ])
except Exception as e:         return f"RETRIEVAL_ERROR: {str(e)}"
@tool def retrieveparentchunks(parent_id: str) -> str:     """Retrieve the full parent chunk for a relevant child search result.
Use this only after searchchildchunks returns a relevant parent_id and     the child excerpt needs more surrounding context. Do not call this for     parent IDs already available in compressed context.          Args:         parentid: Parent chunk ID returned by searchchild_chunks.     """     filename = parentid if parentid.lower().endswith(".json") else f"{parentid}.json"     path = os.path.join(PARENTSTOREPATH, file_name)
if not os.path.exists(path):         return "NOPARENTDOCUMENT"
with open(path, "r", encoding="utf-8") as f:         data = json.load(f)
return (         f"Parent ID: {parent_id}\n"         f"File Name: {data.get('metadata', {}).get('source', 'unknown')}\n"         f"Content: {data.get('page_content', '').strip()}"     )
llmwithtools = llm.bindtools([searchchildchunks, retrieveparent_chunks])

Step 6: Define System Prompts

Define the system prompts for conversation summarization, query rewriting, agent orchestration, context compression, fallback response, and answer aggregation.

def getconversationsummary_prompt() -> str:
    return """## Role
You are a compact memory manager for a retrieval-augmented chat assistant.
Context
The input contains an existing rolling summary plus older user/assistant messages that will be removed from raw chat history.
Instructions
Merge the existing summary with the new older messages.
Preserve context needed for future follow-up questions: topics, user preferences, important facts, unresolved questions, and referenced source file names.
Discard greetings, tool calls, tool outputs, formatting chatter, duplicate details, and resolved misunderstandings.
Keep the summary compact: 30-70 words unless more detail is essential.

Output
Return exactly one merged summary and nothing else.
Do not include labels such as "Updated summary:", "Previous summary:", or "New messages:".
Do not include both old and new summaries.
If there is no meaningful context, return an empty string.
"""

def getrewritequery_prompt() -> str:
    return """## Role
You are a query rewriting specialist for document retrieval in a RAG system.
Instructions
Rewrite the current query so it is clear, self-contained, and useful for retrieval.
Use the conversation summary and recent conversation only to resolve vague follow-ups that refer to prior context.
When an unresolved query and one or more user clarifications are provided, combine all of them into one self-contained retrieval query.
If the query is a follow-up, integrate only the minimal context needed to make it self-contained.
Preserve product names, file names, versions, acronyms, numbers, and technical terms exactly.
If the user asks about a named topic, product, file, acronym, term, or concept, treat the question as clear even if it is new.
Standalone named terms, acronyms, or concepts are valid retrieval queries; do not require prior conversation context.
Split only truly separate information needs, with a maximum of 3 rewritten questions.

Clarification Boundary
Mark the query unclear only when it depends on an unresolved reference such as "it", "that", "this file", or "the previous one".
Do not mark a query unclear because the topic was not mentioned earlier.
Do not ask the user whether a new acronym or term is a typo; preserve it and search for it.
Constraints
Do not add facts, expand acronyms, invent context, or broaden the user's meaning.
"""

def getorchestratorprompt() -> str:
    return """## Role
You are a document-grounded research assistant for an agentic RAG system. Your job is to answer using retrieved document evidence, not general knowledge.
Available Context
Current user question
Optional compressed context from prior retrieval steps
Tools for searching child chunks and loading full parent chunks

Tool Guidance
Search documents before answering unless compressed context already contains enough evidence.
Use 'searchchildchunks' for missing or uncovered parts of the question.
If searched or retrieved context is not useful, use the tools again with a different, simpler query or a more relevant parent chunk.
Continue tool use until the available evidence is enough, tools stop adding useful information, or the operation limit is reached.
Do not repeat search queries or parent IDs listed in compressed context.
Do not retrieve the same parent ID twice.

Response Framework
Check compressed context for already-known evidence and already-used searches or parents.
Search for missing evidence.
Retrieve parent chunks only when child excerpts are relevant but too fragmented.
Answer using the exact terms and scope in the retrieved evidence.
If evidence is incomplete, state the specific gap.

Output
Start directly with the substantive answer. Do not start with generic headings such as "Answer", "Final answer", or "Response".
Provide the direct answer plus the key supporting details from retrieved evidence; avoid one-sentence fragments unless only one fact is available.
Do not mention internal tool calls or reasoning.
When sources exist, end with a Sources section in exactly this format:
  Sources:
  - filename.ext
Put each source filename on its own bullet line. Never write sources inline, such as "Sources: filename.pdf".
Do not invent or infer source filenames.
Strip descriptions after file names, including text in parentheses.
"""

def getfallbackresponse_prompt() -> str:
    return """## Role
You are a constrained evidence synthesizer for a retrieval-augmented assistant after the research loop reached its limit.
Available Context
Compressed Research Context from earlier retrieval steps
Retrieved Data from current tool outputs

Instructions
Use only explicit facts from the provided context.
Start directly with the substantive answer. Do not start with generic headings such as "Answer", "Final answer", or "Response".
Prefer current Retrieved Data over compressed context if they conflict.
If the answer is incomplete, mention only the missing parts that matter to the user query.
Do not describe the retrieval process, limits, or internal reasoning.
Be concise: answer in 1-3 short paragraphs or up to 5 bullets unless the user asks for detail.
Provide the direct answer plus the key supporting details from retrieved evidence; avoid one-sentence fragments unless only one fact is available.
End with a Sources section only when actual source file names are explicitly present in the context.
Use exactly this format:
  Sources:
  - filename.ext
Put each source filename on its own bullet line. Never write sources inline, such as "Sources: filename.pdf".
Include only bare file names with extensions such as .pdf, .docx, .txt, or .md.
Do not invent or infer source filenames.
"""

def getcontextcompression_prompt() -> str:
    return """## Role
You are a research context compressor for an agentic RAG system.
Instructions
Keep only facts relevant to answering the user question.
Preserve exact names, figures, versions, technical terms, configuration details, and source file names.
Remove duplicates, tool chatter, search query wording, parent IDs, chunk IDs, and other internal identifiers.
Organize findings by source file. Each source section heading must be the real filename found in retrieved data.
Add a Gaps section only for missing information relevant to the question.
Target 400-600 words. If there is too much content, keep the most answer-critical facts.

Output
Return only Markdown in this structure:
Research Context Summary
Focus
[Brief technical restatement of the question]
Structured Findings
For each source file, add a level-3 heading with its real filename and bullet the directly relevant facts below it.
Gaps
Missing or incomplete aspects
"""

def getaggregationprompt() -> str:
    return """## Role
You are a final-answer synthesizer for a retrieval-augmented assistant.
Instructions
Use only information present in the retrieved answers.
Start directly with the substantive answer. Do not start with generic headings such as "Answer", "Final answer", or "Response".
Preserve important names, numbers, versions, examples, and definitions.
Do not expand acronyms or interpret terms unless the sources do it.
If answers conflict, mention the conflict plainly.
Be concise: answer in 1-3 short paragraphs or up to 5 bullets unless the user asks for detail.
Provide the direct answer plus the key supporting details from retrieved evidence; avoid one-sentence fragments unless only one fact is available.
End with a Sources section only when actual source file names are explicitly present in the retrieved answers.
Use exactly this format:
  Sources:
  - filename.ext
Put each source filename on its own bullet line. Never write sources inline, such as "Sources: filename.pdf".
Include only bare file names with extensions such as .pdf, .docx, .txt, or .md.
Do not invent or infer source filenames.
If no useful information is available, say: "I couldn't find any information to answer your question in the available sources."
"""

Step 7: Define State and Data Models

Create the state structure for conversation tracking and agent execution.

from langgraph.graph import MessagesState
from pydantic import BaseModel, Field
from typing import List, Annotated, Set
import operator
def accumulateorreset(existing: List[dict], new: List[dict]) -> List[dict]:     if new and any(item.get('reset') for item in new):         return []     return existing + new
def set_union(a: Set[str], b: Set[str]) -> Set[str]:     return a | b
def append_unique(existing: List[str], new: List[str]) -> List[str]:     return list(dict.fromkeys(existing + new))
class State(MessagesState):     questionIsClear: bool = False     conversation_summary: str = ""     originalQuery: str = ""     pendingQuery: str = ""     pendingClarifications: List[str] = []     rewrittenQuestions: List[str] = []     agentanswers: Annotated[List[dict], accumulateor_reset] = []
class AgentState(MessagesState):     toolcallcount: Annotated[int, operator.add] = 0     iteration_count: Annotated[int, operator.add] = 0     question: str = ""     question_index: int = 0     context_summary: str = ""     retrievalkeys: Annotated[Set[str], setunion] = set()     retrievedcontexts: Annotated[List[str], appendunique] = []     final_answer: str = ""     agent_answers: List[dict] = []
class QueryAnalysis(BaseModel):     is_clear: bool = Field(description="Indicates if the user's question is clear and answerable.")     questions: List[str] = Field(description="List of rewritten, self-contained questions.")     clarification_needed: str = Field(description="Explanation if the question is unclear.")

Step 8: Agent Configuration

Hard limits on tool calls and iterations prevent infinite loops. Token counting (via tiktoken) drives context compression decisions.

import tiktoken
from functools import lru_cache
MAXTOOLCALLS = 8       # Maximum tool calls per agent run MAX_ITERATIONS = 10      # Maximum agent loop iterations BASETOKENTHRESHOLD = 2000     # Initial token threshold for compression TOKENGROWTHFACTOR = 0.9       # Multiplier applied after each compression
@lru_cache(maxsize=1) def gettoken_encoding():     try:         return tiktoken.encodingformodel("gpt-4")     except Exception:         try:             return tiktoken.getencoding("cl100kbase")         except Exception:             return None
def estimatecontexttokens(messages: list) -> int:     contents = [         str(msg.content)         for msg in messages         if hasattr(msg, "content") and msg.content     ]     encoding = gettoken_encoding()     if encoding is None:         return sum(max(1, len(content) // 4) for content in contents)     return sum(len(encoding.encode(content)) for content in contents)

Step 9: Build Graph Node and Edge Functions

Create the processing nodes and edges for the LangGraph workflow.

Main Graph Nodes & Edges

from langgraph.types import Send, Command
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, RemoveMessage, ToolMessage
from typing import Literal, Set
MAINHISTORYMESSAGESTOKEEP = 4 if MAINHISTORYMESSAGESTOKEEP < 2:     raise ValueError("MAINHISTORYMESSAGESTOKEEP must be at least 2.") PREANSWERHISTORYMESSAGESTOKEEP = max(MAINHISTORYMESSAGESTO_KEEP - 1, 0)
def isplainconversationmessage(msg) -> bool:     return (         isinstance(msg, (HumanMessage, AIMessage))         and not getattr(msg, "tool_calls", None)         and not getattr(msg, "name", None)     )
def nameinternal_message(message, name):     """Tag a subgraph-only message so it is not treated as chat history."""     return message.model_copy(update={"name": name})
def retrievalcontexts(messages) -> list[str]:     contexts = []     ignored_prefixes = (         "NORELEVANTCHUNKS",         "NOPARENTDOCUMENT",         "RETRIEVAL_ERROR:",         "PARENTRETRIEVALERROR:",     )     for message in messages:         if not isinstance(message, ToolMessage):             continue         content = str(message.content).strip()         if content and not content.startswith(ignored_prefixes):             parts = content.split(CHILDCHUNKSEPARATOR) if message.name == "searchchildchunks" else [content]             contexts.extend(part for part in parts if part)     return list(dict.fromkeys(contexts))
def formatconversation(messages) -> str:     lines = []     for msg in messages:         role = "User" if isinstance(msg, HumanMessage) else "Assistant"         lines.append(f"{role}: {msg.content}")     return "\n".join(lines)
def removemessagesnotin(messages, keep_ids):     removals = []     for msg in messages:         msg_id = getattr(msg, "id", None)         if isinstance(msg, SystemMessage) or not msg_id:             continue         if msgid not in keepids:             removals.append(RemoveMessage(id=msg_id))     return removals
def recentconversation(messages, pending_query="") -> list:     """Return recent context before the current user message."""     plainmessages = [msg for msg in messages if isplainconversation_message(msg)]     recentmessages = plainmessages[:-1]
if pending_query:         for index in range(len(recent_messages) - 1, -1, -1):             msg = recent_messages[index]             if isinstance(msg, HumanMessage) and str(msg.content).strip() == pending_query:                 return recent_messages[:index]
return recent_messages
def summarize_history(state: State):     messages = state.get("messages", [])     updates = {"agent_answers": [{"reset": True}]}
if not messages:         return updates
plainmessages = [msg for msg in messages if isplainconversation_message(msg)]     keepcount = PREANSWERHISTORYMESSAGESTOKEEP     messagestosummarize = plainmessages[:-keepcount] if len(plainmessages) > keepcount else []     keepids = {getattr(msg, "id", None) for msg in plainmessages[-keep_count:]}     keep_ids.discard(None)
removals = removemessagesnotin(messages, keep_ids)     if removals:         updates["messages"] = removals
if not messagestosummarize:         return updates
existingsummary = state.get("conversationsummary", "").strip()     conversation = "Existing summary:\n"     conversation += f"{existing_summary or '(none)'}\n\n"     conversation += "New messages to merge into the summary:\n"     conversation += formatconversation(messagestosummarize)
summary_response = llm.invoke([         SystemMessage(content=getconversationsummary_prompt()),         HumanMessage(content=conversation),     ])     updates["conversationsummary"] = summaryresponse.content.strip()     return updates
def rewrite_query(state: State):     last_message = state["messages"][-1]     currentquery = str(lastmessage.content).strip()     conversationsummary = state.get("conversationsummary", "").strip()     pending_query = state.get("pendingQuery", "").strip()     pending_clarifications = state.get("pendingClarifications", [])     recentmessages = recentconversation(state["messages"], pendingquery)
context_parts = []     if conversation_summary:         contextparts.append(f"Conversation Summary:\n{conversationsummary}")     if recent_messages:         contextparts.append(f"Recent Conversation:\n{formatconversation(recentmessages)}")
if pending_query:         clarifications = [*pendingclarifications, currentquery]         clarification_text = "\n".join(             f"{index}. {value}" for index, value in enumerate(clarifications, start=1)         )         context_parts.append(             f"Unresolved User Query:\n{pending_query}\n\n"             f"User Clarifications:\n{clarification_text}"         )         originalquery = f"{pendingquery}\nClarifications:\n{clarification_text}"     else:         clarifications = []         contextparts.append(f"User Query:\n{currentquery}")         originalquery = currentquery
contextsection = "\n\n".join(contextparts)     llmwithstructure = llm.withstructuredoutput(QueryAnalysis)     response = llmwithstructure.invoke([SystemMessage(content=getrewritequeryprompt()), HumanMessage(content=contextsection)])     clarificationmessageupdate = (         [nameinternalmessage(lastmessage, "clarification_response")]         if pending_query else []     )
if response.questions and response.is_clear:         return {             "questionIsClear": True,             "originalQuery": original_query,             "pendingQuery": "",             "pendingClarifications": [],             "rewrittenQuestions": response.questions,             "messages": clarificationmessageupdate,         }
clarification = response.clarificationneeded if response.clarificationneeded and len(response.clarification_needed.strip()) > 10 else "I need more information to understand your question."     return {         "questionIsClear": False,         "originalQuery": "",         "pendingQuery": pendingquery or currentquery,         "pendingClarifications": clarifications,         "rewrittenQuestions": [],         "messages": clarificationmessageupdate + [             AIMessage(content=clarification, name="clarification")         ],     }
def request_clarification(state: State):     return {}
def routeafterrewrite(state: State) -> Literal["request_clarification", "agent"]:     if not state.get("questionIsClear", False):         return "request_clarification"     else:         return [                 Send("agent", {"question": query, "question_index": idx, "messages": []})                 for idx, query in enumerate(state["rewrittenQuestions"])             ]
def aggregate_answers(state: State):     messages = state.get("messages", [])     plainmessages = [msg for msg in messages if isplainconversation_message(msg)]     keepids = {getattr(msg, "id", None) for msg in plainmessages[-PREANSWERHISTORYMESSAGESTO_KEEP:]}     keep_ids.discard(None)     removals = removemessagesnotin(messages, keep_ids)
if not state.get("agent_answers"):         return {"messages": removals + [AIMessage(c)]}
sortedanswers = sorted(state["agentanswers"], key=lambda x: x["index"])
formatted_answers = ""     for i, ans in enumerate(sorted_answers, start=1):         formatted_answers += (f"\nRetrieved response {i}:\n"f"{ans['answer']}\n")
usermessage = HumanMessage(content=f"""Original user question: {state["originalQuery"]}\nRetrieved answers:{formattedanswers}""")     synthesisresponse = llm.invoke([SystemMessage(content=getaggregationprompt()), usermessage])     return {"messages": removals + [AIMessage(content=synthesis_response.content)]}

Agent Subgraph Nodes & Edges

def orchestrator(state: AgentState):
    contextsummary = state.get("contextsummary", "").strip()
    sysmsg = SystemMessage(content=getorchestrator_prompt())
    summary_injection = (
        [HumanMessage(content=f"[COMPRESSED CONTEXT FROM PRIOR RESEARCH]\n\n{context_summary}")]
        if context_summary else []
    )
    if not state.get("messages"):
        humanmsg = HumanMessage(content=state["question"], name="agentquestion")
        force_search = HumanMessage(c)
        response = llmwithtools.invoke([sysmsg] + summaryinjection + [humanmsg, forcesearch])
        response = nameinternalmessage(response, "agentresponse")
        return {"messages": [humanmsg, response], "toolcallcount": len(response.toolcalls or []), "iteration_count": 1}
response = llmwithtools.invoke([sysmsg] + summaryinjection + state["messages"])     response = nameinternalmessage(response, "agentresponse")     toolcalls = response.toolcalls if hasattr(response, "tool_calls") else []     return {"messages": [response], "toolcallcount": len(toolcalls) if toolcalls else 0, "iteration_count": 1}
def routeafterorchestratorcall(state: AgentState) -> Literal["tools", "fallbackresponse", "collect_answer"]:     iteration = state.get("iteration_count", 0)     toolcount = state.get("toolcall_count", 0)
last_message = state["messages"][-1]     toolcalls = getattr(lastmessage, "tool_calls", None) or []
if not tool_calls:         return "collect_answer"
# Accept a final answer at the iteration boundary, but do not execute     # tool calls that would exceed the configured research budget.     if iteration >= MAXITERATIONS or toolcount > MAXTOOLCALLS:         return "fallback_response"          return "tools"
def fallback_response(state: AgentState):     seen = set()     unique_contents = []     for m in state["messages"]:         if isinstance(m, ToolMessage) and m.content not in seen:             unique_contents.append(m.content)             seen.add(m.content)
contextsummary = state.get("contextsummary", "").strip()
context_parts = []     if context_summary:         contextparts.append(f"## Compressed Research Context (from prior iterations)\n\n{contextsummary}")     if unique_contents:         context_parts.append(             "## Retrieved Data (current iteration)\n\n" +             "\n\n".join(f"--- DATA SOURCE {i} ---\n{content}" for i, content in enumerate(unique_contents, 1))         )
contexttext = "\n\n".join(contextparts) if context_parts else "No data was retrieved from the documents."
prompt_content = (         f"USER QUERY: {state.get('question')}\n\n"         f"{context_text}\n\n"         f"INSTRUCTION:\nProvide the best possible answer using only the data above."     )     response = llm.invoke([SystemMessage(content=getfallbackresponseprompt()), HumanMessage(content=promptcontent)])     response = nameinternalmessage(response, "agentresponse")     return {"messages": [response]}
def shouldcompresscontext(state: AgentState) -> Command[Literal["compress_context", "orchestrator"]]:     messages = state["messages"]
new_ids: Set[str] = set()     for msg in reversed(messages):         if isinstance(msg, AIMessage) and getattr(msg, "tool_calls", None):             for tc in msg.tool_calls:                 if tc["name"] == "retrieveparentchunks":                     raw = tc["args"].get("parent_id") or tc["args"].get("id") or tc["args"].get("ids") or []                     if isinstance(raw, str):                         new_ids.add(f"parent::{raw}")                     else:                         new_ids.update(f"parent::{r}" for r in raw)
elif tc["name"] == "searchchildchunks":                     query = tc["args"].get("query", "")                     if query:                         new_ids.add(f"search::{query}")             break
updatedids = state.get("retrievalkeys", set()) | new_ids
currenttokenmessages = estimatecontexttokens(messages)     currenttokensummary = estimatecontexttokens([HumanMessage(content=state.get("context_summary", ""))])     currenttokens = currenttokenmessages + currenttoken_summary
maxallowed = BASETOKENTHRESHOLD + int(currenttokensummary * TOKENGROWTH_FACTOR)
goto = "compresscontext" if currenttokens > max_allowed else "orchestrator"     return Command(         update={             "retrievalkeys": updatedids,             "retrievedcontexts": retrieval_contexts(messages),         },         goto=goto,     )
def compress_context(state: AgentState):     messages = state["messages"]     existingsummary = state.get("contextsummary", "").strip()
if not messages:         return {}
conversation_text = f"USER QUESTION:\n{state.get('question')}\n\nConversation to compress:\n\n"     if existing_summary:         conversationtext += f"[PRIOR COMPRESSED CONTEXT]\n{existingsummary}\n\n"
for msg in messages[1:]:         if isinstance(msg, AIMessage):             toolcallsinfo = ""             if getattr(msg, "tool_calls", None):                 calls = ", ".join(f"{tc['name']}({tc['args']})" for tc in msg.tool_calls)                 toolcallsinfo = f" | Tool calls: {calls}"             conversationtext += f"[ASSISTANT{toolcalls_info}]\n{msg.content or '(tool call only)'}\n\n"         elif isinstance(msg, ToolMessage):             tool_name = getattr(msg, "name", "tool")             conversationtext += f"[TOOL RESULT — {toolname}]\n{msg.content}\n\n"
summaryresponse = llm.invoke([SystemMessage(content=getcontextcompressionprompt()), HumanMessage(content=conversation_text)])     newsummary = summaryresponse.content
retrievedids: Set[str] = state.get("retrievalkeys", set())     if retrieved_ids:         parentids = sorted(r for r in retrievedids if r.startswith("parent::"))         searchqueries = sorted(r.replace("search::", "") for r in retrievedids if r.startswith("search::"))
block = "\n\n---\nAlready executed (do NOT repeat):\n"         if parent_ids:             block += "Parent chunks retrieved:\n" + "\n".join(f"- {p.replace('parent::', '')}" for p in parent_ids) + "\n"         if search_queries:             block += "Search queries already run:\n" + "\n".join(f"- {q}" for q in search_queries) + "\n"         new_summary += block
return {"contextsummary": newsummary, "messages": [RemoveMessage(id=m.id) for m in messages[1:]]}
def collect_answer(state: AgentState):     last_message = state["messages"][-1]     isvalid = isinstance(lastmessage, AIMessage) and lastmessage.content and not lastmessage.tool_calls     answer = lastmessage.content if isvalid else "Unable to generate an answer."     return {         "final_answer": answer,         "agent_answers": [{             "index": state["question_index"],             "question": state["question"],             "answer": answer,             "contexts": state.get("retrieved_contexts", []),         }]     }

Why this architecture?

• Summarization maintains conversational context without overwhelming the LLM
• Query rewriting ensures search queries are precise and unambiguous, using context intelligently
• Human-in-the-loop catches unclear queries before wasting any retrieval resources
• Parallel execution via Send API spawns independent agent subgraphs for each sub-question simultaneously
• Context compression keeps the agent's working memory lean across long retrieval loops, preventing redundant fetches
• Fallback response ensures graceful degradation — the agent always returns something useful even when the budget runs out
• Answer collection & aggregation extracts clean final answers from agents and aggregates them into a single coherent response

Step 10: Build the LangGraph Graphs

Assemble the complete workflow graph with conversation memory and multi-agent architecture.

from langgraph.graph import START, END, StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent_builder = StateGraph(AgentState) agentbuilder.addnode(orchestrator) agentbuilder.addnode("tools", ToolNode([searchchildchunks, retrieveparentchunks])) agentbuilder.addnode(compress_context) agentbuilder.addnode(fallback_response) agentbuilder.addnode(shouldcompresscontext) agentbuilder.addnode(collect_answer)
agentbuilder.addedge(START, "orchestrator") agentbuilder.addconditionaledges("orchestrator", routeafterorchestratorcall, {"tools": "tools", "fallbackresponse": "fallbackresponse", "collectanswer": "collectanswer"}) agentbuilder.addedge("tools", "shouldcompresscontext") agentbuilder.addedge("compress_context", "orchestrator") agentbuilder.addedge("fallbackresponse", "collectanswer") agentbuilder.addedge("collect_answer", END) agentsubgraph = agentbuilder.compile()
graph_builder = StateGraph(State) graphbuilder.addnode(summarize_history) graphbuilder.addnode(rewrite_query) graphbuilder.addnode(request_clarification) graphbuilder.addnode("agent", agent_subgraph) graphbuilder.addnode(aggregate_answers)
graphbuilder.addedge(START, "summarize_history") graphbuilder.addedge("summarizehistory", "rewritequery") graphbuilder.addconditionaledges("rewritequery", routeafterrewrite) graphbuilder.addedge("requestclarification", "rewritequery") graphbuilder.addedge(["agent"], "aggregate_answers") graphbuilder.addedge("aggregate_answers", END)
agentgraph = graphbuilder.compile(checkpointer=checkpointer, interruptbefore=["requestclarification"])

Graph architecture explained:

The architecture flow diagram can be viewed here.

Agent Subgraph (processes individual questions):

• START → orchestrator (invoke LLM with tools)
• orchestrator → tools (if tool calls needed) OR fallbackresponse (if budget exhausted) OR collectanswer (if done)
• tools → shouldcompresscontext (check token budget)
• shouldcompresscontext → compress_context (if threshold exceeded) OR orchestrator (otherwise)
• compress_context → orchestrator (resume with compressed memory)
• fallbackresponse → collectanswer (package best-effort answer)
• collect_answer → END (clean final answer with index)

• START → summarize_history (roll older chat into summary and keep only recent exchanges)
• summarizehistory → rewritequery (rewrite query with context, check clarity)
• rewritequery → requestclarification (if unclear) OR spawn parallel agent subgraphs via Send (if clear)
• requestclarification → rewritequery (after user provides clarification)
• All agent subgraphs → aggregate_answers (merge all responses)
• aggregate_answers → END (return final synthesized answer)

Step 11: Create Chat Interface

Build a Gradio interface with conversation persistence and human-in-the-loop support. For a complete end-to-end pipeline Gradio interface, including document ingestion, please refer to project/README.md.

Note: The notebook and full project stream the final aggregated answer while showing query analysis and tool activity in separate collapsible blocks. Raw orchestrator, compression, and fallback model output remains internal. The example below is intentionally minimal.

import gradio as gr
import uuid
def createthreadid():     """Generate a unique thread ID for each conversation"""     return {"configurable": {"threadid": str(uuid.uuid4())}, "recursionlimit": 50}
def clear_session():     """Clear thread for new conversation"""     global config     agentgraph.checkpointer.deletethread(config["configurable"]["thread_id"])     config = createthreadid()
def chat(message, history):     currentstate = agentgraph.get_state(config)          if current_state.next:         agentgraph.updatestate(config,{"messages": [HumanMessage(content=message.strip())]})         result = agent_graph.invoke(None, config)     else:         result = agent_graph.invoke({"messages": [HumanMessage(content=message.strip())]}, config)          return result['messages'][-1].content
config = createthreadid()
with gr.Blocks() as demo:     chatbot = gr.Chatbot()     chatbot.clear(clear_session)     gr.ChatInterface(fn=chat, chatbot=chatbot)
demo.launch(theme=gr.themes.Citrus())

You're done! You now have a fully functional Agentic RAG system with conversation memory, hierarchical indexing, and human-in-the-loop query clarification.

Modular Architecture

The app (project/ folder) is organized into modular components — each independently swappable without breaking the system.

📂 Project Structure

project/
├── app.py                    # Main Gradio application entry point
├── config.py                 # Configuration hub (models, chunk sizes, providers)
├── core/                     # RAG system orchestration
├── db/                       # Vector DB and parent chunk storage
├── rag_agent/                # LangGraph workflow (nodes, edges, prompts, tools)
└── ui/                       # Gradio interface

Key customization points: LLM provider, embedding model, chunking strategy, agent workflow, and system prompts — all configurable via config.py or their respective modules.

Full documentation in project/README.md.

Installation & Usage

Sample pdf files can be found here: javascript, blockchain, fortinetTMP.pdf).

Option 1: Quickstart Notebook (Recommended for Testing)

Google Colab: The notebook clones the repository and installs its requirements. Upload PDFs to docs/. Standard hosted Colab does not provide Ollama, so replace the default Ollama model cell with one of the documented cloud-provider examples before running the remaining cells.

Local (Jupyter/VSCode): Optionally create and activate a virtual environment, install dependencies with pip install -r requirements.txt or uv pip install -r requirements.txt, add your PDFs to docs/, then run all cells top to bottom.

The chat interface will appear at the end.

Option 2: Full Python Project (Recommended for Development)

1. Install Dependencies

# Clone the repository
git clone https://github.com/GiovanniPasq/agentic-rag-for-dummies
cd agentic-rag-for-dummies
Option A: pip
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
Option B: uv
uv venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
uv pip install -r requirements.txt

2. Run the Application

python project/app.py

3. Ask Questions

Open the local URL (e.g., http://127.0.0.1:7860) to start chatting.

Option 3: Docker Deployment

See project/README.md for full Docker instructions and system requirements.

Example Conversations

With Conversation Memory:

User: "How do I install SQL?" Agent: [Provides installation steps from documentation]

User: "How do I update it?" Agent: [Understands "it" = SQL, provides update instructions]

With Query Clarification:

User: "Tell me about that thing" Agent: "I need more information. What specific topic are you asking about?"

User: "The installation process for PostgreSQL" Agent: [Retrieves and answers with specific information]

Troubleshooting

| Area | Common Problems | Suggested Solutions | |------|----------------|------------------| | Model Selection | - Responses ignore instructions
- Tools (retrieval/search) used incorrectly
- Poor context understanding
- Hallucinations or incomplete aggregation | - Use more capable LLMs
- Prefer models 8B+ for better reasoning
- Consider cloud-based models if local models are limited | | System Prompt Behavior | - Model answers without retrieving documents
- Query rewriting loses context
- Aggregation introduces hallucinations | - Make retrieval explicit in system prompts
- Keep query rewriting close to user intent | | Retrieval Configuration | - Relevant documents not retrieved
- Too much irrelevant information | - Increase retrieved chunks (k) or lower similarity thresholds to improve recall
- Reduce k or increase thresholds to improve precision | | Chunk Size / Document Splitting | - Answers lack context or feel fragmented
- Retrieval is slow or embedding costs are high | - Increase chunk & parent sizes for more context
- Decrease chunk sizes to improve speed and reduce costs | | Context Compression | - Agent loses important details after compression
- Compressed summaries are too vague | - Tune the compression system prompt
- Increase BASETOKENTHRESHOLD to delay compression
- Increase TOKENGROWTHFACTOR | | Agent Configuration | - Agent gives up too early
- Agent loops too long| - Increase MAXTOOLCALLS / MAX_ITERATIONS for complex queries
- Decrease them to speed up simple queries | | Temperature & Consistency | - Responses inconsistent or overly creative
- Responses too rigid or repetitive | - Set temperature to 0 for factual, consistent output
- Slightly increase temperature for summarization or analysis tasks | | Embedding Model Quality | - Poor semantic search
- Weak performance on domain-specific or multilingual docs | - Use higher-quality or domain-specific embeddings
- Re-index all documents after changing embeddings |

💡 For additional troubleshooting tips see the README Troubleshooting.