Comment déployer un RAG pour la recherche documentaire : Tutoriel complet 2025 🤖

# 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 😱

❌ 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'urgence

# 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
}

# 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/"]
}

@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")

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>
    );
}

# ✅ 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
}

# ✅ 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"}

# ✅ 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."

# ✅ 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
})

# ✅ 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 : 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 : 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 : 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 : 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 : 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%"
    }
}