Imaginez pouvoir poser des questions en langage naturel à n'importe quel repository GitHub et obtenir des réponses précises basées sur le code réel. C'est exactement ce que permet le RAG (Retrieval-Augmented Generation) appliqué aux bases de code. Dans ce tutoriel complet, nous allons construire ensemble un système de问答 pour GitHub from scratch, sans aucune expérience préalable en API ou en machine learning.

Mon parcours : Après 3 ans à développer des chatbots SaaS, j'ai découvert que la vraie valeur réside dans la capacité à interrogerger le code existant. Quand j'ai intégré HolySheep AI pour ses coûts 85% inférieurs à OpenAI et sa latence inférieure à 50ms, mes projets RAG ont pris une nouvelle dimension. Aujourd'hui, je vous partage ma méthode éprouvée.

Prérequis et architecture du système

Ce dont vous avez besoin

Avant de commencer, préparez votre environnement. Vous aurez besoin de Python 3.10+, Git installé, et un compte HolySheep AI (inscription gratuite avec crédits offerts). L'architecture que nous allons construire se compose de quatre piliers : le téléchargement du repository, le chunking intelligent du code, l'indexation vectorielle, et l'interrogation via LLM.

Installation des dépendances

# Créez un environnement virtuel
python -m venv rag-env
source rag-env/bin/activate  # Sur Windows : rag-env\Scripts\activate

Installez les packages nécessaires

pip install requests beautifulsoup4 langchain-openai faiss-cpu numpy pip install gitpython python-dotenv sentence-transformers

Vérifiez l'installation

python -c "import langchain; print('LangChain installé avec succès')"

Récupération et préparation du code source

Cloner un repository GitHub

La première étape consiste à télécharger le code source que vous souhaitez interroger. Nous allons utiliser la bibliothèque GitPython pour automatiser le clonage, avec gestion des sous-modules et des branches spécifiques.

import os
import shutil
from git import Repo

def clone_repository(repo_url, target_dir="temp_repo", branch="main"):
    """
    Clone un repository GitHub avec gestion des erreurs.
    
    Args:
        repo_url: URL du repository (ex: 'https://github.com/user/repo.git')
        target_dir: Dossier de destination
        branch: Branche à cloner (par défaut: 'main')
    """
    # Nettoyage si le dossier existe déjà
    if os.path.exists(target_dir):
        shutil.rmtree(target_dir)
        print(f"🗑️  Ancien dossier '{target_dir}' supprimé")
    
    try:
        print(f"📥 Clonage de {repo_url} (branche: {branch})...")
        Repo.clone_from(repo_url, target_dir, branch=branch, depth=1)
        print(f"✅ Repository cloné avec succès dans '{target_dir}/'")
        return target_dir
    except Exception as e:
        print(f"❌ Erreur lors du clonage: {e}")
        return None

Exemple d'utilisation

repo_path = clone_repository( "https://github.com/langchain-ai/langchain", target_dir="langchain_repo" )

Extraction intelligente des fichiers de code

Tous les fichiers ne sont pas pertinents pour notre système RAG. Nous allons créer un extracteur qui filtre les fichiers par extension, ignore les node_modules et autres dépendances, et extrait intelligemment le contenu.

import os
from pathlib import Path
from typing import List, Dict

Extensions de fichiers de code à indexer

CODE_EXTENSIONS = { '.py', '.js', '.ts', '.jsx', '.tsx', '.java', '.cpp', '.c', '.go', '.rs', '.rb', '.php', '.swift', '.kt', '.scala', '.cs', '.vue', '.svelte', '.html', '.css', '.scss' }

Dossiers à ignorer (trop de bruit, peu de valeur sémantique)

EXCLUDE_DIRS = { 'node_modules', '__pycache__', '.git', 'venv', 'env', 'dist', 'build', '.next', '__pycache__', 'vendor', 'test_files', 'fixtures', 'examples', 'docs/build' } def extract_code_files(directory: str, max_file_size_kb: int = 500) -> List[Dict[str, str]]: """ Extrait tous les fichiers de code d'un répertoire. Returns: Liste de dictionnaires avec 'path', 'extension', 'content', 'lines' """ code_files = [] root_path = Path(directory) for file_path in root_path.rglob('*'): # Ignorer les dossiers if not file_path.is_file(): continue # Vérifier si le dossier parent est exclu if any(excluded in file_path.parts for excluded in EXCLUDE_DIRS): continue # Vérifier l'extension if file_path.suffix not in CODE_EXTENSIONS: continue # Vérifier la taille file_size_kb = file_path.stat().st_size / 1024 if file_size_kb > max_file_size_kb: print(f"⚠️ Fichier ignoré (trop gros): {file_path.name} ({file_size_kb:.1f}KB)") continue try: with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: content = f.read() lines = content.count('\n') + 1 relative_path = str(file_path.relative_to(root_path)) code_files.append({ 'path': relative_path, 'extension': file_path.suffix, 'content': content, 'lines': lines, 'size_kb': round(file_size_kb, 2) }) except Exception as e: print(f"⚠️ Erreur lecture {file_path}: {e}") print(f"📊 {len(code_files)} fichiers de code extraits") return code_files

Utilisation

files = extract_code_files("langchain_repo") for f in files[:5]: print(f" 📄 {f['path']} ({f['lines']} lignes, {f['size_kb']}KB)")

Chunking intelligent pour le RAG de code

Stratégie de découpage basée sur la structure

Le chunking est crucial pour la qualité du RAG. Pour le code, nous utilisons une approche hybride : découpage par fonction/classe, avec une limite de tokens pour éviter de perdre le contexte. Le format de chunk inclut des métadonnées riches pour améliorer la récupération.

import re
from typing import List, Dict, Tuple

def chunk_code_by_function(content: str, path: str, 
                            max_tokens: int = 800,
                            overlap: int = 100) -> List[Dict]:
    """
    Découpe le code en chunks basés sur les fonctions et classes.
    
    Args:
        content: Contenu du fichier
        path: Chemin du fichier (pour identifier le langage)
        max_tokens: Nombre maximum de tokens par chunk
        overlap: Chevauchement entre chunks (en lignes)
    """
    chunks = []
    lines = content.split('\n')
    extension = path.split('.')[-1] if '.' in path else ''
    
    # Patterns pour différents langages
    patterns = {
        'python': r'^(def |class |async def |async class )',
        'javascript': r'^(export |function |const |let |class |async )',
        'java': r'^(public |private |protected |class |interface )',
        'go': r'^(func |type |package )',
    }
    
    # Trouver les points de découpe
    pattern = patterns.get(extension, patterns.get('python', ''))
    split_points = [0]  # Commence par le début
    
    for i, line in enumerate(lines):
        if re.match(pattern, line.strip()):
            split_points.append(i)
    split_points.append(len(lines))
    
    # Créer les chunks
    for i in range(len(split_points) - 1):
        start = split_points[i]
        end = min(split_points[i + 1], start + max_tokens)
        
        chunk_lines = lines[start:end]
        chunk_content = '\n'.join(chunk_lines)
        
        if len(chunk_content.strip()) > 50:  # Ignorer les chunks vides
            chunks.append({
                'content': chunk_content,
                'path': path,
                'start_line': start + 1,
                'end_line': end,
                'language': extension
            })
    
    return chunks

def create_all_chunks(code_files: List[Dict]) -> List[Dict]:
    """Crée tous les chunks à partir des fichiers de code."""
    all_chunks = []
    
    for file_info in code_files:
        chunks = chunk_code_by_function(
            file_info['content'],
            file_info['path']
        )
        all_chunks.extend(chunks)
    
    print(f"✂️  {len(all_chunks)} chunks créés")
    return all_chunks

Application

chunks = create_all_chunks(files) print(f"📦 Échantillon: {chunks[0]['path']} (lignes {chunks[0]['start_line']}-{chunks[0]['end_line']})")

Indexation vectorielle avec HolySheep AI

Génération des embeddings via l'API HolySheep

Maintenant, nous allons générer des embeddings sémantiques pour chaque chunk. HolySheep AI offre des embeddings de haute qualité avec une latence inférieure à 50ms et des prix imbattables (DeepSeek V3.2 à seulement $0.42/MTok en 2026). Pour les embeddings, nous recommandons le modèle text-embedding-3-small de OpenAI, disponible sur HolySheep.

import requests
import numpy as np
from typing import List, Dict
import time

class HolySheepEmbeddings:
    """Client pour les embeddings HolySheep AI."""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.model = "text-embedding-3-small"
    
    def embed_batch(self, texts: List[str], batch_size: int = 100) -> List[List[float]]:
        """
        Génère des embeddings pour un batch de textes.
        
        Args:
            texts: Liste de textes à embedder
            batch_size: Taille du batch (max 1000 pour l'API)
        """
        all_embeddings = []
        
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }
            
            payload = {
                "model": self.model,
                "input": batch
            }
            
            start_time = time.time()
            response = requests.post(
                f"{self.base_url}/embeddings",
                headers=headers,
                json=payload
            )
            latency = (time.time() - start_time) * 1000
            
            if response.status_code != 200:
                raise Exception(f"Erreur API: {response.status_code} - {response.text}")
            
            data = response.json()
            
            # Calcul du coût (prix HolySheep 2026)
            input_tokens = sum(item['usage']['prompt_tokens'] for item in data['data'])
            cost = (input_tokens / 1_000_000) * 0.42  # $0.42/MTok pour DeepSeek
            
            print(f"  ✅ Batch {i//batch_size + 1}: {len(batch)} textes, "
                  f"{input_tokens} tokens, latence {latency:.1f}ms, coût ${cost:.6f}")
            
            embeddings = [item['embedding'] for item in data['data']]
            all_embeddings.extend(embeddings)
        
        return all_embeddings

Initialisation (remplacez par votre clé)

client = HolySheepEmbeddings(api_key="YOUR_HOLYSHEEP_API_KEY")

Génération des embeddings

texts_to_embed = [chunk['content'] for chunk in chunks[:50]] # Test avec 50 chunks embeddings = client.embed_batch(texts_to_embed) print(f"\n📊 {len(embeddings)} embeddings générés") print(f" Dimension: {len(embeddings[0])}")

Index FAISS pour une recherche rapide

import faiss
import numpy as np
from typing import List, Dict, Tuple

class CodeVectorIndex:
    """Index vectoriel optimisé pour la recherche de code."""
    
    def __init__(self, dimension: int = 1536):
        self.dimension = dimension
        self.index = None
        self.chunks = []
        self.embeddings_matrix = None
    
    def build_index(self, chunks: List[Dict], embeddings: List[List[float]]):
        """
        Construit l'index FAISS à partir des chunks et embeddings.
        
        Args:
            chunks: Liste des chunks de code
            embeddings: Liste des vecteurs d'embedding
        """
        self.chunks = chunks
        
        # Conversion en numpy array
        self.embeddings_matrix = np.array(embeddings).astype('float32')
        
        # Normalisation L2 pour une meilleure recherche
        faiss.normalize_L2(self.embeddings_matrix)
        
        # Index IVF (Inverted File) pour rapidité
        nlist = min(50, len(chunks) // 10)
        quantizer = faiss.IndexFlatIP(self.dimension)  # Inner Product pour vecteurs normalisés
        self.index = faiss.IndexIVFFlat(quantizer, self.dimension, nlist, faiss.METRIC_INNER_PRODUCT)
        
        # Entraînement de l'index
        print(f"🎯 Entraînement de l'index ({nlist} clusters)...")
        self.index.train(self.embeddings_matrix)
        
        # Ajout des vecteurs
        print(f"➕ Ajout de {len(embeddings)} vecteurs...")
        self.index.add(self.embeddings_matrix)
        
        # Force l'index à être en mode recherche
        self.index.nprobe = 10  # Nombre de clusters à explorer
        
        print(f"✅ Index créé: {self.index.ntotal} vecteurs indexés")
    
    def search(self, query_embedding: List[float], top_k: int = 5) -> List[Dict]:
        """
        Recherche les chunks les plus pertinents.
        
        Args:
            query_embedding: Vecteur de la requête
            top_k: Nombre de résultats
        """
        query_vector = np.array([query_embedding]).astype('float32')
        faiss.normalize_L2(query_vector)
        
        distances, indices = self.index.search(query_vector, top_k)
        
        results = []
        for i, idx in enumerate(indices[0]):
            if idx < len(self.chunks):
                results.append({
                    'chunk': self.chunks[idx],
                    'score': float(distances[0][i]),
                    'rank': i + 1
                })
        
        return results

Construction de l'index

vector_index = CodeVectorIndex(dimension=len(embeddings[0])) vector_index.build_index(chunks[:50], embeddings)

Test de recherche

test_results = vector_index.search(embeddings[0], top_k=3) print("\n🔍 Résultats de recherche test:") for r in test_results: print(f" #{r['rank']} Score: {r['score']:.3f} - {r['chunk']['path']}")

Interrogation du système avec génération de réponse

Implémentation du système RAG complet

Maintenant que nous avons indexé le code, nous pouvons interroger le système. L'utilisateur pose une question en langage naturel, nous récupérons les chunks pertinents via FAISS, et nous générons une réponse via un LLM. HolySheep AI propose GPT-4.1 à $8/MTok ou le très économique DeepSeek V3.2 à $0.42/MTok.

import requests
import json
from typing import List, Dict, Optional

class CodeRAGSystem:
    """Système complet de问答 pour repositories GitHub."""
    
    def __init__(self, api_key: str, embeddings_client: HolySheepEmbeddings, 
                 vector_index: CodeVectorIndex):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.embeddings_client = embeddings_client
        self.vector_index = vector_index
    
    def query(self, question: str, top_k: int = 5, 
              model: str = "deepseek-chat") -> Dict:
        """
        Interroge le système RAG avec une question en langage naturel.
        
        Args:
            question: Question de l'utilisateur
            top_k: Nombre de chunks de contexte à utiliser
            model: Modèle LLM à utiliser
        """
        # 1. Embedding de la question
        print(f"❓ Question: {question}")
        question_embedding = self.embeddings_client.embed_batch([question])[0]
        
        # 2. Recherche des chunks pertinents
        relevant_chunks = self.vector_index.search(question_embedding, top_k=top_k)
        
        # 3. Construction du contexte
        context = self._build_context(relevant_chunks)
        
        # 4. Génération de la réponse via LLM
        response = self._generate_response(question, context, model)
        
        return {
            'question': question,
            'answer': response['content'],
            'sources': [c['chunk']['path'] for c in relevant_chunks],
            'usage': response.get('usage'),
            'cost': response.get('cost')
        }
    
    def _build_context(self, chunks: List[Dict]) -> str:
        """Construit le prompt contextuel à partir des chunks."""
        context_parts = []
        
        for i, item in enumerate(chunks):
            chunk = item['chunk']
            context_parts.append(
                f"--- Chunk {i+1} ({chunk['path']}, lignes {chunk['start_line']}-{chunk['end_line']}) ---\n"
                f"``\n{chunk['content']}\n``"
            )
        
        return "\n\n".join(context_parts)
    
    def _generate_response(self, question: str, context: str, 
                          model: str) -> Dict:
        """Génère la réponse via l'API HolySheep."""
        prompt = f"""Tu es un assistant expert en analyse de code. Réponds à la question en te basant EXCLUSIVEMENT sur le contexte fourni.

CONTEXTE:
{context}

QUESTION: {question}

RÈGLES:
1. Cite toujours les fichiers sources utilisés (chemin + numéro de ligne)
2. Si l'information n'est pas dans le contexte, dis-le explicitement
3. Explique le code de manière claire et pédagogique
4. Propose du code si pertinent

RÉPONSE:"""
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        # Prix par modèle (2026)
        model_prices = {
            "deepseek-chat": 0.42,
            "gpt-4.1": 8.0,
            "claude-sonnet-4.5": 15.0
        }
        price_per_mtok = model_prices.get(model, 0.42)
        
        payload = {
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.3,
            "max_tokens": 1000
        }
        
        start = time.time()
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload
        )
        latency_ms = (time.time() - start) * 1000
        
        if response.status_code != 200:
            raise Exception(f"Erreur LLM: {response.status_code} - {response.text}")
        
        data = response.json()
        content = data['choices'][0]['message']['content']
        
        # Calcul du coût
        total_tokens = data['usage']['total_tokens']
        cost = (total_tokens / 1_000_000) * price_per_mtok
        
        return {
            'content': content,
            'usage': data['usage'],
            'latency_ms': round(latency_ms, 1),
            'cost': round(cost, 6)
        }

Initialisation du système

rag_system = CodeRAGSystem( api_key="YOUR_HOLYSHEEP_API_KEY", embeddings_client=client, vector_index=vector_index )

Test du système

result = rag_system.query( "Comment fonctionne la classe Chain dans ce code?", top_k=3, model="deepseek-chat" ) print(f"\n🤖 Réponse:") print(result['answer']) print(f"\n📚 Sources: {result['sources']}") print(f"💰 Coût: ${result['cost']}, Latence: {result['latency_ms']}ms")

Création d'une interface web simple

Pour rendre le système accessible, créons une interface web minimaliste avec Flask. Cette interface permet de coller une URL GitHub, d'indexer le repository, puis de poser des questions.

from flask import Flask, request, jsonify, render_template_string
import tempfile
import os

app = Flask(__name__)

Variables globales pour l'état

current_rag = None HTML_TEMPLATE = ''' GitHub Code Q&A

🦄 GitHub Repository Q&A

Interrogez n'importe quel repository GitHub en langage naturel

📥 Étape 1: Charger un Repository

❓ Étape 2: Poser une Question

''' @app.route('/') def index(): return render_template_string(HTML_TEMPLATE) @app.route('/api/index', methods=['POST']) def index_repo(): global current_rag data = request.json # Clonage repo_path = clone_repository(data['repo_url'], branch=data.get('branch', 'main')) if not repo_path: return jsonify({'success': False, 'error': 'Échec du clonage'}) # Extraction et chunking files = extract_code_files(repo_path) chunks = create_all_chunks(files) # Embeddings embeddings = client.embed_batch([c['content'] for c in chunks]) # Index index = CodeVectorIndex() index.build_index(chunks, embeddings) # Création du RAG current_rag = CodeRAGSystem( api_key="YOUR_HOLYSHEEP_API_KEY", embeddings_client=client, vector_index=index ) return jsonify({ 'success': True, 'stats': {'files': len(files), 'chunks': len(chunks)} }) @app.route('/api/query', methods=['POST']) def query(): global current_rag if not current_rag: return jsonify({'success': False, 'error': 'Aucun repository indexé'}) data = request.json result = current_rag.query( data['question'], model=data.get('model', 'deepseek-chat') ) return jsonify({ 'success': True, **result }) if __name__ == '__main__': print("🚀 Serveur démarré sur http://localhost:5000") app.run(debug=True, port=5000)

Erreurs courantes et solutions

Erreur 1: "401 Unauthorized - Invalid API key"

# ❌ ERREUR:

requests.exceptions.HTTPError: 401 Client Error: Unauthorized

{"error": {"message": "Invalid API Key", "type": "invalid_request_error"}}

✅ SOLUTION:

Vérifiez votre clé API dans le dashboard HolySheep

Assurez-vous que la clé est correctement définie

import os from dotenv import load_dotenv

Chargez les variables d'environnement depuis .env

load_dotenv()

Utilisez la clé comme ceci:

api_key = os.getenv("HOLYSHEEP_API_KEY") if not api_key: raise ValueError("HOLYSHEEP_API_KEY non définie dans .env")

Créez le fichier .env à la racine du projet:

HOLYSHEEP_API_KEY=votre_cle_ici

print(f"✅ Clé API chargée: {api_key[:10]}...")

Erreur 2: "Rate limit exceeded"

# ❌ ERREUR:

429 Too Many Requests

{"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}}

✅ SOLUTION:

Implémentez un système de retry avec backoff exponentiel

import time import requests def call_with_retry(url, headers, payload, max_retries=5, base_delay=1): """Appel API avec retry automatique.""" for attempt in range(max_retries): try: response = requests.post(url, headers=headers, json=payload) if response.status_code == 200: return response.json() elif response.status_code == 429: # Rate limit - on attend et on réessaie wait_time = base_delay * (2 ** attempt) print(f"⏳ Rate limit atteint, attente de {wait_time}s...") time.sleep(wait_time) else: response.raise_for_status() except requests.exceptions.RequestException as e: if attempt == max_retries - 1: raise wait_time = base_delay * (2 ** attempt) print(f"⚠️ Erreur: {e}, retry dans {wait_time}s...") time.sleep(wait_time) raise Exception(f"Échec après {max_retries} tentatives")

Erreur 3: "Embedding dimension mismatch"

# ❌ ERREUR:

ValueError: vector dimension mismatch: 1536 != 768

✅ SOLUTION:

Le problème vient du modèle d'embedding utilisé.

HolySheep utilise par défaut text-embedding-3-small (1536 dims).

Vérifiez que votre index utilise la bonne dimension.

from langchain_openai import OpenAIEmbeddings

Configuration correcte pour HolySheep

embedding_model = OpenAIEmbeddings( model="text-embedding-3-small", openai_api_base="https://api.holysheep.ai/v1", openai_api_key="YOUR_HOLYSHEEP_API_KEY" )

Vérifiez la dimension de votre modèle

test_embedding = embedding_model.embed_query("test") print(f"Dimension des embeddings: {len(test_embedding)}")

Si vous utilisez un modèle différent (text-embedding-3-large = 3076 dims),

ajustez l'initialisation de CodeVectorIndex:

vector_index = CodeVectorIndex(dimension=len(test_embedding))

Erreur 4: "Context window exceeded"

# ❌ ERREUR:

400 Bad Request - maximum context length exceeded

✅ SOLUTION:

Réduisez le nombre de chunks ou leur taille

class SmartChunkSelector: """Sélectionne intelligemment les chunks les plus pertinents.""" def __init__(self, rag_system, max_context_tokens=6000): self.rag = rag_system self.max_tokens = max_context_tokens def get_context(self, question: str, top_k: int = 5) -> str: """Retourne un contexte qui tient dans la fenêtre de tokens.""" # Estimation: 1 token ≈ 4 caractères en moyenne char_limit = self.max_tokens * 4 chunks = self.rag.vector_index.search( self.rag.embeddings_client.embed_batch([question])[0], top_k=top_k * 2 # On récupère plus pour filtrer ) context_parts = [] current_length = 0 for item in chunks: chunk_content = item['chunk']['content'] chunk_length = len(chunk_content) # Ajout du chunk si on respecte la limite if current_length + chunk_length <= char_limit: context_parts.append(chunk_content) current_length += chunk_length else: # Tronquer le dernier chunk si nécessaire remaining = char_limit - current_length if remaining > 500: # Garder si reste assez de place context_parts.append(chunk_content[:remaining]) break return "\n\n".join(context_parts)

Erreur 5: "Git clone timeout on large repositories"

# ❌ ERREUR:

git.exc.GitCommandError: Cloning into 'repo'...

fatal: early EOF

fatal: index-pack failed

✅ SOLUTION:

Clone shallow avec profondeur limitée et retry

import subprocess import shutil from pathlib import Path def clone_repository_robust(repo_url, target_dir, branch="main", depth=50): """ Clone robuste avec gestion des erreurs réseau. """ # Nettoyage if Path(target_dir).exists(): shutil.rmtree(target_dir) max_retries = 3 for attempt in range(max_retries): try: # Clone shallow (rapide) avec profundidad limitada result = subprocess.run([ 'git', 'clone', '--depth', str(depth), '--branch', branch, '--single-branch', '--filter=blob:none', # Blobless clone (plus rapide) repo_url, target_dir ], capture_output=True, text=True, timeout=300