Entwurf einer Vier-Ebenen-Agenten-Speicherlösung mit LangGraph und Amazon Bedrock
Erlernen Sie, LLM-Agenten auf Bedrock und LangGraph eine funktionierende episodische, semantische sowie prozedurale Erinnerung zu verleihen und sie vor Vergiftung, Datenleckagen von PII sowie Informationsfluss zu Dritten zu schützen.
Quellen: Agentic Architectures — Article 7: Agent Memory Architectures
Erlaubt man es, wird früher oder später ein Produktionsagent dieselbe Arbeit zweimal erledigen. Nichts ist kaputt und die Anfrage ist in Ordnung; der Agent erinnert sich einfach nicht daran, letzte Woche ein fast identisches Problem gelöst zu haben. Deshalb durchläuft er wieder die gleichen Schritte, macht dieselben Zwischenfehler, kommt zum selben Ergebnis – und man muss erneut für jeden Token bezahlen. Dieser Leitfaden zeigt, wie man dieses Problem mit einer echten Speicherkonstruktion statt mit einem größeren Kontextfenster beheben kann: vier unterschiedliche Speicherebenen, die mit LangGraph auf Amazon Bedrock erstellt wurden, sowie Schutzmaßnahmen, die verhindern, dass der Speicher zu einer Belastung wird.
Stellen Sie sich einen Vertragsanalyse-Agenten vor, der über eine solide Infrastruktur verfügt und in einer funktionsfähigen Mehr-Agenten-Umgebung arbeitet – doch er verwendet bei jeder ähnlichen Vertragsanalyse weiterhin seine ersten Toolaufrufe dazu, Strukturmustern nachzugehen, die er bereits zuvor ermittelt hat. Was ihm fehlt, ist Speicher.
Ein größeres Kontextfenster ist kein Speicher
Sobald der Kontextbereich voll ist, ist die übliche Reaktion ein größeres Fenster oder eine aggressivere Zusammenfassung. Beides hilft während einer Aufgabe – doch keines davon ersetzt das Gedächtnis. Eine nützliche Analogie: Das Kontextfenster ist eine Tafel, die während eines Meetings praktisch ist und am Ende des Meetings ausgewischt wird. Das Gedächtnis hingegen ist das Notizbuch, das man von einem Meeting zum nächsten mitnimmt.
Die Kontextverwaltung versagt in drei konkreten Aspekten:
- Sie verbindet Aufgaben nicht miteinander. Wenn der Agent seine Arbeit beendet und sein Prozess endet, verschwindet der Kontext. Alles, was gelernt wurde, geht verloren, es sei denn, man speichert es explizit.
- Sie überschreitet nicht die Grenzen zwischen Agenten. In einem Mehr-Agentensystem hat jeder Worker seinen eigenen Kontext. Was ein Mitarbeiter für die Sicherheitsprüfung entdeckt, ist für einen Mitarbeiter zur Codeanalyse unsichtbar, es sei denn, man leitet es absichtlich dorthin weiter.
Memory adressiert ein anderes Bedürfnis: einen persistenten, abrufbaren Wissensspeicher, der länger besteht als einzelne Ausführungen und über jede beliebige Agentenarchitektur hinweg geteilt werden kann.
Die vier Ebenen des Agentenmemoriums
Jede Ebene erfordert unterschiedliche Infrastruktur, Abrufmuster und Fehlermodi:
+---------------------+--------------------------------------------------+----------------------------+
| Memory Tier | What It Stores | Survives Run End? |
+---------------------+--------------------------------------------------+----------------------------+
| Working Memory | Active context during current run | No |
| | Tool outputs, reasoning steps, partial results | |
+---------------------+--------------------------------------------------+----------------------------+
| Episodic Memory | Summaries of past runs | Yes |
| | What happened, what worked, what failed | |
+---------------------+--------------------------------------------------+----------------------------+
| Semantic Memory | Structured knowledge about the domain | Yes |
| | Facts, relationships, learned patterns | |
+---------------------+--------------------------------------------------+----------------------------+
| Procedural Memory | Successful approaches to recurring task types | Yes |
| | How to do things, not just what was done | |
+---------------------+--------------------------------------------------+----------------------------+
Die Ebenen interagieren in einem Kreislauf. Das Arbeitsgedächtnis speichert die aktuelle Ausführung. Wenn diese beendet ist, führt ein Destillierungsprozess Schreibvorgänge in die drei persistenten Ebenen durch. Wenn die nächste Ausführung beginnt, holt ein Abrufprozess das relevante Material wieder ab:
Run starts
|
v
+------------------+
| Working Memory | <-- active context, tool outputs, LangGraph state
| (current run) |
+------------------+
|
| Run ends: distillation step
v
+------------------+ +------------------+ +------------------+
| Episodic Memory | | Semantic Memory | | Procedural Memory|
| "what happened" | | "what I know" | | "how I do this" |
| DynamoDB | | Bedrock KB | | DynamoDB + KB |
+------------------+ +------------------+ +------------------+
| | |
+------------------------+------------------------+
|
Next run starts: retrieval step
|
v
Agent bootstrapped with
relevant prior knowledge
Die episodische Erinnerung teilt dem nächsten Lauf mit, was geschehen ist, die semantische Erinnerung enthält das Wissen über den jeweiligen Bereich und die prozedurale Erinnerung beschreibt, wie man eine Aufgabe angehen soll. Alle drei Arten von Erinnerung werden beim Start abgerufen und in das Arbeitsgedächtnis eingefügt, bevor der Agent seinen ersten Schlussfolgerungs Schritt unternimmt.
Ebene 1: Verwaltung des Arbeitsgedächtnisses innerhalb eines Laufs
Das Arbeitsgedächtnis umfasst alles, was während eines einzelnen Laufs vorhanden ist, und Sie verwalten es bereits – unabhängig davon, ob Sie es so nennen oder nicht. In LangGraph ist der Graphzustand das Arbeitsgedächtnis: Jede Nachricht, jede Tool-Ausgabe sowie jedes Zwischenergebnis befinden sich dort.
Das Problem ist, dass es ständig wächst. Die Kontextkompression zwischen den Pipeline-Stufen, eine gängige Multi-Agenten-Technik, ist im Grunde nichts anderes als Memory-Management unter einem anderen Namen. Die untenstehende Klasse macht diese Idee zu einem wiederverwendbaren Komponenten. maybe_compress wird vor jedem Agentenschritt ausgeführt; sobald die Nachrichtenliste einen Schwellenwert überschreitet (standardmäßig 30), werden die Systemnachrichten sowie die 15 neuesten nicht-systemischen Nachrichten unverändert beibehalten, während alles Ältere durch eine einzige Zusammenfassung ersetzt wird, die von einem kleinen, kostengünstigen Modell erzeugt wird. Eine zweite Methode, extract_run_artifacts, wird am Ende der Ausführung genutzt und bittet das Modell um eine strukturierte JSON-Dokumentation der Aufgabe, des Ergebnisses, der Erkenntnisse, des Ansatzes, der Tools sowie der wiederverwendbaren Muster – diese dient anschließend als Eingang für das episodische Gedächtnis. Der Codeausschnitt ist in Python, nicht in JavaScript:
# harness/memory/working_memory.py
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, AIMessage
from langchain_aws import ChatBedrock
import boto3
class WorkingMemoryManager:
"""
Manages the active context within a single agent run.
Applies a sliding window with summarization to prevent context overflow.
"""
def __init__(
self,
max_messages: int = 40,
summarize_after: int = 30,
region: str = "us-east-1",
):
self.max_messages = max_messages
self.summarize_after = summarize_after
bedrock = boto3.client("bedrock-runtime", region_name=region)
# Haiku for summarization: cheap and fast for mechanical compression
self.summarizer = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-haiku-4-5",
model_kwargs={"temperature": 0, "max_tokens": 2000},
)
def maybe_compress(self, messages: List[BaseMessage]) -> List[BaseMessage]:
"""
Called before each agent step. If the message list is getting long,
compress the oldest messages into a summary and keep the recent ones.
"""
if len(messages) < self.summarize_after:
return messages
# Keep the system message and the most recent messages
system_messages = [m for m in messages if isinstance(m, SystemMessage)]
non_system = [m for m in messages if not isinstance(m, SystemMessage)]
# Split: old messages get summarized, recent ones stay verbatim
cutoff = len(non_system) - 15 # always keep last 15 exchanges verbatim
to_summarize = non_system[:cutoff]
to_keep = non_system[cutoff:]
if not to_summarize:
return messages
summary_text = self._summarize(to_summarize)
summary_message = SystemMessage(
content=f"[WORKING MEMORY SUMMARY - earlier in this run]\n{summary_text}"
)
return system_messages + [summary_message] + to_keep
def _summarize(self, messages: List[BaseMessage]) -> str:
messages_text = "\n".join([
f"{type(m).__name__}: {m.content[:500]}"
for m in messages
])
response = self.summarizer.invoke([
SystemMessage(content="""
Summarize these agent messages into a compact working memory record.
Include: decisions made, tools called and their results, key findings,
and anything the agent will need to continue its work.
Discard: failed attempts, repetitive reasoning, tool errors that were resolved.
"""),
HumanMessage(content=messages_text)
])
return response.content
def extract_run_artifacts(self, messages: List[BaseMessage]) -> dict:
"""
At end of run, extracts key artifacts for episodic memory storage.
Separates what happened from how it happened.
"""
response = self.summarizer.invoke([
SystemMessage(content="""
Extract the following from these agent messages and return as JSON:
{
"task_summary": "one sentence description of what was asked",
"outcome": "COMPLETED | PARTIAL | FAILED",
"key_findings": ["list of important discoveries"],
"approach_used": "brief description of the approach that worked",
"tools_used": ["list of tool names"],
"time_taken_steps": 0,
"reusable_patterns": ["any patterns that could help on similar future tasks"]
}
Return only valid JSON.
"""),
HumanMessage(content="\n".join([
f"{type(m).__name__}: {m.content[:300]}"
for m in messages[-30:] # last 30 messages are enough
]))
])
import json
try:
return json.loads(response.content)
except json.JSONDecodeError:
return {"outcome": "UNKNOWN", "key_findings": [], "reusable_patterns": []}
Die Zusammenfassungsfunktion ignoriert fehlgeschlagene Versuche sowie behobene Toolspezifikationsfehler, und die JSON-Parse verwendet bei Fehlern einen UNKNOWN-Status zusammen mit leeren Listen, sodass eine fehlerhafte Antwort den gesamten Ablauf niemals zum Abbruch bringt. Modellidentifikatoren wie die hier angezeigte Haiku-ID ändern sich im Laufe der Zeit und erfordern möglicherweise ein regionsspezifisches Inferenzprofil; prüfen Sie daher vor dem Kopieren die aktuelle Bedrock-Modellliste.
Zum lokalen Entwickeln können Sie den Bedrock-Zusammenfasser durch ein von Ollama bereitgestelltes Modell ersetzen:
# Local dev alternative
# ollama pull llama3.1:8b
from langchain_ollama import ChatOllama
summarizer = ChatOllama(model="llama3.1:8b", temperature=0, format="json")
Ebene 2: episodisches Gedächtnis früherer Ausführungen
Das episodische Gedächtnis beantwortet die Frage „Was ist zuvor passiert?“ Zu Beginn einer Ausführung holt der Agent Zusammenfassungen früherer Ausführungen ab, die der aktuellen Aufgabe ähneln, und nutzt sie, um überflüssige Arbeiten zu vermeiden sowie bekannte Fehler zu umgehen.
Das Design speichert verdichtete Zusammenfassungen anstelle von rohen Ausführungsprotokollen. Vollständige Protokolle sind teuer in der Speicherung und langsam bei der Suche; kompakte Zusammenfassungen hingegen sind günstig und schnell. Die Speicherung erfolgt in DynamoDB, wobei zur Abfrage semantischer Ähnlichkeit Bedrock-Embeddings verwendet werden.
Die in Terraform definierte Tabelle verwendet agent_id als Teilungsschlüssel und run_id als Sortierschlüssel. Ein globaler sekundärer Index auf agent_id und created_at ermöglicht Abfragen nach „den neuesten Episoden für diesen Agenten“, und das TTL-Feature von DynamoDB auf expires_at löscht alte Einträge automatisch:
# infrastructure/memory_tables.tf
resource "aws_dynamodb_table" "episodic_memory" {
name = "agent-episodic-memory"
billing_mode = "PAY_PER_REQUEST"
hash_key = "agent_id"
range_key = "run_id"
attribute {
name = "agent_id"
type = "S"
}
attribute {
name = "run_id"
type = "S"
}
attribute {
name = "created_at"
type = "N"
}
global_secondary_index {
name = "agent-time-index"
hash_key = "agent_id"
range_key = "created_at"
projection_type = "ALL"
}
ttl {
attribute_name = "expires_at"
enabled = true
}
tags = {
Component = "agent-memory"
}
}
Der Store integriert selbst die Zusammenfassung der Aufgabe zusammen mit den wichtigsten Erkenntnissen, speichert die Eintragung mit einer Gültigkeitsdauer von 90 Tagen und vergleicht bei der Abrufung die Darstellung der aktuellen Aufgabe mit jüngsten Episoden mithilfe der Kosinusähnlichkeit. Nur Episoden mit einer Ähnlichkeitsgrenze von über 0,75 werden beibehalten; bei Gleichstand wird bevorzugt das Ergebnis COMPLETED gewählt; und format_for_injection stellt die Gewinner als beschrifteten Block für den Systemprompt dar:
# harness/memory/episodic.py
import boto3
import time
import json
import uuid
from typing import List, Optional
from dataclasses import dataclass
from langchain_aws import BedrockEmbeddings
import numpy as np
@dataclass
class EpisodeRecord:
run_id: str
agent_id: str
task_summary: str
outcome: str
key_findings: List[str]
approach_used: str
tools_used: List[str]
reusable_patterns: List[str]
embedding: Optional[List[float]]
created_at: float
expires_at: float
class EpisodicMemoryStore:
"""
Stores and retrieves summaries of past agent runs.
Uses embeddings for semantic retrieval so the agent can find
relevant episodes even when the wording is different.
"""
RETENTION_DAYS = 90
def __init__(
self,
agent_id: str,
table_name: str = "agent-episodic-memory",
region: str = "us-east-1",
):
self.agent_id = agent_id
self.region = region
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
bedrock = boto3.client("bedrock-runtime", region_name=region)
self.embeddings = BedrockEmbeddings(
client=bedrock,
model_id="amazon.titan-embed-text-v2:0",
)
def store_episode(self, artifacts: dict) -> str:
"""
Called at the end of each run to store what happened.
Returns the run_id of the stored episode.
"""
run_id = str(uuid.uuid4())
now = time.time()
# Embed the task summary for semantic retrieval later
text_to_embed = artifacts.get("task_summary", "") + " " + \
" ".join(artifacts.get("key_findings", []))
embedding = self.embeddings.embed_query(text_to_embed)
record = {
"run_id": run_id,
"agent_id": self.agent_id,
"task_summary": artifacts.get("task_summary", ""),
"outcome": artifacts.get("outcome", "UNKNOWN"),
"key_findings": artifacts.get("key_findings", []),
"approach_used": artifacts.get("approach_used", ""),
"tools_used": artifacts.get("tools_used", []),
"reusable_patterns": artifacts.get("reusable_patterns", []),
"embedding": embedding,
"created_at": int(now),
"expires_at": int(now + self.RETENTION_DAYS * 86400),
}
self.table.put_item(Item=record)
return run_id
def retrieve_relevant_episodes(
self,
current_task: str,
max_results: int = 5,
min_similarity: float = 0.75,
) -> List[EpisodeRecord]:
"""
Finds past episodes semantically similar to the current task.
Returns them ranked by relevance, filtered by outcome quality.
"""
query_embedding = self.embeddings.embed_query(current_task)
# Fetch recent episodes for this agent
# In production, you'd paginate this and use a vector DB for scale
response = self.table.query(
IndexName="agent-time-index",
KeyConditionExpression="agent_id = :aid",
ExpressionAttributeValues={":aid": self.agent_id},
ScanIndexForward=False, # newest first
Limit=100,
)
episodes = []
for item in response.get("Items", []):
stored_embedding = item.get("embedding")
if not stored_embedding:
continue
similarity = self._cosine_similarity(query_embedding, stored_embedding)
if similarity >= min_similarity:
episodes.append((similarity, item))
# Sort by similarity, take top results, prefer COMPLETED outcomes
episodes.sort(key=lambda x: (x[0], x[1].get("outcome") == "COMPLETED"), reverse=True)
return [
EpisodeRecord(**{k: v for k, v in item.items() if k != "embedding"}, embedding=None)
for _, item in episodes[:max_results]
]
def format_for_injection(self, episodes: List[EpisodeRecord]) -> str:
"""
Formats retrieved episodes into a system prompt block
for injection at the start of a new run.
"""
if not episodes:
return ""
lines = ["[EPISODIC MEMORY: relevant past runs]\n"]
for ep in episodes:
lines.append(f"Task: {ep.task_summary}")
lines.append(f"Outcome: {ep.outcome}")
if ep.key_findings:
lines.append("Key findings:")
for f in ep.key_findings:
lines.append(f" - {f}")
if ep.reusable_patterns:
lines.append("Patterns that worked:")
for p in ep.reusable_patterns:
lines.append(f" - {p}")
lines.append("")
return "\n".join(lines)
def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
a, b = np.array(a), np.array(b)
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))
Die Abfrage lädt höchstens 100 aktuelle Episoden und berechnet die Ähnlichkeit im Anwendungscode; wie der Kommentar erwähnt, sollten größere Implementierungen paginiert werden und eine Vektordatenbank verwenden. Beachten Sie außerdem: Die boto3 DynamoDB-Ressource lehnt Python float-Werte ab, weshalb die Embedding-Liste in Decimal umgewandelt werden muss (oder serialisiert, beispielsweise als JSON oder Binärdaten), bevor put_item sie akzeptiert. Das hier verwendete Embedding-Modell wird in der Amazon Titan Embeddings-Dokumentation beschrieben.
Ebene 3: semantisches Gedächtnis mit Bedrock Knowledge Bases
Episodisches Gedächtnis speichert Ereignisse; semantisches Gedächtnis speichert Wissen. Der Unterschied ist subtil, aber wichtig. Ein Episodenbeispiel könnte lauten: „Ein Lieferantenvertrag, der letzten Dienstag analysiert wurde, enthielt eine nicht standardmäßige Klausel 7.3“. Die entsprechende semantische Information lautet: „Klausel 7.3 in Lieferantenverträgen bezieht sich in der Regel auf Haftungsgrenzen; der übliche Bereich beträgt das Doppelte des Vertragswerts, und alles Unterhalb dieses Wertes ist ein Warnsignal.“
Semantische Informationen sind länger haltbar als Episoden. Sie verfallen nicht nach 90 Tagen; sie werden im Laufe der Zeit weiter verfeinert und helfen bei vielen Arten von Aufgaben, nicht nur bei solchen, die einer früheren Ausführung ähneln.
Bedrock Knowledge Bases ist der ideale Ort für diese Art von Informationen, da es das Aufteilen in Blöcke, das Einbetten, die Speicherung und das Abrufen ermöglicht. Was die meisten Teams bisher nicht umgesetzt haben, ist der Schreibweg, da Knowledge Bases in der Regel als nur zum Lesen vorgesehen gelten. Der untenstehende Ansatz deckt beide Richtungen ab:
retrieveführt eine semantische Vektorsuche durch, die optional nach einerdomain-Metadaten-Schlüssel gefiltert wird, und ignoriert Ergebnisse mit einer Bewertung unter 0,6.store_factschreibt die Information als JSON-Dokument in S3 und startet einen Einpflegungsprozess, damit die Wissensdatenbank sie aufnehmen kann.extract_facts_from_episodebittet ein Modell, generalisierbare Fakten aus einer abgeschlossenen Ausführung zu extrahieren, überspringt task-spezifische Details, die in der episodischen Erinnerung aufbewahrt werden sollten, und behält nur Fakten mit einer Zuverlässigkeit von mindestens 0,7 bei.
# harness/memory/semantic.py
import boto3
import json
import uuid
import time
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class SemanticFact:
fact_id: str
content: str
domain: str # "contract_law", "security_patterns", etc.
confidence: float # 0.0 to 1.0
source_run_id: str
created_at: float
last_confirmed_at: float
class SemanticMemoryStore:
"""
Manages domain knowledge that persists across runs.
Reads from Bedrock Knowledge Base.
Writes new facts to S3 and triggers KB sync.
"""
def __init__(
self,
knowledge_base_id: str,
s3_bucket: str,
s3_prefix: str = "semantic-memory/",
region: str = "us-east-1",
):
self.kb_id = knowledge_base_id
self.s3_bucket = s3_bucket
self.s3_prefix = s3_prefix
self.region = region
self.bedrock_agent = boto3.client("bedrock-agent-runtime", region_name=region)
self.bedrock_agent_mgmt = boto3.client("bedrock-agent", region_name=region)
self.s3 = boto3.client("s3", region_name=region)
def retrieve(
self,
query: str,
domain: Optional[str] = None,
max_results: int = 5,
min_score: float = 0.6,
) -> List[str]:
"""
Retrieves relevant semantic facts for the current task.
Called at agent startup to bootstrap domain knowledge.
"""
retrieval_config = {
"vectorSearchConfiguration": {
"numberOfResults": max_results,
"overrideSearchType": "SEMANTIC",
}
}
if domain:
retrieval_config["vectorSearchConfiguration"]["filter"] = {
"equals": {"key": "domain", "value": domain}
}
response = self.bedrock_agent.retrieve(
knowledgeBaseId=self.kb_id,
retrievalQuery={"text": query},
retrievalConfiguration=retrieval_config,
)
return [
r["content"]["text"]
for r in response.get("retrievalResults", [])
if r.get("score", 0) >= min_score
]
def store_fact(self, fact: SemanticFact) -> str:
"""
Writes a new fact to S3 and triggers a KB ingestion job.
The fact will be available for retrieval after the sync completes
(typically 1-5 minutes).
For time-sensitive use cases, consider a separate fast-path
retrieval from DynamoDB alongside the KB retrieval.
"""
document = {
"fact_id": fact.fact_id,
"content": fact.content,
"domain": fact.domain,
"confidence": fact.confidence,
"source_run_id": fact.source_run_id,
"created_at": fact.created_at,
"metadata": {
"domain": fact.domain,
"confidence": str(fact.confidence),
}
}
key = f"{self.s3_prefix}{fact.domain}/{fact.fact_id}.json"
self.s3.put_object(
Bucket=self.s3_bucket,
Key=key,
Body=json.dumps(document),
ContentType="application/json",
)
# Trigger sync so new fact becomes retrievable
self.bedrock_agent_mgmt.start_ingestion_job(
knowledgeBaseId=self.kb_id,
dataSourceId=self._get_data_source_id(),
)
return fact.fact_id
def extract_facts_from_episode(
self,
episode_artifacts: dict,
extractor_model,
domain: str,
) -> List[SemanticFact]:
"""
At end of a successful run, extracts generalizable facts
from the episode and prepares them for semantic memory storage.
Only extracts from COMPLETED runs with confidence.
"""
if episode_artifacts.get("outcome") != "COMPLETED":
return []
from langchain_core.messages import SystemMessage, HumanMessage
response = extractor_model.invoke([
SystemMessage(content=f"""
Extract generalizable domain facts from this completed agent run.
Domain: {domain}
Rules:
- Only extract facts that would be useful across many different tasks
- Do not extract task-specific details (those belong in episodic memory)
- Rate confidence 0.0 to 1.0 based on how certain the agent was
- Skip anything that might be specific to this one instance
Return JSON array:
[
{{"content": "the fact", "confidence": 0.9}},
...
]
Return empty array if no generalizable facts found.
"""),
HumanMessage(content=json.dumps(episode_artifacts))
])
try:
facts_data = json.loads(response.content)
except json.JSONDecodeError:
return []
now = time.time()
return [
SemanticFact(
fact_id=str(uuid.uuid4()),
content=f["content"],
domain=domain,
confidence=f.get("confidence", 0.7),
source_run_id=episode_artifacts.get("run_id", "unknown"),
created_at=now,
last_confirmed_at=now,
)
for f in facts_data
if f.get("confidence", 0) >= 0.7 # only store high-confidence facts
]
def _get_data_source_id(self) -> str:
response = self.bedrock_agent_mgmt.list_data_sources(
knowledgeBaseId=self.kb_id,
maxResults=1,
)
return response["dataSourceSummaries"][0]["dataSourceId"]
Neue Fakten sind nicht sofort abrufbar. Die Einpflege dauert eine Weile – typischerweise einige Minuten, wie aus dem Codekommentar hervorgeht – weshalb zeitkritische Arbeitslasten neben der Wissensdatenbank eine schnelle Abfragemethode (zum Beispiel in DynamoDB) benötigen. Zwei weitere Punkte sollten vor der Verwendung dieses Codes anhand der Dokumentation zur Einpflege in die Wissensdatenbank überprüft werden. Erstens funktioniert das Metadatenfiltern in Wissensdatenbanken mit den in einer separaten Metadatendatei neben jedem S3-Dokument bereitgestellten Metadatenattributen, nicht mit Feldern, die im Dokumentinhalt enthalten sind; daher passt das domain-Filter möglicherweise nicht wie beschrieben zu irgendetwas. Zweitens kann das Starten eines Einpflegevorgangs für jeden einzelnen Fakt zu Kollisionen mit einem bereits laufenden Vorgang an derselben Datengrundlage führen; das Gruppieren der Schreibvorgänge und das Synchronisieren zu festgelegten Zeiten ist in der Regel effektiver.
Tier 4: Prozedurales Gedächtnis mit Validierungsschleuse
Prozedurales Gedächtnis ist sowohl die leistungsstärkste als auch die gefährlichste Ebene. Es speichert, wie etwas ausgeführt wird, und nicht nur, was bereits getan wurde: Ein erfolgreicher Ansatz für eine wiederkehrende Aufgabentypologie wird zu einer wiederverwendbaren Liste von Schritten verarbeitet und in zukünftige Systemanfragen eingefügt.
Die Gefahr besteht darin, dass ein fehlerhafter Ablauf, sobald er gespeichert wurde, bei jeder nachfolgenden Aufgabe ohne Frage befolgt wird. Deshalb benötigt diese Ebene eine Validierungsstufe auf dem Schreibweg. Im untenstehenden Speicher wird propose_procedure nur für abgeschlossene Ausführungen gestartet und speichert neue Abläufe mit validated auf „false“, was bedeutet, dass sie noch ausstehen. Ein anderer Agent oder ein Mensch muss einen Ablauf freigeben, bevor get_procedure ihn zurückgibt, und get_procedure erfordert außerdem mindestens drei vorherige Anwendungen. record_application_result verfolgt, wie oft ein Ablauf angewendet wurde und wie oft er erfolgreich war, während format_for_injection die Schritte zusammen mit der Erfolgsrate sowie einer Erinnerung daran anzeigt, dass es sich um einen Ausgangspunkt und nicht um ein Skript handelt:
# harness/memory/procedural.py
import boto3
import json
import time
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class Procedure:
procedure_id: str
task_type: str # "contract_review", "security_audit", etc.
title: str
steps: List[str]
success_rate: float # updated as procedure is applied to new runs
times_applied: int
times_succeeded: int
created_at: float
last_updated_at: float
validated: bool # human-validated procedures get priority
class ProceduralMemoryStore:
"""
Stores and retrieves learned procedures for recurring task types.
Write path has a validation gate: procedures are stored as PENDING
and must be validated (manually or by a second agent) before use.
This prevents a single bad run from poisoning future behavior.
"""
def __init__(
self,
table_name: str = "agent-procedural-memory",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def get_procedure(self, task_type: str) -> Optional[Procedure]:
"""
Returns the best validated procedure for a task type, if one exists.
Prefers human-validated procedures, then sorts by success rate.
"""
response = self.table.query(
KeyConditionExpression="task_type = :tt",
FilterExpression="validated = :v AND times_applied >= :min",
ExpressionAttributeValues={
":tt": task_type,
":v": True,
":min": 3, # must have been used at least 3 times
}
)
items = response.get("Items", [])
if not items:
return None
# Sort by success rate, prefer human-validated
best = sorted(
items,
key=lambda x: (x.get("validated", False), x.get("success_rate", 0)),
reverse=True
)[0]
return Procedure(**best)
def propose_procedure(
self,
task_type: str,
episode_artifacts: dict,
extractor_model,
) -> Optional[str]:
"""
After a successful run, proposes a new procedure based on
what worked. Stored as PENDING until validated.
Returns procedure_id if a procedure was proposed, None otherwise.
"""
if episode_artifacts.get("outcome") != "COMPLETED":
return None
from langchain_core.messages import SystemMessage, HumanMessage
response = extractor_model.invoke([
SystemMessage(content=f"""
Based on this successful agent run, extract a reusable procedure for
task type: {task_type}
The procedure should be:
- A numbered list of clear, actionable steps
- Generalizable to similar future tasks, not specific to this one
- Based only on what actually worked, not what was attempted and failed
Return JSON:
{{
"title": "short title for this procedure",
"steps": ["step 1", "step 2", ...]
}}
If the run does not contain a clearly reusable procedure, return null.
"""),
HumanMessage(content=json.dumps(episode_artifacts))
])
try:
procedure_data = json.loads(response.content)
except (json.JSONDecodeError, TypeError):
return None
if not procedure_data or not procedure_data.get("steps"):
return None
import uuid
now = time.time()
procedure_id = str(uuid.uuid4())
self.table.put_item(Item={
"procedure_id": procedure_id,
"task_type": task_type,
"title": procedure_data["title"],
"steps": procedure_data["steps"],
"success_rate": 1.0,
"times_applied": 0,
"times_succeeded": 0,
"created_at": now,
"last_updated_at": now,
"validated": False, # starts as PENDING
"source_run_id": episode_artifacts.get("run_id", "unknown"),
})
return procedure_id
def record_application_result(
self,
procedure_id: str,
succeeded: bool,
):
"""
Updates the success rate after a procedure is applied.
If success rate drops below threshold, flag for review.
"""
self.table.update_item(
Key={"procedure_id": procedure_id},
UpdateExpression="""
SET times_applied = times_applied + :one,
times_succeeded = times_succeeded + :inc,
last_updated_at = :now,
success_rate = (times_succeeded + :inc) / (times_applied + :one)
""",
ExpressionAttributeValues={
":one": 1,
":inc": 1 if succeeded else 0,
":now": time.time(),
}
)
def format_for_injection(self, procedure: Procedure) -> str:
"""Formats a procedure for injection into agent system prompt."""
lines = [
f"[PROCEDURAL MEMORY: recommended approach for {procedure.task_type}]",
f"{procedure.title}",
f"(Success rate: {procedure.success_rate:.0%} over {procedure.times_applied} uses)",
"",
]
for i, step in enumerate(procedure.steps, 1):
lines.append(f"{i}. {step}")
lines.append(
"\nThis procedure is a starting point. Adapt it if the current task requires it."
)
return "\n".join(lines)
Betrachten Sie diesen Codeausschnitt eher als Entwurf denn als fertigen Code, da einige Details angepasst werden müssen. Die Update-Expressionen von DynamoDB unterstützen nur Addition und Subtraktion, weshalb die Division der success_rate im Anwendungscode (oder zum Zeitpunkt des Lesens) statt innerhalb der UpdateExpression berechnet werden muss. Die Abfrage verwendet task_type als Schlüssel, während Updates procedure_id nutzen; daher benötigt die Tabelle ein Schlüsselschema oder einen sekundären Index, der beide Zugriffsweisen unterstützt. Die Fließkommazeitstempel sowie success_rate erfordern dieselbe Decimal-Konvertierung wie bereits erwähnt. Schließlich kann die Kombination aus „validiert“ und „mindestens dreimal angewendet“ zu einem Problem führen: Es muss entschieden werden, wie ein frisch validierter Prozess seine ersten Anwendungen erhält – beispielsweise über einen Testmodus oder eine von einem Menschen genehmigte Ausnahme.
/p>
Bewusstes Vergessen: Gedächtnisverfall
Das ewige Aufbewahren jedes Gedächtnisinhalts führt zu zwei Problemen. Der Speicherbedarf für Hunderttausende von Ausführungsübersichten in DynamoDB und S3 ist erheblich. Noch subtiler ist die Tatsache, dass veraltete Informationen aktiv in die Irre führen können: Ein Muster, das vor sechs Monaten funktioniert hat, mag jetzt falsch sein, weil sich der Kontext geändert hat. Ein gut konzipiertes Vergessen ist eine Stärke, kein Mangel.
Der Verfallmanager wendet je nach Kategorie unterschiedliche Richtlinien an. Für Episoden, die länger als eine Mindestaufbewahrungsfrist (standardmäßig sieben Tage) alt sind, werden die Einträge mit dem Ergebnis FAILED gelöscht. Bei Verfahren wird stattdessen ein Flag gesetzt, anstatt sie zu löschen: Alles, was innerhalb von 60 Tagen nicht aktualisiert wurde oder mindestens dreimal mit einer Erfolgsrate unter 0,6 verwendet wurde, erhält ein needs_review-Flag, damit es von einem Menschen überprüft werden kann. Dieser Code ist Python, obwohl er als SQL gekennzeichnet ist:
# harness/memory/decay.py
import boto3
import time
from decimal import Decimal
from typing import List
class MemoryDecayManager:
"""
Manages the lifecycle of stored memories.
Applies different decay strategies per memory tier.
"""
def __init__(self, region: str = "us-east-1"):
self.dynamodb = boto3.resource("dynamodb", region_name=region)
self.region = region
def apply_episodic_decay(
self,
agent_id: str,
table_name: str = "agent-episodic-memory",
max_episodes: int = 500,
min_retention_days: int = 7,
):
"""
Prunes old episodic records when the store gets large.
Keeps: recent episodes, COMPLETED outcomes, high-similarity clusters.
Removes: old FAILED runs, duplicates of similar episodes.
"""
table = self.dynamodb.Table(table_name)
now = time.time()
min_age = now - (min_retention_days * 86400)
response = table.query(
IndexName="agent-time-index",
KeyConditionExpression="agent_id = :aid AND created_at < :cutoff",
ExpressionAttributeValues={
":aid": agent_id,
":cutoff": Decimal(str(min_age)),
},
ScanIndexForward=True, # oldest first
Limit=max_episodes,
)
candidates_for_removal = [
item for item in response.get("Items", [])
if item.get("outcome") == "FAILED"
]
with table.batch_writer() as batch:
for item in candidates_for_removal:
batch.delete_item(Key={
"agent_id": item["agent_id"],
"run_id": item["run_id"],
})
def flag_stale_procedures(
self,
table_name: str = "agent-procedural-memory",
stale_after_days: int = 60,
min_success_rate: float = 0.6,
) -> List[str]:
"""
Identifies procedures that should be reviewed:
- Not used in a long time (might be outdated)
- Success rate has dropped below acceptable threshold
Does NOT delete automatically. Flags for human review.
Returns list of procedure_ids that need attention.
"""
table = self.dynamodb.Table(table_name)
stale_cutoff = time.time() - (stale_after_days * 86400)
response = table.scan(
FilterExpression=(
"last_updated_at < :cutoff OR "
"(times_applied >= :min AND success_rate < :threshold)"
),
ExpressionAttributeValues={
":cutoff": Decimal(str(stale_cutoff)),
":min": 3,
":threshold": Decimal(str(min_success_rate)),
}
)
flagged_ids = []
for item in response.get("Items", []):
table.update_item(
Key={"procedure_id": item["procedure_id"]},
UpdateExpression="SET needs_review = :v",
ExpressionAttributeValues={":v": True},
)
flagged_ids.append(item["procedure_id"])
return flagged_ids
Das Verlieren einer Episode kostet etwas Kontext, während das stille Löschen eines Verfahrens einen mühsam erarbeiteten, von Menschen überprüften Ansatz vernichten könnte – daher die Asymmetrie. flag_stale_procedures verwendet einen Tabellenscan, der sich für geplante Batch-Aufgaben (wie Lambda) eignet, aber nicht für häufig genutzte Abläufe.
Zusammenführung zu Beginn und Ende der Ausführung
Sämtliche Ebenen treffen zu Beginn jeder Ausführung zusammen. Die Bootstrap-Klasse fragt den episodischen Speicher nach bis zu drei ähnlichen Episoden, die Wissensdatenbank nach bis zu fünf Domänenfakten sowie den Verfahrensspeicher nach einem überprüften Verfahren, das zum Aufgabentyp passt. Anschließend umhüllt sie alles, was gefunden wird, mit klar abgegrenzten Markierungen, damit das Modell die abgerufene Erinnerung von der Aufgabe selbst unterscheiden kann:
# harness/memory/bootstrap.py
from typing import Optional
from langchain_core.messages import SystemMessage
from harness.memory.episodic import EpisodicMemoryStore
from harness.memory.semantic import SemanticMemoryStore
from harness.memory.procedural import ProceduralMemoryStore
class MemoryBootstrap:
"""
Assembles the memory context for a new agent run.
Called once at startup, before the agent sees the task.
"""
def __init__(
self,
agent_id: str,
knowledge_base_id: str,
s3_bucket: str,
region: str = "us-east-1",
):
self.episodic = EpisodicMemoryStore(agent_id, region=region)
self.semantic = SemanticMemoryStore(knowledge_base_id, s3_bucket, region=region)
self.procedural = ProceduralMemoryStore(region=region)
def build_memory_context(
self,
task: str,
task_type: Optional[str] = None,
domain: Optional[str] = None,
) -> str:
"""
Retrieves relevant memories from all tiers and assembles
them into a single context block for system prompt injection.
"""
sections = []
# Episodic: what happened on similar tasks before
episodes = self.episodic.retrieve_relevant_episodes(task, max_results=3)
if episodes:
sections.append(self.episodic.format_for_injection(episodes))
# Semantic: what is known about this domain
if domain:
facts = self.semantic.retrieve(task, domain=domain, max_results=5)
if facts:
sections.append(
"[SEMANTIC MEMORY: domain knowledge]\n" +
"\n".join(f"- {f}" for f in facts)
)
# Procedural: how to approach this task type
if task_type:
procedure = self.procedural.get_procedure(task_type)
if procedure:
sections.append(self.procedural.format_for_injection(procedure))
if not sections:
return ""
return (
"\n\n[MEMORY CONTEXT - retrieved from prior runs and domain knowledge]\n\n" +
"\n\n".join(sections) +
"\n\n[END MEMORY CONTEXT]\n"
)
Der Schreibweg ist absichtlich vom Leseweg getrennt. Am Ende einer Ausführung extrahiert der Konsolidator Artefakte aus dem Arbeitsspeicher, speichert die Episode ab und extrahiert sowie speichert semantische Fakten erst nach Abschluss der Ausführung. Zudem wird ein Verfahren vorgeschlagen, das auf eine Validierung wartet. Es wird ein größeres Modell verwendet als bei der Kompression während der Ausführung, da dieser Schritt das zukünftige Verhalten prägt; überprüfen Sie, ob die Modell-ID in Ihrer Region weiterhin verfügbar ist. Der Codeausschnitt verwendet außerdem Optional, ohne es aus typing zu importieren – fügen Sie daher diesen Import hinzu, wenn Sie ihn erneut verwenden:
# harness/memory/consolidator.py
from harness.memory.working_memory import WorkingMemoryManager
from harness.memory.episodic import EpisodicMemoryStore
from harness.memory.semantic import SemanticMemoryStore
from harness.memory.procedural import ProceduralMemoryStore
from langchain_aws import ChatBedrock
import boto3
class MemoryConsolidator:
"""
Runs at the end of each agent run to distill and store memories.
The write path is separate from the read path by design.
"""
def __init__(
self,
agent_id: str,
knowledge_base_id: str,
s3_bucket: str,
region: str = "us-east-1",
):
bedrock = boto3.client("bedrock-runtime", region_name=region)
self.extractor = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
model_kwargs={"temperature": 0, "max_tokens": 4000},
)
self.working = WorkingMemoryManager(region=region)
self.episodic = EpisodicMemoryStore(agent_id, region=region)
self.semantic = SemanticMemoryStore(knowledge_base_id, s3_bucket, region=region)
self.procedural = ProceduralMemoryStore(region=region)
def consolidate(
self,
final_state: dict,
task_type: Optional[str] = None,
domain: Optional[str] = None,
):
"""
End-of-run consolidation. Extracts and stores memories from all tiers.
"""
messages = final_state.get("messages", [])
# Step 1: Extract run artifacts from working memory
artifacts = self.working.extract_run_artifacts(messages)
# Step 2: Store as episodic memory
run_id = self.episodic.store_episode(artifacts)
artifacts["run_id"] = run_id
# Step 3: Extract and store semantic facts (only from successful runs)
if domain and artifacts.get("outcome") == "COMPLETED":
facts = self.semantic.extract_facts_from_episode(
artifacts, self.extractor, domain
)
for fact in facts:
self.semantic.store_fact(fact)
# Step 4: Propose procedural memory update
if task_type:
self.procedural.propose_procedure(task_type, artifacts, self.extractor)
Produktionsfestigung
Schutz vor Speichervergiftung
Ein schlechter Ablauf kann eine falsche Tatsache in das semantische Gedächtnis eintragen oder einen fehlerhaften Ablauf vorschlagen, und in großem Maßstab wird der Fehler verstärkt, denn jeder zukünftige Ablauf, der die Tatsache abruft oder dem Ablauf folgt, erbt ihn. Die Schutzmaßnahmen sind schichtweise angeordnet:
- Schreibe nur aus
COMPLETED-Abläufen, deren Ergebnisse eine hohe Zuverlässigkeit aufweisen, ein. - Fordere vor der Aufwertung einer Tatsache wiederholte, unabhängige Beobachtungen: Ein einzelner Ablauf deutet nur auf etwas hin, während Übereinstimmung in drei Abläufen echter Beweis ist.
- Führe eine menschliche Überprüfung vor dem prozeduralen Gedächtnis durch, bevor ein Ablauf den Status der Validierung erreicht.
Eine einfache Methode, die zweite Schicht umzusetzen, ist ein Bestätigungs-Zähler, der durch einen Hash des Inhalts der Tatsache identifiziert wird. Jede Beobachtung erhöht den Zähler atomar mit if_not_exists, und die Tatsache wird erst dann aufgewertet, wenn die Anzahl den Schwellenwert erreicht hat:
# Simple confirmation counter before promoting a fact
def promote_fact_if_confirmed(
fact_content: str,
confirmation_threshold: int = 3,
table_name: str = "fact-confirmation-log",
region: str = "us-east-1",
):
"""
Tracks how many independent runs have observed the same fact.
Only writes to semantic memory after threshold is reached.
"""
table = boto3.resource("dynamodb", region_name=region).Table(table_name)
import hashlib
fact_hash = hashlib.sha256(fact_content.encode()).hexdigest()[:16]
response = table.update_item(
Key={"fact_hash": fact_hash},
UpdateExpression="SET confirmation_count = if_not_exists(confirmation_count, :z) + :inc, content = :c",
ExpressionAttributeValues={":z": 0, ":inc": 1, ":c": fact_content},
ReturnValues="UPDATED_NEW",
)
return int(response["Attributes"]["confirmation_count"]) >= confirmation_threshold
Weil der genaue Text gehässt wird, werden unterschiedlich formulierte Versionen derselben Tatsache getrennt gezählt; normalisieren oder gruppieren Sie die Fakten zunächst, falls die Promotions zu selten auftreten.
PII aus dem Speicher halten
Der Agent-Speicher sammelt personenbezogene Informationen, es sei denn, Sie verhindern dies aktiv. Ein Vertragsanalyse-Agent speichert Fakten über Verträge, und in diesen Verträgen werden Personen, Gehälter sowie andere persönliche Details erwähnt. Der sauberste Ansatz ist es, den Text vor dem Schreiben in die episodischen oder semantischen Ebenen zu bereinigen. Amazon Comprehend bietet eine PII-Erkennungs-API, die gut mit englischen Texten funktioniert. Die untenstehende Funktion ersetzt jede erkannte Entität durch deren Typetikett und arbeitet dabei von Ende des Strings ausgehend, damit frühere Positionen weiterhin gültig bleiben:
import boto3
def scrub_pii_before_storage(text: str, region: str = "us-east-1") -> str:
comprehend = boto3.client("comprehend", region_name=region)
response = comprehend.detect_pii_entities(Text=text[:5000], LanguageCode="en")
scrubbed = text
# Replace detected PII with type labels, working backwards to preserve offsets
for entity in sorted(response["Entities"], key=lambda x: x["BeginOffset"], reverse=True):
scrubbed = (
scrubbed[:entity["BeginOffset"]] +
f"[{entity['Type']}]" +
scrubbed[entity["EndOffset"]:]
)
return scrubbed
Achten Sie darauf, dass die Aufrufung nur die ersten 5.000 Zeichen prüft, wodurch alles darüber ungesäubert gespeichert wird. Längere Texte müssen in Blöcke aufgeteilt und jeder Block separat gesäubert werden. Unterstützte Entity-Typen sowie Grenzwerte sind in der Comprehend PII-Dokumentation aufgeführt.
Isolierung der Mieter
Falls Agenten mehrere Kunden auf gemeinsamer Infrastruktur bedienen, müssen ihre Speicherinhalte vollständig isoliert werden: Ein Abfragen für einen Kunden darf niemals die Speicherinhalte eines anderen Kunden abrufen. Das einfachste Vorgehen besteht darin, jeden Speicher Schlüssel mit einem Mieter-Identifikator zu versehen und die Grenzen in der Speicherschicht durchzusetzen – beispielsweise mit DynamoDB-Bedingungsexpressionen – anstatt sich allein auf Filterung auf Anwendungsseite zu verlassen:
# Always include tenant_id in all memory operations
def store_episode_isolated(tenant_id: str, agent_id: str, artifacts: dict):
namespaced_agent_id = f"{tenant_id}#{agent_id}"
# All subsequent operations use namespaced_agent_id, never bare agent_id
Da agent_id die episodische Teilungsschlüssel ist, sorgt das Präfix dafür, dass normale Abfragen innerhalb eines Mieters bleiben. Wenden Sie denselben Namensraum auf die Knowledge Base, die prozedurale Tabelle sowie das Bestätigungsprotokoll an.
Was zu erwarten ist, sobald das Gedächtnis aktiv ist
Gedächtnissysteme verbessern die Leistung der Agenten im Laufe der Zeit deutlich, führen aber auch zu Fehlern, die schwieriger zu debuggen sind als alles, was ein stateless System hervorruft:
- Schlechte Erinnerungen sind schlimmer als gar keine. Ein verwirrter stateless Agent verdirbt eine gesamte Ausführung. Ein Agent mit fehlerhaften Informationen verdirbt jede Ausführung, die diese abruft – möglicherweise monatelang. Investieren Sie in Validierungsschritte, bevor Sie die Abdeckung des Gedächtnisses erweitern.
Referenzarchitektur
Der vollständige Lebenszyklus einer Ausführung, von der Initialisierung über die Konsolidierung in die drei persistenten Speicher, sieht wie folgt aus:
New Run Starts
|
v
+-------------------------------+
| Memory Bootstrap |
| 1. Query Episodic Store |
| 2. Query Bedrock KB |
| 3. Query Procedural Store |
| 4. Assemble context block |
+-------------------------------+
|
v
+-------------------------------+
| Agent Run (LangGraph) |
| Working Memory Manager |
| Compression at threshold |
| Tool calls with auth/retry |
+-------------------------------+
|
v
+-------------------------------+
| Memory Consolidator |
| 1. Extract run artifacts |
| 2. Scrub PII |
| 3. Write episodic record |
| 4. Propose semantic facts |
| (if confirmed: store) |
| 5. Propose procedure |
| (pending validation) |
+-------------------------------+
|
v
+------------------+ +------------------+ +------------------+
| DynamoDB | | Bedrock KB | | DynamoDB |
| Episodic Memory | | Semantic Memory | | Procedural Mem. |
| TTL: 90 days | | S3 backed | | Validated flag |
+------------------+ +------------------+ +------------------+
Und das ist der gesamte Infrastruktur-Stack, einschließlich der lokalen Entwicklungsalternative basierend auf Ollama, ChromaDB und Docker Compose, die es ermöglicht, den Speicherstack ohne AWS-Kosten auszuprobieren:
+-----------------------------+---------------------+------------------------------+
| Component | Technology | Role |
+-----------------------------+---------------------+------------------------------+
| Working Memory | LangGraph State + | Active context compression, |
| | MemorySaver | sliding window management |
+-----------------------------+---------------------+------------------------------+
| Episodic Memory Store | DynamoDB + TTL | Past run summaries, semantic |
| | | similarity retrieval |
+-----------------------------+---------------------+------------------------------+
| Episodic Embeddings | Titan Embed v2 | Semantic similarity search |
| | (Bedrock) | over episode store |
+-----------------------------+---------------------+------------------------------+
| Semantic Memory | Bedrock Knowledge | Domain facts, structured |
| | Bases + S3 | knowledge, write via ingest |
+-----------------------------+---------------------+------------------------------+
| Procedural Memory | DynamoDB | Learned task procedures, |
| | | validated flag, success rate |
+-----------------------------+---------------------+------------------------------+
| PII Scrubbing | AWS Comprehend | Before any memory write, |
| | | English PII detection |
+-----------------------------+---------------------+------------------------------+
| Fact Confirmation | DynamoDB | Multi-run confirmation gate |
| | | before semantic write |
+-----------------------------+---------------------+------------------------------+
| Memory Decay | DynamoDB TTL + | Episode pruning, stale |
| | scheduled Lambda | procedure flagging |
+-----------------------------+---------------------+------------------------------+
| Context Compression | Claude Haiku 4.5 | Working memory summarization |
+-----------------------------+---------------------+------------------------------+
| Artifact Extraction | Claude 3.5 Sonnet | End-of-run distillation |
+-----------------------------+---------------------+------------------------------+
| Local Dev Alternative | Ollama + ChromaDB | Full memory stack testable |
| | Docker Compose | without AWS costs |
+-----------------------------+---------------------+------------------------------+
| Infrastructure as Code | Terraform | DynamoDB tables, S3 bucket, |
| | | KB data source, IAM roles |
+-----------------------------+---------------------+------------------------------+
Wichtige Erkenntnisse
- Betrachten Sie das Kontextfenster als Whiteboard und erstellen Sie ein separates, persistentes Notizbuch für Wissen, das eine Ausführung überdauern muss.
- Trennen Sie das Gedächtnis in Arbeits-, episodische, semantische und prozedurale Ebenen auf, da jede andere Speicher-, Abruf- und Fehlerbehandlung erfordert.
- Fassen Sie Ausführungen in kompakte Zusammenfassungen und Fakten zusammen; speichern oder suchen Sie niemals Rohdaten als Gedächtnisinhalte.
- Setzen Sie die strengsten Kontrollmechanismen auf den Ebenen ein, die das Verhalten am stärksten verändern: Bestätigungen für Fakten, menschliche Überprüfung für Prozeduren.
- Entfernen Sie PII und gewährleisten Sie eine Isolierung der Nutzer in der Speicherschicht, bevor irgendein Gedächtnisinhalt geschrieben wird.
- Bereiten Sie von Anfang an Ressourcen für Vergessen, Überwachung des Abrufs, Verzögerungen bei der Eingabe sowie eine spätere Wiedereinbettung vor.