Building Your First RAG System with Python and OpenAI

rag-project/ │ ├── src/ │ ├── __init__.py │ ├── document_loader.py │ ├── text_processor.py │ ├── embeddings_manager.py │ ├── retrieval_system.py │ └── rag_system.py │ ├── data/ │ └── documents/ │ ├── requirements.txt ├── test.py ├── README.md └── .env 

python -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate pip install openai python-dotenv numpy pandas 

openai==1.12.0 python-dotenv==1.0.0 numpy==1.24.3 pandas==2.1.0 

OPENAI_API_KEY=your_api_key_here 

import os from typing import List class DocumentLoader: def __init__(self, documents_path: str): self.documents_path = documents_path def load_documents(self) -> List[str]: documents = [] for filename in os.listdir(self.documents_path): if filename.endswith('.txt'): with open(os.path.join(self.documents_path, filename), 'r') as file: documents.append(file.read()) return documents 

from typing import List class TextProcessor: def __init__(self, chunk_size: int = 1000): self.chunk_size = chunk_size def split_into_chunks(self, text: str) -> List[str]: words = text.split() chunks = [] current_chunk = [] current_size = 0 for word in words: if current_size + len(word) > self.chunk_size: chunks.append(' '.join(current_chunk)) current_chunk = [word] current_size = len(word) else: current_chunk.append(word) current_size += len(word) + 1 if current_chunk: chunks.append(' '.join(current_chunk)) return chunks 

from typing import List import openai import numpy as np class EmbeddingsManager: def __init__(self, api_key: str): openai.api_key = api_key def create_embeddings(self, texts: List[str]) -> List[np.ndarray]: embeddings = [] for text in texts: response = openai.embeddings.create( model="text-embedding-ada-002", input=text ) embeddings.append(np.array(response.data[0].embedding)) return embeddings 

import numpy as np from typing import List, Tuple class RetrievalSystem: def __init__(self, chunks: List[str], embeddings: List[np.ndarray]): self.chunks = chunks self.embeddings = embeddings def find_similar_chunks(self, query_embedding: np.ndarray, top_k: int = 3) -> List[Tuple[str, float]]: similarities = [] for i, embedding in enumerate(self.embeddings): similarity = np.dot(query_embedding, embedding) / ( np.linalg.norm(query_embedding) * np.linalg.norm(embedding) ) similarities.append((self.chunks[i], similarity)) return sorted(similarities, key=lambda x: x[1], reverse=True)[:top_k] 

import os from dotenv import load_dotenv from typing import List import openai from .document_loader import DocumentLoader from .text_processor import TextProcessor from .embeddings_manager import EmbeddingsManager from .retrieval_system import RetrievalSystem class RAGSystem: def __init__(self): load_dotenv() self.api_key = os.getenv('OPENAI_API_KEY') self.loader = DocumentLoader('data/documents') self.processor = TextProcessor() self.embeddings_manager = EmbeddingsManager(self.api_key) # Initialize system  self.initialize_system() def initialize_system(self): # Load and process documents  documents = self.loader.load_documents() self.chunks = [] for doc in documents: self.chunks.extend(self.processor.split_into_chunks(doc)) # Create embeddings  self.embeddings = self.embeddings_manager.create_embeddings(self.chunks) # Initialize retrieval system  self.retrieval_system = RetrievalSystem(self.chunks, self.embeddings) def answer_question(self, question: str) -> str: # Get question embedding  question_embedding = self.embeddings_manager.create_embeddings([question])[0] # Get relevant chunks  relevant_chunks = self.retrieval_system.find_similar_chunks(question_embedding) # Prepare context  context = "\n".join([chunk[0] for chunk in relevant_chunks]) # Create prompt  prompt = f"""Context: {context}\n\nQuestion: {question}\n\nAnswer:""" # Get response from OpenAI  response = openai.chat.completions.create( model="gpt-4-turbo-preview", messages=[ {"role": "system", "content": "You are a helpful assistant. Use the provided context to answer the question."}, {"role": "user", "content": prompt} ] ) return response.choices[0].message.content 

story.txt 

# test.py from src.rag_system import RAGSystem # Initialize the RAG system rag = RAGSystem() # Ask a question question = "What was the answer to the guardian’s riddle, and how did it help Kai?" answer = rag.answer_question(question) print(answer)