Czterowarstwowy projekt pamięci dla agentów AI z informacją o pochodzeniu
Rozdzielenie pamięci krótkoterminowej i długoterminowej, cztery warstwy (bufor, epizodyczna, semantyczna, proceduralna), opcje przechowywania, zapisy asynchroniczne oraz mechanizmy ochronne przed zatruwaniem pamięci.
Załóżmy agenta, który zawodzi na czternastym kroku długiego procesu pracy. To nie wina modelu – agent po prostu już nie pamięta, co wcześniej zrobił. Ten problem występuje według znajomego wzorca: wywołania narzędzi są poprawne, a instrukcje jasne, ale gdy rozmowa przekracza pewną niewidzialną granicę długości, agent zaczyna się pętlić. Powtarza pytania, na które już odpowiedziano, pobiera dane, które i tak posiada, oraz sprzecza się sam ze sobą pomiędzy szóstym a dwunastym krokiem, jakby pierwszych pięciu kroków w ogóle nie było.
Zastąpienie modelu innym rzadko coś w tym zmienia. Wąskim gardłem jest infrastruktura otaczająca model, a dokładniej brak prawdziwej architektury pamięci. Wklejenie ostatnich N wiadomości do instrukcji działa w przypadku chatbotów, ale przestaje być skuteczne, gdy agent musi wykonać dwudziestoetapowy proces i pamiętać z siedemnastego kroku to, co zdecydował na trzecim.
Wzorem rozwiązującym ten problem jest czterowarstwowa pila pamięci, w której za każdą warstwą znajduje się odpowiednia przestrzeń przechowywania. W dalszej części omówiono, jakie funkcje pełni każda warstwa, w jaki sposób zespoły w GitHub, Anthropic i Berkeley opisują swoje wersje oraz jakie są przykłady kodu w Pythonie dla każdej z tych warstw.
Okno kontekstu to nie pamięć
Okno kontekstu stanowi pamięć roboczą: RAM, a nie dysk twardy. Każdy token w nim ma swoją cenę. Wypełnianie całej historii rozmowy do okna o pojemności 128K wydaje się bezpieczne, dopóki nie uruchomimy agenta wymagającego intensywnego wykorzystania narzędzi. Trzydziestopunktowy proces pracy z wywołaniami narzędzi, obserwacjami i śladami rozumowania szybko zużywa 100K tokenów, a gdy okno się wypełni, najstarszy content jest usuwany z początku – tam zazwyczaj znajdują się pierwotne ograniczenia i opis zadania użytkownika.
Bardziej subtelny problem jest jeszcze gorszy. Badanie przeprowadzone przez badaczy ze Stanfordu i UC Berkeley, opublikowane w 2024 roku w czasopiśmie ACL pod tytułem „Lost in the Middle: How Language Models Use Long Contexts”, wykazało, że wydajność modeli językowych ma kształt krzywej w formie litery U w zależności od zakresu kontekstu. Modele radzą sobie dobrze z informacjami znajdującymi się na początku i końcu kontekstu, natomiast wyraźnie gorzej z informacjami ukrytymi pośrodku. Efekt ten zaobserwowano w różnych architekturach, rozmiarach modeli i długościach kontekstu, przy czym testowano okna o długości od 4K do 32K tokenów. Badanie kontynuacyjne z 2025 roku przeprowadzone przez MIT dostarczyło matematycznego wyjaśnienia, argumentując, że zarówno maskowanie uwagi przyczynowej, jak i kodowanie pozycji obrotowej zmniejszają wagę uwagi przy tokenach znajdujących się daleko od obu końców sekwencji.
Dla agentów konsekwencje są oczywiste. Nawet jeśli wszystko pasuje, model nie będzie traktować tego równomiernie. Krytyczne ograniczenie przy poziomie 30 tysięcy tokenów w kontekście o pojemności 60 tysięcy tokenów może faktycznie pozostać niewidoczne.
Model zapomina, ponieważ system dostarcza mu więcej informacji, niż jest w stanie skutecznie przetworzyć, a nic nie kontroluje tego, co trafia do jego obszaru przetwarzania.
Pierwotny podział: pamięć krótkoterminowa i długoterminowa
Przed czterowarstwową taxonomią istnieje prostszy podział, który większość inżynierów ma w głowie. Pamięć w systemach AI dzieli się na krótkoterminową i długoterminową, a zrozumienie tego, czego każda z nich nie potrafi, motywuje do opracowania bardziej szczegółowej architektury.
Pamięć krótkoterminowa jest ograniczona do danej sesji i istnieje tylko podczas jednej interakcji. Przechowuje to, czego agent potrzebuje w tym momencie: stan bieżącego zadania, etap, na którym się znajduje, ostatni wynik użycia narzędzia oraz w trakcie tworzenia ślad rozumowania. W praktyce jest to okno kontekstowe, które jest bardzo dynamiczne, stale aktualizowane, kosztowne pod względem liczby tokenów i usuwane po zakończeniu sesji.
Dokumentacja LangChain dobrze to opisuje, określając pamięć krótkoterminową jako miejsce przechowywania historii rozmowy, wyników użycia narzędzi oraz tymczasowego stanu rozumowania. Nie wspomina jednak, że to właśnie ona jako pierwsza ulega przepełnieniu w większych skali. W procesie pracy składającym się z dwudziestu kroków agenta, pośrednie wyniki, refleksje, wyniki wyszukiwań i odpowiedzi narzędzi gromadzą się i konkurują o ten ograniczony przestrzeń. Przypomina to próbę utrzymania skomplikowanego planu w głowie, podczas gdy ktoś ciągle przerywa, podając nowe fakty.
Pamięć krótkoterminowa składa się z dwóch praktycznych komponentów:
- Stan zadania to roboczy zapis postępów: bieżący krok, zadania podrzędne do wykonania, stan wykonywania narzędzi oraz koordynacja pomiędzy agentami. Agent planujący podróż przechowuje cel podróży, limity budżetowe, sprawdzone loty oraz listę hoteli, aby uniknąć powtarzania pracy z drugiego kroku, gdy dotrze do ósmego.
- Wyniki pośrednie to efekty rozumowania w trakcie pracy: wyniki narzędzi, wygenerowane streszczenia, wyniki z SQL, fragmenty wyszukiwań oraz notatki tymczasowe. Zespół inżynieryjny Anthropic opisał wzorzec występujący we wszystkich systemach produkcyjnych, zgodnie z którym agenci, którzy przechowują notatki poza oknem kontekstowym i pobierają je później, radzą sobie lepiej przy długich zadaniach niż agenci, którzy próbują utrzymać wszystko w kontekście. Robienie notatek samo w sobie jest strategią pamięci krótkoterminowej.
Pamięć długoterminowa przetrwa sesje. Umożliwia ona agentowi przypomnienie sobie, że użytkownik woli zwięzłe odpowiedzi, że wdrożenie z zeszłego tygodnia nie powiodło się z powodu problemu z migracją lub że schemat bazy danych uległ zmianie w styczniu. LangChain definiuje ją jako pamięć, która utrzymuje się między wątkami i sesjami, a w praktyce oznacza to, że dane te wymagają bazy danych, a nie okna kontekstowego.
Istnieją dwa podstawowe rodzaje takiej pamięci:
- Preferencje użytkownika to sygnały personalizacji, które kumulują się z czasem – obejmują one styl pisania, ton, poziom zaawansowania technicznego, wybór narzędzi oraz nawyki formatowania. Asystent do pisania, który pamięta, że zawsze chcesz używać formatu markdown oraz wysokiego poziomu zaawansowania technicznego, staje się przydatny w sposób, którego nie może dorównać żaden system bez stanu, bez względu na jakość jego modelu.
Wiele zespołów popełnia te same błędy. Implementują pamięć krótkoterminową poprzez przepełnianie okna kontekstowego, a pamięć długoterminową poprzez tworzenie jednego magazynu wektorowego. Obie te metody są błędne lub w najlepszym przypadku niekompletne. Pierwsza zawodzi na większą skalę z powodów opisanych wyżej. Druga przestaje działać, ponieważ wyszukiwanie podobieństw to kiepski sposób na pobieranie logów sesji, preferencji czy reguł zachowań, a każdy typ pamięci długoterminowej wymaga innego backendu. Projekt czterowarstwowy to udoskonalona wersja podziału na pamięć krótkoterminową i długoterminową, która przyporządkowuje każdy typ pamięci odpowiedniemu sposobowi przechowywania.
Cztery typy pamięci, jedna taksonomia
Zespoły tworzące produkcyjne systemy AI niezależnie doszły do tej samej taksonomii. Nazwy się różnią, ale struktura pozostaje identyczna.
Projekt MemGPT, przedstawiony przez Charlesa Packera i jego współpracowników z UC Berkeley w 2023 roku, upopularnił ideę traktowania okna kontekstowego modeli LLM jak pamięci RAM – ograniczonej i cennej. System operacyjny rozszerza pamięć poprzez paginację i przechowywanie na dysku, a ramy agentów mogą w podobny sposób symulować długoterminową pamięć, przenosząc starszy kontekst do większego miejsca przechowywania i przywołując go w razie potrzeby. Ta idea później stała się podstawą Letty.
Zespół LangMem z LangChain, framework ReMem od Google DeepMind oraz architektura produkcyjna Mem0 opisują wszystkie warianty czterech typów:
- Pamięć robocza to wszystko, co znajduje się w danym kontekście w tym momencie: aktualne etapy rozmowy, pobraane dokumenty oraz wyniki działania narzędzi. Jest szybka, ma ograniczone zasoby i wymaga dużo zasobów obliczeniowych na każdy token.
- Pamięć epizodyczna przechowuje informacje o wcześniejszych interakcjach, np. to, co agent zrobił dla danego użytkownika w zeszły wtorek i jaki był wynik. Jest uporządkowana chronologicznie i dotyczy konkretnej sesji, zazwyczaj przechowywana w szybkim magazynie typu klucz-wartość lub bazie danych SQL, a dostępna poprzez kryterium najnowszości lub podobieństwa semantycznego.
- Pamięć semantyczna przechowuje ogólne fakty, wiedzę z danej dziedziny oraz informacje nabyte w trakcie pracy. Jest to warstwa RAG, która jest wykorzystywana dzięki podobieństwu embeddingów, gdy zadanie wymaga wiedzy z danej dziedziny lub wcześniejszych ustaleń.
Te typy odzwierciedlają zasady nauk poznawczych, które wyjaśniają, dlaczego zespoły do nich dążą. Odpowiadają one systemom pamięci epizodycznej, semantycznej i proceduralnej z neurobiologii ludzkiej, a pamięć robocza jako okno kontekstowe to coś więcej niż metafora – jest to najbliższy funkcjonalny analog dostępny obecnie.
Wybór sposobu przechowywania: SQL, wektorowy czy grafowy
Która baza danych powinna przechowywać pamięć? Zanim napisze się kod, uczciwa odpowiedź brzmi, że zależy to od rodzaju pamięci.
Bazy danych wektorowych (Weaviate, Pinecone, Qdrant, Chroma) nadają się do przechowywania pamięci semantycznej, ponieważ umożliwiają szybkie wyszukiwanie podobieństw w embeddingach na dużą skalę, a większość z nich oferuje obecnie hybrydowe metody wyszukiwania łączące BM25 z wektorami, które wyraźnie przewyższają czyste wyszukiwanie wektorowe w przypadku tekstu. Brakuje im jednak rozumowania temporalnego oraz zapytań relacyjnych, a dwa sprzeczne fakty o niemal identycznych embeddingach mogą wprowadzić w błąd. Unikaj przechowywania w nich rekordów epizodycznych oraz ustrukturyzowanych śladów audytowych.
Bazy danych SQL, takie jak PostgreSQL i SQLite, nadają się do przechowywania informacji epizodycznych. Zapytania są strukturyzowane według ID użytkownika, sesji, typu zadania lub daty i godziny, objętość zapisów jest niska, a gwarancje ACID są istotne, gdy agent rejestruje to, co zrobił i kiedy. PostgreSQL z rozszerzeniem pgvector oferuje zapytania relacyjne oraz wyszukiwanie wektorowe w jednym systemie. Przy mniej niż dziesięciu milionach rekordów to rozwiązanie najbardziej praktyczne – zapewnia znajome operacje, tani hosting oraz możliwość używania SQL do debugowania.
Bazy danych grafowe, takie jak Neo4j, nadają się do systemów pamięciowych tylko wtedy, gdy dane epizodyczne lub semantyczne mają złożoną strukturę relacyjną wymagającą zapytań wieloetapowych, np. kto zgłosił co do kogo trzy miesiące temu w momencie napisania dokumentu. Proste rekordy zadań agentów nie kwalifikują się do takiego podejścia. Wiele zespołów decyduje się na bazy grafowe, by później odkryć, że w rzeczywistości potrzebowały SQL.
Wdrożenia produkcyjne opierają się głównie na przechowywaniu relacyjnym (SQL lub PostgreSQL) do przechowywania odcinków i profili użytkowników, na magazynie wektorowym do wyszukiwań semantycznych oraz na stanach wewnątrz procesu do utrzymywania kontekstu pracy; rozwiązaniem hybrydowym dla zespołów szukających mniejszego obciążenia operacyjnego jest PostgreSQL w połączeniu z pgvector. Wybierz poziom przechowywania w zależności od funkcji, które musi pełnić typ pamięci, i nie wektoruj wszystkiego.
Szczegół 1: pamięć buforowa z funkcją podsumowywania
Najprostszą formą jest kolejka deklinująca z ograniczeniem liczby wiadomości do określonej wartości:
from collections import deque
from typing import List, Dict
class BufferMemory:
def __init__(self, max_turns: int = 10):
# Each turn = one user message + one assistant message
self.history: deque = deque(maxlen=max_turns * 2)
def add_message(self, role: str, content: str):
self.history.append({"role": role, "content": content})
def get_context(self) -> List[Dict]:
return list(self.history)
def token_estimate(self) -> int:
total_chars = sum(len(m["content"]) for m in self.history)
return total_chars // 4 # rough approximation
def clear(self):
self.history.clear()
To sprawdza się przy krótkich rozmowach. W przypadku dłuższych rozmów starsze wiadomości po prostu znikaą na końcu, co utrudnia pracę agentowi. Lepszym podejściem, które LangMem określa jako metodę „gorącej ścieżki”, jest stworzenie podsumowania ze starszych wiadomości przed ich usunięciem:
import os
from collections import deque
from typing import List, Dict
from openai import OpenAI
class SummarizingBufferMemory:
def __init__(self, max_turns: int = 8, summary_batch: int = 4):
self.client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
self.model = "deepseek/deepseek-v4-flash-0731"
self.recent: deque = deque(maxlen=max_turns * 2)
self.rolling_summary: str = ""
self.summary_batch = summary_batch
def add_message(self, role: str, content: str):
if len(self.recent) == self.recent.maxlen:
self._compress_oldest()
self.recent.append({"role": role, "content": content})
def _compress_oldest(self):
batch = [self.recent.popleft() for _ in range(min(self.summary_batch * 2, len(self.recent)))]
text = "\n".join(f"{m['role']}: {m['content']}" for m in batch)
response = self.client.chat.completions.create(
model=self.model,
max_tokens=300,
messages=[
{
"role": "system",
"content": "You compress conversation history. Reply with the summary only.",
},
{
"role": "user",
"content": (
"Summarize this conversation segment in 2-4 sentences. "
"Preserve any decisions made, constraints stated, and conclusions reached.\n\n"
f"{text}"
),
},
],
)
new_summary = response.choices[0].message.content.strip()
if self.rolling_summary:
self.rolling_summary = f"{self.rolling_summary} | {new_summary}"
else:
self.rolling_summary = new_summary
def get_context(self) -> List[Dict]:
context = []
if self.rolling_summary:
context.append({
"role": "system",
"content": f"[Prior conversation summary: {self.rolling_summary}]",
})
context.extend(list(self.recent))
return context
LangMem oddziela sumaryzację ścieżki aktywnej, która działa w czasie rzeczywistym i powoduje opóźnienie dla użytkownika, od sumaryzacji w tle, która jest wykonywana po zakończeniu sesji. W aplikacjach przeznaczonych dla użytkowników sumaryzacja w tle jest prawie zawsze lepsza: wyniki są lepsze, ponieważ można przeanalizować całą rozmowę, a nikt nie musi czekać. W przypadku agentów autonomicznych działających bez nadzoru ścieżka aktywna jest do przyjęcia, ponieważ i tak nie ma czekającego użytkownika.
Szczegół 2: pamięć epizodyczna, która sprawdza własne cytaty
Tutaj wiele zespołów zawodzi: budują mechanizmy wyszukiwania, ale nigdy nie rejestrują tego, co agent faktycznie zrobił.
Blog inżynieryjny GitHub Copilot opisuje wzorzec wart naśladowania. Każda pozycja pamięci zawiera odniesienia, czyli konkretne adresy takie jak ścieżki plików, URL-y, nazwy narzędzi oraz numery wierszy, które umożliwiają sprawdzenie podczas wykorzystywania danych, czy przechowywana informacja nadal jest aktualna. GitHub nazywa to walidacją w czasie rzeczywistym. Gdy agent ładuje dane z pamięci na początku zadania, sprawdza, czy odwołane źródło nadal istnieje i nadal potwierdza podaną informację. W przeciwnym razie dane są korygowane lub usuwane, dzięki czemu zbiór informacji sam się oczyszcza bez konieczności ręcznej edycji poza siecią.
GitHub odnotował wyraźny efekt: agent do programowania z funkcją pamięci osiągnął 7% wzrostu szybkości łączenia pull requestów – 90% w porównaniu z 83%.
import json
import sqlite3
from datetime import datetime, timedelta
from dataclasses import dataclass, field
@dataclass
class Episode:
task_type: str
user_query: str
outcome: str
tools_used: list
citations: list # source references for validation
duration_seconds: float
success: bool
session_id: str = ""
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
class EpisodicMemory:
EXPIRY_DAYS = 28 # GitHub Copilot's production default
def __init__(self, db_path: str = "episodes.db"):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_schema()
def _init_schema(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS episodes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_type TEXT NOT NULL,
user_query TEXT,
outcome TEXT,
tools_used TEXT,
citations TEXT,
duration_s REAL,
success INTEGER,
session_id TEXT,
created_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_task_type ON episodes (task_type, success, created_at);
""")
self.conn.commit()
def record(self, ep: Episode):
self.conn.execute(
"""
INSERT INTO episodes
(task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
ep.task_type, ep.user_query, ep.outcome,
json.dumps(ep.tools_used), json.dumps(ep.citations),
ep.duration_seconds, int(ep.success),
ep.session_id, ep.created_at,
),
)
self.conn.commit()
def recall_similar(self, task_type: str, limit: int = 3) -> list[Episode]:
cutoff = (datetime.utcnow() - timedelta(days=self.EXPIRY_DAYS)).isoformat()
cursor = self.conn.execute(
"""
SELECT task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at
FROM episodes
WHERE task_type = ? AND success = 1 AND created_at > ?
ORDER BY created_at DESC
LIMIT ?
""",
(task_type, cutoff, limit),
)
return [
Episode(
task_type=r[0], user_query=r[1], outcome=r[2],
tools_used=json.loads(r[3]), citations=json.loads(r[4]),
duration_seconds=r[5], success=bool(r[6]),
session_id=r[7], created_at=r[8],
)
for r in cursor.fetchall()
]
def validate_citations(self, episode: Episode, validator_fn) -> bool:
"""
validator_fn(citation: str) -> bool
Check if cited sources are still valid (file exists, URL responds, etc.)
Return False if any citation fails; the episode should be discarded.
"""
return all(validator_fn(c) for c in episode.citations)
Okres ważności wynoszący 28 dni jest zgodny z wyborem GitHuba w środowisku produkcyjnym. Stare epizody nie są usuwane, ale tracą możliwość odzyskania, a zapytanie LIMIT przeprowadzone na indeksie filtrowanym według daty jest szybkie i nie wymaga dodatkowych zadań kuratorskich w tle.
Szczegół 3: pamięć semantyczna wykorzystująca hybrydowe wyszukiwanie Weaviate
To jest warstwa RAG. Większość przewodników omawia tylko konfigurację i zatrzymanie działania, podczas gdy prawdziwe problemy w środowisku produkcyjnym dotyczą jakości odzyskiwania informacji oraz ich pochodzenia.
Częstym pierwszym błędem jest ustawienie pamięci semantycznej na tryb tylko do odczytu. Agent działający przez długi czas wymaga możliwości zarówno odczytu, jak i zapisu: gdy agent znajdzie fakt podczas wykonywania zadania, zweryfikuje go i uzna, że warto go zachować, powinien zostać zapisany do pamięci semantycznej dla przyszłych sesji. Drugim błędem jest poleganie wyłącznie na podobieństwie kosinowym. Hybrydowe wyszukiwanie Weaviate łączy dopasowywanie słów kluczowych BM25 z podobieństwem wektorowym, co zazwyczaj daje lepsze wyniki niż każda z tych metod osobno przy odzyskiwaniu informacji z pamięci agenta.
Po trzecie, i to właśnie otwiera drogę do zatruwania – wpisy często nie mają żadnego źródła pochodzenia. Każdy z nich wymaga źródła, oceny wiarygodności oraz identyfikatora sesji. Bez nich nie jest możliwe wyszukiwanie z uwzględnieniem wagi zaufania, stare lub wątpliwe wpisy nie mogą zostać obniżone w rankingu, a wykrycie błędnego wyniku wyszukiwania staje się domysłem.
import uuid as uuid_lib
from datetime import datetime
import weaviate
from weaviate.classes.config import Configure, Property, DataType, VectorDistances
from weaviate.classes.query import MetadataQuery, HybridFusion, Filter
def embed(text: str) -> list[float]:
"""
Swap in any embedding source: OpenAI, Cohere, sentence-transformers, etc.
Returns a normalized float vector.
"""
from sentence_transformers import SentenceTransformer
_model = SentenceTransformer("all-MiniLM-L6-v2")
return _model.encode(text, normalize_embeddings=True).tolist()
class WeaviateSemanticMemory:
COLLECTION = "AgentMemory"
def __init__(self, host: str = "localhost", port: int = 8080):
self.client = weaviate.connect_to_local(host=host, port=port)
self._ensure_collection()
def _ensure_collection(self):
if self.client.collections.exists(self.COLLECTION):
return
self.client.collections.create(
name=self.COLLECTION,
# We provide our own vectors; no built-in vectorizer needed.
# Swap to Configure.Vectorizer.text2vec_openai() if you prefer managed embedding.
vectorizer_config=Configure.Vectorizer.none(),
vector_index_config=Configure.VectorIndex.hnsw(
distance_metric=VectorDistances.COSINE
),
properties=[
Property(name="content", data_type=DataType.TEXT),
Property(name="task_type", data_type=DataType.TEXT),
Property(name="source", data_type=DataType.TEXT),
Property(name="citations", data_type=DataType.TEXT),
Property(name="session_id", data_type=DataType.TEXT),
Property(name="confidence", data_type=DataType.NUMBER),
Property(name="created_at", data_type=DataType.TEXT),
],
)
def store(
self,
content: str,
task_type: str = "",
source: str = "agent",
citations: str = "",
session_id: str = "",
confidence: float = 1.0,
) -> str:
collection = self.client.collections.get(self.COLLECTION)
doc_id = str(uuid_lib.uuid4())
collection.data.insert(
properties={
"content": content,
"task_type": task_type,
"source": source,
"citations": citations,
"session_id": session_id,
"confidence": confidence,
"created_at": datetime.utcnow().isoformat(),
},
vector=embed(content),
uuid=doc_id,
)
return doc_id
def retrieve_hybrid(
self,
query: str,
n_results: int = 5,
min_confidence: float = 0.6,
task_type: str = None,
) -> list[dict]:
collection = self.client.collections.get(self.COLLECTION)
# Filter by confidence floor and optionally by task type
confidence_filter = Filter.by_property("confidence").greater_or_equal(min_confidence)
if task_type:
active_filter = (
Filter.by_property("task_type").equal(task_type) & confidence_filter
)
else:
active_filter = confidence_filter
results = collection.query.hybrid(
query=query,
vector=embed(query),
limit=n_results,
fusion_type=HybridFusion.RELATIVE_SCORE,
filters=active_filter,
return_metadata=MetadataQuery(score=True),
)
return [
{
"content": obj.properties["content"],
"score": obj.metadata.score,
"confidence": obj.properties.get("confidence", 1.0),
"source": obj.properties.get("source", ""),
"citations": obj.properties.get("citations", ""),
"uuid": str(obj.uuid),
}
for obj in results.objects
]
def update_confidence(self, doc_id: str, new_confidence: float):
collection = self.client.collections.get(self.COLLECTION)
collection.data.update(
uuid=doc_id,
properties={"confidence": new_confidence},
)
def close(self):
self.client.close()
Dwie informacje zawarte w retrieve_hybrid brakuje w wielu implementacjach. Po pierwsze, koniec hybrydowy Weaviate otrzymuje zarówno surowy tekst zapytania (wykorzystywany do BM25), jak i jego embedding (wykorzystywany do wyszukiwania wektorowego). Dokładne dopasowania słów kluczowych pochodzą z pierwszego elementu, a bliskość semantyczna – z drugiego, przy czym połączenie to zawsze daje lepsze wyniki niż każdy z nich osobno. Po drugie, istnieje filtr min_confidence, który wyklucza z kontekstu pamięci o niskim poziomie zaufania, niezależnie od ich bliskości semantycznej. W praktyce jest to środek bezpieczeństwa, a nie tylko modyfikacja stopnia istotności.
Metoda update_confidence zapewnia ścieżkę zwrotną. Jeśli własne rozumowanie agenta jest sprzeczne z pamięcią pobraną podczas wykonywania zadania, należy obniżyć jej poziom zaufania. Jeśli weryfikacja cytatu to potwierdzi, należy go podnieść. Zbiór pamięci powinien uczyć się na podstawie swojego użytkowania.
Szczegół 4: pamięć proceduralna i ewolucja promptów
To warstwa, która najczęściej jest pomijana. Prosty projekt polega na pliku JSON zawierającym zasady, które agent czyta, przestrzega i rozszerza po niepowodzeniach. Działa to, ale napotyka ograniczenia, ponieważ zasady w oddzielnym pliku konkurują o przestrzeń kontekstową i wymagają wyraźnego pobierania.
Inżynierowie LangMem z LangChain opublikowali pomysł wart skopiowania: w środowisku produkcyjnym pamięć proceduralna istnieje jako sam prompt systemu. Nie jest przechowywana obok promptu ani do niego pobierana – to właśnie jest prompt. Za każdym razem, gdy niepowodzenie uczy agenta nowej zasady postępowania, sam prompt jest przepisywany; praktykę tę nazywają optymalizacją promptu jako pamięci proceduralnej. Nabyte zachowania stają się instrukcjami w języku naturalnym, które model czyta na początku każdej interakcji.
Prostsza wersja z plikiem JSON to punkt wyjścia, od którego powinny zacząć się większość zespołów:
import json
import os
import re
from datetime import datetime, timedelta, timezone
from openai import OpenAI
MODEL = "deepseek/deepseek-v4-flash-0731"
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
class ProceduralMemory:
def __init__(self, rules_path: str = "procedural_rules.json"):
self.rules_path = rules_path
self.rules: list[dict] = self._load()
def _load(self) -> list[dict]:
try:
with open(self.rules_path) as f:
return json.load(f)
except FileNotFoundError:
return []
def add_rule(self, situation: str, action: str, reason: str, confidence: float = 1.0):
self.rules.append({
"situation": situation,
"action": action,
"reason": reason,
"confidence": float(confidence),
"added_at": datetime.now(timezone.utc).isoformat(),
"trigger_count": 0,
})
self._save()
def get_applicable_rules(self, context: str, min_confidence: float = 0.7) -> list[dict]:
relevant = []
ctx_lower = context.lower()
dirty = False
for rule in self.rules:
if rule.get("confidence", 1.0) < min_confidence:
continue
keywords = rule["situation"].lower().split()
if not keywords:
continue
hits = sum(1 for kw in keywords if kw in ctx_lower)
if hits >= max(1, len(keywords) // 3):
rule["trigger_count"] = rule.get("trigger_count", 0) + 1
relevant.append(rule)
dirty = True
if dirty:
self._save()
relevant.sort(key=lambda r: r.get("confidence", 1.0), reverse=True)
return relevant[:5]
def prune_stale(self, max_age_days: int = 60, min_triggers: int = 2):
cutoff = (datetime.now(timezone.utc) - timedelta(days=max_age_days)).isoformat()
self.rules = [
r for r in self.rules
if r.get("added_at", "") > cutoff or r.get("trigger_count", 0) >= min_triggers
]
self._save()
def _save(self):
tmp = f"{self.rules_path}.tmp"
with open(tmp, "w") as f:
json.dump(self.rules, f, indent=2)
os.replace(tmp, self.rules_path)
def _parse_json(text: str) -> dict:
text = text.strip()
fenced = re.search(r"```(?:json)?\s*(.*?)```", text, re.S)
if fenced:
text = fenced.group(1).strip()
return json.loads(text)
def extract_rule_from_failure(failure_trace: str, memory: ProceduralMemory):
response = client.chat.completions.create(
model=MODEL,
max_tokens=250,
response_format={"type": "json_object"},
messages=[
{
"role": "system",
"content": "You return only a JSON object. No prose, no code fences.",
},
{
"role": "user",
"content": (
"A task failed. Extract one behavioral rule to prevent this failure.\n"
"Return ONLY valid JSON with keys: situation, action, reason, "
"confidence (0.0-1.0)\n"
f"Failure trace:\n{failure_trace}"
),
},
],
)
try:
rule = _parse_json(response.choices[0].message.content)
except (json.JSONDecodeError, AttributeError, TypeError):
return None
if not all(k in rule for k in ("situation", "action", "reason")):
return None
memory.add_rule(
situation=str(rule["situation"]),
action=str(rule["action"]),
reason=str(rule["reason"]),
confidence=float(rule.get("confidence", 1.0)),
)
return rule
prune_stale radzi sobie z problemem zapominania na poziomie proceduralnym. Reguła utworzona po jednym nietypowym błędzie i nigdy więcej nie uruchomiona powinna zostać wycofana, natomiast reguła, która jest uruchamiana regularnie, pozostaje bez względu na to, jak stara jest. To prosty narzędzie, ale zapominanie jest najłatwiejsze do zaimplementowania właśnie na tym poziomie.
Gdy agent działa w sposób ciągły, naturalnym następnym krokiem jest włączenie reguł, które są zarówno wiarygodne, jak i często używane, bezpośrednio do promptu systemu. LangMem automatyzuje to za pomocą pętli optymalizacji promptu, natomiast ręczna edycja jest doskonała dla mniejszych konfiguracji.
Łączenie warstw
Następna lista pokazuje, jak cztery warstwy współpracują ze sobą w środowisku produkcyjnym. Jedno ograniczenie jest niepodważalne: zapisywanie do pamięci nigdy nie może opóźniać odpowiedzi.
import json
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
@dataclass
class TaskResult:
task_type: str
query: str
outcome: str
tools_used: list
citations: list
duration_seconds: float
success: bool
new_facts: list[dict] = field(default_factory=list)
session_id: str = ""
class MemoryManager:
def __init__(
self,
weaviate_host: str = "localhost",
episodic_db: str = "episodes.db",
rules_path: str = "procedural_rules.json",
):
self.buffer = SummarizingBufferMemory(max_turns=8)
self.episodic = EpisodicMemory(db_path=episodic_db)
self.semantic = WeaviateSemanticMemory(host=weaviate_host)
self.procedural = ProceduralMemory(rules_path=rules_path)
self._pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="memory_write")
def build_context(self, query: str, task_type: str) -> list[dict]:
context: list[dict] = []
# Procedural rules first: they constrain behavior throughout the task
rules = self.procedural.get_applicable_rules(query)
if rules:
rules_text = "\n".join(
f"- When '{r['situation']}': {r['action']} (reason: {r['reason']})"
for r in rules
)
context.append({"role": "user", "content": f"[Behavioral rules:\n{rules_text}]"})
# Semantic facts: domain knowledge and past discoveries
facts = self.semantic.retrieve_hybrid(query, n_results=5, min_confidence=0.6)
if facts:
facts_text = "\n".join(f"- {f['content']}" for f in facts)
context.append({"role": "user", "content": f"[Relevant knowledge:\n{facts_text}]"})
# Past episodes: outcome templates for similar tasks
episodes = self.episodic.recall_similar(task_type, limit=3)
if episodes:
ep_text = "\n".join(
f"- Outcome: {e.outcome} (tools: {', '.join(e.tools_used)})"
for e in episodes
)
context.append({"role": "user", "content": f"[Past similar tasks:\n{ep_text}]"})
# Current conversation last: the model reads this most carefully
context.extend(self.buffer.get_context())
return context
def record_turn(self, role: str, content: str):
self.buffer.add_message(role, content)
def persist(self, result: TaskResult):
# Submit to thread pool and return immediately; never block the caller
self._pool.submit(self._persist_worker, result)
def _persist_worker(self, result: TaskResult):
ep = Episode(
task_type=result.task_type,
user_query=result.query,
outcome=result.outcome,
tools_used=result.tools_used,
citations=result.citations,
duration_seconds=result.duration_seconds,
success=result.success,
session_id=result.session_id,
)
self.episodic.record(ep)
for fact in result.new_facts:
self.semantic.store(**fact)
if not result.success:
extract_rule_from_failure(result.outcome, self.procedural)
def shutdown(self):
self._pool.shutdown(wait=True)
self.semantic.close()
Kolejność elementów w funkcji build_context jest celowa. Badania nad uprzedzeniami związanymi z pozycją w uwadze potwierdzają, że modele przywiązują większą wagę do początku i końca okna. Zasady behawioralne znajdują się na górze, ponieważ są to ograniczenia, które powinny kształtować wszystko, co następuje później. Obecna rozmowa umieszczona jest na dole, ponieważ to właśnie jej model musi najbardziej uwzględnić, aby udzielić odpowiedzi. Fakty semantyczne oraz wcześniejsze epizody znajdują się pomiędzy nimi, dostarczając informacji referencyjnych bez przeszkadzania w wykonywaniu aktualnego zadania.
Linia self._pool.submit(self._persist_worker, result) stanowi najważniejszy wybór inżynieryjny w tej klasie. Każda operacja zapisu zwraca wynik natychmiast, dzięki czemu odpowiedź LLM jest już przekazywana użytkownikowi przed rozpoczęciem jakiegokolwiek zapisu do bazy danych.
Trzy decyzje, które kształtują architekturę
Gdy już są wdrożone cztery warstwy, trzy decyzje architektoniczne wpływają na wszystko inne, a wiele zespołów popełnia przynajmniej jedną błąd.
Zapisy w trybie „hot-path” lub w tle. Zespół LangMem włożył w to znaczne wysiłki. Zapisy w trybie „hot-path” odbywają się podczas rozmowy, gdy agent wykrywa fakt wart przechowania i natychmiast dokonuje zapisu. Zapisy w tle następują po zakończeniu sesji i analizują całą rozmowę, aby określić, co należy zachować. Tryb „hot-path” powoduje opóźnienia, ponieważ każda funkcja semantic.store() wywołana w pętli odpowiedzi stanowi operację zatwierdzenia w bazie danych, której użytkownik musi czekać. Zapisy w tle poprawiają jakość przypominania informacji, ponieważ można przeanalizować tematy i decyzje z całej rozmowy oraz przechować bardziej szczegółowe podsumowania. W aplikacjach produkcyjnych dostępnych dla użytkowników rozwiązanie oparte na zapisach w tle jest niemal zawsze najlepsze. Zbiór wątków w funkcji persist() zapewnia większość korzyści, unikając przy tym całkowicie asynchronicznego procesu.
Pobieranie lub streszczanie. Rozwiązują one różne problemy, a zespoły często je mylą. Pobieranie oznacza znalezienie odpowiedniego fragmentu w dużym korpusie. Zapytanie jest konkretne, a korpus duży i dość stabilny, np. dokumentacja, katalogi produktów lub kod. Metoda ta zawodzi, gdy zapytanie jest złożone, istotne informacje są rozproszone w wielu fragmentach lub korpus zmienia się szybciej niż indeks embeddingów. Streszczanie oznacza kompresję rozmowy lub dokumentu, aby pasował do kontekstu. Działa dobrze, gdy dane wejściowe są sekwencyjne, jak w historii rozmów, i dopuszczalna jest pewna utrata informacji; zawodzi natomiast, gdy pominięte szczegóły okazują się ważne. W 2025 roku branża zaczęła zastępować oba te terminy pojęciem „inżynieria kontekstu”: pytanie nie brzmi już o różnicy między pobieraniem a streszczaniem, lecz o tym, jaka kombinacja przechowywanej wiedzy, pobranych fragmentów i skompresowanej historii zapewni najlepsze wyniki.
To kontekst dla tej konkretnej operacji wnioskowania. Metodabuild_context menedżera pamięci stanowi element inżynierii kontekstu, ponieważ decyduje, z których czterech źródeł pobrać dane, ile z każdego oraz w jakiej kolejności połączyć te elementy. Pochodzenie od samego początku. Nie można dodać informacji o pochodzeniu do systemu pamięci w późniejszym etapie. Już przy pierwszej operacji zapisu każda pozycja powinna zawierać informację o jej źródle, ocenę wiarygodności, identyfikator sesji oraz czas utworzenia. Bez tych danych nie ma możliwości wyszukiwania z uwzględnieniem stopnia zaufania, filtrowania według terminu ostatniej modyfikacji ani audytowania błędnych wyników wyszukiwania. Zespoły, które pomijają kwestię pochodzenia już w pierwszym etapie pracy, kończą na tygodniach poświęconych dodawaniu tej informacji do bazy danych, która i tak zawiera tysiące pozycji bez niej.
Zatrucie pamięcią: wzorce ataków i metody obrony
To jest część, której przewodniki projektowania pamięci zazwyczaj nie uwzględniają, a powinna znaleźć się w każdym z nich.
W artykule opublikowanym na arXiv w 2025 roku przedstawiono framework ataku MINJA, który należy do najbardziej szczegółowych publicznych analiz zatruwania pamięci w agentach LLM. Jego autorzy donoszą, że poprzez zwykłe interakcje użytkownika, bez uprawnień administratora, atakujący mogą osiągnąć wskaźniki skuteczności iniekcji przekraczające 95% przy użyciu trzechetapowego schematu. Etap łączący powiązuje niewinny, legalny temat z celem, który zamierza osiągnąć atakujący. Etap wskazujący dostarcza subtelnych sygnałów, które sprawiają, że szkodliwa pamięć wydaje się wiarygodna. Etap stopniowego skracania kompresuje iniekcję do minimalnych rozmiarów, które przetrwają ekstrakcję pamięci.
To, co odróżnia zatruwanie pamięci od zwykłej iniekcji promptu, to fakt, że zatrute pamięci przenikają między sesjami i się kumulują. Jedna udana iniekcja w pierwszej sesji może wpływać na wszystkie kolejne sesje, nieograniczenie długo.
Odrębny artykuł z 2025 roku na temat bayesowskich metod obrony przed nieufnością opisał pięciowarstwową architekturę obejmującą całą powierzchnię ataku:
- Moderacja danych wejściowych w momencie zapisu. Przed każdym zapisem do pamięci obliczana jest łączna ocena nieufności na podstawie kilku niezależnych wskaźników: pochodzenia źródła, analizy semantycznej wzorców instrukcji oraz wykrywania anomalii w kontekście historii użytkownika. Żaden pojedynczy wskaźnik nie jest wystarczający, a metody typu MINJA są zaprojektowane tak, by przechodzić testy oparte na pojedynczych wskaźnikach.
- Oczyszczanie instrukcji przed zapisem. Jeśli dane do zapisu przypominają polecenie ("zawsze", ">nigdy", ">musisz", ">nie rób"), należy je oczyścić lub odrzucić zamiast je zapisać. Prawdziwe, faktograficzne dane w pamięci nigdy nie muszą wydawać poleceń agentowi.
min_confidence w warstwie 3 realizowanej przez Weaviate. Informacje o niskim pochodzeniu pozostają poza kontekstem, niezależnie od ich podobieństwa. W pełniejszej implementacji pole zaufania jest aktualizowane na podstawie wyników weryfikacji, dzięki czemu informacje potwierdzone zyskują zaufanie, a te sprzeczne je tracą.Dla większości zespołów rozsądnym porządkiem jest najpierw umieszczanie tagów wskazujących na pochodzenie przy każdej operacji zapisu, następnie wykrywanie instrukcji w momencie zapisu, a potem wyszukiwanie z uwzględnieniem wagi zaufania. Monitorowanie zachowań jest na końcu, ponieważ wymaga bazowego profilu zachowań, a jego stworzenie potrzebuje tygodni danych z rzeczywistej pracy systemu.
Jeszcze jedno uwagi: badania przeprowadzone na rzeczywistych systemach pamięci z już istniejącymi legalnymi wpisami wykazały, że realistyczne warunki znacznie obniżają skuteczność ataków w porównaniu z warunkami laboratoryjnymi, ponieważ skażone wpisy konkurują z legalnymi danymi o miejsca w pamięci. To nie jest powód, by pomijać środki ochronne – to powód, by utrzymywać legalne wpisy na wysokim poziomie jakości i wiarygodności, ponieważ stanowią one pierwszą linię obrony.
Co zbudować najpierw
Nie wszystkie cztery warstwy są potrzebne od razu. W przypadku większości zastosowań AI dwa z nich pokrywają około 80% problemów z pamięcią: Warstwa 1 (bufor plus streszczanie) oraz Warstwa 3 (Weaviate z wyszukiwaniem hybrydowym, filtrem pewności i informacjami o pochodzeniu). Razem radzą sobie z dwoma najczęstszymi problemami – agentem, który zapomina w trakcie wykonywania zadania, oraz takim, który nie posiada wiedzy specjalistycznej. Zbuduj obie warstwy, porównaj stopień ukończenia zadań przed i po ich wdrożeniu, a dopiero wtedy oceniaj, czego brakuje.
Dodaj pamięć epizodyczną, gdy zauważysz, że zadajesz sobie pytanie, dlaczego agent zapomniał o tym, co osiągnął w poprzedniej sesji – to właśnie takie pytanie jest sygnałem. Nie dodawaj jej wcześniej. Dodaj pamięć proceduralną, gdy ta sama klasa błędów powtarza się w różnych zadaniach, a modyfikacja zapytań nie pomaga, ponieważ wzorzec zachowań wymaga pamięci behawioralnej.
PostgreSQL z pgvector stanowi rozsądny standard dla większości zespołów. Jedna instancja bazy danych z poleceniem CREATE EXTENSION IF NOT EXISTS vector zapewnia zgodność z zasadami ACID, elastyczny SQL, wyszukiwanie podobieństw wektorowych oraz hybrydową ocenę wyników; według jednych szacunków koszt hostingu wynosi około 45 dolarów miesięcznie przy 10 tysiącach pamięci. Samodzielna baza danych wektorowa staje się opłacalna dopiero wtedy, gdy ilość semantycznych pamięci rośnie do takiego poziomu, że indeks HNSW w PostgreSQL staje się powodem spowolnienia – co większość zespołów nigdy nie doświadczy.
Zrób tak, aby każda operacja zapisu pamięci była asynchroniczna. Reakcja na żądanie użytkownika nie powinna zależeć od potwierdzenia zmiany w bazie danych, dlatego przekaż tę pracę do puli wątków i idź dalej.
Pomijaj również lokalne magazyny wektorów w środowisku produkcyjnym. Rozwiązanie bez trwałego przechowywania traci całą zgromadzoną wiedzę po pierwszym restartie serwera – to błąd, którego niewielu ludzi popełnia dwa razy.
Literatura pokrewna
- Pamięć agenta semantycznego w LangGraph z Qdrant i przywoływanie danych między wątkami — Nadaj agentowi LangGraph pamięć długoterminową za pomocą Qdrant: umieść tam fakty użytkowników, przywołuj je przy różnych identyfikatorach wątków oraz dodaj filtry ładunku danych dla izolacji wielu użytkowników.
- Lokalny agent programistyczny w Pythonie: narzędzie Ollama, które pisze i uruchamia kod — Stwórz prosty cykl agenta przy użyciu Ollama i llama3.1:8b, który tworzy pliki w języku Python, je uruchamia i naprawia błędy, bez konieczności korzystania z kluczy API czy ponoszenia kosztów chmurowych.