Task 4: RAG for Job-Specific Questions

Generic interview questions are fine, but job-specific questions are better.

• Load job descriptions, tech specs, company info
• Convert them to searchable embeddings
• Retrieve relevant context for each question
• Generate targeted questions

RAG Pipeline Overview

┌─────────────────────────────────────────────────────────────────────┐
│                         RAG Pipeline                                 │
│                                                                      │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐       │
│  │  Load    │ -> │  Split   │ -> │  Embed   │ -> │  Store   │       │
│  │  Docs    │    │  Chunks  │    │  Vectors │    │ VectorDB │       │
│  └──────────┘    └──────────┘    └──────────┘    └──────────┘       │
│                                                                      │
│  At Query Time:                                                      │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐       │
│  │  Query   │ -> │ Retrieve │ -> │  Context │ -> │ Generate │       │
│  │          │    │ Similar  │    │   + LLM  │    │ Response │       │
│  └──────────┘    └──────────┘    └──────────┘    └──────────┘       │
└─────────────────────────────────────────────────────────────────────┘

Step 1: Create Sample Job Descriptions

# data/job_descriptions/senior_python.txt
"""
Senior Python Developer - TechCorp

About the Role:
We're looking for a Senior Python Developer to join our Platform team.
You'll be building scalable microservices that process millions of events daily.

Requirements:
- 5+ years Python experience
- Strong understanding of async programming (asyncio, aiohttp)
- Experience with FastAPI or Django
- Knowledge of message queues (Kafka, RabbitMQ)
- PostgreSQL and Redis experience
- Docker and Kubernetes familiarity
- CI/CD pipeline experience

Nice to Have:
- Experience with event-driven architecture
- Knowledge of distributed systems
- GraphQL experience
- Monitoring with Prometheus/Grafana

Tech Stack:
Python 3.11, FastAPI, PostgreSQL, Redis, Kafka, Docker, Kubernetes, AWS
"""

Step 2: Load and Split Documents

# rag/loader.py
from langchain_community.document_loaders import (
    TextLoader,
    PyPDFLoader,
    Docx2txtLoader,
    DirectoryLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from pathlib import Path

def load_job_description(file_path: str):
    """Load a single job description file."""

    path = Path(file_path)

    if path.suffix == '.pdf':
        loader = PyPDFLoader(file_path)
    elif path.suffix == '.docx':
        loader = Docx2txtLoader(file_path)
    else:  # Default to text
        loader = TextLoader(file_path)

    documents = loader.load()

    # Add metadata
    for doc in documents:
        doc.metadata['source'] = path.name
        doc.metadata['type'] = 'job_description'

    return documents

def load_all_documents(directory: str):
    """Load all documents from a directory."""

    loader = DirectoryLoader(
        directory,
        glob="**/*.txt",
        loader_cls=TextLoader
    )
    return loader.load()

def split_documents(documents, chunk_size=500, chunk_overlap=50):
    """Split documents into smaller chunks for embedding."""

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", ". ", " ", ""]
    )

    return splitter.split_documents(documents)

Step 3: Create Vector Store

# rag/retriever.py
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from typing import List

def create_vector_store(documents: List[Document], persist_directory: str = None):
    """Create a vector store from documents."""

    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    if persist_directory:
        vector_store = Chroma.from_documents(
            documents=documents,
            embedding=embeddings,
            persist_directory=persist_directory
        )
    else:
        vector_store = Chroma.from_documents(
            documents=documents,
            embedding=embeddings
        )

    return vector_store

def load_vector_store(persist_directory: str):
    """Load an existing vector store."""

    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    return Chroma(
        persist_directory=persist_directory,
        embedding_function=embeddings
    )

def create_retriever(vector_store, k: int = 4):
    """Create a retriever that finds similar documents."""

    return vector_store.as_retriever(
        search_type="similarity",
        search_kwargs={"k": k}
    )

Step 4: RAG Chain for Question Generation

# chains/question_generator.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

def format_docs(docs):
    """Format retrieved documents for the prompt."""
    return "\n\n".join(doc.page_content for doc in docs)

def create_question_generator(retriever):
    """Create a RAG chain for generating interview questions."""

    prompt = ChatPromptTemplate.from_messages([
        ("system", """You are an expert technical interviewer.

Generate a relevant interview question based on the job requirements below.
The question should:
- Test specific skills mentioned in the job description
- Be appropriate for the candidate's level
- Be clear and focused on one topic

Job Requirements Context:
{context}
"""),
        ("human", """Generate a {difficulty} level question about {topic}.

Previous questions asked: {previous_questions}

Ensure this question is different from previous ones.""")
    ])

    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

    chain = (
        {
            "context": lambda x: format_docs(retriever.invoke(x["topic"])),
            "difficulty": lambda x: x["difficulty"],
            "topic": lambda x: x["topic"],
            "previous_questions": lambda x: x.get("previous_questions", "None")
        }
        | prompt
        | llm
        | StrOutputParser()
    )

    return chain

Step 5: Complete RAG Setup

# rag/setup.py
from rag.loader import load_job_description, split_documents
from rag.retriever import create_vector_store, create_retriever
from chains.question_generator import create_question_generator

def setup_interview_rag(job_description_path: str):
    """Complete setup for RAG-powered interviews."""

    # 1. Load documents
    print("Loading job description...")
    docs = load_job_description(job_description_path)

    # 2. Split into chunks
    print("Splitting into chunks...")
    chunks = split_documents(docs, chunk_size=300, chunk_overlap=30)
    print(f"Created {len(chunks)} chunks")

    # 3. Create vector store
    print("Creating embeddings and vector store...")
    vector_store = create_vector_store(chunks)

    # 4. Create retriever
    retriever = create_retriever(vector_store, k=3)

    # 5. Create question generator
    question_generator = create_question_generator(retriever)

    return {
        "vector_store": vector_store,
        "retriever": retriever,
        "question_generator": question_generator
    }

Step 6: Generate Job-Specific Questions

# main.py
from rag.setup import setup_interview_rag

def run_rag_interview():
    # Setup RAG
    rag_components = setup_interview_rag("data/job_descriptions/senior_python.txt")
    question_gen = rag_components["question_generator"]

    # Topics to cover (could also be extracted from JD)
    topics = [
        "async programming",
        "FastAPI",
        "message queues",
        "database design",
        "containerization"
    ]

    previous_questions = []

    print("=" * 50)
    print("RAG-Powered Interview Questions")
    print("=" * 50)

    for i, topic in enumerate(topics, 1):
        question = question_gen.invoke({
            "topic": topic,
            "difficulty": "senior",
            "previous_questions": ", ".join(previous_questions) if previous_questions else "None"
        })

        print(f"\nQ{i} ({topic}): {question}")
        previous_questions.append(question[:50] + "...")

        answer = input("\nYour answer: ")
        print("-" * 30)

run_rag_interview()

Q1 (async programming): Given that our platform processes millions of
events daily using asyncio, can you explain how you would design an
async pipeline that handles backpressure when downstream services
slow down? What patterns would you use?

Q2 (FastAPI): Our services use FastAPI extensively. How would you
implement a rate-limiting middleware that works correctly with
async request handlers? Walk me through your approach.

Step 7: Extract Topics from Job Description

Let the LLM identify topics to cover:

# chains/topic_extractor.py
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List

class InterviewTopics(BaseModel):
    """Topics to cover in the interview."""
    must_have: List[str] = Field(description="Required skills to assess")
    nice_to_have: List[str] = Field(description="Optional skills to assess")
    soft_skills: List[str] = Field(description="Soft skills to evaluate")

def create_topic_extractor():
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    return llm.with_structured_output(InterviewTopics)

# Usage
extractor = create_topic_extractor()
topics = extractor.invoke("""Extract interview topics from this JD:

Senior Python Developer - requires async programming, FastAPI,
message queues, PostgreSQL. Nice to have: distributed systems, GraphQL.
""")

print(topics.must_have)
# ['async programming', 'FastAPI', 'message queues', 'PostgreSQL']

Advanced: Multi-Source RAG

Combine multiple sources:

# rag/multi_source.py
def setup_multi_source_rag(sources: dict):
    """Setup RAG with multiple document sources."""

    all_docs = []

    for source_type, path in sources.items():
        docs = load_job_description(path)
        for doc in docs:
            doc.metadata['source_type'] = source_type
        all_docs.extend(docs)

    chunks = split_documents(all_docs)
    vector_store = create_vector_store(chunks)

    return create_retriever(vector_store, k=5)

# Usage
retriever = setup_multi_source_rag({
    "job_description": "data/jd.txt",
    "company_values": "data/values.txt",
    "tech_stack": "data/tech_stack.md"
})

Key Takeaways

• Document Loaders handle various file formats
• Text Splitters chunk documents for better retrieval
• Embeddings convert text to vectors for similarity search
• Vector Stores (Chroma, Pinecone, etc.) store and search embeddings
• Retrievers find relevant context for queries
• RAG Chains combine retrieval with generation

Next Up