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.

Was ist Buffer Memory?

Funktionsweise

Buffer Memory funktioniert wie eine FIFO-Warteschlange (First In, First Out) für Nachrichten:

┌─────────────────────────────────────────────────────────────┐
│                      BUFFER MEMORY                           │
├─────────────────────────────────────────────────────────────┤
│  Message 1: [User] Hallo, ich suche einen Computer           │
│  Message 2: [AI] Hallo! Wofür möchten Sie ihn nutzen?        │
│  Message 3: [User] Hauptsächlich Gaming                      │
│  Message 4: [AI] Ich empfehle Ihnen eine Konfig mit RTX...   │
│  Message 5: [User] Und wie sieht es mit dem Budget aus?      │
│  Message 6: [AI] Rechnen Sie mit 1200 bis 2000 Euro...       │
│  ...                                                         │
└─────────────────────────────────────────────────────────────┘
                            │
                            ▼
              ┌─────────────────────────┐
              │   Einfügung in den      │
              │   Prompt des LLM        │
              └─────────────────────────┘

Wenn der Puffer sein Limit erreicht (in Anzahl der Nachrichten oder Tokens), werden die ältesten Nachrichten automatisch gelöscht, um Platz für neue zu schaffen.

Vergleich der Speicheransätze

Schlüsselstatistik: Ein Buffer Memory mit 10 Gesprächsrunden verbraucht durchschnittlich 2000-3000 Tokens, also etwa 15-20% des GPT-4-Kontextfensters. Für 80% der Kundensupport-Anwendungsfälle reichen 6-8 Runden aus.

Grundlegende Implementierung mit Python

Buffer Memory von Grund auf

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 in eine RAG-Pipeline

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. Nutzernachricht hinzufügen
        self.memory.add_message("user", user_message)

        # 2. Relevante Dokumente abrufen
        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. Prompt mit dem Verlauf erstellen
        history = self.memory.get_context_string()

        prompt = f"""Du bist ein hilfreicher Assistent. Nutze die bereitgestellten Dokumente
und den Gesprächsverlauf, um zu antworten.

Gesprächsverlauf:
{history}

Relevante Dokumente:
{context}

Aktuelle Frage: {user_message}

Antworte präzise und berücksichtige dabei den Kontext des Gesprächs."""

        # 4. Antwort generieren
        response = self.llm.invoke(prompt)

        # 5. Antwort speichern
        self.memory.add_message("assistant", response)

        return response

    def reset(self) -> None:
        """Setzt das Gespräch zurück"""
        self.memory.clear()

Intelligentes Token-Management

Das Hauptproblem von Buffer Memory ist der Token-Verbrauch. Hier ist eine optimierte Version:

DEVELOPERpython
import tiktoken

class TokenAwareBufferMemory:
    """
    Buffer Memory mit Token-Verwaltung
    Löscht alte Nachrichten, wenn das Limit erreicht ist
    """
    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:
        """Fügt eine Nachricht hinzu und passt bei Bedarf an"""
        self.messages.append({"role": role, "content": content})
        self._trim_to_token_limit()

    def _count_tokens(self) -> int:
        """Zählt die Gesamtzahl der Tokens"""
        text = " ".join([m["content"] for m in self.messages])
        return len(self.encoding.encode(text))

    def _trim_to_token_limit(self) -> None:
        """Löscht die ältesten Nachrichten, wenn das Limit überschritten ist"""
        while self._count_tokens() > self.max_tokens and len(self.messages) > 2:
            # Immer mindestens den letzten Austausch behalten
            self.messages.pop(0)

    def get_messages_for_api(self) -> List[Dict]:
        """Gibt die Nachrichten im OpenAI-API-Format zurück"""
        return [
            {"role": m["role"], "content": m["content"]}
            for m in self.messages
        ]

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

Implementierung mit LangChain

LangChain bietet gebrauchsfertige Implementierungen:

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

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

# Mit einem gleitenden Fenster (behält die letzten k Austausche)
from langchain.memory import ConversationBufferWindowMemory

memory_window = ConversationBufferWindowMemory(
    k=5,  # Behält die letzten 5 Austausche
    memory_key="history",
    return_messages=True
)

# Integration mit einer konversationellen Chain
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# Verwendung
response = conversation.predict(input="Ich suche einen Gaming-Laptop")
print(response)

response = conversation.predict(input="Budget maximal 1500 Euro")
print(response)  # Versteht den Kontext des Laptops

Integration mit ConversationalRetrievalChain

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

# Konfiguration des Vector Stores
vectorstore = Qdrant(
    client=qdrant_client,
    collection_name="products",
    embeddings=OpenAIEmbeddings()
)

# Memory für konversationelles RAG
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="answer"
)

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

# Natürliche Konversation
result = rag_chain({"question": "Wie lautet Ihre Rückgaberichtlinie?"})
result2 = rag_chain({"question": "Und bei Elektronikprodukten?"})
# Versteht, dass sich "Und bei" auf die Rückgaberichtlinie bezieht

Implementierung mit LlamaIndex

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

# Index erstellen
index = VectorStoreIndex.from_documents(documents)

# Buffer Memory mit Token-Limit
memory = ChatMemoryBuffer.from_defaults(
    token_limit=3000
)

# Chat Engine mit Memory
chat_engine = CondensePlusContextChatEngine.from_defaults(
    retriever=index.as_retriever(similarity_top_k=4),
    memory=memory,
    llm=OpenAI(model="gpt-4"),
    context_prompt=(
        "Relevanter Kontext:\n{context_str}\n\n"
        "Antworte unter Verwendung dieses Kontexts und des Verlaufs."
    ),
    verbose=True
)

# Konversation
response1 = chat_engine.chat("Wie sind Ihre Öffnungszeiten?")
response2 = chat_engine.chat("Und am Wochenende?")  # Versteht den Kontext

Best Practices und Optimierungen

1. Komprimierung langer Nachrichten

DEVELOPERpython
def compress_message(content: str, max_chars: int = 500) -> str:
    """Kürzt zu lange Nachrichten"""
    if len(content) <= max_chars:
        return content

    # Anfang und Ende behalten
    half = max_chars // 2
    return f"{content[:half]}... [gekürzt] ...{content[-half:]}"

2. Speicherung und Persistenz

DEVELOPERpython
import json
from pathlib import Path

class PersistentBufferMemory(ConversationBufferMemory):
    """Buffer mit automatischer Speicherung"""

    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 und Metriken

DEVELOPERpython
class MonitoredBufferMemory(TokenAwareBufferMemory):
    """Buffer mit Monitoring-Metriken"""

    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()
        )

Wann Buffer Memory verwenden?

Ideale Anwendungsfälle

• Kurze Gespräche: Weniger als 10 Austausche
• Einfache Folgefragen: "Und bei Zubehör?", "Wie viel kostet das?"
• Kundensupport Level 1: FAQ mit aktuellem Kontext
• FAQ-Chatbots: Unabhängige Fragen mit wenig Nachfragen

Wann zu einer anderen Lösung wechseln

Weiterführende Ressourcen

• Summary Memory - Für lange Gespräche
• Entity Memory - Zum Merken von Entitäten
• Konversationelles RAG - Überblick

Buffer Memory mit Ailog

Die Implementierung und Optimierung eines Buffer Memory erfordert Zeit. Mit Ailog profitieren Sie von einer schlüsselfertigen Gesprächsspeicherverwaltung:

• Optimierter Buffer Memory mit automatischer Token-Verwaltung
• Adaptives Fenster je nach Gesprächskomplexität
• Automatische Persistenz mit Sitzungswiederaufnahme
• Analytics zur Speichernutzung und Gesprächsqualität
• Konfigurationsfrei: Funktioniert sofort einsatzbereit

Ailog kostenlos testen und einen Chatbot mit Gedächtnis in 3 Minuten bereitstellen.