Intermédiaire
🧠 Fondamentaux
20 XP
0 personnes ont réussi
Assembler le pipeline RAG
Tu as toutes les briques : appel API, gestion de conversation, chargement de documents, chunking, base vectorielle, recherche. Maintenant tu vas les assembler en un pipeline RAG complet.
RAG veut dire Retrieval Augmented Generation. Le principe : 1. L'utilisateur pose une question 2. Tu cherches les chunks les plus pertinents dans ta base 3. Tu construis un prompt avec ces chunks comme contexte 4. Tu envoies le tout au LLM 5. Le LLM repond en se basant sur les documents, pas sur ses connaissances generales
C'est la difference entre un chatbot qui invente des reponses (hallucinations) et un assistant qui cite ses sources. En entreprise, cette difference est cruciale : tu ne veux pas qu'un assistant RH invente une politique de conges qui n'existe pas.
Le prompt RAG ressemble a ca :
Tu es un assistant FAQ de l'entreprise. Reponds uniquement en te basant sur les documents suivants. Si la reponse n'est pas dans les documents, dis que tu ne sais pas.
Ecris une fonction construire_prompt_rag(question, resultats_recherche, system_prompt=None) qui prend la question de l'utilisateur, les resultats de la recherche (liste de dicts avec "texte" et "source"), et un system prompt optionnel. Elle renvoie un dictionnaire avec deux cles : "system" (le system prompt complet avec les documents) et "user" (la question). Si system_prompt est None, utilise le system prompt par defaut ci-dessus.
Ecris aussi une fonction repondre_rag(question, base, fonction_llm=None, k=3) qui fait le pipeline complet. Elle prend la question, la base simulee, une fonction LLM optionnelle, et k. Elle renvoie un dictionnaire avec "reponse" (le texte genere), "sources" (la liste des sources utilisees, sans doublons), et "nb_chunks" (le nombre de chunks utilises).
Si fonction_llm est None, la fonction renvoie juste le contexte formate au lieu d'une reponse generee (pour les tests).
Exemple :
base = { "documents": ["Les conges sont de 25 jours.", "Le teletravail est autorise 3j/sem."], "metadatas": [{"source": "rh.txt"}, {"source": "rh.txt"}], "ids": ["c0", "c1"], "count": 2, }
prompt = construire_prompt_rag("Combien de jours de conges ?", [ {"texte": "Les conges sont de 25 jours.", "source": "rh.txt", "score": 0.8} ]) prompt["system"] # contient le system prompt avec le chunk prompt["user"] # "Combien de jours de conges ?"
Tests (4/5)
construire_prompt_rag retourne le bon format
resultats = [{"texte": "Info importante", "source": "doc.txt", "score": 0.9}]
prompt = construire_prompt_rag("Ma question ?", resultats)
assert isinstance(prompt, dict), "La fonction doit retourner un dictionnaire"
assert "system" in prompt, "Le dict doit avoir une cle 'system'"
assert "user" in prompt, "Le dict doit avoir une cle 'user'"
assert prompt["user"] == "Ma question ?"
Le contexte contient les documents
resultats = [
{"texte": "Les conges sont de 25 jours.", "source": "rh.txt", "score": 0.8},
{"texte": "Le teletravail est autorise.", "source": "guide.txt", "score": 0.6},
]
prompt = construire_prompt_rag("Question ?", resultats)
assert "Les conges sont de 25 jours." in prompt["system"], "Le system prompt doit contenir le texte des chunks"
assert "rh.txt" in prompt["system"], "Le system prompt doit contenir la source des chunks"
assert "Le teletravail est autorise." in prompt["system"]
repondre_rag retourne le bon format
base = {
"documents": ["Doc 1 conges jours", "Doc 2 teletravail"],
"metadatas": [{"source": "rh.txt"}, {"source": "rh.txt"}],
"ids": ["c0", "c1"],
"count": 2,
}
r = repondre_rag("conges", base, k=2)
assert isinstance(r, dict)
assert "reponse" in r, "Le resultat doit contenir 'reponse'"
assert "sources" in r, "Le resultat doit contenir 'sources'"
assert "nb_chunks" in r, "Le resultat doit contenir 'nb_chunks'"
Sources sans doublons
base = {
"documents": ["A jours", "B jours", "C jours"],
"metadatas": [{"source": "rh.txt"}, {"source": "rh.txt"}, {"source": "autre.txt"}],
"ids": ["c0", "c1", "c2"],
"count": 3,
}
r = repondre_rag("jours", base, k=3)
# rh.txt apparait 2 fois dans les metadatas mais 1 seule fois dans les sources
assert len(r["sources"]) == len(set(r["sources"])), "Les sources ne doivent pas contenir de doublons"
+ 0 tests cachés
Indices (3 disponibles)
Solution officielle
def construire_prompt_rag(question, resultats_recherche, system_prompt=None):
if system_prompt is None:
system_prompt = "Tu es un assistant FAQ de l'entreprise. Reponds uniquement en te basant sur les documents suivants. Si la reponse n'est pas dans les documents, dis que tu ne sais pas."
contexte = ""
for r in resultats_recherche:
contexte += f"\n---\n{r['texte']}\nSource : {r['source']}\n"
contexte += "---"
system_complet = f"{system_prompt}\n\nDocuments :{contexte}"
return {
"system": system_complet,
"user": question,
}
def repondre_rag(question, base, fonction_llm=None, k=3):
# Etape 1 : recherche
resultats = rechercher_simple(base, question, k=k)
# Etape 2 : construire le prompt
prompt = construire_prompt_rag(question, resultats)
# Etape 3 : appeler le LLM (ou simuler)
if fonction_llm:
reponse = fonction_llm(prompt["system"], prompt["user"])
else:
reponse = f"Contexte utilise : {len(resultats)} chunks trouves."
# Etape 4 : extraire les sources uniques
sources = list(dict.fromkeys(r["source"] for r in resultats))
return {
"reponse": reponse,
"sources": sources,
"nb_chunks": len(resultats),
}
def rechercher_simple(base, question, k=3):
mots_question = set(question.lower().split())
scores = []
for i, doc in enumerate(base["documents"]):
mots_doc = set(doc.lower().split())
mots_communs = mots_question & mots_doc
score = len(mots_communs) / len(mots_question) if mots_question else 0
scores.append({
"texte": doc,
"source": base["metadatas"][i].get("source", "inconnu"),
"score": round(score, 4),
})
scores.sort(key=lambda x: x["score"], reverse=True)
return scores[:k]