Summary Memory: Zusammenfassen für langfristiges Merken

Summary Memory ist eine fortgeschrittene Technik des Gesprächsspeichers, die den Austauschverlauf zu aufeinanderfolgenden Zusammenfassungen verdichtet. Sie ermöglicht es, den Kontext über Dutzende, ja sogar Hunderte von Austauschen beizubehalten, ohne das Token-Budget zu sprengen. Dieser Leitfaden erklärt, wie man sie effektiv implementiert.

Warum Summary Memory?

Das Problem langer Gespräche

Bei einem klassischen Buffer Memory verbraucht jede Nachricht Tokens. Nach 15-20 Austauschen stößt man schnell an die Grenzen:

┌─────────────────────────────────────────────────────────────┐
│               PROBLEM DES BUFFER MEMORY                      │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  Austausch 1-5:    ~800 Tokens    ██░░░░░░░░                │
│  Austausch 1-10:   ~1600 Tokens   ████░░░░░░                │
│  Austausch 1-15:   ~2400 Tokens   ██████░░░░                │
│  Austausch 1-20:   ~3200 Tokens   ████████░░   GRENZE!      │
│  Austausch 1-30:   ~4800 Tokens   ████████████ ÜBERSCHRITTEN!│
│                                                              │
│  GPT-4-Fenster: ~8000 nutzbare Tokens                        │
│  (Rest des Kontexts = RAG-Dokumente + Prompt)                │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Die Lösung: Progressive Zusammenfassung

┌─────────────────────────────────────────────────────────────┐
│               SUMMARY MEMORY                                 │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌────────────────────────────────────────────────────┐     │
│  │ ZUSAMMENFASSUNG (200-300 Tokens)                    │     │
│  │ "Der Nutzer sucht ein Gaming-Laptop mit Budget      │     │
│  │  1500€. Er hat das MSI abgelehnt, da zu schwer.     │     │
│  │  Interessiert an ASUS ROG, Fragen zur Garantie."    │     │
│  └────────────────────────────────────────────────────┘     │
│                            +                                 │
│  ┌────────────────────────────────────────────────────┐     │
│  │ AKTUELLE NACHRICHTEN (letzte 4-6 Austausche)        │     │
│  │ User: "Und die Grafikkarte?"                        │     │
│  │ AI: "Das ASUS ROG hat eine RTX 4060..."             │     │
│  │ User: "Ist das mit meinem Bildschirm kompatibel?"   │     │
│  │ AI: "Ja, der HDMI-2.1-Anschluss..."                 │     │
│  └────────────────────────────────────────────────────┘     │
│                                                              │
│  GESAMT: ~600 Tokens (vs. 3000+ bei Buffer)                 │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Schlüsselstatistik: Summary Memory reduziert den Token-Verbrauch um 70-80% bei langen Gesprächen und bewahrt dabei 90%+ des relevanten Kontexts.

Vergleich der Ansätze

Grundlegende Implementierung

Summary Memory von Grund auf

DEVELOPERpython
from typing import List, Dict
from datetime import datetime

class SummaryMemory:
    """
    Gesprächsspeicher mit progressiver Zusammenfassung
    """
    def __init__(
        self,
        llm,
        max_messages_before_summary: int = 6,
        summary_max_tokens: int = 300
    ):
        self.llm = llm
        self.max_messages = max_messages_before_summary
        self.summary_max_tokens = summary_max_tokens
        self.summary: str = ""
        self.recent_messages: List[Dict] = []

    def add_message(self, role: str, content: str) -> None:
        """Fügt eine Nachricht hinzu und löst bei Bedarf eine Zusammenfassung aus"""
        self.recent_messages.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })

        # Bei zu vielen Nachrichten die alten zusammenfassen
        if len(self.recent_messages) > self.max_messages:
            self._summarize_old_messages()

    def _summarize_old_messages(self) -> None:
        """Fasst die ältesten Nachrichten zusammen"""
        # Aufteilen: alte zusammenfassen, aktuelle behalten
        split_point = len(self.recent_messages) // 2
        to_summarize = self.recent_messages[:split_point]
        self.recent_messages = self.recent_messages[split_point:]

        # Zu formatierende Konversation
        conversation = "\n".join([
            f"{m['role'].capitalize()}: {m['content']}"
            for m in to_summarize
        ])

        # Neue Zusammenfassung generieren
        prompt = f"""Aktualisiere eine Gesprächszusammenfassung.

Bestehende Zusammenfassung:
{self.summary if self.summary else "Keine vorherige Zusammenfassung."}

Neu zu integrierende Austausche:
{conversation}

Anweisungen:
- Integriere die wichtigen Informationen der neuen Austausche in die Zusammenfassung
- Behalte getroffene Entscheidungen, geäußerte Präferenzen, gelöste Probleme
- Entferne redundante oder veraltete Details
- Die Zusammenfassung muss prägnant sein (max. {self.summary_max_tokens} Tokens)
- Schreibe in der dritten Person ("Der Nutzer...", "Der Assistent...")

Aktualisierte Zusammenfassung:"""

        self.summary = self.llm.invoke(prompt)

    def get_context(self) -> str:
        """Gibt den vollständigen Kontext für den Prompt zurück"""
        parts = []

        if self.summary:
            parts.append(f"Zusammenfassung des vorherigen Gesprächs:\n{self.summary}")

        if self.recent_messages:
            recent = "\n".join([
                f"{m['role'].capitalize()}: {m['content']}"
                for m in self.recent_messages
            ])
            parts.append(f"Aktuelle Austausche:\n{recent}")

        return "\n\n".join(parts)

    def clear(self) -> None:
        """Setzt den Speicher zurück"""
        self.summary = ""
        self.recent_messages = []

Integration mit RAG

DEVELOPERpython
class RAGWithSummaryMemory:
    """
    Vollständige RAG-Pipeline mit Summary Memory
    """
    def __init__(self, vector_store, llm):
        self.vector_store = vector_store
        self.llm = llm
        self.memory = SummaryMemory(llm, max_messages_before_summary=6)

    def query(self, user_message: str) -> str:
        """Führt eine RAG-Abfrage mit zusammengefasstem Kontext 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)
        doc_context = "\n\n".join([
            f"Dokument {i+1}:\n{d.page_content}"
            for i, d in enumerate(docs)
        ])

        # 3. Prompt mit Speicher aufbauen
        memory_context = self.memory.get_context()

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

{memory_context}

Relevante Dokumente:
{doc_context}

Aktuelle Frage: {user_message}

Antworte präzise und kontextbezogen."""

        # 4. Antwort generieren und speichern
        response = self.llm.invoke(prompt)
        self.memory.add_message("assistant", response)

        return response

Fortgeschrittene Techniken

Inkrementelle Zusammenfassung (effizienter)

Anstatt die vollständige Zusammenfassung neu zu generieren, kann man sie inkrementell anreichern:

DEVELOPERpython
class IncrementalSummaryMemory(SummaryMemory):
    """
    Summary Memory mit inkrementeller Aktualisierung
    Schneller und tokensparsamer
    """

    def _summarize_old_messages(self) -> None:
        """Inkrementelle Aktualisierung der Zusammenfassung"""
        split_point = len(self.recent_messages) // 2
        to_summarize = self.recent_messages[:split_point]
        self.recent_messages = self.recent_messages[split_point:]

        conversation = "\n".join([
            f"{m['role'].capitalize()}: {m['content']}"
            for m in to_summarize
        ])

        # Optimierter Prompt für inkrementelle Aktualisierung
        if self.summary:
            prompt = f"""Aktuelle Zusammenfassung: {self.summary}

Neue Gesprächselemente:
{conversation}

Aktualisiere die Zusammenfassung, indem du:
1. die neuen wichtigen Informationen hinzufügst
2. veraltete Informationen entfernst
3. die Zusammenfassung prägnant hältst (max. 200 Wörter)

Aktualisierte Zusammenfassung:"""
        else:
            prompt = f"""Fasse dieses Gespräch zusammen und extrahiere:
- Das Hauptthema
- Getroffene Entscheidungen
- Präferenzen des Nutzers
- Offene Fragen

Gespräch:
{conversation}

Zusammenfassung (max. 200 Wörter):"""

        self.summary = self.llm.invoke(prompt)

Strukturierte Zusammenfassung

Für eine bessere Organisation nutzen Sie eine strukturierte Zusammenfassung:

DEVELOPERpython
import json

class StructuredSummaryMemory(SummaryMemory):
    """
    Summary Memory mit strukturierter Zusammenfassung in JSON
    Ermöglicht präziseren Zugriff auf Informationen
    """

    def __init__(self, llm, **kwargs):
        super().__init__(llm, **kwargs)
        self.structured_summary: Dict = {
            "main_topic": "",
            "user_preferences": [],
            "decisions_made": [],
            "pending_questions": [],
            "key_facts": []
        }

    def _summarize_old_messages(self) -> None:
        """Generiert eine strukturierte Zusammenfassung"""
        split_point = len(self.recent_messages) // 2
        to_summarize = self.recent_messages[:split_point]
        self.recent_messages = self.recent_messages[split_point:]

        conversation = "\n".join([
            f"{m['role'].capitalize()}: {m['content']}"
            for m in to_summarize
        ])

        prompt = f"""Analysiere dieses Gespräch und aktualisiere die strukturierte Zusammenfassung.

Aktuelle Zusammenfassung:
{json.dumps(self.structured_summary, ensure_ascii=False, indent=2)}

Neue Austausche:
{conversation}

Gib ein JSON mit folgender Struktur zurück:
{{
    "main_topic": "Hauptthema des Gesprächs",
    "user_preferences": ["Liste der geäußerten Präferenzen"],
    "decisions_made": ["Liste der getroffenen Entscheidungen"],
    "pending_questions": ["ungelöste Fragen"],
    "key_facts": ["wichtige zu merkende Fakten"]
}}

Aktualisiertes JSON:"""

        response = self.llm.invoke(prompt)
        try:
            self.structured_summary = json.loads(response)
        except json.JSONDecodeError:
            # Fallback auf Textzusammenfassung
            self.summary = response

    def get_context(self) -> str:
        """Formatiert die strukturierte Zusammenfassung für den Prompt"""
        parts = []

        if self.structured_summary.get("main_topic"):
            parts.append(f"Thema: {self.structured_summary['main_topic']}")

        if self.structured_summary.get("user_preferences"):
            prefs = ", ".join(self.structured_summary["user_preferences"])
            parts.append(f"Nutzerpräferenzen: {prefs}")

        if self.structured_summary.get("decisions_made"):
            decisions = ", ".join(self.structured_summary["decisions_made"])
            parts.append(f"Getroffene Entscheidungen: {decisions}")

        if self.structured_summary.get("pending_questions"):
            questions = ", ".join(self.structured_summary["pending_questions"])
            parts.append(f"Offene Fragen: {questions}")

        summary_text = "\n".join(parts) if parts else ""

        if self.recent_messages:
            recent = "\n".join([
                f"{m['role'].capitalize()}: {m['content']}"
                for m in self.recent_messages
            ])
            return f"Kontext:\n{summary_text}\n\nAktuelle Austausche:\n{recent}"

        return f"Kontext:\n{summary_text}"

Implementierung mit LangChain

LangChain bietet eine native Implementierung:

Hinweis: Seit LangChain 0.3 sind diese Memory-Klassen (ConversationSummaryMemory, ConversationSummaryBufferMemory, ConversationChain) veraltet und werden in LangChain 1.0 entfernt; der empfohlene Ansatz nutzt inzwischen LangGraph (Checkpointing). Das Prinzip bleibt gleich; das folgende Beispiel veranschaulicht die Logik.

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

# LLM für Generierung und für die Zusammenfassung
llm = ChatOpenAI(model="gpt-4", temperature=0.7)

# Native LangChain Summary Memory
memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="history",
    return_messages=True
)

# Gesprächskette
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# Nutzung - die Zusammenfassung erfolgt automatisch
response = conversation.predict(input="Ich suche ein Gaming-Laptop")
response = conversation.predict(input="Budget max. 1500 Euro")
response = conversation.predict(input="Ich habe schon das MSI angeschaut, aber zu schwer")
# ... nach mehreren Austauschen enthält der Speicher eine Zusammenfassung

# Zusammenfassung inspizieren
print(memory.buffer)  # Zeigt die aktuelle Zusammenfassung an

Kombination Buffer + Summary

Der beste Ansatz kombiniert beide:

DEVELOPERpython
from langchain.memory import ConversationSummaryBufferMemory

# Behält aktuelle Nachrichten UND eine Zusammenfassung der alten
memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=1000,  # Zusammenfassung bei Überschreitung von 1000 Tokens
    memory_key="history",
    return_messages=True
)

# Effizienter als reine Summary für den aktuellen Kontext

Implementierung mit LlamaIndex

DEVELOPERpython
from llama_index.core.memory import ChatSummaryMemoryBuffer
from llama_index.core.chat_engine import CondensePlusContextChatEngine
from llama_index.llms.openai import OpenAI

# LLM
llm = OpenAI(model="gpt-4", temperature=0.7)

# Memory mit automatischer Zusammenfassung
memory = ChatSummaryMemoryBuffer.from_defaults(
    llm=llm,
    token_limit=1500,  # Grenze vor Zusammenfassung
    summarize_prompt=(
        "Fasse das vorherige Gespräch zusammen und behalte:\n"
        "- Das Hauptthema\n"
        "- Getroffene Entscheidungen\n"
        "- Wichtigen Kontext\n\n"
        "Gespräch:\n{conversation}\n\n"
        "Zusammenfassung:"
    )
)

# Chat Engine mit Speicher
chat_engine = CondensePlusContextChatEngine.from_defaults(
    retriever=index.as_retriever(similarity_top_k=4),
    memory=memory,
    llm=llm
)

# Langes Gespräch - automatische Zusammenfassung
for question in user_questions:
    response = chat_engine.chat(question)
    print(response)

Best Practices

1. Den richtigen Zusammenfassungsschwellwert wählen

DEVELOPERpython
# Empfehlungen je nach Anwendungsfall
THRESHOLDS = {
    "support_client": 6,     # Häufige Zusammenfassung, aktueller Kontext wichtig
    "consultation": 10,      # Mehr Kontext vor Zusammenfassung
    "tutoriel": 4,          # Sehr häufige Zusammenfassung
    "conversation_libre": 8  # Ausgewogen
}

memory = SummaryMemory(
    llm=llm,
    max_messages_before_summary=THRESHOLDS["support_client"]
)

2. Qualität der Zusammenfassung validieren

DEVELOPERpython
class ValidatedSummaryMemory(SummaryMemory):
    """Summary Memory mit Validierung der Zusammenfassung"""

    def _summarize_old_messages(self) -> None:
        super()._summarize_old_messages()

        # Prüfen, dass die Zusammenfassung nicht leer oder zu kurz ist
        if len(self.summary.split()) < 20:
            # Mit einem expliziteren Prompt neu generieren
            self._force_detailed_summary()

    def _force_detailed_summary(self) -> None:
        """Generiert bei Bedarf eine ausführlichere Zusammenfassung"""
        prompt = f"""Die vorherige Zusammenfassung war zu kurz.
Erweitere sie und beziehe ein:
- Wer der Nutzer ist und was er sucht
- Die genannten wichtigen Kriterien
- Die diskutierten Optionen
- Den aktuellen Stand des Gesprächs

Aktuelle Zusammenfassung: {self.summary}

Erweiterte Zusammenfassung:"""
        self.summary = self.llm.invoke(prompt)

3. Themenwechsel handhaben

DEVELOPERpython
class TopicAwareSummaryMemory(SummaryMemory):
    """Erkennt und verwaltet Themenwechsel"""

    def add_message(self, role: str, content: str) -> None:
        # Themenwechsel erkennen
        if role == "user" and self._is_topic_change(content):
            # Zusammenfassung des vorherigen Themas abschließen
            self._finalize_current_topic()

        super().add_message(role, content)

    def _is_topic_change(self, message: str) -> bool:
        """Erkennt, ob die Nachricht das Thema wechselt"""
        indicators = [
            "etwas anderes",
            "neues thema",
            "andere frage",
            "ich hätte auch gerne",
            "außerdem"
        ]
        return any(ind in message.lower() for ind in indicators)

    def _finalize_current_topic(self) -> None:
        """Archiviert das aktuelle Thema vor dem Wechsel"""
        if self.summary:
            self.archived_topics.append({
                "topic": self.summary,
                "timestamp": datetime.now().isoformat()
            })
            self.summary = ""

Wann Summary Memory verwenden?

Ideale Anwendungsfälle

• Umfassender Kundensupport: Mehrstufige Fehlerbehebung
• Beratungen: Personalisierte Empfehlungen über längere Zeit
• Onboarding: Begleitung über mehrere Sitzungen
• Komplexe Gespräche: Verhandlungen, detaillierte Konfigurationen

Wann vermeiden

Weiterführende Ressourcen

• Buffer Memory - Für kurze Gespräche
• Entity Memory - Um spezifische Entitäten zu behalten
• Konversations-RAG - Überblick

Summary Memory mit Ailog

Die Implementierung einer robusten Summary Memory erfordert besondere Aufmerksamkeit für die Qualität der Zusammenfassungen. Mit Ailog profitieren Sie von optimierter Speicherverwaltung:

• Automatische Zusammenfassung mit adaptiven Schwellwerten je nach Gespräch
• Erkennung von Themenwechseln für relevante Zusammenfassungen
• Strukturierte Zusammenfassung für präzisen Zugriff auf Informationen
• Multi-Session-Persistenz mit Kontextwiederaufnahme
• Analytics zur Qualität der Zusammenfassungen und Informationserhaltung

Ailog kostenlos testen und einen Chatbot mit Langzeitgedächtnis bereitstellen.