Memory & Experience Replay für Agenten

Episodisches Gedächtnis, Kurzzeit- vs. Langzeitspeicher, Vektor-basiertes Retrieval und Prioritized Experience Replay – wie LLM-Agenten aus vergangenen Erfahrungen lernen und bei neuen Aufgaben gezielt auf bewährte Strategien zugreifen.

1. Das Gedächtnis-Problem

Die vorherigen Artikel haben gezeigt, wie ein Agent aus Feedback lernt (Artikel 02) und wie automatisierte Tests als Reward-Signal dienen (Artikel 03). Doch ein fundamentales Problem blieb ungelöst: Der Agent vergisst alles nach jeder Aufgabe.

Ohne Gedächtnis wiederholt ein Agent dieselben Fehler. Eine Login-Implementierung, die beim dritten Versuch gelang, muss beim nächsten Login-Feature von Null aufgebaut werden. Das Feedback-Signal aus der vorherigen Aufgabe ist verloren.

„Experience replay allows the agent to learn from past experiences by storing transitions in a replay buffer and sampling from them during training." — Mnih et al. (2015), Human-level control through deep reinforcement learning, Nature 518, S. 529–533

In Deep Reinforcement Learning löst Experience Replay genau dieses Problem: Vergangene Erfahrungen werden gespeichert und bei neuen Entscheidungen als Kontext bereitgestellt. Für LLM-Agenten bedeutet das: Episoden aus (Aufgabe, Lösung, Feedback, Ergebnis) werden persistent gespeichert und bei ähnlichen Aufgaben als Kontext in den Prompt eingebettet.

2. Episoden als Datentyp

Der erste Schritt: Eine klare Datenstruktur für Erfahrungen definieren. Jede Interaktion des Agenten – von der Aufgabenstellung über die Lösung bis zum Feedback – bildet eine Episode:

class OutcomeStatus(Enum):
    SUCCESS = "success"
    PARTIAL = "partial"
    FAILURE = "failure"

@dataclass(frozen=True)
class Episode:
    episode_id: str
    task: TaskDescription
    solution: str
    feedback: Feedback
    outcome: OutcomeStatus
    timestamp: float = 0.0
    iterations: int = 1
    strategy_used: str = "default"

Die Aufgabe selbst ist ebenfalls strukturiert – mit Kategorie und Komplexität:

class TaskCategory(Enum):
    CODE_GENERATION = "code_generation"
    DEBUGGING = "debugging"
    REFACTORING = "refactoring"
    TESTING = "testing"
    ARCHITECTURE = "architecture"
    UNKNOWN = "unknown"

@dataclass(frozen=True)
class TaskDescription:
    description: str
    category: TaskCategory = TaskCategory.UNKNOWN
    complexity: float = 0.5

Das Feedback enthält neben dem Reward aus Artikel 03 auch textuelle Informationen:

@dataclass(frozen=True)
class Feedback:
    reward: float
    message: str = ""
    tests_passed: int = 0
    tests_total: int = 0

    @property
    def pass_rate(self) -> float:
        return self.tests_passed / self.tests_total if self.tests_total > 0 else 0.0

Drei Design-Entscheidungen: - Frozen Dataclasses: Episoden sind unveränderlich. Was passiert ist, bleibt gespeichert – keine nachträgliche Manipulation. - Kategorisierung: TaskCategory ermöglicht Skill-Tracking (Abschnitt 8). - Zeitstempel: timestamp ist essentiell für Temporal Decay (Abschnitt 5).

3. Embedding-basiertes Retrieval

Ein Agent braucht nicht alle vergangenen Episoden – nur die relevanten. Die Kernfrage: Wie findet man zu einer neuen Aufgabe die ähnlichsten Erfahrungen?

Die Antwort: Vektor-Embeddings und Cosine Similarity.

„Vector databases store data as high-dimensional vectors, enabling efficient similarity search. They are the backbone of retrieval-augmented generation (RAG) systems." — Ramirez, J. (2024), Generative AI with LangChain, 2. Aufl., S. 189

Cosine Similarity

Zwei Texte werden in Vektoren gleicher Dimension transformiert. Ihre Ähnlichkeit ergibt sich aus dem Kosinus des Winkels zwischen den Vektoren:

def cosine_similarity(a: Embedding, b: Embedding) -> float:
    mag_a = _magnitude(a)
    mag_b = _magnitude(b)
    if mag_a == 0.0 or mag_b == 0.0:
        return 0.0
    return _dot_product(a, b) / (mag_a * mag_b)

Embedding-Abstraktion

Die Embedding-Funktion ist als injizierbarer Parameter gestaltet. Für Tests genügt eine deterministische Hash-basierte Funktion; in Produktion wird sie durch OpenAI Embeddings, Sentence-Transformers oder ähnliche Modelle ersetzt:

Embedding = tuple[float, ...]
EmbedFn = Callable[[str], Embedding]

def simple_hash_embedding(text: str, dim: int = 64) -> Embedding:
    """Deterministische Pseudo-Embedding-Funktion für Tests."""
    values = []
    for i in range(dim):
        h = hash(text + str(i)) % 10000
        values.append((h - 5000) / 5000.0)
    mag = _magnitude(tuple(values))
    if mag == 0.0:
        return tuple(0.0 for _ in range(dim))
    return tuple(v / mag for v in values)

Diese Abstraktion entkoppelt die Architektur von konkreten Embedding-Modellen. Der gesamte Code funktioniert ohne externe Abhängigkeiten – das Embedding-Modell ist ein austauschbarer Baustein.

4. Der Embedding-Store

Der EmbeddingStore verwaltet Paare aus (Embedding, Episode) und ermöglicht Ähnlichkeitssuche:

@dataclass
class EmbeddingStore:
    embed_fn: EmbedFn = field(default=simple_hash_embedding)
    _entries: list[tuple[Embedding, Episode]] = field(
        default_factory=list, init=False
    )

    def add(self, episode: Episode) -> None:
        embedding = self.embed_fn(episode.task.description)
        self._entries.append((embedding, episode))

    def search(self, query: str, top_k: int = 3) -> tuple[SearchResult, ...]:
        if not self._entries:
            return ()
        query_embedding = self.embed_fn(query)
        scored = [
            SearchResult(episode=ep, similarity=cosine_similarity(query_embedding, emb))
            for emb, ep in self._entries
        ]
        scored.sort(key=lambda r: r.similarity, reverse=True)
        return tuple(scored[:top_k])

❌ Naiver Ansatz: Alle Episoden als Kontext

Agent erhält Aufgabe → Alle 500 vergangenen Episoden in den Prompt → Context Window überläuft

Das funktioniert nicht. LLMs haben begrenzte Context Windows, und irrelevante Informationen verschlechtern die Qualität der Antworten.

✅ Vektor-basiertes Retrieval: Top-K ähnlichste Episoden

Agent erhält Aufgabe → Embedding berechnen → Top-3 ähnlichste Episoden abrufen → Fokussierter Kontext

In Produktion wird der lineare Scan durch FAISS, ChromaDB oder Pinecone ersetzt – die Schnittstelle bleibt identisch. Die EmbedFn-Abstraktion ermöglicht diesen Wechsel ohne Änderung der Architektur.

5. Temporal Decay: Aktualität zählt

Nicht alle Erfahrungen sind gleich relevant. Eine Episode von gestern ist wertvoller als eine von vor drei Monaten – die Codebase hat sich weiterentwickelt, Patterns haben sich geändert.

Exponentieller Zeitverfall löst das Problem:

def temporal_decay(
    similarity: float,
    episode_time: float,
    current_time: float,
    half_life: float = 86400.0,
) -> float:
    age = max(0.0, current_time - episode_time)
    decay_factor = math.exp(-math.log(2) * age / half_life)
    return similarity * decay_factor

Die Halbwertszeit bestimmt, nach welcher Zeit das Gewicht auf 50% fällt:

Das Konzept stammt aus der Psychologie: Das Vergessen folgt einer exponentiellen Zerfallskurve (Ebbinghaus, 1885). Für Agenten bedeutet das: Alte Erfahrungen verschwinden nicht – sie verlieren an Gewicht. Eine brillante Lösung von vor einem Monat wird immer noch gefunden, aber eine ähnliche Lösung von gestern wird bevorzugt.

6. Prioritized Experience Replay

Semantische Ähnlichkeit allein reicht nicht. Eine Episode kann ähnlich, aber alt sein. Oder ähnlich, aber fehlgeschlagen. Vier Signale bestimmen den Wert einer Erfahrung:

@dataclass(frozen=True)
class ReplayWeights:
    similarity: float = 0.4
    recency: float = 0.3
    reward: float = 0.2
    outcome_bonus: float = 0.1

Die Prioritäts-Berechnung kombiniert alle vier Signale:

def compute_priority(
    result: SearchResult,
    current_time: float,
    weights: ReplayWeights | None = None,
    half_life: float = 86400.0,
) -> PrioritizedResult:
    w = weights or ReplayWeights()

    sim_score = max(0.0, result.similarity)
    recency_score = temporal_decay(1.0, result.episode.timestamp, current_time, half_life)
    reward_score = result.episode.feedback.reward
    outcome_score = 1.0 if result.episode.outcome == OutcomeStatus.SUCCESS else (
        0.5 if result.episode.outcome == OutcomeStatus.PARTIAL else 0.0
    )

    priority = (
        w.similarity * sim_score
        + w.recency * recency_score
        + w.reward * reward_score
        + w.outcome_bonus * outcome_score
    )

    return PrioritizedResult(
        episode=result.episode,
        similarity=result.similarity,
        priority=priority,
    )

„Prioritized experience replay samples transitions with probability proportional to their temporal-difference error, allowing the agent to learn more frequently from surprising or informative transitions." — Schaul, T. et al. (2016), Prioritized Experience Replay, ICLR 2016

Die Gewichte sind als ReplayWeights-Dataclass konfigurierbar. In der Praxis verschiebt sich die Gewichtung je nach Phase:

In der Frühphase dominiert die Ähnlichkeit – jede relevante Erfahrung zählt. In der Spätphase verschiebt sich das Gewicht zur Recency – bei Hunderten von Episoden sind die neuesten am wertvollsten.

7. Das episodische Gedächtnis

Die bisherigen Bausteine fügen sich zusammen zu einem zweistufigen Gedächtnis:

@dataclass
class EpisodicMemory:
    short_term_limit: int = 5
    embed_fn: EmbedFn = field(default=simple_hash_embedding)
    _short_term: list[Episode] = field(default_factory=list, init=False)
    _long_term: EmbeddingStore = field(init=False)

    def record(self, episode: Episode) -> None:
        self._short_term.append(episode)
        if len(self._short_term) > self.short_term_limit:
            self._short_term = self._short_term[-self.short_term_limit:]
        self._long_term.add(episode)

    def recall(
        self,
        query: str,
        top_k: int = 3,
        current_time: float | None = None,
        weights: ReplayWeights | None = None,
        half_life: float = 86400.0,
    ) -> tuple[PrioritizedResult, ...]:
        now = current_time if current_time is not None else time.time()
        candidates = self._long_term.search(query, top_k=top_k * 2)
        prioritized = tuple(
            compute_priority(r, now, weights, half_life)
            for r in candidates
        )
        ranked = sorted(prioritized, key=lambda p: p.priority, reverse=True)
        return tuple(ranked[:top_k])

Zwei Speicherebenen:

Das Kurzzeitgedächtnis ist ein Sliding Window – es enthält immer die letzten N Episoden, unabhängig von ihrer Relevanz. Das Langzeitgedächtnis speichert alles und ermöglicht gezieltes Retrieval über Embeddings.

Die recall-Methode kombiniert alle bisherigen Konzepte: 1. Semantische Suche im Embedding-Store (Abschnitt 4) 2. Temporal Decay auf die Ergebnisse (Abschnitt 5) 3. Prioritized Replay für die finale Rangfolge (Abschnitt 6)

8. Experience Replay: Kontext für den Agenten

Rohe Episoden sind kein brauchbarer Prompt. Der ReplayContext aggregiert die relevanten Informationen:

@dataclass(frozen=True)
class ReplayContext:
    episodes: tuple[Episode, ...]
    strategies_used: tuple[str, ...]
    avg_reward: float
    success_rate: float

Die Aufbereitung extrahiert Strategien, berechnet Statistiken und dedupliziert:

def build_replay_context(results: tuple[PrioritizedResult, ...]) -> ReplayContext:
    if not results:
        return ReplayContext(episodes=(), strategies_used=(), avg_reward=0.0, success_rate=0.0)

    episodes = tuple(r.episode for r in results)
    strategies = tuple(dict.fromkeys(ep.strategy_used for ep in episodes))
    rewards = [ep.feedback.reward for ep in episodes]
    successes = sum(1 for ep in episodes if ep.outcome == OutcomeStatus.SUCCESS)

    return ReplayContext(
        episodes=episodes,
        strategies_used=strategies,
        avg_reward=sum(rewards) / len(rewards),
        success_rate=successes / len(episodes),
    )

Der formatierte Prompt liefert dem Agenten gezielten Kontext:

def format_replay_prompt(context: ReplayContext, task: TaskDescription) -> str:
    if not context.episodes:
        return f"Aufgabe: {task.description}\nKeine relevanten Erfahrungen vorhanden."

    lines = [
        f"Aufgabe: {task.description}",
        f"Kategorie: {task.category.value}",
        "",
        f"Relevante Erfahrungen ({len(context.episodes)}):",
        f"  Durchschnittlicher Reward: {context.avg_reward:.2f}",
        f"  Erfolgsrate: {context.success_rate:.0%}",
        f"  Bewährte Strategien: {', '.join(context.strategies_used)}",
    ]
    # ... Episode-Details
    return "\n".join(lines)

Das Ergebnis: Der Agent sieht nicht nur die aktuelle Aufgabe, sondern auch, welche Strategien bei ähnlichen Aufgaben funktioniert haben, wie hoch der durchschnittliche Reward war und welche konkreten Fehler in der Vergangenheit auftraten.

9. Skill-Tracking: Stärken und Schwächen erkennen

Episoden enthalten Kategorie-Informationen (TaskCategory). Aggregiert ergibt sich ein Kompetenzprofil:

@dataclass(frozen=True)
class SkillProfile:
    category: TaskCategory
    total_episodes: int
    successes: int
    avg_reward: float
    avg_iterations: float

    @property
    def success_rate(self) -> float:
        return self.successes / self.total_episodes if self.total_episodes > 0 else 0.0

Die Berechnung gruppiert Episoden nach Kategorie:

def compute_skill_profiles(episodes: tuple[Episode, ...]) -> tuple[SkillProfile, ...]:
    by_category: dict[TaskCategory, list[Episode]] = {}
    for ep in episodes:
        cat = ep.task.category
        if cat not in by_category:
            by_category[cat] = []
        by_category[cat].append(ep)

    profiles = []
    for cat in sorted(by_category, key=lambda c: c.value):
        eps = by_category[cat]
        successes = sum(1 for e in eps if e.outcome == OutcomeStatus.SUCCESS)
        rewards = [e.feedback.reward for e in eps]
        iterations = [e.iterations for e in eps]
        profiles.append(SkillProfile(
            category=cat,
            total_episodes=len(eps),
            successes=successes,
            avg_reward=sum(rewards) / len(rewards),
            avg_iterations=sum(iterations) / len(iterations),
        ))
    return tuple(profiles)

Schwache Kategorien werden identifiziert und beeinflussen die Strategie-Wahl:

def identify_weak_skills(
    profiles: tuple[SkillProfile, ...],
    threshold: float = 0.5,
) -> tuple[SkillProfile, ...]:
    return tuple(p for p in profiles if p.success_rate < threshold)

10. Adaptive Strategie-Selektion

Das Gedächtnis ermöglicht erfahrungsbasierte Strategie-Wahl. Statt einer festen Strategie wählt der Agent basierend auf drei Signalen:

STRATEGY_POOL: dict[str, str] = {
    "conservative": "Generiere minimalen, sicheren Code. Vermeide Komplexität.",
    "exploratory": "Probiere neue Ansätze. Nutze moderne Patterns.",
    "defensive": "Fokus auf Error-Handling und Edge-Cases.",
    "iterative": "Starte einfach, erweitere schrittweise.",
    "example_driven": "Orientiere dich an erfolgreichen Beispielen.",
}

Die Selektion folgt einer klaren Hierarchie:

def select_strategy(
    task: TaskDescription,
    memory: EpisodicMemory,
    current_time: float | None = None,
) -> str:
    now = current_time if current_time is not None else time.time()
    results = memory.recall(task.description, top_k=5, current_time=now)
    context = build_replay_context(results)

    # 1. Erfolgreiche Strategien bevorzugen
    if context.success_rate > 0.6 and context.strategies_used:
        strategy = context.strategies_used[0]
        if strategy in STRATEGY_POOL:
            return strategy

    # 2. Schwache Kategorien → defensive Strategie
    # ... (Skill-Tracking-basiert)

    # 3. Komplexitätsbasierte Auswahl
    if task.complexity > 0.7:
        return "iterative"
    if task.complexity < 0.3:
        return "exploratory"

    return "iterative"

Die Entscheidungslogik: 1. Erfahrung: Falls ähnliche Aufgaben mit einer bestimmten Strategie erfolgreich waren → dieselbe Strategie verwenden 2. Schwäche: Falls die Aufgabenkategorie als schwach identifiziert wurde → defensive Strategie 3. Komplexität: Hohe Komplexität → iterativ. Niedrige Komplexität → explorativ

Diese drei Ebenen bilden einen Fallback-Mechanismus: Ohne Erfahrung entscheidet die Komplexität. Mit Erfahrung dominiert das Wissen aus vergangenen Episoden.

11. Lernkurve und Diagnostik

Die Episodenhistorie ist nicht nur Kontext für den Agenten – sie ist ein Diagnosewerkzeug:

@dataclass(frozen=True)
class LearningSnapshot:
    episode_number: int
    cumulative_reward: float
    rolling_success_rate: float
    rolling_avg_reward: float

Die Lernkurve berechnet gleitende Durchschnitte über ein konfigurierbares Fenster:

def compute_learning_curve(
    episodes: tuple[Episode, ...],
    window: int = 5,
) -> tuple[LearningSnapshot, ...]:
    snapshots = []
    cumulative = 0.0
    for i, ep in enumerate(episodes):
        cumulative += ep.feedback.reward
        start = max(0, i - window + 1)
        window_eps = episodes[start:i + 1]
        rolling_successes = sum(
            1 for e in window_eps if e.outcome == OutcomeStatus.SUCCESS
        )
        rolling_rewards = [e.feedback.reward for e in window_eps]
        snapshots.append(LearningSnapshot(
            episode_number=i + 1,
            cumulative_reward=cumulative,
            rolling_success_rate=rolling_successes / len(window_eps),
            rolling_avg_reward=sum(rolling_rewards) / len(rolling_rewards),
        ))
    return tuple(snapshots)

Drei Metriken zeigen, ob das Memory-System funktioniert:

Wenn die Rolling Success Rate steigt, nutzt der Agent vergangene Erfahrungen effektiv. Wenn sie stagniert, ist das Retrieval möglicherweise nicht präzise genug – oder die gespeicherten Erfahrungen sind nicht aussagekräftig.

12. Vergleich: Memory-Architekturen

Die Empfehlung: Starte mit JSON-Log und Context Window. Wechsle zum episodischen Gedächtnis mit Vektor-DB, sobald der Agent regelmäßig ähnliche Aufgaben bearbeitet und die Erfahrungsbasis wächst.

13. Fazit

Memory & Experience Replay transformieren einen zustandslosen Agenten in ein lernendes System:

• Episoden strukturieren Erfahrungen als unveränderliche Datensätze mit Aufgabe, Lösung, Feedback und Ergebnis
• Embedding-basiertes Retrieval findet relevante Erfahrungen via Cosine Similarity, ohne alle Episoden durchsuchen zu müssen
• Temporal Decay gewichtet aktuelle Erfahrungen höher als veraltete – analog zum menschlichen Vergessen
• Prioritized Experience Replay kombiniert Ähnlichkeit, Aktualität, Reward und Outcome zu einem robusten Relevanz-Score
• Zweistufiges Gedächtnis trennt Kurzzeit (Sliding Window) und Langzeit (Vektor-Store) für unterschiedliche Zugriffsmuster
• Skill-Tracking identifiziert Stärken und Schwächen, um die Strategie-Wahl anzupassen
• Lernkurven machen den Fortschritt sichtbar und diagnostizierbar

Die Architektur ist bewusst modular: Die Embedding-Funktion ist injizierbar, der Store ist austauschbar (von In-Memory bis FAISS), und die Prioritäts-Gewichtung ist konfigurierbar. Die Intelligenz sitzt nicht im Speicher, sondern in der Kombination aus Retrieval, Priorisierung und Kontextaufbereitung.

Der nächste Artikel der Serie behandelt Das virtuelle Dev-Team – wie mehrere spezialisierte Agenten mit unterschiedlichen Rollen zusammenarbeiten und sich gegenseitig verbessern.

Quellen & Ressourcen

• Mnih, V. et al. (2015): Human-level control through deep reinforcement learning. Nature 518, S. 529–533. – Experience Replay als Kernmechanismus im Deep RL.
• Schaul, T. et al. (2016): Prioritized Experience Replay. ICLR 2016. – Priorisierung nach TD-Error, übertragbar auf Agenten-Systeme.
• Ramirez, J. (2024): Generative AI with LangChain. 2. Aufl. Packt. S. 189 ff. – Vektor-Datenbanken, Embedding-Modelle, RAG-Architektur.
• Labonne, M. et al. (2025): LLM Engineer's Handbook. Packt. S. 331 ff. – RLHF, Feedback-Loops, Policy-Optimierung.
• Vaid, S. & Lund, B. (2025): LLM Design Patterns. Packt. S. 452 ff. – Memory-Patterns für LLM-Agenten, Kontext-Management.
• Ebbinghaus, H. (1885): Über das Gedächtnis. Duncker & Humblot. – Vergessenskurve als Grundlage für Temporal Decay.

→ Vollständiger Code: src/