Ein vierlagiges Speicherdesign für KI-Agenten mit Herkunftsverfolgung
Trennung von Kurz- und Langzeitgedächtnis, vier Schichten (Puffer, episodisch, semantisch, prozedural), Speicheroptionen, asynchrone Schreibvorgänge sowie Schutzmaßnahmen gegen Gedächtnisvergiftung.
Stellen Sie sich einen Agenten vor, der in Schritt vierzehn eines langen Workflows versagt. Das Modell trägt keine Schuld – der Agent weiß einfach nicht mehr, was er bereits erledigt hat. Der Fehler folgt einem bekannten Muster: Die Aufrufe der Tools sind einwandfrei und die Anfragen klar formuliert, doch sobald das Gespräch eine bestimmte Länge überschreitet, beginnt der Agent im Kreis zu laufen. Er wiederholt bereits beantwortete Fragen, holt Daten ab, die er bereits besitzt, und widerspricht sich zwischen Schritt sechs und Schritt zwölf, als wären die ersten fünf Schritte nie stattgefunden.
Das Einführen eines anderen Frontier-Modells ändert das in der Regel kaum. Das Engpass ist die Infrastruktur um das Modell herum – genauer gesagt das Fehlen einer echten Speicherverwaltung. Das Einfügen der letzten N Nachrichten in die Anfrage funktioniert bei Chatbots, bricht jedoch zusammen, wenn ein Agent einen zwanzigstufigen Workflow ausführen muss und in Schritt siebzehn noch daran erinnert werden muss, was er in Schritt drei entschieden hat.
Ein Muster, das dieses Problem löst, ist ein Vier-Schichten-Gedächtnisturm, bei dem hinter jeder Schicht die passende Speicherung vorhanden ist. Im Folgenden wird erläutert, was jede Schicht bewirkt, wie Teams bei GitHub, Anthropic und Berkeley ihre Versionen beschreiben, sowie Python-Beispiele für jeden Bestandteil.
Ein Kontextfenster ist kein Gedächtnis
Ein Kontextfenster stellt Arbeitsspeicher dar: RAM, nicht eine Festplatte. Jeder Token darin hat einen Aufwand. Die Einpflege der gesamten Konversationsgeschichte in ein 128K-Fenster erscheint sicher, solange man keinen tool-intensiven Agenten verwendet. Ein 30-stufiger Workflow mit Toolaufrufen, Beobachtungen und Schlussfolgerungsverläufen verbraucht schnell 100K Token, und wenn das Fenster voll ist, wird der älteste Inhalt von vorne entfernt – meist dort, wo sich die ursprünglichen Einschränkungen und Aufgabenbeschreibungen des Benutzers befinden.
Ein subtileres Problem ist noch schlimmer. Eine Studie von Forschern von Stanford und UC Berkeley, die 2024 in einer ACL-Zeitschrift unter dem Titel „Lost in the Middle: How Language Models Use Long Contexts“ veröffentlicht wurde, ergab, dass die Leistung von LLMs im Hinblick auf den Kontextumfang einer U-förmigen Kurve folgt. Die Modelle performen gut mit Informationen am Anfang und am Ende des Kontexts, leiden jedoch deutlich unter Informationen, die in der Mitte liegen. Dieses Phänomen trat bei verschiedenen Architekturen, Modellgrößen und Kontextlängen auf, wobei Kontextfenster von 4K bis 32K Token getestet wurden. Eine im Jahr 2025 veröffentlichte Fortsetzungsstudie vom MIT lieferte eine mathematische Erklärung und argumentierte, dass sowohl kausales Attention-Masking als auch rotatorische Positionskodierung die Aufmerksamkeitsgewichte für Tokens, die weit von den Enden der Sequenz entfernt sind, verringern.
Für Agenten sind die Folgen klar. Selbst wenn alles passt, wird das Modell es nicht gleichmäßig berücksichtigen. Eine kritische Einschränkung bei 30.000 Token in einem Kontext von 60.000 Token hat tatsächlich die Möglichkeit, faktisch unsichtbar zu bleiben.
Das Modell vergisst, weil das System ihm mehr Informationen zuführt, als es zuverlässig verarbeiten kann, und es gibt keine Kontrolle darüber, was in den Verarbeitungsbereich gelangt.
Die grundlegende Einteilung: Kurzfristige und langfristige Erinnerung
Vor der vierstufigen Taxonomie gibt es eine einfachere Einteilung, die die meisten Ingenieure im Kopf haben. Die Erinnerung in KI-Systemen lässt sich in kurzfristige und langfristige Erinnerung einteilen, und das Verständnis dessen, was jede Art von Erinnerung nicht leisten kann, motiviert die Entwicklung einer feineren Architektur.
Kurzfristiges Gedächtnis ist auf eine Sitzung beschränkt und existiert nur für eine einzige Interaktion. Es enthält das, was der Agent im Moment benötigt: den Zustand der aktuellen Aufgabe, den erreichten Schritt, die letzte Ausgabe eines Tools sowie den laufend erstellten Schlussfolgerungsverlauf. In der Praxis handelt es sich dabei um das Kontextfenster, das äußerst dynamisch ist, ständig aktualisiert wird, pro Token Kosten verursacht und beim Ende der Sitzung gelöscht wird.
In der Dokumentation von LangChain wird dies gut beschrieben, wobei das Kurzfristige Gedächtnis als Speicher für Konversationsgeschichte, Tool-Ausgaben und vorübergehenden Schlussfolgerungszustand bezeichnet wird. Unausgesprochen bleibt dabei, dass genau dies auch zuerst bei größeren Skalen nachlässt. In einem Workflow mit zwanzig agierenden Schritten häufen sich Zwischenergebnisse, Reflexionen, Suchergebnisse und Tool-Antworten an und konkurrieren um denselben begrenzten Speicherplatz. Das gleicht dem Versuch, einen komplizierten Plan im Kopf zu behalten, während jemand ständig mit neuen Fakten unterbricht.
Das Kurzzeitgedächtnis hat zwei praktische Komponenten:
- Aufgabenzustand ist die laufende Dokumentation des Fortschritts: der aktuelle Schritt, ausstehende Teilaufgaben, der Ausführungszustand von Tools sowie die Koordination zwischen Agenten. Ein Reiseplanungsagent speichert das Ziel, die Budgetgrenzen, bereits geprüfte Flüge sowie vorgeschlagene Hotels, damit er bei Schritt acht nicht die Arbeit von Schritt zwei wiederholen muss.
- Zwischenergebnisse sind während des Verarbeitungsprozesses entstehende Ergebnisse: Tool-Ergebnisse, generierte Zusammenfassungen, SQL-Ausgaben, Suchauszüge sowie Notizen im Zwischenspeicher. Das Ingenieurteam von Anthropic hat ein Muster in Produktionsystemen beschrieben, bei dem Agenten, die Notizen außerhalb des Kontextfensters speichern und diese später abrufen, bei langen Aufgaben besser abschneiden als Agenten, die versuchen, alles im Kontext zu behalten. Das Notieren selbst ist eine Strategie des Kurzzeitgedächtnisses.
Langfristiges Gedächtnis überdauert Sitzungen. Es ermöglicht es einem Agenten, sich daran zu erinnern, dass ein Benutzer kurze Antworten bevorzugt, dass die Bereitstellung letzter Woche aufgrund eines Migrationsproblems fehlgeschlagen ist oder dass sich das Datenbankschema im Januar geändert hat. LangChain definiert es als Gedächtnis, das über Threads und Sitzungen hinweg bestehen bleibt – praktisch bedeutet das, dass diese Daten eine Datenbank benötigen und nicht ein Kontextfenster.
Es gibt zwei grundlegende Arten davon:
- Benutzerpräferenzen sind Personalisierungssignale, die im Laufe der Zeit entstehen, einschließlich Schreibstil, Tonfall, technischer Tiefe, Wahl der Werkzeuge sowie Formatierungsgewohnheiten. Ein Schreibassistent, der sich daran erinnert, dass Sie immer Markdown und einen hohen technischen Anspruch wünschen, ist auf eine Weise nützlich, die kein zustandsloses System erreichen kann – egal wie gut sein Modell auch ist.
Viele Teams machen denselben Fehler. Sie implementieren Kurzzeitgedächtnis, indem sie das Kontextfenster füllen, und Langzeitgedächtnis, indem sie einen Vektorlagers aufbauen. Beides ist falsch oder bestenfalls unvollständig. Das Erste versagt in größeren Skalen aus den oben genannten Gründen. Das Zweite bricht zusammen, da eine Ähnlichkeitssuche keine geeignete Methode ist, um Sitzungsnachrichten, Präferenzen oder Verhaltensregeln abzurufen – außerdem benötigt jedes Arten von Langzeitgedächtnis einen anderen Backend. Das Vier-Schichten-Design ist eine verfeinerte Variante der Trennung in Kurz-/Langzeitgedächtnis, bei der jeder Gedächtnistyp auf die erforderliche Speicherlösung abgebildet wird.
Vier Gedächtnistypen, eine Taxonomie
Teams, die produktive KI-Systeme entwickeln, sind unabhängig voneinander zu derselben Taxonomie gelangt. Die Bezeichnungen variieren, doch die Struktur bleibt dieselbe.
Das im Jahr 2023 von Charles Packer und seinen Kollegen an der UC Berkeley vorgestellte MemGPT-Projekt popularisierte die Idee, das Kontextfenster eines LLM wie RAM zu behandeln – als begrenzte und wertvolle Ressource. Ein Betriebssystem erweitert den Speicher durch Paging und Festplattenspeicher, und Agent-Frameworks können auf ähnliche Weise Langzeitgedächtnis simulieren, indem sie älteren Kontext in größeren Speicherbereiche verschieben und ihn bei Bedarf abrufen. Diese Idee wurde später zur Grundlage von Letta.
Das LangMem-Team von LangChain, das ReMem-Framework von Google DeepMind sowie die Produktionsarchitektur von Mem0 beschreiben alle Varianten von vier Typen:
- Arbeitsspeicher umfasst alles, was im aktuellen Kontext vorhanden ist: aktuelle Gesprächspunkte, abgerufene Dokumente sowie Ausgaben von Tools. Er ist schnell, begrenzt und kostspielig pro Token.
- Episodisches Gedächtnis speichert Aufzeichnungen vergangener Interaktionen, wie beispielsweise, was der Agent für einen bestimmten Benutzer letzten Dienstag getan hat und wie das Ergebnis ausgesehen hat. Es ist chronologisch strukturiert und auf die jeweilige Sitzung beschränkt, wird in der Regel in einem schnellen Schlüssel-Wert-Speicher oder einer SQL-Datenbank gespeichert und über Aktualität oder semantische Ähnlichkeit abgerufen.
- Semantisches Gedächtnis enthält allgemeine Fakten, Fachwissen sowie im Laufe der Zeit erlernte Informationen. Es handelt sich dabei um die RAG-Schicht, die bei jeder Anfrage nach Fachwissen oder früheren Erkenntnissen durch ähnlichkeitsbasierte Embeddings herangezogen wird.
Diese Kategorien spiegeln die Kognitionswissenschaft wider, was erklärt, warum sich Teams auf sie einigen. Sie entsprechen den episodischen, semantischen und prozeduralen Gedächtnissystemen der menschlichen Neurowissenschaften, und das Arbeitsgedächtnis als Kontextfenster ist mehr als nur eine Metapher – es handelt sich dabei um den nächstbesten funktionalen Analogon.
Speicherwahl: SQL, Vektor oder Graph
Welche Datenbank sollte das Gedächtnis speichern? Bevor Code geschrieben wird, lautet die ehrliche Antwort, dass dies vom jeweiligen Gedächtnistyp abhängt.
Vektordatenbanken (Weaviate, Pinecone, Qdrant, Chroma) eignen sich für semantisches Gedächtnis, da sie Suchvorgänge nach Ähnlichkeit in großen Mengen an Vektoren schnell durchführen können. Die meisten bieten außerdem eine hybride Suchmethode, die BM25 mit Vektoren kombiniert und bei der Textsuche deutlich über einer reinen Vektorsuche liegt. Ihnen fehlt jedoch die Fähigkeit zum zeitlichen Reasoning sowie relationale Abfragen, und zwei widersprüchliche Fakten mit nahezu identischen Vektoren können zu Fehlinterpretationen führen. Halten Sie episodische Aufzeichnungen sowie strukturierte Audit-Trail-Daten davon getrennt.
SQL-Datenbanken wie PostgreSQL und SQLite eignen sich für episodisches Gedächtnis. Die Abfragen sind strukturiert nach Benutzer-ID, Sitzung, Aufgabentyp oder Zeitstempel, das Schreibvolumen ist gering, und die ACID-Garantien sind wichtig, wenn ein Agent aufzeichnet, was er getan hat und wann. PostgreSQL mit der pgvector-Erweiterung bietet relationale Abfragen sowie Vektorabfragen in einem einzigen System. Bei weniger als zehn Millionen Datensätzen ist das die praktikable Wahl – mit vertrauten Operationen, günstigem Hosting und SQL zur Fehlersuche.
Graphdatenbanken wie Neo4j gehören nur in ein Speichersystem, wenn episodische oder semantische Daten eine komplexe relationale Struktur aufweisen, die mehrstufige Abfragen erfordert – beispielsweise wer vor drei Monaten, als ein Dokument erstellt wurde, wem Bericht erstattet hat. Einfache Aufgabendaten von Agenten erfüllen diese Kriterien nicht. Viele Teams wenden sich an Graphdatenbanken und stellen fest, dass sie eigentlich SQL benötigen.
Das vorherrschende Muster bei Produktivumgebungen ist die Verwendung relationeller Speicherlösungen (SQL oder PostgreSQL) für Episoden und Benutzerprofile, ein Vektor-Speicher für semantische Abfragen sowie ein In-Process-Speicher für den Arbeitskontext. Für Teams, die weniger Betriebsaufwand wünschen, eignet sich eine Kombination aus PostgreSQL und pgvector. Wählen Sie die Speicherebene entsprechend den Anforderungen des Speichertyps aus – vectorisieren Sie nicht alles.
Ebene 1: Puffermemoria mit Zusammenfassung
Die einfachste Form besteht aus einem Deque, das durch eine feste Anzahl an Nachrichten begrenzt wird:
from collections import deque
from typing import List, Dict
class BufferMemory:
def __init__(self, max_turns: int = 10):
# Each turn = one user message + one assistant message
self.history: deque = deque(maxlen=max_turns * 2)
def add_message(self, role: str, content: str):
self.history.append({"role": role, "content": content})
def get_context(self) -> List[Dict]:
return list(self.history)
def token_estimate(self) -> int:
total_chars = sum(len(m["content"]) for m in self.history)
return total_chars // 4 # rough approximation
def clear(self):
self.history.clear()
Das funktioniert bei kurzen Gesprächen. Bei längeren Gesprächen verschwinden die älteren Nachrichten einfach am Ende, wodurch der Agent den Überblick verliert. Ein besseres Verfahren, das von LangMem als „Hot Path“-Methode bezeichnet wird, fasst die alten Nachrichten zu einem Zusammenfassung zusammen, bevor sie verworfen werden:
import os
from collections import deque
from typing import List, Dict
from openai import OpenAI
class SummarizingBufferMemory:
def __init__(self, max_turns: int = 8, summary_batch: int = 4):
self.client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
self.model = "deepseek/deepseek-v4-flash-0731"
self.recent: deque = deque(maxlen=max_turns * 2)
self.rolling_summary: str = ""
self.summary_batch = summary_batch
def add_message(self, role: str, content: str):
if len(self.recent) == self.recent.maxlen:
self._compress_oldest()
self.recent.append({"role": role, "content": content})
def _compress_oldest(self):
batch = [self.recent.popleft() for _ in range(min(self.summary_batch * 2, len(self.recent)))]
text = "\n".join(f"{m['role']}: {m['content']}" for m in batch)
response = self.client.chat.completions.create(
model=self.model,
max_tokens=300,
messages=[
{
"role": "system",
"content": "You compress conversation history. Reply with the summary only.",
},
{
"role": "user",
"content": (
"Summarize this conversation segment in 2-4 sentences. "
"Preserve any decisions made, constraints stated, and conclusions reached.\n\n"
f"{text}"
),
},
],
)
new_summary = response.choices[0].message.content.strip()
if self.rolling_summary:
self.rolling_summary = f"{self.rolling_summary} | {new_summary}"
else:
self.rolling_summary = new_summary
def get_context(self) -> List[Dict]:
context = []
if self.rolling_summary:
context.append({
"role": "system",
"content": f"[Prior conversation summary: {self.rolling_summary}]",
})
context.extend(list(self.recent))
return context
LangMem trennt die Live-Summarisierung des Hot-Paths, die in Echtzeit abläuft und dem Benutzer Latenz verursacht, von der Hintergrundsummarisierung, die nach der Sitzung ausgeführt wird. Bei benutzerorientierten Anwendungen ist die Hintergrundsummarisierung fast immer besser: Die Zusammenfassungen verbessern sich, da die gesamte Konversation analysiert werden kann, und niemand muss warten. Bei autonomen Agenten, die unüberwacht laufen, ist der Hot-Path akzeptabel, da ohnehin kein Benutzer wartet.
Schicht 2: episodisches Gedächtnis, das seine eigenen Zitate überprüft
Hier versagen viele Teams: Sie implementieren zwar die Abruffunktion, dokumentieren aber nie, was der Agent tatsächlich getan hat.
Der Engineering-Blog von GitHub Copilot beschreibt ein Muster, das es wert ist, nachgeahmt zu werden. Jeder Speicher-Eintrag enthält Quellenangaben, d.h. konkrete Adressen wie Dateipfade, URLs, Tool-Names und Zeilennummern, die es ermöglichen, zum Zeitpunkt der Abfrage zu überprüfen, ob die gespeicherte Information weiterhin gültig ist. GitHub bezeichnet dies als Just-in-Time-Validierung. Wenn der Agent zu Beginn einer Aufgabe einen Speicher-Eintrag lädt, prüft er, ob die angegebenen Quellen weiterhin existieren und die Aussage noch stützen. Falls nicht, wird der Speicher-Eintrag korrigiert oder verworfen, sodass sich der Pool ohne manuelle Nachbearbeitung selbst regeneriert.
GitHub berichtete von einem messbaren Effekt: Der Code-Agent mit Speicherfunktion zeigte eine Steigerung der Fusionsrate von Pull-Requests um 7 %, von 83 % auf 90 %.
import json
import sqlite3
from datetime import datetime, timedelta
from dataclasses import dataclass, field
@dataclass
class Episode:
task_type: str
user_query: str
outcome: str
tools_used: list
citations: list # source references for validation
duration_seconds: float
success: bool
session_id: str = ""
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
class EpisodicMemory:
EXPIRY_DAYS = 28 # GitHub Copilot's production default
def __init__(self, db_path: str = "episodes.db"):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_schema()
def _init_schema(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS episodes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_type TEXT NOT NULL,
user_query TEXT,
outcome TEXT,
tools_used TEXT,
citations TEXT,
duration_s REAL,
success INTEGER,
session_id TEXT,
created_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_task_type ON episodes (task_type, success, created_at);
""")
self.conn.commit()
def record(self, ep: Episode):
self.conn.execute(
"""
INSERT INTO episodes
(task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
ep.task_type, ep.user_query, ep.outcome,
json.dumps(ep.tools_used), json.dumps(ep.citations),
ep.duration_seconds, int(ep.success),
ep.session_id, ep.created_at,
),
)
self.conn.commit()
def recall_similar(self, task_type: str, limit: int = 3) -> list[Episode]:
cutoff = (datetime.utcnow() - timedelta(days=self.EXPIRY_DAYS)).isoformat()
cursor = self.conn.execute(
"""
SELECT task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at
FROM episodes
WHERE task_type = ? AND success = 1 AND created_at > ?
ORDER BY created_at DESC
LIMIT ?
""",
(task_type, cutoff, limit),
)
return [
Episode(
task_type=r[0], user_query=r[1], outcome=r[2],
tools_used=json.loads(r[3]), citations=json.loads(r[4]),
duration_seconds=r[5], success=bool(r[6]),
session_id=r[7], created_at=r[8],
)
for r in cursor.fetchall()
]
def validate_citations(self, episode: Episode, validator_fn) -> bool:
"""
validator_fn(citation: str) -> bool
Check if cited sources are still valid (file exists, URL responds, etc.)
Return False if any citation fails; the episode should be discarded.
"""
return all(validator_fn(c) for c in episode.citations)
Die 28-tägige Gültigkeitsdauer folgt der Produktionsentscheidung von GitHub. Alte Episoden werden nicht gelöscht, sind aber nicht mehr abrufbar, und eine LIMIT-Anfrage an einen nach Datum gefilterten Index ist schnell und erfordert keine Hintergrundbearbeitung.
Ebene 3: semantisches Gedächtnis mit Weaviates hybrider Suche
Das ist die RAG-Ebene. Die meisten Anleitungen behandeln nur die Einrichtung und den Betrieb, während die eigentlichen Produktionsprobleme in der Qualität der Abrufe und der Herkunft der Informationen liegen.
Ein häufiger erster Fehler besteht darin, das semantische Gedächtnis nur für Lesen zu konfigurieren. Ein langzeitiger Agent benötigt es jedoch in Lese-/Schreibmodus: Wenn der Agent während einer Aufgabe eine Tatsache findet, diese überprüft und für sinnvoll hält, sollte sie im semantischen Gedächtnis gespeichert werden, damit sie in zukünftigen Sitzungen verfügbar ist. Der zweite Fehler besteht darin, sich ausschließlich auf die Kosinusähnlichkeit zu verlassen. Weaviates hybride Suche kombiniert das BM25-Schlüsselwortabgleichen mit der Vektorähnlichkeit und weist in der Regel eine bessere Leistung bei der Abrufung aus dem Agenten-Gedächtnis auf als jede Methode für sich genommen.
Drittens – und das ist es, was der Vergiftung von Daten Vorschub leistet – weisen die Einträge oft keine Herkunftsangabe auf. Jeder Eintrag benötigt eine Quelle, einen Zuverlässigkeitswert sowie einen Sitzungsidentifikator. Ohne diese sind eine vertrauensbasierte Suche unmöglich, veraltete oder zweifelhafte Einträge können nicht herabgestuft werden, und das Nachverfolgen einer fehlerhaften Suche wird zur Vermutungssuche.
import uuid as uuid_lib
from datetime import datetime
import weaviate
from weaviate.classes.config import Configure, Property, DataType, VectorDistances
from weaviate.classes.query import MetadataQuery, HybridFusion, Filter
def embed(text: str) -> list[float]:
"""
Swap in any embedding source: OpenAI, Cohere, sentence-transformers, etc.
Returns a normalized float vector.
"""
from sentence_transformers import SentenceTransformer
_model = SentenceTransformer("all-MiniLM-L6-v2")
return _model.encode(text, normalize_embeddings=True).tolist()
class WeaviateSemanticMemory:
COLLECTION = "AgentMemory"
def __init__(self, host: str = "localhost", port: int = 8080):
self.client = weaviate.connect_to_local(host=host, port=port)
self._ensure_collection()
def _ensure_collection(self):
if self.client.collections.exists(self.COLLECTION):
return
self.client.collections.create(
name=self.COLLECTION,
# We provide our own vectors; no built-in vectorizer needed.
# Swap to Configure.Vectorizer.text2vec_openai() if you prefer managed embedding.
vectorizer_config=Configure.Vectorizer.none(),
vector_index_config=Configure.VectorIndex.hnsw(
distance_metric=VectorDistances.COSINE
),
properties=[
Property(name="content", data_type=DataType.TEXT),
Property(name="task_type", data_type=DataType.TEXT),
Property(name="source", data_type=DataType.TEXT),
Property(name="citations", data_type=DataType.TEXT),
Property(name="session_id", data_type=DataType.TEXT),
Property(name="confidence", data_type=DataType.NUMBER),
Property(name="created_at", data_type=DataType.TEXT),
],
)
def store(
self,
content: str,
task_type: str = "",
source: str = "agent",
citations: str = "",
session_id: str = "",
confidence: float = 1.0,
) -> str:
collection = self.client.collections.get(self.COLLECTION)
doc_id = str(uuid_lib.uuid4())
collection.data.insert(
properties={
"content": content,
"task_type": task_type,
"source": source,
"citations": citations,
"session_id": session_id,
"confidence": confidence,
"created_at": datetime.utcnow().isoformat(),
},
vector=embed(content),
uuid=doc_id,
)
return doc_id
def retrieve_hybrid(
self,
query: str,
n_results: int = 5,
min_confidence: float = 0.6,
task_type: str = None,
) -> list[dict]:
collection = self.client.collections.get(self.COLLECTION)
# Filter by confidence floor and optionally by task type
confidence_filter = Filter.by_property("confidence").greater_or_equal(min_confidence)
if task_type:
active_filter = (
Filter.by_property("task_type").equal(task_type) & confidence_filter
)
else:
active_filter = confidence_filter
results = collection.query.hybrid(
query=query,
vector=embed(query),
limit=n_results,
fusion_type=HybridFusion.RELATIVE_SCORE,
filters=active_filter,
return_metadata=MetadataQuery(score=True),
)
return [
{
"content": obj.properties["content"],
"score": obj.metadata.score,
"confidence": obj.properties.get("confidence", 1.0),
"source": obj.properties.get("source", ""),
"citations": obj.properties.get("citations", ""),
"uuid": str(obj.uuid),
}
for obj in results.objects
]
def update_confidence(self, doc_id: str, new_confidence: float):
collection = self.client.collections.get(self.COLLECTION)
collection.data.update(
uuid=doc_id,
properties={"confidence": new_confidence},
)
def close(self):
self.client.close()
Zwei Aspekte in retrieve_hybrid fehlen in vielen Implementierungen. Erstens erhält Weaviates Hybrid-Endpunkt sowohl den rohen Abfragedtext (für BM25) als auch dessen Embedding (für die Vektorsuche). Genaue Schlüsselworttreffer stammen vom ersten und semantische Nähe vom zweiten, wobei die Kombination zuverlässig bessere Ergebnisse liefert als jede Methode für sich allein. Zweitens ist da der min_confidence-Filter, der Erinnerungen mit geringem Vertrauensgrad unabhängig von ihrer semantischen Nähe aus dem Kontext ausschließt. Im Grunde handelt es sich dabei um eine Sicherheitsmaßnahme und nicht nur um eine Anpassung der Relevanz.
Die update_confidence-Methode stellt den Rückkopplungsweg bereit. Wenn das eigene Denken des Agents während einer Aufgabe einer abgerufenen Erinnerung widerspricht, sollte dessen Vertrauensgrad gesenkt werden. Bestätigt eine Zitatenprüfung dies, sollte der Vertrauensgrad erhöht werden. Der Pool sollte aus der Nutzung lernen.
Layer 4: Prozedurales Gedächtnis und Prompt-Entwicklung
Diese Schicht wird am häufigsten übersehen. Das naive Design besteht aus einer JSON-Datei mit Regeln, die vom Agenten gelesen, befolgt und bei Fehlern erweitert werden. Es funktioniert zwar, erreicht aber eine Grenze, da die Regeln in einer separaten Datei um den Kontextraum konkurrieren und eine explizite Abrufung erfordern.
Die Ingenieure von LangMem bei LangChain haben eine Idee veröffentlicht, die sich lohnt: In der Produktion existiert das prozedurale Gedächtnis als der Systemprompt selbst. Es wird weder neben dem Prompt gespeichert noch darin abgerufen – es ist der Prompt. Immer wenn ein Fehler dem Agenten eine neue Verhaltensregel beibringt, wird der Prompt selbst umgeschrieben; diese Praxis nennen sie Prompt-Optimierung als prozedurales Gedächtnis. Gelernte Verhaltensweisen werden zu Anweisungen in natürlicher Sprache, die das Modell am Beginn jeder Interaktion liest.
Die einfachere Version mit JSON-Datei ist der Ausgangspunkt, den die meisten Teams wählen sollten:
import json
import os
import re
from datetime import datetime, timedelta, timezone
from openai import OpenAI
MODEL = "deepseek/deepseek-v4-flash-0731"
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
class ProceduralMemory:
def __init__(self, rules_path: str = "procedural_rules.json"):
self.rules_path = rules_path
self.rules: list[dict] = self._load()
def _load(self) -> list[dict]:
try:
with open(self.rules_path) as f:
return json.load(f)
except FileNotFoundError:
return []
def add_rule(self, situation: str, action: str, reason: str, confidence: float = 1.0):
self.rules.append({
"situation": situation,
"action": action,
"reason": reason,
"confidence": float(confidence),
"added_at": datetime.now(timezone.utc).isoformat(),
"trigger_count": 0,
})
self._save()
def get_applicable_rules(self, context: str, min_confidence: float = 0.7) -> list[dict]:
relevant = []
ctx_lower = context.lower()
dirty = False
for rule in self.rules:
if rule.get("confidence", 1.0) < min_confidence:
continue
keywords = rule["situation"].lower().split()
if not keywords:
continue
hits = sum(1 for kw in keywords if kw in ctx_lower)
if hits >= max(1, len(keywords) // 3):
rule["trigger_count"] = rule.get("trigger_count", 0) + 1
relevant.append(rule)
dirty = True
if dirty:
self._save()
relevant.sort(key=lambda r: r.get("confidence", 1.0), reverse=True)
return relevant[:5]
def prune_stale(self, max_age_days: int = 60, min_triggers: int = 2):
cutoff = (datetime.now(timezone.utc) - timedelta(days=max_age_days)).isoformat()
self.rules = [
r for r in self.rules
if r.get("added_at", "") > cutoff or r.get("trigger_count", 0) >= min_triggers
]
self._save()
def _save(self):
tmp = f"{self.rules_path}.tmp"
with open(tmp, "w") as f:
json.dump(self.rules, f, indent=2)
os.replace(tmp, self.rules_path)
def _parse_json(text: str) -> dict:
text = text.strip()
fenced = re.search(r"```(?:json)?\s*(.*?)```", text, re.S)
if fenced:
text = fenced.group(1).strip()
return json.loads(text)
def extract_rule_from_failure(failure_trace: str, memory: ProceduralMemory):
response = client.chat.completions.create(
model=MODEL,
max_tokens=250,
response_format={"type": "json_object"},
messages=[
{
"role": "system",
"content": "You return only a JSON object. No prose, no code fences.",
},
{
"role": "user",
"content": (
"A task failed. Extract one behavioral rule to prevent this failure.\n"
"Return ONLY valid JSON with keys: situation, action, reason, "
"confidence (0.0-1.0)\n"
f"Failure trace:\n{failure_trace}"
),
},
],
)
try:
rule = _parse_json(response.choices[0].message.content)
except (json.JSONDecodeError, AttributeError, TypeError):
return None
if not all(k in rule for k in ("situation", "action", "reason")):
return None
memory.add_rule(
situation=str(rule["situation"]),
action=str(rule["action"]),
reason=str(rule["reason"]),
confidence=float(rule.get("confidence", 1.0)),
)
return rule
prune_stale bekämpft das Vergessen auf prozeduraler Ebene. Eine nach einem ungewöhnlichen Fehler erstellte Regel, die danach nie wieder ausgelöst wird, sollte nach und nach entfernt werden, während eine Regel, die regelmäßig aktiviert wird, unabhängig von ihrem Alter bestehen bleibt. Es handelt sich um ein einfaches Werkzeug, doch das Vergessen lässt sich genau in dieser Schicht am einfachsten umsetzen.
Sobald ein Agent kontinuierlich läuft, ist der natürliche nächste Schritt, Regeln, denen Vertrauen entgegengebracht wird und die häufig verwendet werden, direkt in den Systemprompt aufzunehmen. LangMem automatisiert dies mithilfe eines Prompt-Optimierungszyklus, während eine manuelle Auswahl für kleinere Systeme völlig ausreicht.
Die Schichten miteinander verbinden
In der folgenden Auflistung wird gezeigt, wie die vier Schichten in der Praxis zusammenwirken. Eine Einschränkung ist unverhandelbar: Das Schreiben in den Speicher darf auf keinen Fall die Antwortzeit verlängern.
import json
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
@dataclass
class TaskResult:
task_type: str
query: str
outcome: str
tools_used: list
citations: list
duration_seconds: float
success: bool
new_facts: list[dict] = field(default_factory=list)
session_id: str = ""
class MemoryManager:
def __init__(
self,
weaviate_host: str = "localhost",
episodic_db: str = "episodes.db",
rules_path: str = "procedural_rules.json",
):
self.buffer = SummarizingBufferMemory(max_turns=8)
self.episodic = EpisodicMemory(db_path=episodic_db)
self.semantic = WeaviateSemanticMemory(host=weaviate_host)
self.procedural = ProceduralMemory(rules_path=rules_path)
self._pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="memory_write")
def build_context(self, query: str, task_type: str) -> list[dict]:
context: list[dict] = []
# Procedural rules first: they constrain behavior throughout the task
rules = self.procedural.get_applicable_rules(query)
if rules:
rules_text = "\n".join(
f"- When '{r['situation']}': {r['action']} (reason: {r['reason']})"
for r in rules
)
context.append({"role": "user", "content": f"[Behavioral rules:\n{rules_text}]"})
# Semantic facts: domain knowledge and past discoveries
facts = self.semantic.retrieve_hybrid(query, n_results=5, min_confidence=0.6)
if facts:
facts_text = "\n".join(f"- {f['content']}" for f in facts)
context.append({"role": "user", "content": f"[Relevant knowledge:\n{facts_text}]"})
# Past episodes: outcome templates for similar tasks
episodes = self.episodic.recall_similar(task_type, limit=3)
if episodes:
ep_text = "\n".join(
f"- Outcome: {e.outcome} (tools: {', '.join(e.tools_used)})"
for e in episodes
)
context.append({"role": "user", "content": f"[Past similar tasks:\n{ep_text}]"})
# Current conversation last: the model reads this most carefully
context.extend(self.buffer.get_context())
return context
def record_turn(self, role: str, content: str):
self.buffer.add_message(role, content)
def persist(self, result: TaskResult):
# Submit to thread pool and return immediately; never block the caller
self._pool.submit(self._persist_worker, result)
def _persist_worker(self, result: TaskResult):
ep = Episode(
task_type=result.task_type,
user_query=result.query,
outcome=result.outcome,
tools_used=result.tools_used,
citations=result.citations,
duration_seconds=result.duration_seconds,
success=result.success,
session_id=result.session_id,
)
self.episodic.record(ep)
for fact in result.new_facts:
self.semantic.store(**fact)
if not result.success:
extract_rule_from_failure(result.outcome, self.procedural)
def shutdown(self):
self._pool.shutdown(wait=True)
self.semantic.close()
Die Reihenfolge des Kontexts in build_context ist beabsichtigt. Forschungen zum Bias bei positionaler Aufmerksamkeit stützen die Annahme, dass Modelle den Anfang und das Ende des Fensters stärker gewichten. Verhaltensregeln kommen an die Spitze, da es sich um Einschränkungen handelt, die alles Folgende prägen sollten. Der aktuelle Dialog steht ganz unten, weil er das ist, was das Modell zum nächsten Antworten am meisten benötigt. Semantische Fakten sowie frühere Episoden befinden sich dazwischen und bieten Referenzen, ohne die aktuelle Aufgabe zu überlasten.
Die Zeile self._pool.submit(self._persist_worker, result) ist die wichtigste ingenieurtechnische Entscheidung in der Klasse. Jeder Schreibaufruf erfolgt sofort, sodass die Antwort des LLM bereits dem Benutzer zugesendet wird, bevor eine Datenbankschreibung beginnt.
Drei Entscheidungen, die die Architektur prägen
Sobald vier Schichten vorhanden sind, beeinflussen drei architektonische Entscheidungen alles Weitere, und viele Teams treffen mindestens eine falsche Entscheidung.
Hot-Path- oder Hintergrundschreibvorgänge. Das LangMem-Team hat viel Mühe in diese Entwicklung investiert. Hot-Path-Schreibvorgänge finden während des Gesprächs statt, wenn der Agent eine Information erkennt, die gespeichert werden sollte, und sofort eine Schreiboperation ausführt. Hintergrundschreibvorgänge erfolgen nach Beendigung der Sitzung, wobei das gesamte Gespräch analysiert wird, um herauszufinden, was aufbewahrt werden sollte. Der Hot-Path verursacht Verzögerungen, da jede semantic.store()-Aufruf im Antwortzyklus eine Datenbankkommitierung darstellt, die der Benutzer mitmachen muss. Hintergrundschreibvorgänge verbessern hingegen die Qualität der Wiedergabe, da Themen und Entscheidungen im gesamten Verlauf untersucht werden können und umfassendere Zusammenfassungen gespeichert werden. Für an Benutzer gerichtete Produktionsanwendungen ist der Hintergrundansatz fast immer der richtige. Der Thread-Pool in persist() ermöglicht den größten Nutzen, ohne einen vollständig asynchronen Ablauf benötigen zu müssen.
Auswahl oder Zusammenfassung. Beides löst unterschiedliche Probleme, und Teams verwechseln sie oft. Bei der Auswahl geht es darum, den relevanten Abschnitt in einem großen Korpus zu finden. Die Abfrage ist spezifisch und der Korpus groß sowie relativ stabil, wie bei Dokumentationen, Produktkatalogen oder Code. Sie versagt, wenn die Abfrage komplex ist, die relevanten Informationen über viele Abschnitte verteilt sind oder sich der Korpus schneller ändert als der Embedding-Index. Bei der Zusammenfassung geht es darum, ein Gespräch oder Dokument zu komprimieren, um es in den Kontext zu passen. Sie funktioniert, wenn die Eingabe sequentiell ist, wie in einer Gesprächshistorie, und ein gewisser Verlust tolerierbar ist; sie versagt hingegen, wenn die weggelassenen Details wichtig sind. Im Jahr 2025 begann die Branche, beide Begriffe durch „Context Engineering“ zu ersetzen: Die Frage lautet nicht mehr Auswahl gegenüber Zusammenfassung, sondern welche Kombination aus gespeichertem Wissen, ausgewählten Abschnitten und komprimierter Historie die beste Lösung bietet.
Der Kontext für diesen spezifischen Inferenzaufruf. Diebuild_context-Methode des Speichermanagers ist eine Form des Kontextengineering, da sie entscheidet, aus welchen der vier Speicherquellen gelesen werden soll, wie viel von jeder genommen wird und in welcher Reihenfolge die Daten zusammengefügt werden. Ursprungsverfolgbarkeit von Anfang an. Man kann Ursprungsverfolgbarkeit nicht später in ein Speichersystem einbauen. Bereits bei der ersten Schreiboperation sollte jede Eintragung ihre Quelle, ihren Zuverlässigkeitswert, die Sitzungs-ID sowie das Erstellungsdatum enthalten. Ohne diese Elemente ist weder eine auf Vertrauenswerten basierende Abruflogik möglich, noch ein Filtern nach Aktualität oder eine Überprüfung fehlerhafter Abfragen. Teams, die in dem ersten Entwicklungszyklus auf Ursprungsverfolgbarkeit verzichten, verbringen anschließend Wochen damit, sie in eine Datenbank einzufügen, die bereits Tausende von Einträgen ohne diese Funktion enthält.
Speichervergiftung: Angriffsmodelle und Abwehrmaßnahmen
Dies ist der Abschnitt, den Speicherdesign-Leitfäden in der Regel weglassen – doch er sollte in jedem enthalten sein.
In einer arXiv-Publikation aus dem Jahr 2025 wurde das MINJA-Angriffsschema vorgestellt, eines der ausführlichsten öffentlichen Analysen zum Memory-Poisoning in LLM-Agenten. Seine Autoren berichten, dass Angreifer durch gewöhnliche Benutzerinteraktionen ohne erhöhte Rechte mit einem dreistufigen Muster Erfolgsraten bei Injektionen von über 95 % erzielen können. Ein Verbindungs Schritt verknüpft ein unschuldiges, legitimes Thema mit dem Zielthema des Angreifers. Ein Indikations Schritt liefert subtile Hinweise, die das schädliche Memory glaubwürdig erscheinen lassen. Ein schrittweises Verkürzungs Schritt komprimiert die Injektion zu einer minimalen Größe, die bei der Speicherauswertung erhalten bleibt.
Was Memory-Poisoning von gewöhnlicher Prompt-Injektion unterscheidet, ist, dass vergiftete Memos die Grenzen zwischen Sitzungen überschreiten und sich gegenseitig verstärken. Eine erfolgreiche Injektion in der ersten Sitzung kann jede nachfolgende Sitzung dauerhaft beeinflussen.
In einer separaten Veröffentlichung aus dem Jahr 2025 zu bayesschen Schutzmechanismen gegen Misstrauensangriffe wurde ein fünfschichtiges Design vorgestellt, das die Angriffsfläche abdeckt:
- Moderation der Eingaben zum Zeitpunkt der Schreibvorgänge. Vor jeder Speicherung wird eine kombinierte Vertrauensbewertung anhand mehrerer unabhängiger Signale berechnet: Herkunft der Quelle, semantische Analyse von Befehmuster sowie Erkennung von Anomalien im Verhalten des Benutzers. Kein einzelnes Signal reicht aus, und MINJA-Injektionen sind so konzipiert, dass sie Prüfungen mit einem einzigen Signal überstehen.
- Ausfiltern von Befehlen vor der Speicherung. Wenn ein potenzieller Datensatz wie ein Befehl aussieht („immer“, „niemals“, „Sie müssen“, „tun Sie nicht“), wird er entweder bereinigt oder weggeworfen, anstatt gespeichert zu werden. Echte, faktenbasierte Datensätze müssen dem Agenten niemals Befehle erteilen.
min_confidence-Filter auf der Weaviate-Seite der dritten Ebene. Erinnerungen mit geringer Provenienz werden unabhängig von ihrer Ähnlichkeit außerhalb des Kontexts belassen. In einer vollständigeren Implementierung wird das Vertrauensfeld anhand von Validierungsresultaten aktualisiert, sodass Erinnerungen, die bestätigt werden, an Vertrauenswerten gewinnen und solche, die widerlegt werden, daran verlieren.Für die meisten Teams ist die sinnvolle Reihenfolge zunächst das Anhängen von Herkunftsmetadaten bei jeder Schreiboperation, anschließend die Erkennung von Anweisungen zum Zeitpunkt der Schreibung und schließlich die abhängige Suche nach Informationen nach dem Vertrauensgrad. Die Verhaltensüberwachung kommt zuletzt, da sie eine Verhaltensgrundlinie benötigt, und deren Erstellung erfordert wochenlange Daten aus der Produktivumgebung.
Noch ein Punkt: Untersuchungen an realen Speichersystemen mit bereits vorhandenen legitimen Einträgen zeigten, dass realistische Bedingungen die Effektivität von Angriffen im Vergleich zu Laboreinrichtungen erheblich verringern, da vergiftete Einträge um die Speicherplätze mit legitimen Informationen konkurrieren. Das ist kein Grund, Schutzmaßnahmen zu vernachlässigen. Es ist vielmehr ein Grund, die legitimen Einträge von hoher Qualität und hohem Vertrauensgrad zu halten, da sie die erste Verteidigungslinie bilden.
Was zuerst entwickelt werden sollte
Nicht alle vier Schichten sind sofort notwendig. Für die meisten KI-Anwendungen decken zwei davon etwa 80 Prozent der Speicherverfehlungen ab: Schicht 1 (Puffer plus Zusammenfassung) und Schicht 3 (Weaviate mit hybrider Suche, Vertrauensfilter und Herkunftsangabe). Gemeinsam bewältigen sie die beiden häufigsten Probleme – ein Agent, der mitten in einer Aufgabe etwas vergisst, und ein Agent ohne Fachwissen. Entwickeln Sie beide Schichten, vergleichen Sie die Aufgabenerledigung vor und nach deren Einsatz und beurteilen Sie erst danach, was fehlt.
Fügen Sie episodisches Gedächtnis hinzu, wenn Sie feststellen, dass Sie sich fragen, warum der Agent vergessen hat, was er in der letzten Sitzung erreicht hat – schließlich ist diese Frage das Signal. Fügen Sie es nicht früher hinzu. Fügen Sie prozedurales Gedächtnis hinzu, wenn derselbe Fehlertyp in verschiedenen Aufgaben wiederkehrt und Prompt-Engineering ihn nicht beheben kann, denn ein Verhaltensmuster benötigt prozedurales Gedächtnis.
PostgreSQL mit pgvector ist die sinnvolle Standardlösung für die meisten Teams. Eine Datenbankinstanz mit CREATE EXTENSION IF NOT EXISTS vector bietet ACID-Konformität, flexible SQL-Anfragen, Vektorähnlichkeitssuche sowie hybride Bewertungsmethoden. Schätzungen zufolge liegen die Hostingkosten bei etwa 45 US-Dollar pro Monat für 10.000 Speicherplätze. Eine eigenständige Vektordatenbank lohnt sich erst dann, wenn das semantische Speichervolumen so groß wird, dass der HNSW-Index von PostgreSQL die Leistung beeinträchtigt – etwas, was die meisten Teams niemals erleben werden.
Sorgen Sie dafür, dass jede Schreiboperation im Speicher asynchron abläuft. Die Reaktion auf den Benutzer sollte nicht von einer Datenbankbestätigung abhängen; übertragen Sie die Aufgabe daher auf einen Thread-Pool und machen Sie weiter.
Achten Sie in der Produktion auf die Vermeidung von im Speicher liegenden Vektorlagern. Eine Implementierung ohne persistenten Speicher verliert alle gesammelten Informationen beim ersten Serverneustart – ein Fehler, den nur wenige Menschen zweimal begehen.