agentic-rag

Agentic RAG

Build RAG systems that reason, plan, and adaptively retrieve information.

When to Use

• Questions require multiple retrieval steps
• Need to combine information from different sources
• Query needs decomposition into sub-queries
• Results need validation or refinement
• Complex reasoning over retrieved documents

Simple RAG vs Agentic RAG

Simple RAG:
Query → Retrieve → Generate → Answer

Agentic RAG:
Query → Plan → [Retrieve → Analyze → Decide]*n → Synthesize → Answer

Core Architecture

┌─────────────────────────────────────────────────────────┐
│                     User Question                        │
└─────────────────────────┬───────────────────────────────┘
                          │
                          ▼
                ┌───────────────────┐
                │   Query Analyzer  │
                │   (Decompose?)    │
                └─────────┬─────────┘
                          │
         ┌────────────────┼────────────────┐
         │                │                │
         ▼                ▼                ▼
   ┌──────────┐    ┌──────────┐    ┌──────────┐
   │ Sub-Q 1  │    │ Sub-Q 2  │    │ Sub-Q 3  │
   └────┬─────┘    └────┬─────┘    └────┬─────┘
        │               │               │
        ▼               ▼               ▼
   ┌──────────┐    ┌──────────┐    ┌──────────┐
   │ Retrieve │    │ Retrieve │    │ Retrieve │
   └────┬─────┘    └────┬─────┘    └────┬─────┘
        │               │               │
        └───────────────┼───────────────┘
                        │
                        ▼
              ┌───────────────────┐
              │    Synthesizer    │
              │  (Combine & Cite) │
              └─────────┬─────────┘
                        │
                        ▼
              ┌───────────────────┐
              │   Final Answer    │
              └───────────────────┘

Implementation with LangGraph

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, List, Annotated
import operator

class AgentState(TypedDict):
    question: str
    sub_questions: List[str]
    retrieved_docs: Annotated[List, operator.add]
    current_step: int
    final_answer: str

# Nodes
def analyze_query(state: AgentState) -> AgentState:
    """Decompose complex query into sub-questions."""
    llm = ChatOpenAI(model="gpt-4")

    prompt = f"""Analyze this question and break it into sub-questions if needed.
    Question: {state['question']}

    Return a JSON list of sub-questions, or just the original if simple."""

    response = llm.invoke(prompt)
    sub_questions = parse_questions(response.content)

    return {"sub_questions": sub_questions, "current_step": 0}

def retrieve_for_subquery(state: AgentState) -> AgentState:
    """Retrieve documents for current sub-question."""
    current_q = state["sub_questions"][state["current_step"]]
    docs = retriever.invoke(current_q)

    return {
        "retrieved_docs": docs,
        "current_step": state["current_step"] + 1
    }

def should_continue(state: AgentState) -> str:
    """Check if more sub-questions to process."""
    if state["current_step"] < len(state["sub_questions"]):
        return "retrieve"
    return "synthesize"

def synthesize_answer(state: AgentState) -> AgentState:
    """Combine all retrieved info into final answer."""
    llm = ChatOpenAI(model="gpt-4")

    context = "\n\n".join([doc.page_content for doc in state["retrieved_docs"]])

    prompt = f"""Based on the following context, answer the question.
    Cite sources using [1], [2], etc.

    Question: {state['question']}

    Context:
    {context}
    """

    response = llm.invoke(prompt)
    return {"final_answer": response.content}

# Build graph
workflow = StateGraph(AgentState)

workflow.add_node("analyze", analyze_query)
workflow.add_node("retrieve", retrieve_for_subquery)
workflow.add_node("synthesize", synthesize_answer)

workflow.set_entry_point("analyze")
workflow.add_edge("analyze", "retrieve")
workflow.add_conditional_edges("retrieve", should_continue, {
    "retrieve": "retrieve",
    "synthesize": "synthesize"
})
workflow.add_edge("synthesize", END)

agent = workflow.compile()

# Run
result = agent.invoke({"question": "Compare AWS and GCP pricing for ML workloads"})