AdvancedEstimated time: 60 mins

Agentic RAG with LangGraph

Build an autonomous RAG agent that evaluates retrieval quality, rewrites queries conditionally, and invokes tools.

Tech Stack:PythonLangGraphLangChainTavily Search APIChromaDB

System Architecture

Query ➔ Agent State ➔ Retrieve Node ➔ Grade Documents Node ➔ [If Relevant ➔ Generate Answer] / [If Irrelevant ➔ Rewrite Query ➔ Web Search ➔ Generate]

Prerequisites

  • Understanding of State Graphs (Nodes, Edges, State)
  • LangGraph core concepts

Installation Commands

terminalbash
pip install langgraph langchain langchain-openai tavily-python chromadb

Folder Structure

Directory Layouttext
agentic-rag/
├── graph.py
├── state.py
└── nodes.py

Step-by-Step Implementation

1

Define Graph State

Define the TypedDict representing our graph workflow state.

state.pypython
from typing import List, TypedDict

class GraphState(TypedDict):
    question: str
    generation: str
    documents: List[str]
    iteration_count: int
2

Construct Nodes and Conditional Routing in LangGraph

Implement document grading, query rewriting, and the cyclic graph workflow.

graph.pypython
from langgraph.graph import END, StateGraph
from state import GraphState

def retrieve(state: GraphState):
    print("---RETRIEVING FROM VECTOR STORE---")
    # Simulation: retrieve docs
    return {"documents": ["Doc 1 context..."]}

def grade_documents(state: GraphState):
    print("---GRADING DOCUMENTS FOR RELEVANCE---")
    # Evaluator LLM checks relevance
    return state

def decide_to_generate(state: GraphState):
    # Conditional edge logic
    if len(state["documents"]) > 0:
        return "generate"
    return "rewrite_query"

workflow = StateGraph(GraphState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges("grade_documents", decide_to_generate, {
    "generate": END,
    "rewrite_query": "retrieve"
})
app = workflow.compile()

Example Execution (Input & Output)

Input Query:
Query: "What happened in the latest 2026 AI safety conference?"
Generated Grounded Output:
Workflow executed: Retrieve (Local Store: 0 relevant hits) ➔ Grade (Failed) ➔ Rewrite Query ➔ Web Search Fallback ➔ Grounded Synthesis.

Why this works: The agent detected that local documents were outdated or irrelevant, autonomously falling back to web search.

Troubleshooting Common Gotchas

Infinite graph loop when documents are never deemed relevant

Enforce a strict maximum iteration counter in GraphState (e.g., max 3 attempts).

Known Limitations

  • •Higher latency and token usage due to iterative decision-making cycles.

Security Considerations

  • •Constrain external web tool access with strict domain whitelisting.
Back to all projects