┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Ingest │────▶│ Embedding │────▶│ Retrieval │────▶│ Generation │ │ Pipeline │ │ & Index │ │ Pipeline │ │ + Guardrails └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ │ ▼ ▼ ▼ ▼ Chunking Vector DB Hybrid Search Structured Output Metadata (Qdrant) (Dense+Sparse) Citations Deduplication Quantization Reranking Guardrails Incremental Replication Filtering Evaluation 

# ingestion/pipeline.py from dataclasses import dataclass from typing import List, Optional from pathlib import Path import hashlib import asyncio from datetime import datetime @dataclass class Document: id: str content: str metadata: dict source: str created_at: str chunk_hash: str = "" @dataclass class Chunk: id: str document_id: str content: str metadata: dict token_count: int embedding: List[float] = None class IngestionPipeline: def __init__( self, chunker, embedder, vector_store, doc_store, deduplicator ): self.chunker = chunker self.embedder = embedder self.vector_store = vector_store self.doc_store = doc_store self.deduplicator = deduplicator async def ingest_file(self, file_path: Path, metadata: dict = None) -> dict: """Ingest a single file.""" # 1. Load document  doc = await self._load_document(file_path, metadata) # 2. Check for duplicates  if await self.deduplicator.is_duplicate(doc.chunk_hash): return {"status": "duplicate", "document_id": doc.id} # 3. Chunk  chunks = self.chunker.chunk(doc) # 4. Embed  texts = [c.content for c in chunks] embeddings = await self.embedder.embed_batch(texts) for chunk, emb in zip(chunks, embeddings): chunk.embedding = emb # 5. Store  await self.doc_store.save_document(doc) await self.vector_store.upsert_chunks(chunks) await self.deduplicator.mark_seen(doc.chunk_hash) return {"status": "success", "document_id": doc.id, "chunks": len(chunks)} async def ingest_batch(self, file_paths: List[Path]) -> dict: """Parallel batch ingestion.""" semaphore = asyncio.Semaphore(10) async def limited_ingest(path): async with semaphore: return await self.ingest_file(path) results = await asyncio.gather(*[limited_ingest(p) for p in file_paths]) return { "total": len(results), "successful": sum(1 for r in results if r["status"] == "success"), "duplicates": sum(1 for r in results if r["status"] == "duplicate"), "details": results } async def _load_document(self, path: Path, metadata: dict) -> Document: # Implement based on file type (PDF, MD, HTML, etc.)  content = path.read_text(encoding="utf-8") chunk_hash = hashlib.sha256(content.encode()).hexdigest()[:16] return Document( id=hashlib.md5(str(path).encode()).hexdigest()[:12], content=content, metadata=metadata or {}, source=str(path), created_at=datetime.utcnow().isoformat(), chunk_hash=chunk_hash ) 

# ingestion/chunker.py from dataclasses import dataclass import re from typing import List @dataclass class Chunk: content: str metadata: dict token_count: int class SemanticChunker: """Chunk by semantic boundaries (headers, paragraphs).""" def __init__(self, max_tokens: int = 512, overlap: int = 50): self.max_tokens = max_tokens self.overlap = overlap def chunk(self, doc: Document) -> List[Chunk]: # Split by markdown headers  sections = self._split_by_headers(doc.content) all_chunks = [] for section_content, section_meta in sections: section_chunks = self._chunk_section( section_content, {**doc.metadata, **section_meta} ) all_chunks.extend(section_chunks) return all_chunks def _split_by_headers(self, text: str) -> List[tuple[str, dict]]: pattern = r'^(#{1,6})\s+(.+)$' lines = text.split('\n') sections = [] current_header = {} current_content = [] for line in lines: match = re.match(pattern, line) if match: if current_content: header_path = " > ".join(v for v in current_header.values() if v) sections.append(("\n".join(current_content), {"header_path": header_path})) level = len(match.group(1)) header_text = match.group(2) current_header[f"h{level}"] = header_text for l in range(level + 1, 7): current_header[f"h{l}"] = "" current_content = [line] else: current_content.append(line) if current_content: header_path = " > ".join(v for v in current_header.values() if v) sections.append(("\n".join(current_content), {"header_path": header_path})) return sections def _chunk_section(self, content: str, metadata: dict) -> List[Chunk]: paragraphs = [p.strip() for p in content.split('\n\n') if p.strip()] chunks = [] current_chunk = [] current_tokens = 0 for para in paragraphs: para_tokens = self._estimate_tokens(para) if current_tokens + para_tokens > self.max_tokens and current_chunk: chunk_text = "\n\n".join(current_chunk) chunks.append(Chunk(chunk_text, metadata, current_tokens)) overlap_text = self._get_overlap(current_chunk) current_chunk = [overlap_text, para] if overlap_text else [para] current_tokens = self._estimate_tokens("\n\n".join(current_chunk)) else: current_chunk.append(para) current_tokens += para_tokens if current_chunk: chunk_text = "\n\n".join(current_chunk) chunks.append(Chunk(chunk_text, metadata, current_tokens)) return chunks def _estimate_tokens(self, text: str) -> int: return len(text) // 4 def _get_overlap(self, chunks: List[str]) -> str: if not chunks: return "" last = chunks[-1] words = last.split() overlap_words = words[-(self.overlap // 4):] return " ".join(overlap_words) 

# retrieval/pipeline.py from dataclasses import dataclass from typing import List, Optional, Callable import asyncio @dataclass class RetrievedDoc: content: str metadata: dict score: float rank: int source: str # "dense", "sparse", "hybrid"  class HybridRetrievalPipeline: def __init__( self, dense_embedder: Callable, sparse_embedder: Callable, vector_store, sparse_store, reranker: Callable, dense_weight: float = 0.7, sparse_weight: float = 0.3 ): self.dense_embedder = dense_embedder self.sparse_embedder = sparse_embedder self.vector_store = vector_store self.sparse_store = sparse_store self.reranker = reranker self.dense_weight = dense_weight self.sparse_weight = sparse_weight async def retrieve( self, query: str, k: int = 5, fetch_k: int = 20, filters: dict = None, use_reranker: bool = True ) -> List[RetrievedDoc]: # Stage 1: Parallel dense + sparse retrieval  dense_task = self._dense_search(query, fetch_k, filters) sparse_task = self._sparse_search(query, fetch_k, filters) dense_results, sparse_results = await asyncio.gather(dense_task, sparse_task) # Stage 2: Fuse results (RRF - Reciprocal Rank Fusion)  fused = self._reciprocal_rank_fusion(dense_results, sparse_results, k * 2) # Stage 3: Rerank  if use_reranker and fused: fused = await self._rerank(query, fused) # Stage 4: Final selection  return fused[:k] async def _dense_search(self, query: str, k: int, filters: dict) -> List[RetrievedDoc]: query_vector = await self.dense_embedder(query) results = await self.vector_store.search(query_vector, k, filters) return [ RetrievedDoc( content=r["content"], metadata=r["metadata"], score=r["score"] * self.dense_weight, rank=i, source="dense" ) for i, r in enumerate(results) ] async def _sparse_search(self, query: str, k: int, filters: dict) -> List[RetrievedDoc]: sparse_vector = await self.sparse_embedder(query) results = await self.sparse_store.search(sparse_vector, k, filters) return [ RetrievedDoc( content=r["content"], metadata=r["metadata"], score=r["score"] * self.sparse_weight, rank=i, source="sparse" ) for i, r in enumerate(results) ] def _reciprocal_rank_fusion( self, dense: List[RetrievedDoc], sparse: List[RetrievedDoc], k: int, k_constant: int = 60 ) -> List[RetrievedDoc]: """RRF fusion for hybrid search.""" # Combine by content hash  doc_scores = {} for results, weight in [(dense, self.dense_weight), (sparse, self.sparse_weight)]: for rank, doc in enumerate(results): content_hash = hash(doc.content[:200]) if content_hash not in doc_scores: doc_scores[content_hash] = { "doc": doc, "rrf_score": 0 } # RRF score: 1 / (k + rank)  doc_scores[content_hash]["rrf_score"] += weight / (k_constant + rank + 1) # Sort by RRF score  fused = sorted( doc_scores.values(), key=lambda x: x["rrf_score"], reverse=True ) return [ RetrievedDoc( content=item["doc"].content, metadata=item["doc"].metadata, score=item["rrf_score"], rank=i, source="hybrid" ) for i, item in enumerate(fused[:k]) ] async def _rerank(self, query: str, docs: List[RetrievedDoc]) -> List[RetrievedDoc]: contents = [d.Source：https://dev.to/ipsphil_digital/production-rag-with-langchain-langgraph-retrieval-reranking-agentic-workflows-1kd9
My favorites