Mémoire et Contexte Long-Terme pour Agents IA

┌─────────────────────────────────────────────────┐
│              SYSTÈME DE MÉMOIRE                 │
├─────────────────────────────────────────────────┤
│                                                 │
│  📝 COURT-TERME   (Buffer, derniers messages)  │
│  🧠 LONG-TERME    (Vector DB, faits permanents)│
│  🎬 ÉPISODIQUE    (Historique des actions)     │
│  📚 SÉMANTIQUE    (Connaissances structurées)  │
│                                                 │
└─────────────────────────────────────────────────┘

class ShortTermMemory:
    """Mémoire court-terme : buffer de conversation"""

    def __init__(self, max_messages: int = 10):
        self.messages = []
        self.max_messages = max_messages

    def add(self, role: str, content: str):
        """Ajoute un message"""
        self.messages.append({"role": role, "content": content})

        # Garder seulement les N derniers messages
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]

    def get_context(self) -> list:
        """Retourne le contexte pour le LLM"""
        return self.messages

    def clear(self):
        """Efface la mémoire"""
        self.messages = []

# Utilisation
memory = ShortTermMemory(max_messages=10)

memory.add("user", "Je m'appelle Alice")
memory.add("assistant", "Enchanté Alice !")
memory.add("user", "Quelle est la météo ?")

# Obtenir le contexte
context = memory.get_context()

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter
from datetime import datetime

class LongTermMemory:
    """Mémoire long-terme avec RAG"""

    def __init__(self):
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = None
        self.documents = []

    def remember(self, text: str, metadata: dict = None):
        """Stocke une information en mémoire long-terme"""
        # Ajouter timestamp
        if metadata is None:
            metadata = {}
        metadata["timestamp"] = datetime.now().isoformat()

        # Splitter si texte long
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )
        chunks = splitter.split_text(text)

        # Créer ou mettre à jour le vectorstore
        if self.vectorstore is None:
            self.vectorstore = FAISS.from_texts(
                chunks,
                self.embeddings,
                metadatas=[metadata] * len(chunks)
            )
        else:
            self.vectorstore.add_texts(chunks, metadatas=[metadata] * len(chunks))

    def recall(self, query: str, k: int = 3) -> list:
        """Récupère les K informations les plus pertinentes"""
        if self.vectorstore is None:
            return []

        docs = self.vectorstore.similarity_search_with_score(query, k=k)

        return [
            {
                "content": doc.page_content,
                "metadata": doc.metadata,
                "score": score
            }
            for doc, score in docs
        ]

    def recall_recent(self, n: int = 5) -> list:
        """Récupère les N souvenirs les plus récents"""
        if self.vectorstore is None:
            return []

        # Trier par timestamp
        all_docs = self.vectorstore.similarity_search("", k=100)
        sorted_docs = sorted(
            all_docs,
            key=lambda d: d.metadata.get("timestamp", ""),
            reverse=True
        )

        return sorted_docs[:n]

# Utilisation
long_term = LongTermMemory()

# Stocker des faits
long_term.remember("L'utilisateur s'appelle Alice", {"type": "user_info"})
long_term.remember("Alice travaille chez TechCorp en tant que data scientist", {"type": "user_info"})
long_term.remember("Alice aime le machine learning et Python", {"type": "preferences"})

# Récupérer par similarité
results = long_term.recall("Quel est le travail de l'utilisateur ?", k=2)
for result in results:
    print(f"💡 {result['content']} (score: {result['score']:.2f})")

# Sortie :
# 💡 Alice travaille chez TechCorp en tant que data scientist (score: 0.82)
# 💡 Alice aime le machine learning et Python (score: 0.65)

from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class Episode:
    """Un épisode : une action et son résultat"""
    timestamp: datetime
    action: str
    input: dict
    output: str
    success: bool
    error: Optional[str] = None

class EpisodicMemory:
    """Mémoire épisodique : historique des actions"""

    def __init__(self):
        self.episodes = []

    def record(self, action: str, input: dict, output: str, success: bool, error: str = None):
        """Enregistre un épisode"""
        episode = Episode(
            timestamp=datetime.now(),
            action=action,
            input=input,
            output=output,
            success=success,
            error=error
        )
        self.episodes.append(episode)

    def get_recent(self, n: int = 10) -> list:
        """Récupère les N derniers épisodes"""
        return self.episodes[-n:]

    def get_failed_episodes(self) -> list:
        """Récupère tous les épisodes échoués"""
        return [ep for ep in self.episodes if not ep.success]

    def get_by_action(self, action_name: str) -> list:
        """Récupère tous les épisodes d'une action spécifique"""
        return [ep for ep in self.episodes if ep.action == action_name]

    def get_summary(self) -> dict:
        """Résumé statistique"""
        total = len(self.episodes)
        successful = sum(1 for ep in self.episodes if ep.success)

        action_counts = {}
        for ep in self.episodes:
            action_counts[ep.action] = action_counts.get(ep.action, 0) + 1

        return {
            "total_episodes": total,
            "successful": successful,
            "failed": total - successful,
            "success_rate": successful / total if total > 0 else 0,
            "actions_used": action_counts
        }

# Utilisation
episodic = EpisodicMemory()

# Enregistrer des actions
episodic.record(
    action="search_web",
    input={"query": "météo Paris"},
    output="18°C, nuageux",
    success=True
)

episodic.record(
    action="send_email",
    input={"to": "[email protected]"},
    output="",
    success=False,
    error="SMTP connection failed"
)

# Analyser l'historique
summary = episodic.get_summary()
print(f"📊 {summary['successful']}/{summary['total_episodes']} actions réussies")

# Identifier les problèmes récurrents
failed = episodic.get_failed_episodes()
if failed:
    print(f"⚠️ {len(failed)} actions échouées")
    for ep in failed:
        print(f"  - {ep.action}: {ep.error}")

from typing import Dict, List, Set

class SemanticMemory:
    """Mémoire sémantique : graphe de connaissances"""

    def __init__(self):
        self.facts = {}  # {entity: {attribute: value}}
        self.relations = {}  # {entity: {relation: [entities]}}

    def add_fact(self, entity: str, attribute: str, value: str):
        """Ajoute un fait : entity.attribute = value"""
        if entity not in self.facts:
            self.facts[entity] = {}
        self.facts[entity][attribute] = value

    def add_relation(self, entity1: str, relation: str, entity2: str):
        """Ajoute une relation : entity1 -[relation]-> entity2"""
        if entity1 not in self.relations:
            self.relations[entity1] = {}
        if relation not in self.relations[entity1]:
            self.relations[entity1][relation] = []
        self.relations[entity1][relation].append(entity2)

    def get_fact(self, entity: str, attribute: str) -> str:
        """Récupère un fait"""
        return self.facts.get(entity, {}).get(attribute)

    def get_all_facts(self, entity: str) -> dict:
        """Récupère tous les faits sur une entité"""
        return self.facts.get(entity, {})

    def get_relations(self, entity: str, relation: str = None) -> dict:
        """Récupère les relations d'une entité"""
        if relation:
            return self.relations.get(entity, {}).get(relation, [])
        return self.relations.get(entity, {})

    def query(self, entity: str) -> dict:
        """Récupère toutes les informations sur une entité"""
        return {
            "facts": self.get_all_facts(entity),
            "relations": self.get_relations(entity)
        }

# Utilisation
semantic = SemanticMemory()

# Ajouter des connaissances
semantic.add_fact("Alice", "nom_complet", "Alice Dupont")
semantic.add_fact("Alice", "profession", "Data Scientist")
semantic.add_fact("Alice", "entreprise", "TechCorp")

semantic.add_fact("TechCorp", "secteur", "Intelligence Artificielle")
semantic.add_fact("TechCorp", "effectif", "250 employés")

# Ajouter des relations
semantic.add_relation("Alice", "travaille_pour", "TechCorp")
semantic.add_relation("Alice", "utilise", "Python")
semantic.add_relation("Alice", "utilise", "TensorFlow")

# Requêter
info = semantic.query("Alice")
print(f"👤 Alice :")
print(f"  Profession : {info['facts']['profession']}")
print(f"  Entreprise : {info['facts']['entreprise']}")
print(f"  Outils : {', '.join(info['relations'].get('utilise', []))}")

# Sortie :
# 👤 Alice :
#   Profession : Data Scientist
#   Entreprise : TechCorp
#   Outils : Python, TensorFlow

class HybridMemory:
    """Système de mémoire hybride"""

    def __init__(self):
        self.short_term = ShortTermMemory(max_messages=10)
        self.long_term = LongTermMemory()
        self.episodic = EpisodicMemory()
        self.semantic = SemanticMemory()

    def add_message(self, role: str, content: str):
        """Ajoute un message (court-terme)"""
        self.short_term.add(role, content)

        # Aussi en long-terme si important
        if role == "user":
            self.long_term.remember(f"User said: {content}", {"type": "conversation"})

    def add_fact(self, entity: str, attribute: str, value: str):
        """Ajoute un fait structuré"""
        self.semantic.add_fact(entity, attribute, value)

        # Aussi en long-terme (format texte)
        self.long_term.remember(f"{entity}'s {attribute} is {value}", {"type": "fact"})

    def record_action(self, action: str, input: dict, output: str, success: bool):
        """Enregistre une action"""
        self.episodic.record(action, input, output, success)

    def build_context(self, current_query: str, max_tokens: int = 2000) -> dict:
        """Construit le contexte optimal pour le LLM"""

        context = {
            "short_term": self.short_term.get_context(),
            "relevant_memories": [],
            "recent_actions": [],
            "semantic_facts": {}
        }

        # Récupérer les souvenirs pertinents (RAG)
        relevant = self.long_term.recall(current_query, k=3)
        context["relevant_memories"] = [r["content"] for r in relevant]

        # Ajouter les actions récentes
        recent_episodes = self.episodic.get_recent(n=5)
        context["recent_actions"] = [
            f"{ep.action}: {ep.output}" for ep in recent_episodes if ep.success
        ]

        # TODO: Estimer les tokens et tronquer si nécessaire

        return context

    def get_system_prompt(self, context: dict) -> str:
        """Génère le prompt système avec contexte"""

        parts = ["Tu es un assistant IA avec mémoire."]

        # Ajouter les souvenirs pertinents
        if context["relevant_memories"]:
            parts.append("\n**Informations pertinentes :**")
            for memory in context["relevant_memories"]:
                parts.append(f"- {memory}")

        # Ajouter les actions récentes
        if context["recent_actions"]:
            parts.append("\n**Actions récentes :**")
            for action in context["recent_actions"]:
                parts.append(f"- {action}")

        return "\n".join(parts)


# ========== EXEMPLE D'UTILISATION ==========

memory = HybridMemory()

# Session 1
memory.add_message("user", "Je m'appelle Alice")
memory.add_message("assistant", "Enchanté Alice !")
memory.add_fact("user", "name", "Alice")

memory.add_message("user", "Je travaille sur un projet de NLP")
memory.add_fact("user", "current_project", "NLP")

# Session 2 (plus tard)
memory.add_message("user", "Comment avance mon projet ?")

# Construire le contexte
context = memory.build_context("Comment avance mon projet ?")

# Le contexte inclura automatiquement :
# - La conversation actuelle (court-terme)
# - Le fait que l'utilisateur s'appelle Alice (long-terme)
# - Le fait qu'elle travaille sur un projet NLP (sémantique)
# - Les actions récentes (épisodique)

system_prompt = memory.get_system_prompt(context)
print(system_prompt)

Tu es un assistant IA avec mémoire.

**Informations pertinentes :**
- L'utilisateur s'appelle Alice
- Alice travaille sur un projet de NLP
- User said: Je travaille sur un projet de NLP

**Actions récentes :**
- search_web: Résultats sur les transformers en NLP
- get_weather: 18°C à Paris

def summarize_conversation(messages: list, llm) -> str:
    """Résume une conversation en 2-3 phrases"""

    conversation_text = "\n".join([
        f"{msg['role']}: {msg['content']}" for msg in messages
    ])

    prompt = f"""Résume cette conversation en 2-3 phrases concises.
Capture les points clés et les décisions importantes.

Conversation :
{conversation_text}

Résumé :"""

    response = llm(prompt)
    return response

# Utilisation
class SummarizingMemory:
    """Mémoire avec résumé automatique"""

    def __init__(self, llm, window_size=10, summary_threshold=20):
        self.llm = llm
        self.messages = []
        self.summary = ""
        self.window_size = window_size
        self.summary_threshold = summary_threshold

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

        # Résumer si trop de messages
        if len(self.messages) > self.summary_threshold:
            # Résumer les anciens messages
            old_messages = self.messages[:-self.window_size]
            new_summary = summarize_conversation(old_messages, self.llm)

            # Combiner avec le résumé existant
            if self.summary:
                self.summary = f"{self.summary}\n{new_summary}"
            else:
                self.summary = new_summary

            # Garder seulement les messages récents
            self.messages = self.messages[-self.window_size:]

    def get_context(self) -> dict:
        return {
            "summary": self.summary,
            "recent_messages": self.messages
        }

import tiktoken

class TokenAwareMemory:
    """Mémoire qui respecte un budget de tokens"""

    def __init__(self, model="gpt-4", max_tokens=4000):
        self.encoding = tiktoken.encoding_for_model(model)
        self.max_tokens = max_tokens
        self.messages = []

    def count_tokens(self, text: str) -> int:
        """Compte le nombre de tokens"""
        return len(self.encoding.encode(text))

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

    def get_context_within_budget(self) -> list:
        """Retourne le contexte qui rentre dans le budget"""

        total_tokens = 0
        context = []

        # Partir de la fin (messages les plus récents)
        for msg in reversed(self.messages):
            msg_tokens = self.count_tokens(msg["content"])

            if total_tokens + msg_tokens > self.max_tokens:
                break

            context.insert(0, msg)
            total_tokens += msg_tokens

        return context

# Utilisation
memory = TokenAwareMemory(max_tokens=2000)

# Ajouter beaucoup de messages
for i in range(100):
    memory.add_message("user", f"Message {i}")

# Ne garde que ce qui rentre dans le budget
context = memory.get_context_within_budget()
print(f"✅ {len(context)} messages gardés sur {len(memory.messages)}")

class HierarchicalMemory:
    """Mémoire hiérarchique avec résumés multi-niveaux"""

    def __init__(self, llm):
        self.llm = llm
        self.current_session = []  # Messages actuels
        self.session_summaries = []  # Résumés de sessions
        self.weekly_summaries = []  # Résumés hebdomadaires

    def add_message(self, role: str, content: str):
        self.current_session.append({"role": role, "content": content})

    def end_session(self):
        """Termine la session et crée un résumé"""
        if not self.current_session:
            return

        summary = summarize_conversation(self.current_session, self.llm)
        self.session_summaries.append({
            "timestamp": datetime.now(),
            "summary": summary,
            "message_count": len(self.current_session)
        })

        self.current_session = []

        # Si 7 sessions, créer un résumé hebdomadaire
        if len(self.session_summaries) >= 7:
            weekly_summary = self._create_weekly_summary()
            self.weekly_summaries.append(weekly_summary)
            self.session_summaries = []

    def _create_weekly_summary(self) -> str:
        """Résume une semaine de sessions"""
        all_summaries = "\n".join([s["summary"] for s in self.session_summaries])

        prompt = f"""Résume cette semaine d'interactions en un paragraphe.
Capture les thèmes récurrents et les évolutions.

Résumés quotidiens :
{all_summaries}

Résumé hebdomadaire :"""

        return self.llm(prompt)

    def get_full_context(self) -> dict:
        return {
            "weekly_summaries": self.weekly_summaries,
            "recent_session_summaries": self.session_summaries[-3:],
            "current_session": self.current_session
        }

📅 Semaine 1 (résumé)
  └─ 📆 Session 1 (résumé)
  └─ 📆 Session 2 (résumé)
  └─ 📆 Session 3 (résumé)
  ...
📅 Semaine 2 (résumé)
  └─ 📆 Session 8 (messages complets)  ← Session actuelle

class SelectiveMemory:
    """Mémoire qui oublie sélectivement"""

    def __init__(self, llm):
        self.llm = llm
        self.messages = []

    def add_message(self, role: str, content: str):
        # Calculer l'importance
        importance = self._calculate_importance(content)

        self.messages.append({
            "role": role,
            "content": content,
            "importance": importance,
            "timestamp": datetime.now()
        })

    def _calculate_importance(self, content: str) -> float:
        """Calcule l'importance d'un message (0-1)"""

        # Heuristiques simples
        importance = 0.5  # Base

        # Augmenter si contient des infos personnelles
        if any(word in content.lower() for word in ["je m'appelle", "mon nom", "j'habite"]):
            importance += 0.3

        # Augmenter si contient des décisions
        if any(word in content.lower() for word in ["décider", "choisir", "je veux"]):
            importance += 0.2

        # Diminuer si question simple
        if content.endswith("?") and len(content.split()) < 10:
            importance -= 0.1

        return min(max(importance, 0), 1)

    def forget_unimportant(self, threshold: float = 0.3):
        """Oublie les messages peu importants"""
        self.messages = [
            msg for msg in self.messages
            if msg["importance"] >= threshold
        ]

    def get_context(self) -> list:
        """Retourne le contexte, trié par importance"""
        return sorted(self.messages, key=lambda m: m["importance"], reverse=True)

from llmlingua import PromptCompressor

compressor = PromptCompressor(
    model_name="microsoft/llmlingua-2-xlm-roberta-large-meetingbank"
)

# Contexte long
long_context = """
Alice Dupont travaille chez TechCorp en tant que data scientist depuis 3 ans.
Elle est spécialisée en NLP et a récemment travaillé sur un projet de
classification de sentiments pour les réseaux sociaux. Alice utilise
principalement Python et TensorFlow. Elle a un master en IA de l'université
Paris-Saclay. Son projet actuel concerne l'analyse de données textuelles
pour améliorer le service client. Elle collabore avec l'équipe produit
et l'équipe data engineering. Alice aime le machine learning et passe
son temps libre à contribuer à des projets open source.
"""

# Compresser
compressed = compressor.compress_prompt(
    long_context,
    instruction="Garde les informations essentielles sur Alice",
    rate=0.5  # Compresser à 50%
)

print(f"Original: {len(long_context)} caractères")
print(f"Compressé: {len(compressed['compressed_prompt'])} caractères")
print(f"\nTexte compressé:\n{compressed['compressed_prompt']}")

# Sortie exemple :
# Original: 580 caractères
# Compressé: 290 caractères
#
# Texte compressé:
# Alice Dupont data scientist TechCorp 3 ans. NLP classification sentiments.
# Python TensorFlow. Master IA Paris-Saclay. Projet analyse textes service client.

class SlidingWindowMemory:
    """Fenêtre glissante avec ancrage"""

    def __init__(self, window_size=10, anchor_messages=None):
        self.window_size = window_size
        self.anchor_messages = anchor_messages or []  # Messages importants toujours gardés
        self.messages = []

    def add_anchor(self, role: str, content: str):
        """Ajoute un message d'ancrage (toujours présent)"""
        self.anchor_messages.append({"role": role, "content": content})

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

        # Garder seulement les N derniers
        if len(self.messages) > self.window_size:
            self.messages = self.messages[-self.window_size:]

    def get_context(self) -> list:
        """Retourne : ancres + fenêtre glissante"""
        return self.anchor_messages + self.messages

# Utilisation
memory = SlidingWindowMemory(window_size=5)

# Définir les ancres (toujours présentes)
memory.add_anchor("system", "Tu es un assistant pour Alice Dupont, data scientist")
memory.add_anchor("user", "Je travaille sur un projet NLP")

# Ajouter des messages normaux (fenêtre glissante)
for i in range(20):
    memory.add_message("user", f"Message {i}")

# Le contexte inclura :
# - Les 2 ancres (toujours)
# - Les 5 derniers messages (fenêtre)
context = memory.get_context()
print(f"Contexte : {len(context)} messages")
# → 7 messages (2 ancres + 5 récents)

import pickle
from pathlib import Path

class PersistentAgent:
    """Agent avec mémoire persistante entre sessions"""

    def __init__(self, user_id: str, memory_dir: str = "./memories"):
        self.user_id = user_id
        self.memory_path = Path(memory_dir) / f"{user_id}.pkl"
        self.memory = self._load_memory()

    def _load_memory(self) -> HybridMemory:
        """Charge la mémoire depuis le disque"""
        if self.memory_path.exists():
            with open(self.memory_path, "rb") as f:
                return pickle.load(f)
        return HybridMemory()

    def _save_memory(self):
        """Sauvegarde la mémoire sur le disque"""
        self.memory_path.parent.mkdir(exist_ok=True)
        with open(self.memory_path, "wb") as f:
            pickle.dump(self.memory, f)

    def chat(self, message: str) -> str:
        """Converse avec mémoire"""
        # Ajouter le message
        self.memory.add_message("user", message)

        # Construire le contexte
        context = self.memory.build_context(message)

        # Appeler le LLM avec le contexte
        system_prompt = self.memory.get_system_prompt(context)

        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": system_prompt},
                *context["short_term"]
            ]
        )

        answer = response.choices[0].message.content

        # Sauvegarder la réponse
        self.memory.add_message("assistant", answer)

        # Persister
        self._save_memory()

        return answer

# Utilisation
agent = PersistentAgent(user_id="alice_123")

# Session 1
agent.chat("Je m'appelle Alice")
agent.chat("Je travaille sur un projet NLP")

# [Fermer l'application]

# Session 2 (plus tard)
agent = PersistentAgent(user_id="alice_123")  # Charge la mémoire
response = agent.chat("Comment s'appelle mon projet ?")
# → "Votre projet est sur le NLP" (se souvient !)