14. Memoire persistante pour le test-time scaling

Navigation : Index | << Précédent | Suivant >>
15. Tree of Thoughts >>

Phase 2 de l’epic #2926 — pont entre NB-12 (moteur Reflexion, memoire in-memory), NB-13 (routeur agentique) et NB-05 (RAG : embeddings + vector store).

Le problème de la memoire volatile

Dans NB-12, le moteur Reflexion accumule des lecons (quels tests echouent, quel diagnostic) dans une liste Python. Cette memoire est perdue a la fin de l’exécution : a chaque nouveau problème, le moteur repart de zero, même si un problème similaire a déjà ete resolu (et diagnostique) precedemment.

Le projet de reference ICR (Iterative-Contextual-Refinements) dispose d’un agent “Memory” qui persiste les lecons d’une session a l’autre dans un vector store, et les retrocede par similarite quand un problème apparente se presente. C’est exactement ce que ce notebook implemente (Phase 2 de #2926).

Ide centrale : remplacer la liste Python volatile de NB-12 par une memoire vectorielle persistante. Avant de lancer Reflexion sur un nouveau problème, on retrocede les diagnostics des problemes passes les plus similaires (cosinus), et on les injecte dans le feedback du générateur. Les lecos survit aux runs.

Plan

  1. Rappel compact du moteur Reflexion (memoire in-memory, pont NB-12).
  2. Embeddings : baseline BoW transparente (numpy) + pont vers les embeddings OpenAI de NB-05.
  3. Memoire vectorielle persistante (stockage JSON cross-run + retrieval cosinus top-k).
  4. Reflexion avec memoire : injection des lecons retrocede.
  5. Demonstration : la memoire accelere un problème similaire (transfer de lecon).
  6. Pont production (NB-05 embeddings, NB-09 cout) + limites honnetes.

0. Setup — même infrastructure que NB-12 / NB-13

Client OpenRouter (claudish-proof) + chargeur .env robuste. La couche memoire (embeddings BoW + persistence JSON) est sans dépendance API ; seuls les appels du moteur Reflexion passent par le LLM.

%pip install -q openai python-dotenv numpy

import os, re, json
from pathlib import Path
from collections import Counter
from openai import OpenAI
from dotenv import load_dotenv

_env_path = None
_current = Path.cwd()
for _i in range(10):
    if (_current / ".env").exists():
        _env_path = _current / ".env"; break
    if _current.name in ("GenAI", "MyIA.AI.Notebooks"):
        break
    _current = _current.parent
if _env_path is None:
    for _cand in (Path.cwd() / "MyIA.AI.Notebooks" / "GenAI" / ".env",
                  Path.cwd() / "GenAI" / ".env"):
        if _cand.exists():
            _env_path = _cand; break
if _env_path is not None:
    load_dotenv(_env_path); print(f".env charge depuis : {_env_path.name}")
else:
    print("ATTENTION : aucun .env trouve.")

FAST_MODEL = os.getenv("OPENAI_MODEL_FAST", "meta-llama/llama-3.3-70b-instruct")
BIG_MODEL = os.getenv("OPENAI_MODEL_BIG", "openai/gpt-5-nano")
BATCH_MODE = os.getenv("BATCH_MODE", "true").lower() in ("1", "true", "yes")

client = OpenAI(
    api_key=os.getenv("OPENROUTER_API_KEY"),
    base_url=os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1"),
)
print(f"FAST_MODEL={FAST_MODEL} | BIG_MODEL={BIG_MODEL} | BATCH_MODE={BATCH_MODE}")
Note: you may need to restart the kernel to use updated packages.
.env charge depuis : .env
FAST_MODEL=meta-llama/llama-3.3-70b-instruct | BIG_MODEL=openai/gpt-5-nano | BATCH_MODE=False

Lecture du setup. Le message .env charge depuis : .env confirme que les variables d’environnement (dont FAST_MODEL=meta-llama/llama-3.3-70b-instruct et BIG_MODEL=openai/gpt-5-nano) sont disponibles pour le notebook, et la notice pip indique simplement qu’une mise à jour est possible mais non bloquante.

def chat(prompt, system=None, model=FAST_MODEL, temperature=0.0, max_tokens=2000):
    """Appel LLM unique. Renvoie le texte ou '' en cas d'erreur (BATCH_MODE safe)."""
    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})
    try:
        resp = client.chat.completions.create(
            model=model, messages=messages, temperature=temperature, max_tokens=max_tokens)
        return resp.choices[0].message.content or ""
    except Exception as exc:
        print(f"  [chat] erreur ({model}) : {exc}")
        return ""

_ping = chat("Reponds uniquement par : OK", max_tokens=10)
print("Ping :", repr(_ping[:40]) if _ping else "ECHEC")
Ping : 'OK'

1. Rappel compact du moteur Reflexion (memoire in-memory)

On redefinit les utilitaires de NB-12/NB-13 (generation + exécution de code) puis le moteur Reflexion dans sa forme sans memoire persistante : le feedback des tests echoues est stocke dans une variable locale feedback, perdue a la sortie de la fonction.

Confinement (pont 13_Agentic_Orchestration) : executer_tests ci-dessous exécute le code généré par le modèle dans un sous-processus confiné — builtins restreints, imports en liste blanche, borne de temps réelle. Le pourquoi (injection de prompt, OWASP LLM01) et la démonstration mesurée : NB-13.

def extraire_code(reponse):
    m = re.search(r"```(?:python)?\s*(.*?)```", reponse or "", re.DOTALL)
    return m.group(1).strip() if m else (reponse or "").strip()

import subprocess, sys, json

# --- Bac a sable : le code genere par le modele s'execute en SOUS-PROCESSUS confine ---
_NOMS_BUILTINS_SURS = (
    "abs", "all", "any", "bool", "chr", "dict", "divmod", "enumerate", "filter",
    "float", "format", "frozenset", "hasattr", "hash", "int", "isinstance", "len",
    "list", "map", "max", "min", "next", "oct", "ord", "pow", "print", "range",
    "repr", "reversed", "round", "set", "slice", "sorted", "str", "sum", "tuple", "zip",
    "Exception", "ValueError", "TypeError", "IndexError", "KeyError",
    "ZeroDivisionError", "ArithmeticError", "AttributeError", "RuntimeError",
    "StopIteration", "NotImplementedError", "OverflowError", "RecursionError",
    "AssertionError", "OSError",
)
_MODULES_SURS = ("math", "itertools", "functools", "collections", "string",
                 "heapq", "re", "statistics", "random", "datetime")


def _runner_source(code, tests):
    """Source du processus ENFANT. Le runner lui-meme tourne en confiance pleine ;
    seule la portion exec(code, ...) voit le dictionnaire de builtins restreint.
    Le resultat transite par une ligne __RESULT__ sur stdout (les print du code
    genere ne polluent pas le canal)."""
    return (
        "import json, builtins as _b, importlib\n"
        "_MODULES_SURS = " + repr(_MODULES_SURS) + "\n"
        "def _import_sur(nom, *a, **k):\n"
        "    if nom not in _MODULES_SURS:\n"
        "        raise ImportError('module non autorise dans le bac a sable : ' + str(nom))\n"
        "    return importlib.import_module(nom)\n"
        "_surs = {n: getattr(_b, n) for n in " + repr(_NOMS_BUILTINS_SURS) + "}\n"
        "_surs['__import__'] = _import_sur\n"
        "ns = {'__builtins__': _surs}\n"
        "res = {'passes': 0, 'fails': [], 'erreur': None}\n"
        "try:\n"
        "    exec(" + repr(code) + ", ns)\n"
        "except BaseException as e:\n"
        "    res['erreur'] = type(e).__name__ + ': ' + str(e)\n"
        "if res['erreur'] is None:\n"
        "    for _t in " + repr(list(tests)) + ":\n"
        "        try:\n"
        "            if eval(_t, ns):\n"
        "                res['passes'] += 1\n"
        "            else:\n"
        "                res['fails'].append(_t)\n"
        "        except BaseException:\n"
        "            res['fails'].append(_t)\n"
        "print('__RESULT__' + json.dumps(res))\n"
    )


def executer_tests_detaille(code, probleme, timeout=5):
    """Execute `code` genere par le modele dans un sous-processus confine.

    Ce qui est reellement garanti (et demontre sur sortie committee plus bas) :
    - builtins restreints : ni open, ni eval, ni exec, ni __import__ libre ;
    - imports limites a une liste blanche de modules de calcul ;
    - borne de temps REELLE : subprocess.run(timeout=...) tue le processus enfant,
      y compris sous Windows (ou signal.alarm n'existe pas).

    Ce qui n'est PAS garanti : ce n'est pas une frontiere de securite dure — c'est une
    ceinture de securite pedagogique pour du code de benchmark ; du code reellement
    non fiable demande un conteneur jetable ou un interpreteur WASM.

    Renvoie (passes, total, fails, erreur)."""
    tests = probleme["tests"]
    total = len(tests)
    if not code.strip():
        return 0, total, [], "code vide"
    try:
        proc = subprocess.run(
            [sys.executable, "-c", _runner_source(code, tests)],
            capture_output=True, text=True, timeout=timeout,
            stdin=subprocess.DEVNULL)
    except subprocess.TimeoutExpired:
        return 0, total, list(tests), f"timeout {timeout}s : execution interrompue de force"
    for ligne in proc.stdout.splitlines():
        if ligne.startswith("__RESULT__"):
            res = json.loads(ligne[len("__RESULT__"):])
            if res["erreur"] is not None:
                return 0, total, list(tests), "echec de chargement intercepte : " + res["erreur"]
            return res["passes"], total, res["fails"], None
    return 0, total, list(tests), "pas de resultat (sortie anormale du bac a sable)"


def executer_tests(code, probleme, timeout=5):
    """(API inchangee) Renvoie (nb_tests_passes, nb_tests_total, tests_echoues).
    Confinement effectif : voir executer_tests_detaille (NB-13 pour la demonstration).
    Le parametre timeout est APPLIQUE (subprocess.run(timeout=...))."""
    passes, total, fails, err = executer_tests_detaille(code, probleme, timeout=timeout)
    if err:
        return passes, total, [err]
    return passes, total, fails


def _gen_code(spec, model=FAST_MODEL, temperature=0.0, max_tokens=2000):
    resp = chat(spec + "\n\nReponds UNIQUEMENT avec le code Python dans un bloc ```python```.",
                model=model, temperature=temperature, max_tokens=max_tokens)
    return extraire_code(resp)

# Banque : deux problemes partageant un pitfall (logique modulo concatenative).
PROBLEMES = {
    "fizzbuzz": {
        "spec": "Ecris `fizzbuzz(n)` renvoyant une liste de longueur n (1-indexee) ou l'element i "
                "vaut 'Fizz' si i multiple de 3, 'Buzz' si multiple de 5, 'FizzBuzz' si multiple "
                "des deux, sinon i.",
        "tests": ["fizzbuzz(5)==[1,2,'Fizz',4,'Buzz']", "fizzbuzz(15)[-1]=='FizzBuzz'"]},
    "fizzbuzz_etendu": {
        "spec": "Ecris `fizzbuzz_etendu(n)` renvoyant une liste de longueur n (1-indexee) ou "
                "l'element i vaut 'Fizz' si i multiple de 3, 'Buzz' si 5, 'Jazz' si 7, en "
                "CONCATENANT les regles qui s'appliquent (ex: 105 -> 'FizzBuzzJazz'), sinon i.",
        "tests": ["fizzbuzz_etendu(3)==[1,2,'Fizz']", "fizzbuzz_etendu(35)[-1]=='BuzzJazz'",
                  "fizzbuzz_etendu(21)[-1]=='FizzJazz'"]},
}

def reflexion_sans_memoire(probleme, model=FAST_MODEL, iterations=3):
    """Reflexion NB-12 : boucle generation -> tests -> critique -> regeneration.
    Le feedback est IN-MEMORY (variable locale), PERDU a la sortie. Renvoie (passes, total, lecon)."""
    total = len(probleme["tests"])
    meilleur, feedback, code, lecon = 0, "", "", ""
    for it in range(iterations):
        if feedback:
            prompt = (probleme["spec"] + "\n\nCode precedent ECHEC :\n```python\n" + code +
                      "\n```\nTests echouant :\n" + feedback +
                      "\nCorrige. Reponds UNIQUEMENT avec le code dans ```python```.")
        else:
            prompt = (probleme["spec"] +
                      "\n\nReponds UNIQUEMENT avec le code dans ```python```.")
        code = _gen_code(prompt, model=model)
        passes, _, fails = executer_tests(code, probleme)
        meilleur = max(meilleur, passes)
        if passes >= total:
            lecon = f"Resolu en {it+1} iteration(s)."
            break
        feedback = "\n".join(fails)
        # Synthese d'un diagnostic court (la "lecon") -- c'est ce qu'on va memoriser
        lecon = feedback[:200]
    return meilleur, total, lecon

print("Moteur Reflexion (memoire in-memory) recharge depuis NB-12/NB-13.")
Moteur Reflexion (memoire in-memory) recharge depuis NB-12/NB-13.

2. Embeddings — baseline BoW transparente + pont NB-05

NB-05 (RAG) vectorise le texte via client.embeddings.create(model="text-embedding-3-large") de l’API OpenAI. Ici, pour garder le notebook executable sans dependence API sur la couche memoire (claudish-proof, CPU-safe) et rendre le mécanisme de retrieval transparent (boite blanche), on implemente une baseline Bag-of-Words + cosine en numpy.

Pont production (NB-05) : l’interface embedder.embed(text) -> vecteur est identique. En production, remplacez BoWEmbedder() par un embedder OpenAI (text-embedding-3-large) ou un modèle local (sentence-transformers). Voir Exercice 1. Le reste du notebook (MemoireVectorielle, reflexion_avec_memoire) fonctionne a l’identique.

import numpy as np

class BoWEmbedder:
    """Embedder Bag-of-Words transparent (boite blanche, numpy, sans API).
    Vocabulaire construit sur .fit() ; embed() produit un vecteur TF normalise.
    En production : remplacer par OpenAIEmbedder (cf NB-05 + Exercice 1)."""

    def __init__(self):
        self.vocab = {}

    def _tokens(self, text):
        # Tokenisation simple : minuscules, mots alphanumeriques >= 3 chars.
        return [w for w in re.findall(r"[a-z0-9_]{3,}", (text or "").lower())]

    def fit(self, textes):
        idx = 0
        for t in textes:
            for tok in self._tokens(t):
                if tok not in self.vocab:
                    self.vocab[tok] = idx; idx += 1
        print(f"BoWEmbedder fitte : vocabulaire de {len(self.vocab)} tokens.")
        return self

    def embed(self, text):
        v = np.zeros(len(self.vocab))
        for tok in self._tokens(text):
            if tok in self.vocab:
                v[self.vocab[tok]] += 1
        n = np.linalg.norm(v)
        return v / n if n > 0 else v

    @staticmethod
    def cosine(a, b):
        na, nb = np.linalg.norm(a), np.linalg.norm(b)
        if na == 0 or nb == 0:
            return 0.0
        return float(np.dot(a, b) / (na * nb))

# Smoke test : deux specs similaires doivent avoir un cosine eleve.
emb = BoWEmbedder().fit([PROBLEMES["fizzbuzz"]["spec"], PROBLEMES["fizzbuzz_etendu"]["spec"]])
_sim = BoWEmbedder.cosine(emb.embed(PROBLEMES["fizzbuzz"]["spec"]),
                          emb.embed(PROBLEMES["fizzbuzz_etendu"]["spec"]))
print(f"Similarite cosine(fizzbuzz, fizzbuzz_etendu) = {_sim:.3f}  (eleve attendu : pitfall partage)")
BoWEmbedder fitte : vocabulaire de 24 tokens.
Similarite cosine(fizzbuzz, fizzbuzz_etendu) = 0.599  (eleve attendu : pitfall partage)

3. Memoire vectorielle persistante

La memoire stocke des lecons : un diagnostic court associe a un problème resolu/en cours, vectorise par l’embedder. Elle persiste sur disque (JSON) pour survivre aux runs, et retrocede les k lecons les plus similaires (cosinus) a un nouveau problème.

class MemoireVectorielle:
    """Vector store de lecons, persistant (JSON). Pont NB-05 (VectorStore sklearn) -- ici
    on reste numpy pur pour la transparence. Chaque entree : {id, spec, lecon, vecteur}."""

    def __init__(self, embedder, chemin="tmp/memoire_reflexion.json"):
        self.embedder = embedder
        self.chemin = chemin
        self.entrees = []  # liste de dicts {id, spec, lecon, vecteur(list)}

    def _prochain_id(self):
        return max([e["id"] for e in self.entrees], default=-1) + 1

    def ajouter(self, spec, lecon):
        """Memorise une lecon (diagnostic) associee a la spec d'un probleme."""
        if not lecon or not lecon.strip():
            return None
        entree = {"id": self._prochain_id(), "spec": spec,
                  "lecon": lecon, "vecteur": self.embedder.embed(spec).tolist()}
        self.entrees.append(entree)
        return entree["id"]

    def retroceder(self, spec_query, k=3, seuil=0.05):
        """Renvoie les k lecons les plus similaires (cosinus >= seuil) a la requete."""
        if not self.entrees:
            return []
        q = self.embedder.embed(spec_query)
        scorees = []
        for e in self.entrees:
            s = BoWEmbedder.cosine(q, np.array(e["vecteur"]))
            if s >= seuil:
                scorees.append((s, e))
        scorees.sort(key=lambda x: x[0], reverse=True)
        return [{"score": round(s, 3), "id": e["id"], "lecon": e["lecon"],
                 "spec": e["spec"][:80]} for s, e in scorees[:k]]

    def sauvegarder(self):
        Path(self.chemin).parent.mkdir(parents=True, exist_ok=True)
        with open(self.chemin, "w", encoding="utf-8") as f:
            json.dump(self.entrees, f, ensure_ascii=False, indent=1)
        return len(self.entrees)

    def charger(self):
        if Path(self.chemin).exists():
            with open(self.chemin, encoding="utf-8") as f:
                self.entrees = json.load(f)
        return len(self.entrees)

print("MemoireVectorielle definie (ajouter / retroceder / sauvegarder / charger).")
MemoireVectorielle definie (ajouter / retroceder / sauvegarder / charger).

4. Reflexion AVEC memoire — injection des lecons retrocede

On enrichit le moteur Reflexion : avant la 1ere generation, on retrocede les lecons des problemes passes les plus similaires, et on les injecte dans le prompt du générateur. C’est l’agent “Memory” d’ICR. Après resolution (ou epuisement des itérations), on memorise la nouvelle lecon (diagnostic final) pour les runs futurs.

def reflexion_avec_memoire(probleme, memoire, model=FAST_MODEL, iterations=3, k=3, verbose=False):
    """Reflexion enrichi d'une memoire vectorielle persistante.
    1. Retrocede k lecons similaires -> injectees dans le 1er prompt.
    2. Boucle generation -> tests -> critique (feedback in-memory local).
    3. Memorise le diagnostic final dans la memoire (persistant)."""
    total = len(probleme["tests"])
    lecons_passees = memoire.retroceder(probleme["spec"], k=k)
    if verbose and lecons_passees:
        print(f"  [memoire] {len(lecons_passees)} lecon(s) retrocede(s) : "
              + " | ".join(f"#{l['id']}(score={l['score']})" for l in lecons_passees))
    contexte = ""
    if lecons_passees:
        contexte = ("\n\nLecons de problemes SIMILAIRES deja rencontres (applique-les si pertinent) :\n"
                    + "\n".join(f"- {l['lecon']}" for l in lecons_passees))

    meilleur, feedback, code, diagnostic = 0, "", "", ""
    for it in range(iterations):
        prompt = probleme["spec"] + contexte
        if feedback:
            prompt += ("\n\nCode precedent ECHEC :\n```python\n" + code +
                       "\n```\nTests echouant :\n" + feedback)
        prompt += "\n\nReponds UNIQUEMENT avec le code dans ```python```."
        code = _gen_code(prompt, model=model)
        passes, _, fails = executer_tests(code, probleme)
        meilleur = max(meilleur, passes)
        if passes >= total:
            diagnostic = f"Resolu en {it+1} iteration(s). Pitfall maitrise."
            break
        feedback = "\n".join(fails)
        diagnostic = (f"Echec partiel ({meilleur}/{total}). Tests echouant : "
                      + "; ".join(t[:60] for t in fails[:3]))
    # Memorise la lecon pour les runs futurs (persistant).
    memoire.ajouter(probleme["spec"], f"[{probleme.get('id','?')}] {diagnostic}")
    return meilleur, total, diagnostic, lecons_passees

print("reflexion_avec_memoire definie (pont ICR 'Memory' agent).")
reflexion_avec_memoire definie (pont ICR 'Memory' agent).

5. Demonstration — le mécanisme du transfer de lecon

On resout d’abord fizzbuzz a froid (memoire vide), on persiste sa lecon, puis on resout fizzbuzz_etendu (même pitfall : concatenation de règles modulo). On compare le taux de reussite avec vs sans memoire retrocedee.

Ce que la demo montre reellement : le mécanisme (persistence cross-run + retrocession par similarite + injection dans le prompt) fonctionne — les lecons survivent et sont bien retrocedees (score cosine 0.599). En revanche, sur ces deux problemes simples resolus en 1 itération, le delta mesure vaut 0 : la baseline froide resout déjà fizzbuzz_etendu seule. De plus, la lecon stockee ici est un diagnostic de statut (“Resolu en N itération(s). Pitfall maitrise.”), pas une description technique du pitfall : elle ne transporte donc pas d’information exploitable par le générateur. Pour qu’un transfer mesurable apparaisse, il faudrait une lecon instructive (ex. le feedback des tests echoues) ET des problemes ou la baseline froide echoue — conditions explorees dans la conclusion (G.2) et les exercices.

# --- Run 1 : fizzbuzz a FROID (memoire vide) ---
import shutil
if Path("tmp/memoire_reflexion.json").exists():
    shutil.rmtree("tmp/memoire_reflexion.json", ignore_errors=True)

embedder = BoWEmbedder().fit([p["spec"] for p in PROBLEMES.values()])
mem = MemoireVectorielle(embedder, chemin="tmp/memoire_reflexion.json")

print("=" * 64)
print("RUN 1 — fizzbuzz A FROID (memoire vide)")
print("=" * 64)
p1 = dict(PROBLEMES["fizzbuzz"]); p1["id"] = "fizzbuzz"
pass1, tot1, diag1, _ = reflexion_avec_memoire(p1, mem, iterations=3, verbose=True)
n_avant = mem.sauvegarder()
print(f"Resultat : {pass1}/{tot1} tests | memoire persistee : {n_avant} entree(s)")
print(f"Diagnostic memorise : {diag1[:120]}")
BoWEmbedder fitte : vocabulaire de 24 tokens.
================================================================
RUN 1 — fizzbuzz A FROID (memoire vide)
================================================================
Resultat : 2/2 tests | memoire persistee : 1 entree(s)
Diagnostic memorise : Resolu en 1 iteration(s). Pitfall maitrise.

Lecture du Run 1 à froid. La sortie Resultat : 2/2 tests | memoire persistee : 1 entree(s) montre que le problème fizzbuzz a été résolu en 1 itération avec le pitfall maîtrisé, et cette leçon a été automatiquement stockée dans la mémoire — une première entraînante pour le système.

# --- Run 2 : fizzbuzz_etendu AVEC memoire (lecon de fizzbuzz retrocedee) ---
print("=" * 64)
print("RUN 2 — fizzbuzz_etendu AVEC MEMOIRE (lecon fizzbuzz retrocedee)")
print("=" * 64)
p2 = dict(PROBLEMES["fizzbuzz_etendu"]); p2["id"] = "fizzbuzz_etendu"
pass2, tot2, diag2, lecons = reflexion_avec_memoire(p2, mem, iterations=3, verbose=True)
print(f"Resultat : {pass2}/{tot2} tests")
print(f"Lecons retrocede(s) : {len(lecons)} (la 1ere : '{lecons[0]['lecon'][:80]}' "
      f"si non vide)" if lecons else "Aucune lecon retrocedee.")

print("\n--- Comparaison : la memoire a-t-elle aide ? ---")
# Baseline : fizzbuzz_etendu SANS memoire (memoire fraiche), pour comparaison honnete (G.2).
mem_vider = MemoireVectorielle(embedder, chemin="tmp/memoire_vide.json")
pass2_base, tot2_base, diag2_base, _ = reflexion_avec_memoire(p2, mem_vider, iterations=3)
print(f"fizzbuzz_etendu AVEC memoire    : {pass2}/{tot2} tests")
print(f"fizzbuzz_etendu SANS memoire    : {pass2_base}/{tot2_base} tests  (baseline froide)")
print(f"Delta memoire                   : {pass2 - pass2_base:+d} test(s)")
================================================================
RUN 2 — fizzbuzz_etendu AVEC MEMOIRE (lecon fizzbuzz retrocedee)
================================================================
  [memoire] 1 lecon(s) retrocede(s) : #0(score=0.599)
Resultat : 3/3 tests
Lecons retrocede(s) : 1 (la 1ere : '[fizzbuzz] Resolu en 1 iteration(s). Pitfall maitrise.' si non vide)

--- Comparaison : la memoire a-t-elle aide ? ---
fizzbuzz_etendu AVEC memoire    : 3/3 tests
fizzbuzz_etendu SANS memoire    : 3/3 tests  (baseline froide)
Delta memoire                   : +0 test(s)

Lecture du Run 2 avec mémoire. Le résultat Resultat : 3/3 tests avec la mention [memoire] 1 lecon(s) retrocede(s) : #0(score=0.599) montre que la leçon de fizzbuzz a bien été rétrocedée pour fizzbuzz_etendu — mais le tableau final est honnête : fizzbuzz_etendu SANS memoire : 3/3 tests (baseline froide) et Delta memoire : +0 test(s) — sur ce problème, la mémoire n’a pas changé le score.

# --- Persistence cross-run : on recharge la memoire depuis le disque ---
print("=" * 64)
print("PERSISTENCE CROSS-RUN — rechargement depuis le disque")
print("=" * 64)
mem_rechargee = MemoireVectorielle(embedder, chemin="tmp/memoire_reflexion.json")
n = mem_rechargee.charger()
print(f"Memoire rechargee depuis tmp/memoire_reflexion.json : {n} entree(s) persistee(s).")
for e in mem_rechargee.entrees:
    print(f"  #{e['id']} : {e['lecon'][:90]}")
print("=> Les lecons SURVIVENT aux runs (vs NB-12 ou la liste Python etait perdue).")
================================================================
PERSISTENCE CROSS-RUN — rechargement depuis le disque
================================================================
Memoire rechargee depuis tmp/memoire_reflexion.json : 1 entree(s) persistee(s).
  #0 : [fizzbuzz] Resolu en 1 iteration(s). Pitfall maitrise.
=> Les lecons SURVIVENT aux runs (vs NB-12 ou la liste Python etait perdue).

Lecture des trois runs : le mécanisme fonctionne — mais le bénéfice mesuré est nul

Les sorties committées ci-dessus résument les trois axes de la démonstration :

Run Ce qu’il éprouve Sortie committée Verdict
1 — fizzbuzz à froid Écriture de la leçon sur disque 2/2 tests, memoire persistee : 1 entree(s) La leçon est captée et persistée
2 — fizzbuzz_etendu AVEC mémoire Rétrocession par similarité leçon rétrocédée #0 (score=0.599), 3/3 tests Le recall se déclenche (similarité BoW borderline)
3 — rechargement cross-run Survie hors RAM 1 entree(s) persistee(s) rechargée depuis tmp/memoire_reflexion.json La mémoire survit à un redémarrage

Le point critique — et le plus honnête — est la ligne de comparaison du Run 2 :

fizzbuzz_etendu AVEC memoire : 3/3 tests
fizzbuzz_etendu SANS memoire : 3/3 tests  (baseline froide)
Delta memoire                : +0 test(s)

Le mécanisme s’est déclenché, et pourtant il n’a rien rapporté. C’est la leçon la plus importante de cette démonstration : « la mémoire a été consultée » n’est pas « la mémoire a aidé ». La cellule calcule exprès le delta pour séparer ces deux faits.

Pourquoi le delta est +0 ici :

  • fizzbuzz_etendu reste soluble à froid. Le LLM le résout en une passe même sans aide. Le pitfall partagé (concaténer les règles modulo) n’est pas assez bloquant pour qu’un échec cold-start se produise — donc il n’y a rien à débloquer.
  • La leçon rétrocédée est générique. « [fizzbuzz] Resolu en 1 iteration(s). Pitfall maitrise. » confirme un succès passé ; elle ne porte pas de correctif spécifique (un piège précis, une tactique de déblocage) qui aurait changé l’issue. Une leçon corrective (« attention au cas N=15 où Fizz et Buzz se chevauchent : tester la divisibilité cumulée d’abord ») aurait un pouvoir de transfert réel.
  • La similarité BoW (0.599) est borderline. Le BoWEmbedder (24 tokens, sac de mots) est grossier : il détecte que fizzbuzz et fizzbuzz_etendu se ressemblent partiellement — assez pour déclencher la rétrocession, mais sans garantir que la leçon est pertinente pour le nouveau problème.

Quand le transfert se rentabiliserait-il ? Il faudrait réunir trois conditions : un problème cible où le cold-solving échoue au pitfall partagé, une leçon qui porte un correctif spécifique (pas juste un acquittement), et un embedder assez fin pour distinguer « problèmes au pitfall commun » de « problèmes qui partagent juste des mots ». L’Exercice 1 ci-dessous (embedder OpenAI text-embedding-3-large, pont NB-05) adresse précisément le troisième levier — un embedder sémantique aurait donné un score de similarité plus discriminant que 0.599.

À retenir. Cette démonstration éprouve le mécanisme (persister, rétrocéder, survivre au redémarrage) et, accessoirement, montre sa limite : sur un problème trop simple, le bénéfice est nul et mesurablement nul. C’est exactement pourquoi on instrumente le delta plutôt que de supposer que « mémoire = mieux ».

6. Travaux pratiques

Les exercices sont a completer (convention C.1 : pas d’erreur volontaire, le notebook tourne de bout en bout même non complete).

Exercice 1 : embedder OpenAI (pont NB-05)

Implemente une classe OpenAIEmbedder qui vectorise via client.embeddings.create (modèle text-embedding-3-large, cf NB-05 cellule create_embedding). Elle doit exposer la même interface que BoWEmbedder : méthode embed(text) -> vecteur (np.array). Vérifie qu’en la branchant dans MemoireVectorielle, le notebook fonctionne a l’identique (seul l’embedder change).

Indice : wrappe client.embeddings.create(model=..., input=[text]) dans un try/except (les embeddings via OpenRouter peuvent echouer ; garde BoWEmbedder comme fallback).

class OpenAIEmbedder:
    """Exercice 1 : embedder production pont NB-05 (text-embedding-3-large).
    Doit exposer .embed(text) -> np.array, interface identique a BoWEmbedder."""
    def __init__(self, model="text-embedding-3-large"):
        self.model = model
        # TODO etudiant : implemente embed(text) via client.embeddings.create.
        pass

    def embed(self, text):
        # TODO etudiant : appeler client.embeddings.create(model=self.model, input=[text]),
        # retourner np.array(response.data[0].embedding). Gerer l'echec (return vecteur nul).
        return None

_o = OpenAIEmbedder()
print(f"Exercice 1 - OpenAIEmbedder : {'implemente' if _o.embed('test') is not None else 'a completer'}")
Exercice 1 - OpenAIEmbedder : a completer

Exercice 2 : politique d’oubli (eviction LRU + taille max)

Une memoire non bornee grossit indefiniment (cout de retrieval + stockage). Implemente une politique d’eviction dans MemoireVectorielle : taille maximale max_entrees, et eviction de l’entree la moins recentement utilisee (LRU) quand on depasse. Ajoute un compteur nb_acces par entree, incremente a chaque retroceder.

Indice : garde une liste ordonnee par recence d’acces ; ajoute depasse les max_entrees -> supprime la plus ancienne. Teste avec max_entrees=2 en ajoutant 3 lecons.

def ajouter_avec_eviction(memoire, spec, lecon, max_entrees=50):
    """Exercice 2 : ajoute une lecon en evictant la moins recente (LRU) si > max_entrees.
    Reutilise memoire.ajouter + une logique LRU sur memoire.entrees."""
    # TODO etudiant : implementer l'eviction LRU quand len(memoire.entrees) >= max_entrees.
    return None

print(f"Exercice 2 - eviction LRU : {'implemente' if False else 'a completer'}")
Exercice 2 - eviction LRU : a completer

Exercice 3 (avance) : memoire partagee multi-domaines (transfer cross-task)

Etends la memoire pour gerer des domaines (ex: ‘code’, ‘math’, ‘logique’). Au retroceder, filtre optionnellement par domaine. Etudie si une lecon apprise sur un domaine transfer vers un autre (ex: pitfall ‘off-by-one’ en code vs en arithmetique).

Indice : ajoute un champ ‘domaine’ aux entrees ; paramètre domaine=None (tous) dans retroceder. Conclus honnetement (G.2) si le transfer est reel ou illusoire.

def retroceder_par_domaine(memoire, spec_query, domaine=None, k=3):
    """Exercice 3 : retrieval filtre par domaine (transfer cross-task)."""
    # TODO etudiant : filtrer memoire.entrees par domaine avant le tri cosinus.
    return None

print(f"Exercice 3 - memoire multi-domaines : {'implemente' if False else 'a completer'}")
Exercice 3 - memoire multi-domaines : a completer

7. Conclusion et suite

On a remplace la memoire volatile (liste Python) du moteur Reflexion de NB-12 par une memoire vectorielle persistante qui retrocede les lecons par similarite cosinus — l’agent “Memory” d’ICR. Les lecons survivent aux runs (persistence JSON) et sont injectees dans le générateur quand un problème apparente se presente.

Limites honnetes (G.2) : - Embeddings BoW : la baseline bag-of-words capte la similarite lexicale, pas sémantique fine. Deux problemes decrits avec des mots différents mais de logique identique ne seront pas retrocedes. En production : embeddings OpenAI / sentence-transformers (Exercice 1, pont NB-05). - Benefice marginal sur 2 problemes : la demo montre le mécanisme (transfer de lecon). Le gain mesure (delta memoire) depend fortement des problemes ; sur un benchmark large, la memoire aide surtout quand beaucoup de problemes similaires se repetent. - Cout : la retrocession allonge le 1er prompt (k lecons injectees). A budget fixe, cela reduit le budget generation (pont NB-09). - Pas d’oubli : la memoire courante grossit indefiniment (Exercice 2 = eviction LRU).

Suite de l’epic #2926 : - Phase 3 — ToT sur de vrais problemes de recherche (cryptarithmetic SEND+MORE=MONEY, CSP) ou single-shot echoue systematiquement (pont series Search/Sudoku). - Phase 4 — courbes de scaling Snell 2024 (quand echantillonner large vs chercher profond). - Phase 5 — hand-rolled vs modèles a raisonnement natif (gpt-5-thinking), cout-normalise. - Phase 6 — plugin Semantic Kernel (pont series SemanticKernel).

References : ICR repo ; Shinn et al. 2023 (Reflexion) ; Snell et al. 2024 (scaling) ; NB-05 (RAG embeddings), NB-12 (moteurs), NB-13 (routeur agentique).

Retour au sommet