Agent-Gedächtnissysteme

Was macht Agent-Gedächtnissysteme?

Gedächtnis ist der Eckpfeiler intelligenter Agenten. Ohne es beginnt jede Interaktion bei Null. Diese Kompetenz behandelt die Architektur des Agentengedächtnisses: Kurzzeitgedächtnis (Kontextfenster), Langzeitgedächtnis (Vektorspeicher) und die kognitiven Architekturen, die sie organisieren.

Wichtige Erkenntnis: Gedächtnis ist nicht nur Speicherung – es ist Abruf. Eine Million gespeicherter Fakten bedeuten nichts, wenn man das richtige nicht finden kann. Chunking-, Embedding- und Retrieval-Strategien entscheiden darüber, ob Ihr Agent sich erinnert oder vergisst.

Das Feld ist fragmentiert mit inkonsistenter Terminologie. Wir verwenden das CoALA-Framework (Cognitive Architectures for Language Agents): semantisches Gedächtnis (Fakten), episodisches Gedächtnis (Erfahrungen) und prozedurales Gedächtnis (Anleitungswissen).

Prinzipien

• Gedächtnisqualität = Retrieval-Qualität, nicht Speichermenge
• Chunken für den Abruf, nicht für die Speicherung
• Kontextisolation ist der Feind des Gedächtnisses
• Richtiger Gedächtnistyp für die richtige Information
• Alte Erinnerungen verfallen lassen – nicht alles sollte ewig bleiben
• Testen Sie die Retrieval-Genauigkeit vor der Produktion
• Hintergrundgedächtnisbildung ist besser als Echtzeit

Fähigkeiten

• agent-memory
• long-term-memory
• short-term-memory
• working-memory
• episodic-memory
• semantic-memory
• procedural-memory
• memory-retrieval
• memory-formation
• memory-decay

Anwendungsbereich

• vector-database-operations → data-engineer
• rag-pipeline-architecture → llm-architect
• embedding-model-selection → ml-engineer
• knowledge-graph-design → knowledge-engineer

Werkzeuge

Memory_frameworks

• LangMem (LangChain) - Wann: LangGraph-Agenten mit persistentem Gedächtnis Hinweis: Semantische, episodische, prozedurale Gedächtnistypen
• MemGPT / Letta - Wann: Virtuelles Kontextmanagement, OS-ähnliches Gedächtnis Hinweis: Hierarchische Gedächtnisebenen, automatisches Paging
• Mem0 - Wann: Benutzergedächtnisschicht für Personalisierung Hinweis: Entwickelt für Benutzerpräferenzen und -historie

Vector_stores

• Pinecone - Wann: Verwaltet, Unternehmensmaßstab (Milliarden von Vektoren) Hinweis: Beste Abfrageleistung, höchste Kosten
• Qdrant - Wann: Komplexe Metadatenfilterung, Open-Source Hinweis: Rust-basiert, hervorragende Filterung
• Weaviate - Wann: Hybride Suche, Wissensgraphfunktionen Hinweis: GraphQL-Schnittstelle, gut für Beziehungen
• ChromaDB - Wann: Prototyping, kleine/mittelgroße Anwendungen Hinweis: Entwicklerfreundlich, ~20ms p50 bei 100K Vektoren
• pgvector - Wann: Bereits PostgreSQL im Einsatz, einfachere Einrichtung Hinweis: Gut für <1M Vektoren, vertraute Werkzeuge

Embedding_models

• OpenAI text-embedding-3-large - Wann: Beste Qualität, 3072 Dimensionen Hinweis: 0,13 $/1M Tokens
• OpenAI text-embedding-3-small - Wann: Gute Balance, 1536 Dimensionen Hinweis: 0,02 $/1M Tokens, 5x günstiger
• nomic-embed-text-v1.5 - Wann: Open-Source, lokale Bereitstellung Hinweis: 768 Dimensionen, gute Qualität
• all-MiniLM-L6-v2 - Wann: Leichtgewichtig, schnelles lokales Embedding Hinweis: 384 Dimensionen, niedrigste Latenz

Muster

Gedächtnistyp-Architektur

Auswahl des richtigen Gedächtnistyps für verschiedene Informationen

Wann verwenden: Entwurf eines Agentengedächtnissystems

GEDÄCHTNISTYP-ARCHITEKTUR (CoALA-Framework):

""" Drei Gedächtnistypen für verschiedene Zwecke:

1. Semantisches Gedächtnis: Fakten und Wissen

   • Was Sie über die Welt wissen
   • Benutzerpräferenzen, Domänenwissen
   • Gespeichert in Profilen (strukturiert) oder Sammlungen (unstrukturiert)

2. Episodisches Gedächtnis: Erfahrungen und Ereignisse

   • Was passiert ist (zeitgestempelte Ereignisse)
   • Frühere Gespräche, Aufgabenergebnisse
   • Wird zum Lernen aus Erfahrung verwendet

3. Prozedurales Gedächtnis: Wie man Dinge tut

   • Regeln, Fähigkeiten, Arbeitsabläufe
   • Oft als Few-Shot-Beispiele implementiert
   • "Wie habe ich das früher gelöst?" """

LangMem-Implementierung

""" from langmem import MemoryStore from langgraph.graph import StateGraph

Gedächtnisspeicher initialisieren

memory = MemoryStore( connection_string=os.environ["POSTGRES_URL"] )

Semantisches Gedächtnis: Benutzerprofil

await memory.semantic.upsert( namespace="user_profile", key=user_id, content={ "name": "Alice", "preferences": ["Dark Mode", "prägnante Antworten"], "expertise_level": "Entwickler", } )

Episodisches Gedächtnis: vergangene Interaktion

await memory.episodic.add( namespace="conversations", content={ "timestamp": datetime.now(), "summary": "Habe beim Debuggen eines Authentifizierungsproblems geholfen", "outcome": "gelöst", "key_insights": ["Token-Ablauf war die Ursache"], }, metadata={"user_id": user_id, "topic": "debugging"} )

Prozedurales Gedächtnis: gelerntes Muster

await memory.procedural.add( namespace="skills", content={ "task_type": "debug_auth", "steps": ["Token-Ablauf prüfen", "Refresh-Flow überprüfen"], "example_interaction": few_shot_example, } ) """

Gedächtnisabruf zur Laufzeit

""" async def prepare_context(user_id, query): # Benutzerprofil abrufen (semantisch) profile = await memory.semantic.get( namespace="user_profile", key=user_id )

# Relevante vergangene Erfahrungen finden (episodisch)
similar_experiences = await memory.episodic.search(
    namespace="conversations",
    query=query,
    filter={"user_id": user_id},
    limit=3
)

# Relevante Fähigkeiten finden (prozedural)
relevant_skills = await memory.procedural.search(
    namespace="skills",
    query=query,
    limit=2
)

return {
    "profile": profile,
    "past_experiences": similar_experiences,
    "relevant_skills": relevant_skills,
}

"""

Vektorspeicher-Auswahlmuster

Auswahl der richtigen Vektordatenbank für Ihren Anwendungsfall

Wann verwenden: Einrichtung persistenter Gedächtnisspeicherung

VEKTORSPEICHER-AUSWAHL:

""" Entscheidungsmatrix:

	Pinecone	Qdrant	Weaviate	ChromaDB	pgvector
Maßstab	Milliarden	100M+	100M+	1M	1M
Verwaltet	Ja	Beide	Beide	Selbst	Selbst
Filterung	Einfach	Am besten	Gut	Einfach	SQL
Hybrid	Nein	Ja	Am besten	Nein	Ja
Kosten	Hoch	Mittel	Mittel	Kostenlos	Kostenlos
Latenz	5ms	7ms	10ms	20ms	15ms
"""

Pinecone (Unternehmensmaßstab)

""" from pinecone import Pinecone

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) index = pc.Index("agent-memory")

Upsert mit Metadaten

index.upsert( vectors=[ { "id": f"memory-{uuid4()}", "values": embedding, "metadata": { "user_id": user_id, "timestamp": datetime.now().isoformat(), "type": "episodic", "content": memory_text, } } ], namespace=namespace )

Abfrage mit Filter

results = index.query( vector=query_embedding, filter={"user_id": user_id, "type": "episodic"}, top_k=5, include_metadata=True ) """

Qdrant (Komplexe Filterung)

""" from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, Filter, FieldCondition

client = QdrantClient(url="http://localhost:6333")

Komplexe Filterung mit Qdrant

results = client.search( collection_name="agent_memory", query_vector=query_embedding, query_filter=Filter( must=[ FieldCondition(key="user_id", match={"value": user_id}), FieldCondition(key="type", match={"value": "semantic"}), ], should=[ FieldCondition(key="topic", match={"any": ["auth", "security"]}), ] ), limit=5 ) """

ChromaDB (Prototyping)

""" import chromadb

client = chromadb.PersistentClient(path="./memory_db") collection = client.get_or_create_collection("agent_memory")

Einfach und schnell für Prototypen

collection.add( ids=[str(uuid4())], embeddings=[embedding], documents=[memory_text], metadatas=[{"user_id": user_id, "type": "episodic"}] )

results = collection.query( query_embeddings=[query_embedding], n_results=5, where={"user_id": user_id} ) """

Chunking-Strategie-Muster

Aufteilen von Dokumenten in abrufbare Stücke

Wann verwenden: Verarbeitung von Dokumenten zur Gedächtnisspeicherung

CHUNKING-STRATEGIEN:

""" Das Chunking-Dilemma:

• Zu groß: Vektor verliert Spezifität
• Zu klein: Verliert Kontext

Die optimale Chunk-Größe hängt ab von:

• Dokumenttyp (Code vs. Prosa vs. Daten)
• Abfragemuster (faktenbasiert vs. explorativ)
• Embedding-Modell (jedes hat einen Sweet Spot)

Allgemeine Richtlinie: 256-512 Tokens für die meisten Anwendungsfälle """

Festgrößen-Chunking (Basislinie)

""" from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter( chunk_size=500, # Zeichen chunk_overlap=50, # Überlappung verhindert das Durchtrennen von Sätzen separators=["\n\n", "\n", ". ", " ", ""] # Prioritätsreihenfolge )

chunks = splitter.split_text(document) """

Semantisches Chunking (Bessere Qualität)

""" from langchain_experimental.text_splitter import SemanticChunker from langchain_openai import OpenAIEmbeddings

Aufteilung basierend auf semantischer Ähnlichkeit

splitter = SemanticChunker( embeddings=OpenAIEmbeddings(), breakpoint_threshold_type="percentile", breakpoint_threshold_amount=95 )

chunks = splitter.split_text(document) """

Strukturbewusstes Chunking (Dokumente mit Hierarchie)

""" from langchain.text_splitter import MarkdownHeaderTextSplitter

Respektiert die Dokumentstruktur

splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "Überschrift 1"), ("##", "Überschrift 2"), ("###", "Überschrift 3"), ] )

chunks = splitter.split_text(markdown_doc)

Jeder Chunk hat Kopfzeilen-Metadaten für den Kontext

"""

Kontextuelles Chunking (Anthropics Ansatz)

"""

Vor dem Einbetten Kontext zu jedem Chunk hinzufügen

Reduziert Abruffehler um 35%

def add_context_to_chunk(chunk, document_summary): context_prompt = f''' Dokumentenzusammenfassung: {document_summary}

Das Folgende ist ein Chunk aus diesem Dokument:
{chunk}
'''
return context_prompt

Den kontextualisierten Chunk einbetten, nicht den rohen Chunk

for chunk in chunks: contextualized = add_context_to_chunk(chunk, summary) embedding = embed(contextualized) store(chunk, embedding) # Original speichern, Kontextualisiertes einbetten """

Code-spezifisches Chunking

""" from langchain.text_splitter import Language, RecursiveCharacterTextSplitter

Sprachbewusste Aufteilung

python_splitter = RecursiveCharacterTextSplitter.from_language( language=Language.PYTHON, chunk_size=1000, chunk_overlap=200 )

Respektiert Funktions-/Klassengrenzen

chunks = python_splitter.split_text(python_code) """

Hintergrundgedächtnisbildung

Asynchrone Verarbeitung von Erinnerungen für bessere Qualität

Wann verwenden: Sie möchten eine höhere Trefferquote ohne Verlangsamung von Interaktionen

HINTERGRUNDGEDÄCHTNISBILDUNG:

""" Die Gedächtnisextraktion in Echtzeit verlangsamt Gespräche und erhöht die Komplexität von Agenten-Werkzeugaufrufen. Die Hintergrundverarbeitung nach Gesprächen liefert qualitativ hochwertigere Erinnerungen.

Muster: Unterbewusste Gedächtnisbildung """

LangGraph-Hintergrundverarbeitung

""" from langgraph.graph import StateGraph from langgraph.checkpoint.postgres import PostgresSaver

async def background_memory_processor(thread_id: str): # Nach Gesprächsende oder Leerlauf ausführen conversation = await load_conversation(thread_id)

# Erkenntnisse ohne Zeitdruck extrahieren
insights = await llm.invoke('''
    Analysiere dieses Gespräch und extrahiere:
    1. Wichtige Fakten über den Benutzer, die gelernt wurden
    2. Offenbarte Benutzerpräferenzen
    3. Abgeschlossene oder anstehende Aufgaben
    4. Muster im Benutzerverhalten

    Sei gründlich – dies läuft im Hintergrund.

    Gespräch:
    {conversation}
''')

# Im Langzeitgedächtnis speichern
for insight in insights:
    await memory.semantic.upsert(
        namespace="user_insights",
        key=generate_key(insight),
        content=insight,
        metadata={"source_thread": thread_id}
    )

Bei Gesprächsende oder Leerlauf-Timeout auslösen

@on_conversation_idle(timeout_minutes=5) async def process_conversation(thread_id): await background_memory_processor(thread_id) """

Gedächtniskonsolidierung (Wie Schlaf)

"""

Erinnerungen regelmäßig konsolidieren und deduplizieren

async def consolidate_memories(user_id: str): # Alle Erinnerungen für den Benutzer abrufen memories = await memory.semantic.list( namespace="user_insights", filter={"user_id": user_id} )

# Ähnliche Erinnerungen finden (potenzielle Duplikate)
clusters = cluster_by_similarity(memories, threshold=0.9)

# Ähnliche Erinnerungen zusammenführen
for cluster in clusters:
    if len(cluster) > 1:
        merged = await llm.invoke(f'''
            Konsolidiere diese verwandten Erinnerungen zu einer:
            {cluster}

            Bewahre alle wichtigen Informationen.
        ''')
        await memory.semantic.upsert(
            namespace="user_insights",
            key=generate_key(merged),
            content=merged
        )
        # Originale löschen
        for old in cluster:
            await memory.semantic.delete(old.id)

"""

Gedächtnisverfall-Muster

Vergessen alter, irrelevanter Erinnerungen

Wann verwenden: Gedächtnis wird groß, Abruf wird langsamer

GEDÄCHTNISVERFALL:

""" Nicht alle Erinnerungen sollten ewig leben:

• Alte Präferenzen können veraltet sein
• Aufgabendetails verlieren an Relevanz
• Sich widersprechende Erinnerungen verwirren den Abruf

Intelligenten Verfall implementieren basierend auf:

• Aktualität (wann wurde es erstellt/abgerufen?)
• Häufigkeit (wie oft wird es abgerufen?)
• Wichtigkeit (ist es eine Kernfakten oder ein Detail?) """

Zeitbasierter Verfall

""" from datetime import datetime, timedelta

async def decay_old_memories(namespace: str, max_age_days: int): cutoff = datetime.now() - timedelta(days=max_age_days)

old_memories = await memory.episodic.list(
    namespace=namespace,
    filter={"last_accessed": {"$lt": cutoff.isoformat()}}
)

for mem in old_memories:
    # Weiches Löschen (als archiviert markieren)
    await memory.episodic.update(
        id=mem.id,
        metadata={"archived": True, "archived_at": datetime.now().isoformat()}
    )

"""

Wichtigkeitsschwellenwert

""" async def decay_by_importance(namespace: str, min_importance: float): # Erinnerungen mit geringer Wichtigkeit finden low_importance = await memory.semantic.search( namespace=namespace, query_filter={"importance": {"$lt": min_importance}}, limit=1000 )

for mem in low_importance:
    # Relevanz prüfen bevor gelöscht wird
    if not await is_relevant_now(mem):
        await memory.semantic.delete(mem.id)

"""

Zusammenführungsstrategie (Bei Konflikten)

""" async def resolve_memory_conflict(user_id: str, topic: str): # Sich widersprechende Erinnerungen finden memories = await memory.semantic.search( namespace="user_insights", query=topic, filter={"user_id": user_id}, limit=10 )

if has_conflicts(memories):
    # KI zur Lösung nutzen
    resolved = await llm.invoke(f'''
        Diese Erinnerungen widersprechen sich:
        {memories}

        Welche ist basierend auf neueren Informationen oder
        zuverlässigeren Quellen am wahrscheinlichsten korrekt?
        Antworte nur mit der korrigierten Version.
    ''')
    # Alte Erinnerungen durch die aufgelöste Version ersetzen
    await memory.semantic.upsert(
        namespace="user_insights",
        key=generate_key(resolved),
        content=resolved,
        metadata={"resolved_conflict": True, "timestamp": datetime.now()}
    )
    # In Konflikt stehende Originale löschen
    for mem in memories:
        await memory.semantic.delete(mem.id)

"""

Nützliche Notizen

• Fangen Sie klein an: ChromaDB + MiniLM für Prototypen, dann skalieren
• Einbettungen vor der Produktion testen: Nutzen Sie pytest mit Abruffälschungen
• Das CoALA-Framework bietet eine gemeinsame Sprache für Gedächtnisentwürfe
• Gedächtnisbildung im Hintergrund nach Gesprächen produziert reichhaltigere Erinnerungen
• Vergessen ist in Ordnung: Implementieren Sie Verfall für gesunde Gedächtnissysteme
• Kontextuelles Chunking (Anthropics Ansatz) reduziert Abruffehler um bis zu 35 %
• Strukturbewusstes Chunking (Markdown, Code) bewahrt den semantischen Kontext