┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Ingest │────▶│ Embedding │────▶│ Retrieval │────▶│ Generation │ │ (Documents)│ │ (Vectors) │ │ (Top-K) │ │ (Answer) │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ │ ▼ ▼ ▼ ▼ Chunking Vector DB Reranking LLM + Context Strategy (Qdrant/ (Cross- (Citations, (Semantic/ Pinecone/ encoder/ Guardrails) Fixed-size) Weaviate) MMR) 

# ingestion/chunker.py from typing import List from dataclasses import dataclass import re @dataclass class Chunk: content: str metadata: dict token_count: int class SemanticChunker: """Chunk by semantic boundaries (headers, paragraphs) not fixed size.""" def __init__(self, max_tokens: int = 512, overlap: int = 50): self.max_tokens = max_tokens self.overlap = overlap def chunk(self, text: str, metadata: dict) -> List[Chunk]: # Split by markdown headers first  sections = self._split_by_headers(text) chunks = [] for section in sections: section_chunks = self._chunk_section(section, metadata) chunks.extend(section_chunks) return chunks def _split_by_headers(self, text: str) -> List[tuple[str, dict]]: """Split by # ## ### headers, preserve hierarchy.""" pattern = r'^(#{1,6})\s+(.+)$' lines = text.split('\n') sections = [] current_header = {"h1": "", "h2": "", "h3": ""} current_content = [] for line in lines: match = re.match(pattern, line) if match: # Save previous section  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})) # Update header hierarchy  level = len(match.group(1)) header_text = match.group(2) current_header[f"h{level}"] = header_text # Clear deeper levels  for l in range(level + 1, 7): current_header[f"h{l}"] = "" current_content = [line] else: current_content.append(line) # Last section  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, section: tuple[str, dict], base_metadata: dict) -> List[Chunk]: content, section_meta = section metadata = {**base_metadata, **section_meta} # Split by paragraphs  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: # Finalize current chunk  chunk_text = "\n\n".join(current_chunk) chunks.append(Chunk(chunk_text, metadata, current_tokens)) # Start new chunk with overlap  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 # Last chunk  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 # Rough estimate  def _get_overlap(self, chunks: List[str]) -> str: if not chunks: return "" last = chunks[-1] words = last.split() overlap_words = words[-(self.overlap // 4):] # Rough token->word  return " ".join(overlap_words) 

# ingestion/enrich.py from datetime import datetime import hashlib def enrich_metadata(doc: dict, chunk: Chunk) -> dict: """Add rich metadata for filtering and debugging.""" return { **chunk.metadata, "doc_id": doc["id"], "doc_title": doc.get("title", ""), "doc_source": doc.get("source", ""), "doc_created": doc.get("created_at", ""), "chunk_hash": hashlib.md5(chunk.content.encode()).hexdigest()[:12], "ingested_at": datetime.utcnow().isoformat(), "token_count": chunk.token_count, # Custom fields for your domain  "department": doc.get("department", "general"), "confidentiality": doc.get("confidentiality", "public"), "version": doc.get("version", "1.0"), } 

# vectorstore/qdrant_client.py from qdrant_client import QdrantClient from qdrant_client.models import ( Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue, SearchParams, HnswConfigDiff ) from typing import List, Optional import uuid class QdrantVectorStore: def __init__(self, url: str, api_key: str, collection_name: str): self.client = QdrantClient(url=url, api_key=api_key) self.collection_name = collection_name def create_collection(self, vector_size: int, distance: Distance = Distance.COSINE): self.client.create_collection( collection_name=self.collection_name, vectors_config=VectorParams( size=vector_size, distance=distance, hnsw_config=HnswConfigDiff( m=16, # Connections per layer  ef_construct=100 # Build-time accuracy  ) ) ) def upsert_chunks(self, chunks: List[Chunk], embeddings: List[List[float]]): points = [ PointStruct( id=str(uuid.uuid4()), vector=emb, payload={ "content": chunk.content, "metadata": chunk.metadata } ) for chunk, emb in zip(chunks, embeddings) ] self.client.upsert( collection_name=self.collection_name, points=points ) def search( self, query_vector: List[float], limit: int = 10, filter_conditions: Optional[dict] = None, score_threshold: float = 0.7 ) -> List[dict]: query_filter = None if filter_conditions: conditions = [ FieldCondition(key=f"metadata.{k}", match=MatchValue(value=v)) for k, v in filter_conditions.items() ] query_filter = Filter(must=conditions) results = self.client.search( collection_name=self.collection_name, query_vector=query_vector, limit=limit, query_filter=query_filter, score_threshold=score_threshold, search_params=SearchParams( hnsw_ef=128, # Search-time accuracy  exact=False ) ) return [ { "content": r.payload["content"], "metadata": r.payload["metadata"], "score": r.score } for r in results ] def hybrid_search( self, query_vector: List[float], query_text: str, limit: int = 10, sparse_weight: float = 0.3 ) -> List[dict]: """Combine dense + sparse (BM25) search.""" # Requires Qdrant hybrid index setup  # This is a simplified version  dense_results = self.search(query_vector, limit * 2) # In production, use Qdrant's native hybrid search  # or combine with separate BM25 index  return dense_results[:limit] 

# retrieval/pipeline.py from typing import List, Callable from dataclasses import dataclass @dataclass class RetrievedDoc: content: str metadata: dict score: float rank: int class RetrievalPipeline: def __init__( self, embedder: Callable[[str], List[float]], vector_store, reranker: Callable[[str, List[str]], List[float]] = None ): self.embedder = embedder self.vector_store = vector_store self.reranker = reranker def retrieve( self, query: str, k: int = 5, fetch_k: int = 20, filters: dict = None, use_reranker: bool = True ) -> List[RetrievedDoc]: # Stage 1: Embed query  query_vector = self.embedder(query) # Stage 2: Vector search (fetch more for reranking)  raw_results = self.vector_store.search( query_vector=query_vector, limit=fetch_k, filter_conditions=filters ) # Stage 3: Rerank (optional but recommended)  if use_reranker and self.reranker and raw_results: contents = [r["content"] for r in raw_results] rerank_scores = self.reranker(query, contents) for r, score in zip(raw_results, rerank_scores): r["rerank_score"] = score # Sort by rerank score  raw_results.sort(key=lambda x: x.get("rerank_score", 0), reverse=True) # Stage 4: Final selection  final = raw_results[:k] return [ RetrievedDoc( content=r["content"], metadata=r["metadata"], score=r.get("rerank_score", r["score"]), rank=i ) for i, r in enumerate(final) ] # Cross-encoder reranker (best quality) from sentence_transformers import CrossEncoder reranker_model = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2") def rerank(query: str, documents: List[str]) -> List[float]: pairs = [(query, doc) for doc in documents] scores = reranker_model.predict(pairs) return scores.tolist() # MMR (Maximal Marginal Relevance) for diversity def mmr_retrieve( query_vector: List[float], candidate_vectors: List[List[float]], candidate_docs: List[RetrievedDoc], k: int = 5, lambda_mult: float = 0.5 ) -> List[RetrievedDoc]: """Select diverse results using MMR.""" import numpy as np selected = [] remaining = list(range(len(candidate_docs))) # First: highest score  best_idx = max(remaining, key=lambda i: candidate_docs[i].score) selected.append(best_idx) remaining.remove(best_idx) while len(selected) < k and remaining: mmr_scores = [] for idx in remaining: # Relevance to query  relevance = candidate_docs[idx].score # Max similarity to already selected  max_sim = max( cosine_similarity( candidat...