RAG Coding Questions Python Student Notes
RAG Coding Questions Python Student Notes
RAG CODING
WITH PYTHON
Learning objectives
By the end of these notes, a learner can build a small Retrieval-Augmented Generation (RAG) application that
uploads documents, stores their meaning as vectors, retrieves relevant text, and asks an LLM to answer only from
that text.
What is RAG?
RAG means Retrieval-Augmented Generation. Before the LLM writes an answer, the application searches a
private document collection and supplies the most relevant passages as context. This improves grounding and
lets the response show its source.
Core flow
Document -> extracted pages -> chunks -> embeddings -> vector database. A question follows the same
embedding step, then similarity search returns context, and the LLM produces the final grounded answer.
Project setup
Recommended project structure
rag-project/
|-- [Link]
|-- uploads/
|-- chroma_data/
`-- .env
Installation commands
py -m venv venv
venv\Scripts\activate
py -m pip install fastapi uvicorn python-multipart openai chromadb pymupdf python-docx
python-dotenv
Never paste a real API key into source code or commit it to Git.
import chromadb
client = OpenAI()
chroma_client = [Link](path='./chroma_data')
collection = chroma_client.get_or_create_collection(name='documents')
def load_pdf(file_path):
pages = []
pdf = [Link](file_path)
[Link]()
return pages
pages = load_pdf('[Link]')
print(pages[0]['text'])
Explanation
The function returns a list of dictionaries. Every dictionary contains page text and its human-readable page
number. Keeping page numbers now makes source display easy later.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
return chunks
Explanation
Here, each chunk has at most 500 characters. The next chunk starts 50 characters before the previous one
ended. In production, token-aware or sentence-aware splitting is usually better.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
Explanation
The function sends one text string to the embedding model and returns its vector. Use the same model for
document chunks and questions.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
[Link](
ids=['chunk-1', 'chunk-2'],
documents=chunks,
embeddings=embeddings,
metadatas=[
{'source': '[Link]', 'page': 1, 'category': 'python'},
{'source': '[Link]', 'page': 2, 'category': 'api'}
]
)
print('Stored successfully')
Explanation
documents keeps readable passages; embeddings enables semantic search; metadatas keeps filename, page
and category; ids identifies every stored chunk.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
Explanation
The first five numbers are printed only for demonstration. The complete vector is sent to ChromaDB for
comparison.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
top_chunks = results['documents'][0]
Explanation
n_results=3 means top-k is three. The first [0] is required because ChromaDB can accept multiple questions in
one query and returns one result list per question.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
prompt = f'''
Use the context to answer the question.
Context:
{context}
Question:
{question}
'''
response = [Link](
model='gpt-5.5',
input=prompt
)
answer = response.output_text
print(answer)
Explanation
Retrieval and generation are separate steps: ChromaDB selects evidence; the LLM turns that evidence into a
natural-language answer. Use a text model available to your OpenAI project.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
seen = set()
for item in results['metadatas'][0]:
source = f"{item['source']} - Page {item['page']}"
if source not in seen:
print(source)
[Link](source)
Explanation
The set prevents the same filename and page from appearing repeatedly when two retrieved chunks came from
the same page.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
history = '\n'.join(
f"{m['role']}: {m['content']}"
for m in conversation_memory
)
response = [Link](
model='gpt-5.5',
input=f'''History:
{history}
Context:
{context}
Question:
{question}
Answer only from context.'''
)
answer = response.output_text
conversation_memory.append({
'role': 'assistant', 'content': answer
})
return answer
Explanation
This is teaching-level memory. It disappears when the application restarts and is shared by all users. A real
application should store memory by session ID in a database.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
app = FastAPI()
[Link]('uploads', exist_ok=True)
@[Link]('/upload')
async def upload(file: UploadFile = File(...)):
path = [Link]('uploads', [Link])
pages = load_document(path)
count = 0
Explanation
UploadFile receives multipart file data. UUID creates a unique ID for every chunk. The helper load_document is
defined in Question 12.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
class QuestionRequest(BaseModel):
question: str
@[Link]('/ask')
def ask(request: QuestionRequest):
vector = create_embedding([Link])
results = [Link](
query_embeddings=[vector], n_results=3
)
context = '\n\n'.join(results['documents'][0])
prompt = f'''Answer only from this context.
{context}
Question: {[Link]}
If missing, say: Answer not found in the uploaded documents.'''
response = [Link](
model='gpt-5.5', input=prompt
)
return {
'answer': response.output_text,
'sources': results['metadatas'][0]
}
Explanation
A POST body such as {"question": "What is RAG?"} is validated by Pydantic. The response contains evidence
metadata for traceability.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
def load_document(file_path):
ext = file_path.lower().split('.')[-1]
if ext == 'pdf':
return load_pdf(file_path)
if ext == 'docx':
doc = Document(file_path)
text = '\n'.join([Link] for p in [Link])
return [{'text': text, 'page': 1}]
if ext == 'txt':
with open(file_path, encoding='utf-8') as file:
return [{'text': [Link](), 'page': 1}]
Explanation
PDF has natural pages. A basic DOCX or TXT loader treats the entire file as page 1. More advanced DOCX
source tracking can use headings or paragraph numbers.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
Rules:
1. Use only the supplied context.
2. Do not use outside knowledge.
3. Do not guess.
4. If the answer is missing, respond exactly:
Answer not found in the uploaded documents.
Context:
{context}
Question:
{question}
'''
Explanation
Prompting reduces unsupported answers but cannot mathematically guarantee zero hallucination. Good
chunking, retrieval thresholds, evaluations and source display provide additional protection.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
return {
'message': 'Document and embeddings deleted',
'filename': filename
}
Explanation
The metadata filter deletes every chunk belonging to that document. In a production application, prefer an internal
document ID to avoid ambiguity between files with the same name.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
Explanation
Filtering is useful when the user selects a document, department, course or category. Store that value in
metadata during upload so it is available during search.
Student checkpoint
Run the snippet, print its output, and explain what data enters and leaves this step.
db = [Link](path="./chroma_data")
collection = db.get_or_create_collection(name="documents")
class QuestionRequest(BaseModel):
question: str
def load_pdf(path):
pages = []
pdf = [Link](path)
for index, page in enumerate(pdf):
[Link]({"text": page.get_text(), "page": index + 1})
[Link]()
return pages
def load_document(path):
ext = [Link]().split(".")[-1]
if ext == "pdf":
return load_pdf(path)
if ext == "docx":
doc = Document(path)
text = "\n".join([Link] for p in [Link])
return [{"text": text, "page": 1}]
if ext == "txt":
with open(path, encoding="utf-8") as file:
return [{"text": [Link](), "page": 1}]
raise ValueError("Only PDF, DOCX and TXT are supported")
def create_embedding(text):
result = [Link](
model="text-embedding-3-small", input=text
)
return [Link][0].embedding
@[Link]("/upload")
async def upload(file: UploadFile = File(...)):
path = [Link]("uploads", [Link])
with open(path, "wb") as output:
[Link](await [Link]())
count = 0
for page in load_document(path):
for chunk in split_text(page["text"]):
if not [Link]():
continue
[Link](
ids=[str(uuid.uuid4())],
documents=[chunk],
embeddings=[create_embedding(chunk)],
metadatas=[{"source": [Link], "page": page["page"]}]
)
count += 1
@[Link]("/ask")
def ask(request: QuestionRequest):
results = [Link](
query_embeddings=[create_embedding([Link])],
n_results=3
)
context = "\n\n".join(results["documents"][0])
prompt = f"""Use only the context below.
If the answer is missing, say: Answer not found in the uploaded documents.
Context:
{context}
Question:
{[Link]}"""
response = [Link](model="gpt-5.5", input=prompt)
return {"answer": response.output_text,
"sources": results["metadatas"][0]}
@[Link]("/documents/{filename}")
def delete_document(filename: str):
[Link](where={"source": filename})
path = [Link]("uploads", filename)
if [Link](path):
[Link](path)
return {"message": "Deleted successfully", "filename": filename}
2. Open Swagger UI
[Link]
3. Test in order
• POST /upload: choose a PDF, DOCX or TXT file.
• Read the answer and verify its filename and page metadata.
Common errors
Problem Likely solution
Model not found Choose a text model available in your OpenAI project.
• Why must questions and chunks use the same embedding model?
Hands-on assignments
• Add a category field to POST /upload and save it in metadata.
Quick revision
RAG = Retrieve relevant chunks + Augment the prompt + Generate a grounded answer. The vector database does
not write the final answer; it finds evidence. The LLM does not search ChromaDB automatically; application code
retrieves and supplies the context.
References
OpenAI Vector Embeddings: [Link]