Buffer Memory : Historique de conversation simple

La buffer memory est la forme la plus simple et la plus intuitive de memoire conversationnelle. Elle conserve les N derniers messages de la conversation dans un tampon (buffer) et les injecte directement dans le contexte du LLM. Simple mais efficace pour les conversations courtes a moyennes, elle constitue souvent le premier choix pour implementer un chatbot RAG conversationnel.

Qu'est-ce que la Buffer Memory ?

Principe de fonctionnement

Le Buffer Memory fonctionne comme une file d'attente FIFO (First In, First Out) pour les messages :

┌─────────────────────────────────────────────────────────────┐
│                      BUFFER MEMORY                           │
├─────────────────────────────────────────────────────────────┤
│  Message 1: [User] Bonjour, je cherche un ordinateur        │
│  Message 2: [AI] Bonjour ! Quel usage prevoyez-vous ?       │
│  Message 3: [User] Gaming principalement                     │
│  Message 4: [AI] Je vous recommande une config avec RTX...  │
│  Message 5: [User] Et pour le budget ?                       │
│  Message 6: [AI] Comptez entre 1200 et 2000 euros...        │
│  ...                                                         │
└─────────────────────────────────────────────────────────────┘
                            │
                            ▼
              ┌─────────────────────────┐
              │   Injection dans le     │
              │   prompt du LLM         │
              └─────────────────────────┘

Quand le buffer atteint sa limite (en nombre de messages ou en tokens), les messages les plus anciens sont automatiquement supprimes pour faire de la place aux nouveaux.

Comparaison des approches memoire

Statistique cle : Un Buffer Memory de 10 tours de conversation consomme en moyenne 2000-3000 tokens, soit environ 15-20% de la fenetre de contexte d'un GPT-4. Pour 80% des cas d'usage en support client, 6-8 tours suffisent.

Implementation basique avec Python

Buffer Memory from scratch

DEVELOPERpython
from collections import deque
from typing import List, Dict
from datetime import datetime

class ConversationBufferMemory:
    """
    Memoire tampon simple pour conversations
    Garde les N derniers messages dans une deque
    """
    def __init__(self, max_messages: int = 10):
        self.messages: deque = deque(maxlen=max_messages)
        self.created_at = datetime.now()

    def add_message(self, role: str, content: str) -> None:
        """Ajoute un message au buffer"""
        self.messages.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })

    def get_history(self) -> List[Dict]:
        """Retourne tous les messages"""
        return list(self.messages)

    def get_context_string(self) -> str:
        """Formate l'historique pour injection dans le prompt"""
        return "\n".join([
            f"{m['role'].capitalize()}: {m['content']}"
            for m in self.messages
        ])

    def clear(self) -> None:
        """Vide le buffer"""
        self.messages.clear()

    def __len__(self) -> int:
        return len(self.messages)

Integration avec un pipeline RAG

DEVELOPERpython
class ConversationalRAG:
    """
    Pipeline RAG complet avec Buffer Memory
    """
    def __init__(self, vector_store, llm, max_history: int = 6):
        self.vector_store = vector_store
        self.llm = llm
        self.memory = ConversationBufferMemory(max_messages=max_history)

    def query(self, user_message: str) -> str:
        """Execute une requete RAG avec contexte conversationnel"""
        # 1. Ajouter le message utilisateur
        self.memory.add_message("user", user_message)

        # 2. Recuperer les documents pertinents
        docs = self.vector_store.similarity_search(user_message, k=3)
        context = "\n\n".join([
            f"Document {i+1}:\n{d.page_content}"
            for i, d in enumerate(docs)
        ])

        # 3. Construire le prompt avec l'historique
        history = self.memory.get_context_string()

        prompt = f"""Tu es un assistant utile. Utilise les documents fournis
et l'historique de conversation pour repondre.

Historique de conversation:
{history}

Documents pertinents:
{context}

Question actuelle: {user_message}

Reponds de maniere precise en tenant compte du contexte de la conversation."""

        # 4. Generer la reponse
        response = self.llm.invoke(prompt)

        # 5. Sauvegarder la reponse
        self.memory.add_message("assistant", response)

        return response

    def reset(self) -> None:
        """Reinitialise la conversation"""
        self.memory.clear()

Gestion intelligente des tokens

Le probleme principal du Buffer Memory est la consommation de tokens. Voici une version optimisee :

DEVELOPERpython
import tiktoken

class TokenAwareBufferMemory:
    """
    Buffer Memory avec gestion des tokens
    Supprime les anciens messages quand la limite est atteinte
    """
    def __init__(self, max_tokens: int = 2000, model: str = "gpt-4o"):
        self.max_tokens = max_tokens
        self.encoding = tiktoken.encoding_for_model(model)
        self.messages: List[Dict] = []

    def add_message(self, role: str, content: str) -> None:
        """Ajoute un message et ajuste si necessaire"""
        self.messages.append({"role": role, "content": content})
        self._trim_to_token_limit()

    def _count_tokens(self) -> int:
        """Compte le nombre total de tokens"""
        text = " ".join([m["content"] for m in self.messages])
        return len(self.encoding.encode(text))

    def _trim_to_token_limit(self) -> None:
        """Supprime les messages les plus anciens si limite depassee"""
        while self._count_tokens() > self.max_tokens and len(self.messages) > 2:
            # Toujours garder au moins le dernier echange
            self.messages.pop(0)

    def get_messages_for_api(self) -> List[Dict]:
        """Retourne les messages au format OpenAI API"""
        return [
            {"role": m["role"], "content": m["content"]}
            for m in self.messages
        ]

    @property
    def token_usage(self) -> dict:
        """Statistiques d'utilisation"""
        current = self._count_tokens()
        return {
            "current_tokens": current,
            "max_tokens": self.max_tokens,
            "utilization": current / self.max_tokens,
            "messages_count": len(self.messages)
        }

Implementation avec LangChain

LangChain fournit des implementations pretes a l'emploi :

DEVELOPERpython
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI

# Buffer Memory basique
memory = ConversationBufferMemory(
    memory_key="history",
    return_messages=True
)

# Avec une fenetre glissante (garde les k derniers echanges)
from langchain.memory import ConversationBufferWindowMemory

memory_window = ConversationBufferWindowMemory(
    k=5,  # Garde les 5 derniers echanges
    memory_key="history",
    return_messages=True
)

# Integration avec une chaine conversationnelle
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# Utilisation
response = conversation.predict(input="Je cherche un laptop gaming")
print(response)

response = conversation.predict(input="Budget 1500 euros max")
print(response)  # Comprend le contexte du laptop

Integration avec ConversationalRetrievalChain

DEVELOPERpython
from langchain.chains import ConversationalRetrievalChain
from langchain_community.vectorstores import Qdrant

# Configuration du vector store
vectorstore = Qdrant(
    client=qdrant_client,
    collection_name="products",
    embeddings=OpenAIEmbeddings()
)

# Memory pour RAG conversationnel
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="answer"
)

# Chaine RAG avec memoire
rag_chain = ConversationalRetrievalChain.from_llm(
    llm=ChatOpenAI(model="gpt-4"),
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    memory=memory,
    return_source_documents=True
)

# Conversation naturelle
result = rag_chain({"question": "Quelle est votre politique de retour ?"})
result2 = rag_chain({"question": "Et pour les produits electroniques ?"})
# Comprend que "Et pour" fait reference a la politique de retour

Implementation avec LlamaIndex

DEVELOPERpython
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.chat_engine import CondensePlusContextChatEngine
from llama_index.core import VectorStoreIndex

# Creer l'index
index = VectorStoreIndex.from_documents(documents)

# Buffer memory avec limite de tokens
memory = ChatMemoryBuffer.from_defaults(
    token_limit=3000
)

# Chat engine avec memoire
chat_engine = CondensePlusContextChatEngine.from_defaults(
    retriever=index.as_retriever(similarity_top_k=4),
    memory=memory,
    llm=OpenAI(model="gpt-4"),
    context_prompt=(
        "Contexte pertinent:\n{context_str}\n\n"
        "Reponds en utilisant ce contexte et l'historique."
    ),
    verbose=True
)

# Conversation
response1 = chat_engine.chat("Quels sont vos horaires ?")
response2 = chat_engine.chat("Et le week-end ?")  # Comprend le contexte

Bonnes pratiques et optimisations

1. Compression des messages longs

DEVELOPERpython
def compress_message(content: str, max_chars: int = 500) -> str:
    """Tronque les messages trop longs"""
    if len(content) <= max_chars:
        return content

    # Garder le debut et la fin
    half = max_chars // 2
    return f"{content[:half]}... [tronque] ...{content[-half:]}"

2. Sauvegarde et persistence

DEVELOPERpython
import json
from pathlib import Path

class PersistentBufferMemory(ConversationBufferMemory):
    """Buffer avec sauvegarde automatique"""

    def __init__(self, session_id: str, storage_dir: str = "./sessions"):
        super().__init__()
        self.session_id = session_id
        self.storage_path = Path(storage_dir) / f"{session_id}.json"
        self._load()

    def _load(self) -> None:
        if self.storage_path.exists():
            with open(self.storage_path, "r") as f:
                data = json.load(f)
                for msg in data.get("messages", []):
                    self.messages.append(msg)

    def save(self) -> None:
        self.storage_path.parent.mkdir(exist_ok=True)
        with open(self.storage_path, "w") as f:
            json.dump({"messages": list(self.messages)}, f)

    def add_message(self, role: str, content: str) -> None:
        super().add_message(role, content)
        self.save()

3. Monitoring et metriques

DEVELOPERpython
class MonitoredBufferMemory(TokenAwareBufferMemory):
    """Buffer avec metriques de monitoring"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.stats = {
            "total_messages": 0,
            "messages_evicted": 0,
            "peak_tokens": 0
        }

    def add_message(self, role: str, content: str) -> None:
        count_before = len(self.messages)
        super().add_message(role, content)

        self.stats["total_messages"] += 1
        self.stats["messages_evicted"] += max(0, count_before + 1 - len(self.messages))
        self.stats["peak_tokens"] = max(
            self.stats["peak_tokens"],
            self._count_tokens()
        )

Quand utiliser Buffer Memory ?

Cas d'usage ideaux

• Conversations courtes : Moins de 10 echanges
• Questions de suivi simples : "Et pour les accessoires ?", "Combien ca coute ?"
• Support client niveau 1 : FAQ avec contexte recent
• Chatbots FAQ : Questions independantes avec peu de suivi

Quand passer a autre chose

Pour aller plus loin

• Summary Memory - Pour les conversations longues
• Entity Memory - Pour retenir les entites
• RAG Conversationnel - Vue d'ensemble

Buffer Memory avec Ailog

Implementer et optimiser un Buffer Memory demande du temps. Avec Ailog, beneficiez d'une gestion de memoire conversationnelle cle en main :

• Buffer Memory optimise avec gestion automatique des tokens
• Fenetre adaptative selon la complexite de la conversation
• Persistence automatique avec reprise de session
• Analytics sur l'utilisation memoire et la qualite des conversations
• Zero configuration : ca marche out of the box

Testez Ailog gratuitement et deployez un chatbot avec memoire en 3 minutes.