agentic-rag-patterns（agentic RAG 模式）

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

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

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)

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_content))
        relevance_scores[doc_id] = 1.0 if score.binary_score == "yes" else 0.0

        if score.binary_score == "yes":
            filtered_docs.append(doc)

    # Trigger web search if too many docs filtered out
    web_search_needed = len(filtered_docs) < len(documents) // 2

    return {
        "documents": filtered_docs,
        "web_search_needed": web_search_needed,
        "relevance_scores": relevance_scores
    }

Query Transformation

def transform_query(state: RAGState) -> dict:
    """Transform query for better retrieval."""
    question = state["question"]

    better_question = question_rewriter.invoke({
        "question": question,
        "feedback": "Rephrase to improve retrieval. Be specific."
    })

    return {
        "question": better_question,
        "retry_count": state.get("retry_count", 0) + 1
    }

Web Search Fallback (CRAG)

def web_search(state: RAGState) -> dict:
    """Fallback to web search when documents insufficient."""
    question = state["question"]

    web_results = tavily_client.search(
        question,
        max_results=5,
        search_depth="advanced"
    )

    web_docs = [
        Document(
            page_content=r["content"],
            metadata={"source": r["url"], "type": "web"}
        )
        for r in web_results
    ]

    return {"documents": web_docs, "web_search_needed": False}

Generation Node

def generate(state: RAGState) -> dict:
    """Generate answer from documents."""
    question = state["question"]
    documents = state["documents"]

    context = "\n\n".join([
        f"[{i+1}] {doc.page_content}"
        for i, doc in enumerate(documents)
    ])

    generation = rag_chain.invoke({
        "context": context,
        "question": question
    })

    return {"generation": generation}

Conditional Routing

def route_after_grading(state: RAGState) -> str:
    """Route based on document quality."""
    if state["web_search_needed"]:
        if state.get("retry_count", 0) < 2:
            return "transform_query"  # Try rewriting first
        return "web_search"  # Fallback to web
    return "generate"  # Documents are good

workflow.add_conditional_edges(
    "grade",
    route_after_grading,
    {
        "generate": "generate",
        "transform_query": "transform_query",
        "web_search": "web_search"
    }
)

Complete CRAG Workflow

def build_crag_workflow() -> StateGraph:
    """Build Corrective-RAG workflow with web fallback."""
    workflow = StateGraph(RAGState)

    # Add nodes
    workflow.add_node("retrieve", retrieve)
    workflow.add_node("grade", grade_documents)
    workflow.add_node("generate", generate)
    workflow.add_node("web_search", web_search)
    workflow.add_node("transform_query", transform_query)

    # Define edges
    workflow.add_edge(START, "retrieve")
    workflow.add_edge("retrieve", "grade")

    # Conditional routing based on document quality
    workflow.add_conditional_edges(
        "grade",
        route_after_grading,
        {
            "generate": "generate",
            "transform_query": "transform_query",
            "web_search": "web_search"
        }
    )

    # After query transform, retry retrieval
    workflow.add_edge("transform_query", "retrieve")

    # Web search leads to generation
    workflow.add_edge("web_search", "generate")

    workflow.add_edge("generate", END)

    return workflow.compile()

Pattern Comparison

Key Decisions

Common Mistakes

• No fallback path (hangs on bad queries)
• Infinite rewrite loops (no retry limit)
• Web search on every query (expensive)
• Not tracking relevance scores (can't debug)

Related Skills

• rag-retrieval - Basic RAG patterns this enhances
• langgraph-routing - Conditional edge patterns
• langgraph-state - State design with reducers
• contextual-retrieval - Anthropic's context-prepending
• reranking-patterns - Post-retrieval reranking

Capability Details

self-rag

Keywords: self-rag, adaptive retrieval, reflection tokens Solves:

• Build self-correcting RAG systems
• Implement adaptive retrieval logic
• Add reflection tokens for quality

corrective-rag

Keywords: crag, document grading, web fallback Solves:

• Implement CRAG workflows
• Grade document relevance
• Add web search fallback

knowledge-graph-rag

Keywords: graphrag, neo4j, entity extraction Solves:

• Combine KG with vector search
• Entity-based retrieval
• Multi-hop reasoning

adaptive-retrieval

Keywords: query routing, multi-source, orchestration Solves:

• Route queries to optimal sources
• Multi-retriever orchestration
• Dynamic retrieval strategies