Si vous ne connaissez pas chatPDF, je vous recommande de le tester sur le site en uploadant un fichier pdf, et interrogez le document comme si vous interrogiez chatGPT. L’intérêt est que vous pouvez connaitre c’e qui vous intéresse du document sans le lire entièrement.
Par exemple vous souscrivez à une assurance, et il y a un long contrat en petits caractères à lire, plutôt que de perdre des heures à décortiquer le document, vous interrogez chatGPT sur le document.
Techniquement ce n’est pas comme ça. En fait vous injecter le texte du document dans la base vectorielle FAISS, vous interrogez FAISS, qui vous retournera un résultat (qui est un contexte) et vous le soumettez à chatGPT.
Vous contextualisez très précisément votre prompt.
Contenu
Installation des composants
La base de données vectorielle FAISS
FAISS est un logiciel fait par Facebook, acronyme de Facebook AI Similarity Search. Elle tokenise le pdf et le stocke sous forme vectorielle.
Le parseur de PDF
PyPDF2 est un très bon parseur de pdf.
Un tokeniser
tiktoken va transformer le texte en token. Un token est une unité qui a une signification mathématique pour le LLM. tiktoken travaille de concert avec le modèle de OpenAI.
Openai
Permet d’appeler le webservice d’OpenAI facilement
Une clé API par exemple chatGPT
Il vous faudra acheter des crédits pour le webservice chatGPT, 10 euros est pas mal et vous permettra de faire beaucoup d’appels.
Le code Python
#!/usr/bin/env python3
"""
ChatPDF Clone avec Base de Données Vectorielle
Utilise des embeddings et une recherche vectorielle pour une meilleure précision
"""
import PyPDF2
from openai import OpenAI
import os
import argparse
import sys
import numpy as np
from typing import List, Optional, Dict, Tuple
import tiktoken
import re
import faiss
import pickle
from sentence_transformers import SentenceTransformer
from dataclasses import dataclass
import json
@dataclass
class TextChunk:
text: str
page_num: int
chunk_id: int
embedding: Optional[np.ndarray] = None
class VectorChatPDF:
def __init__(self, api_key: Optional[str] = None, embedding_model: str = "all-MiniLM-L6-v2"):
"""
Initialise ChatPDF avec base vectorielle
"""
self.api_key = "sk-proj-BBFnLZzRMt8EZp4LStJXyfzc_6knvyKX2jOhYXIRh27MmGHIkxzcGFfcisYMS3uiRYBP0r2qT3BlbkFJObsGY8qW-VjRQNLsB9Fr-iPhSmCs0Aaj2Y8hSF34dBQQ4dEgiq7mQO30BCWDItJC775SUA"
if not self.api_key:
raise ValueError("Clé API OpenAI requise. Définissez OPENAI_API_KEY ou passez api_key")
self.client = OpenAI(api_key=self.api_key)
# Modèle d'embeddings local (gratuit)
print("Chargement du modèle d'embeddings...")
self.embedding_model = SentenceTransformer(embedding_model)
# Base de données vectorielle FAISS
self.vector_db = None
self.chunks: List[TextChunk] = []
self.pdf_path = ""
self.encoding = tiktoken.get_encoding("cl100k_base")
def extract_text_from_pdf(self, pdf_path: str) -> List[TextChunk]:
"""
Extrait le texte du PDF et le divise en chunks
"""
try:
with open(pdf_path, 'rb') as file:
pdf_reader = PyPDF2.PdfReader(file)
chunks = []
chunk_id = 0
print(f"Extraction du texte de {len(pdf_reader.pages)} pages...")
for page_num, page in enumerate(pdf_reader.pages):
try:
page_text = page.extract_text()
if page_text.strip():
# Diviser la page en chunks
page_chunks = self._split_text_into_chunks(page_text, max_tokens=500)
for chunk_text in page_chunks:
chunk = TextChunk(
text=chunk_text,
page_num=page_num + 1,
chunk_id=chunk_id
)
chunks.append(chunk)
chunk_id += 1
except Exception as e:
print(f"Erreur lors de l'extraction de la page {page_num + 1}: {e}")
continue
self.chunks = chunks
self.pdf_path = pdf_path
print(f"Extraction terminée. {len(chunks)} chunks créés.")
return chunks
except FileNotFoundError:
raise FileNotFoundError(f"Fichier PDF non trouvé: {pdf_path}")
except Exception as e:
raise Exception(f"Erreur lors de la lecture du PDF: {e}")
def _split_text_into_chunks(self, text: str, max_tokens: int = 500) -> List[str]:
"""
Divise le texte en chunks optimaux pour les embeddings
"""
# Nettoyer le texte
text = re.sub(r'\s+', ' ', text).strip()
# Diviser par phrases
sentences = re.split(r'[.!?]+', text)
chunks = []
current_chunk = ""
for sentence in sentences:
sentence = sentence.strip()
if not sentence:
continue
test_chunk = current_chunk + " " + sentence if current_chunk else sentence
if len(self.encoding.encode(test_chunk)) <= max_tokens:
current_chunk = test_chunk
else:
if current_chunk:
chunks.append(current_chunk)
current_chunk = sentence
if current_chunk:
chunks.append(current_chunk)
return [chunk for chunk in chunks if len(chunk.strip()) > 20] # Filtrer les chunks trop courts
def create_embeddings(self):
"""
Crée les embeddings pour tous les chunks et construit l'index FAISS
"""
if not self.chunks:
raise ValueError("Aucun chunk disponible. Extrayez d'abord le texte du PDF.")
print("Création des embeddings...")
texts = [chunk.text for chunk in self.chunks]
# Créer les embeddings par batch pour l'efficacité
embeddings = self.embedding_model.encode(texts, show_progress_bar=True)
# Associer les embeddings aux chunks
for i, chunk in enumerate(self.chunks):
chunk.embedding = embeddings[i]
# Créer l'index FAISS
dimension = embeddings.shape[1]
self.vector_db = faiss.IndexFlatIP(dimension) # Index par produit scalaire (cosine similarity)
# Normaliser les vecteurs pour la similarité cosine
normalized_embeddings = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
self.vector_db.add(normalized_embeddings.astype('float32'))
print(f"Index vectoriel créé avec {len(self.chunks)} embeddings.")
def search_similar_chunks(self, query: str, top_k: int = 5) -> List[Tuple[TextChunk, float]]:
"""
Recherche les chunks les plus similaires à la requête
"""
if not self.vector_db:
raise ValueError("Index vectoriel non créé. Appelez create_embeddings() d'abord.")
# Créer l'embedding de la requête
query_embedding = self.embedding_model.encode([query])
query_embedding = query_embedding / np.linalg.norm(query_embedding)
# Rechercher les chunks similaires
scores, indices = self.vector_db.search(query_embedding.astype('float32'), top_k)
results = []
for score, idx in zip(scores[0], indices[0]):
if idx < len(self.chunks): # Vérification de sécurité
results.append((self.chunks[idx], float(score)))
return results
def ask_question(self, question: str, model: str = "gpt-3.5-turbo", top_k: int = 5) -> Dict:
"""
Pose une question en utilisant la recherche vectorielle
"""
if not self.vector_db:
return {"error": "Index vectoriel non créé. Appelez create_embeddings() d'abord."}
# Rechercher les chunks pertinents
similar_chunks = self.search_similar_chunks(question, top_k=top_k)
if not similar_chunks:
return {"error": "Aucun contenu pertinent trouvé."}
# Construire le contexte à partir des chunks les plus pertinents
context_parts = []
source_info = []
for chunk, score in similar_chunks:
context_parts.append(f"[Page {chunk.page_num}] {chunk.text}")
source_info.append({
"page": chunk.page_num,
"chunk_id": chunk.chunk_id,
"similarity_score": score,
"text_preview": chunk.text[:100] + "..." if len(chunk.text) > 100 else chunk.text
})
context = "\n\n".join(context_parts)
prompt = f"""
Vous êtes un assistant expert qui répond aux questions basées sur le contenu d'un document PDF.
Contexte pertinent extrait du document:
{context}
Question: {question}
Instructions:
- Répondez uniquement basé sur le contexte fourni
- Si l'information n'est pas suffisante, dites-le clairement
- Citez les numéros de page quand c'est pertinent
- Soyez précis et structuré dans votre réponse
Réponse:"""
try:
response = self.client.chat.completions.create(
model=model,
messages=[
{"role": "system",
"content": "Vous êtes un assistant spécialisé dans l'analyse de documents PDF avec une expertise en recherche d'information."},
{"role": "user", "content": prompt}
],
max_tokens=800,
temperature=0.1
)
answer = response.choices[0].message.content.strip()
return {
"answer": answer,
"sources": source_info,
"context_used": len(similar_chunks)
}
except Exception as e:
return {"error": f"Erreur lors de la génération de la réponse: {e}"}
def save_index(self, filepath: str):
"""
Sauvegarde l'index vectoriel et les métadonnées
"""
if not self.vector_db:
raise ValueError("Aucun index à sauvegarder.")
# Sauvegarder l'index FAISS
faiss.write_index(self.vector_db, f"{filepath}.faiss")
# Sauvegarder les métadonnées
metadata = {
"chunks": [
{
"text": chunk.text,
"page_num": chunk.page_num,
"chunk_id": chunk.chunk_id
} for chunk in self.chunks
],
"pdf_path": self.pdf_path
}
with open(f"{filepath}.json", 'w', encoding='utf-8') as f:
json.dump(metadata, f, ensure_ascii=False, indent=2)
print(f"Index sauvegardé: {filepath}.faiss et {filepath}.json")
def load_index(self, filepath: str):
"""
Charge un index vectoriel sauvegardé
"""
try:
# Charger l'index FAISS
self.vector_db = faiss.read_index(f"{filepath}.faiss")
# Charger les métadonnées
with open(f"{filepath}.json", 'r', encoding='utf-8') as f:
metadata = json.load(f)
# Reconstruire les chunks
self.chunks = []
for chunk_data in metadata["chunks"]:
chunk = TextChunk(
text=chunk_data["text"],
page_num=chunk_data["page_num"],
chunk_id=chunk_data["chunk_id"]
)
self.chunks.append(chunk)
self.pdf_path = metadata["pdf_path"]
print(f"Index chargé: {len(self.chunks)} chunks disponibles.")
except FileNotFoundError:
raise FileNotFoundError(f"Fichiers d'index non trouvés: {filepath}.faiss ou {filepath}.json")
def main():
parser = argparse.ArgumentParser(description="ChatPDF avec Base Vectorielle")
parser.add_argument("pdf_path", help="Chemin vers le fichier PDF")
parser.add_argument("--api-key", help="Clé API OpenAI")
parser.add_argument("--model", default="gpt-3.5-turbo", help="Modèle OpenAI à utiliser")
parser.add_argument("--save-index", help="Sauvegarder l'index vectoriel")
parser.add_argument("--load-index", help="Charger un index vectoriel existant")
parser.add_argument("--top-k", type=int, default=5, help="Nombre de chunks à récupérer")
args = parser.parse_args()
try:
# Initialiser ChatPDF vectoriel
chat_pdf = VectorChatPDF(api_key=args.api_key)
if args.load_index:
# Charger un index existant
print(f"Chargement de l'index: {args.load_index}")
chat_pdf.load_index(args.load_index)
else:
# Traiter un nouveau PDF
print(f"Chargement du PDF: {args.pdf_path}")
chat_pdf.extract_text_from_pdf(args.pdf_path)
chat_pdf.create_embeddings()
if args.save_index:
chat_pdf.save_index(args.save_index)
# Mode interactif
print("\n=== MODE QUESTIONS-RÉPONSES VECTORIEL ===")
print("Posez vos questions sur le document (tapez 'quit' pour quitter):")
print("Tapez 'stats' pour voir les statistiques de l'index")
while True:
try:
question = input("\n❓ Votre question: ").strip()
if question.lower() in ['quit', 'exit', 'q']:
break
if question.lower() == 'stats':
print(f"📊 Statistiques:")
print(f" - Nombre de chunks: {len(chat_pdf.chunks)}")
print(f" - Pages: {max(c.page_num for c in chat_pdf.chunks) if chat_pdf.chunks else 0}")
print(f" - Document: {os.path.basename(chat_pdf.pdf_path)}")
continue
if not question:
continue
print("🔍 Recherche vectorielle en cours...")
result = chat_pdf.ask_question(question, model=args.model, top_k=args.top_k)
if "error" in result:
print(f"❌ {result['error']}")
else:
print(f"\n💡 Réponse:\n{result['answer']}")
print(f"\n📚 Sources utilisées ({result['context_used']} chunks):")
for i, source in enumerate(result['sources'][:3], 1):
print(f" {i}. Page {source['page']} (score: {source['similarity_score']:.3f})")
print(f" {source['text_preview']}")
except KeyboardInterrupt:
break
except Exception as e:
print(f"Erreur: {e}")
print("\nAu revoir!")
except Exception as e:
print(f"Erreur: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
Lancer le programme en ligne de commande
lancement du script en mode normal, cela va créer un fichier binaire mon_index.faiss qui est la base vectorielle
python3 avecBaseVectorielle.py document.pdf --save-index mon_index
La même chose en plus léger
python3 avecBaseVectorielle.py document.pdf --fast --save-index mon_index
La même chose avec un LLM personnalisé
python3 avecBaseVectorielle.py document.pdf --embedding-model "paraphrase-MiniLM-L3-v2"
La même chose en chargeant une base vectorielle déjà calculée
python3 avecBaseVectorielle.py document.pdf --load-index mon_index
Lors du lancement du script en mode normal, une base vectorielle va être créée. Vous allez poser une question qui va requêter sur cette base de données. Par exemple, « Combien de jours de formation y a t il dans cette formation? », la requête ne va pas partir vers chatGPT directement mais va extraire un contexte de la base vectorielle, qui va être mis dans le prompt vers chatGPT.
Sans le vouloir vraiment, vous venez de faire du RAG, Retrieval Augmented Generation, autrement dit intelligence générative augmentée par une extraction de données.
Modèle d’embedding c’est quoi?
Un modèle d’embedding est un système qui transforme du texte (mots, phrases, paragraphes) en vecteurs numériques de dimension fixe. Ces vecteurs capturent le sens sémantique du texte de manière à ce que des textes similaires aient des représentations vectorielles proches dans l’espace mathématique.
Comment ça fonctionne
Les modèles d’embedding utilisent généralement des réseaux de neurones entraînés sur de grandes quantités de texte pour apprendre les relations sémantiques. Par exemple, les mots « chien » et « animal » auront des vecteurs plus proches que « chien » et « voiture ».
Utilité pratique
Ces représentations vectorielles permettent de :
- Mesurer la similarité sémantique entre textes
- Effectuer des recherches par similarité
- Regrouper des documents par thème
- Alimenter des systèmes de question-réponse
Le modèle « paraphrase-MiniLM-L3-v2 » est un modèle d’embedding compact et efficace développé par Sentence Transformers. Il est particulièrement adapté pour :
- Identifier des paraphrases (textes exprimant la même idée différemment)
- Créer des bases de données vectorielles pour la recherche sémantique
Le mode –fast
C’est u mode d’optimisation qui accélère le traitement en faisant certains compromis.
Le mode fast améliore généralement la vitesse au détriment de :
- La précision de la recherche sémantique
- La qualité des réponses
- La finesse de l’analyse du document