Module 10 — Projet : un assistant documentaire complet
Les neuf modules précédents ont construit chacun un maillon. Ce module les assemble en un service utilisable, met en place la partie qui n'a pas encore été traitée — les permissions — et fournit une méthode d'analyse des échecs pour continuer à l'améliorer après la mise en service.
L'architecture cible
utilisateur --> API --> [reecriture] --> [recherche hybride]
|
[filtre permissions]
|
[reclassement]
|
[consigne + generation]
|
[journal + cache]
|
reponse
Une seule requête HTTP en entrée, une réponse structurée en sortie. On peut ajouter des variantes — mode conversationnel avec historique, sortie en flux — mais elles sont des surcouches. Le cœur est ce chemin linéaire.
Le service en une centaine de lignes
FastAPI est un bon défaut : léger, typé, documenté automatiquement.
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
app = FastAPI()
class Requete(BaseModel):
question: str
class Extrait(BaseModel):
numero: int
texte: str
source: str
page: int | None
date_maj: str | None
class Reponse(BaseModel):
reponse: str
extraits: list[Extrait]
abstention: bool
cout_eur: float
temps_ms: int
@app.post("/questions", response_model=Reponse)
def poser(req: Requete, x_utilisateur: str = Header(...)):
profil = charger_profil(x_utilisateur)
if not profil:
raise HTTPException(status_code=401)
debut = time.time()
question = enrichir(reecrire_question(req.question, appeler_modele))
vec = plonger([question])[0]
candidats = fusion_rrf([
chercher_dense(vec, k=20, filtre={"confidentialite": profil["niveaux"]}),
chercher_bm25(question, k=20),
])
candidats = charger_passages(candidats[:20])
candidats = [c for c in candidats if profil["peut_lire"](c)]
top5 = diversifier_par_source(reclasser(question, candidats), k=5)
version = version_index_courante()
cle = clef_reponse(req.question, version)
si_cache = cache_reponses.obtenir(cle)
if si_cache:
return si_cache
messages = construire_consigne(req.question, top5)
reponse_txt, cout = appeler_modele_avec_cout(messages, max_tokens=400)
resultat = Reponse(
reponse=reponse_txt,
extraits=[Extrait(numero=i+1, texte=p["texte"], source=p["source"],
page=p.get("page"), date_maj=p.get("date_maj"))
for i, p in enumerate(top5)],
abstention="Je ne trouve pas" in reponse_txt,
cout_eur=cout,
temps_ms=int((time.time() - debut) * 1000),
)
cache_reponses.poser(cle, resultat)
journaliser(req.question, top5, reponse_txt, resultat.temps_ms, cout)
return resultat
Cent lignes suffisent parce que chaque brique — plonger, chercher_dense, fusion_rrf, reclasser, construire_consigne — a été mise au point dans les modules précédents.
Les permissions par document
Ni le module 4 ni le module 5 ne suffisent seuls : le filtre par métadonnées limite ce que la recherche remonte, mais un vrai contrôle d'accès exige aussi de vérifier au moment de servir, car les métadonnées peuvent avoir été modifiées depuis l'indexation.
Deux niveaux :
Filtre à la recherche : n'index remonte que ce que l'utilisateur a le droit de voir. Optimal pour la performance, mais dépend de métadonnées à jour.
Filtre à la génération : on charge chaque candidat retenu, on vérifie ses permissions actuelles (dans une base séparée, source de vérité), on écarte ceux qui ne passent pas. Nécessaire dès qu'un document peut changer de niveau après indexation.
def peut_lire(profil, passage):
doc = base_documents.charger(passage["document_id"])
if doc.confidentialite == "public":
return True
if doc.confidentialite == "interne":
return profil.est_salarie
if doc.confidentialite == "confidentielle":
return doc.equipe in profil.equipes or profil.est_admin
return False
L'appel à charger doit être caché (cache 60 secondes suffit) sinon la latence explose. Mais jamais de mise en cache à durée illimitée sur les permissions : c'est le point sur lequel une négligence a le plus de conséquences.
Ajouter dans la consigne « ne réponds pas si l'utilisateur n'a pas le droit » n'est pas une politique de sécurité, c'est un vœu pieux. Un modèle peut être manipulé, mais surtout, il ne connaît pas les permissions internes. Un utilisateur n'a jamais accès à un document parce que la fonction peut_lire a renvoyé False, jamais parce que le modèle « a compris » qu'il ne fallait pas répondre.
Une interface minimale
Un formulaire HTML sert de départ. Deux éléments comptent : un champ de question, et une liste des extraits cités cliquables qui ouvrent le document source à la bonne page.
<form onsubmit="poser(event)">
<input name="question" placeholder="Votre question…" required>
<button>Chercher</button>
</form>
<div id="reponse"></div>
<ul id="extraits"></ul>
Une preuve d'utilisabilité tient dans une règle : chaque affirmation de la réponse est accompagnée d'un lien cliquable vers le passage exact. Un utilisateur qui doute d'une phrase vérifie en un clic, et cela suffit à faire adopter le système.
L'analyse des échecs
Le vrai travail commence après la mise en service. Chaque semaine, extraire du journal les cas suivants :
- Abstentions inattendues : questions pour lesquelles le système a répondu par la phrase de repli alors qu'une réponse existe dans le corpus. Cause typique : découpage ou recherche défaillants sur ce type de question.
- Questions à faible utilité : réponses signalées explicitement par les utilisateurs comme incorrectes ou inutiles. Ces retours doivent alimenter le jeu annoté du module 8.
- Répétitions massives d'une même question : signale un besoin réel non couvert par le corpus, à documenter en amont.
- Coûts anormalement élevés : questions dont le coût dépasse trois écarts-types de la moyenne. Souvent une réponse qui déborde du plafond ; parfois un dérèglement du modèle.
def analyser_journal(chemin, jours=7):
seuil = datetime.now(timezone.utc) - timedelta(days=jours)
lignes = [json.loads(l) for l in open(chemin) if l.strip()]
recents = [l for l in lignes if datetime.fromisoformat(l["date"]) > seuil]
abstentions = [l for l in recents if l["abstention"]]
couts = [l["cout_eur"] for l in recents]
print(f"Questions sur {jours} jours : {len(recents)}")
print(f"Taux d'abstention : {len(abstentions) / len(recents):.1%}")
print(f"Cout total : {sum(couts):.2f} EUR")
print(f"Cout par question : {sum(couts) / len(recents):.4f} EUR")
Que rejeter, que corriger
Chaque échec est classé en une des trois causes des modules précédents :
| Symptôme observé | Cause probable | Module à retravailler |
|---|---|---|
| Abstention alors que le fait existe | Rappel de recherche | 3, 4, 5 |
| Réponse contredit le passage cité | Fidélité de la génération | 7, 8 |
| Réponse cite un passage hors sujet | Reclassement défaillant | 6 |
| Fuite d'un document confidentiel | Filtre par permissions | 4, 10 |
| Coût par question qui monte | Consigne trop longue ou modèle bavard | 7, 9 |
L'arbre est simple, mais c'est lui qui distingue l'équipe qui améliore son système chaque semaine de l'équipe qui ajoute des fonctionnalités sans regarder les indicateurs.
Cinq chiffres suffisent : volume de questions, taux d'abstention, taux de succès (retour utilisateur), coût moyen, temps de réponse médian. Publiés chaque lundi, ils orientent la semaine mieux qu'une réunion. La tentation d'y ajouter dix métriques doit être combattue : ce sont les cinq qui bougent qui portent le signal.
En résumé
- Cent lignes de FastAPI assemblent les modules 2 à 9 en un service utilisable, dominé par un chemin linéaire sans branche exceptionnelle.
- Les permissions se contrôlent en deux endroits : à la recherche (métadonnées) et à la génération (base de vérité), jamais dans la consigne au modèle.
- Une interface minimale avec citations cliquables suffit à faire adopter le système ; c'est la vérifiabilité qui construit la confiance.
- L'analyse des échecs hebdomadaire, avec un arbre de causes simple, transforme un système livré en un système qui s'améliore ; sans elle, la qualité stagne ou régresse.
Module suivant : récapituler ces dix modules et présenter l'examen final.