Construire un chatbot PDF propulsé par l’OCR avec un chiffon et une conscience de l’histoire contextuelle | par Rudra Bhaskar | Mai 2025

importer fitz # pyMupdf
De Langchain.Text_Splitter Import RecursiVECHarAtteTextSplitter
Importer Numpy comme NP
importer google.generativaia en tant que Genai
Importer un système d’exploitation
de la liste des importations de dactylographie, dict
à partir de Dotenv Import Load_Dotenv
à partir de l’image d’importation PIL
Importer IO
load_dotenv ()
Google_api_key = os.getenv (‘google_api_key’)
Si ce n’est pas Google_API_KEY:
River ValueError (« Veuillez définir la variable d’environnement Google_API_KEY »)
Class ChunkwithSource:
Def __init __ (Self, Texte: Str, Source: Str):
self.text = texte
self.source = source
def process_documents (dossier_path: str) -> list (chunkwithsource):
chunks_with_sources = ()
files = get_files (dossier_path)
# Process PDFS
pour pdf_path dans les fichiers (‘pdfs’):
filename = os.path.basename (pdf_path)
text = extract_text_from_pdf (pdf_path)
chunks = split_text_into_chunks (texte)
pour un morceau en morceaux:
chunks_with_sources.append (chunkwithsource (chunk, nom de fichier)))
# Processus des images avec les Gémeaux
pour image_path dans les fichiers (‘images’):
filename = os.path.basename (image_path)
# Utilisez des Gémeaux pour OCR au lieu de Tesseract
text = process_image_with_gemini (image_path)
Si le texte:
chunks = split_text_into_chunks (texte)
pour un morceau en morceaux:
chunks_with_sources.append (chunkwithsource (chunk, nom de fichier)))
return chunks_with_sources
def get_files (dossier_path: str) -> dict (str, list (str)):
« » « Obtenez tous les fichiers PDF et image du dossier spécifié. » « »
Si pas os.path.exists (Folder_Path):
os.makedirs (dossier_path)
print (f « répertoire créé: {dossier_path} »)
return {‘pdfs’: (), ‘images’: ()}
files = {
‘pdfs’: (),
‘images’: ()
}
pour le fichier dans os.listdir (dossier_path):
Lower_file = file.lower ()
si bower_file.endswith (‘. pdf’):
fichiers (‘pdfs’). APPEND (os.path.join (dossier_path, fichier))
elif bower_file.endswith ((‘. png’, ‘.jpg’, ‘.jpeg’, ‘.tiff’, ‘.bmp’)):
fichiers (‘images’). Ajouter (os.path.join (dossier_path, fichier))
return les fichiers
def get_pdf_files (dossier_path: str) -> list (str):
« » « Obtenez tous les fichiers PDF à partir du dossier spécifié. » « »
Si pas os.path.exists (Folder_Path):
os.makedirs (dossier_path)
print (f « répertoire créé: {dossier_path} »)
retour ()
pdf_files = ()
pour le fichier dans os.listdir (dossier_path):
si file.endswith (‘. pdf’):
pdf_files.append (os.path.join (dossier_path, fichier))
retourner pdf_files
def process_image_with_gemini (image_path: str) -> str:
« » « Processus Image en utilisant API Vision Gemini » « »
essayer:
# Lire l’image comme des octets
avec ouvert (image_path, ‘rb’) comme f:
image_bytes = f.read ()
# Configurez l’API
genai.configure (api_key = google_api_key)
# Initialiser le modèle
modèle = Genai.GenerativeModel (‘Gemini-2.0-Flash’)
# Créez l’invite pour OCR
prompt = « Extraire et renvoyer tout le texte visible dans cette image. Format clairement. »
# Générer du contenu avec l’image
réponse = modélisation.generate_content (((
{
« mime_type »: f’image / {image_path.split (« . ») (- 1) .lower ()} ‘,
« Données »: image_bytes
},
rapide
))
Retour Response.Text
sauf exception comme e:
Imprimer (F « Image de traitement d’erreur avec Gemini: {str (e)} »)
retour « »
Class ConversationMemory:
def __init __ (self, max_history: int = 5):
self.history: list (dict) = ()
self.max_history = max_history
Def add_interaction (self, requête: str, réponse: str, contexte: str):
self.history.append ({
« Query »: requête,
« Réponse »: réponse,
« contexte »: contexte
})
Si Len (self.history)> self.max_history:
self.history.pop (0)
def get_formatted_history (self) -> str:
formated = « »
pour l’interaction dans l’auto-hith:
formaté + = f « Question: {interaction (‘query’)} n »
formaté + = f « Réponse: {interaction (‘réponse’)} n »
Retour format
def extract_text_from_pdf (pdf_path):
text = « »
avec fitz.open (pdf_path) comme doc:
Pour page_num, page en énumération (doc, start = 1):
text + = page.get_text ()
RETOUR
def Split_text_into_chunks (texte, chunk_size = 1000, chunk_overlap = 200):
Splitter = récursiveCharAtteTtexTsPlitter (
chunk_size = chunk_size,
chunk_overlap = chunk_overlap,
séparateurs = ( » n n », » n », « . », « ! », « ? », « », « »)
)
return Splitter.Split_Text (texte)
def phrase_encode (phrases):
Model = SentenTetransFormer (‘Sheason Transformateurs / All-Minilm-L6-V2’)
Embeddings = Model.Encode (Sentins)
Retour intégrer
def cosine_similarity (a, b):
a = np.array (a)
b = np.array (b)
retour np.dot (a, b) / (np.linalg.norm (a) * np.linalg.norm (b))
Si __name__ == « __main__ »:
# Spécifiez votre chemin de dossier PDFS
Folder_path = os.path.join (os.path.dirname (__ file__), « Folder_With_pdfs »)
# Processus tous les PDF avec suivi de la source
chunks_with_sources = process_documents (dossier_path)
Si ce n’est pas Chunks_With_Sources:
print (« Pas de fichiers PDF trouvés dans le dossier spécifié! »)
print (f « Veuillez ajouter des fichiers PDF à: {Folder_Path} »)
sortie()
# Extraire juste le texte pour les intégres
all_chunks = (chunk.text pour morceau dans chunks_with_sources)
print (f « Total Chunks créé: {len (all_chunks)} »)
# Créer des intégres pour tous les morceaux
chunk_vectors = phrase_encode (all_chunks)
# Initialiser la mémoire de la conversation
mémoire = ConversationMemory ()
Bien que vrai:
# Obtenez la saisie des utilisateurs
Query = Input ( » nenter votre question (ou ‘quit’ pour quitter): »)
Si query.lower () == ‘quit’:
casser
query_vector = phrase_encode ((requête))
top_k = 3
similitudes = ()
pour idx, Chunk_Vec en énumération (Chunk_Vecteurs):
sim = cosine_similarity (chunk_vec, query_vector (0))
similitudes.APPEND ((SIM, IDX))
imprimer (« similitudes: », similitudes)
imprimer (« == » * 20)
# Triez par similitude descendant et obtenez les indices TOP_K
top_chunks = tri (similitudes, inverse = true) (: top_k)
top_indices = (idx pour _, idx dans top_chunks)
Imprimer (« Indices de morceaux supérieurs: », top_indices)
new_context = « »
pour i dans top_indices:
new_context + = all_chunks (i) + » n »
# Créer une invite consciente de l’histoire
Conversation_History = Memory.get_formatted_history ()
prompt_template = f « » « Vous êtes un assistant utile avec accès au contexte de conversation précédent et à la question actuelle.
Conversation précédente:
{Conversation_history}
Contexte actuel (de {chunks_with_sources (top_indices (0)). Source}):
{new_context}
Question actuelle: {Query}
Veuillez fournir une réponse cohérente qui prend en compte à la fois l’historique de la conversation et le contexte actuel. Mentionnez également quel (s) fichier PDF contenait les informations pertinentes. « » «
essayer:
# Configurez l’API
genai.configure (api_key = google_api_key)
# Initialiser correctement le modèle
modèle = Genai.GenerativeModel (‘Gemini-2.0-Flash’)
# Générer une réponse avec l’invite réelle
Response = Model.generate_content (prompt_template)
print ( » nResponse: »)
Imprimer (Response.Text)
# Stocker l’interaction dans la mémoire
mémoire.add_interaction (query, réponse.Text, new_context)
sauf exception comme e:
Print (F « Réponse de génération d’erreur: {str (e)} »)