RAG Agent — LangGraph

Agentic RAG system built with LangGraph and LangChain: hybrid retrieval, parallel query expansion, hallucination-checked generation, and a human-in-the-loop gate before any web fallback. Runs fully local — no API keys required.

Why it's non-trivial: most RAG demos are a linear chain. This one is a graph with feedback loops (query rewriting on weak retrieval), parallel fan-out (Send() map-reduce), self-grading (every generation is checked for hallucination against retrieved context before it reaches the user), and an explicit human checkpoint — the agent asks permission before leaving the local knowledge base for the web.

Tech Stack

Layer	Technology
Orchestration	LangGraph 1.x (StateGraph, subgraphs, Send(), HITL, checkpointing)
LLM & Embeddings	Ollama (qwen2.5:7b, nomic-embed-text) — fully local
Vector Store	ChromaDB
Hybrid Retrieval	BM25 + vector (EnsembleRetriever), optional FlashrankRerank
Web Fallback	DuckDuckGo (free, no API key) / Tavily fallback
Tests	pytest — 18 unit tests, run offline, no Ollama required

Architecture

START
  └─► retrieve  ────────────────────────────────────── retrieval subgraph
  │     expand_queries (LLM generates N variants)
  │       └─► Send() × N ──► retrieve_single (parallel)
  │                                └─► grade_documents
  │
  ├─► generate  ────────────────────────────────────── generation subgraph
  │     generate → grade_generation (hallucination check)
  │
  ├─► transform_query → retrieve  (loop, up to MAX_LOOP_COUNT=2)
  │
  └─► web_search_confirm  ◄── HITL interrupt (user approves)
        └─► web_search → generate

LangGraph Features Demonstrated

Graph design

• Subgraphs (subgraphs/retrieval.py, subgraphs/generation.py) — compiled sub-StateGraphs as reusable units
• Send() / map-reduce parallel retrieval — expand_queries fans out N parallel retrieve_single calls; raw_documents: Annotated[list, operator.add] accumulates results; grade_documents runs once on the merged pool
• Human-in-the-loop — interrupt_before=["web_search_confirm"]; the CLI asks the user before any web search
• MemorySaver checkpointing — thread-scoped conversation memory; state persists between turns
• astream_events(version="v2") — token-level streaming with node progress labels
• langgraph.json — ready for LangGraph Studio

Engineering quality

• Annotated[list, operator.add] state reducers — nodes return only their delta; LangGraph merges
• Prompts extracted to prompts.py, separated from graph logic
• Structured logging, proper package structure, pinned requirements
• 18 pytest unit tests that run offline — graph logic is testable without a model

Project Structure

rag_agent/
├── agent.py              # Main graph (composes subgraphs, HITL)
├── state.py              # GraphState, RetrievalState, GenerationState
├── prompts.py            # All LLM prompts
├── tools.py              # Web search tool factory
├── ingest.py             # Document loading, chunking, indexing
├── chat_cli.py           # Interactive CLI (streaming v2, HITL flow)
├── subgraphs/
│   ├── retrieval.py      # Parallel Send() retrieval subgraph
│   └── generation.py     # Generate + hallucination check subgraph
├── tests/
│   └── test_nodes.py     # 18 unit tests (no Ollama required)
├── data/                 # Source documents (PDFs / notebooks)
├── langgraph.json        # LangGraph Studio config
└── requirements.txt      # Pinned versions

Setup

# 1. Dependencies
pip install -r requirements.txt

# 2. Local models
ollama pull nomic-embed-text
ollama pull qwen2.5:7b
ollama serve   # if not already running

# 3. Web search (optional, free, no API key)
pip install duckduckgo-search ddgs
# Tavily alternative: pip install langchain-tavily + TAVILY_API_KEY in .env
# Auto-selects: DuckDuckGo → Tavily → disabled (priority order in tools.py)

# 4. Index documents (place PDFs/notebooks in data/ first)
python -m rag_agent.ingest

Run

# Interactive chat (streaming + HITL)
python -m rag_agent.chat_cli

# Tests (offline, no Ollama needed)
pytest tests/ -v

Programmatic single question:

from rag_agent.agent import app, DEFAULT_CONFIG

result = app.invoke(
    {"question": "What is gradient descent?", "steps": [], "loop_count": 0,
     "documents": [], "generation": "", "relevance": ""},
    config=DEFAULT_CONFIG,
)
print(result["generation"])
print(result["steps"])   # ["retrieve", "generate"]

How HITL Works

When the local knowledge base can't answer (after MAX_LOOP_COUNT=2 query rewrites), the graph pauses at web_search_confirm before touching the internet:

🔍 Retrieving documents...
🔄 Rephrasing question...
🔍 Retrieving documents...
⏸  Paused — web search needed

⚠️  The local knowledge base didn't have a good answer.
   The agent wants to search the web.
   Allow web search? [y/n]: y

🌐 Searching the web...
💡 ANSWER:
...

Type n to skip and get an answer from partial context.

Data Source

The bundled knowledge base is built from Andrew Ng's "Supervised Machine Learning" course materials (slides and labs). Swap in any PDFs/notebooks via data/ + ingest.py.

Credits

Started as a capstone for LangChain Academy — LangGraph coursework; some Graph-RAG patterns adapted from this Habr article. Extended well beyond the course scope (subgraphs, Send() parallelism, HITL, hallucination grading, offline test suite).