agentic-rag-for-dummies

# Agentic RAG for Dummies > Skill by [ara.so](https://ara.so) — AI Agent Skills collection. This skill enables you to build modular Agentic RAG (Retrieval-Augmented Generation) systems using LangGraph. The framework provides hierarchical document indexing, conversation memory, query clarification with human-in-the-loop, multi-agent map-reduce for complex queries, self-correction, and context compression. ## What This Project Does Agentic RAG for Dummies is a production-ready framework for building intelligent document retrieval systems that go beyond basic RAG: - **Hierarchical Indexing**: Search small child chunks for precision, retrieve large parent chunks for context - **Conversation Memory**: Maintains dialogue context across multiple questions - **Query Clarification**: Rewrites ambiguous queries or pauses for human clarification - **Multi-Agent Orchestration**: Decomposes complex queries into parallel sub-agents using LangGraph - **Self-Correction**: Automatically re-queries when initial results are insufficient - **Context Compression**: Prevents redundant retrievals across long conversations - **Provider Agnostic**: Works with Ollama, OpenAI, Anthropic, Google, or any LangChain-supported LLM ## Installation ### Clone and Set Up Environment ```bash git clone https://github.com/GiovanniPasq/agentic-rag-for-dummies.git cd agentic-rag-for-dummies python -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate pip install -r requirements.txt ``` ### Install Ollama (for Local LLMs) ```bash # Download from https://ollama.com or use: curl -fsSL https://ollama.com/install.sh | sh # Pull a recommended model (7B+ for reliable tool calling) ollama pull qwen3:4b-instruct-2507-q4_K_M # Or for better performance: ollama pull llama3.1:8b-instruct-q4_K_M ``` ### For Cloud Providers ```bash # OpenAI pip install langchain-openai export OPENAI_API_KEY="your-key-here" # Anthropic pip install langchain-anthropic export ANTHROPIC_API_KEY="your-key-here" # Google pip install langchain-google-genai export GOOGLE_API_KEY="your-key-here" ``` ## Core Configuration ### Initialize Components ```python import os from pathlib import Path from langchain_huggingface import HuggingFaceEmbeddings from langchain_qdrant.fastembed_sparse import FastEmbedSparse from qdrant_client import QdrantClient from langchain_ollama import ChatOllama # Directory structure DOCS_DIR = "docs" # Your PDF files MARKDOWN_DIR = "markdown_docs" # Converted markdown PARENT_STORE_PATH = "parent_store" # Parent chunk storage CHILD_COLLECTION = "document_child_chunks" # Vector DB collection os.makedirs(DOCS_DIR, exist_ok=True) os.makedirs(MARKDOWN_DIR, exist_ok=True) os.makedirs(PARENT_STORE_PATH, exist_ok=True) # Initialize LLM (swap provider easily) llm = ChatOllama(model="qwen3:4b-instruct-2507-q4_K_M", temperature=0) # Embeddings for hybrid search dense_embeddings = HuggingFaceEmbeddings( model_name="sentence-transformers/all-mpnet-base-v2" ) sparse_embeddings = FastEmbedSparse(model_name="Qdrant/bm25") # Vector database client = QdrantClient(path="qdrant_db") ``` ### Switch LLM Providers ```python # OpenAI from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # Anthropic from langchain_anthropic import ChatAnthropic llm = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0) # Google from langchain_google_genai import ChatGoogleGenerativeAI llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0) ``` ## Document Processing Pipeline ### 1. Convert PDFs to Markdown ```python import pymupdf import pymupdf4llm import glob def pdf_to_markdown(pdf_path, output_dir): """Convert a single PDF to Markdown.""" doc = pymupdf.open(pdf_path) md = pymupdf4llm.to_markdown( doc, header=False, footer=False, page_separators=True, ignore_images=True ) md_cleaned = md.encode('utf-8', errors='surrogatepass').decode('utf-8', errors='ignore') output_path = Path(output_dir) / Path(doc.name).stem Path(output_path).with_suffix(".md").write_bytes(md_cleaned.encode('utf-8')) def pdfs_to_markdowns(path_pattern, overwrite=False): """Convert all PDFs matching pattern.""" output_dir = Path(MARKDOWN_DIR) for pdf_path in map(Path, glob.glob(path_pattern)): md_path = (output_dir / pdf_path.stem).with_suffix(".md") if overwrite or not md_path.exists(): pdf_to_markdown(pdf_path, output_dir) # Convert all PDFs in docs directory pdfs_to_markdowns(f"{DOCS_DIR}/*.pdf") ``` ### 2. Hierarchical Chunking (Parent/Child) ```python from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter import json def process_document_hierarchical(markdown_path): """Split document into parent and child chunks.""" content = Path(markdown_path).read_text(encoding='utf-8') # Parent chunks: split by headers header_splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ], strip_headers=False ) parent_chunks = header_splitter.split_text(content) # Child chunks: fixed-size from each parent child_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=100 ) parent_ids = [] child_chunks = [] for i, parent in enumerate(parent_chunks): parent_id = f"{Path(markdown_path).stem}_parent_{i}" parent_ids.append(parent_id) # Store parent chunk parent_data = { "id": parent_id, "content": parent.page_content, "metadata": parent.metadata } parent_file = Path(PARENT_STORE_PATH) / f"{parent_id}.json" parent_file.write_text(json.dumps(parent_data, ensure_ascii=False)) # Create child chunks children = child_splitter.split_documents([parent]) for j, child in enumerate(children): child.metadata["parent_id"] = parent_id child.metadata["child_index"] = j child_chunks.append(child) return parent_ids, child_chunks ``` ### 3. Index Documents in Vector Database ```python from qdrant_client.http import models as qmodels from langchain_qdrant import QdrantVectorStore, RetrievalMode def ensure_collection(collection_name): """Create Qdrant collection if it doesn't exist.""" embedding_dimension = len(dense_embeddings.embed_query("test")) if not client.collection_exists(collection_name): client.create_collection( collection_name=collection_name, vectors_config=qmodels.VectorParams( size=embedding_dimension, distance=qmodels.Distance.COSINE ), sparse_vectors_config={ "sparse": qmodels.SparseVectorParams() }, ) def index_documents(markdown_files): """Index all documents with hierarchical chunking.""" ensure_collection(CHILD_COLLECTION) vector_store = QdrantVectorStore( client=client, collection_name=CHILD_COLLECTION, embedding=dense_embeddings, sparse_embedding=sparse_embeddings, retrieval_mode=RetrievalMode.HYBRID, ) all_child_chunks = [] for md_file in glob.glob(f"{MARKDOWN_DIR}/*.md"): parent_ids, child_chunks = process_document_hierarchical(md_file) all_child_chunks.extend(child_chunks) print(f"Processed {Path(md_file).name}: {len(parent_ids)} parents, {len(child_chunks)} children") # Batch index all child chunks vector_store.add_documents(all_child_chunks) return vector_store # Index all markdown documents vector_store = index_documents(f"{MARKDOWN_DIR}/*.md") ``` ## Building the Agentic RAG System ### Define Agent Tools ```python from langchain_core.tools import tool @tool def retrieve_documents(query: str) -> list[str]: """ Search the knowledge base using hybrid search (dense + sparse embeddings). Returns relevant document chunks. Args: query: The search query """ results = vector_store.similarity_search(query, k=5) return [doc.page_content for doc in results] @tool def get_parent_context(parent_id: str) -> str: """ Retrieve the full parent chunk for additional context. Args: parent_id: The parent chunk identifier """ parent_file = Path(PARENT_STORE_PATH) / f"{parent_id}.json" if parent_file.exists(): data = json.loads(parent_file.read_text()) return data["content"] return "Parent chunk not found." tools = [retrieve_documents, get_parent_context] ``` ### Define System Prompts ```python CONVERSATION_SUMMARIZER_PROMPT = """You are a conversation summarizer. Extract key context from the conversation history that is relevant to the current query. Focus on: entities mentioned, topics discussed, user intent. Conversation History: {history} Current Query: {query} Provide a concise summary of relevant context.""" QUERY_CLARIFICATION_PROMPT = """You are a query clarification assistant. Analyze the query and conversation context. If the query is: - Ambiguous or contains pronouns without clear referents: Rewrite it clearly - Multi-part (multiple questions): Split into focused sub-queries - Clear and focused: Return it unchanged Context: {context} Query: {query} Return a JSON object: {{ "needs_clarification": boolean, "clarification_question": string or null, "rewritten_queries": [list of clear, focused queries] }}""" AGENT_PROMPT = """You are a RAG agent. Use the retrieve_documents tool to search for information. If results are insufficient, try rephrasing your search query. If you find relevant parent_id metadata, use get_parent_context for full context. Available tools: - retrieve_documents(query: str): Search the knowledge base - get_parent_context(parent_id: str): Get full parent chunk Question: {query} Context: {context} Provide a comprehensive answer based on retrieved documents.""" ``` ### Define State Models ```python from typing import TypedDict, Annotated, Sequence from langgraph.graph import MessagesState from langchain_core.messages import BaseMessage class AgentState(TypedDict): """State for individual RAG agents.""" messages: Annotated[Sequence[BaseMessage], "The messages in the conversation"] query: str context: str retrieved_docs: list[str] parent_contexts: list[str] search_attempts: int max_searches: int answer: str class OrchestratorState(TypedDict): """State for the main orchestration graph.""" user_query: str conversation_history: list[dict] conversation_summary: str clarified_queries: list[str] needs_human_input: bool clarification_question: str agent_results: list[dict] final_answer: str ``` ### Build LangGraph Agent ```python from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from langchain_core.messages import HumanMessage, AIMessage def should_continue(state: AgentState) -> str: """Decide if agent should continue searching or finish.""" if state["answer"]: return "end" if state["search_attempts"] >= state["max_searches"]: return "end" return "continue" def agent_node(state: AgentState) -> AgentState: """Main agent reasoning node.""" llm_with_tools = llm.bind_tools(tools) messages = state["messages"] if not messages: messages = [HumanMessage(content=AGENT_PROMPT.format( query=state["query"], context=state.get("context", "") ))] response = llm_with_tools.invoke(messages) # Check if we have a final answer (no tool calls) if not response.tool_calls: return { **state, "answer": response.content, "messages": messages + [response] } return { **state, "messages": messages + [response], "search_attempts": state["search_attempts"] + 1 } def build_agent_graph(): """Build the RAG agent graph.""" workflow = StateGraph(AgentState) workflow.add_node("agent", agent_node) workflow.add_node("tools", ToolNode(tools)) workflow.set_entry_point("agent") workflow.add_conditional_edges( "agent", should_continue, { "continue": "tools", "end": END } ) workflow.add_edge("tools", "agent") return workflow.compile() agent_graph = build_agent_graph() ``` ### Multi-Agent Orchestration ```python from langgraph.graph import StateGraph, END import json def summarize_conversation(state: OrchestratorState) -> OrchestratorState: """Summarize conversation history for context.""" history_text = "\n".join([ f"{msg['role']}: {msg['content']}" for msg in state["conversation_history"][-5:] # Last 5 messages ]) summary_prompt = CONVERSATION_SUMMARIZER_PROMPT.format( history=history_text, query=state["user_query"] ) summary = llm.invoke(summary_prompt).content