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

à partir de phrase_transformateurs Importer SentenTentransformateur
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)} »)