Introduction : Pourquoi la Recherche Vectorielle Change Tout pour la Médecine
En tant qu'ingénieur qui a travaillé pendant trois ans sur des systèmes RAG (Retrieval-Augmented Generation) pour des bases de données médicales, je peux vous dire que le plus grand défi n'est pas de comprendre les modèles de langage — c'est de faire en sorte que votre système comprenne réellement les termes médicaux spécialisés. Quand j'ai commencé, je pensais qu'un simple embedding suffirait. J'avais tort. Un patient dont le dossier mentionne une « cardiomyopathie hypertrophique » ne doit pas être confondu avec une simple « insuffisance cardiaque ». Cette nuance peut sauver des vies.
Dans ce tutoriel, je vais vous guider pas à pas depuis zéro. Aucun prérequis en API ou en machine learning n'est nécessaire. Nous construirons ensemble un système de recherche vectorielle optimisé pour la littérature médicale francophone et anglophone, capable de comprendre les synonymes médicaux, les abréviations cliniques et le contexte sémantique propre à la médecine.
Avertissement important : Les exemples médicaux présentés sont uniquement à des fins pédagogiques. Ne les utilisez jamais pour des décisions cliniques réelles sans supervision médicale qualifiée.
Comprendre les Bases : Qu'est-ce que le RAG Médical ?
Le Principe Fondamental Expliqué Simplement
Imaginez que vous avez une bibliothèque immense contenant des milliers d'articles médicaux en français et en anglais. Un patient vous demande : « J'ai des douleurs thoraciques, est-ce que c'est grave ? » Un système RAG médical fonctionne comme un assistant médical virtuelle qui :
- Comprend votre question en la convertissant en « coordonnées numériques » (vecteur)
- Recherche dans la bibliothèque les passages les plus pertinents
- Synthétise une réponse en s'appuyant sur les documents trouvés
Le problème ? Les termes médicaux sont truffés de synonymes, d'abréviations et de variations linguistiques. « AVC » peut signifier « accident vasculaire cérébral » mais aussi « attack vasculaire cérébrale » selon les régions. « HTA » désigne l'hypertension artérielle. Notre système doit comprendre que « tension artérielle élevée », « HTA » et « hypertension » désignent la même pathologie.
Pourquoi les Modèles Standards Échouent
Les modèles d'embedding génériques comme text-embedding-ada-002 traitent les termes médicaux comme des mots ordinaires. Ils ne savent pas que « infarctus du myocarde », « crise cardiaque » et « IDM » sont synonymes. C'est pourquoi nous avons besoin d'une optimisation spécifique au domaine médical.
Architecture de Notre Système RAG Médical
Vue d'Ensemble en 5 Étapes
┌─────────────────────────────────────────────────────────────────┐
│ ARCHITECTURE RAG MÉDICAL │
├─────────────────────────────────────────────────────────────────┤
│ │
│ Étape 1 : INGESTION DES DOCUMENTS │
│ ├── Extraction du texte (PDF, XML, JSON médical) │
│ ├── Découpage intelligent (pas par caractères !) │
│ └── Normalisation terminologique │
│ │
│ Étape 2 : AUGMENTATION TERMINOLOGIQUE │
│ ├── Expansion avec synonymes médicaux (UMLS, SNOMED-CT) │
│ ├── Gestion des abréviations cliniques │
│ └── Traduction croisée FR ↔ EN │
│ │
│ Étape 3 : GÉNÉRATION DES VECTEURS │
│ ├── Embedding contextuel médical │
│ ├── Métadonnées structurées │
│ └── Indexation dans la base vectorielle │
│ │
│ Étape 4 : RECHERCHE HYBRIDE │
│ ├── Recherche vectorielle (similarité sémantique) │
│ ├── Recherche par mots-clés (BM25) │
│ └── Fusion des résultats │
│ │
│ Étape 5 : GÉNÉRATION DE LA RÉPONSE │
│ ├── Contexte pertinents identifiés │
│ ├── Prompt médical structuré │
│ └── Réponse citant les sources │
│ │
└─────────────────────────────────────────────────────────────────┘
Les Technologies que Nous Utiliserons
- Base URL API : https://api.holysheep.ai/v1 (infrastructure haute performance, latence moyenne 47ms)
- Modèle d'embedding : text-embedding-3-large avec optimisation médicale
- Base vectorielle : ChromaDB (open source, simple à déployer)
- Client HTTP : requests en Python (accessible aux débutants)
Installation de l'Environnement de Développement
Prérequis et Configuration Initiale
Ouvrez votre terminal et tapez les commandes suivantes. Ne vous inquiétez pas si vous ne comprenez pas tout immédiatement — je vais expliquer chaque ligne.
# Installation des dépendances Python
pip install requests chromadb numpy python-dotenv pdfplumber
Vérification de l'installation
python -c "import requests, chromadb; print('✓ Modules installés avec succès')"
Création du fichier de configuration
mkdir -p medical_rag_project
cd medical_rag_project
touch .env
Indications capture d'écran : Vous devriez voir « ✓ Modules installés avec succès » s'afficher en vert dans votre terminal. Si vous obtenez un message d'erreur rouge, vérifiez que Python 3.8+ est installé avec python --version.
Configuration de la Clé API HolySheep
Modifiez le fichier .env créé précédemment :
# Contenu du fichier .env
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
BASE_URL=https://api.holysheep.ai/v1
EMBEDDING_MODEL=text-embedding-3-large
GENERATION_MODEL=gpt-4.1
Important : Remplacez YOUR_HOLYSHEEP_API_KEY par votre clé réelle obtenue lors de l'inscription sur la plateforme HolySheheep AI. La plateforme propose des crédits gratuits pour les nouveaux utilisateurs — idéal pour expérimenter sans engagement financier initial.
Étape 1 : Création du Module d'Embedding Médical
La Classe de Base pour les Embeddings
Créez un fichier nommé medical_embedder.py et collez le code suivant. Cette classe encapsule toute la logique de communication avec l'API HolySheep pour générer des vecteurs sémantiques.
import os
import requests
import numpy as np
from typing import List, Dict, Optional
from dotenv import load_dotenv
load_dotenv()
class MedicalEmbedder:
"""
Classe pour générer des embeddings médicaux optimisés.
Gère la communication avec l'API HolySheep AI.
"""
def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None):
"""
Initialisation du client d'embedding.
Args:
api_key: Clé API HolySheep (par défaut: variable d'environnement)
base_url: URL de base de l'API (par défaut: variable d'environnement)
"""
self.api_key = api_key or os.getenv("HOLYSHEEP_API_KEY")
self.base_url = base_url or os.getenv("BASE_URL", "https://api.holysheep.ai/v1")
if not self.api_key:
raise ValueError(
"Clé API manquante. "
"Inscrivez-vous sur https://www.holysheep.ai/register pour obtenir votre clé."
)
def _call_api(self, texts: List[str]) -> List[List[float]]:
"""
Appel interne à l'API HolySheep pour obtenir les embeddings.
"""
url = f"{self.base_url}/embeddings"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"input": texts,
"model": "text-embedding-3-large"
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
# Gestion des erreurs HTTP
if response.status_code != 200:
error_detail = response.json().get("error", {}).get("message", response.text)
raise Exception(f"Erreur API HolySheep ({response.status_code}): {error_detail}")
data = response.json()
return [item["embedding"] for item in data["data"]]
def embed_single(self, text: str) -> np.ndarray:
"""
Génère un embedding pour un texte unique.
Args:
text: Texte médical à convertir en vecteur
Returns:
Vecteur numpy de dimension 3072
"""
vectors = self._call_api([text])
return np.array(vectors[0])
def embed_batch(self, texts: List[str], batch_size: int = 100) -> List[np.ndarray]:
"""
Génère des embeddings pour plusieurs textes par lots.
Plus économique pour le traitement de documents volumineux.
Args:
texts: Liste de textes médicaux
batch_size: Nombre de textes par appel API (défaut: 100)
Returns:
Liste de vecteurs numpy
"""
all_vectors = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
vectors = self._call_api(batch)
all_vectors.extend([np.array(v) for v in vectors])
# Indicateur de progression pour les longues opérations
if (i + batch_size) % 500 == 0:
print(f" Progression: {min(i + batch_size, len(texts))}/{len(texts)} textes traités")
return all_vectors
def compute_similarity(self, vec1: np.ndarray, vec2: np.ndarray) -> float:
"""
Calcule la similarité cosinus entre deux vecteurs.
Args:
vec1: Premier vecteur
vec2: Deuxième vecteur
Returns:
Score de similarité entre -1 et 1
"""
dot_product = np.dot(vec1, vec2)
norm_product = np.linalg.norm(vec1) * np.linalg.norm(vec2)
return dot_product / norm_product if norm_product > 0 else 0.0
Test unitaire rapide
if __name__ == "__main__":
print("Test du module d'embedding médical...")
embedder = MedicalEmbedder()
# Test avec termes médicaux synonymes
term1 = "infarctus du myocarde"
term2 = "crise cardiaque"
term3 = "diabète de type 2"
vec1 = embedder.embed_single(term1)
vec2 = embedder.embed_single(term2)
vec3 = embedder.embed_single(term3)
sim_synonyme = embedder.compute_similarity(vec1, vec2)
sim_different = embedder.compute_similarity(vec1, vec3)
print(f" Similarité '{term1}' vs '{term2}' (synonymes): {sim_synonyme:.4f}")
print(f" Similarité '{term1}' vs '{term3}' (différent): {sim_different:.4f}")
assert sim_synonyme > sim_different, "Les synonymes devraient être plus similaires !"
print("✓ Test réussi : le modèle distingue correctement les termes médicaux.")
Exécutez ce script avec python medical_embedder.py. Vous devriez voir un score de similarité élevé (supérieur à 0.7) entre « infarctus du myocarde » et « crise cardiaque », et un score plus faible entre « infarctus » et « diabète ».
Comprendre le Code Pas à Pas
Si vous êtes débutant, voici ce que chaque partie fait :
import requests: Cette bibliothèque permet d'envoyer des requêtes HTTP, comme quand votre navigateur demande une page web — sauf qu'on demande des embeddings à l'APInp.ndarray: Un tableau de nombres très efficace pour les calculs mathématiques. Notre texte devient une longue liste de 3072 nombresbatch_size=100: Au lieu d'envoyer un texte à la fois (lent), on en envoie 100 d'un coup (économique). L'API HolySheep facture par token, donc moins d'appels = moins de coûts
Étape 2 : Expansion Terminologique Médicale
Le Problème des Synonymes en Médecine
Un même concept médical peut être exprimé de dizaines de façons différentes. Voici quelques exemples concrets que j'ai rencontrés dans mes projets :
┌──────────────────────────────────────────────────────────────────────┐
│ TABLE DE SYNONYMES MÉDICAUX │
├──────────────────────────────────────────────────────────────────────┤
│ Terme principal │ Synonymes │
├──────────────────────────────────────────────────────────────────────┤
│ Hypertension artérielle │ HTA, tension élevée, haute pression │
│ Diabète │ DT2, diabète type II, sugar diabetes │
│ Accident vasculaire │ AVC, attack, brain attack, ictus │
│ Infarctus │ IDM, crise cardiaque, cardiac event │
│ Bronchopneumopathie │ BPCO, maladie pulmonaire obstructive │
│ Insuffisance rénale │ IRC, rein faible, renal failure │
└──────────────────────────────────────────────────────────────────────┘
Quand un médecin tape « HTA » dans votre système RAG, vous devez retrouver tous les documents parlant d’hypertension artérielle, que ce soit en français ou en anglais.
Le Module d'Expansion Terminologique
import re
from typing import List, Dict, Set
class MedicalTermExpander:
"""
Expanseur de terminologie médicale pour améliorer la récupération.
Utilise un dictionnaire de synonymes médicaux pour enrichir les requêtes.
"""
# Dictionary expanded with French-English medical terminology
SYNONYM_DICTIONARY = {
# Cardiovascular
"hypertension artérielle": ["HTA", "tension artérielle élevée", "haute pression",
"hypertension", "high blood pressure", "elevated BP"],
"hypotension": ["basse pression", "pression basse", "low blood pressure", "hypotension orthostatique"],
"infarctus du myocarde": ["IDM", "crise cardiaque", "infarctus", "cardiac infarction",
"myocardial infarction", "heart attack"],
"insuffisance cardiaque": ["ICC", "cardiaque", "cardiac failure", "heart failure", "défaillance cardiaque"],
"accident vasculaire cérébral": ["AVC", "ictus", "brain attack", "stroke", "attaque cérébrale"],
"arythmie": ["trouble du rythme", "fibrillation", "arrhythmia", "atrial fibrillation", "FA"],
# Metabolic / Endocrine
"diabète de type 2": ["DT2", "diabète type II", "diabète adulte", "NIDDM",
"type 2 diabetes", "adult-onset diabetes"],
"diabète de type 1": ["DT1", "diabète juvénile", "insulinodépendant",
"type 1 diabetes", "juvenile diabetes"],
"obésité": ["surpoids sévère", "IMC élevé", "obèse", "adiposité",
"obesity", "overweight", "bariatric"],
"dyslipidémie": ["cholestérol élevé", "triglycérides hauts", "hyperlipidémie",
"hypercholesterolemia", "high cholesterol"],
# Respiratory
"bronchopneumopathie chronique obstructive": ["BPCO", "bronchite chronique",
"emphysème", "COPD", "chronic obstructive pulmonary disease"],
"asthme": ["bronchospasme", "asthme allergique", "asthme bronchique", "asthma"],
"pneumonie": ["infection pulmonaire", "bronchopneumonie", "pneumonie bactérienne",
"pneumonia", "lung infection"],
# Renal
"insuffisance rénale chronique": ["IRC", "maladie rénale chronique", "MRC",
"chronic kidney disease", "CKD", "renal failure"],
"néphropathie": ["maladie du rein", "rein", "nephropathy", "kidney disease"],
# Neurological
"maladie d'Alzheimer": ["Alzheimer", "démence sénile", "troubles cognitifs",
"Alzheimer's disease", "dementia", "cognitive decline"],
"épilepsie": ["convulsions", "crises épileptiques", "seizures", "epilepsy"],
# General medical terms
"douleur thoracique": ["thoracic pain", "chest pain", "angine", "angor", "precordialgia"],
"dyspnée": ["essoufflement", "difficulté à respirer", "shortness of breath", "SOB", "dyspnea"],
"oedème": ["gonflement", "rétention d'eau", "swelling", "edema"],
"nausées": ["envie de vomir", "vomissements", "nausea", "vomiting", "emesis"],
}
# Medical abbreviations (bidirectional mapping)
ABBREVIATIONS = {
"HTA": ["hypertension artérielle", "high blood pressure"],
"IDM": ["infarctus du myocarde", "myocardial infarction"],
"AVC": ["accident vasculaire cérébral", "stroke"],
"BPCO": ["bronchopneumopathie chronique obstructive", "COPD"],
"DT1": ["diabète de type 1", "type 1 diabetes"],
"DT2": ["diabète de type 2", "type 2 diabetes"],
"IRC": ["insuffisance rénale chronique", "chronic kidney disease"],
"ICC": ["insuffisance cardiaque congestive", "congestive heart failure"],
"VHC": ["virus de l'hépatite C", "Hepatitis C virus"],
"VIH": ["virus de l'immunodéficience humaine", "HIV", "AIDS"],
"PCR": ["protéine C réactive", "C-reactive protein"],
"HbA1c": ["hémoglobine glyquée", "glycated hemoglobin", "glycosylated hemoglobin"],
"TSH": ["thyréostimuline", "thyroid stimulating hormone"],
"ECG": ["électrocardiogramme", "electrocardiogram"],
"IRM": ["imagerie par résonance magnétique", "MRI", "magnetic resonance imaging"],
"TDM": ["tomodensitométrie", "scanner", "CT scan"],
}
def __init__(self):
"""Initialisation de l'expanseur terminologique."""
# Build reverse mapping (expansion → canonical)
self.expansion_to_canonical = {}
for canonical, expansions in self.SYNONYM_DICTIONARY.items():
for exp in expansions:
self.expansion_to_canonical[exp.lower()] = canonical
for abbrev, expansions in self.ABBREVIATIONS.items():
for exp in expansions:
self.expansion_to_canonical[exp.lower()] = abbrev
def expand_query(self, query: str) -> List[str]:
"""
Enrichit une requête avec tous les termes synonymes.
Args:
query: Requête utilisateur originale
Returns:
Liste de toutes les variations synonymiques
"""
query_lower = query.lower()
terms = [query] # Start with original query
# Find direct matches
for canonical, synonyms in self.SYNONYM_DICTIONARY.items():
if canonical.lower() in query_lower or query_lower in [s.lower() for s in synonyms]:
terms.append(canonical)
terms.extend(synonyms)
# Expand abbreviations
words = re.findall(r'\b\w+\b', query_lower)
for word in words:
if word in self.ABBREVIATIONS:
terms.extend(self.ABBREVIATIONS[word])
if word in self.expansion_to_canonical:
canonical = self.expansion_to_canonical[word]
if canonical in self.SYNONYM_DICTIONARY:
terms.extend(self.SYNONYM_DICTIONARY[canonical])
# Remove duplicates while preserving order
seen = set()
unique_terms = []
for term in terms:
if term.lower() not in seen:
seen.add(term.lower())
unique_terms.append(term)
return unique_terms
def get_canonical_term(self, term: str) -> str:
"""
Retourne le terme médical standard pour un terme donné.
Args:
term: Terme à normaliser
Returns:
Terme médical canonique
"""
term_lower = term.lower()
if term_lower in self.expansion_to_canonical:
return self.expansion_to_canonical[term_lower]
return term
def generate_medical_context(self, terms: List[str]) -> str:
"""
Génère un contexte médical enrichi pour améliorer les embeddings.
Args:
terms: Liste de termes à contextualiser
Returns:
Texte enrichi pour l'embedding
"""
context_parts = []
for term in terms:
canonical = self.get_canonical_term(term)
context_parts.append(term)
# Add canonical + main synonyms for semantic enrichment
if canonical in self.SYNONYM_DICTIONARY:
synonyms = self.SYNONYM_DICTIONARY[canonical][:3] # Top 3
context_parts.extend(synonyms)
return " | ".join(context_parts)
Demonstration of the expander
if __name__ == "__main__":
expander = MedicalTermExpander()
# Test queries
test_queries = [
"Patient avec HTA et DT2",
"Douleur thoracique + dyspnée",
"Suivi infarctus du myocarde"
]
print("Démonstration de l'expansion terminologique:\n")
for query in test_queries:
expanded = expander.expand_query(query)
print(f"Requête: '{query}'")
print(f" Étendue à {len(expanded)} termes: {expanded[:6]}...")
print()
Ce module est crucial pour la performance de votre système. Quand un utilisateur tape « HTA », le système génère des embeddings pour « HTA », « hypertension artérielle », « haute pression », « high blood pressure » et tous les synonymes apparentés. La similarité avec les documents pertinents augmente significativement.
Étape 3 : Indexation de Documents Médicaux
Création de la Base Vectorielle avec ChromaDB
ChromaDB est une base de données vectorielle open source, simple et performante. Elle stocke vos embeddings et permet des recherches de similarité ultra-rapides.
import chromadb
from chromadb.config import Settings
import json
from datetime import datetime
from typing import List, Dict, Optional
class MedicalVectorStore:
"""
Gestionnaire de base de données vectorielle pour documents médicaux.
Utilise ChromaDB pour le stockage et la recherche de similarité.
"""
def __init__(self, persist_directory: str = "./medical_db"):
"""
Initialisation de la base vectorielle.
Args:
persist_directory: Chemin pour sauvegarder les données
"""
self.client = chromadb.PersistentClient(path=persist_directory)
self.collection = self.client.get_or_create_collection(
name="medical_documents",
metadata={"description": "Documents médicaux indexés pour RAG"}
)
# Embedder pour la génération de vecteurs
self.embedder = None
def set_embedder(self, embedder):
"""Associe un embedder médical au store."""
self.embedder = embedder
def add_document(
self,
doc_id: str,
content: str,
metadata: Optional[Dict] = None,
expand_terms: bool = True
) -> bool:
"""
Ajoute un document médical à l'index.
Args:
doc_id: Identifiant unique du document
content: Contenu textuel du document
metadata: Métadonnées (auteur, date, type, etc.)
expand_terms: Si True, enrichit avec synonymes médicaux
Returns:
True si succès, False sinon
"""
if not self.embedder:
raise RuntimeError("Aucun embedder configuré. Appelez set_embedder() d'abord.")
try:
# Enrichissement terminologique si demandé
if expand_terms and hasattr(self.embedder, 'expander'):
enriched_content = self.embedder.expander.generate_medical_context(
content.split()[:50] # Top 50 terms
)
# Combine original + enriched for better semantic capture
final_content = f"{content}\n\nTerminologie associée: {enriched_content}"
else:
final_content = content
# Génération de l'embedding
vector = self.embedder.embed_single(final_content)
# Préparation des métadonnées
doc_metadata = metadata or {}
doc_metadata.update({
"indexed_at": datetime.now().isoformat(),
"content_length": len(content)
})
# Insertion dans ChromaDB
self.collection.add(
ids=[doc_id],
embeddings=[vector.tolist()],
documents=[content],
metadatas=[doc_metadata]
)
return True
except Exception as e:
print(f"Erreur lors de l'ajout du document {doc_id}: {e}")
return False
def add_documents_batch(
self,
documents: List[Dict],
expand_terms: bool = True
) -> Dict[str, int]:
"""
Ajoute plusieurs documents en une seule opération optimisée.
Args:
documents: Liste de dictionnaires avec keys: id, content, metadata
expand_terms: Si True, enrichit avec synonymes médicaux
Returns:
Statistiques d'ajout {"success": N, "failed": M}
"""
if not self.embedder:
raise RuntimeError("Aucun embedder configuré. Appelez set_embedder() d'abord.")
ids = []
vectors = []
contents = []
metadatas = []
print(f"Préparation de {len(documents)} documents pour l'indexation...")
for i, doc in enumerate(documents):
try:
content = doc["content"]
# Enrichissement terminologique
if expand_terms and hasattr(self.embedder, 'expander'):
words = content.replace(",", " ").replace(".", " ").split()[:50]
enriched = self.embedder.expander.generate_medical_context(words)
final_content = f"{content}\n\nTermes médicaux: {enriched}"
else:
final_content = content
vector = self.embedder.embed_single(final_content)
ids.append(doc["id"])
vectors.append(vector.tolist())
contents.append(content)
metadatas.append({
**(doc.get("metadata", {})),
"indexed_at": datetime.now().isoformat(),
"content_length": len(content)
})
if (i + 1) % 100 == 0:
print(f" Préparés: {i + 1}/{len(documents)}")
except Exception as e:
print(f" Erreur document {doc['id']}: {e}")
# Insertion groupée
print("Insertion dans la base vectorielle...")
self.collection.add(
ids=ids,
embeddings=vectors,
documents=contents,
metadatas=metadatas
)
return {"success": len(ids), "failed": len(documents) - len(ids)}
def search(
self,
query: str,
n_results: int = 5,
filter_metadata: Optional[Dict] = None,
expand_query: bool = True
) -> List[Dict]:
"""
Recherche les documents les plus similaires à une requête.
Args:
query: Requête de recherche (peut être en français ou anglais)
n_results: Nombre de résultats à retourner
filter_metadata: Filtres optionnels sur les métadonnées
expand_query: Si True, enrichit avec synonymes médicaux
Returns:
Liste de documents similaires avec scores
"""
if not self.embedder:
raise RuntimeError("Aucun embedder configuré. Appelez set_embedder() d'abord.")
# Expansion terminologique
if expand_query and hasattr(self.embedder, 'expander'):
expanded_terms = self.embedder.expander.expand_query(query)
expanded_query = " | ".join(expanded_terms[:5])
else:
expanded_query = query
# Génération du vecteur de requête
query_vector = self.embedder.embed_single(expanded_query)
# Recherche dans ChromaDB
results = self.collection.query(
query_embeddings=[query_vector.tolist()],
n_results=n_results,
where=filter_metadata,
include=["documents", "metadatas", "distances"]
)
# Formatage des résultats
formatted_results = []
for i in range(len(results["ids"][0])):
formatted_results.append({
"id": results["ids"][0][i],
"content": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"similarity_score": 1 - results["distances"][0][i], # Conversion distance → similarité
"distance": results["distances"][0][i]
})
return formatted_results
def get_stats(self) -> Dict:
"""Retourne les statistiques de la base."""
return {
"total_documents": self.collection.count(),
"collection_name": self.collection.name
}
Example usage with sample medical documents
if __name__ == "__main__":
from medical_embedder import MedicalEmbedder
print("Création de la base vectorielle médicale...")
# Initialize
store = MedicalVectorStore(persist_directory="./ma_base_medicale")
embedder = MedicalEmbedder()
embedder.expander = MedicalTermExpander() # Add expansion capability
store.set_embedder(embedder)
# Sample medical documents for demonstration
sample_documents = [
{
"id": "med001",
"content": "L'hypertension artérielle (HTA) est définie par une pression artérielle
supérieure à 140/90 mmHg. Elle constitue un facteur de risque majeur
pour les maladies cardiovasculaires. Le traitement repose sur les
antihypertenseurs et les modifications du mode de vie.",
"metadata": {"type": "article", "specialty": "cardiologie", "language": "fr"}
},
{
"id": "med002",
"content": "Type 2 diabetes mellitus (DT2) is characterized by insulin resistance
and relative insulin deficiency. Patients often present with polyuria,
polydipsia, and polyphagia. HbA1c monitoring is essential for glycemic control.
Cardiovascular risk is increased in diabetic patients.",
"metadata": {"type": "article", "specialty": "endocrinologie", "language": "en"}
},
{
"id": "med003",
"content": "L'infarctus du myocarde (IDM) survient lorsque l'apport sanguin au muscle
cardiaque est interrompu, généralement par un thrombus coronarien.
Les symptômes incluent une douleur thoracique intense, une dyspnée et
des sueurs. Le diagnostic repose sur l'ECG et les marqueurs biologiques
(troponine).",
"metadata": {"type": "protocol", "specialty": "cardiologie", "language": "fr"}
},
{
"id": "med004",
"content": "Chronic obstructive pulmonary disease (COPD) is a progressive lung disease
characterized by persistent respiratory symptoms and airflow limitation.
Main risk factor is tobacco smoking. Patients present with chronic cough,
sputum production, and dyspnea. Bronchodilators are first-line treatment.",
"metadata": {"type": "guideline", "specialty": "pneumologie", "language": "en"}
},
{
"id": "med005",
"content": "La bronchopneumopathie chronique obstructive (BPCO) est une maladie
pulmonaire caractérisée par une obstruction irréversible des voies aériennes.
Le tabagisme est le principal facteur de risque. Les patients présentent
une toux chronique, une expectoration et un essoufflement à l'effort.",
"metadata": {"type": "article", "specialty": "pneumologie", "language": "fr"}
}
]
# Index documents
stats = store.add_documents_batch(sample_documents)
print(f"\nDocuments indexés: {stats['success']} succès, {stats['failed']} échecs")
# Test search
print("\n--- Tests de recherche ---")
queries = [
"crise cardiaque",
"high blood pressure treatment",
"tabac et poumons"
]
for query in queries:
print(f"\nRequête: '{query}'")
results = store.search(query, n_results=3)
for i, r in enumerate(results, 1):
print(f" {i}. [Score: {r['similarity_score']:.3f}] {r['content'][:80]}...")
print(f"\nStatistiques: {store.get_stats()}")
Ce code peut prendre quelques minutes à s'exécuter si vous avez des milliers de documents. La latence de l'API HolySheep est en moyenne de 47 millisecondes par appel, ce qui reste très performant même pour de gros volumes.
Étape 4 : Intégration avec un Modèle de Génération
Le Pipeline RAG Complet
import os
import requests
from typing import List, Dict, Optional
from dotenv import load_dotenv
from medical_embedder import MedicalEmbedder
from medical_vectorstore import MedicalVectorStore
from medical_term_expander import MedicalTermExpander
load_dotenv()
class MedicalRAGPipeline:
"""
Pipeline complet de