#Capstone Exercise:
A capstone project is a comprehensive, culminating academic assignment that learners complete at the end of a course such as GenAI training. It requires you to apply the skills and knowledge you've acquired throughout your studies to investigate, design a solution for, or evaluate a specific, real-world problem or research question
This Capstone project demonstrates:
✅ Chroma vector store
✅ text-embedding-3-small embeddings
✅ GPT generation
✅ Semantic retrieval
✅ Semantic + threshold retrieval
✅ Hybrid retrieval (BM25 + semantic)
✅ Hallucination-resistant prompt
✅ No-answer fallback
✅ Top-K ≤ 3
✅ Outputs submission.csv
#!/usr/bin/python
import os
import json
import numpy as np
import pandas as pd
from dotenv import load_dotenv
from tenacity import (
retry,
stop_after_attempt,
wait_random_exponential
)
from langchain_community.document_loaders import CSVLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import (
AzureOpenAIEmbeddings,
AzureChatOpenAI
)
from rank_bm25 import BM25Okapi
# ============================================================
# CONFIGURATION
# ============================================================
load_dotenv("./Data/vars.env")
DATASET_FILE = "./Data/capstone1_rag_dataset.csv"
TEST_FILE = "./Data/capstone1_rag_test_questions.csv"
VECTOR_DB_DIR = "./chroma_capstone_db"
AZURE_OPENAI_ENDPOINT = os.environ["MODEL_ENDPOINT"]
OPENAI_API_VERSION = os.environ["API_VERSION"]
CHAT_DEPLOYMENT_NAME = os.environ["MODEL_NAME"]
PROJECT_ID = os.environ["PROJECT_ID"]
EMBEDDINGS_DEPLOYMENT_NAME = os.environ["EMBEDDINGS_DEPLOYMENT_NAME "]
# Required by Chroma in many enterprise environments
os.environ["ANONYMIZED_TELEMETRY"] = "False"
# ============================================================
# AUTHENTICATION
# ============================================================
def get_access_token():
auth = "https://<your-provider-endpoint>/oauth2/token"
scope = "https:// <your-provider-endpoint>/.default"
grant_type = "client_credentials"
with httpx.Client() as client:
body = {
"grant_type": grant_type,
"scope": scope,
"client_id": dbutils.secrets.get(scope="AIML_Training", key="client_id"),
"client_secret": dbutils.secrets.get(scope="AIML_Training", key="client_secret"),
}
headers = {"Content-Type": "application/x-www-form-urlencoded"}
resp = client.post(auth, headers=headers, data=body, timeout=60)
access_token = resp.json()["access_token"]
return access_token
# ============================================================
# MODELS
# ============================================================
embeddings = AzureOpenAIEmbeddings(
azure_deployment=EMBEDDINGS_DEPLOYMENT_NAME,
azure_endpoint=AZURE_OPENAI_ENDPOINT,
api_version=OPENAI_API_VERSION,
azure_ad_token_provider=get_access_token,
default_headers={
"projectId": PROJECT_ID,
"model-usage-type": "prod"
}
)
llm = AzureChatOpenAI(
azure_deployment=CHAT_DEPLOYMENT_NAME,
azure_endpoint=AZURE_OPENAI_ENDPOINT,
api_version=OPENAI_API_VERSION,
azure_ad_token_provider=get_access_token,
default_headers={
"projectId": PROJECT_ID,
"model-usage-type": "prod"
},
temperature=0.1
)
# ============================================================
# DATA LOADING
# ============================================================
def load_dataset():
loader = CSVLoader(
file_path=DATASET_FILE,
encoding="utf-8"
)
return loader.load()
# ============================================================
# VECTOR STORE
# ============================================================
@retry(
wait=wait_random_exponential(min=2, max=30),
stop=stop_after_attempt(5),
reraise=True
)
def build_vector_store(documents):
return Chroma.from_documents(
documents=documents,
embedding=embeddings,
persist_directory=VECTOR_DB_DIR
)
# ============================================================
# BM25 INDEX
# ============================================================
def build_bm25_index(documents):
corpus = [
doc.page_content
for doc in documents
]
tokenized = [
text.lower().split()
for text in corpus
]
bm25 = BM25Okapi(tokenized)
return bm25, corpus
# ============================================================
# RETRIEVAL STRATEGY #1
# Semantic Search
# ============================================================
def semantic_retrieval(
query,
vectorstore,
top_k=3
):
results = vectorstore.similarity_search(
query,
k=top_k
)
docs = [
doc.page_content
for doc in results
]
return docs
# ============================================================
# RETRIEVAL STRATEGY #2
# Semantic + Threshold Filtering
# ============================================================
def threshold_retrieval(
query,
vectorstore,
threshold=0.70,
top_k=3
):
try:
results = vectorstore.similarity_search_with_relevance_scores(
query,
k=10
)
filtered_docs = []
for doc, score in results:
if score >= threshold:
filtered_docs.append(
doc.page_content
)
return filtered_docs[:top_k]
except Exception:
return semantic_retrieval(
query,
vectorstore,
top_k
)
# ============================================================
# RETRIEVAL STRATEGY #3
# Hybrid BM25 + Semantic
# ============================================================
def hybrid_retrieval(
query,
vectorstore,
bm25,
corpus,
top_k=3
):
semantic_results = vectorstore.similarity_search(
query,
k=10
)
semantic_texts = {
doc.page_content
for doc in semantic_results
}
bm25_scores = bm25.get_scores(
query.lower().split()
)
ranked_idx = np.argsort(
bm25_scores
)[::-1][:10]
bm25_texts = {
corpus[idx]
for idx in ranked_idx
}
combined_docs = list(
semantic_texts.union(
bm25_texts
)
)
scored_docs = []
query_embedding = embeddings.embed_query(
query
)
for doc_text in combined_docs:
try:
doc_embedding = embeddings.embed_query(
doc_text[:8000]
)
cosine = np.dot(
query_embedding,
doc_embedding
) / (
np.linalg.norm(query_embedding)
* np.linalg.norm(doc_embedding)
)
scored_docs.append(
(
doc_text,
float(cosine)
)
)
except Exception:
pass
scored_docs.sort(
key=lambda x: x[1],
reverse=True
)
return [
doc
for doc, _
in scored_docs[:top_k]
]
# ============================================================
# GENERATION
# ============================================================
@retry(
wait=wait_random_exponential(min=2, max=30),
stop=stop_after_attempt(5),
reraise=True
)
def generate_answer(
query,
retrieved_docs
):
if len(retrieved_docs) == 0:
return (
"The question cannot be answered "
"using the available documents."
)
context = "\n\n".join(
retrieved_docs
)
prompt = f"""
You are a clinical intelligence assistant.
IMPORTANT RULES:
1. Use ONLY the provided context.
2. Do NOT use prior medical knowledge.
3. Do NOT hallucinate.
4. If the answer is not present in the context,
say:
"The question cannot be answered using the available documents."
5. Cite information only from context.
6. Keep responses concise and factual.
CONTEXT:
{context}
QUESTION:
{query}
ANSWER:
"""
response = llm.invoke(
prompt
)
return response.content
# ============================================================
# MAIN RAG PIPELINE
# ============================================================
def rag_pipeline(
query,
vectorstore,
bm25,
corpus,
retrieval_strategy="hybrid"
):
if retrieval_strategy == "semantic":
docs = semantic_retrieval(
query,
vectorstore
)
elif retrieval_strategy == "threshold":
docs = threshold_retrieval(
query,
vectorstore
)
else:
docs = hybrid_retrieval(
query,
vectorstore,
bm25,
corpus
)
answer = generate_answer(
query,
docs
)
return {
"retrieved_documents": docs,
"generated_answer": answer
}
# ============================================================
# MAIN
# ============================================================
if __name__ == "__main__":
print("Loading dataset...")
documents = load_dataset()
print(
f"Documents Loaded: {len(documents)}"
)
print("Building vector store...")
vectorstore = build_vector_store(
documents
)
print("Building BM25 index...")
bm25, corpus = build_bm25_index(
documents
)
print("Loading questions...")
questions_df = pd.read_csv(
TEST_FILE,
dtype=str
).fillna("")
questions_df[
"retrieved_documents"
] = ""
questions_df[
"generated_answer"
] = ""
for idx, row in questions_df.iterrows():
question = row["question"]
print("\n" + "=" * 80)
print(
f"QUESTION {idx + 1}:"
)
print(question)
result = rag_pipeline(
query=question,
vectorstore=vectorstore,
bm25=bm25,
corpus=corpus,
retrieval_strategy="hybrid"
)
print("\nANSWER:")
print(
result["generated_answer"]
)
questions_df.loc[
idx,
"retrieved_documents"
] = json.dumps(
result[
"retrieved_documents"
]
)
questions_df.loc[
idx,
"generated_answer"
] = result[
"generated_answer"
]
submission = questions_df[
[
"question",
"retrieved_documents",
"generated_answer"
]
]
submission.to_csv(
"submission.csv",
index=False
)
print("\nsubmission.csv created.")
print(
f"Rows: {len(submission)}"
)
# ============================================================
# SAMPLE OUTPUT
# ============================================================
Loading dataset...
Creating vector store...
Failed to send telemetry event ClientStartEvent: capture() takes 1 positional argument but 3 were given
Failed to send telemetry event ClientCreateCollectionEvent: capture() takes 1 positional argument but 3 were given
Loading questions...
Processing: What are the key features of …
Failed to send telemetry event CollectionQueryEvent: capture() takes 1 positional argument but 3 were given
document_id: 94
document_url: https://...
context: Auto…
---
document_id: 769
document_url: https://...
context: Palm…
---
document_id: 784
document_url: https://...
context: La...
questions_df.loc[idx, "retrieved_documents"]
questions_df.loc[idx, "generated_answer"]
No comments:
Post a Comment