Agentic RAG Pipelines

Naive RAG vs Agentic RAG

The standard RAG pipeline looks like this:

flowchart LR
    Q([User Query]) --> E[Embed Query] --> R[Retrieve top-K docs] --> G[LLM] --> A([Answer])

This works for simple factual questions over a well-structured knowledge base. It breaks down for:

• Multi-hop questions (“What did the CEO say about the product that won the 2024 award?”)
• Ambiguous queries where the first retrieval misses the intent
• Synthesis tasks that require information from multiple independent documents
• Verification (“Is this fact actually in the retrieved documents?”)

Agentic RAG replaces the single-step retrieval with an agent loop that can retrieve multiple times, reformulate queries, grade results, and synthesize intelligently.

The Agentic RAG Architecture

flowchart TD
    Q([User Query]) --> QA["Query Analyzer\nSimple or complex?\nDecompose if needed"]
    QA --> RET["Retriever\nVector DB + Full-text search"]
    RET --> GRD["Relevance Grader\nScore each doc\nDiscard irrelevant"]
    GRD -->|Enough relevant docs| SYN["Synthesizer\nGenerate answer"] --> ANS([Answer])
    GRD -->|Not enough| REF["Query Reformulator\nRewrite for better recall"]
    REF --> RET
    style QA fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style RET fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style GRD fill:#EEF0F7,stroke:#818CF8,color:#0F172A
    style SYN fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style REF fill:#fff0f0,stroke:#f87171,color:#0F172A

Document Ingestion Pipeline

Before retrieval, you need to ingest documents. Quality chunking and metadata extraction at index time pays dividends at query time.

# src/rag/ingestion.py
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader
import hashlib

class DocumentIngester:
    def __init__(self, vector_store: AgentMemoryStore):
        self.vector_store = vector_store
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=800,     # ~200-250 tokens per chunk, small enough to stay focused, large enough for context
            chunk_overlap=100,  # overlap prevents cutting mid-sentence and losing cross-boundary context
            separators=["\n\n", "\n", ". ", " "],  # try paragraph breaks before line breaks before word breaks
        )
    
    async def ingest_pdf(self, pdf_path: str, metadata: dict = {}) -> int:
        """Load, chunk, and index a PDF document."""
        loader = PyPDFLoader(pdf_path)
        pages = loader.load()  # returns one Document per page, preserving page numbers
        
        chunks = []
        for page in pages:
            page_chunks = self.splitter.split_text(page.page_content)
            for i, chunk in enumerate(page_chunks):
                chunks.append({
                    "content": chunk,
                    "metadata": {
                        **metadata,
                        "source": pdf_path,
                        "page": page.metadata.get("page", 0),  # preserve page number for citations
                        "chunk_index": i,
                        # short hash to detect duplicate chunks during re-ingestion
                        "chunk_id": hashlib.md5(chunk.encode()).hexdigest()[:8],
                    }
                })
        
        # Batch embed and store
        for i in range(0, len(chunks), 50):  # embed in batches of 50 to stay within API limits
            batch = chunks[i:i+50]
            contents = [c["content"] for c in batch]
            embeddings = await embed_batch(contents)  # one API call per batch, not per chunk
            
            for chunk, embedding in zip(batch, embeddings):
                await self.vector_store.store_memory(
                    content=chunk["content"],
                    metadata=chunk["metadata"],
                )
        
        return len(chunks)
    
    async def ingest_web(self, url: str, metadata: dict = {}) -> int:
        """Scrape, chunk, and index a web page."""
        loader = WebBaseLoader(url)
        docs = loader.load()
        
        # join all page sections into one text before chunking
        text = " ".join([d.page_content for d in docs])
        chunks = self.splitter.split_text(text)
        
        for i, chunk in enumerate(chunks):
            await self.vector_store.store_memory(
                content=chunk,
                metadata={"source": url, "chunk_index": i, **metadata}
            )
        
        return len(chunks)

Building the Agentic RAG Graph

This LangGraph implementation wires together the four pipeline stages: decompose, retrieve, grade, and generate, with a conditional loop that retries retrieval up to 3 times if the first results are insufficient.

# src/rag/agentic_rag.py
from langgraph.graph import StateGraph, END
from pydantic import BaseModel

class RagState(TypedDict):
    question: str
    sub_questions: list[str]        # decomposed sub-questions for multi-hop retrieval
    retrieved_docs: list[dict]
    graded_docs: list[dict]         # docs that passed relevance grading
    generation: str | None
    retrieval_attempts: int         # tracks how many times we've tried to retrieve

class RelevanceGrade(BaseModel):
    is_relevant: bool
    reasoning: str

# Structured output ensures grader always returns a well-typed boolean, not free text
grader_llm = llm.with_structured_output(RelevanceGrade)

async def decompose_question(state: RagState) -> dict:
    """Break complex questions into simpler sub-questions."""
    prompt = f"""Analyze this question: {state['question']}

If it's simple and self-contained, return just this question as a list with one item.
If it requires multi-step reasoning, break it into 2-4 simpler sub-questions.

Return as JSON: {{"sub_questions": ["question1", "question2", ...]}}"""
    
    result = llm.invoke([HumanMessage(content=prompt)])
    import json
    data = json.loads(result.content)
    return {"sub_questions": data["sub_questions"]}

async def retrieve(state: RagState) -> dict:
    """Retrieve documents for each sub-question."""
    all_docs = []
    
    for sub_q in state["sub_questions"]:
        docs = await vector_store.search(
            query=sub_q,
            top_k=4,                    # 4 docs per sub-question; total can grow with multiple sub-questions
            score_threshold=0.65,       # lower threshold than 0.75 to cast a wider net before grading
        )
        all_docs.extend(docs)
    
    # Deduplicate by content hash
    seen = set()
    unique_docs = []
    for doc in all_docs:
        key = doc["content"][:100]  # first 100 chars as a cheap deduplication fingerprint
        if key not in seen:
            seen.add(key)
            unique_docs.append(doc)
    
    return {
        "retrieved_docs": unique_docs,
        "retrieval_attempts": state["retrieval_attempts"] + 1,
    }

async def grade_documents(state: RagState) -> dict:
    """Grade each retrieved document for relevance."""
    graded = []
    
    for doc in state["retrieved_docs"]:
        # ask the LLM to judge relevance: more accurate than relying on cosine score alone
        grade = await grader_llm.ainvoke([
            HumanMessage(content=f"""Question: {state['question']}
            
Document: {doc['content'][:500]}

Is this document relevant to answering the question?""")
        ])
        
        if grade.is_relevant:
            graded.append(doc)
    
    return {"graded_docs": graded}

def check_relevance(state: RagState) -> str:
    """Route based on whether we have enough relevant docs."""
    if len(state["graded_docs"]) >= 2:  # 2+ relevant docs is enough to generate a grounded answer
        return "generate"
    if state["retrieval_attempts"] >= 3:  # safety valve, don't loop forever
        return "generate"  # give up and generate with what we have
    return "reformulate"  # not enough relevant docs, try a better query

async def reformulate_query(state: RagState) -> dict:
    """Reformulate questions when retrieval is insufficient."""
    prompt = f"""The original question was: {state['question']}

We retrieved {len(state['retrieved_docs'])} documents but only {len(state['graded_docs'])} were relevant.

Generate improved search queries that might find better information.
Return JSON: {{"sub_questions": ["better query 1", "better query 2"]}}"""
    
    result = llm.invoke([HumanMessage(content=prompt)])
    import json
    data = json.loads(result.content)
    # reset retrieved_docs so the next retrieval starts fresh with the new queries
    return {"sub_questions": data["sub_questions"], "retrieved_docs": []}

async def generate_answer(state: RagState) -> dict:
    """Generate final answer from graded documents."""
    # prefer graded docs; fall back to all retrieved docs if grading yielded nothing
    context = "\n\n---\n\n".join([
        f"[Source: {d['metadata'].get('source', 'unknown')}]\n{d['content']}"
        for d in (state["graded_docs"] or state["retrieved_docs"])[:8]  # cap at 8 docs to stay within context limits
    ])
    
    result = llm.invoke([
        SystemMessage(content="You are a precise research assistant. Answer based strictly on the provided context. If the context doesn't contain the answer, say so."),
        HumanMessage(content=f"Question: {state['question']}\n\nContext:\n{context}")
    ])
    
    return {"generation": result.content}

# Build the graph
rag_builder = StateGraph(RagState)
rag_builder.add_node("decompose", decompose_question)
rag_builder.add_node("retrieve", retrieve)
rag_builder.add_node("grade", grade_documents)
rag_builder.add_node("reformulate", reformulate_query)
rag_builder.add_node("generate", generate_answer)

rag_builder.set_entry_point("decompose")
rag_builder.add_edge("decompose", "retrieve")
rag_builder.add_edge("retrieve", "grade")
rag_builder.add_conditional_edges(
    "grade",
    check_relevance,
    {"generate": "generate", "reformulate": "reformulate"}  # conditional routing based on grading result
)
rag_builder.add_edge("reformulate", "retrieve")  # loop back, retry with better queries
rag_builder.add_edge("generate", END)

rag_agent = rag_builder.compile()

Agentic RAG as a Tool for Other Agents

The most powerful pattern: expose your RAG pipeline as a tool that any agent can call:

@research_agent.tool
async def search_knowledge_base(
    ctx: RunContext[AgentDeps],
    query: str,
    scope: str = "all",
) -> str:
    """Search the internal knowledge base for relevant information.
    
    Args:
        query: Natural language search query
        scope: Filter scope - "all", "docs", "code", or "policies"
    
    Returns:
        Relevant passages from the knowledge base with source citations.
    """
    filter_by = {"scope": scope} if scope != "all" else None
    
    # Run the full agentic RAG pipeline: the calling agent doesn't need to know retrieval details
    result = await rag_agent.ainvoke({
        "question": query,
        "sub_questions": [],
        "retrieved_docs": [],
        "graded_docs": [],
        "generation": None,
        "retrieval_attempts": 0,  # reset attempt counter for each fresh invocation
    })
    
    return result["generation"] or "No relevant information found."

This approach makes your entire knowledge base available to any agent without that agent needing to understand retrieval internals. The RAG agent handles query decomposition, relevance grading, and reformulation transparently.