File size: 7,324 Bytes
d5067c9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 | """Apprentissage depuis documents — cycle PDF/URL → chunk → KB → retrieval + citations.
Réfute les gaps audit C7/C8/H10 :
* C7 : parse_pdf existait mais n'écrivait PAS dans la KB → aucun apprentissage.
* C8 : URLMemory stockait le texte brut, pas de retrieval sémantique exploitable.
* H10 : KnowledgeBase retrieval cosinus sur mots-id, pas de chunking texte + citations.
Ici on câble les briques existantes (parse_pdf, fetch_url, chunk_document,
KnowledgeBase, LearnedVocab) en un VRAI cycle d'apprentissage :
document (PDF/URL/texte) → chunks → embedding (hash stable) → KB.store
→ retrieve(query) retourne le chunk + sa source (citation) + confiance.
L'embedding texte = hash de n-grammes stable (pas un BERT — honnête, cf audit H17
qui demande des embeddings sémantiques réels ; ici on fournit le PIPELINE complet,
l'embedding est remplaçable par un vrai encodeur sans changer l'API).
Abstention conservée : si aucun chunk n'atteint le seuil → "je ne sais pas" (épistémique).
"""
from __future__ import annotations
import hashlib
from typing import List, Optional, Tuple
import torch
from .knowledge_base import chunk_document
from .learned_vocab import LearnedVocab
def text_embedding(text: str, dim: int = 64) -> torch.Tensor:
"""Embedding texte stable par hash de n-grammes de caractères (dim-d).
Honnête : hash déterministe (pas sémantique) — remplaçable par un vrai encodeur.
Suffisant pour retrieval exact/near-exact sur chunks appris."""
v = torch.zeros(dim)
s = text.lower().strip()
if not s:
return v
# n-grammes de caractères (3,4,5) + mots
for n in (3, 4, 5):
for i in range(max(0, len(s) - n + 1)):
gram = s[i:i + n]
h = int(hashlib.md5(gram.encode()).hexdigest(), 16)
v[h % dim] += 1.0
for w in s.split():
h = int(hashlib.md5(w.encode()).hexdigest(), 16)
v[h % dim] += 2.0 # mots pèsent +
nrm = v.norm() + 1e-8
return v / nrm
class DocumentLearner:
"""Cycle complet : apprend des documents (PDF/URL/texte), retrieve + citations.
Memoire = liste de (chunk_text, source, embedding). Retrieval = plus proche
voisin cosinus. Abstention si confiance < seuil."""
def __init__(self, threshold: float = 0.55, chunk_size: int = 200, overlap: int = 40,
margin: float = 0.08):
"""threshold (absolu) + margin (cos1−cos2) : double critère d'abstention.
HONNÊTE : le hash n-gramme est peu discriminant sémantiquement (audit H17) ;
la marge+seuil filtre les OOD grossiers mais un vrai encodeur sémantique
(mini-BERT/word2vec) reste nécessaire pour un retrieval de qualité. Le
PIPELINE (learn→retrieve→citation→abstention) est complet et l'embedding
est remplaçable sans changer l'API."""
self.threshold = threshold
self.margin = margin
self.chunk_size = chunk_size
self.overlap = overlap
self.chunks: List[str] = []
self.sources: List[str] = []
self.embeddings: List[torch.Tensor] = []
# ---- apprentissage (écrit dans la mémoire) ----
def learn_text(self, text: str, source: str = "text") -> int:
"""Découpe un texte en chunks et les apprend. Retourne le nb de chunks appris."""
if not text or text.startswith("[") and "error" in text.lower():
return 0
n_before = len(self.chunks)
for ch in chunk_document(text, self.chunk_size, self.overlap):
self.chunks.append(ch)
self.sources.append(source)
self.embeddings.append(text_embedding(ch))
return len(self.chunks) - n_before
def learn_pdf(self, path: str, max_pages: int = 10) -> int:
"""C7 : apprend depuis un PDF réel (PyMuPDF) → chunks → mémoire."""
from .web_tools import parse_pdf
text = parse_pdf(path, max_pages=max_pages)
if not text or text.startswith("["):
return 0
return self.learn_text(text, source=f"pdf:{path}")
def learn_url(self, url: str) -> int:
"""C8 : apprend depuis une URL réelle (HTTP fetch) → chunks → mémoire."""
from .web_tools import fetch_url
try:
text = fetch_url(url)
except Exception:
return 0
return self.learn_text(text, source=f"url:{url}")
# ---- retrieval (avec abstention + citation) ----
@torch.no_grad()
def retrieve(self, query: str, top_k: int = 1
) -> List[Tuple[str, str, float]]:
"""Retourne les top_k chunks pertinents (chunk, source, confiance).
Double critère d'abstention : cos1 ≥ threshold ET (cos1−cos2) ≥ margin.
→ robuste aux faux-positifs du hash embedding (OOD rejeté)."""
if not self.chunks:
return []
q = text_embedding(query)
sims = [(float((q @ e).item()), i) for i, e in enumerate(self.embeddings)]
sims.sort(reverse=True)
out = []
for rank, (sim, i) in enumerate(sims):
if rank >= top_k:
break
# marge vs 2e meilleur (cos1−cos2) : distingue match net de match ambigu
second = sims[rank + 1][0] if rank + 1 < len(sims) else 0.0
if sim >= self.threshold and (sim - second) >= self.margin:
out.append((self.chunks[i], self.sources[i], sim))
return out
def answer(self, query: str) -> Tuple[Optional[str], Optional[str], float]:
"""RAG : retourne (meilleur_chunk, source_citation, confiance) ou abstention."""
res = self.retrieve(query, top_k=1)
if not res:
return None, None, 0.0 # abstention : "je ne sais pas"
chunk, source, conf = res[0]
return chunk, source, conf
def knows(self, query: str) -> bool:
return len(self.retrieve(query)) > 0
def size(self) -> int:
return len(self.chunks)
# ---------------- démo / self-test ----------------
def _demo() -> dict:
dl = DocumentLearner()
n1 = dl.learn_text(
"La photosynthèse convertit l'énergie lumineuse en énergie chimique. "
"Les plantes utilisent le dioxyde de carbone et l'eau pour produire du glucose. "
"L'oxygène est libéré comme sous-produit de la photosynthèse.",
source="bio:textbook")
n2 = dl.learn_text(
"La deuxième loi de Newton stipule que la force égale la masse fois l'accélération. "
"F=ma est l'équation fondamentale de la mécanique classique.",
source="phys:textbook")
a1 = dl.answer("Que produit la photosynthèse ?")
a2 = dl.answer("Quelle est la deuxième loi de Newton ?")
a3 = dl.answer("Quelle est la capitale du Brésil ?") # OOD → abstention
return {
"chunks_appris": dl.size(),
"photosynthese": {"source": a1[1], "confiance": round(a1[2], 3),
"contient_glucose": "glucose" in (a1[0] or "")},
"newton": {"source": a2[1], "confiance": round(a2[2], 3),
"contient_Fma": "F=ma" in (a2[0] or "")},
"ood_bresil": {"abstention": a3[0] is None, "confiance": round(a3[2], 3)},
}
if __name__ == "__main__":
import json
print(json.dumps(_demo(), indent=2, ensure_ascii=False))
|