RAG Conversationnel : Memoire et contexte multi-sessions

Un RAG statique repond a chaque question isolement. Un RAG conversationnel maintient le fil de la discussion, se souvient des preferences utilisateur et personnalise ses reponses au fil du temps. Ce guide explique comment implementer une memoire efficace.

Pourquoi la memoire est essentielle

Les limites du RAG sans memoire

Sans memoire, chaque requete est traitee independamment :

Utilisateur: Quelle est votre politique de retour ?
Assistant: Vous avez 30 jours pour retourner un produit.

Utilisateur: Et pour les produits electroniques ?
Assistant: [Pas de contexte] Que voulez-vous savoir sur les produits electroniques ?

Avec memoire :

Utilisateur: Quelle est votre politique de retour ?
Assistant: Vous avez 30 jours pour retourner un produit.

Utilisateur: Et pour les produits electroniques ?
Assistant: Pour les produits electroniques, le delai de retour est egalement de 30 jours,
          mais ils doivent etre non ouverts sauf en cas de defaut.

Types de memoire

Architecture conversationnelle

┌─────────────────────────────────────────────────────────────┐
│                    REQUETE UTILISATEUR                       │
└───────────────────────────┬─────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────────┐
│               CONTEXT MANAGER                                │
│  ┌─────────────┐ ┌─────────────┐ ┌───────────────────────┐  │
│  │  Historique │ │   User      │ │      Memoire          │  │
│  │  Session    │ │  Profile    │ │      Long Terme       │  │
│  └──────┬──────┘ └──────┬──────┘ └───────────┬───────────┘  │
│         └───────────────┼───────────────────┘               │
│                         ▼                                    │
│               ┌─────────────────┐                           │
│               │ Context Builder │                           │
│               └────────┬────────┘                           │
└────────────────────────┼────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│                    RAG PIPELINE                              │
│  ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│  │  Query       │ │   Retrieval  │ │     Generation       │ │
│  │  Rewriting   │ │   + Context  │ │     + Memory         │ │
│  └──────────────┘ └──────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

Gestion de l'historique de session

Structure de conversation

DEVELOPERpython
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional

@dataclass
class Message:
    role: str  # "user" ou "assistant"
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: dict = field(default_factory=dict)

@dataclass
class Conversation:
    id: str
    user_id: str
    messages: List[Message] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)
    context: dict = field(default_factory=dict)

class ConversationManager:
    def __init__(self, storage, max_messages: int = 20):
        self.storage = storage
        self.max_messages = max_messages

    async def add_message(
        self,
        conversation_id: str,
        role: str,
        content: str,
        metadata: dict = None
    ):
        """Ajoute un message a la conversation"""
        conversation = await self.storage.get(conversation_id)

        message = Message(
            role=role,
            content=content,
            metadata=metadata or {}
        )

        conversation.messages.append(message)

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

        await self.storage.save(conversation)

    async def get_context(
        self,
        conversation_id: str,
        max_turns: int = 5
    ) -> str:
        """Recupere le contexte de conversation formate"""
        conversation = await self.storage.get(conversation_id)

        # Prendre les derniers tours
        recent_messages = conversation.messages[-(max_turns * 2):]

        context_parts = []
        for msg in recent_messages:
            role = "Utilisateur" if msg.role == "user" else "Assistant"
            context_parts.append(f"{role}: {msg.content}")

        return "\n".join(context_parts)

Compression de l'historique

Pour les longues conversations, compresser l'historique ancien :

DEVELOPERpython
class HistoryCompressor:
    def __init__(self, llm):
        self.llm = llm

    async def compress(
        self,
        messages: List[Message],
        keep_recent: int = 4
    ) -> dict:
        """
        Compresse l'historique en gardant les messages recents intacts
        """
        if len(messages) <= keep_recent:
            return {
                "summary": None,
                "recent_messages": messages
            }

        # Messages a compresser
        to_compress = messages[:-keep_recent]
        recent = messages[-keep_recent:]

        # Generer un resume
        history_text = "\n".join([
            f"{m.role}: {m.content}" for m in to_compress
        ])

        prompt = f"""
        Resume cette conversation en conservant :
        - Les informations cles echangees
        - Les decisions ou engagements pris
        - Le contexte important pour la suite

        Conversation :
        {history_text}

        Resume concis (max 200 mots) :
        """

        summary = await self.llm.generate(prompt, temperature=0)

        return {
            "summary": summary,
            "recent_messages": recent
        }

Query Rewriting avec contexte

Resolution des co-references

DEVELOPERpython
class QueryRewriter:
    def __init__(self, llm):
        self.llm = llm

    async def rewrite(
        self,
        query: str,
        conversation_context: str
    ) -> str:
        """
        Reecrit la requete en resolvant les co-references
        """
        prompt = f"""
        Tu dois reecrire la question de l'utilisateur pour la rendre autonome,
        en resolvant les pronoms et references au contexte.

        Historique de conversation :
        {conversation_context}

        Derniere question : {query}

        Question reecrite (autonome, sans pronoms ambigus) :
        """

        rewritten = await self.llm.generate(prompt, temperature=0)
        return rewritten.strip()

    async def extract_search_queries(
        self,
        query: str,
        context: str
    ) -> List[str]:
        """
        Genere plusieurs requetes de recherche optimisees
        """
        prompt = f"""
        A partir de cette conversation et question, genere 2-3 requetes
        de recherche optimisees pour trouver les informations pertinentes.

        Contexte :
        {context}

        Question : {query}

        Requetes de recherche (une par ligne) :
        """

        result = await self.llm.generate(prompt, temperature=0.3)
        queries = [q.strip() for q in result.strip().split("\n") if q.strip()]

        return queries[:3]

Exemple d'application

DEVELOPERpython
# Avant rewriting
context = """
Utilisateur: Je cherche un laptop pour le gaming
Assistant: Je vous recommande le ASUS ROG Strix G15, excellent rapport qualite-prix.
Utilisateur: Il a combien de RAM ?
"""

query = "Et la carte graphique ?"

rewritten = await rewriter.rewrite(query, context)
# "Quelle est la carte graphique du ASUS ROG Strix G15 ?"

Memoire long terme

Stockage du profil utilisateur

DEVELOPERpython
from datetime import datetime, timedelta

class UserMemory:
    def __init__(self, db):
        self.db = db

    async def update_preference(
        self,
        user_id: str,
        key: str,
        value: any,
        confidence: float = 1.0
    ):
        """
        Met a jour une preference utilisateur
        """
        await self.db.upsert(
            "user_preferences",
            {
                "user_id": user_id,
                "key": key,
                "value": value,
                "confidence": confidence,
                "last_updated": datetime.now()
            },
            conflict_keys=["user_id", "key"]
        )

    async def get_preferences(self, user_id: str) -> dict:
        """
        Recupere toutes les preferences d'un utilisateur
        """
        prefs = await self.db.find(
            "user_preferences",
            {"user_id": user_id}
        )

        return {p["key"]: p["value"] for p in prefs}

    async def log_interaction(
        self,
        user_id: str,
        query: str,
        response: str,
        topic: str,
        satisfaction: float = None
    ):
        """
        Log une interaction pour analyse future
        """
        await self.db.insert("user_interactions", {
            "user_id": user_id,
            "query": query,
            "response": response,
            "topic": topic,
            "satisfaction": satisfaction,
            "timestamp": datetime.now()
        })

    async def extract_preferences_from_history(
        self,
        user_id: str,
        llm
    ) -> dict:
        """
        Extrait des preferences des interactions passees
        """
        # Recuperer les dernieres interactions
        interactions = await self.db.find(
            "user_interactions",
            {
                "user_id": user_id,
                "timestamp": {"$gte": datetime.now() - timedelta(days=30)}
            },
            limit=50
        )

        if not interactions:
            return {}

        history = "\n".join([
            f"Q: {i['query']}\nA: {i['response']}"
            for i in interactions
        ])

        prompt = f"""
        Analyse cet historique de conversation et extrait les preferences
        et caracteristiques de l'utilisateur.

        Historique :
        {history}

        Extrais en JSON :
        - preferred_language: langue preferee
        - technical_level: debutant/intermediaire/expert
        - topics_of_interest: liste des sujets frequents
        - communication_style: formel/informel
        - preferences: autres preferences detectees

        JSON :
        """

        result = await llm.generate(prompt, temperature=0)
        return self._parse_json(result)

Personnalisation des reponses

DEVELOPERpython
class PersonalizedRAG:
    def __init__(self, rag_pipeline, user_memory, llm):
        self.rag = rag_pipeline
        self.memory = user_memory
        self.llm = llm

    async def query(
        self,
        user_id: str,
        conversation_id: str,
        query: str
    ) -> dict:
        """
        Requete RAG personnalisee
        """
        # 1. Recuperer le profil utilisateur
        preferences = await self.memory.get_preferences(user_id)

        # 2. Recuperer le contexte de conversation
        conv_context = await self.conv_manager.get_context(conversation_id)

        # 3. Reecrire la requete avec contexte
        rewritten_query = await self.query_rewriter.rewrite(query, conv_context)

        # 4. Executer le RAG standard
        rag_result = await self.rag.query(rewritten_query)

        # 5. Personnaliser la reponse
        personalized = await self._personalize_response(
            query=query,
            rag_response=rag_result["answer"],
            preferences=preferences,
            context=conv_context
        )

        return {
            "answer": personalized,
            "sources": rag_result["sources"],
            "original_query": query,
            "rewritten_query": rewritten_query
        }

    async def _personalize_response(
        self,
        query: str,
        rag_response: str,
        preferences: dict,
        context: str
    ) -> str:
        """
        Adapte la reponse aux preferences utilisateur
        """
        tech_level = preferences.get("technical_level", "intermediaire")
        style = preferences.get("communication_style", "professionnel")

        prompt = f"""
        Adapte cette reponse selon le profil utilisateur.

        Profil :
        - Niveau technique : {tech_level}
        - Style prefere : {style}

        Contexte de conversation :
        {context}

        Question : {query}

        Reponse originale :
        {rag_response}

        Reponse adaptee :
        """

        return await self.llm.generate(prompt, temperature=0.3)

Memoire semantique partagee

Apprentissage des conversations

DEVELOPERpython
class SemanticMemory:
    def __init__(self, vector_db, embedder, llm):
        self.vector_db = vector_db
        self.embedder = embedder
        self.llm = llm

    async def learn_from_conversation(
        self,
        conversation: Conversation
    ):
        """
        Extrait et stocke les faits appris d'une conversation
        """
        # Extraire les faits
        facts = await self._extract_facts(conversation)

        for fact in facts:
            # Verifier si le fait existe deja
            existing = await self._find_similar_fact(fact)

            if existing:
                # Mettre a jour la confiance
                await self._update_fact_confidence(existing, fact)
            else:
                # Stocker le nouveau fait
                await self._store_fact(fact)

    async def _extract_facts(self, conversation: Conversation) -> List[dict]:
        """
        Extrait les faits factuels d'une conversation
        """
        messages_text = "\n".join([
            f"{m.role}: {m.content}" for m in conversation.messages
        ])

        prompt = f"""
        Extrais les faits nouveaux appris de cette conversation.
        Un fait doit etre :
        - Factuel et verifiable
        - Utile pour de futures conversations
        - Nouveau (pas une info deja connue)

        Conversation :
        {messages_text}

        Faits (format JSON array) :
        [
          {{"fact": "...", "category": "...", "confidence": 0.9}},
          ...
        ]
        """

        result = await self.llm.generate(prompt, temperature=0)
        return self._parse_json(result)

    async def recall(
        self,
        query: str,
        top_k: int = 5
    ) -> List[dict]:
        """
        Recupere les faits pertinents pour une requete
        """
        query_embedding = self.embedder.encode(query)

        results = await self.vector_db.search(
            collection="semantic_memory",
            query_vector=query_embedding,
            limit=top_k
        )

        return [r.payload for r in results]

Gestion des sessions

Multi-device et continuite

DEVELOPERpython
import redis
import json

class SessionManager:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.session_ttl = 3600 * 24  # 24 heures

    async def create_session(
        self,
        user_id: str,
        device_id: str = None
    ) -> str:
        """
        Cree une nouvelle session
        """
        session_id = f"session_{user_id}_{datetime.now().timestamp()}"

        session_data = {
            "user_id": user_id,
            "device_id": device_id,
            "created_at": datetime.now().isoformat(),
            "messages": [],
            "context": {}
        }

        await self.redis.setex(
            session_id,
            self.session_ttl,
            json.dumps(session_data)
        )

        return session_id

    async def resume_or_create(
        self,
        user_id: str,
        device_id: str = None
    ) -> str:
        """
        Reprend la derniere session ou en cree une nouvelle
        """
        # Chercher une session active recente
        pattern = f"session_{user_id}_*"
        keys = await self.redis.keys(pattern)

        if keys:
            # Trier par date de creation
            sessions = []
            for key in keys:
                data = json.loads(await self.redis.get(key))
                sessions.append((key, data))

            # Prendre la plus recente
            sessions.sort(key=lambda x: x[1]["created_at"], reverse=True)
            return sessions[0][0]

        # Creer une nouvelle session
        return await self.create_session(user_id, device_id)

    async def merge_sessions(
        self,
        session_ids: List[str]
    ) -> str:
        """
        Fusionne plusieurs sessions en une seule
        """
        all_messages = []

        for session_id in session_ids:
            data = json.loads(await self.redis.get(session_id))
            all_messages.extend(data.get("messages", []))

        # Trier par timestamp
        all_messages.sort(key=lambda x: x.get("timestamp", ""))

        # Creer une session fusionnee
        merged_data = {
            "messages": all_messages,
            "merged_from": session_ids,
            "created_at": datetime.now().isoformat()
        }

        merged_id = f"session_merged_{datetime.now().timestamp()}"
        await self.redis.setex(
            merged_id,
            self.session_ttl,
            json.dumps(merged_data)
        )

        return merged_id

Metriques et evaluation

Qualite conversationnelle

DEVELOPERpython
class ConversationMetrics:
    def __init__(self, db):
        self.db = db

    async def calculate_metrics(
        self,
        conversation_id: str
    ) -> dict:
        """
        Calcule les metriques d'une conversation
        """
        conversation = await self.db.get("conversations", conversation_id)

        return {
            # Longueur et engagement
            "total_turns": len(conversation["messages"]) // 2,
            "avg_user_message_length": self._avg_length(
                [m for m in conversation["messages"] if m["role"] == "user"]
            ),
            "avg_assistant_message_length": self._avg_length(
                [m for m in conversation["messages"] if m["role"] == "assistant"]
            ),

            # Resolution
            "ended_naturally": self._check_natural_end(conversation),
            "required_clarification": self._count_clarifications(conversation),

            # Coherence
            "topic_coherence": await self._calculate_topic_coherence(conversation),
            "context_usage": self._measure_context_usage(conversation)
        }

    async def _calculate_topic_coherence(self, conversation: dict) -> float:
        """
        Mesure si la conversation reste coherente dans son sujet
        """
        # Embedding de chaque message
        messages = conversation["messages"]
        embeddings = [
            self.embedder.encode(m["content"])
            for m in messages
        ]

        # Calculer la similarite entre messages consecutifs
        similarities = []
        for i in range(1, len(embeddings)):
            sim = cosine_similarity(embeddings[i-1], embeddings[i])
            similarities.append(sim)

        return sum(similarities) / len(similarities) if similarities else 0

Bonnes pratiques

1. Limiter la taille du contexte

Ne pas surcharger le prompt avec trop d'historique. Compresser ou resumer.

2. Gerer les changements de sujet

Detecter quand l'utilisateur change de sujet pour ne pas polluer le contexte.

3. Respecter la vie privee

Permettre aux utilisateurs de supprimer leur historique et preferences.

4. Fallback gracieux

Si la memoire n'est pas disponible, le systeme doit fonctionner en mode sans etat.

Pour aller plus loin

• Fondamentaux du Retrieval - Optimiser la recherche
• Generation LLM - Ameliorer les reponses
• Prompt Engineering RAG - Optimiser les prompts

RAG Conversationnel avec Ailog

Implementer une memoire conversationnelle robuste est complexe. Avec Ailog, beneficiez de ces fonctionnalites natives :

• Historique de session automatique avec compression
• Memoire utilisateur persistante et personnalisation
• Query rewriting pour resoudre les co-references
• Multi-device avec reprise de conversation
• Analytics conversationnelles pour mesurer l'engagement
• Conformite RGPD avec export et suppression des donnees

Testez Ailog gratuitement et deployez un assistant conversationnel intelligent.