"The cat sat on the mat"  →  [0.12, -0.34, 0.56, ..., 0.78]  (1536 dimensions)
"A feline rested on a rug" →  [0.11, -0.33, 0.55, ..., 0.77]  (similar vector!)
"Stock prices rose today"  →  [-0.45, 0.67, -0.12, ..., 0.23]  (very different)

import openai

def get_embeddings(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    """Generate embeddings for a list of texts."""
    response = openai.embeddings.create(input=texts, model=model)
    return [item.embedding for item in response.data]

# Open-source alternative with sentence-transformers
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-large-en-v1.5")
embeddings = model.encode(["Hello world", "Semantic search is powerful"])

import psycopg2
from pgvector.psycopg2 import register_vector

conn = psycopg2.connect("postgresql://localhost/ragdb")
register_vector(conn)
cur = conn.cursor()

# Create table with vector column
cur.execute("""
    CREATE TABLE IF NOT EXISTS documents (
        id SERIAL PRIMARY KEY,
        content TEXT NOT NULL,
        metadata JSONB,
        embedding vector(1536)
    );
    CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);
""")

# Insert document with embedding
cur.execute(
    "INSERT INTO documents (content, metadata, embedding) VALUES (%s, %s, %s)",
    (chunk_text, json.dumps(metadata), embedding_vector)
)

# Query: find top-5 most similar documents
cur.execute("""
    SELECT content, metadata, 1 - (embedding <=> %s::vector) AS similarity
    FROM documents
    ORDER BY embedding <=> %s::vector
    LIMIT 5
""", (query_embedding, query_embedding))

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

client = QdrantClient(url="http://localhost:6333")

# Create collection
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

# Upsert documents
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(id=i, vector=emb, payload={"text": text, "source": source})
        for i, (emb, text, source) in enumerate(zip(embeddings, texts, sources))
    ]
)

# Search with metadata filtering
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    query_filter={"must": [{"key": "source", "match": {"value": "technical-docs"}}]},
    limit=5
)

Too small (50 tokens):   "The API returns" → lacks context, meaningless in isolation
Too large (2000 tokens): Entire page → dilutes the embedding, retrieves irrelevant content
Sweet spot (200-500):    Complete paragraph/section → meaningful, focused, retrievable

from langchain_text_splitters import RecursiveCharacterTextSplitter

# Recursive splitting — tries larger separators first, falls back to smaller
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""],  # Priority order
    length_function=len,
)

chunks = splitter.split_text(document_text)

# Semantic chunking — splits at meaning boundaries
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

semantic_splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95,  # Split when similarity drops below 95th percentile
)

semantic_chunks = semantic_splitter.split_text(document_text)

┌─────────────────────────────────────────────────────┐
│ What type of content?                               │
├──────────────────┬──────────────────────────────────┤
│ Structured docs  │ Use document-aware (headers,     │
│ (markdown, HTML) │ sections as natural boundaries)  │
├──────────────────┼──────────────────────────────────┤
│ Code             │ Split by function/class, keep    │
│                  │ imports with each chunk           │
├──────────────────┼──────────────────────────────────┤
│ Prose            │ Recursive (paragraph → sentence) │
│                  │ with 10-20% overlap              │
├──────────────────┼──────────────────────────────────┤
│ Mixed/unknown    │ Semantic chunking                │
└──────────────────┴──────────────────────────────────┘

def create_chunk_with_metadata(text: str, source: str, page: int, section: str) -> dict:
    return {
        "text": text,
        "metadata": {
            "source": source,
            "page": page,
            "section": section,
            "chunk_size": len(text),
            "created_at": datetime.utcnow().isoformat(),
        }
    }

Query: "How do I fix a memory leak in Python?"
Retrieves: "Debugging memory issues in Python applications" (semantically similar)

Query: "error code ECONNREFUSED"
BM25 finds: Documents containing exactly "ECONNREFUSED"
Dense might miss: If the embedding doesn't capture the specific error code

from rank_bm25 import BM25Okapi

# Index documents
tokenized_corpus = [doc.lower().split() for doc in documents]
bm25 = BM25Okapi(tokenized_corpus)

# Query
tokenized_query = "memory leak python".lower().split()
scores = bm25.get_scores(tokenized_query)
top_indices = scores.argsort()[-5:][::-1]

def reciprocal_rank_fusion(rankings: list[list[str]], k: int = 60) -> list[str]:
    """Combine multiple ranked lists using RRF."""
    scores = {}
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking):
            scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank + 1)

    return sorted(scores.keys(), key=lambda x: scores[x], reverse=True)

# Combine dense and sparse results
dense_results = ["doc_3", "doc_7", "doc_1", "doc_5"]
sparse_results = ["doc_7", "doc_2", "doc_3", "doc_9"]
fused = reciprocal_rank_fusion([dense_results, sparse_results])

Bi-encoder (retrieval):   embed(query) · embed(doc)     → fast but shallow
Cross-encoder (reranking): model(query + doc) → score   → slow but accurate

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")

# Score query-document pairs
query = "How to handle database migrations in production?"
documents = [retrieved_doc_1, retrieved_doc_2, retrieved_doc_3, ...]

pairs = [[query, doc] for doc in documents]
scores = reranker.predict(pairs)

# Rerank by score
reranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
top_documents = [doc for doc, score in reranked[:5]]

import cohere

co = cohere.Client("your-api-key")

results = co.rerank(
    model="rerank-v3.5",
    query="database migration strategies",
    documents=retrieved_documents,
    top_n=5,
    return_documents=True,
)

for result in results.results:
    print(f"Score: {result.relevance_score:.3f} | {result.document.text[:100]}")

┌──────────────────────────────────────────────────────────────┐
│ Stage 1: Retrieval (fast, high recall)                       │
│   Input: query → Output: top 20-50 candidates               │
│   Method: ANN search (HNSW) + optional BM25                 │
│   Latency: ~10-50ms                                         │
├──────────────────────────────────────────────────────────────┤
│ Stage 2: Reranking (slower, high precision)                  │
│   Input: 20-50 candidates → Output: top 3-5 most relevant   │
│   Method: Cross-encoder scoring                              │
│   Latency: ~100-300ms                                        │
└──────────────────────────────────────────────────────────────┘

class RAGPipeline:
    def __init__(self, vector_store, embedding_model, reranker, llm):
        self.vector_store = vector_store
        self.embedding_model = embedding_model
        self.reranker = reranker
        self.llm = llm

    def query(self, question: str, top_k: int = 5) -> dict:
        # 1: Embed the query
        query_embedding = self.embedding_model.embed(question)

        # 2: Retrieve candidates (cast wide net)
        candidates = self.vector_store.search(query_embedding, limit=20)

        # 3: Rerank for precision
        reranked = self.reranker.rerank(question, candidates, top_n=top_k)

        # 4: Assemble context
        context = "\n\n---\n\n".join([doc.text for doc in reranked])

        # 5: Generate answer
        prompt = f"""Answer the question based on the provided context.
If the context doesn't contain enough information, say so.

Context:
{context}

Question: {question}

Answer:"""

        answer = self.llm.generate(prompt)

        return {
            "answer": answer,
            "sources": [{"text": doc.text, "metadata": doc.metadata} for doc in reranked],
        }

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset

# Prepare evaluation dataset
eval_data = Dataset.from_dict({
    "question": ["What is the capital of France?", ...],
    "answer": ["Paris is the capital of France.", ...],
    "contexts": [["France is a country in Europe. Its capital is Paris."], ...],
    "ground_truth": ["The capital of France is Paris.", ...],
})

results = evaluate(
    eval_data,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)
print(results)
# {'faithfulness': 0.95, 'answer_relevancy': 0.92, 'context_precision': 0.88, ...}

Question: "What is the market cap of the company that acquired Twitter?"

Hop 1: "Twitter was acquired by X Corp (Elon Musk) in 2022"
Hop 2: "X Corp is private and not publicly traded" → need related: "Tesla market cap..."

def multi_hop_rag(question: str, max_hops: int = 3) -> str:
    """Iteratively retrieve and reason across multiple documents."""
    context_so_far = []
    current_query = question

    for hop in range(max_hops):
        # Retrieve for current query
        new_docs = retrieve(current_query)
        context_so_far.extend(new_docs)

        # Ask LLM: do we have enough to answer?
        check = llm.generate(f"""
Given this context: {format_context(context_so_far)}
Can you fully answer: {question}
If yes, provide the answer. If no, what additional information is needed?
""")

        if "ANSWER:" in check:
            return check.split("ANSWER:")[1].strip()

        # Generate follow-up query for next hop
        current_query = extract_followup_query(check)

    # Final attempt with all gathered context
    return llm.generate(f"Context: {format_context(context_so_far)}\nQuestion: {question}")

def corrective_rag(question: str) -> str:
    """RAG with retrieval quality assessment and correction."""
    # Standard retrieval
    documents = retrieve(question, top_k=5)

    # Evaluate retrieval quality
    relevance_scores = evaluate_relevance(question, documents)

    if max(relevance_scores) > 0.8:
        # High confidence: use retrieved docs
        context = filter_relevant(documents, relevance_scores, threshold=0.5)
    elif max(relevance_scores) > 0.4:
        # Medium confidence: refine with web search
        web_results = web_search(question)
        context = merge_and_deduplicate(documents, web_results)
    else:
        # Low confidence: fall back to web search entirely
        context = web_search(question)

    # Generate with corrected context
    return generate_answer(question, context)

def hyde_retrieval(question: str) -> list:
    """Hypothetical Document Embeddings — generate a fake answer, then search for real ones."""
    # Generate hypothetical answer (doesn't need to be correct)
    hypothetical = llm.generate(
        f"Write a short paragraph that would answer this question:\n{question}"
    )

    # Embed the hypothetical answer (captures the "shape" of a good answer)
    hyde_embedding = embed(hypothetical)

    # Search with the hypothetical embedding
    return vector_store.search(hyde_embedding, limit=10)

┌─────────────┐     ┌──────────┐     ┌──────────┐     ┌──────────┐
│ Data Sources│────▶│ Extract  │────▶│ Chunk    │────▶│ Embed    │
│ (docs, APIs)│     │ & Clean  │     │ & Enrich │     │ & Store  │
└─────────────┘     └──────────┘     └──────────┘     └──────────┘
                         │                                    │
                         ▼                                    ▼
                    ┌──────────┐                      ┌──────────────┐
                    │ Dedup &  │                      │ Vector DB +  │
                    │ Version  │                      │ Metadata     │
                    └──────────┘                      └──────────────┘