IETM_FIXE\ietm6
e26d11a99f
A la premiere analyse d une source, chaque lot est resume (1 appel LLM, cache disque, purge avec la source) et son resume embedde. Aux questions suivantes, la question est comparee aux resumes et seuls les lots proches du meilleur score (marge 0.10, plancher 3 lots) sont relus -> 3-5x moins d appels sur un gros livre pour les questions ciblees. Selection volontairement conservatrice ; best-effort (tout echec -> plein scan) ; desactivable via DEEP_SUMMARY_FILTER=false (exhaustivite maximale). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
286 lines
13 KiB
Python
286 lines
13 KiB
Python
"""Use case « Analyse approfondie » d'un notebook : map-reduce sur TOUT le document.
|
|
|
|
Contrairement au chat RAG (qui ne ramène que les top-k extraits), ce mode lit
|
|
l'INTÉGRALITÉ des sources par lots :
|
|
- MAP : pour chaque lot, le modèle extrait ce qui est pertinent pour la question
|
|
(ou « RAS » si rien) ;
|
|
- REDUCE : il synthétise toutes les notes en une réponse finale (streamée).
|
|
|
|
→ Répond aux questions globales/exhaustives (« liste tous les… ») quel que soit le
|
|
modèle, au prix de plusieurs appels (comme l'import). Le lot est dimensionné par
|
|
`batch_tokens` (= taille de morceau d'import) : avec un modèle gros-contexte, peu de
|
|
lots ; avec un petit modèle local, plus de lots (mais ça reste exhaustif).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import AsyncIterator
|
|
|
|
import tiktoken
|
|
|
|
from app.application.llm_retry import generate_with_retry
|
|
from app.application.query_rewrite import standalone_question
|
|
from app.domain.models import ChatMessage
|
|
from app.domain.ports import LLMChatProvider, LLMProvider, LLMProviderError
|
|
from app.infrastructure import vector_store
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_NO_MATCH = "RAS"
|
|
_MAP_TEMPERATURE = 0.2
|
|
|
|
# --- Index de résumés (pré-filtrage des lots) --------------------------------
|
|
# Sans index : CHAQUE question relit TOUT le document (1 appel LLM par lot).
|
|
# Avec : les résumés de lots (construits UNE fois, cache disque) sont comparés
|
|
# à la question par embedding, et seuls les lots plausiblement pertinents sont
|
|
# relus. Sélection volontairement CONSERVATRICE (on préfère relire un lot de
|
|
# trop que rater une mention) ; désactivable via deep_summary_filter=False.
|
|
_SUMMARY_PROMPT = """Résume l'EXTRAIT ci-dessous en 4 à 8 puces factuelles : lieux, PNJ et
|
|
créatures nommés, objets notables, évènements, règles particulières. Pas d'analyse, pas
|
|
d'introduction — uniquement les puces, pour servir d'index de recherche.
|
|
|
|
--- EXTRAIT ---
|
|
{excerpt}
|
|
--- FIN EXTRAIT ---
|
|
|
|
Résumé :"""
|
|
|
|
# Un lot est gardé si son score est proche du meilleur (marge) OU bon dans
|
|
# l'absolu ; et on garde toujours au moins _MIN_KEPT lots.
|
|
_SELECT_MARGIN = 0.10
|
|
_SELECT_FLOOR = 0.5
|
|
_MIN_KEPT = 3
|
|
|
|
_MAP_PROMPT = """Voici un EXTRAIT d'un document. Extrais UNIQUEMENT les informations
|
|
pertinentes pour répondre à la question ci-dessous. Conserve les détails utiles et
|
|
indique les numéros de page (format « p. X »). Si l'extrait ne contient RIEN de
|
|
pertinent, réponds EXACTEMENT « {no_match} » et rien d'autre.
|
|
|
|
QUESTION : {question}
|
|
|
|
--- EXTRAIT ---
|
|
{excerpt}
|
|
--- FIN EXTRAIT ---
|
|
|
|
Informations pertinentes (ou « {no_match} ») :"""
|
|
|
|
_REDUCE_SYSTEM = """Tu es l'assistant-MJ d'un jeu de rôle. Tu réponds à la demande du MJ en
|
|
t'appuyant sur TROIS sources : (1) des NOTES extraites de l'ENSEMBLE du document source (vue
|
|
complète — mais POSSIBLEMENT VIDE si rien d'utile n'y figure), (2) le contexte de sa CAMPAGNE,
|
|
(3) la conversation ci-dessous.
|
|
|
|
- Si les notes contiennent des éléments utiles : exploite-les et CITE les pages (« p. X »).
|
|
- Si les notes sont VIDES ou pauvres (cas fréquent d'une demande CRÉATIVE portant sur des
|
|
éléments INVENTÉS par le MJ) : ne te bloque surtout PAS. Aide-le quand même en t'appuyant
|
|
sur sa CAMPAGNE, la CONVERSATION et ta connaissance du genre — propose des adaptations
|
|
concrètes (arcs, chapitres, scènes, PNJ), structurées et jouables.
|
|
- Sois concret et utile. N'affirme rien de FAUX sur le contenu du document.
|
|
|
|
{context_block}
|
|
--- NOTES EXTRAITES DE TOUT LE DOCUMENT ---
|
|
{notes_block}
|
|
--- FIN DES NOTES ---
|
|
|
|
Réponds en français."""
|
|
|
|
|
|
class NotebookDeepUseCase:
|
|
def __init__(
|
|
self,
|
|
llm: LLMProvider,
|
|
batch_tokens: int = 10000,
|
|
map_concurrency: int = 1,
|
|
embedder=None,
|
|
summary_filter: bool = True,
|
|
) -> None:
|
|
self._llm = llm
|
|
self._batch_tokens = max(2000, batch_tokens)
|
|
# Lots MAP traités par vagues de cette taille (parallélisme LLM).
|
|
self._map_concurrency = max(1, map_concurrency)
|
|
# EmbeddingProvider (duck typing) pour l'index de résumés ; None = pas
|
|
# de pré-filtrage (plein scan, comportement historique).
|
|
self._embedder = embedder
|
|
self._summary_filter = summary_filter
|
|
|
|
async def stream(
|
|
self,
|
|
source_ids: list[str],
|
|
messages: list[ChatMessage],
|
|
context: str = "",
|
|
history_limit: int = 8,
|
|
) -> AsyncIterator[dict]:
|
|
"""Yield des évènements : {type:'progress',current,total}, {type:'token',token},
|
|
{type:'done'}. (Les erreurs LLM des lots sont tolérées : lot ignoré.)
|
|
|
|
La dernière question utilisateur sert à la LECTURE du document (map) ; la
|
|
SYNTHÈSE (reduce) reçoit les `history_limit` derniers messages → les relances
|
|
conversationnelles (« et pour les autres ? ») fonctionnent aussi en approfondi.
|
|
"""
|
|
# Question autonome : la phase MAP lit chaque lot avec LA question — sur
|
|
# une relance conversationnelle, il faut y résoudre les références
|
|
# implicites, sinon les lots sont filtrés sur un texte sans sujet.
|
|
question = await standalone_question(self._llm, messages)
|
|
# Lots PAR SOURCE (l'index de résumés est caché par source).
|
|
per_source: list[tuple[str, list[dict]]] = []
|
|
for sid in source_ids:
|
|
chunks = vector_store.all_chunks(sid)
|
|
for batch in self._group(chunks):
|
|
per_source.append((sid, batch))
|
|
if not per_source:
|
|
yield {"type": "token", "token": "Aucune source indexée à analyser."}
|
|
yield {"type": "done"}
|
|
return
|
|
|
|
# Pré-filtrage par index de résumés (best-effort : tout échec → plein scan).
|
|
selected: set[int] | None = None
|
|
if self._summary_filter and self._embedder is not None:
|
|
try:
|
|
async for ev_or_result in self._select_batches(per_source, question):
|
|
if isinstance(ev_or_result, dict):
|
|
yield ev_or_result # progress de construction de l'index
|
|
else:
|
|
selected = ev_or_result
|
|
except Exception as exc: # noqa: BLE001 — le filtre ne doit jamais bloquer
|
|
logger.warning("Index de résumés ignoré (échec) : %s", exc)
|
|
selected = None
|
|
if selected is not None:
|
|
logger.info(
|
|
"Analyse approfondie : %s/%s lot(s) retenus via l'index de résumés.",
|
|
len(selected), len(per_source))
|
|
|
|
indices = sorted(selected) if selected is not None else list(range(len(per_source)))
|
|
total = len(indices)
|
|
notes: list[str] = []
|
|
# Lots traités par VAGUES parallèles ; les notes restent dans l'ordre du
|
|
# document (gather préserve l'ordre des tâches de la vague).
|
|
for start in range(0, total, self._map_concurrency):
|
|
yield {"type": "progress", "current": start, "total": total}
|
|
wave = indices[start:start + self._map_concurrency]
|
|
results = await asyncio.gather(
|
|
*(self._map_batch(question, per_source[i][1]) for i in wave),
|
|
return_exceptions=True)
|
|
for j, res in enumerate(results):
|
|
if isinstance(res, LLMProviderError):
|
|
logger.warning(
|
|
"Analyse approfondie : lot %s/%s ignoré : %s", start + j + 1, total, res)
|
|
elif isinstance(res, BaseException):
|
|
raise res # bug inattendu : ne pas l'avaler
|
|
elif res:
|
|
notes.append(res)
|
|
yield {"type": "progress", "current": total, "total": total}
|
|
|
|
notes_block = "\n\n".join(notes) if notes else "(aucune information pertinente trouvée dans le document)"
|
|
context_block = (
|
|
f"--- TA CAMPAGNE (structure, PNJ, univers) ---\n{context.strip()}\n--- FIN CAMPAGNE ---\n\n"
|
|
if context.strip() else ""
|
|
)
|
|
system_prompt = _REDUCE_SYSTEM.format(context_block=context_block, notes_block=notes_block)
|
|
# Historique récent pour la cohérence des relances ; on garantit que le
|
|
# dernier message est bien la question courante.
|
|
reduce_messages = messages[-history_limit:] if messages else [ChatMessage(role="user", content=question)]
|
|
llm_chat: LLMChatProvider = self._llm # type: ignore[assignment]
|
|
produced = False
|
|
async for token in llm_chat.stream_chat(reduce_messages, system_prompt=system_prompt):
|
|
if token:
|
|
produced = True
|
|
yield {"type": "token", "token": token}
|
|
if not produced:
|
|
# Jamais de bulle vide : message de repli + orientation vers le mode rapide,
|
|
# mieux adapté aux demandes créatives (et qui propose des cartes d'action).
|
|
yield {"type": "token", "token": (
|
|
"Je n'ai pas trouvé d'éléments pertinents dans le document pour cette demande "
|
|
"(elle porte sans doute sur des éléments que tu as inventés). Pour une "
|
|
"**adaptation créative** — proposer des arcs, chapitres, scènes ou PNJ — "
|
|
"utilise plutôt le bouton **« Envoyer »** (mode rapide) : il est conversationnel, "
|
|
"voit ta campagne, et te propose des cartes « Créer dans la campagne »."
|
|
)}
|
|
yield {"type": "done"}
|
|
|
|
# --- Index de résumés ------------------------------------------------------
|
|
|
|
async def _select_batches(self, per_source: list[tuple[str, list[dict]]], question: str):
|
|
"""Générateur : yield des évènements `progress` pendant la construction de
|
|
l'index (1ère analyse d'une source), puis le set des indices retenus —
|
|
ou None si le filtre n'apporte rien (tous retenus)."""
|
|
# 1. Charge/construit les résumés par source (cache disque).
|
|
by_sid: dict[str, list[int]] = {}
|
|
for i, (sid, _) in enumerate(per_source):
|
|
by_sid.setdefault(sid, []).append(i)
|
|
vectors: list[list[float] | None] = [None] * len(per_source)
|
|
|
|
to_build = []
|
|
for sid, idxs in by_sid.items():
|
|
cached = vector_store.load_summaries(sid, self._batch_tokens)
|
|
if cached is not None and len(cached) == len(idxs):
|
|
for i, entry in zip(idxs, cached):
|
|
vectors[i] = entry.get("vector")
|
|
else:
|
|
to_build.append((sid, idxs))
|
|
|
|
total_build = sum(len(idxs) for _, idxs in to_build)
|
|
done_build = 0
|
|
for sid, idxs in to_build:
|
|
summaries: list[str] = []
|
|
for start in range(0, len(idxs), self._map_concurrency):
|
|
yield {"type": "progress", "current": done_build, "total": total_build}
|
|
wave = idxs[start:start + self._map_concurrency]
|
|
results = await asyncio.gather(
|
|
*(self._summarize_batch(per_source[i][1]) for i in wave))
|
|
summaries.extend(results)
|
|
done_build += len(wave)
|
|
vecs = await self._embedder.embed(summaries, kind="document")
|
|
entries = [{"summary": s, "vector": v} for s, v in zip(summaries, vecs)]
|
|
vector_store.save_summaries(sid, self._batch_tokens, entries)
|
|
for i, entry in zip(idxs, entries):
|
|
vectors[i] = entry["vector"]
|
|
|
|
# 2. Score de chaque lot face à la question, sélection conservatrice.
|
|
qv = (await self._embedder.embed([question], kind="query"))[0]
|
|
scores = [
|
|
vector_store.cosine_similarity(qv, v) if v else 0.0
|
|
for v in vectors
|
|
]
|
|
best = max(scores)
|
|
keep = {i for i, s in enumerate(scores) if s >= best - _SELECT_MARGIN or s >= _SELECT_FLOOR}
|
|
floor = min(_MIN_KEPT, len(scores))
|
|
if len(keep) < floor:
|
|
keep = set(sorted(range(len(scores)), key=lambda i: -scores[i])[:floor])
|
|
yield keep if len(keep) < len(scores) else None
|
|
|
|
async def _summarize_batch(self, batch: list[dict]) -> str:
|
|
excerpt = "\n\n".join(c.get("text", "").strip() for c in batch)
|
|
raw = await generate_with_retry(
|
|
self._llm, _SUMMARY_PROMPT.format(excerpt=excerpt), temperature=_MAP_TEMPERATURE)
|
|
return (raw or "").strip()
|
|
|
|
async def _map_batch(self, question: str, batch: list[dict]) -> str:
|
|
"""Phase MAP d'un lot : extrait les infos pertinentes ('' si RAS)."""
|
|
excerpt = "\n\n".join(
|
|
f"(p. {c['page']}) {c['text'].strip()}" if c.get("page") else c["text"].strip()
|
|
for c in batch
|
|
)
|
|
prompt = _MAP_PROMPT.format(no_match=_NO_MATCH, question=question, excerpt=excerpt)
|
|
raw = await generate_with_retry(self._llm, prompt, temperature=_MAP_TEMPERATURE)
|
|
answer = raw.strip()
|
|
if answer and answer.upper().rstrip(".") != _NO_MATCH:
|
|
return answer
|
|
return ""
|
|
|
|
def _group(self, chunks: list[dict]) -> list[list[dict]]:
|
|
"""Regroupe les extraits en lots ~`batch_tokens` (compte tiktoken)."""
|
|
enc = tiktoken.get_encoding("cl100k_base")
|
|
batches: list[list[dict]] = []
|
|
current: list[dict] = []
|
|
current_tokens = 0
|
|
for c in chunks:
|
|
t = len(enc.encode(c.get("text", "")))
|
|
if current and current_tokens + t > self._batch_tokens:
|
|
batches.append(current)
|
|
current, current_tokens = [], 0
|
|
current.append(c)
|
|
current_tokens += t
|
|
if current:
|
|
batches.append(current)
|
|
return batches
|