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