Capstone Project: Enterprise Technical Q&A System

In this hands-on capstone project, you will build an end-to-end Multi-Source Technical Documentation Q&A Engine with LCEL, real-time token streaming, and verified source metadata attribution.

1. System Requirements

The system must:

1. Ingest technical data platform reference guides.
2. Index chunks into a vector store with system category metadata (spark, snowflake, airflow).
3. Accept user questions with optional metadata filtering.
4. Execute an LCEL RAG chain that streams the answer with exact source citations.

flowchart LR
    Q["User Question"] --> Filter["Metadata Filter Tagging"]
    Filter --> Retriever["Vector Store Retriever"]
    Retriever --> LCEL["LCEL Prompt & Model Chain"]
    LCEL --> Stream["Live Streaming Token Generator"]

    style Q fill:#334155,color:#fff
    style Filter fill:#2563eb,color:#fff
    style Retriever fill:#059669,color:#fff
    style LCEL fill:#7c3aed,color:#fff
    style Stream fill:#10b981,color:#fff

2. Complete Runnable Implementation

from typing import List, Dict, Generator

# Step 1: Simulated Technical Knowledge Base
TECH_DOCS = [
    {
        "id": "doc_sf_01",
        "system": "snowflake",
        "title": "Snowflake Clustering Best Practices",
        "content": "For tables over 1TB with high query volumes, define cluster keys on columns frequently used in WHERE and JOIN clauses (e.g. order_date)."
    },
    {
        "id": "doc_spark_01",
        "system": "spark",
        "title": "PySpark Adaptive Query Execution",
        "content": "Adaptive Query Execution (AQE) dynamically coalesces shuffle partitions, converts sort-merge joins to broadcast joins, and handles data skew at runtime."
    },
    {
        "id": "doc_af_01",
        "system": "airflow",
        "title": "Airflow Dynamic Task Mapping",
        "content": "Dynamic task mapping (expand) allows Airflow tasks to generate parallel execution instances at runtime based on the output of upstream tasks."
    }
]

class TechnicalQASystem:
    def __init__(self, documents: List[Dict]):
        self.docs = documents

    def retrieve(self, query: str, target_system: str = None) -> List[Dict]:
        """Semantic search simulation with metadata pre-filtering."""
        pool = self.docs
        if target_system:
            pool = [d for d in pool if d["system"] == target_system.lower()]

        # Return top matches
        return pool[:2]

    def stream_answer(self, question: str, target_system: str = None) -> Generator[str, None, None]:
        matched_chunks = self.retrieve(question, target_system)

        context_str = "

".join([
            f"[Source: {d['title']} ({d['system'].upper()})]
{d['content']}"
            for d in matched_chunks
        ])

        simulated_response = (
            f"Based on our internal {target_system.upper() if target_system else 'data platform'} documentation:
"
            f"{matched_chunks[0]['content']}

"
            f"Citation: [{matched_chunks[0]['title']}]"
        )

        # Simulate real-time token streaming generator
        for token in simulated_response.split(" "):
            yield token + " "

if __name__ == "__main__":
    qa_engine = TechnicalQASystem(TECH_DOCS)

    print("=== Streaming Query: Spark AQE ===")
    stream = qa_engine.stream_answer(
        question="How does Spark handle data skew dynamically?",
        target_system="spark"
    )
    for token in stream:
        print(token, end="", flush=True)
    print("
")

3. Production Deployment Notes

• Expose the streaming generator using FastAPI StreamingResponse with media_type="text/event-stream" (Server-Sent Events).
• Connect frontend clients using the standard JavaScript EventSource API for a ChatGPT-like streaming typing effect.