Interroger, récupérer et générer
Écrivez le pipeline de requête qui transforme la question de l’utilisateur en embedding, récupère les k segments les mieux classés, formate un prompt enrichi, appelle le LLM et renvoie une réponse citée.
Interroger, récupérer et générer est une leçon AI Engineering Academy gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Engineering Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Engineering Academy comprend 4 leçons au total.
Le pipeline de requête : de bout en bout
Le pipeline de requête constitue la partie en ligne de RAG : c'est le code qui s'exécute en temps réel lorsqu'un utilisateur pose une question. Il relie tous les composants créés lors de l'indexation : le modèle de représentation vectorielle, le magasin de vecteurs, le modèle de prompt et le LLM. Un pipeline de requête correctement implémenté s'exécute en moins de 500 ms pour la plupart des charges de travail et produit des réponses étayées et accompagnées de citations. Dans cette leçon, nous construisons chaque étape à partir de zéro.
Étape 1 : vectoriser la requête de l'utilisateur
La première étape consiste à convertir la question formulée en langage naturel par l'utilisateur en une représentation vectorielle à l'aide du même modèle que celui utilisé lors de l'indexation. Cette représentation encode le sens sémantique de la question et sera comparée aux représentations vectorielles des segments de documents dans le magasin de vecteurs. Veillez à ce que cette étape soit rapide : utilisez un modèle léger comme text-embedding-3-small et mettez en cache les représentations des requêtes identiques répétées.
from openai import OpenAI
client = OpenAI()
def embed_query(question: str) -> list:
response = client.embeddings.create(
model='text-embedding-3-small',
input=[question]
)
return response.data[0].embedding
user_question = 'What is our remote work policy?'
query_vector = embed_query(user_question)
print(f'Query embedded: {len(query_vector)}-dim vector')Étape 2 : récupérer les K premiers segments
Envoyez le vecteur de requête au magasin de vecteurs afin de trouver les K segments les plus similaires sur le plan sémantique. Les correspondances renvoyées sont classées selon un score de similarité cosinus (généralement compris entre 0,0 et 1,0 ; plus le score est élevé, meilleur est le résultat). La valeur idéale de K établit un équilibre entre la richesse du contexte et le coût de la fenêtre de contexte : K=5 constitue un point de départ courant. Vous pouvez également appliquer ici des filtres de métadonnées pour limiter la récupération à un service, un type de document ou une plage de dates précis.
def retrieve_chunks(query_vector, index, top_k=5, filters=None):
query_params = {
'vector': query_vector,
'top_k': top_k,
'include_metadata': True
}
if filters:
query_params['filter'] = filters
results = index.query(**query_params)
chunks = []
for match in results.matches:
chunks.append({
'score': match.score,
'text': match.metadata['text'],
'source': match.metadata.get('source', ''),
'page': match.metadata.get('page', '')
})
return chunksÉtape 3 : filtrer selon un seuil de score
Les segments récupérés ne sont pas tous réellement pertinents : certains peuvent avoir de faibles scores de similarité tout en figurant parmi les K premiers parce que la requête se situe en dehors de la couverture de l'index. Appliquez un seuil de score minimal pour éliminer les correspondances peu fiables. Si tous les segments récupérés sont sous ce seuil, renvoyez une réponse indiquant qu'aucune information n'a été trouvée plutôt que d'envoyer un contexte non pertinent au LLM, ce qui produirait une réponse moins bonne qu'un refus formulé correctement.
MIN_SCORE_THRESHOLD = 0.75
def filter_by_score(chunks, threshold=MIN_SCORE_THRESHOLD):
relevant = [c for c in chunks if c['score'] >= threshold]
if not relevant:
print(f'No chunks above threshold {threshold}. Scores: {[c["score"] for c in chunks]}')
return relevant
retrieved = retrieve_chunks(query_vector, index, top_k=5)
filtered = filter_by_score(retrieved)
if not filtered:
print('Responding: no relevant information found')Étape 4 : mettre en forme le bloc de contexte
Assemblez les segments récupérés dans un bloc de contexte structuré que le LLM pourra lire. Étiquetez chaque segment avec sa source afin que le modèle puisse le citer correctement. Ajoutez un séparateur entre les segments pour améliorer la lisibilité. Maintenez le contexte total dans les limites de votre budget de jetons : comptez les jetons avec tiktoken et tronquez ou supprimez les segments ayant les scores les plus faibles si vous dépassez la limite. Le bloc de contexte est inséré dans le prompt entre l'instruction système et la question de l'utilisateur.
def format_context(chunks):
parts = []
for i, chunk in enumerate(chunks, start=1):
source_label = chunk['source']
if chunk.get('page'):
source_label += f", page {chunk['page']}"
parts.append(
f'[Document {i} | Source: {source_label}]\n{chunk["text"]}'
)
return '\n\n---\n\n'.join(parts)
context = format_context(filtered)
print(f'Context block: {len(context)} characters')Étape 5 : créer le prompt augmenté
Combinez le bloc de contexte, l'instruction système et la question de l'utilisateur dans le prompt final. Le message système indique au modèle d'utiliser uniquement le contexte fourni et de citer ses sources. Le message utilisateur contient le contexte mis en forme, suivi de la question. Cette séparation claire empêche le modèle de mélanger le contenu du contexte avec la question et rend la frontière entre les données récupérées et la saisie de l'utilisateur parfaitement explicite.
def build_prompt(question, context):
system_message = (
'You are a helpful assistant. Answer the question using ONLY '
'the information in the provided documents. '
'Cite the document number(s) used, like [Doc 1]. '
'If the documents do not contain the answer, say so.'
)
user_message = (
f'Documents:\n\n{context}\n\n'
f'Question: {question}'
)
return system_message, user_messageÉtape 6 : appeler le LLM et obtenir la réponse
Envoyez le prompt assemblé au LLM à l'aide de l'API Chat Completions. Utilisez une température basse (de 0,0 à 0,3) pour les questions-réponses factuelles afin d'obtenir des réponses cohérentes et étayées. Des températures plus élevées produisent des réponses plus créatives, mais augmentent le risque que le modèle ajoute des informations absentes du contexte. Analysez la réponse et renvoyez à la fois le texte de la réponse et les sources récupérées afin que votre application puisse afficher les citations à l'utilisateur.
def generate_answer(question, context, sources):
system_msg, user_msg = build_prompt(question, context)
response = client.chat.completions.create(
model='gpt-4o',
temperature=0.1, # low temperature for factual Q&A
messages=[
{'role': 'system', 'content': system_msg},
{'role': 'user', 'content': user_msg}
]
)
answer = response.choices[0].message.content
return {
'answer': answer,
'sources': sources,
'tokens_used': response.usage.total_tokens
}Assembler le tout
Le pipeline de requête complet exécute ces étapes dans l'ordre. Chaque étape est une fonction pure que vous pouvez tester indépendamment, et les données circulent proprement d'une étape à la suivante. L'ajout d'une journalisation à chaque étape rend le pipeline observable : vous pouvez voir précisément quels segments ont été récupérés, quel était leur score, comment le contexte a été assemblé et combien de jetons ont été utilisés. Cette visibilité est essentielle pour déboguer le système et améliorer la qualité de la récupération.
def answer_question(user_question, vector_index):
# Step 1: Embed query
q_vector = embed_query(user_question)
# Step 2: Retrieve
chunks = retrieve_chunks(q_vector, vector_index, top_k=5)
# Step 3: Filter low-confidence matches
chunks = filter_by_score(chunks, threshold=0.70)
if not chunks:
return {'answer': 'I do not have information about that topic.', 'sources': []}
# Step 4 & 5: Format and build prompt
context = format_context(chunks)
sources = [c['source'] for c in chunks]
# Step 6: Generate
return generate_answer(user_question, context, sources)Optimisation de la latence
Le pipeline de requête comporte deux étapes limitées par les entrées-sorties : l'appel de représentation vectorielle et l'appel au LLM. Exécutez-les sans attentes inutiles : l'appel de représentation vectorielle est rapide (<100 ms), tandis que l'appel au LLM est lent (500 ms à 3 s). Pour réduire la latence perçue, diffusez la réponse du LLM en continu afin que les jetons apparaissent au fur et à mesure de leur génération, au lieu d'attendre la réponse complète. Mettez en cache la représentation des requêtes identiques répétées pour éviter les appels d'API redondants.
async def answer_question_streaming(question, index):
q_vector = embed_query(question)
chunks = retrieve_chunks(q_vector, index, top_k=5)
chunks = filter_by_score(chunks)
if not chunks:
yield 'I do not have information about that topic.'
return
context = format_context(chunks)
system_msg, user_msg = build_prompt(question, context)
stream = await client.chat.completions.create(
model='gpt-4o',
stream=True,
messages=[
{'role': 'system', 'content': system_msg},
{'role': 'user', 'content': user_msg}
]
)
async for chunk in stream:
delta = chunk.choices[0].delta.content or ''
yield deltaJournalisation pour l'observabilité
Les pipelines RAG utilisés en production ont besoin d'une journalisation structurée afin que vous puissiez diagnostiquer les échecs de récupération ou les mauvaises réponses du LLM. Journalisez pour chaque requête la requête, les ID et les scores des segments récupérés, le nombre de jetons du contexte, la réponse et la latence. Stockez ces journaux dans une base de données ou une plateforme d'observabilité. Lorsque des utilisateurs signalent de mauvaises réponses, vous pouvez rejouer la requête exacte et examiner quels segments ont été récupérés et pourquoi ils étaient insuffisants.
import time
import logging
import json
def answer_question_with_logging(question, index):
start = time.time()
q_vector = embed_query(question)
chunks = retrieve_chunks(q_vector, index, top_k=5)
chunks = filter_by_score(chunks)
context = format_context(chunks)
result = generate_answer(question, context, [c['source'] for c in chunks])
latency_ms = (time.time() - start) * 1000
log_entry = {
'question': question,
'num_chunks_retrieved': len(chunks),
'chunk_scores': [c['score'] for c in chunks],
'tokens_used': result.get('tokens_used'),
'latency_ms': round(latency_ms)
}
logging.info(json.dumps(log_entry))
return resultMise en cache des représentations de requêtes
Si votre application reçoit de nombreuses requêtes répétées ou presque identiques — par exemple dans des robots FAQ où les utilisateurs posent souvent les mêmes questions — la mise en cache des représentations de requêtes constitue une optimisation simple et très efficace. Calculez une empreinte de la chaîne de requête, recherchez la représentation correspondante dans un cache Redis et n'appelez l'API de représentation vectorielle qu'en cas d'absence dans le cache. Dans les robots FAQ et d'assistance utilisés en production, les taux d'accès au cache des représentations atteignent couramment 30 à 60 %, ce qui élimine une part importante des coûts d'API et réduit la latence de 50 à 100 ms par requête mise en cache.
import hashlib
import json
import redis
r = redis.Redis(host='localhost', port=6379)
EMBED_CACHE_TTL = 86400 # 24 hours
def embed_query_cached(question):
cache_key = 'embed:' + hashlib.sha256(question.encode()).hexdigest()
cached = r.get(cache_key)
if cached:
return json.loads(cached) # cache hit
# Cache miss: call the API
vector = embed_query(question)
r.setex(cache_key, EMBED_CACHE_TTL, json.dumps(vector))
return vectorVérification rapide
Vérifiez votre compréhension des concepts d'ingénierie de l'IA présentés dans cette leçon.
Récapitulatif de la leçon
Dans cette leçon, vous avez appris les six étapes du pipeline de requête (vectoriser la requête, récupérer les segments, filtrer selon le score, mettre en forme le contexte, créer le prompt, générer la réponse), le filtrage selon un seuil de score pour gérer les requêtes situées en dehors de la couverture de l'index, ainsi que les améliorations destinées à la production, notamment la diffusion continue des réponses, la journalisation structurée et l'optimisation de la latence. Nous allons maintenant apprendre à évaluer si votre système RAG complet fonctionne réellement correctement.
Questions Fréquemment Posées
La leçon « Interroger, récupérer et générer » est-elle gratuite ?
Oui — le texte complet de « Interroger, récupérer et générer » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Engineering Academy, passe à CoddyKit PRO. Le cours AI Engineering Academy comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Interroger, récupérer et générer » ?
Écrivez le pipeline de requête qui transforme la question de l’utilisateur en embedding, récupère les k segments les mieux classés, formate un prompt enrichi, appelle le LLM et renvoie une réponse ci… Tu pratiques AI Engineering Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer AI Engineering Academy ?
Aucune expérience préalable n'est requise. AI Engineering Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.
Combien de temps prend la leçon « Interroger, récupérer et générer » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon AI Engineering Academy ?
Oui. Chaque leçon AI Engineering Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Chargement des documents et extraction de texte
- Stratégies de segmentation : fixe, par phrases ou récursive
- Indexer : créer et stocker les segments
- Interroger, récupérer et générer