RAG Konversationell : Gedächtnis und Multi-Session-Kontext

Ein statisches RAG beantwortet jede Frage isoliert. Ein konversationelles RAG behält den Gesprächsfaden bei, erinnert sich an Nutzerpräferenzen und personalisiert seine Antworten im Laufe der Zeit. Dieser Leitfaden erklärt, wie man ein effektives Gedächtnis implementiert.

Warum Gedächtnis essenziell ist

Die Grenzen von RAG ohne Gedächtnis

Ohne Gedächtnis wird jede Anfrage unabhängig behandelt :

Nutzer: Wie ist Ihre Rückgabepolitik ?
Assistent: Sie haben 30 Tage, um ein Produkt zurückzugeben.

Nutzer: Und bei elektronischen Produkten ?
Assistent: [Kein Kontext] Was möchten Sie über elektronische Produkte wissen ?

Mit Gedächtnis :

Nutzer: Wie ist Ihre Rückgabepolitik ?
Assistent: Sie haben 30 Tage, um ein Produkt zurückzugeben.

Nutzer: Und bei elektronischen Produkten ?
Assistent: Bei elektronischen Produkten beträgt die Rückgabefrist ebenfalls 30 Tage,
          sie müssen jedoch ungeöffnet sein, außer bei einem Defekt.

Arten von Gedächtnis

Konversationelle Architektur

┌─────────────────────────────────────────────────────────────┐
│                    NUTZERANFRAGE                              │
└───────────────────────────┬─────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────────┐
│               CONTEXT MANAGER                                │
│  ┌─────────────┐ ┌─────────────┐ ┌───────────────────────┐  │
│  │  Sitzungs-  │ │   User      │ │      Langzeit-        │  │
│  │  verlauf    │ │  Profile    │ │      Gedächtnis       │  │
│  └──────┬──────┘ └──────┬──────┘ └───────────┬───────────┘  │
│         └───────────────┼───────────────────┘               │
│                         ▼                                    │
│               ┌─────────────────┐                           │
│               │ Context Builder │                           │
│               └────────┬────────┘                           │
└────────────────────────┼────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│                    RAG PIPELINE                              │
│  ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│  │  Query       │ │   Retrieval  │ │     Generation       │ │
│  │  Rewriting   │ │   + Context  │ │     + Memory         │ │
│  └──────────────┘ └──────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

Verwaltung des Sessionverlaufs

Struktur der Konversation

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

@dataclass
class Message:
    role: str  # "user" oder "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
    ):
        """Fügt eine Nachricht zur Konversation hinzu"""
        conversation = await self.storage.get(conversation_id)

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

        conversation.messages.append(message)

        # Nur die letzten N Nachrichten behalten
        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:
        """Holt den formatierten Konversationskontext"""
        conversation = await self.storage.get(conversation_id)

        # Die letzten Runden nehmen
        recent_messages = conversation.messages[-(max_turns * 2):]

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

        return "\n".join(context_parts)

Kompression des Verlaufs

Für sehr lange Gespräche sollte der ältere Verlauf komprimiert werden :

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

    async def compress(
        self,
        messages: List[Message],
        keep_recent: int = 4
    ) -> dict:
        """
        Komprimiert den Verlauf und behält die aktuellen Nachrichten unverändert bei
        """
        if len(messages) <= keep_recent:
            return {
                "summary": None,
                "recent_messages": messages
            }

        # Nachrichten, die komprimiert werden sollen
        to_compress = messages[:-keep_recent]
        recent = messages[-keep_recent:]

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

        prompt = f"""
        Fasse dieses Gespräch zusammen und behalte dabei bei :
        - Die ausgetauschten Schlüsselinformationen
        - Getroffene Entscheidungen oder Zusagen
        - Den wichtigen Kontext für die Fortsetzung

        Konversation :
        {history_text}

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

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

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

Query Rewriting mit Kontext

Auflösung von Koreferenzen

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

    async def rewrite(
        self,
        query: str,
        conversation_context: str
    ) -> str:
        """
        Schreibt die Anfrage um und löst Koreferenzen auf
        """
        prompt = f"""
        Schreibe die Frage des Nutzers so um, dass sie eigenständig verständlich ist,
        indem du Pronomen und Kontextverweise auflöst.

        Gesprächsverlauf :
        {conversation_context}

        Letzte Frage : {query}

        Umgeschriebene Frage (eigenständig, ohne mehrdeutige Pronomen) :
        """

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

    async def extract_search_queries(
        self,
        query: str,
        context: str
    ) -> List[str]:
        """
        Erzeugt mehrere optimierte Suchanfragen
        """
        prompt = f"""
        Erzeuge auf Basis dieser Konversation und Frage 2-3 optimierte
        Suchanfragen, um die relevanten Informationen zu finden.

        Kontext :
        {context}

        Frage : {query}

        Suchanfragen (eine pro Zeile) :
        """

        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]

Anwendungsbeispiel

DEVELOPERpython
# Vor dem Rewriting
context = """
Nutzer: Ich suche einen Laptop für Gaming
Assistent: Ich empfehle Ihnen den ASUS ROG Strix G15, ein hervorragendes Preis-Leistungs-Verhältnis.
Nutzer: Wie viel RAM hat er ?
"""

query = "Und die Grafikkarte ?"

rewritten = await rewriter.rewrite(query, context)
# "Welche Grafikkarte hat das ASUS ROG Strix G15 ?"

Langzeitgedächtnis

Speicherung des Nutzerprofils

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
    ):
        """
        Aktualisiert eine Benutzerpräferenz
        """
        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:
        """
        Ruft alle Präferenzen eines Benutzers ab
        """
        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
    ):
        """
        Loggt eine Interaktion für spätere Analysen
        """
        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:
        """
        Extrahiert Präferenzen aus vergangenen Interaktionen
        """
        # Die letzten Interaktionen abrufen
        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"""
        Analysiere diesen Gesprächsverlauf und extrahiere die Präferenzen
        und Merkmale des Nutzers.

        Verlauf :
        {history}

        Extrahiere als JSON :
        - preferred_language: bevorzugte Sprache
        - technical_level: Anfänger/Fortgeschritten/Experte
        - topics_of_interest: Liste der häufigen Themen
        - communication_style: formell/informell
        - preferences: weitere erkannte Präferenzen

        JSON :
        """

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

Personalisierung der Antworten

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:
        """
        Personalisierte RAG-Anfrage
        """
        # 1. Profil des Benutzers abrufen
        preferences = await self.memory.get_preferences(user_id)

        # 2. Gesprächskontext abrufen
        conv_context = await self.conv_manager.get_context(conversation_id)

        # 3. Anfrage mit Kontext umschreiben
        rewritten_query = await self.query_rewriter.rewrite(query, conv_context)

        # 4. Standardmäßigen RAG ausführen
        rag_result = await self.rag.query(rewritten_query)

        # 5. Antwort personalisieren
        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:
        """
        Passt die Antwort an die Nutzerpräferenzen an
        """
        tech_level = preferences.get("technical_level", "intermediaire")
        style = preferences.get("communication_style", "professionnel")

        prompt = f"""
        Passe diese Antwort an das Nutzerprofil an.

        Profil :
        - Technisches Niveau : {tech_level}
        - Bevorzugter Stil : {style}

        Gesprächskontext :
        {context}

        Frage : {query}

        Ursprüngliche Antwort :
        {rag_response}

        Angepasste Antwort :
        """

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

Geteiltes semantisches Gedächtnis

Lernen aus Gesprächen

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
    ):
        """
        Extrahiert und speichert aus einer Konversation gelernte Fakten
        """
        # Fakten extrahieren
        facts = await self._extract_facts(conversation)

        for fact in facts:
            # Prüfen, ob das Fakt bereits existiert
            existing = await self._find_similar_fact(fact)

            if existing:
                # Vertrauen (confidence) aktualisieren
                await self._update_fact_confidence(existing, fact)
            else:
                # Neues Fakt speichern
                await self._store_fact(fact)

    async def _extract_facts(self, conversation: Conversation) -> List[dict]:
        """
        Extrahiert faktische Fakten aus einer Konversation
        """
        messages_text = "\n".join([
            f"{m.role}: {m.content}" for m in conversation.messages
        ])

        prompt = f"""
        Extrahiere die neuen Fakten, die aus diesem Gespräch gelernt wurden.
        Ein Fakt muss sein :
        - Faktisch und überprüfbar
        - Nützlich für zukünftige Gespräche
        - Neu (keine bereits bekannte Information)

        Konversation :
        {messages_text}

        Fakten (JSON-Array-Format) :
        [
          {{"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]:
        """
        Ruft die relevanten Fakten für eine Anfrage ab
        """
        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]

Sitzungsverwaltung

Multi-Device und Kontinuität

DEVELOPERpython
import redis
import json

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

    async def create_session(
        self,
        user_id: str,
        device_id: str = None
    ) -> str:
        """
        Erstellt eine neue 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:
        """
        Setzt die letzte Session fort oder erstellt eine neue
        """
        # Nach einer kürzlich aktiven Session suchen
        pattern = f"session_{user_id}_*"
        keys = await self.redis.keys(pattern)

        if keys:
            # Nach Erstellungsdatum sortieren
            sessions = []
            for key in keys:
                data = json.loads(await self.redis.get(key))
                sessions.append((key, data))

            # Die aktuellste nehmen
            sessions.sort(key=lambda x: x[1]["created_at"], reverse=True)
            return sessions[0][0]

        # Neue Session erstellen
        return await self.create_session(user_id, device_id)

    async def merge_sessions(
        self,
        session_ids: List[str]
    ) -> str:
        """
        Fasst mehrere Sessions zu einer einzigen zusammen
        """
        all_messages = []

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

        # Nach Timestamp sortieren
        all_messages.sort(key=lambda x: x.get("timestamp", ""))

        # Eine zusammengeführte Session erstellen
        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

Metriken und Evaluation

Konversationelle Qualität

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

    async def calculate_metrics(
        self,
        conversation_id: str
    ) -> dict:
        """
        Berechnet Metriken für eine Konversation
        """
        conversation = await self.db.get("conversations", conversation_id)

        return {
            # Länge und 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"]
            ),

            # Auflösung
            "ended_naturally": self._check_natural_end(conversation),
            "required_clarification": self._count_clarifications(conversation),

            # Kohärenz
            "topic_coherence": await self._calculate_topic_coherence(conversation),
            "context_usage": self._measure_context_usage(conversation)
        }

    async def _calculate_topic_coherence(self, conversation: dict) -> float:
        """
        Misst, ob die Konversation thematisch kohärent bleibt
        """
        # Einbettung jeder Nachricht
        messages = conversation["messages"]
        embeddings = [
            self.embedder.encode(m["content"])
            for m in messages
        ]

        # Ähnlichkeit zwischen aufeinanderfolgenden Nachrichten berechnen
        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

Best Practices

1. Kontextgröße begrenzen

Überladen Sie das Prompt nicht mit zuviel Verlauf. Komprimieren oder zusammenfassen.

2. Umgang mit Themenwechseln

Erkennen Sie, wenn der Nutzer das Thema wechselt, um den Kontext nicht zu verschmutzen.

3. Privatheit respektieren

Ermöglichen Sie Nutzern, ihren Verlauf und ihre Präferenzen zu löschen.

4. Eleganter Fallback

Wenn das Gedächtnis nicht verfügbar ist, sollte das System im zustandslosen Modus funktionieren.

Zum Weiterlesen

• Grundlagen des Retrieval - Suche optimieren
• LLM-Generierung - Antworten verbessern
• Prompt Engineering RAG - Prompts optimieren

RAG Konversationell mit Ailog

Die Implementierung eines robusten konversationellen Gedächtnisses ist komplex. Mit Ailog profitieren Sie von diesen nativen Funktionen:

• Sitzungsverlauf automatisch mit Kompression
• Nutzergedächtnis dauerhaft und personalisiert
• Query Rewriting zur Auflösung von Koreferenzen
• Multi-Device mit Fortsetzung der Konversation
• Konversations-Analytics zur Messung des Engagements
• DSGVO-Konformität mit Export und Löschung von Daten

Testen Sie Ailog kostenlos und setzen Sie einen intelligenten konversationellen Assistenten ein.