Projektowanie czteropoziomowej pamięci agenta przy użyciu LangGraph i Amazon Bedrock
Naucz się tworzyć agentów LLM z pamięcią epizodyczną, semantyczną i proceduralną w środowiskach Bedrock i LangGraph, a także chronić je przed zatruwaniem, wyciekami danych osobowych i utratą informacji.
Źródła: Agentic Architectures — Article 7: Agent Memory Architectures
Ranо czy późno agent produkcyjny zaczyna wykonywać tę samą pracę dwa razy. Nic nie jest uszkodzone, a polecenie jest w porządku; agent po prostu nie pamięta, że w zeszłym tygodniu rozwiązywał niemal identyczny problem, więc postępuje według tych samych kroków, powtarza te same błędy pośrednie, dochodzi do tego samego wyniku, a ty ponownie płacisz za każdy token. Ten przewodnik pokazuje, jak to naprawić za pomocą rzeczywistej infrastruktury pamięci, a nie większego okna kontekstowego: cztery odrębne poziomy pamięci stworzone za pomocą LangGraph na Amazon Bedrock, wraz z mechanizmami bezpieczeństwa produkcyjnymi, które zapobiegają temu, by pamięć stała się obciążeniem.
Załóżmy agenta do analizy umów wyposażonego w solidne narzędzia oraz działającą architekturę wielu agentów, który mimo to poświęca pierwsze kilka wywołań narzędzi na każdą podobną umowę na ponowne odkrywanie wzorców strukturalnych, które już wcześniej ustalił. Brakuje mu pamięci.
Większe okno kontekstowe to nie pamięć
Gdy kontekst się wypełnia, typową reakcją jest zwiększenie rozmiaru okna lub bardziej agresywne streszczanie. Oba rozwiązania są przydatne w trakcie jednej sesji; żadne z nich nie stanowi pamięci. Przydatna analogia: okno kontekstu to tablica pisana. Jest użyteczne podczas spotkania i zostaje wycierane po jego zakończeniu. Pamięć to notatnik, który nosisz od jednego spotkania do drugiego.
Zarządzanie kontekstem ma trzy konkretne ograniczenia:
- Nie łączy sesji. Gdy agent kończy pracę i jego proces się zamyka, kontekst znika. Wszystko, czego się nauczył, przepada, chyba że wyraźnie to zapisazesz.
- Nie przekracza granic agentów. W systemie wieloagentowym każdy pracownik ma swój własny kontekst. To, co odkrywa pracownik zajmujący się audytem bezpieczeństwa, jest niewidoczne dla pracownika analizującego kod, chyba że celowo to tam skierujesz.
Pamięć służy innemu celowi: stanowi trwałe, dostępne do wyszukiwania repozytorium wiedzy, które przetrwa poszczególne wykonywania zadań i może być udostępniane w dowolnej topologii agentów.
Cztery poziomy pamięci agenta
Każdy poziom wiąże się z inną infrastrukturą, sposobami wyszukiwania informacji oraz mechanizmami awarii:
+---------------------+--------------------------------------------------+----------------------------+
| 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 | |
+---------------------+--------------------------------------------------+----------------------------+
Poziomy te oddziałują ze sobą w cyklu. Pamięć robocza przechowuje aktualne wykonywanie zadań. Gdy wykonywanie się kończy, krok destylacji zapisuje dane do trzech trwałych poziomów pamięci. Gdy rozpoczyna się kolejne wykonywanie, krok wyszukiwania przywołuje istotne informacje z tych poziomów:
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
Pamięć epizodyczna informuje o tym, co wydarzyło się podczas kolejnego uruchomienia, pamięć semantyczna o tym, co jest znane na dany temat, a pamięć proceduralna o tym, jak podejść do zadania. Wszystkie trzy są pobierane przy uruchomieniu i umieszczane w pamięci roboczej, zanim agent wykonuje pierwszy krok rozumowania.
Poziom 1: zarządzanie pamięcią roboczą podczas jednego uruchomienia
Pamięć robocza to wszystko, co istnieje podczas jednego uruchomienia, i zarządzasz nią niezależnie od tego, czy ją tak nazywasz, czy nie. W LangGraph stan grafu stanowi pamięć roboczą – tam znajdują się każde wiadomość, wynik narzędzia oraz rezultat pośredni.
Problem polega na tym, że ta tendencja stale się nasila. Kompresja kontekstu pomiędzy etapami procesu, często stosowana w technikach wieloagentowych, jest w rzeczywistości zarządzaniem pamięcią roboczą pod inną nazwą. Poniższa klasa przekształca tę ideę w komponent wielokrotnie używalny. Funkcja maybe_compress jest wykonywana przed każdym krokiem agenta; gdy lista wiadomości przekroczy określony próg (domyślnie 30), zachowuje wiadomości systemowe oraz 15 najnowszych wiadomości niebędących wiadomościami systemowymi w niezmienionej formie, a wszystko starsze zastępuje jednym streszczeniem wygenerowanym przez mały, tani model. Druga metoda, extract_run_artifacts, jest wykonywana na końcu procesu i prosi model o ustrukturyzowany zapis w formacie JSON zawierający informacje o zadaniu, wyniku, ustaleniach, podejściu, narzędziach oraz wzorcach wielokrotnie używalnych – ten zapis staje się danymi wejściowymi dla pamięci epizodycznej. Ten fragment jest napisany w języku Python, a nie 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": []}
Zapytanie dotyczące streszczania odrzuca nieudane próby oraz błędy narzędzi, a parsowanie JSON przyjmuje wartość UNKNOWN wraz z pustymi listami, dzięki czemu błędna odpowiedź nigdy nie powoduje awarii całego procesu. Identyfikatory modeli, takie jak pokazany tutaj ID Haiku, zmieniają się z czasem i mogą wymagać profilu inferencji specyficznego dla danego regionu, dlatego przed ich skopiowaniem sprawdź aktualną listę modeli Bedrock.
Dla rozwoju lokalnego możesz zastąpić generator streszczeń Bedrock modelem dostarczanym przez Ollamę:
# Local dev alternative
# ollama pull llama3.1:8b
from langchain_ollama import ChatOllama
summarizer = ChatOllama(model="llama3.1:8b", temperature=0, format="json")
Poziom 2: pamięć epizodyczna wcześniejszych wykonywań
Pamięć epizodyczna odpowiada na pytanie „co się wydarzyło wcześniej?”. Na początku wykonywania zadania agent pobiera streszczenia wcześniejszych wykonań, które przypominają bieżące zadanie, i wykorzystuje je, aby uniknąć powtarzania pracy oraz znanych błędów.
System przechowuje streszczenia w formie skondensowanej, a nie surowe logi wykonywań. Pełne logi są kosztowne w utrzymywaniu i wolne w wyszukiwaniu; zwięzłe streszczenia są tanie i szybkie. Przechowywanie odbywa się w DynamoDB, a wyszukiwanie wykorzystuje embeddingi Bedrock do określania podobieństwa semantycznego.
Tabela zdefiniowana w Terraform używa agent_id jako klucza podziału oraz run_id jako klucza sortowania. Globalny indeks wtórny dla agent_id i created_at umożliwia zapytania typu „najnowsze epizody dla danego agenta”, a funkcja TTL w DynamoDB dla expires_at automatycznie usuwa stare rekordy:
# 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"
}
}
Sam sklep włącza podsumowanie zadania wraz z kluczowymi ustaleniami, zapisuje rekord o terminie ważności 90 dni i przy pobieraniu porównuje zawartość aktualnego zadania z najnowszymi epizodami przy użyciu podobieństwa kosinowego. Zachowywane są tylko te epizody, których podobieństwo przekracza próg 0,75; w przypadku remisów decyduje się na wynik COMPLETED; natomiast funkcja format_for_injection przedstawia wybrane elementy jako oznaczone bloki do promptu systemowego:
# 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))
Zapytanie pobiera maksymalnie 100 najnowszych odcinków i oblicza podobieństwo w kodzie aplikacji; jak zaznaczono w komentarzu, większe implementacje powinny korzystać z paginacji oraz bazy danych wektorowych. Należy również zauważyć, że zasób boto3 DynamoDB odrzuca wartości typu float w Pythonie, dlatego lista embeddingów musi zostać przekształcona na typ Decimal (lub zserializowana, na przykład jako JSON lub dane binarne), zanim funkcja put_item będzie w stanie ją przyjąć. Model embeddingów użyty tutaj jest opisany w dokumentacji Amazon Titan Embeddings.
Poziom 3: pamięć semantyczna z bazami wiedzy Bedrock
Pamięć epizodyczna przechowuje wydarzenia; pamięć semantyczna przechowuje wiedzę. Różnica jest subtelna, ale istotna. Epizod może brzmieć tak: „Umowa z dostawcą przeanalizowana w zeszły wtorek zawierała niestandardowy punkt 7.3”. Odpowiadający mu fakt semantyczny brzmi: „Punkt 7.3 w umowach z dostawcami zazwyczaj dotyczy limitów odpowiedzialności; normalny zakres to dwukrotność wartości umowy, a każda wartość poniżej tej kwoty jest sygnałem ostrzegawczym”.
Fakty semantyczne trwają dłużej niż epizody. Nie wygasają po 90 dniach; są udoskonalane z upływem czasu i przydają się przy wielu typach zadań, a nie tylko przy tych, które przypominają poprzednie realizacje.
Bedrock Knowledge Bases stanowi naturalne środowisko dla tego poziomu, ponieważ obsługuje dzielenie na fragmenty, embedding, przechowywanie i odzyskiwanie danych. Większość zespołów nie stworzyła jednak mechanizmu zapisu, ponieważ bazy wiedzy są zazwyczaj traktowane jako tylko do odczytu. Poniższa struktura obsługuje obie strony procesu:
retrieveprzeprowadza semantyczne wyszukiwanie wektorowe, opcjonalnie filtrowane za pomocą klucza metadanychdomain, i odrzuca wyniki o ocenie poniżej 0,6.store_factzapisuje fakt jako dokument JSON do S3 i uruchamia zadanie pobierania, aby Baza Wiedzy mogła go przetworzyć.extract_facts_from_episodeprosi model o wydobycie faktów uniwersalnych z zakończonego procesu, pomijając szczegóły specyficzne dla zadania, które należą do pamięci epizodycznej, i zachowując tylko te fakty o poziomie pewności co najmniej 0,7.
# 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"]
Nowe fakty nie mogą być natychmiast odzyskane. Proces przyjmowania danych zajmuje trochę czasu, zazwyczaj kilka minut, jak wskazuje komentarz w kodzie, dlatego obciążenia wymagające szybkiego dostępu do informacji (na przykład w DynamoDB) muszą istnieć obok Bazy Wiedzy. Dwa dodatkowe punkty wymagają weryfikacji na podstawie dokumentacji dotyczącej przyjmowania danych do Bazy Wiedzy przed użyciem tego kodu. Po pierwsze, filtrowanie metadanych w Bazach Wiedzy odbywa się na podstawie atrybutów metadanych umieszczonych w oddzielnym pliku metadanych obok każdego dokumentu S3, a nie na podstawie pól zawartych w treści dokumentu, więc filtr domain może nie pasować do żadnych danych w takiej formie. Po drugie, rozpoczynanie zadania przyjmowania danych dla każdego pojedynczego faktu może kolidować z zadaniem już uruchomionym na tej samej bazie danych; grupowanie zapisów i synchronizacja według harmonogramu jest zazwyczaj bardziej efektywne.
Poziom 4: pamięć proceduralna z bramką walidacji
Pamięć proceduralna jest zarówno najpotężniejszym, jak i najniebezpieczniejszym poziomem. Przechowuje informacje o tym, jak coś zrobić, a nie tylko to, co zostało zrobione: skuteczna metoda realizacji powtarzającego się typu zadania jest sprowadzana do listy kroków nadających się do ponownego użycia i włączana do przyszłych zapytań systemu.
Niebezpieczeństwo polega na tym, że błędna procedura, po jej zapisaniu, jest bezkrytycznie stosowana przy każdym kolejnym zadaniu. Dlatego ten poziom wymaga bramy walidacji na ścieżce zapisu. W poniższym magazynie funkcja propose_procedure jest uruchamiana tylko w przypadku zakończonych procesów i zapisuje nowe procedury z ustawieniem validated na false, co oznacza stan w oczekiwaniu. Inny agent lub człowiek musi zatwierdzić procedurę, zanim get_procedure ją zwróci, a funkcja ta wymaga dodatkowo co najmniej trzech wcześniejszych zastosowań. Funkcja record_application_result rejestruje, jak często procedura była stosowana i ile razy odniosła sukces, natomiast format_for_injection przedstawia kroki wraz z wskaźnikiem skuteczności oraz przypomnieniem, że stanowią one jedynie punkt wyjścia, a nie skrypt:
# 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)
Traktuj ten fragment jako szkic projektu, a nie gotowy kod do bezpośredniego użycia, ponieważ kilka szczegółów wymaga dostosowania. Wyrażenia aktualizacyjne w DynamoDB obsługują jedynie dodawanie i odejmowanie, więc dzielenie wartości success_rate musi być obliczane w kodzie aplikacji (lub podczas odczytu), a nie wewnątrz UpdateExpression. Zapytanie używa task_type jako klucza, podczas gdy aktualizacje wykorzystują procedure_id, więc tabela wymaga schematu klucza lub indeksu wtórnego, który obsłuży oba te wzorce dostępu. Zmienne typu float reprezentujące daty oraz success_rate wymagają tej samej konwersji na typ Decimal, o której mowa wcześniej. Wreszcie połączenie kryterium „zweryfikowanego” z wymogiem „zastosowania co najmniej trzy razy” może prowadzić do sytuacji typu „kura czy jajko”: trzeba zdecydować, w jaki sposób świeżo zweryfikowana procedura może uzyskać swoje pierwsze zastosowania, na przykład poprzez tryb testowy lub wyjątek zatwierdzony przez człowieka.
Celowane zapominanie: degradacja pamięci
Zachowywanie każdej informacji na zawsze powoduje dwa problemy. Przechowywanie setek tysięcy podsumowań wykonań w DynamoDB i S3 wiąże się z rzeczywistymi kosztami. Co bardziej subtelne, przestarzałe informacje mogą błędnie wprowadzać w błąd – schemat, który działał sześć miesięcy temu, może teraz być nieprawidłowy ze względu na zmiany w obszarze, którym się zajmujemy. Dobrze zaprojektowane zapominanie jest zaletą, a nie wadą.
Menedżer degradacji stosuje różne zasady w zależności od poziomu. W przypadku epizodów starszych niż minimalny okres przechowywania (siedem dni jako standard) usuwa rekordy, których wynik to FAILED. W przypadku procedur zamiast usuwać, oznacza je jako wymagające sprawdzenia: wszystko, co nie zostało zaktualizowane w ciągu 60 dni lub co zostało użyte co najmniej trzy razy przy wskaźniku skuteczności poniżej 0,6, otrzymuje oznaczenie needs_review, aby mogło zostać przeanalizowane przez człowieka. Ten kod jest napisany w Pythonie, mimo że jest oznaczony jako SQL:
# 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
Utrata odcinka kosztuje tylko niewielką ilość kontekstu, natomiast ciche usunięcie procedury może doprowadzić do utraty trudno zdobytego, zweryfikowanego przez ludzi podejścia, stąd ta asymetria. flag_stale_procedures wykorzystuje przeszukiwanie tabeli, co nadaje się do zadań batchowych uruchamianych według harmonogramu (np. Lambda), ale nie do intensywnie używanych ścieżek.
Łączenie wszystkiego na początku i końcu wykonywania
Wszystkie warstwy spotykają się na początku każdego wykonywania. Klasa bootstrap przeszukuje magazyn odcinków w poszukiwaniu maksymalnie trzech podobnych odcinków, bazę wiedzy w poszukiwaniu maksymalnie pięciu faktów z danej dziedziny oraz magazyn procedur w poszukiwaniu zweryfikowanej procedury pasującej do typu zadania. Następnie otacza to, co znajdzie, wyraźnie zdefiniowanymi markerami, aby model mógł odróżnić pobraaną pamięć od samego zadania:
# 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"
)
Ścieżka zapisu jest celowo oddzielona od ścieżki odczytu. Pod koniec sesji program konsolidujący wyodrębnia dane z pamięci roboczej, przechowuje epizod, a fakty semantyczne wydobywa i zapisuje dopiero po zakończeniu sesji, proponując jednocześnie procedurę czekającą na weryfikację. Wykorzystuje on większy model niż kompresja w trakcie sesji, ponieważ ten krok wpływa na przyszłe zachowanie; sprawdź, czy identyfikator modelu jest nadal dostępny w twoim regionie. Fragment ten używa również typu Optional bez jego importowania z modułu typing, więc dodaj ten import, jeśli zamierzasz go ponownie użyć:
# 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)
Wzmocnienie systemu w produkcji
Ochrona przed zatruwaniem pamięci
Jedna nieudana próba może zapisać błędny fakt do pamięci semantycznej lub zasugerować nieprawidłową procedurę, a w skali masowej błąd się powiększa, ponieważ każda kolejna próba, która pobiera ten fakt lub stosuje tę procedurę, odziedzicza go. Mechanizmy obronne są warstwowe:
- Zapisuj tylko wyniki z prób o stanie
COMPLETED, których dane charakteryzują się wysokim poziomem pewności. - Wymagaj powtarzanych, niezależnych obserwacji przed uznaniem faktu za prawdziwy: pojedyncza próba jedynie sugeruje coś, natomiast zgodność w trzech próbach stanowi rzeczywisty dowód.
- Zanim procedura osiągnie status zweryfikowany, umożliw ludzką weryfikację przed jej zastosowaniem.
Prostym sposobem na wdrożenie drugiej warstwy jest użycie licznika potwierdzeń opartego na haszu treści faktu. Każda obserwacja atomowo zwiększa licznik za pomocą funkcji if_not_exists, a fakt zostaje uznany za prawdziwy dopiero wtedy, gdy liczba osiągnie określony próg:
# 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
Ponieważ tekst jest haszowany, różne sformułowania tego samego faktu są liczone oddzielnie; jeśli promocje okazują się zbyt rzadkie, należy najpierw znormalizować lub skupić fakty.
Zachowywanie danych PII poza pamięcią
Pamięć agenta będzie gromadzić informacje identyfikujące osobę, chyba że aktywnie temu zapobiegniesz. Agent do analizy umów przechowuje dane o umowach, a te z kolei zawierają informacje o osobach, pensjach i innych szczegółach osobistych. Najlepszym rozwiązaniem jest wyczyszczenie tekstu przed zapisaniem go do poziomów epizodycznych lub semantycznych. Amazon Comprehend oferuje API do wykrywania danych PII, które dobrze funkcjonuje na tekście w języku angielskim. Poniższa funkcja zastępuje każdą wykrytą entytet jej etykietą typu, iterując od końca ciągu znaków, dzięki czemu wcześniejsze pozycje pozostają ważne:
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
Należy pamiętać, że funkcja ta sprawdza tylko pierwsze 5 000 znaków, więc wszystko, co znajduje się poza tym zakresem, jest przechowywane bez żadnej obróbki. Dłuższe teksty muszą być dzielone na fragmenty, a każdy z nich należy przetworzyć osobno. Typy i limity obsługiwanych entytetów są wymienione w dokumentacji Comprehend PII.
Izolacja użytkowników
Jeśli agenci obsługują kilku klientów na wspólnej infrastrukturze, ich dane muszą być w pełni izolowane: proces przetwarzania danych dla jednego klienta nie może nigdy uzyskać dostępu do danych innego klienta. Najprostszym rozwiązaniem jest nadanie każdemu kluczu danych identyfikatora użytkownika oraz wprowadzenie ograniczeń na poziomie warstwy przechowywania, np. za pomocą wyrażeń warunkowych w DynamoDB, zamiast polegać wyłącznie na filtrowaniu na poziomie aplikacji:
# 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
Ponieważ agent_id jest kluczem podziału epizodycznego, przedrostek zapewnia, że zwykłe zapytania pozostają w obrębie jednego użytkownika. Ten sam schemat nazewniczy powinien być zastosowany w Bazie Wiedzy, tabeli proceduralnej oraz dzienniku potwierdzeń.
Czego oczekiwać po uruchomieniu pamięci
Pamięć sprawia, że agenci z czasem stają się znacznie lepsi, ale jednocześnie wprowadza modele awarii trudniejsze do debugowania niż te, które powstają w systemach bezstanowych:
- Złe informacje są gorsze niż żadne. Zdezorientowany agent bezstanowy psuje jeden przebieg działania, natomiast agent z fałszywymi danymi psuje każdy przebieg, w którym te dane są wykorzystywane, być może przez całe miesiące. Inwestuj w mechanizmy walidacji przed rozszerzaniem zakresu wykorzystania pamięci.
Architektura referencyjna
Pełny cykl życia procesu, od startu po konsolidację w trzech trwałych zasobach, wygląda następująco:
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 |
+------------------+ +------------------+ +------------------+
A oto pełna lista komponentów infrastruktury, w tym alternatywa do rozwoju lokalnego oparta na Ollama, ChromaDB i Docker Compose, która umożliwia korzystanie z tej infrastruktury bez kosztów związanych z AWS:
+-----------------------------+---------------------+------------------------------+
| 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 |
+-----------------------------+---------------------+------------------------------+
Główne wnioski
- Traktuj okno kontekstowe jak tablicę pisarską i twórz oddzielny, trwały notatnik dla wiedzy, która musi przetrwać cały proces.
- Rozdziel pamięć na poziomy roboczy, epizodyczny, semantyczny i proceduralny, ponieważ każdy z nich wymaga innego sposobu przechowywania, wyszukiwania oraz radzenia sobie z awariami.
- Przekształcaj wykonywane operacje w zwięzłe streszczenia i fakty; nigdy nie przechowuj ani nie wyszukuj surowych zapisów jako części pamięci.
- Zastosuj najstricterze zasady na tych poziomach, które najbardziej wpływają na zachowanie: potwierdzenia są istotne dla faktów, a weryfikacja przez ludzi – dla procedur.
- Zanim zostanie napisana choćby pierwsza informacja do pamięci, usuń dane osobowe i zapewnij izolację poszczególnych użytkowników na warstwie przechowywania.
- Od samego początku przewidź budżet na zapominanie, monitorowanie dostępu do informacji, opóźnienia w ich przetwarzaniu oraz ewentualne ponowne włączenie do systemu.
Literatura pokrewna
- Wewnątrz agenta AI: modele, narzędzia, pamięć i uświadomione myślenie wyjaśnione — Opisuje podstawową architekturę agentów AI – cele, modele, narzędzia, pamięć i procesy myślowe – oraz przedstawia praktyczny przykład agenta badawczego opartego na Pythonie.
- Projektowanie wydolnych grafów agentów AI: próby ponowne, plany zastępcze i GraphRAG — Dowiedz się, jak budować odporną na błędy architekturę agentów AI przy użyciu specjalnych ścieżek obsługi awarii, logiki prób ponownych, LangGraph oraz sytuacji, w których GraphRAG przewyższa zwykłe rozwiązania RAG lub pętle agentów.