zenith / src /document_processing /chunking.py
hello-yaaash25's picture
Initial commit without binary files
29c4d1d
Raw History Blame Contribute Delete
1.93 kB
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents.base import Document
from typing import List, Optional
import os
import chardet
import PyPDF2
import io
def chunk_text(text, chunk_size=1000, overlap=100):
chunks = []
start = 0
while start < len(text):
end = min(start + chunk_size, len(text))
chunk = text[start:end]
chunks.append(chunk)
start += chunk_size - overlap # Move the start index for the next chunk
return chunks
def process_document(document_path: str) -> List[str]:
file_extension = os.path.splitext(document_path)[1].lower()
if file_extension == '.pdf':
text = ""
with open(document_path, 'rb') as file:
pdf_reader = PyPDF2.PdfReader(file)
for page in pdf_reader.pages:
text += page.extract_text() + "\n"
return chunk_text(text)
else:
with open(document_path, 'rb') as file:
raw_data = file.read()
result = chardet.detect(raw_data)
encoding = result['encoding'] if result['confidence'] > 0.7 else 'latin-1'
try:
content = raw_data.decode(encoding)
except UnicodeDecodeError:
content = raw_data.decode('latin-1')
return chunk_text(content)
def process_text_file_and_store_vectors(file_path: str, chunk_size: int, vector_store_path: str):
chunks = process_document(file_path)
documents = [Document(page_content=chunk, metadata={"source": file_path}) for chunk in chunks]
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2",
model_kwargs={'device': 'cpu'}
)
vector_store = FAISS.from_documents(documents, embeddings)
vector_store.save_local(vector_store_path)
return vector_store