Module 3 Hands-On Coding Lab & Practical Scenarios

from typing import List, Dict

def reciprocal_rank_fusion(vector_rankings: List[str], keyword_rankings: List[str], k: int = 60) -> List[Dict]:
    """
    Merge ranked document IDs from vector search and keyword search
    using Reciprocal Rank Fusion (RRF).
    """
    scores = {}

    # Process vector rankings (1-indexed rank)
    for rank, doc_id in enumerate(vector_rankings, start=1):
        if doc_id not in scores:
            scores[doc_id] = 0.0
        scores[doc_id] += 1.0 / (k + rank)

    # Process keyword rankings
    for rank, doc_id in enumerate(keyword_rankings, start=1):
        if doc_id not in scores:
            scores[doc_id] = 0.0
        scores[doc_id] += 1.0 / (k + rank)

    # Sort by descending RRF score
    sorted_docs = sorted(scores.items(), key=lambda item: item[1], reverse=True)

    return [{"doc_id": doc_id, "rrf_score": round(score, 6)} for doc_id, score in sorted_docs]

# Example Demonstration
vec_search = ["doc_A", "doc_B", "doc_C", "doc_D"]
kw_search  = ["doc_B", "doc_E", "doc_A", "doc_F"]

merged = reciprocal_rank_fusion(vec_search, kw_search)
print("=== Reciprocal Rank Fusion Results ===")
for item in merged:
    print(f"Document: {item['doc_id']} | Score: {item['rrf_score']}")
# Notice how 'doc B' wins the top rank because it scored rank 1 in keyword and rank 2 in vector!

import numpy as np
from typing import List

def mock_sentence_vector(text: str) -> np.ndarray:
    """Simulates sentence embeddings based on topical keywords."""
    v = np.zeros(4) # [pyspark, database, cooking, weather]
    t = text.lower()
    if any(w in t for w in ["spark", "dataframe", "shuffle", "executor"]):
        v[0] = 1.0
    if any(w in t for w in ["sql", "postgres", "table", "schema"]):
        v[1] = 1.0
    if any(w in t for w in ["recipe", "dough", "bake", "pizza"]):
        v[2] = 1.0
    if any(w in t for w in ["rain", "sunny", "cloudy", "weather"]):
        v[3] = 1.0

    norm = np.linalg.norm(v)
    return v / norm if norm > 0 else np.array([0.25, 0.25, 0.25, 0.25])

def cosine_similarity(u, v):
    return float(np.dot(u, v) / (np.linalg.norm(u) * np.linalg.norm(v)))

def split_by_semantic_distance(sentences: List[str], threshold: float = 0.5) -> List[List[str]]:
    if not sentences:
        return []

    chunks = []
    current_chunk = [sentences[0]]

    for i in range(len(sentences) - 1):
        v1 = mock_sentence_vector(sentences[i])
        v2 = mock_sentence_vector(sentences[i + 1])
        sim = cosine_similarity(v1, v2)

        # If semantic similarity drops below threshold, start a new chunk
        if sim < threshold:
            chunks.append(current_chunk)
            current_chunk = [sentences[i + 1]]
        else:
            current_chunk.append(sentences[i + 1])

    if current_chunk:
        chunks.append(current_chunk)
    return chunks

test_sentences = [
    "PySpark DataFrames run on distributed executors across the cluster.",
    "Shuffle operations move data partitions across the network.",
    "Making sourdough pizza requires high hydration bread flour.",
    "Bake the crust at 500 degrees Fahrenheit for 8 minutes.",
    "PostgreSQL tables should define foreign keys for referential integrity."
]

semantic_chunks = split_by_semantic_distance(test_sentences, threshold=0.5)
for idx, chunk in enumerate(semantic_chunks, 1):
    print(f"
--- Semantic Chunk {idx} ---")
    for s in chunk:
        print(f"  • {s}")

import numpy as np
from typing import List, Dict

DOCUMENTS_STORE = [
    {"id": "doc_1", "tenant_id": "ORG_42", "text": "Annual revenue report for ORG_42"},
    {"id": "doc_2", "tenant_id": "ORG_99", "text": "Confidential acquisition strategy for ORG_99"},
    {"id": "doc_3", "tenant_id": "ORG_42", "text": "Quarterly tax filings for ORG_42"}
]

def mock_vector(text: str) -> np.ndarray:
    np.random.seed(abs(hash(text)) % 2**32)
    v = np.random.randn(32)
    return v / np.linalg.norm(v)

def secure_multi_tenant_search(query: str, tenant_id: str, top_k: int = 2) -> List[Dict]:
    q_vec = mock_vector(query)

    # 1. HARD PRE-FILTER: Never compute similarity on unauthorized tenant records!
    allowed_docs = [d for d in DOCUMENTS_STORE if d["tenant_id"] == tenant_id]

    # 2. Score only authorized documents
    ranked = []
    for doc in allowed_docs:
        d_vec = mock_vector(doc["text"])
        score = float(np.dot(q_vec, d_vec))
        ranked.append((doc, score))

    ranked.sort(key=lambda x: x[1], reverse=True)
    return [{"doc": d, "score": s} for d, s in ranked[:top_k]]

results = secure_multi_tenant_search("acquisition strategy", tenant_id="ORG_42")
print("Search Results for ORG_42 (Notice ORG_99 document is completely absent):")
for r in results:
    print(f"ID: {r['doc']['id']} | Tenant: {r['doc']['tenant_id']} | Text: {r['doc']['text']}")