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 :

Utilisateur: Quelle est votre politique de retour ?
Assistant: Vous avez 30 jours pour retourner un produit.

Utilisateur: Et pour les produits electroniques ?
Assistant: [Pas de contexte] Que voulez-vous savoir sur les produits electroniques ?

Mit Gedächtnis :

Utilisateur: Quelle est votre politique de retour ?
Assistant: Vous avez 30 jours pour retourner un produit.

Utilisateur: Et pour les produits electroniques ?
Assistant: Pour les produits electroniques, le delai de retour est egalement de 30 jours,
          mais ils doivent etre non ouverts sauf en cas de defaut.

Arten von Gedächtnis

Konversationelle Architektur

┌─────────────────────────────────────────────────────────────┐
│                    REQUETE UTILISATEUR                       │
└───────────────────────────┬─────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────────┐
│               CONTEXT MANAGER                                │
│  ┌─────────────┐ ┌─────────────┐ ┌───────────────────────┐  │
│  │  Historique │ │   User      │ │      Memoire          │  │
│  │  Session    │ │  Profile    │ │      Long Terme       │  │
│  └──────┬──────┘ └──────┬──────┘ └───────────┬───────────┘  │
│         └───────────────┼───────────────────┘               │
│                         ▼                                    │
│               ┌─────────────────┐                           │
│               │ 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" ou "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 = "Utilisateur" if msg.role == "user" else "Assistant"
            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"""
        Resume cette conversation en conservant :
        - Les informations cles echangees
        - Les decisions ou engagements pris
        - Le contexte important pour la suite

        Conversation :
        {history_text}

        Resume concis (max 200 mots) :
        """

        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"""
        Tu dois reecrire la question de l'utilisateur pour la rendre autonome,
        en resolvant les pronoms et references au contexte.

        Historique de conversation :
        {conversation_context}

        Derniere question : {query}

        Question reecrite (autonome, sans pronoms ambigus) :
        """

        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"""
        A partir de cette conversation et question, genere 2-3 requetes
        de recherche optimisees pour trouver les informations pertinentes.

        Contexte :
        {context}

        Question : {query}

        Requetes de recherche (une par ligne) :
        """

        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 = """
Utilisateur: Je cherche un laptop pour le gaming
Assistant: Je vous recommande le ASUS ROG Strix G15, excellent rapport qualite-prix.
Utilisateur: Il a combien de RAM ?
"""

query = "Et la carte graphique ?"

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"""
        Analyse cet historique de conversation et extrait les preferences
        et caracteristiques de l'utilisateur.

        Historique :
        {history}

        Extrais en JSON :
        - preferred_language: langue preferee
        - technical_level: debutant/intermediaire/expert
        - topics_of_interest: liste des sujets frequents
        - communication_style: formel/informel
        - preferences: autres preferences detectees

        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"""
        Adapte cette reponse selon le profil utilisateur.

        Profil :
        - Niveau technique : {tech_level}
        - Style prefere : {style}

        Contexte de conversation :
        {context}

        Question : {query}

        Reponse originale :
        {rag_response}

        Reponse adaptee :
        """

        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"""
        Extrais les faits nouveaux appris de cette conversation.
        Un fait doit etre :
        - Factuel et verifiable
        - Utile pour de futures conversations
        - Nouveau (pas une info deja connue)

        Conversation :
        {messages_text}

        Faits (format JSON array) :
        [
          {{"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

• Fondamentaux du Retrieval - Optimiser la recherche
• Generation LLM - Ameliorer les reponses
• Prompt Engineering RAG - Optimiser les prompts

RAG Konversationell mit Ailog

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

• Historique de session automatique avec compression
• Memoire utilisateur persistante et personnalisation
• Query rewriting pour resoudre les co-references
• Multi-device avec reprise de conversation
• Analytics conversationnelles pour mesurer l'engagement
• Conformite RGPD avec export et suppression des donnees

Testez Ailog gratuitement et deployez un assistant conversationnel intelligent.