User Query → Retrieve Relevant Docs → Assemble Context → LLM Generation → Output Validation → Response

┌─────────────────────────────────────────────────────┐
│                    RAG Pipeline                       │
│                                                       │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────┐ │
│  │ Document  │→ │ Chunking │→ │ Embedding│→ │Vector│ │
│  │ Ingestion │  │ Engine   │  │ Service  │  │ Store│ │
│  └──────────┘  └──────────┘  └──────────┘  └──────┘ │
│                                                       │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────┐ │
│  │  Query   │→ │ Retriever│→ │ Context  │→ │  LLM │ │
│  │ Processor│  │          │  │ Assembler│  │      │ │
│  └──────────┘  └──────────┘  └──────────┘  └──────┘ │
└─────────────────────────────────────────────────────┘

from dataclasses import dataclass
from enum import Enum

class PipelineMode(Enum):
    OFFLINE = "offline"  # Indexing pipeline
    ONLINE = "online"    # Query pipeline

@dataclass
class RAGConfig:
    """RAG system configuration"""
    # Chunking config
    chunk_size: int = 512
    chunk_overlap: int = 64
    chunking_strategy: str = "semantic"  # semantic | fixed | recursive

    # Retrieval config
    top_k: int = 10
    rerank_top_k: int = 5
    similarity_threshold: float = 0.7

    # Generation config
    model_name: str = "gpt-4"
    max_output_tokens: int = 2048
    temperature: float = 0.1

    # Quality assurance
    enable_citation: bool = True
    enable_hallucination_check: bool = True
    enable_answer_relevance_check: bool = True

from abc import ABC, abstractmethod
from pathlib import Path

class DocumentParser(ABC):
    """Base document parser"""

    @abstractmethod
    def parse(self, file_path: Path) -> list[DocumentSection]:
        """Parse document into structured sections"""
        ...

class PDFParser(DocumentParser):
    """PDF parser — preserves structural information"""

    def parse(self, file_path: Path) -> list[DocumentSection]:
        sections = []
        # 1. Extract text with layout information
        pages = self._extract_pages_with_layout(file_path)
        for page in pages:
            # 2. Detect headings, paragraphs, tables, images
            elements = self._detect_elements(page)
            for element in elements:
                sections.append(DocumentSection(
                    content=element.text,
                    element_type=element.type,  # heading, paragraph, table, etc.
                    page_number=page.number,
                    metadata={
                        "source": str(file_path),
                        "bbox": element.bounding_box,
                        "font_size": element.font_size,
                    }
                ))
        return sections

class SemanticChunker:
    """Semantic similarity-based chunker"""

    def __init__(self, embedding_model, similarity_threshold: float = 0.5):
        self.embedding_model = embedding_model
        self.threshold = similarity_threshold

    def chunk(self, sentences: list[str]) -> list[Chunk]:
        """Group sentences by semantic boundaries"""
        if not sentences:
            return []

        # 1. Generate embeddings for each sentence
        embeddings = self.embedding_model.encode(sentences)

        # 2. Compute cosine similarity between adjacent sentences
        similarities = []
        for i in range(len(embeddings) - 1):
            sim = cosine_similarity(embeddings[i], embeddings[i + 1])
            similarities.append(sim)

        # 3. Split where similarity falls below threshold
        chunks = []
        current_chunk_sentences = [sentences[0]]

        for i, sim in enumerate(similarities):
            if sim < self.threshold:
                # Semantic breakpoint — create new chunk
                chunks.append(Chunk(
                    text="\n".join(current_chunk_sentences),
                    metadata={"start_sentence": i - len(current_chunk_sentences) + 1}
                ))
                current_chunk_sentences = [sentences[i + 1]]
            else:
                current_chunk_sentences.append(sentences[i + 1])

        # Handle last chunk
        if current_chunk_sentences:
            chunks.append(Chunk(text="\n".join(current_chunk_sentences)))

        return chunks

@dataclass
class ChunkMetadata:
    """Document chunk metadata"""
    # Source information
    source_document: str          # Original document path/URI
    document_title: str           # Document title
    section_title: str | None     # Section heading
    page_number: int | None       # Page number

    # Position information
    chunk_index: int              # Sequence number within document
    total_chunks: int             # Total chunks in document
    char_offset_start: int        # Start position in original text
    char_offset_end: int          # End position in original text

    # Temporal information
    document_created_at: str      # Document creation time
    document_updated_at: str      # Document update time
    indexed_at: str               # Indexing time

    # Classification information
    document_type: str            # Document type (policy, manual, report...)
    access_level: str             # Access level
    department: str               # Owning department

    # Context window
    parent_chunk_id: str | None   # Parent chunk ID (for hierarchical retrieval)
    sibling_chunk_ids: list[str]  # Adjacent chunk IDs (for context expansion)

class VectorIndexConfig:
    """Vector index configuration"""

    @staticmethod
    def hnsw_config() -> dict:
        """HNSW — high recall, high memory"""
        return {
            "index_type": "HNSW",
            "metric_type": "COSINE",
            "params": {
                "M": 16,                # Max connections per node
                "ef_construction": 200,  # Search width during construction
                "ef_search": 128,        # Search width during query
            },
            "use_case": "Real-time queries, data size < 10M",
        }

    @staticmethod
    def ivf_pq_config() -> dict:
        """IVF_PQ — large scale, low memory"""
        return {
            "index_type": "IVF_PQ",
            "metric_type": "L2",
            "params": {
                "nlist": 4096,   # Number of cluster centers
                "m": 16,         # PQ subspaces
                "nbits": 8,      # Bits per subspace
                "nprobe": 64,    # Clusters scanned during query
            },
            "use_case": "Massive datasets, acceptable recall trade-off",
        }

class HybridRetriever:
    """Hybrid retriever: vector + keyword"""

    def __init__(self, vector_store, keyword_store, alpha: float = 0.7):
        self.vector_store = vector_store
        self.keyword_store = keyword_store
        self.alpha = alpha  # Vector score weight

    def retrieve(self, query: str, top_k: int = 10) -> list[RetrievalResult]:
        # 1. Vector retrieval
        vector_results = self.vector_store.search(
            query_embedding=self.embed(query),
            top_k=top_k * 2,  # Over-retrieve
        )

        # 2. Keyword retrieval (BM25)
        keyword_results = self.keyword_store.search(
            query=query,
            top_k=top_k * 2,
        )

        # 3. Normalize scores
        vector_scores = self._normalize_scores(vector_results)
        keyword_scores = self._normalize_scores(keyword_results)

        # 4. Weighted fusion (Reciprocal Rank Fusion)
        fused = self._reciprocal_rank_fusion(
            vector_results=vector_scores,
            keyword_results=keyword_scores,
            k=60,
        )

        return fused[:top_k]

    def _reciprocal_rank_fusion(self, vector_results, keyword_results, k=60):
        """RRF fusion algorithm"""
        scores = {}
        for rank, result in enumerate(vector_results):
            scores[result.id] = scores.get(result.id, 0) + 1 / (k + rank + 1)
        for rank, result in enumerate(keyword_results):
            scores[result.id] = scores.get(result.id, 0) + 1 / (k + rank + 1)

        sorted_results = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return [RetrievalResult(id=doc_id, score=score) for doc_id, score in sorted_results]

class QueryProcessor:
    """Query processor"""

    def __init__(self, llm_client):
        self.llm = llm_client

    async def process(self, raw_query: str) -> ProcessedQuery:
        """Process user query"""
        # 1. Query classification
        intent = await self._classify_intent(raw_query)

        # 2. Query rewriting
        rewritten = await self._rewrite_query(raw_query, intent)

        # 3. Query expansion (generate multiple retrieval queries)
        expanded_queries = await self._expand_query(rewritten)

        # 4. Extract filter conditions
        filters = await self._extract_filters(raw_query)

        return ProcessedQuery(
            original=raw_query,
            rewritten=rewritten,
            expanded=expanded_queries,
            intent=intent,
            filters=filters,
        )

    async def _expand_query(self, query: str) -> list[str]:
        """HyDE: Generate hypothetical documents as retrieval queries"""
        prompt = f"""Given the question: "{query}"
        Generate 3 different versions of this question that capture different aspects
        and would help retrieve relevant documents. Return as JSON array."""

        response = await self.llm.complete(prompt)
        return json.loads(response)

class MultiHopRetriever:
    """Multi-hop retriever for complex cross-document queries"""

    def __init__(self, retriever, llm, max_hops: int = 3):
        self.retriever = retriever
        self.llm = llm
        self.max_hops = max_hops

    async def retrieve(self, query: str) -> list[RetrievalResult]:
        collected_context = []
        current_query = query

        for hop in range(self.max_hops):
            # 1. Retrieve documents for current query
            results = self.retriever.retrieve(current_query, top_k=5)
            collected_context.extend(results)

            # 2. Assess whether more retrieval is needed
            assessment = await self._assess_completeness(
                query, collected_context
            )

            if assessment.is_sufficient:
                break

            # 3. Generate follow-up query
            current_query = await self._generate_followup_query(
                original_query=query,
                current_context=collected_context,
                missing_info=assessment.missing_information,
            )

        return self._deduplicate(collected_context)

class ContextAssembler:
    """Context assembler"""

    def __init__(self, max_context_tokens: int = 8000):
        self.max_tokens = max_context_tokens

    def assemble(
        self,
        query: str,
        retrieved_chunks: list[Chunk],
        system_prompt: str,
    ) -> AssembledContext:
        """Assemble LLM input context"""
        # 1. Already sorted by relevance from retriever
        # 2. Deduplicate and merge adjacent chunks
        merged = self._merge_adjacent_chunks(retrieved_chunks)

        # 3. Token budget allocation
        system_tokens = self._count_tokens(system_prompt)
        query_tokens = self._count_tokens(query)
        available = self.max_tokens - system_tokens - query_tokens - 200  # Buffer

        # 4. Greedy filling (prioritize high-relevance chunks)
        selected_chunks = []
        used_tokens = 0
        for chunk in merged:
            chunk_tokens = self._count_tokens(chunk.text)
            if used_tokens + chunk_tokens <= available:
                selected_chunks.append(chunk)
                used_tokens += chunk_tokens

        # 5. Re-sort by original document order (maintain reading coherence)
        selected_chunks.sort(key=lambda c: (c.metadata.source_document, c.metadata.chunk_index))

        return AssembledContext(
            chunks=selected_chunks,
            total_tokens=used_tokens,
            coverage_ratio=len(selected_chunks) / len(retrieved_chunks),
        )

RAG_PROMPT_TEMPLATE = """You are an enterprise knowledge assistant. Answer the user's question
based on the following retrieved document excerpts.

## Rules
1. Answer ONLY based on the provided documents — do not use your training knowledge
2. If documents lack sufficient information, explicitly state "Based on available documents,
   I cannot answer this question"
3. Every key assertion must cite the source document: [Source: document name, page]
4. If different documents contain contradictory information, note the contradiction
   and present each document's position

## Retrieved Documents
{context}

## User Question
{query}

## Answer
"""

class HallucinationDetector:
    """Verifies whether generated content is supported by source documents"""

    def __init__(self, nli_model):
        self.nli_model = nli_model  # Natural Language Inference model

    def detect(self, generated_answer: str, source_chunks: list[Chunk]) -> list[HallucinationFlag]:
        flags = []
        # Split answer into independent claims
        claims = self._extract_claims(generated_answer)

        for claim in claims:
            # Check each claim against source documents
            support_scores = []
            for chunk in source_chunks:
                score = self.nli_model.predict_entailment(
                    premise=chunk.text,
                    hypothesis=claim.text,
                )
                support_scores.append(score)

            max_support = max(support_scores) if support_scores else 0

            if max_support < 0.5:
                flags.append(HallucinationFlag(
                    claim=claim.text,
                    confidence=1 - max_support,
                    suggestion="This claim lacks sufficient support in source documents",
                ))

        return flags

class AnswerRelevanceEvaluator:
    """Evaluates relevance of generated answer to user question"""

    async def evaluate(self, query: str, answer: str) -> float:
        """Returns a 0-1 relevance score"""
        prompt = f"""Rate the relevance of the following answer to the question.
        Score from 0 (completely irrelevant) to 1 (perfectly relevant).

        Question: {query}
        Answer: {answer}

        Return only the numeric score."""

        score = float(await self.llm.complete(prompt))
        return min(max(score, 0), 1)

class CitationVerifier:
    """Citation verifier"""

    def verify(self, answer: str, chunks: list[Chunk]) -> VerificationResult:
        citations = self._extract_citations(answer)
        verified = []
        unverified = []

        for citation in citations:
            match = self._find_in_chunks(citation.text, chunks)
            if match and match.similarity > 0.85:
                verified.append(VerifiedCitation(
                    citation=citation,
                    source_chunk=match.chunk,
                    similarity=match.similarity,
                ))
            else:
                unverified.append(citation)

        return VerificationResult(
            verified=verified,
            unverified=unverified,
            verification_rate=len(verified) / len(citations) if citations else 1.0,
        )

class AgenticRAG:
    """Agent-enhanced RAG system"""

    def __init__(self, retriever, llm, tools):
        self.retriever = retriever
        self.llm = llm
        self.tools = tools  # Calculator, database query, API calls, etc.

    async def answer(self, query: str) -> AgentResponse:
        plan = await self._plan(query)

        results = []
        for step in plan.steps:
            if step.action == "retrieve":
                docs = self.retriever.retrieve(step.query)
                results.append(("context", docs))
            elif step.action == "compute":
                output = await self.tools.execute(step.tool, step.args)
                results.append(("computation", output))
            elif step.action == "synthesize":
                answer = await self._synthesize(query, results)
                results.append(("answer", answer))

        return AgentResponse(answer=results[-1][1], trace=results)