📚 RAG — Retrieval-Augmented Generation Master Guide

"""Smart chunker with metadata enrichment."""
from langchain.text_splitter import RecursiveCharacterTextSplitter
from datetime import datetime

def chunk_document(
    text: str,
    source: str,
    chunk_size: int = 512,
    overlap: int = 64,
    metadata: dict = None,
) -> list[dict]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
        length_function=len,
    )
    chunks = splitter.split_text(text)

    base_meta = {
        "source": source,
        "indexed_at": datetime.utcnow().isoformat(),
        "chunk_count": len(chunks),
        **(metadata or {}),
    }

    return [
        {
            "content": chunk,
            "metadata": {
                **base_meta,
                "chunk_index": i,
                "char_count": len(chunk),
                # Approx token count (rough: 4 chars/token)
                "token_estimate": len(chunk) // 4,
            },
        }
        for i, chunk in enumerate(chunks)
    ]

# Usage:
# chunks = chunk_document(
#     text=my_document,
#     source="hr_policy_2025.pdf",
#     metadata={
#         "department": "HR",
#         "allowed_roles": ["employee", "manager"],
#         "doc_type": "policy",
#         "version": "2.1",
#     }
# )

"""Hybrid BM25 + vector retrieval with RRF fusion."""
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Qdrant

def build_hybrid_retriever(docs: list, k: int = 20):
    """
    Build a hybrid retriever that fuses BM25 and vector search.
    Consistently outperforms either alone.
    """
    # BM25 (keyword / sparse)
    bm25_retriever = BM25Retriever.from_documents(docs)
    bm25_retriever.k = k

    # Dense vector (semantic)
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Qdrant.from_documents(
        docs,
        embeddings,
        location=":memory:",
        collection_name="docs",
    )
    vector_retriever = vectorstore.as_retriever(
        search_kwargs={"k": k}
    )

    # EnsembleRetriever uses RRF fusion internally
    # weights=[0.5, 0.5] = equal blend; tune on your eval set
    return EnsembleRetriever(
        retrievers=[bm25_retriever, vector_retriever],
        weights=[0.4, 0.6],  # slightly favor semantic
    )

# For production: add a cross-encoder reranker on top
# from langchain_cohere import CohereRerank
# from langchain.retrievers import ContextualCompressionRetriever
# reranker = CohereRerank(model="rerank-v3.5", top_n=5)
# final = ContextualCompressionRetriever(
#     base_compressor=reranker, base_retriever=hybrid
# )

"""Role-based access control for RAG retrieval.
Pattern: tag chunks at ingestion, filter at query time."""
import jwt
from functools import wraps
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchAny

client = QdrantClient(":memory:")

def get_roles_from_token(token: str) -> list[str]:
    """Extract roles from JWT token."""
    try:
        payload = jwt.decode(token, options={"verify_signature": False})
        return payload.get("roles", [])
    except Exception:
        return []  # deny on error

def index_chunk_with_rbac(
    chunk_text: str,
    embedding: list[float],
    allowed_roles: list[str],
    metadata: dict,
):
    """Index a chunk with role-based access metadata."""
    client.upsert(
        collection_name="docs",
        points=[{
            "id": generate_id(),
            "vector": embedding,
            "payload": {
                "content": chunk_text,
                "allowed_roles": allowed_roles,  # the key field
                **metadata,
            }
        }]
    )

def secure_search(
    query_embedding: list[float],
    user_token: str,
    k: int = 10,
) -> list:
    """Search with mandatory RBAC filter. Never call without filter."""
    # 1. Get user's roles from auth token (never trust client claims)
    user_roles = get_roles_from_token(user_token)
    if not user_roles:
        return []  # deny if no valid roles

    # 2. Build filter: doc's allowed_roles must overlap user's roles
    rbac_filter = Filter(
        must=[
            FieldCondition(
                key="allowed_roles",
                match=MatchAny(any=user_roles)
            )
        ]
    )

    # 3. Search with MANDATORY filter (never omit this)
    results = client.search(
        collection_name="docs",
        query_vector=query_embedding,
        query_filter=rbac_filter,
        limit=k,
    )
    # User can only see docs where their role is in allowed_roles
    return results

"""Complete production RAG chain with citations."""
from langchain_anthropic import ChatAnthropic
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

RAG_PROMPT = PromptTemplate.from_template("""
You are a helpful assistant. Answer the question using ONLY
the provided context. If the answer is not in the context,
say "I don't have information about that."

Always cite your sources at the end using [Source: filename, page N].

Context:
{context}

Question: {question}

Answer:""")

def build_rag_chain(retriever):
    llm = ChatAnthropic(
        model="claude-sonnet-4-6",
        temperature=0,  # deterministic for RAG
    )
    return (
        {"context": retriever, "question": lambda x: x}
        | RAG_PROMPT
        | llm
        | StrOutputParser()
    )

# With evaluation hooks:
def rag_with_eval(chain, retriever, query: str) -> dict:
    """Run RAG and capture retrieval metadata for eval."""
    docs = retriever.invoke(query)
    answer = chain.invoke(query)
    return {
        "query": query,
        "answer": answer,
        "retrieved_docs": [
            {"source": d.metadata.get("source"),
             "content": d.page_content[:200]}
            for d in docs
        ],
        "retrieval_count": len(docs),
        # Feed these to RAGAS for evaluation:
        # contexts=[d.page_content for d in docs],
    }

"""PII detection and anonymization before indexing.
pip install presidio-analyzer presidio-anonymizer"""
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

PII_ENTITIES = [
    "PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER",
    "CREDIT_CARD", "US_SSN", "IP_ADDRESS",
    "LOCATION", "DATE_TIME",  # optional
]

def process_chunk_for_indexing(text: str) -> dict:
    """
    Analyze and optionally anonymize a chunk before indexing.
    Returns cleaned text + PII report for metadata.
    """
    results = analyzer.analyze(
        text=text,
        entities=PII_ENTITIES,
        language="en",
    )

    has_pii = len(results) > 0
    pii_types = list({r.entity_type for r in results})

    if has_pii:
        # Replace PII with type tokens e.g. [PERSON], [EMAIL]
        anonymized = anonymizer.anonymize(
            text=text,
            analyzer_results=results,
        )
        clean_text = anonymized.text
    else:
        clean_text = text

    return {
        "text_for_index": clean_text,  # use this for embedding
        "original_text": text,          # store separately if needed
        "has_pii": has_pii,
        "pii_types": pii_types,
        "metadata": {
            "pii_flag": has_pii,
            "pii_types": pii_types,
            # Increase access restriction if PII detected
            "min_access_level": "restricted" if has_pii else "internal",
        }
    }

# Usage in ingestion pipeline:
# for chunk in chunks:
#     processed = process_chunk_for_indexing(chunk["content"])
#     if processed["has_pii"]:
#         chunk["metadata"]["allowed_roles"] = ["hr", "admin"]
#     embed_and_index(processed["text_for_index"], chunk["metadata"])

"""Evaluate your RAG pipeline with RAGAS.
pip install ragas langchain-openai"""
from ragas import evaluate
from ragas.metrics import (
    faithfulness,          # does answer use only retrieved context?
    answer_relevancy,      # does answer address the question?
    context_precision,     # are retrieved chunks actually useful?
    context_recall,        # did we retrieve all relevant chunks?
)
from datasets import Dataset

def evaluate_rag_pipeline(
    queries: list[str],
    answers: list[str],
    contexts: list[list[str]],
    ground_truths: list[str] = None,
) -> dict:
    """
    Run RAGAS evaluation on a batch of RAG outputs.

    Args:
        queries:       The user questions
        answers:       The RAG-generated answers
        contexts:      List of retrieved chunks per query
        ground_truths: Expected answers (for context_recall)
    """
    data = {
        "question":  queries,
        "answer":    answers,
        "contexts":  contexts,
    }
    if ground_truths:
        data["ground_truth"] = ground_truths

    dataset = Dataset.from_dict(data)
    metrics = [faithfulness, answer_relevancy, context_precision]
    if ground_truths:
        metrics.append(context_recall)

    result = evaluate(dataset, metrics=metrics)
    return {
        "faithfulness":     result["faithfulness"],
        "answer_relevancy": result["answer_relevancy"],
        "context_precision":result["context_precision"],
        "context_recall":   result.get("context_recall"),
        "overall":          sum(result.values()) / len(result),
    }

# Interpretation:
# faithfulness     < 0.8 → hallucination risk
# answer_relevancy < 0.8 → prompt or retrieval issue
# context_precision< 0.7 → too many irrelevant chunks retrieved
# context_recall   < 0.7 → missing relevant chunks (chunking issue)

"""Advanced retrieval patterns for hard queries."""
from langchain.retrievers import MultiQueryRetriever
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import PromptTemplate

llm = ChatAnthropic(model="claude-haiku-4-5", temperature=0.3)

# ── PATTERN 1: Multi-Query ──────────────────────────────────
# Generate N query variants to improve recall on ambiguous queries
MULTI_QUERY_PROMPT = PromptTemplate.from_template("""
Generate 3 different phrasings of this query that capture
the same intent but use different words.
Output as a numbered list.

Original query: {question}
""")

multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=base_retriever,
    llm=llm,
    prompt=MULTI_QUERY_PROMPT,
)
# Retrieves for all 3 variants, deduplicates results

# ── PATTERN 2: HyDE (Hypothetical Document Embeddings) ─────
# LLM writes a fake answer → embed that → search
# Works well for vague or abstract queries
HYDE_PROMPT = PromptTemplate.from_template("""
Write a short, factual passage that would answer this question.
Be specific, not vague. 2-3 sentences.

Question: {question}
Passage:""")

def hyde_retrieval(query: str, retriever, k: int = 5):
    """Generate hypothetical doc, embed it, search with that."""
    chain = HYDE_PROMPT | llm
    # Generate hypothetical answer
    hypo_doc = chain.invoke({"question": query}).content
    # Embed the hypothetical doc (not the original query)
    hypo_embedding = embeddings.embed_query(hypo_doc)
    # Search with the hypothetical embedding
    return retriever.vectorstore.similarity_search_by_vector(
        hypo_embedding, k=k
    )

# When to use which:
# Multi-query → queries with multiple interpretations
# HyDE → abstract or vague queries where exact match is unlikely
# Ensemble → always (combine with above)

"""LLM extracts metadata filters from natural language queries.
'Show me HR docs updated after 2025' → filter + semantic search"""
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo
from langchain_anthropic import ChatAnthropic

# Tell the LLM what metadata fields exist and their types
metadata_field_info = [
    AttributeInfo(
        name="department",
        description="Department that owns this doc: HR, Finance, Engineering, Legal",
        type="string",
    ),
    AttributeInfo(
        name="doc_type",
        description="Document type: policy, report, guide, procedure",
        type="string",
    ),
    AttributeInfo(
        name="updated_at",
        description="Last update date in YYYY-MM-DD format",
        type="string",
    ),
    AttributeInfo(
        name="access_level",
        description="Access restriction: public, internal, confidential, restricted",
        type="string",
    ),
    AttributeInfo(
        name="version",
        description="Document version number",
        type="string",
    ),
]

retriever = SelfQueryRetriever.from_llm(
    llm=ChatAnthropic(model="claude-haiku-4-5"),
    vectorstore=vectorstore,
    document_contents="Internal company documents including policies and reports",
    metadata_field_info=metadata_field_info,
    verbose=True,  # prints the generated filters for debugging
)

# These natural language queries auto-generate structured filters:
# "HR policies" → filter: {department: "HR", doc_type: "policy"}
# "Finance reports from 2025" → filter: {department: "Finance", updated_at: {$gte: "2025-01-01"}}
# "Latest version of the remote work guide" → filter: {doc_type: "guide"} + semantic "remote work"