Optimisation du RAG des ateliers : seuil de pertinence, recherche hybride, cache et overlap
- Seuil rag_min_score (defaut 0.30) : plus d'extraits hors-sujet injectes dans le prompt - Recherche hybride : cosinus + bonus lexical (noms propres JdR mieux retrouves) - Cache memoire du vector store (invalidation mtime) : plus de re-parse JSON par question - Overlap de 80 tokens entre extraits RAG consecutifs (phrases a cheval retrouvables) - Script de non-regression brain/scripts/sanity_rag_check.py Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -13,8 +13,19 @@ from __future__ import annotations
|
||||
CHUNK_TARGET_TOKENS = 6000
|
||||
|
||||
|
||||
def chunk_text(full_text: str, target_tokens: int = CHUNK_TARGET_TOKENS) -> list[str]:
|
||||
"""Découpe `full_text` en morceaux ~`target_tokens` tokens (frontières de §)."""
|
||||
def chunk_text(
|
||||
full_text: str,
|
||||
target_tokens: int = CHUNK_TARGET_TOKENS,
|
||||
overlap_tokens: int = 0,
|
||||
) -> list[str]:
|
||||
"""Découpe `full_text` en morceaux ~`target_tokens` tokens (frontières de §).
|
||||
|
||||
`overlap_tokens` > 0 : chaque morceau reprend la fin du précédent (les derniers
|
||||
paragraphes, jusqu'à ~`overlap_tokens` tokens). Utile pour le RAG : une phrase-clé
|
||||
à cheval sur deux morceaux reste retrouvable dans au moins l'un des deux. À
|
||||
laisser à 0 pour les imports (recopie) : un overlap y DUPLIQUERAIT du texte.
|
||||
Un morceau peut légèrement dépasser la cible (jusqu'à target + overlap).
|
||||
"""
|
||||
if not full_text.strip():
|
||||
return []
|
||||
|
||||
@@ -26,32 +37,65 @@ def chunk_text(full_text: str, target_tokens: int = CHUNK_TARGET_TOKENS) -> list
|
||||
chunks: list[str] = []
|
||||
current: list[str] = []
|
||||
current_tokens = 0
|
||||
fresh = False # `current` contient-il du contenu pas encore émis ? (évite de
|
||||
# ré-émettre un morceau composé uniquement de l'overlap en fin de texte)
|
||||
for para in paragraphs:
|
||||
para_tokens = len(enc.encode(para))
|
||||
# Un paragraphe seul plus gros que la cible : on le coupe en sous-blocs.
|
||||
if para_tokens > target_tokens:
|
||||
if current:
|
||||
if current and fresh:
|
||||
chunks.append("\n\n".join(current))
|
||||
current, current_tokens = [], 0
|
||||
chunks.extend(_split_oversized(para, enc, target_tokens))
|
||||
current, current_tokens, fresh = [], 0, False
|
||||
chunks.extend(_split_oversized(para, enc, target_tokens, overlap_tokens))
|
||||
continue
|
||||
if current_tokens + para_tokens > target_tokens and current:
|
||||
chunks.append("\n\n".join(current))
|
||||
current, current_tokens = [], 0
|
||||
if fresh:
|
||||
chunks.append("\n\n".join(current))
|
||||
current, current_tokens = _overlap_tail(current, enc, overlap_tokens)
|
||||
fresh = False
|
||||
current.append(para)
|
||||
current_tokens += para_tokens
|
||||
fresh = True
|
||||
|
||||
if current:
|
||||
if current and fresh:
|
||||
chunks.append("\n\n".join(current))
|
||||
return chunks
|
||||
|
||||
|
||||
def _split_oversized(paragraph: str, enc, target_tokens: int) -> list[str]:
|
||||
"""Coupe un paragraphe géant en sous-blocs ~`target_tokens` tokens."""
|
||||
def _overlap_tail(parts: list[str], enc, overlap_tokens: int) -> tuple[list[str], int]:
|
||||
"""Derniers paragraphes de `parts` totalisant au plus `overlap_tokens` tokens —
|
||||
le « rappel » recopié en tête du morceau suivant."""
|
||||
if overlap_tokens <= 0 or not parts:
|
||||
return [], 0
|
||||
tail: list[str] = []
|
||||
total = 0
|
||||
for para in reversed(parts):
|
||||
para_tokens = len(enc.encode(para))
|
||||
if total + para_tokens > overlap_tokens:
|
||||
break
|
||||
tail.insert(0, para)
|
||||
total += para_tokens
|
||||
if not tail:
|
||||
# Aucun paragraphe entier ne tient dans le budget (paragraphes longs) :
|
||||
# on reprend la FIN du dernier paragraphe pour garantir le recouvrement.
|
||||
tokens = enc.encode(parts[-1])
|
||||
tail = [enc.decode(tokens[-overlap_tokens:])]
|
||||
total = min(overlap_tokens, len(tokens))
|
||||
return tail, total
|
||||
|
||||
|
||||
def _split_oversized(paragraph: str, enc, target_tokens: int, overlap_tokens: int = 0) -> list[str]:
|
||||
"""Coupe un paragraphe géant en sous-blocs ~`target_tokens` tokens (fenêtre
|
||||
glissante avec recouvrement si `overlap_tokens` > 0)."""
|
||||
tokens = enc.encode(paragraph)
|
||||
step = max(1, target_tokens - overlap_tokens)
|
||||
out: list[str] = []
|
||||
for i in range(0, len(tokens), target_tokens):
|
||||
i = 0
|
||||
while i < len(tokens):
|
||||
out.append(enc.decode(tokens[i : i + target_tokens]))
|
||||
if i + target_tokens >= len(tokens):
|
||||
break
|
||||
i += step
|
||||
return out
|
||||
|
||||
|
||||
|
||||
@@ -20,6 +20,9 @@ from app.infrastructure import vector_store
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_RAG_CHUNK_TOKENS = 600
|
||||
# Recouvrement entre extraits consécutifs (~13% de la cible) : une phrase-clé à
|
||||
# cheval sur deux extraits reste retrouvable dans au moins l'un des deux.
|
||||
_RAG_OVERLAP_TOKENS = 80
|
||||
# Un extrait avec quasi aucun texte réel (en-tête/pied de page, fragment de numéro
|
||||
# de page isolé « 249 250 ») ne sert à rien en RAG → on l'écarte. Seuil bas et
|
||||
# conservateur : on ne coupe QUE les fragments quasi-vides, jamais une vraie phrase.
|
||||
@@ -36,10 +39,14 @@ class NotebookRagUseCase:
|
||||
extractor: PdfTextExtractor,
|
||||
embedder: EmbeddingProvider,
|
||||
chunk_target_tokens: int = _RAG_CHUNK_TOKENS,
|
||||
min_score: float = 0.0,
|
||||
) -> None:
|
||||
self._extractor = extractor
|
||||
self._embedder = embedder
|
||||
self._chunk_target_tokens = chunk_target_tokens
|
||||
# Cosinus minimal pour qu'un extrait soit injecté dans le prompt : sous ce
|
||||
# seuil, l'extrait n'a aucun rapport avec la question → bruit. 0 = désactivé.
|
||||
self._min_score = min_score
|
||||
|
||||
async def index_source(self, source_id: str, pdf_bytes: bytes) -> dict:
|
||||
"""Extrait, découpe PAR PAGE (pour garder le n° de page → citations), embed
|
||||
@@ -48,7 +55,9 @@ class NotebookRagUseCase:
|
||||
chunks: list[str] = []
|
||||
pages: list[int] = []
|
||||
for page in doc.pages:
|
||||
for piece in chunk_text(page.text, self._chunk_target_tokens):
|
||||
for piece in chunk_text(
|
||||
page.text, self._chunk_target_tokens, overlap_tokens=_RAG_OVERLAP_TOKENS
|
||||
):
|
||||
if not _has_enough_text(piece):
|
||||
continue # fragment quasi-vide (en-tête/pied/numéro) → ignoré
|
||||
chunks.append(piece)
|
||||
@@ -69,11 +78,17 @@ class NotebookRagUseCase:
|
||||
}
|
||||
|
||||
async def retrieve(self, source_ids: list[str], query: str, top_k: int = 6) -> list[dict]:
|
||||
"""Passages les plus pertinents (toutes sources) pour `query`."""
|
||||
"""Passages les plus pertinents (toutes sources) pour `query`.
|
||||
|
||||
Recherche hybride (cosinus + bonus lexical sur les mots de la question) ;
|
||||
peut renvoyer moins de `top_k` passages si le seuil de pertinence écarte
|
||||
les extraits hors-sujet."""
|
||||
ids = [s for s in source_ids if vector_store.exists(s)]
|
||||
if not ids or not query.strip():
|
||||
return []
|
||||
query_vectors = await self._embedder.embed([query])
|
||||
if not query_vectors:
|
||||
return []
|
||||
return vector_store.search(ids, query_vectors[0], top_k)
|
||||
return vector_store.search(
|
||||
ids, query_vectors[0], top_k, query_text=query, min_score=self._min_score
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user