agentic-rag-patterns

# Agentic RAG Patterns

Build self-correcting retrieval systems with LLM-driven decision making.

> **LangGraph 1.0.6** (Jan 2026): langgraph-checkpoint 4.0.0, compile-time checkpointer validation, namespace sanitization.

## Architecture Overview

```
Query → [Retrieve] → [Grade] → [Generate/Rewrite/Web Search] → Response
              ↓           ↓
         Documents    Quality Check
                          ↓
                   Route Decision:
                   - Good docs → Generate
                   - Poor docs → Rewrite query
                   - No docs → Web fallback
```

## Self-RAG State Definition

```python
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List, Annotated
from langchain_core.documents import Document
import operator

class RAGState(TypedDict):
    """State for agentic RAG workflows."""
    question: str
    documents: Annotated[List[Document], operator.add]
    generation: str
    web_search_needed: bool
    retry_count: int
    relevance_scores: dict[str, float]
```

## Core Retrieval Node

```python
def retrieve(state: RAGState) -> dict:
    """Retrieve documents from vector store."""
    question = state["question"]
    documents = retriever.invoke(question)
    return {"documents": documents, "question": question}
```

## Document Grading (Self-RAG Core)

```python
from pydantic import BaseModel, Field

class GradeDocuments(BaseModel):
    """Binary score for document relevance."""
    binary_score: str = Field(
        description="Relevance score 'yes' or 'no'"
    )

def grade_documents(state: RAGState) -> dict:
    """Grade documents for relevance - core Self-RAG pattern."""
    question = state["question"]
    documents = state["documents"]

    filtered_docs = []
    relevance_scores = {}

    for doc in documents:
        score = retrieval_grader.invoke({
            "question": question,
            "document": doc.page_content
        })
        doc_id = doc.metadata.get("id", hash(doc.page_