Entity Memory : Erwähnte Entitäten behalten

Die Entity Memory extrahiert und speichert die in der Konversation erwähnten Schlüsselen(titäten): Personen, Produkte, Unternehmen, Orte, Daten usw. Im Gegensatz zu Buffer oder Summary Memory, die den Rohtext bewahren, baut die Entity Memory einen dynamischen "Knowledge Graph" auf, der im Verlauf der Interaktionen angereichert wird. Das ermöglicht natürliche Referenzen wie "er", "dieses Produkt", "diese Option".

Warum die Entity Memory?

Das Problem mit Referenzen auf Entitäten

In einer natürlichen Konversation beziehen sich Benutzer ständig auf zuvor erwähnte Entitäten:

User: "Ich suche das Dell XPS 15"
AI: "Das Dell XPS 15 ist ein hervorragendes Laptop..."

User: "Wie lange ist die Garantie?"     <- "die Garantie" = Dell XPS 15
AI: "Die Garantie des Dell XPS 15..."

User: "Und im Vergleich zum Lenovo?"    <- "zum Lenovo" = welches Lenovo-Produkt?
AI: ???

Ohne Entity Memory muss das LLM raten, auf welche Entität sich der Nutzer bezieht.

Die Architektur der Entity Memory

┌─────────────────────────────────────────────────────────────┐
│                     ENTITY MEMORY                            │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌────────────────────────────────────────────────────┐     │
│  │ ENTITY STORE                                        │     │
│  ├────────────────────────────────────────────────────┤     │
│  │                                                     │     │
│  │ "Dell XPS 15" (PRODUCT)                            │     │
│  │   - Type: Laptop                                   │     │
│  │   - Prix: 1599 euros                               │     │
│  │   - Garantie: 2 ans                               │     │
│  │   - Mentions: 3                                    │     │
│  │                                                     │     │
│  │ "Lenovo ThinkPad X1" (PRODUCT)                     │     │
│  │   - Type: Laptop                                   │     │
│  │   - Prix: 1799 euros                               │     │
│  │   - Mentions: 1                                    │     │
│  │                                                     │     │
│  └────────────────────────────────────────────────────┘     │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Wichtiges Ergebnis : Die Entity Memory verbessert die Relevanz der Antworten um 40–60% bei Konversationen mit häufigen Referenzen auf dieselben Objekte.

Anwendungsfälle

Grundimplementierung

Struktur der Entitäten

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

@dataclass
class Entity:
    """Stellt eine extrahierte Entität dar"""
    name: str
    type: str  # PERSON, PRODUCT, ORGANIZATION, etc.
    attributes: Dict = field(default_factory=dict)
    mentions: int = 1
    first_seen: datetime = field(default_factory=datetime.now)
    last_seen: datetime = field(default_factory=datetime.now)

    def update(self, new_attributes: Dict) -> None:
        """Aktualisiert die Attribute der Entität"""
        self.attributes.update(new_attributes)
        self.mentions += 1
        self.last_seen = datetime.now()

    def to_dict(self) -> Dict:
        """Wandelt in ein Dictionary um"""
        return {
            "name": self.name,
            "type": self.type,
            "attributes": self.attributes,
            "mentions": self.mentions
        }

Entity Memory komplett

DEVELOPERpython
class EntityMemory:
    """
    Konversationsspeicher basierend auf Entitäten
    """
    def __init__(self, llm, entity_types: List[str] = None):
        self.llm = llm
        self.entity_types = entity_types or [
            "PERSON", "PRODUCT", "ORGANIZATION",
            "LOCATION", "DATE", "MONEY", "CONCEPT"
        ]
        self.entities: Dict[str, Entity] = {}
        self.recent_messages: List[Dict] = []

    def add_message(self, role: str, content: str) -> None:
        """Fügt eine Nachricht hinzu und extrahiert Entitäten"""
        self.recent_messages.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })

        # Entitäten aus der Nachricht extrahieren
        entities = self._extract_entities(content)
        self._update_entity_store(entities)

    def _extract_entities(self, text: str) -> List[Dict]:
        """Extrahiert Entitäten aus einem Text mit dem LLM"""
        prompt = f"""Extrahiere die benannten Entitäten aus diesem Text.

Zu erkennende Entitätstypen: {', '.join(self.entity_types)}

Text: "{text}"

Gib ein JSON-Array zurück:
[{{"name": "name", "type": "TYPE", "attributes": {{}}}}, ...]

Falls keine Entität gefunden wird, gib zurück: []

JSON:"""

        response = self.llm.invoke(prompt)
        try:
            return json.loads(response)
        except json.JSONDecodeError:
            return []

    def _update_entity_store(self, entities: List[Dict]) -> None:
        """Aktualisiert den Entitäten-Store"""
        for entity_data in entities:
            name = entity_data.get("name", "").lower()
            if not name:
                continue

            if name in self.entities:
                self.entities[name].update(entity_data.get("attributes", {}))
            else:
                self.entities[name] = Entity(
                    name=entity_data.get("name"),
                    type=entity_data.get("type", "UNKNOWN"),
                    attributes=entity_data.get("attributes", {})
                )

    def get_recent_entities(self, limit: int = 5) -> List[Entity]:
        """Gibt die zuletzt erwähnten Entitäten zurück"""
        sorted_entities = sorted(
            self.entities.values(),
            key=lambda e: e.last_seen,
            reverse=True
        )
        return sorted_entities[:limit]

    def resolve_reference(self, reference: str) -> Optional[str]:
        """Löst eine Referenz wie 'il', 'ce produit' auf"""
        if not self.entities:
            return None

        entities_json = json.dumps([e.to_dict() for e in self.get_recent_entities(5)])

        prompt = f"""Aufzulösende Referenz: "{reference}"

Bekannte Entitäten:
{entities_json}

Welche Entität entspricht "{reference}"?
Antworte nur mit dem Namen oder "NONE", falls keine Übereinstimmung gefunden wird."""

        result = self.llm.invoke(prompt).strip()
        return result if result != "NONE" else None

    def get_context(self) -> str:
        """Erzeugt den Kontext, der in das Prompt eingefügt werden soll"""
        if not self.entities:
            return ""

        recent = self.get_recent_entities(5)
        entity_lines = []
        for e in recent:
            attrs = ", ".join([f"{k}: {v}" for k, v in e.attributes.items()])
            line = f"- {e.name} ({e.type})"
            if attrs:
                line += f": {attrs}"
            entity_lines.append(line)

        return "Entitäten im Kontext:\n" + "\n".join(entity_lines)

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

Integration mit RAG

DEVELOPERpython
class RAGWithEntityMemory:
    """RAG-Pipeline mit Entity Memory"""

    def __init__(self, vector_store, llm):
        self.vector_store = vector_store
        self.llm = llm
        self.memory = EntityMemory(llm)

    def query(self, user_message: str) -> str:
        """Führt eine RAG-Anfrage mit Entity-Kontext aus"""
        # 1. Referenzen auflösen
        resolved_message = self._resolve_references(user_message)

        # 2. Entitäten extrahieren
        self.memory.add_message("user", resolved_message)

        # 3. Dokumente abrufen
        docs = self.vector_store.similarity_search(resolved_message, k=3)
        doc_context = "\n\n".join([d.page_content for d in docs])

        # 4. Prompt erstellen
        entity_context = self.memory.get_context()

        prompt = f"""{entity_context}

Relevante Dokumente:
{doc_context}

Frage: {user_message}

Antworte unter Berücksichtigung der erwähnten Entitäten."""

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

        return response

    def _resolve_references(self, text: str) -> str:
        """Löst pronominale Referenzen"""
        references = ["er", "sie", "dieses Produkt", "das", "dieser hier"]

        for ref in references:
            if ref in text.lower():
                resolved = self.memory.resolve_reference(ref)
                if resolved:
                    text = text.replace(ref, resolved, 1)

        return text

Fortgeschrittene Techniken

Extraktion mit Relationen

DEVELOPERpython
class RelationalEntityMemory(EntityMemory):
    """Entity Memory mit Relationen zwischen Entitäten"""

    def __init__(self, llm, **kwargs):
        super().__init__(llm, **kwargs)
        self.relations: List[Dict] = []

    def _extract_entities(self, text: str) -> List[Dict]:
        """Extrahiert Entitäten UND Relationen"""
        prompt = f"""Analysiere diesen Text und extrahiere:
1. Die benannten Entitäten
2. Die Beziehungen zwischen den Entitäten

Beziehungstypen: COMPARES_TO, PREFERS, OWNS, INTERESTED_IN

Text: "{text}"

JSON:
{{
  "entities": [{{"name": "...", "type": "...", "attributes": {{}}}}],
  "relations": [{{"source": "e1", "relation": "TYPE", "target": "e2"}}]
}}"""

        response = self.llm.invoke(prompt)
        try:
            data = json.loads(response)
            for rel in data.get("relations", []):
                self.relations.append(rel)
            return data.get("entities", [])
        except json.JSONDecodeError:
            return []

    def get_related_entities(self, entity_name: str) -> List[Dict]:
        """Findet verbundene Entitäten"""
        related = []
        name_lower = entity_name.lower()

        for rel in self.relations:
            if rel["source"].lower() == name_lower:
                related.append({"entity": rel["target"], "relation": rel["relation"]})
            elif rel["target"].lower() == name_lower:
                related.append({"entity": rel["source"], "relation": rel["relation"]})

        return related

Persistenz über mehrere Sessions

DEVELOPERpython
from pathlib import Path

class PersistentEntityMemory(EntityMemory):
    """Entity Memory mit persistenter Speicherung"""

    def __init__(self, llm, user_id: str, storage_dir: str = "./entity_store"):
        super().__init__(llm)
        self.user_id = user_id
        self.storage_path = Path(storage_dir) / f"{user_id}.json"
        self._load()

    def _load(self) -> None:
        """Lädt die Entitäten aus der Datei"""
        if self.storage_path.exists():
            with open(self.storage_path, "r") as f:
                data = json.load(f)
                for name, entity_data in data.get("entities", {}).items():
                    self.entities[name] = Entity(
                        name=entity_data["name"],
                        type=entity_data["type"],
                        attributes=entity_data.get("attributes", {}),
                        mentions=entity_data.get("mentions", 1)
                    )

    def save(self) -> None:
        """Speichert die Entitäten"""
        self.storage_path.parent.mkdir(exist_ok=True)
        data = {
            "user_id": self.user_id,
            "entities": {name: e.to_dict() for name, e in self.entities.items()}
        }
        with open(self.storage_path, "w") as f:
            json.dump(data, f, indent=2)

    def add_message(self, role: str, content: str) -> None:
        super().add_message(role, content)
        self.save()

Implementation mit LangChain

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

llm = ChatOpenAI(model="gpt-4", temperature=0.7)

# Native Entity Memory in LangChain
memory = ConversationEntityMemory(llm=llm, return_messages=True)

conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# Verwendung
response = conversation.predict(input="Ich suche ein Dell XPS 15")
response = conversation.predict(input="Was kostet es?")

# Entitäten inspizieren
print(memory.entity_store)

Hinweis (Juli 2026): ConversationEntityMemory und ConversationChain sind seit LangChain 0.3.1 veraltet (deprecated). Mit dem Erscheinen von LangChain 1.0 wurden diese Klassen aus dem Hauptpaket langchain herausgelöst und in das Paket langchain-classic verschoben (Installation via pip install langchain-classic, Import aus langchain_classic). Für neue Projekte empfiehlt LangChain inzwischen die Persistenz über LangGraph-Checkpointing (mit LangGraph Store für Langzeitgedächtnis). Die oben gezeigte manuelle Implementierung bleibt gültig und ist von LangChain unabhängig.

Best Practices

1. Entity-Typen je nach Domäne

DEVELOPERpython
# E-commerce
ECOMMERCE_TYPES = ["PRODUCT", "BRAND", "CATEGORY", "PRICE", "FEATURE"]

# Support
SUPPORT_TYPES = ["PRODUCT", "ISSUE", "SOLUTION", "TICKET_ID", "PERSON"]

# HR
HR_TYPES = ["PERSON", "POSITION", "COMPANY", "SKILL", "EDUCATION"]

2. Ablauf (Expiration) von Entitäten

DEVELOPERpython
from datetime import timedelta

class ExpiringEntityMemory(EntityMemory):
    def __init__(self, llm, ttl_minutes: int = 30, **kwargs):
        super().__init__(llm, **kwargs)
        self.ttl = timedelta(minutes=ttl_minutes)

    def _cleanup_expired(self) -> None:
        now = datetime.now()
        expired = [name for name, e in self.entities.items()
                   if now - e.last_seen > self.ttl]
        for name in expired:
            del self.entities[name]

    def get_context(self) -> str:
        self._cleanup_expired()
        return super().get_context()

3. Intelligente Priorisierung

DEVELOPERpython
def get_prioritized_entities(self, query: str, limit: int = 5) -> List[Entity]:
    """Gibt die für eine Anfrage relevantesten Entitäten zurück"""
    scored = []
    for entity in self.entities.values():
        score = 0
        recency = (datetime.now() - entity.last_seen).total_seconds()
        score += max(0, 100 - recency / 60)
        score += entity.mentions * 10
        if entity.name.lower() in query.lower():
            score += 50
        scored.append((entity, score))

    scored.sort(key=lambda x: x[1], reverse=True)
    return [e for e, _ in scored[:limit]]

Wann Entity Memory verwenden?

Ideale Fälle

• E-commerce : Verfolgung verglichener Produkte
• Technischer Support : Verfolgung von Problemen und Lösungen
• CRM/Vertrieb : Verfolgung von Kontakten und Opportunities
• Beratung : Behalten von Präferenzen und Einschränkungen

Wann vermeiden

Verwandte Guides

• Buffer Memory - Für einfache Konversationen
• Summary Memory - Für lange Konversationen
• RAG Conversationnel - Überblick

Entity Memory mit Ailog

Mit Ailog profitieren Sie von einer nativen Entitätenverwaltung:

• Automatische Extraktion der Entitäten mit konfigurierbaren Typen
• Coreference-Resolution ("il", "celui-ci", etc.)
• Mehrsessionen-Persistenz der Benutzerentitäten
• Relationengraph zwischen Entitäten
• Analytics zu den am häufigsten erwähnten Entitäten

Testen Sie Ailog kostenlos und setzen Sie einen Chatbot auf, der die wichtigen Entitäten behält.