Tu connais cette sensation ? Tu es SRE, il est 3h du matin, un service est down, et tu dois retrouver LA procĂ©dure de rollback dans une doc de 500 pages... Tu scrolles, tu cherches, tu maudis le collĂšgue qui a Ă©crit "voir section prĂ©cĂ©dente" sans dire laquelle. đ©
Plot twist : Et si je te disais qu'on peut transformer cette galĂšre en discussion fluide avec un chatbot IA qui connaĂźt toute ta documentation par cĆur ? Chez nous, avec 50 SRE qui jonglent entre incidents et maintenance, l'implĂ©mentation d'un systĂšme RAG a divisĂ© par 3 le temps de recherche d'infos critiques.
Dans ce tutoriel RAG, tu vas apprendre à implémenter un systÚme complet de Retrieval-Augmented Generation sur une documentation MkDocs Material en utilisant LangChain, ChromaDB et FastAPI. Ce guide étape par étape te montre comment construire un assistant documentaire intelligent qui fournit des réponses précises avec citations des sources.
Picture this : 50 SRE, 7 équipes, une documentation MkDocs Material avec :
Le quotidien de nos équipes :
# Scénario classique à 3h du matin
1. Incident dĂ©tectĂ© â Service X down
2. Recherche procĂ©dure â 15 minutes de navigation
3. "Ah non, c'est pas la bonne version"
4. Re-recherche â 10 minutes de plus
5. ProcĂ©dure trouvĂ©e â ENFIN !
Total : 25 minutes perdues en situation critique đ±Les stats qui piquent :
Insight clé : Le problÚme n'était pas la qualité de notre doc, mais son accessibilité cognitive !
Bien que MkDocs Material soit excellent, la recherche par mots-clés traditionnelle a des limites importantes qu'une implémentation RAG peut résoudre :
â Recherche par mots-clĂ©s uniquement (pas de comprĂ©hension sĂ©mantique)
â Pas de comprĂ©hension du contexte entre documents
â RĂ©sultats parfois trop nombreux ou hors-sujet
â Impossible de poser des questions en langage naturel
â Pas d'agrĂ©gation d'infos cross-documents
â Recherche limitĂ©e aux titres et premiers paragraphes
â Aucune notion de prioritĂ© ou d'urgenceLa communautĂ© demande depuis longtemps l'amĂ©lioration de la recherche sĂ©mantique, comme en tĂ©moigne cette issue GitHub qui reste ouverte depuis plusieurs annĂ©es.
Comparaison avec d'autres solutions :
| Solution | Recherche sémantique | IA conversationnelle | Intégration existante |
|---|---|---|---|
| MkDocs Material natif | â | â | â |
| Algolia DocSearch | â ïž LimitĂ©e | â | â ïž Setup complexe |
| RAG + LLM | â | â | â |
| GitBook | â | â ïž Basique | â Migration requise |
Exemple concret :
Voici comment on a construit notre assistant documentaire IA avec une stack RAG complĂšte :
# Stack technique complĂšte pour RAG documentaire
TECH_STACK = {
"backend": "FastAPI", # API REST rapide pour endpoints RAG
"embeddings": "OpenAI text-embedding-3-small", # Embeddings vectoriels (512 dimensions, $0.02/1M tokens)
"vector_db": "ChromaDB", # Base de données vectorielle pour recherche sémantique (alternative: Pinecone, Weaviate)
"llm": "GPT-4o-mini", # LLM pour génération de réponses ($0.15/1M input tokens)
"framework": "LangChain", # Framework d'orchestration RAG
"docs_source": "MkDocs Material",
"deployment": "Docker + K8s",
"monitoring": "Prometheus + Grafana", # Suivi métriques RAG
"cache": "Redis", # Cache sémantique pour performance
}Workflow du RAG :


Le génie de notre approche : pas besoin de modifier MkDocs ! Ce tutoriel RAG te montre comment aspirer le contenu existant et construire un index de base de données vectorielle en parallÚle, permettant la recherche sémantique sans changer ta configuration de documentation actuelle.
# Configuration de base pour l'indexation MkDocs
MKDOCS_CONFIG = {
"docs_path": "/app/docs",
"base_url": "https://docs.company.com",
"chunk_size": 1000, # Optimal pour les runbooks
"chunk_overlap": 200, # Maintient la cohérence
"file_types": [".md"],
"exclude_patterns": ["temp/", "drafts/"]
}Voici l'implémentation FastAPI complÚte pour notre systÚme RAG avec embeddings OpenAI et streaming :
@app.post("/ask")
def ask_question_stream(request: QuestionRequest):
question = request.question
model = rag.llm
# Configuration pour l'URL de base
BASE_DOCS_URL = "https://docs.company.com"
# Retriever optimisé pour les docs techniques
retriever = rag.vector_store.as_retriever(
search_type="mmr", # Maximum Marginal Relevance
search_kwargs={
"k": 8, # 8 chunks pour du contexte riche
"fetch_k": 20, # Pool initial plus large
"lambda_mult": 0.7 # Balance pertinence/diversité
}
)
retrieved_docs = retriever.invoke(question)
if not retrieved_docs:
def empty_response():
yield "â No relevant context found. Are you in the correct folder?"
return StreamingResponse(empty_response(), media_type="text/plain")
# Construction du contexte avec métadonnées enrichies
context_with_metadata = []
sources_found = set()
for i, doc in enumerate(retrieved_docs):
relative_path = doc.metadata.get("relative_path", "Unknown")
file_name = doc.metadata.get("file_name", "Unknown")
chunk_id = doc.metadata.get("chunk_id", i)
# Génération de l'URL cliquable (sans extension)
clean_path = relative_path.replace('.md', '').replace('.mdx', '')
doc_url = f"{BASE_DOCS_URL}/{clean_path}/"
sources_found.add((relative_path, doc_url))
# Contexte enrichi avec headers de section
context_piece = f"""
Source: {relative_path}
URL: {doc_url}
Section: Chunk {chunk_id + 1}
Content:
{doc.page_content}
---"""
context_with_metadata.append(context_piece)
context = "\n".join(context_with_metadata)
# Le prompt systÚme optimisé pour les SRE
system_prompt = (
"đ§ You are a specialized SRE documentation assistant. "
"Your role is to help Site Reliability Engineers find accurate, "
"actionable information quickly during incidents and maintenance.\n\n"
"đ RESPONSE GUIDELINES:\n"
"- Provide clear, step-by-step answers when possible\n"
"- Prioritize emergency procedures and troubleshooting steps\n"
"- Always cite specific documentation sources\n"
"- Include direct links to full documentation\n"
"- If multiple approaches exist, mention alternatives\n\n"
"đŻ FORMAT YOUR RESPONSE:\n"
"## Answer\n"
"[Detailed response with actionable steps]\n\n"
"## đ Sources\n"
"[List each source with clickable links]\n\n"
"Only use information from the provided context. "
"If unsure, acknowledge limitations explicitly."
)
messages = [
{"role": "system", "content": system_prompt},
{"role": "system", "content": f"Context:\n{context}"},
{"role": "user", "content": f"Question: {question}"}
]
def response_stream():
yield f"đ Analyzing {len(retrieved_docs)} chunks from {len(sources_found)} documentation files...\n\n"
for chunk in model.stream(messages):
if chunk.content:
yield chunk.content
# Sources cliquables en fin de réponse
yield "\n\n---\nđ **Complete documentation links:**\n"
for relative_path, doc_url in sorted(sources_found):
yield f"âą [{relative_path}]({doc_url})\n"
return StreamingResponse(response_stream(), media_type="text/plain")Les améliorations clés qu'on a ajoutées :
đŻ Retrieval amĂ©liorĂ© :
đĄ Pro tip : Ces paramĂštres ont Ă©tĂ© ajustĂ©s aprĂšs 2 semaines de tests avec nos Ă©quipes SRE !
import os
import yaml
from pathlib import Path
from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Indexeur de documents basé sur LangChain pour implémentation RAG
class MkDocsIndexer:
def __init__(self, docs_path: str, base_url: str):
self.docs_path = Path(docs_path)
self.base_url = base_url
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=["\n## ", "\n### ", "\n\n", "\n", " ", ""]
)
def load_mkdocs_config(self):
"""Charge la config MkDocs pour respecter la structure"""
config_path = self.docs_path / "mkdocs.yml"
if config_path.exists():
with open(config_path, 'r') as f:
return yaml.safe_load(f)
return {}
def extract_metadata(self, file_path: Path) -> dict:
"""Extrait métadonnées enrichies pour les docs SRE"""
relative_path = file_path.relative_to(self.docs_path)
# Parse front matter pour tags et metadata
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
metadata = {
"source": str(file_path),
"relative_path": str(relative_path),
"file_name": file_path.stem,
"last_modified": file_path.stat().st_mtime
}
# Détection automatique du type de doc
if "runbook" in str(relative_path).lower():
metadata["doc_type"] = "runbook"
elif "api" in str(relative_path).lower():
metadata["doc_type"] = "api_doc"
elif "troubleshoot" in str(relative_path).lower():
metadata["doc_type"] = "troubleshooting"
else:
metadata["doc_type"] = "general"
return metadata
def process_documents(self):
"""Traite tous les documents MkDocs"""
loader = DirectoryLoader(
str(self.docs_path),
glob="**/*.md",
loader_cls=None,
show_progress=True
)
documents = loader.load()
processed_docs = []
for doc in documents:
# Enrichit avec métadonnées
enhanced_metadata = self.extract_metadata(Path(doc.metadata["source"]))
doc.metadata.update(enhanced_metadata)
# Splitting intelligent par sections
chunks = self.text_splitter.split_documents([doc])
# Ajoute chunk_id pour navigation
for i, chunk in enumerate(chunks):
chunk.metadata["chunk_id"] = i
processed_docs.append(chunk)
return processed_docs
# Usage
indexer = MkDocsIndexer("/app/docs", "https://docs.company.com")
documents = indexer.process_documents()# Classification automatique par type de contenu
DOC_TYPES_CONFIG = {
"runbook": {
"weight": 1.5, # Priorité élevée pour incidents
"keywords": ["incident", "rollback", "emergency", "critical"]
},
"api_doc": {
"weight": 1.2,
"keywords": ["endpoint", "authentication", "request", "response"]
},
"troubleshooting": {
"weight": 1.4, # Priorité élevée pour debug
"keywords": ["error", "debug", "logs", "diagnostic"]
},
"general": {
"weight": 1.0,
"keywords": []
}
}# docker-compose.yml pour dev local
version: '3.8'
services:
rag-api:
build: .
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- DOCS_PATH=/app/docs
- BASE_DOCS_URL=https://docs.company.com
volumes:
- ./docs:/app/docs:ro
- ./vector_db:/app/vector_db
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
# Frontend simple pour les tests
rag-frontend:
build: ./frontend
ports:
- "3000:3000"
environment:
- REACT_APP_API_URL=http://localhost:8000// Component React simple mais efficace
function RAGChat() {
const [question, setQuestion] = useState('');
const [response, setResponse] = useState('');
const [loading, setLoading] = useState(false);
const askQuestion = async () => {
setLoading(true);
setResponse('');
try {
const response = await fetch('/api/ask', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ question })
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
setResponse(prev => prev + chunk);
}
} catch (error) {
setResponse('â Error: ' + error.message);
}
setLoading(false);
};
return (
<div className="rag-chat">
<div className="quick-questions">
<h3>đ Questions rapides SRE :</h3>
<button onClick={() => setQuestion("Comment rollback le service auth ?")}>
Rollback Auth Service
</button>
<button onClick={() => setQuestion("Procédure incident critique ?")}>
Incident Critique
</button>
<button onClick={() => setQuestion("Debug erreur 502 gateway ?")}>
Debug 502 Error
</button>
</div>
<textarea
value={question}
onChange={(e) => setQuestion(e.target.value)}
placeholder="Pose ta question sur notre documentation..."
rows={3}
/>
<button onClick={askQuestion} disabled={loading}>
{loading ? 'đ Recherche...' : 'đ€ Demander au RAG'}
</button>
{response && (
<div className="response"
dangerouslySetInnerHTML={{__html: marked(response)}} />
)}
</div>
);
}đŻ Chunking et indexation
# â
DO : Respecter la structure logique des docs
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # Optimal pour docs techniques
chunk_overlap=200, # Maintient le contexte
separators=[
"\n## ", # Sections principales d'abord
"\n### ", # Puis sous-sections
"\n\n", # Paragraphes
"\n", " ", "" # Enfin mots/caractĂšres
]
)
# â
DO : Enrichir les métadonnées
metadata = {
"doc_type": "runbook", # Classification
"urgency": "critical", # Niveau de priorité
"last_updated": timestamp, # FraĂźcheur
"team": "platform", # Ownership
"tags": ["k8s", "auth"] # Concepts clés
}đ Configuration de retrieval intelligente
# â
DO : Ajuster selon le type de question
def get_retriever_config(question_type):
if "emergency" in question.lower() or "incident" in question.lower():
return {"k": 12, "doc_types": ["runbook", "troubleshooting"]}
elif "api" in question.lower():
return {"k": 6, "doc_types": ["api_doc"]}
else:
return {"k": 8, "doc_types": "all"}đ§ Prompt engineering adaptatif
# â
DO : Adapter le prompt selon le contexte SRE
def build_system_prompt(urgency_level, doc_types):
base_prompt = "You are a specialized SRE assistant."
if urgency_level == "critical":
return base_prompt + """
đš CRITICAL INCIDENT MODE:
- Prioritize immediate actionable steps
- Include rollback procedures when relevant
- Mention escalation contacts if available
- Be concise but complete
"""
elif "api" in doc_types:
return base_prompt + """
đĄ API DOCUMENTATION MODE:
- Provide exact endpoint syntax
- Include authentication details
- Show request/response examples
- Mention rate limits and error codes
"""
return base_prompt + "Standard documentation assistance mode."đ Monitoring et mĂ©triques
# â
DO : Tracker les métriques importantes
METRICS_TO_TRACK = {
"usage": ["questions_per_day", "unique_users", "peak_hours"],
"quality": ["avg_response_time", "user_satisfaction", "sources_clicked"],
"content": ["most_asked_topics", "unused_docs", "missing_answers"],
"performance": ["search_latency", "llm_response_time", "error_rate"]
}
# â
DO : Logs structurés pour analytics
logger.info("rag_query", extra={
"question": hash(question), # Privacy-safe
"doc_count": len(retrieved_docs),
"response_time": response_time,
"user_id": user_id,
"urgency": urgency_level
})đ SĂ©curitĂ© et confidentialitĂ©
# â
DO : Implémenter des guardrails
def validate_question(question: str) -> bool:
"""Vérifie que la question est appropriée"""
# Pas de données sensibles dans les logs
if any(pattern in question.lower() for pattern in
["password", "secret", "token", "key"]):
return False
# Limite de taille pour éviter l'abus
if len(question) > 500:
return False
return True
# â
DO : Anonymiser les logs
def sanitize_for_logs(text: str) -> str:
"""Supprime les infos sensibles des logs"""
patterns = [
r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', # IPs
r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', # Emails
r'\b(?:api[-_]?key|token|secret)[-_]?\w*\b' # Credentials
]
for pattern in patterns:
text = re.sub(pattern, '[REDACTED]', text, flags=re.IGNORECASE)
return textâ DON'T : NĂ©gliger la fraĂźcheur des donnĂ©es
# â DON'T : Index statique sans mise Ă jour
# ProblĂšme : Docs obsolĂštes = mauvais conseils en incident !
# â
DO : SystĂšme de mise Ă jour automatique
def schedule_index_updates():
"""Met Ă jour l'index quand les docs changent"""
# Webhook depuis Git pour déclenchements temps réel
@app.post("/webhook/docs-updated")
def handle_docs_update():
asyncio.create_task(reindex_documents())
# Backup : scan périodique des modifications
scheduler.add_job(
func=check_for_updates,
trigger="interval",
minutes=30,
id='docs_freshness_check'
)â DON'T : Ignorer le contexte utilisateur
# â DON'T : RĂ©ponse identique pour tous
# ProblÚme : Junior vs Senior SRE = besoins différents
# â
DO : Adapter selon l'utilisateur
def personalize_response(user_profile, question, base_answer):
if user_profile.experience_level == "junior":
return add_explanatory_context(base_answer)
elif user_profile.team == "security":
return emphasize_security_aspects(base_answer)
elif user_profile.on_call_status:
return prioritize_quick_actions(base_answer)
return base_answerâ DON'T : Faire confiance aveuglĂ©ment au LLM
# â DON'T : Pas de validation des rĂ©ponses critiques
# ProblÚme : Hallucination = incident aggravé !
# â
DO : Validation pour procédures critiques
def validate_critical_response(question, response, doc_sources):
"""Valide les réponses pour procédures sensibles"""
critical_keywords = ["delete", "drop", "destroy", "remove", "rollback"]
if any(keyword in question.lower() for keyword in critical_keywords):
# Exige une source explicite et récente
if not doc_sources or not has_recent_source(doc_sources):
return add_validation_warning(response)
# Double-check avec pattern matching
if not validate_procedure_steps(response):
return add_uncertainty_disclaimer(response)
return response
def add_validation_warning(response):
return f"""
â ïž **ATTENTION : ProcĂ©dure critique dĂ©tectĂ©e**
Cette réponse concerne une opération sensible.
Veuillez vérifier dans la documentation officielle avant d'exécuter.
{response}
đ **Validation requise** : Consultez un Senior SRE si doute
"""â DON'T : Oublier la performance en production
# â DON'T : Pas de mise en cache intelligente
# ProblÚme : Questions répétitives = coûts OpenAI explosés
# â
DO : Cache sémantique avec TTL adaptatif
from functools import lru_cache
import hashlib
class SemanticCache:
def __init__(self):
self.cache = {}
self.similarity_threshold = 0.92
def get_cache_key(self, question: str) -> str:
"""Clé basée sur l'embedding de la question"""
embedding = get_question_embedding(question)
return hashlib.md5(str(embedding).encode()).hexdigest()
def should_cache_response(self, question: str) -> bool:
"""DĂ©cide si une rĂ©ponse mĂ©rite d'ĂȘtre cachĂ©e"""
# Cache les questions fréquentes plus longtemps
if any(term in question.lower() for term in
["how to", "what is", "explain"]):
return True
# Pas de cache pour questions avec timestamps/IDs
if re.search(r'\b\d{10,}\b', question):
return False
return Trueâ DON'T : NĂ©gliger l'expĂ©rience utilisateur
# â DON'T : RĂ©ponses trop techniques pour tous
# ProblÚme : Manager qui pose une question = réponse illisible
# â
DO : Adaptation automatique du niveau
def adjust_technical_level(response: str, user_role: str) -> str:
"""Adapte le niveau technique selon l'utilisateur"""
if user_role in ["manager", "product", "business"]:
return simplify_technical_terms(response)
elif user_role in ["intern", "junior"]:
return add_educational_context(response)
elif user_role in ["senior", "staff", "principal"]:
return add_advanced_details(response)
return response
def simplify_technical_terms(text: str) -> str:
"""Remplace le jargon par des termes simples"""
replacements = {
"rollback": "revenir à la version précédente",
"pod": "conteneur d'application",
"ingress": "point d'entrée du trafic",
"namespace": "environnement isolé"
}
for tech_term, simple_term in replacements.items():
text = text.replace(tech_term, f"{simple_term} ({tech_term})")
return text# Métriques avant/aprÚs RAG
RESULTS = {
"temps_recherche_moyen": {
"avant": "18 minutes/jour/SRE",
"aprĂšs": "6 minutes/jour/SRE",
"amélioration": "-67%"
},
"résolution_incidents": {
"avant": "MTTR = 23 minutes",
"aprĂšs": "MTTR = 16 minutes",
"amélioration": "-30%"
},
"satisfaction_équipe": {
"avant": "6.2/10",
"aprĂšs": "8.7/10",
"amélioration": "+40%"
}
}Top 5 des questions les plus posées au RAG :
Au final, dĂ©ployer un systĂšme RAG sur une documentation MkDocs Material, c'est un peu comme avoir embauchĂ© un SRE senior qui connaĂźt toutes les procĂ©dures par cĆur, ne dort jamais, et rĂ©pond instantanĂ©ment en situation d'urgence. Cette solution de recherche sĂ©mantique transforme l'accĂšs aux connaissances.
Les bénéfices concrets de notre implémentation RAG :
Le plus beau dans tout ça ? Le systÚme RAG s'améliore automatiquement grùce à la récupération intelligente de LangChain. Plus tes équipes posent de questions, plus la base de données vectorielle devient efficace pour trouver le contenu pertinent.
PrĂȘt Ă implĂ©menter RAG pour ta documentation ? Ce tutoriel te donne tout ce qu'il faut pour construire un assistant documentaire IA avec FastAPI, ChromaDB et OpenAI. Ton "3h du matin future-self" te remerciera ! đ
đ„ Challenge bonus : Mesure le temps que tes Ă©quipes passent Ă chercher de l'info cette semaine. Puis remesure dans un mois aprĂšs avoir implĂ©mentĂ© ton RAG. Les rĂ©sultats vont te surprendre !
Merci de me suivre dans cette aventure ! đ
Cet article a Ă©tĂ© Ă©crit avec â€ïž pour la communautĂ© DevOps.