Buffer Memory : Einfacher Gesprächsverlauf

Die Buffer Memory ist die einfachste und intuitivste Form des Gesprächsspeichers. Sie behält die letzten N Nachrichten des Gesprächs in einem Puffer (buffer) und injiziert sie direkt in den Kontext des LLM. Einfach, aber effektiv für kurze bis mittellange Gespräche; oft die erste Wahl, um einen konversationellen RAG-Chatbot zu implementieren.

Qu'est-ce que la Buffer Memory ?

Funktionsweise

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:
    """
    Einfacher Puffer-Speicher für Gespräche
    Behält die letzten N Nachrichten in einer 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:
        """Fügt eine Nachricht zum Puffer hinzu"""
        self.messages.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })

    def get_history(self) -> List[Dict]:
        """Gibt alle Nachrichten zurück"""
        return list(self.messages)

    def get_context_string(self) -> str:
        """Formatiert den Verlauf zur Injektion in das Prompt"""
        return "\n".join([
            f"{m['role'].capitalize()}: {m['content']}"
            for m in self.messages
        ])

    def clear(self) -> None:
        """Leert den Puffer"""
        self.messages.clear()

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

Integration avec un pipeline RAG

DEVELOPERpython
class ConversationalRAG:
    """
    Komplette RAG-Pipeline mit 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:
        """Führt eine RAG-Anfrage mit Gesprächskontext aus"""
        # 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.